src/genro_asgi/applications/server_sections/users_section.py¶
Source from this local checkout, regenerated when the reader rebuilds.
Line links use #L<number>; a GitHub line range opens its first line.
1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 # https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """The ``_server/users`` section: SUPERADMIN-gated user management.16 17 ``UsersSection`` is a ``RoutingClass`` the ``ServerApplication`` attaches under18 ``users`` (endpoints at ``/_server/users/...``). Every route is gated19 ``auth_rule="SUPERADMIN"`` — the section is ALWAYS declared (fixed structure,20 D26), and each handler answers the ``{"error": ...}`` shape (coherent with21 ``login``) when the server has no ``user_store`` wired.22 23 The credential invariant: ``password_hash`` NEVER crosses the wire. ``list``24 and ``get`` strip it from the record; ``save`` never accepts it (it merges the25 non-credential fields of the body over the stored record, preserving the hash);26 a password enters the system only as plaintext through ``create_user`` and27 ``set_password``, hashed server-side via ``UserStore.hash_password``.28 29 Route responsibilities are separated:30 31 - ``create_user`` — births a NEW record (error if the identity exists): the32 body carries ``password``/``password_confirm`` (the server checks they match)33 plus the metadata and ``tags``;34 - ``save`` — updates an EXISTING record only (error if absent): merges the35 body's metadata/tags over the stored record, ``password_hash`` untouched. The36 body is taken whole (``body_data``) so tomorrow's metadata fields need no37 signature change;38 - ``set_password`` — changes the credential of an existing record only, with the39 same ``password``/``password_confirm`` server-side check;40 - ``delete`` — removes a record.41 42 Parent (dual relationship): the ServerApplication, stored as43 ``self.application``; the store is reached via ``self.application.server.user_store``.44 """45 46 from __future__ import annotations47 48 from typing import TYPE_CHECKING, Any49 50 from genro_routes import RoutingClass, route51 52 if TYPE_CHECKING:53 from ...auth.user_store import UserStore54 from ..server_app import ServerApplication55 56 __all__ = ["UsersSection"]57 58 NO_STORE_ERROR = {"error": "User management is not available"}59 60 61 class UsersSection(RoutingClass):62 """The ``_server/users`` mount: SUPERADMIN CRUD over the server's UserStore.63 64 Note:65 Parent (dual relationship): the ServerApplication, stored as66 ``self.application``. The store is ``self.application.server.user_store``.67 """68 69 def __init__(self, application: ServerApplication) -> None:70 """Bind the section to its ServerApplication (dual relationship)."""71 self.application = application72 73 @property74 def user_store(self) -> UserStore | None:75 """The server's UserStore, or ``None`` when identity is unconfigured."""76 return getattr(self.application.server, "user_store", None)77 78 def public_record(self, record: dict[str, Any]) -> dict[str, Any]:79 """A record without its ``password_hash`` — the wire-safe projection."""80 return {k: v for k, v in record.items() if k != "password_hash"}81 82 def matched_password(self, body: dict[str, Any]) -> str | None:83 """The password when it is present and matches its confirmation, else None.84 85 The caller distinguishes a mismatch (this returns None) from an absent86 password by checking the body itself: only ``create_user`` requires one.87 """88 password = body.get("password")89 if password is None or password != body.get("password_confirm"):90 return None91 return password92 93 @route(auth_rule="SUPERADMIN")94 def list(self) -> dict[str, Any]:95 """Every user record, ``password_hash`` stripped."""96 store = self.user_store97 if store is None:98 return NO_STORE_ERROR99 return {"users": [self.public_record(r) for r in store.load_all()]}100 101 @route(auth_rule="SUPERADMIN")102 def get(self, identity: str = "") -> dict[str, Any]:103 """One user record, ``password_hash`` stripped."""104 store = self.user_store105 if store is None:106 return NO_STORE_ERROR107 record = store.get(identity)108 if record is None:109 return {"error": f"No such user: {identity}"}110 return self.public_record(record)111 112 @route(auth_rule="SUPERADMIN", openapi_method="post")113 def create_user(self, identity: str = "", body_data: dict | None = None) -> dict[str, Any]:114 """Create a NEW user from the body (metadata + tags + password/confirm).115 116 Errors if the identity already exists. The password is validated against117 its confirmation and hashed server-side; ``password_hash`` never arrives118 pre-formed. Metadata and ``tags`` from the body land on the new record.119 A new record defaults to ``enabled: True`` and ``tags: []`` (creating a120 user with a password means letting them log in — the body can still say121 ``enabled: false`` to create it disabled); ``verify`` requires122 ``enabled`` and the login ``Avatar`` requires ``tags``, so a minimal123 body must still produce a working user.124 """125 store = self.user_store126 if store is None:127 return NO_STORE_ERROR128 if not identity:129 return {"error": "Identity is required"}130 if store.get(identity) is not None:131 return {"error": f"User already exists: {identity}"}132 body = body_data or {}133 password = self.matched_password(body)134 if password is None:135 return {"error": "Password and confirmation are required and must match"}136 record = self.metadata_of(body)137 record.setdefault("enabled", True)138 record.setdefault("tags", [])139 record["identity"] = identity140 record["password_hash"] = store.hash_password(password)141 store.save(record)142 return self.public_record(record)143 144 @route(auth_rule="SUPERADMIN", openapi_method="post")145 def save(self, identity: str = "", body_data: dict | None = None) -> dict[str, Any]:146 """Update an EXISTING record's metadata/tags; ``password_hash`` untouched.147 148 Errors if the user does not exist (creation is ``create_user``'s job).149 The body is merged whole over the stored record — new metadata fields150 persist with no signature change — but the credential never moves here.151 """152 store = self.user_store153 if store is None:154 return NO_STORE_ERROR155 record = store.get(identity)156 if record is None:157 return {"error": f"No such user: {identity}"}158 record.update(self.metadata_of(body_data or {}))159 store.save(record)160 return self.public_record(record)161 162 @route(auth_rule="SUPERADMIN", openapi_method="post")163 def set_password(self, identity: str = "", body_data: dict | None = None) -> dict[str, Any]:164 """Change an existing user's password (plaintext in, hashed server-side).165 166 The body carries ``password``/``password_confirm``; the server checks167 they match. Errors if the user does not exist.168 """169 store = self.user_store170 if store is None:171 return NO_STORE_ERROR172 record = store.get(identity)173 if record is None:174 return {"error": f"No such user: {identity}"}175 password = self.matched_password(body_data or {})176 if password is None:177 return {"error": "Password and confirmation are required and must match"}178 record["password_hash"] = store.hash_password(password)179 store.save(record)180 return {"identity": identity, "updated": True}181 182 @route(auth_rule="SUPERADMIN", openapi_method="post")183 def delete(self, identity: str = "") -> dict[str, Any]:184 """Remove a user; reports whether a record was actually removed."""185 store = self.user_store186 if store is None:187 return NO_STORE_ERROR188 return {"identity": identity, "deleted": store.delete(identity)}189 190 def metadata_of(self, body: dict[str, Any]) -> dict[str, Any]:191 """The non-credential, non-key fields of a body (never password/hash/identity).192 193 Credential and key fields are owned by their dedicated routes, so they194 are dropped here: whatever else the body carries is user metadata.195 """196 dropped = {"password", "password_confirm", "password_hash", "identity"}197 return {k: v for k, v in body.items() if k not in dropped}