src/genro_asgi/applications/server_sections/tokens_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/tokens`` section: SUPERADMIN-gated issued credentials.16 17 ``TokensSection`` is the ONE section for credentials the server issues: API18 keys (``gak_``) and short-lived JWTs. It is a ``RoutingClass`` the19 ``ServerApplication`` attaches under ``tokens`` (endpoints at20 ``/_server/tokens/...``), ALWAYS declared (fixed structure, D26). Every route is21 ``auth_rule="SUPERADMIN"`` and answers the ``{"error": ...}`` shape when the22 server has no ``api_key_store`` wired.23 24 The secret invariant mirrors the users section: an api key's ``secret_hash``25 NEVER crosses the wire (``list`` strips it), and the full ``gak_`` key is26 returned ONLY by ``issue``, once — it is never retrievable again (the record27 keeps only its hash).28 29 ``create_jwt`` mints a JWT signed with the FIRST symmetric (``HS*``) verifier in30 the ``auth`` config (``AuthCore.signing_jwt_config``): the token verifies against31 the same config, so no new key material is introduced. With no symmetric verifier32 configured it answers the error shape.33 34 Parent (dual relationship): the ServerApplication, stored as35 ``self.application``; the stores are reached via ``self.application.server``.36 """37 38 from __future__ import annotations39 40 import time41 from typing import TYPE_CHECKING, Any42 43 import jwt44 from genro_routes import RoutingClass, route45 46 if TYPE_CHECKING:47 from ...auth.api_key_store import ApiKeyStore48 from ...auth.core import AuthCore49 from ..server_app import ServerApplication50 51 __all__ = ["TokensSection"]52 53 NO_STORE_ERROR = {"error": "Token management is not available"}54 NO_JWT_ERROR = {"error": "No symmetric JWT signing key is configured"}55 56 57 class TokensSection(RoutingClass):58 """The ``_server/tokens`` mount: SUPERADMIN api-key registry + JWT minting.59 60 Note:61 Parent (dual relationship): the ServerApplication, stored as62 ``self.application``. The stores live on ``self.application.server``.63 """64 65 def __init__(self, application: ServerApplication) -> None:66 """Bind the section to its ServerApplication (dual relationship)."""67 self.application = application68 69 @property70 def api_key_store(self) -> ApiKeyStore | None:71 """The server's ApiKeyStore, or ``None`` when tokens are unconfigured."""72 return getattr(self.application.server, "api_key_store", None)73 74 @property75 def auth_core(self) -> AuthCore | None:76 """The server's AuthCore (carries the JWT signing seam), or ``None``."""77 return getattr(self.application.server, "auth_core", None)78 79 def public_record(self, record: dict[str, Any]) -> dict[str, Any]:80 """A key record without its ``secret_hash`` — the wire-safe projection."""81 return {k: v for k, v in record.items() if k != "secret_hash"}82 83 @route(auth_rule="SUPERADMIN")84 def list(self) -> dict[str, Any]:85 """Every api-key record, ``secret_hash`` stripped."""86 store = self.api_key_store87 if store is None:88 return NO_STORE_ERROR89 return {"tokens": [self.public_record(r) for r in store.load_all()]}90 91 @route(auth_rule="SUPERADMIN", openapi_method="post")92 def issue(self, body_data: dict | None = None) -> dict[str, Any]:93 """Mint an api key; return the full ``gak_`` key ONCE (never again).94 95 Body: ``label`` (required), ``tags`` (default ``[]``), ``expires_at``96 (POSIX timestamp or absent for a key that never expires).97 """98 store = self.api_key_store99 if store is None:100 return NO_STORE_ERROR101 body = body_data or {}102 label = body.get("label")103 if not label:104 return {"error": "Label is required"}105 key = store.issue(label, body.get("tags") or [], body.get("expires_at"))106 return {"key": key, "label": label}107 108 @route(auth_rule="SUPERADMIN", openapi_method="post")109 def revoke(self, key_id: str = "") -> dict[str, Any]:110 """Disable a key (the record stays, listed, for audit)."""111 store = self.api_key_store112 if store is None:113 return NO_STORE_ERROR114 return {"key_id": key_id, "revoked": store.revoke(key_id)}115 116 @route(auth_rule="SUPERADMIN", openapi_method="post")117 def delete(self, key_id: str = "") -> dict[str, Any]:118 """Remove a key record entirely."""119 store = self.api_key_store120 if store is None:121 return NO_STORE_ERROR122 return {"key_id": key_id, "deleted": store.delete(key_id)}123 124 @route(auth_rule="SUPERADMIN", openapi_method="post")125 def create_jwt(self, body_data: dict | None = None) -> dict[str, Any]:126 """Mint a JWT signed with the first symmetric verifier of the auth config.127 128 Body: ``sub`` (required — the token's subject), ``tags`` (default ``[]``),129 ``expires_in`` (seconds; absent for no ``exp`` claim). The token verifies130 against the same config it was signed with (no new key material). Answers131 the error shape when no symmetric verifier is configured.132 """133 core = self.auth_core134 config = core.signing_jwt_config if core is not None else None135 if config is None:136 return NO_JWT_ERROR137 body = body_data or {}138 sub = body.get("sub")139 if not sub:140 return {"error": "Subject 'sub' is required"}141 claims: dict[str, Any] = {"sub": sub, "tags": body.get("tags") or []}142 expires_in = body.get("expires_in")143 if expires_in is not None:144 claims["exp"] = int(time.time()) + int(expires_in)145 token = jwt.encode(claims, config["secret"], algorithm=config["algorithm"])146 return {"token": token, "sub": sub}