tests/core/test_login_flow.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 """Password login surface tests (core 1d wave 1, Phase 4).16 17 The flow is driven through a full hand-built ``AsgiServer`` at the ASGI level18 (no uvicorn), the same driving style as ``test_session.py``: JSON POST to19 ``/_server/login`` verifies against a seeded in-memory ``UserStore`` and20 attaches the avatar to the request's session in place21 (``request.session.attach_avatar``) — the id never changes, so no login-time22 cookie is issued and handlers never touch cookies. The HTML page, the public23 ``login_methods``, ``logout``, the ``AuthMethod``/``AuthSection`` contract and24 the ``safe_next_path`` guard are covered alongside.25 """26 27 from __future__ import annotations28 29 import json30 import time31 from pathlib import Path32 from typing import Any33 34 import pytest35 from cryptography.fernet import Fernet36 37 from tests.storage_support import site_storage38 39 from genro_asgi import (40 AsgiServer,41 AuthMethod,42 AuthSection,43 BaseApplication,44 FileUserStore,45 PasswordMethod,46 ServerApplication,47 UserStore,48 )49 from genro_asgi.auth.auth_method import safe_next_path50 from genro_asgi.types import Message, Scope51 52 53 class MemoryUserStore(UserStore):54 """In-memory ``UserStore`` backend: the contract suite over a plain dict."""55 56 __slots__ = ("_records",)57 58 def __init__(self) -> None:59 self._records: dict[str, dict[str, Any]] = {}60 61 def load_all(self) -> list[dict[str, Any]]:62 return list(self._records.values())63 64 def get(self, identity: str) -> dict[str, Any] | None:65 return self._records.get(identity)66 67 def save(self, record: dict[str, Any]) -> None:68 self._records[record["identity"]] = record69 70 def delete(self, identity: str) -> bool:71 return self._records.pop(identity, None) is not None72 73 74 def make_server(with_users: bool = True) -> AsgiServer:75 """A full hand-built server; ``with_users`` seeds alice/wonder on a user store."""76 if not with_users:77 return AsgiServer(applications=[BaseApplication(mount="")])78 store = MemoryUserStore()79 store.save(80 {81 "identity": "alice",82 "password_hash": store.hash_password("wonder"),83 "tags": ["admin"],84 "enabled": True,85 }86 )87 store.save(88 {89 "identity": "mallory",90 "password_hash": store.hash_password("evil"),91 "tags": [],92 "enabled": False,93 }94 )95 return AsgiServer(applications=[BaseApplication(mount="")], users=store)96 97 98 async def drive(99 server: AsgiServer,100 path: str,101 method: str = "GET",102 cookie: str | None = None,103 body: dict[str, Any] | None = None,104 ) -> tuple[Scope, list[Message]]:105 """Drive one request through ``server`` at the ASGI level (JSON body when given)."""106 headers: list[tuple[bytes, bytes]] = []107 raw = b""108 if body is not None:109 headers.append((b"content-type", b"application/json"))110 raw = json.dumps(body).encode()111 if cookie is not None:112 headers.append((b"cookie", cookie.encode()))113 path, _, query = path.partition("?")114 scope: Scope = {115 "type": "http",116 "method": method,117 "path": path,118 "query_string": query.encode(),119 "headers": headers,120 }121 sent: list[Message] = []122 123 async def receive() -> Message:124 return {"type": "http.request", "body": raw, "more_body": False}125 126 async def send(message: Message) -> None:127 sent.append(message)128 129 await server(scope, receive, send)130 return scope, sent131 132 133 def response_status(sent: list[Message]) -> int:134 return next(m["status"] for m in sent if m["type"] == "http.response.start")135 136 137 def response_headers(sent: list[Message]) -> dict[bytes, bytes]:138 start = next(m for m in sent if m["type"] == "http.response.start")139 return {name: value for name, value in start["headers"] if name != b"set-cookie"}140 141 142 def response_body(sent: list[Message]) -> bytes:143 return b"".join(m.get("body", b"") for m in sent if m["type"] == "http.response.body")144 145 146 def json_body(sent: list[Message]) -> Any:147 return json.loads(response_body(sent))148 149 150 def set_cookie_value(sent: list[Message]) -> str | None:151 start = next(m for m in sent if m["type"] == "http.response.start")152 for name, value in start["headers"]:153 if name == b"set-cookie":154 return value.decode()155 return None156 157 158 def cookie_token(sent: list[Message]) -> str:159 cookie = set_cookie_value(sent)160 assert cookie is not None161 return cookie.split(";")[0].split("=", 1)[1]162 163 164 class Clock:165 """Controllable stand-in for ``time.time`` — drives the lockout window tests."""166 167 def __init__(self, now: float = 1_000_000.0) -> None:168 self.now = now169 170 def __call__(self) -> float:171 return self.now172 173 def advance(self, seconds: float) -> None:174 self.now += seconds175 176 177 def make_lockout_server(178 policy: dict[str, Any] | None = None,179 ) -> tuple[AsgiServer, MemoryUserStore]:180 """A server seeded with alice/wonder; ``policy`` rides the ``login()`` config lift."""181 store = MemoryUserStore()182 store.save(183 {184 "identity": "alice",185 "password_hash": store.hash_password("wonder"),186 "tags": ["admin"],187 "enabled": True,188 }189 )190 kwargs: dict[str, Any] = {"server_app": {"login": policy}} if policy else {}191 return AsgiServer(applications=[BaseApplication(mount="")], users=store, **kwargs), store192 193 194 async def login_attempt(195 server: AsgiServer, session_id: str, password: str, identity: str = "alice"196 ) -> Any:197 """One login POST riding ``session_id``; returns the JSON payload."""198 _, sent = await drive(199 server,200 "/_server/login",201 "POST",202 cookie=f"session_id={session_id}",203 body={"identity": identity, "password": password},204 )205 return json_body(sent)206 207 208 class TestLoginHappyPath:209 async def test_login_attaches_the_avatar_and_keeps_the_session_id(self) -> None:210 server = make_server()211 anonymous = server.session_store.create()212 anonymous.data["cart"] = "kept"213 _, sent = await drive(214 server,215 "/_server/login",216 "POST",217 cookie=f"session_id={anonymous.id}",218 body={"identity": "alice", "password": "wonder"},219 )220 assert response_status(sent) == 200221 payload = json_body(sent)222 assert payload["identity"] == "alice"223 assert payload["tags"] == ["admin"]224 assert payload["session_id"] == anonymous.id # the id never changes at login225 assert set_cookie_value(sent) is None # the client's cookie is still valid226 promoted = server.session_store.get(anonymous.id)227 assert promoted is anonymous228 assert promoted.avatar() is not None229 assert promoted.avatar().identity == "alice"230 assert promoted.avatar().tags == ["admin"]231 assert promoted.data["cart"] == "kept" # the cart survives the login232 233 async def test_login_green_path_against_a_file_user_store(self, tmp_path: Path) -> None:234 storage = site_storage(tmp_path)235 storage.set_encryption_keys(Fernet.generate_key().decode())236 store = FileUserStore(storage)237 store.save(238 {239 "identity": "alice",240 "password_hash": store.hash_password("wonder"),241 "tags": ["admin"],242 "enabled": True,243 }244 )245 server = AsgiServer(applications=[BaseApplication(mount="")], users=store)246 anonymous = server.session_store.create()247 _, sent = await drive(248 server,249 "/_server/login",250 "POST",251 cookie=f"session_id={anonymous.id}",252 body={"identity": "alice", "password": "wonder"},253 )254 assert response_status(sent) == 200255 payload = json_body(sent)256 assert payload["identity"] == "alice"257 assert payload["tags"] == ["admin"]258 259 async def test_login_accepts_the_pages_form_encoded_post(self) -> None:260 server = make_server()261 anonymous = server.session_store.create()262 raw = b"identity=alice&password=wonder"263 headers = [264 (b"content-type", b"application/x-www-form-urlencoded"),265 (b"cookie", f"session_id={anonymous.id}".encode()),266 ]267 scope: Scope = {268 "type": "http",269 "method": "POST",270 "path": "/_server/login",271 "query_string": b"",272 "headers": headers,273 }274 sent: list[Message] = []275 276 async def receive() -> Message:277 return {"type": "http.request", "body": raw, "more_body": False}278 279 async def send(message: Message) -> None:280 sent.append(message)281 282 await server(scope, receive, send)283 payload = json_body(sent)284 assert payload["identity"] == "alice"285 assert payload["session_id"] == anonymous.id # id kept, no cookie re-issue286 assert set_cookie_value(sent) is None287 288 async def test_login_on_first_contact_rides_the_new_session_cookie(self) -> None:289 server = make_server()290 _, sent = await drive(291 server,292 "/_server/login",293 "POST",294 body={"identity": "alice", "password": "wonder"},295 )296 payload = json_body(sent)297 assert cookie_token(sent) == payload["session_id"]298 promoted = server.session_store.get(payload["session_id"])299 assert promoted is not None and promoted.avatar() is not None300 301 302 class TestLoginFailures:303 async def test_invalid_credentials_answer_the_error_and_no_cookie(self) -> None:304 server = make_server()305 anonymous = server.session_store.create()306 _, sent = await drive(307 server,308 "/_server/login",309 "POST",310 cookie=f"session_id={anonymous.id}",311 body={"identity": "alice", "password": "nope"},312 )313 assert response_status(sent) == 200314 assert json_body(sent) == {"error": "Invalid credentials", "remaining_attempts": 4}315 assert set_cookie_value(sent) is None316 317 async def test_disabled_user_never_authenticates(self) -> None:318 server = make_server()319 anonymous = server.session_store.create()320 _, sent = await drive(321 server,322 "/_server/login",323 "POST",324 cookie=f"session_id={anonymous.id}",325 body={"identity": "mallory", "password": "evil"},326 )327 # disabled users have a record: their failures count and surface the counter328 assert json_body(sent) == {"error": "Invalid credentials", "remaining_attempts": 4}329 assert set_cookie_value(sent) is None330 331 async def test_missing_credentials_answer_the_error(self) -> None:332 server = make_server()333 anonymous = server.session_store.create()334 _, sent = await drive(335 server,336 "/_server/login",337 "POST",338 cookie=f"session_id={anonymous.id}",339 body={"identity": "alice"},340 )341 assert json_body(sent) == {"error": "Identity and password are required"}342 assert set_cookie_value(sent) is None343 344 async def test_server_without_a_user_store_answers_the_error(self) -> None:345 server = make_server(with_users=False)346 anonymous = server.session_store.create()347 _, sent = await drive(348 server,349 "/_server/login",350 "POST",351 cookie=f"session_id={anonymous.id}",352 body={"identity": "alice", "password": "wonder"},353 )354 assert json_body(sent) == {"error": "Login is not available"}355 assert set_cookie_value(sent) is None356 357 358 class TestLoginLockout:359 @pytest.fixture360 def clock(self, monkeypatch: pytest.MonkeyPatch) -> Clock:361 """Freeze ``time.time`` on a controllable clock."""362 frozen = Clock()363 monkeypatch.setattr(time, "time", frozen)364 return frozen365 366 async def test_max_attempts_failures_lock_even_the_correct_password(367 self, clock: Clock368 ) -> None:369 server, store = make_lockout_server()370 session = server.session_store.create()371 for expected_remaining in (4, 3, 2, 1, 0):372 payload = await login_attempt(server, session.id, "nope")373 assert payload == {374 "error": "Invalid credentials",375 "remaining_attempts": expected_remaining,376 }377 payload = await login_attempt(server, session.id, "wonder")378 assert payload == {"error": "Too many failed attempts"}379 assert store.get("alice")["failed_attempts"] == 5380 381 async def test_locked_attempt_does_not_touch_the_counter(self, clock: Clock) -> None:382 server, store = make_lockout_server({"max_attempts": 2, "backoff": 10})383 session = server.session_store.create()384 await login_attempt(server, session.id, "nope")385 await login_attempt(server, session.id, "nope")386 record = store.get("alice")387 assert record is not None388 locked_at = record["last_failed_at"]389 clock.advance(5) # still inside the 10s window390 payload = await login_attempt(server, session.id, "nope")391 assert payload == {"error": "Too many failed attempts"}392 assert record["failed_attempts"] == 2 # the refused attempt never counted393 assert record["last_failed_at"] == locked_at # nor extended the lock394 395 async def test_window_expiry_re_allows_and_success_resets(self, clock: Clock) -> None:396 server, store = make_lockout_server({"max_attempts": 2, "backoff": 10})397 session = server.session_store.create()398 await login_attempt(server, session.id, "nope")399 await login_attempt(server, session.id, "nope")400 clock.advance(11) # past the 10s window401 payload = await login_attempt(server, session.id, "wonder")402 assert payload["identity"] == "alice"403 assert payload["session_id"] == session.id404 assert store.get("alice")["failed_attempts"] == 0405 406 async def test_success_resets_the_counter(self, clock: Clock) -> None:407 server, store = make_lockout_server()408 session = server.session_store.create()409 await login_attempt(server, session.id, "nope")410 payload = await login_attempt(server, session.id, "nope")411 assert payload == {"error": "Invalid credentials", "remaining_attempts": 3}412 payload = await login_attempt(server, session.id, "wonder")413 assert payload["identity"] == "alice"414 assert store.get("alice")["failed_attempts"] == 0415 payload = await login_attempt(server, session.id, "nope")416 assert payload == {"error": "Invalid credentials", "remaining_attempts": 4}417 418 async def test_unknown_identity_has_no_counter(self, clock: Clock) -> None:419 server, store = make_lockout_server()420 session = server.session_store.create()421 payload = await login_attempt(server, session.id, "nope", identity="ghost")422 assert payload == {"error": "Invalid credentials"} # no remaining_attempts423 assert store.get("ghost") is None # no record is ever created424 425 async def test_policy_is_tunable_via_the_login_config(self, clock: Clock) -> None:426 server, store = make_lockout_server({"max_attempts": 2, "backoff": 10})427 session = server.session_store.create()428 await login_attempt(server, session.id, "nope")429 payload = await login_attempt(server, session.id, "nope")430 assert payload == {"error": "Invalid credentials", "remaining_attempts": 0}431 payload = await login_attempt(server, session.id, "wonder")432 assert payload == {"error": "Too many failed attempts"}433 434 async def test_backoff_grows_exponentially(self, clock: Clock) -> None:435 server, store = make_lockout_server({"max_attempts": 2, "backoff": 10})436 session = server.session_store.create()437 await login_attempt(server, session.id, "nope")438 await login_attempt(server, session.id, "nope") # locked: window 10 * 2**0439 clock.advance(11) # first window expired440 payload = await login_attempt(server, session.id, "nope")441 assert payload == {"error": "Invalid credentials", "remaining_attempts": 0}442 clock.advance(11) # the third failure doubled the window to 20s443 payload = await login_attempt(server, session.id, "wonder")444 assert payload == {"error": "Too many failed attempts"}445 clock.advance(10) # 21s past the third failure — window passed446 payload = await login_attempt(server, session.id, "wonder")447 assert payload["identity"] == "alice"448 449 450 class TestLoginPage:451 async def test_login_page_serves_the_descriptor_driven_html(self) -> None:452 server = make_server()453 _, sent = await drive(server, "/_server/login_page")454 assert response_status(sent) == 200455 assert response_headers(sent)[b"content-type"].startswith(b"text/html")456 page = response_body(sent).decode()457 assert "<title>Sign in</title>" in page458 assert "/_server/login_methods" in page459 460 async def test_login_page_binds_the_next_query_param(self) -> None:461 server = make_server()462 _, sent = await drive(server, "/_server/login_page?next=/app/page")463 assert response_status(sent) == 200464 465 466 class TestLoginMethods:467 async def test_login_methods_is_public_and_lists_the_password_descriptor(self) -> None:468 server = make_server(with_users=False)469 _, sent = await drive(server, "/_server/login_methods")470 assert response_status(sent) == 200471 assert json_body(sent) == {472 "methods": [473 {474 "id": "password",475 "kind": "form",476 "label": "Sign in",477 "action": "/_server/login",478 }479 ]480 }481 482 483 class TestLogout:484 async def test_logout_deletes_the_session(self) -> None:485 server = make_server(with_users=False)486 session = server.session_store.create()487 _, sent = await drive(488 server, "/_server/logout", "POST", body={"session_id": session.id}489 )490 assert json_body(sent) == {"status": "ok"}491 assert server.session_store.get(session.id) is None492 493 async def test_logout_without_a_session_id_still_answers_ok(self) -> None:494 server = make_server(with_users=False)495 _, sent = await drive(server, "/_server/logout", "POST", body={})496 assert json_body(sent) == {"status": "ok"}497 498 499 class TestAuthMethodContract:500 def test_password_method_owns_zero_routes(self) -> None:501 server = make_server(with_users=False)502 app = server.applications["_server"]503 assert isinstance(app, ServerApplication)504 assert app.auth_section is not None505 method = app.auth_section.methods["password"]506 assert isinstance(method, PasswordMethod)507 assert method.route.nodes() == {}508 509 async def test_password_method_is_never_attached_to_the_routing_tree(self) -> None:510 server = make_server(with_users=False)511 app = server.applications["_server"]512 assert isinstance(app, ServerApplication)513 assert app.auth_section is not None514 routers = app.auth_section.route.nodes(lazy=True).get("routers") or {}515 assert "password" not in routers516 _, sent = await drive(server, "/_server/auth/password/anything")517 assert response_status(sent) == 404518 519 def test_a_method_owning_routes_is_attached(self) -> None:520 from genro_routes import route521 522 class RoutedMethod(AuthMethod):523 kind = "redirect"524 525 @route(media_type="application/json")526 def start(self) -> dict[str, str]:527 """Entry route of the redirect method."""528 return {"ok": "start"}529 530 server = make_server(with_users=False)531 app = server.applications["_server"]532 assert isinstance(app, ServerApplication)533 assert app.auth_section is not None534 app.register_auth_method(RoutedMethod(app, "routed"))535 routers = app.auth_section.route.nodes(lazy=True).get("routers") or {}536 assert "routed" in routers537 538 def test_password_method_is_registered_under_the_auth_section(self) -> None:539 server = make_server(with_users=False)540 app = server.applications["_server"]541 assert isinstance(app, ServerApplication)542 assert isinstance(app.auth_section, AuthSection)543 assert app.sections["auth"] is app.auth_section544 assert list(app.auth_section.methods) == ["password"]545 method = app.auth_section.methods["password"]546 assert method.application is app547 assert method.server is server548 549 def test_auth_section_carries_the_server_and_describes_its_methods(self) -> None:550 server = make_server(with_users=False)551 app = server.applications["_server"]552 assert isinstance(app, ServerApplication)553 assert isinstance(app.auth_section, AuthSection)554 assert app.auth_section.server is server555 # A section with nothing registered describes nothing...556 assert AuthSection(app).descriptors() == []557 # ...and the app's own section describes exactly its registered methods.558 method = app.auth_section.methods["password"]559 assert app.auth_section.descriptors() == [method.descriptor()]560 561 def test_duplicate_method_id_is_rejected(self) -> None:562 server = make_server(with_users=False)563 app = server.applications["_server"]564 assert isinstance(app, ServerApplication)565 with pytest.raises(ValueError, match="already registered"):566 app.register_auth_method(PasswordMethod(app, "password"))567 568 569 class TestSafeNextPath:570 @pytest.mark.parametrize(571 "value",572 ["/app/page", "/", "/a?b=c"],573 )574 def test_same_origin_relative_paths_pass(self, value: str) -> None:575 assert safe_next_path(value) == value576 577 @pytest.mark.parametrize(578 "value",579 [None, "", "app/page", "//evil.example", "https://evil.example", "/a\\b", "javascript:x"],580 )581 def test_unsafe_values_collapse_to_the_default(self, value: str | None) -> None:582 assert safe_next_path(value) == "/"583 assert safe_next_path(value, default="/home") == "/home"