src/genro_asgi/auth/core.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 """Auth core: credential configuration and header verification (salvage Q5).16 17 ``AuthCore`` holds the whole credential configuration and the verification18 logic for the ``basic``, ``bearer`` and ``jwt`` schemes. It is router-free —19 configured per instance from ``{'basic': ..., 'bearer': ..., 'jwt': [...]}``20 and consumed by the server-side ``AuthMixin`` (never by walking wrappers).21 22 Config sections (``_configure_<type>`` dynamic dispatch, unknown sections23 ignored)::24 25 basic: {username: {password: "...", tags: "..."}} — O(1) lookup keyed26 by the base64 username:password the Basic header carries.27 bearer: {name: {token: "...", tags: "..."}} — O(1) lookup keyed by the28 token value.29 jwt: [{secret|public_key, algorithm, tags, name}, ...] — a list of30 verifier configs, each tried in turn and decoded with pyjwt.31 32 ``authenticate(scope)`` reads the ``Authorization`` header via the shared33 ``headers_dict`` cache, dispatches ``_auth_<scheme>`` and returns an ``Avatar``34 (the ONE identity type, unified with sessions) or ``None`` when NO credentials35 are presented. Credentials PRESENT but invalid raise ``HTTPUnauthorized`` —36 never a silent fallback; a malformed header (no scheme/value separator) is37 present-but-invalid, so it raises too. Tag normalization (e.g. a JWT null38 ``tags`` claim) happens at the ``Avatar`` boundary, never ``list(None)``.39 """40 41 from __future__ import annotations42 43 import base6444 from typing import TYPE_CHECKING, Any45 46 import jwt47 48 from ..exceptions import HTTPUnauthorized49 from ..middleware.base import headers_dict50 from ..session.avatar import Avatar51 from .api_key_store import API_KEY_PREFIX52 53 if TYPE_CHECKING:54 from ..types import Scope55 from .api_key_store import ApiKeyStore56 57 __all__ = ["AuthCore"]58 59 # HMAC algorithms share one secret for sign and verify, so a verifier using them60 # can also MINT tokens (the tokens surface's create_jwt); asymmetric verifiers61 # (RS*/ES*, a public_key) can only verify.62 SYMMETRIC_JWT_ALGORITHMS = frozenset({"HS256", "HS384", "HS512"})63 64 65 def _split_and_strip(value: str | list[str] | None) -> list[str]:66 """Split a comma-separated string into a stripped list; pass a list through."""67 if value is None:68 return []69 if isinstance(value, str):70 return [item.strip() for item in value.split(",")]71 return list(value)72 73 74 def _basic_auth_key(username: str, password: str) -> str:75 """Base64-encode ``username:password`` as the HTTP Basic header carries it."""76 return base64.b64encode(f"{username}:{password}".encode()).decode()77 78 79 class AuthCore:80 """Per-instance credential store and header verifier for basic/bearer/jwt."""81 82 def __init__(self, api_key_store: ApiKeyStore | None = None, **entries: Any) -> None:83 """Build the credential store from the configured sections.84 85 ``api_key_store`` is the wired registry the ``AuthMixin`` passes: when86 present, an inbound ``gak_`` bearer token is verified against it before87 the static-bearer/JWT chain.88 """89 self._api_key_store = api_key_store90 self._basic: dict[str, dict[str, Any]] = {}91 self._bearer: dict[str, dict[str, Any]] = {}92 self._jwt: list[dict[str, Any]] = []93 for auth_type, credentials in entries.items():94 configure = getattr(self, f"_configure_{auth_type}", self._configure_default)95 configure(credentials)96 97 def _configure_basic(self, credentials: dict[str, Any]) -> None:98 """Index Basic credentials by their base64 ``username:password`` key."""99 for username, config in credentials.items():100 password = config.get("password")101 if not password:102 raise ValueError(f"Basic auth user '{username}' missing 'password'")103 self._basic[_basic_auth_key(username, password)] = {104 "identity": username,105 "tags": _split_and_strip(config.get("tags")),106 "backend": "basic",107 }108 109 def _configure_bearer(self, credentials: dict[str, Any]) -> None:110 """Index Bearer credentials by their token value."""111 for name, config in credentials.items():112 token = config.get("token")113 if not token:114 raise ValueError(f"Bearer token '{name}' missing 'token' value")115 self._bearer[token] = {116 "identity": name,117 "tags": _split_and_strip(config.get("tags")),118 "backend": "bearer",119 }120 121 def _configure_jwt(self, credentials: list[dict[str, Any]]) -> None:122 """Store the JWT verifier configs as an ordered list of secrets/algorithms.123 124 ``signing`` records the key's provenance: only a key configured as125 ``secret`` (shared HMAC material) can SIGN — a ``public_key`` folded126 into the same slot verifies but must never mint tokens, whatever the127 (defaulted) algorithm says.128 """129 for config in credentials:130 self._jwt.append(131 {132 "name": config.get("name"),133 "secret": config.get("secret") or config.get("public_key"),134 "algorithm": config.get("algorithm", "HS256"),135 "tags": _split_and_strip(config.get("tags")),136 "signing": bool(config.get("secret")),137 }138 )139 140 def _configure_default(self, credentials: Any) -> None:141 """Unknown config section — ignored (salvage semantics)."""142 143 @property144 def signing_jwt_config(self) -> dict[str, Any] | None:145 """The first symmetric (``HS*``) JWT verifier config, usable for signing.146 147 A read-only seam for the tokens surface: HMAC verifiers share one secret148 for sign and verify, so a token minted here verifies against the same149 config with zero new key material. Asymmetric verifiers (``RS*``/``ES*``,150 a ``public_key``) can only verify and are skipped — the ``signing``151 provenance flag guards them even when the algorithm was defaulted, so152 public key material can never mint tokens. ``None`` when no symmetric153 verifier is configured.154 """155 for config in self._jwt:156 if config["algorithm"] in SYMMETRIC_JWT_ALGORITHMS and config["signing"]:157 return config158 return None159 160 def authenticate(self, scope: Scope) -> Avatar | None:161 """Verify the request's credentials; return an ``Avatar`` or ``None``.162 163 ``None`` when no ``Authorization`` header is presented; a present but164 invalid credential — including a malformed header with no scheme/value165 separator — raises ``HTTPUnauthorized`` (no silent fallback) carrying a166 ``WWW-Authenticate: Bearer`` challenge header.167 """168 header = headers_dict(scope).get("authorization")169 if not header:170 return None171 challenge = [(b"www-authenticate", b"Bearer")]172 if " " not in header:173 raise HTTPUnauthorized("Malformed Authorization header", headers=challenge)174 scheme, credentials = header.split(" ", 1)175 verify = getattr(self, f"_auth_{scheme.lower()}", self._auth_default)176 result = verify(credentials)177 if result is None:178 raise HTTPUnauthorized("Invalid or expired credentials", headers=challenge)179 return Avatar(result["identity"], result["tags"])180 181 def _auth_basic(self, credentials: str) -> dict[str, Any] | None:182 """Look up Basic credentials by their raw base64 header value."""183 return self._basic.get(credentials)184 185 def _auth_bearer(self, credentials: str) -> dict[str, Any] | None:186 """Resolve a Bearer token: a ``gak_`` api key first, then static/JWT.187 188 A ``gak_``-prefixed token is unambiguously an api key: it is verified189 against the registry and NEVER falls through to the JWT chain — a miss190 (revoked, expired, unknown, or no store wired) is a hard ``None``.191 Any other token keeps the static-bearer → JWT chain unchanged.192 """193 if credentials.startswith(API_KEY_PREFIX):194 return self._auth_api_key(credentials)195 entry = self._bearer.get(credentials)196 if entry is not None:197 return entry198 return self._auth_jwt(credentials)199 200 def _auth_api_key(self, credentials: str) -> dict[str, Any] | None:201 """Verify a ``gak_`` key against the wired registry (identity = label)."""202 if self._api_key_store is None:203 return None204 record = self._api_key_store.verify(credentials)205 if record is None:206 return None207 return {208 "identity": record["label"],209 "tags": record["tags"],210 "backend": "api_key",211 }212 213 def _auth_jwt(self, credentials: str) -> dict[str, Any] | None:214 """Try each configured JWT verifier in turn."""215 for config in self._jwt:216 result = self._verify_jwt(credentials, config)217 if result is not None:218 return result219 return None220 221 def _auth_default(self, credentials: str) -> dict[str, Any] | None:222 """Unknown scheme — always rejected."""223 return None224 225 def _verify_jwt(self, credentials: str, config: dict[str, Any]) -> dict[str, Any] | None:226 """Decode a JWT with one verifier config; return the identity/tags or ``None``."""227 secret = config.get("secret")228 if not secret:229 return None230 name = config.get("name")231 try:232 payload = jwt.decode(credentials, secret, algorithms=[config["algorithm"]])233 except jwt.InvalidTokenError:234 return None235 return {236 "identity": payload.get("sub"),237 "tags": payload.get("tags", config["tags"]),238 "backend": f"jwt:{name}" if name else "jwt",239 }