tests/core/test_auth.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 tests (Macro 2 Phases 5+9): AuthCore verification + §5.5 identity precedence.16 17 Header verification (basic/bearer/jwt, wrong credentials → 401) is driven at18 the ASGI level through an ``AuthMixin/MiddlewareMixin/BaseServer`` composition,19 the same driving style as ``test_session.py``. The §5.5 precedence — header20 identity wins over the session, an invalid header is a 401 with no fallback —21 is exercised both by calling ``server.authenticate(scope)`` at app-dispatch22 time and END-TO-END through the REAL combined chain (session middleware order23 400 OUTSIDE auth 450) on the full24 ``AuthMixin/SessionMixin/MiddlewareMixin/BaseServer`` composition.25 """26 27 from __future__ import annotations28 29 import base6430 from pathlib import Path31 32 import jwt33 import pytest34 from cryptography.fernet import Fernet35 36 from tests.storage_support import site_storage37 38 from genro_asgi import (39 ApiKeyStore,40 AsgiServer,41 AuthCore,42 AuthMixin,43 Avatar,44 BaseApplication,45 BaseServer,46 FileApiKeyStore,47 FileUserStore,48 Session,49 SessionMixin,50 UserStore,51 )52 from genro_asgi.exceptions import HTTPUnauthorized53 from genro_asgi.middleware import MiddlewareMixin54 from genro_asgi.types import Message, Receive, Scope, Send55 56 57 class MemoryUserStore(UserStore):58 """In-memory ``UserStore`` backend used as a ready store instance in tests."""59 60 __slots__ = ("_records",)61 62 def __init__(self) -> None:63 self._records: dict[str, dict] = {}64 65 def load_all(self) -> list[dict]:66 return list(self._records.values())67 68 def get(self, identity: str) -> dict | None:69 return self._records.get(identity)70 71 def save(self, record: dict) -> None:72 self._records[record["identity"]] = record73 74 def delete(self, identity: str) -> bool:75 return self._records.pop(identity, None) is not None76 77 78 class MemoryApiKeyStore(ApiKeyStore):79 """In-memory ``ApiKeyStore`` backend: the shared issue/verify logic over a dict."""80 81 __slots__ = ("_records",)82 83 def __init__(self) -> None:84 self._records: dict[str, dict] = {}85 86 def load_all(self) -> list[dict]:87 return list(self._records.values())88 89 def get(self, key_id: str) -> dict | None:90 return self._records.get(key_id)91 92 def save(self, record: dict) -> None:93 self._records[record["key_id"]] = record94 95 def delete(self, key_id: str) -> bool:96 return self._records.pop(key_id, None) is not None97 98 99 def basic_header(username: str, password: str) -> str:100 """The value of a Basic ``Authorization`` header for these credentials."""101 raw = base64.b64encode(f"{username}:{password}".encode()).decode()102 return f"Basic {raw}"103 104 105 AUTH_CONFIG = {106 "basic": {"alice": {"password": "wonderland", "tags": "admin,ops"}},107 "bearer": {"svc": {"token": "sk_live_xyz", "tags": "api"}},108 "jwt": [{"secret": "topsecret", "algorithm": "HS256"}],109 }110 111 112 # --- AuthCore unit ---113 114 115 class TestAuthCore:116 def test_no_header_returns_none(self) -> None:117 core = AuthCore(**AUTH_CONFIG)118 assert core.authenticate({"headers": []}) is None119 120 def test_basic_ok_returns_avatar(self) -> None:121 core = AuthCore(**AUTH_CONFIG)122 scope: Scope = {"headers": [(b"authorization", basic_header("alice", "wonderland").encode())]}123 avatar = core.authenticate(scope)124 assert avatar is not None125 assert avatar.identity == "alice"126 assert avatar.tags == ["admin", "ops"]127 128 def test_bearer_ok_returns_avatar(self) -> None:129 core = AuthCore(**AUTH_CONFIG)130 scope: Scope = {"headers": [(b"authorization", b"Bearer sk_live_xyz")]}131 avatar = core.authenticate(scope)132 assert avatar is not None and avatar.identity == "svc"133 134 def test_jwt_ok_returns_avatar(self) -> None:135 core = AuthCore(**AUTH_CONFIG)136 token = jwt.encode({"sub": "carol", "tags": ["reader"]}, "topsecret", algorithm="HS256")137 scope: Scope = {"headers": [(b"authorization", f"Bearer {token}".encode())]}138 avatar = core.authenticate(scope)139 assert avatar is not None140 assert avatar.identity == "carol"141 assert avatar.tags == ["reader"]142 143 def test_signing_config_is_the_symmetric_secret(self) -> None:144 core = AuthCore(**AUTH_CONFIG)145 signing = core.signing_jwt_config146 assert signing is not None147 assert signing["secret"] == "topsecret"148 149 def test_public_key_never_becomes_the_signing_config(self) -> None:150 # A verifier configured with public_key only (algorithm defaulted):151 # it can verify, but public key material must never mint tokens.152 core = AuthCore(jwt=[{"public_key": "not-a-secret"}])153 assert core.signing_jwt_config is None154 155 def test_wrong_password_raises_401(self) -> None:156 core = AuthCore(**AUTH_CONFIG)157 scope: Scope = {"headers": [(b"authorization", basic_header("alice", "nope").encode())]}158 with pytest.raises(HTTPUnauthorized) as excinfo:159 core.authenticate(scope)160 assert excinfo.value.status == 401161 assert (b"www-authenticate", b"Bearer") in excinfo.value.headers162 163 def test_malformed_header_raises_401_with_challenge(self) -> None:164 core = AuthCore(**AUTH_CONFIG)165 scope: Scope = {"headers": [(b"authorization", b"Basicabc123")]}166 with pytest.raises(HTTPUnauthorized) as excinfo:167 core.authenticate(scope)168 assert (b"www-authenticate", b"Bearer") in excinfo.value.headers169 170 def test_unknown_scheme_raises_401(self) -> None:171 core = AuthCore(**AUTH_CONFIG)172 scope: Scope = {"headers": [(b"authorization", b"Weird abc")]}173 with pytest.raises(HTTPUnauthorized):174 core.authenticate(scope)175 176 def test_missing_basic_password_config_raises(self) -> None:177 with pytest.raises(ValueError, match="missing 'password'"):178 AuthCore(basic={"bob": {"tags": "user"}})179 180 def test_missing_bearer_token_config_raises(self) -> None:181 with pytest.raises(ValueError, match="missing 'token'"):182 AuthCore(bearer={"svc": {"tags": "api"}})183 184 185 # --- ASGI-level header authentication ---186 187 188 class HeaderAuthServer(AuthMixin, MiddlewareMixin, BaseServer):189 """Header-only auth composition: no session capability."""190 191 192 class EchoAuthApp(BaseApplication):193 """Echoes the identity of the avatar published on ``scope["auth"]``."""194 195 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:196 auth = scope.get("auth")197 body = auth.identity.encode() if auth is not None else b"anonymous"198 await send({"type": "http.response.start", "status": 200, "headers": []})199 await send({"type": "http.response.body", "body": body})200 201 202 async def http_get(203 server: BaseServer, authorization: str | None = None, cookie: str | None = None204 ) -> tuple[Scope, list[Message]]:205 """Drive one GET through ``server`` at the ASGI level; return the scope and what it sent."""206 headers: list[tuple[bytes, bytes]] = []207 if authorization is not None:208 headers.append((b"authorization", authorization.encode()))209 if cookie is not None:210 headers.append((b"cookie", cookie.encode()))211 scope: Scope = {"type": "http", "method": "GET", "path": "/", "headers": headers}212 sent: list[Message] = []213 214 async def receive() -> Message:215 return {"type": "http.request"}216 217 async def send(message: Message) -> None:218 sent.append(message)219 220 await server(scope, receive, send)221 return scope, sent222 223 224 def response_status(sent: list[Message]) -> int:225 start = next(m for m in sent if m["type"] == "http.response.start")226 return start["status"]227 228 229 def response_body(sent: list[Message]) -> bytes:230 return b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")231 232 233 def set_cookie_value(sent: list[Message]) -> str | None:234 start = next(m for m in sent if m["type"] == "http.response.start")235 for name, value in start["headers"]:236 if name == b"set-cookie":237 return value.decode()238 return None239 240 241 def header_value(sent: list[Message], header: bytes) -> bytes | None:242 start = next(m for m in sent if m["type"] == "http.response.start")243 for name, value in start["headers"]:244 if name == header:245 return value246 return None247 248 249 class TestHeaderAuthFlow:250 async def test_basic_ok(self) -> None:251 server = HeaderAuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)252 scope, sent = await http_get(server, basic_header("alice", "wonderland"))253 assert response_status(sent) == 200254 assert response_body(sent) == b"alice"255 assert scope["auth"].identity == "alice"256 257 async def test_bearer_ok(self) -> None:258 server = HeaderAuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)259 _, sent = await http_get(server, "Bearer sk_live_xyz")260 assert response_body(sent) == b"svc"261 262 async def test_jwt_ok(self) -> None:263 server = HeaderAuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)264 token = jwt.encode({"sub": "carol"}, "topsecret", algorithm="HS256")265 _, sent = await http_get(server, f"Bearer {token}")266 assert response_body(sent) == b"carol"267 268 async def test_wrong_password_yields_401(self) -> None:269 server = HeaderAuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)270 _, sent = await http_get(server, basic_header("alice", "nope"))271 assert response_status(sent) == 401272 273 async def test_invalid_credentials_401_carries_www_authenticate(self) -> None:274 server = HeaderAuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)275 _, sent = await http_get(server, basic_header("alice", "nope"))276 assert response_status(sent) == 401277 assert header_value(sent, b"www-authenticate") == b"Bearer"278 279 async def test_malformed_credentials_401_carries_www_authenticate(self) -> None:280 server = HeaderAuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)281 _, sent = await http_get(server, "Basicabc123")282 assert response_status(sent) == 401283 assert header_value(sent, b"www-authenticate") == b"Bearer"284 285 async def test_no_header_is_anonymous(self) -> None:286 server = HeaderAuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)287 scope, sent = await http_get(server)288 assert scope["auth"] is None289 assert response_body(sent) == b"anonymous"290 291 async def test_explicit_auth_false_disarms_the_middleware(self) -> None:292 server = HeaderAuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG, middleware={"auth": False})293 scope, _ = await http_get(server, basic_header("alice", "wonderland"))294 assert "auth" not in scope295 296 297 # --- §5.5 identity precedence (server.authenticate at app-dispatch time) ---298 299 300 class AuthServer(AuthMixin, SessionMixin, MiddlewareMixin, BaseServer):301 """The shipped-shape composition: header auth over sessions and the chain."""302 303 304 def scope_with(session: Session | None, authorization: str | None) -> Scope:305 """A scope carrying an attached session and/or an Authorization header."""306 headers = [(b"authorization", authorization.encode())] if authorization is not None else []307 scope: Scope = {"type": "http", "method": "GET", "path": "/", "headers": headers}308 if session is not None:309 scope["session"] = session310 return scope311 312 313 class TestIdentityPrecedence:314 def test_header_wins_over_session(self) -> None:315 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)316 session = Session("sid", avatar=Avatar("sessionuser"), ttl=3600)317 scope = scope_with(session, basic_header("alice", "wonderland"))318 avatar = server.authenticate(scope)319 assert avatar.identity == "alice"320 321 def test_invalid_header_is_401_no_session_fallback(self) -> None:322 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)323 session = Session("sid", avatar=Avatar("sessionuser"), ttl=3600)324 scope = scope_with(session, basic_header("alice", "nope"))325 with pytest.raises(HTTPUnauthorized):326 server.authenticate(scope)327 328 def test_no_header_falls_back_to_session(self) -> None:329 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)330 session = Session("sid", avatar=Avatar("sessionuser", ["member"]), ttl=3600)331 scope = scope_with(session, None)332 avatar = server.authenticate(scope)333 assert avatar.identity == "sessionuser"334 assert avatar.tags == ["member"]335 336 def test_no_header_anonymous_session_is_none(self) -> None:337 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)338 session = Session("sid", avatar=None, ttl=3600)339 assert server.authenticate(scope_with(session, None)) is None340 341 def test_no_header_no_session_is_none(self) -> None:342 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)343 assert server.authenticate(scope_with(None, None)) is None344 345 346 # --- end-to-end through the REAL combined chain (Phase 9: B1/B2/B3) ---347 348 349 class TestCombinedChainFlow:350 async def test_header_auth_without_cookie_gets_anonymous_session(self) -> None:351 # B1: header identity on the scope, anonymous session, Set-Cookie, no 500.352 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)353 scope, sent = await http_get(server, basic_header("alice", "wonderland"))354 assert response_status(sent) == 200355 assert scope["auth"].identity == "alice"356 assert set_cookie_value(sent) is not None357 assert scope["session"].avatar() is None358 359 async def test_session_identity_flows_through_the_chain(self) -> None:360 # B2: cookie only — the §5.5 session fallback works through the chain.361 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)362 session = server.session_store.create(avatar=Avatar("sessionuser", ["member"]))363 scope, sent = await http_get(server, cookie=f"session_id={session.id}")364 assert response_status(sent) == 200365 assert response_body(sent) == b"sessionuser"366 assert scope["auth"].identity == "sessionuser"367 368 async def test_header_wins_over_session_through_the_chain(self) -> None:369 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)370 session = server.session_store.create(avatar=Avatar("sessionuser"))371 _, sent = await http_get(372 server, basic_header("alice", "wonderland"), cookie=f"session_id={session.id}"373 )374 assert response_body(sent) == b"alice"375 376 async def test_jwt_null_tags_claim_yields_empty_tags(self) -> None:377 # B3: a validly signed token with "tags": null normalizes to empty tags.378 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)379 token = jwt.encode({"sub": "carol", "tags": None}, "topsecret", algorithm="HS256")380 scope, sent = await http_get(server, f"Bearer {token}")381 assert response_status(sent) == 200382 assert scope["auth"].tags == []383 384 async def test_malformed_header_is_401(self) -> None:385 server = AuthServer(applications=[EchoAuthApp(mount="")], auth=AUTH_CONFIG)386 _, sent = await http_get(server, "Basicabc123")387 assert response_status(sent) == 401388 389 390 # --- composition without the auth capability ---391 392 393 class TestWithoutAuthMixin:394 def test_base_server_authenticate_is_none(self) -> None:395 server = BaseServer(applications=[EchoAuthApp(mount="")])396 assert server.authenticate({"headers": []}) is None397 398 def test_session_only_composition_authenticate_is_none(self) -> None:399 class SessionOnly(SessionMixin, MiddlewareMixin, BaseServer):400 pass401 402 server = SessionOnly(applications=[EchoAuthApp(mount="")])403 assert server.authenticate({"headers": []}) is None404 405 406 def encrypted_server(tmp_path, **kwargs) -> AsgiServer:407 """An ``AsgiServer`` whose site storage carries key material, for store wiring."""408 storage = site_storage(tmp_path, storage_key=Fernet.generate_key().decode())409 return AsgiServer(applications=[BaseApplication(mount="")], storage=storage, **kwargs)410 411 412 class TestStoreWiring:413 def test_stores_are_none_when_unconfigured(self) -> None:414 server = AsgiServer(applications=[BaseApplication(mount="")])415 assert server.user_store is None416 assert server.api_key_store is None417 418 def test_users_config_dict_builds_a_file_store(self, tmp_path: Path) -> None:419 server = encrypted_server(tmp_path, users={})420 assert isinstance(server.user_store, FileUserStore)421 assert server.api_key_store is None422 423 def test_tokens_config_dict_builds_a_file_store(self, tmp_path: Path) -> None:424 server = encrypted_server(tmp_path, tokens={})425 assert isinstance(server.api_key_store, FileApiKeyStore)426 427 def test_config_dict_honours_mount_and_prefix(self, tmp_path: Path) -> None:428 server = encrypted_server(tmp_path, users={"prefix": "people"})429 server.user_store.save(430 {"identity": "bob", "password_hash": "x", "tags": [], "enabled": True}431 )432 node = server.storage.node("site:people/bob.json")433 assert node.exists()434 435 def test_ready_instance_is_passed_through(self) -> None:436 store = MemoryUserStore()437 server = AsgiServer(applications=[BaseApplication(mount="")], users=store)438 assert server.user_store is store439 440 def test_config_dict_without_storage_is_a_boot_error(self) -> None:441 class NoStorageServer(AuthMixin, MiddlewareMixin, BaseServer):442 pass443 444 with pytest.raises(RuntimeError, match="storage"):445 NoStorageServer(applications=[BaseApplication(mount="")], users={})446 447 448 class TestBootstrapAdmin:449 def test_admin_password_seeds_the_superadmin(self) -> None:450 store = MemoryUserStore()451 server = AsgiServer(applications=[BaseApplication(mount="")], users=store, admin_password="pw")452 record = server.user_store.get("admin")453 assert record is not None454 # administration AND observation: the identity that configures the455 # server also reaches its monitor on a fresh install456 assert record["tags"] == ["SUPERADMIN", "SERVER_ADMIN"]457 assert record["enabled"] is True458 assert server.user_store.verify("admin", "pw") is not None459 460 def test_admin_password_without_users_implies_the_default_store(self, tmp_path: Path) -> None:461 server = encrypted_server(tmp_path, admin_password="pw")462 assert isinstance(server.user_store, FileUserStore)463 assert server.user_store.verify("admin", "pw") is not None464 465 def test_bootstrap_is_an_upsert_config_wins(self) -> None:466 store = MemoryUserStore()467 store.save(468 {"identity": "admin", "password_hash": "stale", "tags": [], "enabled": False}469 )470 server = AsgiServer(applications=[BaseApplication(mount="")], users=store, admin_password="fresh")471 record = server.user_store.get("admin")472 assert record["enabled"] is True473 assert record["tags"] == ["SUPERADMIN", "SERVER_ADMIN"]474 assert server.user_store.verify("admin", "fresh") is not None475 476 477 def bearer_scope(token: str) -> Scope:478 """A scope carrying ``Authorization: Bearer <token>``."""479 return {"headers": [(b"authorization", f"Bearer {token}".encode())]}480 481 482 class TestApiKeyBearer:483 def _core_with_key(self, tags: list[str]) -> tuple[AuthCore, str, MemoryApiKeyStore]:484 store = MemoryApiKeyStore()485 key = store.issue("ci-bot", tags)486 return AuthCore(api_key_store=store), key, store487 488 def test_valid_gak_key_authenticates_with_label_identity(self) -> None:489 core, key, _ = self._core_with_key(["ci", "deploy"])490 avatar = core.authenticate(bearer_scope(key))491 assert avatar is not None492 assert avatar.identity == "ci-bot" # identity is the key label493 assert avatar.tags == ["ci", "deploy"]494 495 def test_revoked_key_is_unauthorized_not_a_jwt_fallthrough(self) -> None:496 core, key, store = self._core_with_key(["ci"])497 key_id = key[len("gak_") :].partition("_")[0]498 store.revoke(key_id)499 with pytest.raises(HTTPUnauthorized):500 core.authenticate(bearer_scope(key))501 502 def test_unknown_gak_key_is_unauthorized(self) -> None:503 core, _, _ = self._core_with_key(["ci"])504 with pytest.raises(HTTPUnauthorized):505 core.authenticate(bearer_scope("gak_deadbeef_nonexistentsecret"))506 507 def test_gak_key_with_no_store_is_unauthorized(self) -> None:508 core = AuthCore() # no api_key_store wired509 with pytest.raises(HTTPUnauthorized):510 core.authenticate(bearer_scope("gak_deadbeef_secret"))511 512 def test_non_gak_bearer_still_uses_the_static_chain(self) -> None:513 store = MemoryApiKeyStore()514 core = AuthCore(515 bearer={"svc": {"token": "sk_live_xyz", "tags": "service"}},516 api_key_store=store,517 )518 avatar = core.authenticate(bearer_scope("sk_live_xyz"))519 assert avatar is not None520 assert avatar.identity == "svc"521 assert avatar.tags == ["service"]522 523 def test_end_to_end_gak_key_authenticates_through_the_server(self) -> None:524 store = MemoryApiKeyStore()525 server = AsgiServer(applications=[BaseApplication(mount="")], tokens=store)526 key = store.issue("robot", ["worker"])527 avatar = server.auth_core.authenticate(bearer_scope(key))528 assert avatar is not None529 assert avatar.identity == "robot"530 assert avatar.tags == ["worker"]