t

"""The access policy walked case by case, then through the middleware on the real app. The pure `decide()` with every class, the `next` cases the SPA shares, the must-change gate; then members-mode answers (302, 401, robots, machine 404s, docs off, X-Robots-Tag), unchanged under any Accept header; the traversal probes through the middleware; the origin requirement on every write; an act-mode write producing exactly one `act_as.write` row naming both ids and a read-only View As write producing none. Until 10 Sep the middleware covered writes only and read every page open; the plan (accounts-2026-09 §4.7) closes reads too. In-process on SQLite for the session-backed cases, a raw ASGI scope for the probes. """ import asyncio import json import pathlib import pytest from sqlalchemy import select from app import main from app.config import settings from app.models import Account, AccountLevel, AuditLog, DiscussionComment from app.services import access, accounts from tests import _accounts as T from tests.kit import _env MAIN = pathlib.Path(__file__).resolve().parents[1] SHELL = "t

" ADMITTED = None def dec(**kw): base = dict(method="GET", key="GET /api/products", page=None, site_open=False, signed_in=False, active=False, must_change=False, holds=lambda p: False, read_only=False, origin_ok=True, wants_html=False) base.update(kw) return access.decide(**base) class TestThePolicyAsAFunction: def test_an_unclassified_route_is_refused_before_anything_else(self): assert dec(key="GET /api/new-thing", signed_in=True, active=True, holds=lambda p: True).code == "ROUTE_NOT_CLASSIFIED" assert dec(key=None).code == "ROUTE_NOT_CLASSIFIED" def test_public_always_passes_anonymously_in_every_mode(self): for key in ("GET /api/health", "GET /api/bw/me", "POST /api/auth/login", "GET /robots.txt", "MOUNT /assets"): assert dec(key=key, method=key.split()[0]) is ADMITTED, key assert dec(key=access.CATCH_ALL, page="/login", wants_html=True) is ADMITTED def test_storefront_reads_need_a_session_while_members_only(self): assert dec(key="GET /api/products").code == "NOT_SIGNED_IN" assert dec(key="GET /api/products").status == 401 assert dec(key="GET /api/products", site_open=True) is ADMITTED assert dec(key="GET /api/products", signed_in=True, active=True) is ADMITTED assert dec(key="GET /api/products", signed_in=True, active=False).code == "ACCOUNT_DISABLED" def test_anonymous_html_is_one_redirect_known_route_or_not(self): v = dec(key=access.CATCH_ALL, page="/products", wants_html=True) assert (v.status, v.location) == (302, "/login?next=/products") v = dec(key=access.CATCH_ALL, page="/", wants_html=True) assert (v.status, v.location) == (302, "/login") v = dec(key=access.CATCH_ALL, page="/no-such-page", wants_html=True) assert (v.status, v.location) == (302, "/login?next=/no-such-page") v = dec(key="GET /products/{slug}", page=None, wants_html=True) assert v.status == 302 # Open: the storefront passes and the not-found shell is the route's own 404. assert dec(key=access.CATCH_ALL, page="/no-such-page", wants_html=True, site_open=True) is ADMITTED def test_missing_files_and_docs_are_plain_404s_in_every_mode(self): for page in ("/missing.xml", "/openapi.json", "/docs", "/redoc"): v = dec(key=access.CATCH_ALL, page=page, wants_html=True) assert v.status == 404, page v = dec(key=access.CATCH_ALL, page=page, wants_html=True, site_open=True, signed_in=True, active=True) assert v.status == 404, page for key in sorted(access.MACHINE_FILES): assert dec(key=key).status == 404 assert dec(key=key, signed_in=True, active=True).status == 404 assert dec(key=key, site_open=True) is ADMITTED def test_member_and_permission_classes(self): assert dec(key="POST /api/auth/heartbeat", method="POST").code == "NOT_SIGNED_IN" assert dec(key="POST /api/auth/heartbeat", method="POST", signed_in=True, active=True) is ADMITTED assert dec(key="GET /api/plan", signed_in=True, active=True).code == "FORBIDDEN" assert dec(key="GET /api/plan", signed_in=True, active=True, holds=lambda p: p == "plan.view") is ADMITTED assert dec(key="GET /api/ops/impersonations", signed_in=True, active=True, holds=lambda p: p == "plan.view").code == "FORBIDDEN" assert dec(key="GET /api/ops/impersonations", signed_in=True, active=True, holds=lambda p: p == "owner") is ADMITTED # A client is not told /plan exists: the 404 shell, not a 403. v = dec(key=access.CATCH_ALL, page="/plan", wants_html=True, signed_in=True, active=True) assert (v.status, v.code) == (404, "NOT_FOUND") v = dec(key=access.CATCH_ALL, page="/sources", wants_html=True, signed_in=True, active=True) assert (v.status, v.code) == (403, "FORBIDDEN") def test_origin_and_read_only_come_before_everything_but_classification(self): for key in ("POST /api/auth/login", "POST /api/subscribers", "POST /api/bw/levels", "POST /api/items/{item_id}/decide"): assert dec(key=key, method="POST", origin_ok=False, signed_in=True, active=True, holds=lambda p: True).code == "ORIGIN_MISMATCH", key assert dec(key="GET /api/products", origin_ok=False, signed_in=True, active=True) is ADMITTED assert dec(key="POST /api/bw/levels", method="POST", signed_in=True, active=True, read_only=True).code == "VIEW_AS_READ_ONLY" for key in sorted(access.READ_ONLY_EXEMPT): assert dec(key=key, method="POST", signed_in=True, active=True, read_only=True) is ADMITTED, key def test_the_must_change_gate(self): held = dict(signed_in=True, active=True, must_change=True, holds=lambda p: True) for key in sorted(access.MUST_CHANGE_EXEMPT): assert dec(key=key, method=key.split()[0], **held) is ADMITTED, key assert dec(key=access.CATCH_ALL, page="/products", wants_html=True, **held) is ADMITTED # the shell; the SPA draws the form assert dec(key="GET /api/products", **held).code == "PASSWORD_CHANGE_REQUIRED" assert dec(key="GET /api/products", **held).status == 403 v = dec(key="GET /products/{slug}", wants_html=True, **held) assert (v.status, v.location) == (302, "/account") assert dec(key="POST /api/bw/levels", method="POST", **held).code == "PASSWORD_CHANGE_REQUIRED" def test_next_cases_shared_with_the_spa(self): cases = json.loads((MAIN / "tests" / "fixtures" / "next_cases.json").read_text()) for case in cases: assert access.safe_next(case["input"]) == case["expected"], case assert access.login_redirect("//evil.example") == "/login" assert access.login_redirect("/products/x y") == "/login?next=/products/x%20y" def test_wants_html_comes_from_the_key_alone(self): assert access.wants_html(access.CATCH_ALL) and access.wants_html("GET /products/{slug}") assert not access.wants_html("GET /api/products") and not access.wants_html("GET /sitemap.xml") @pytest.fixture def world(monkeypatch): T.fresh(monkeypatch) monkeypatch.setattr(settings, "site_access", "members") T.person("rian") with _env.TestSessionLocal() as db: db.add(AccountLevel(name="admin", permissions=list(accounts.SEED_LEVELS["admin"]["permissions"]), assignable=[])) db.commit() T.person("adam", level="admin") T.person("gone", status="disabled") yield @pytest.fixture def site(tmp_path): (tmp_path / "assets").mkdir() (tmp_path / "index.html").write_text(SHELL) (tmp_path / "logo.png").write_bytes(b"png-bytes") (tmp_path.parent / "outside.txt").write_text("above the shell root") before = len(main.app.routes) main.mount_site(main.app, tmp_path) try: yield tmp_path finally: del main.app.routes[before:] access.ROOT_FILES = frozenset() if hasattr(main.app.state, "not_found"): del main.app.state.not_found class TestMembersMode: @pytest.mark.parametrize("accept", ["application/json", "text/html", None]) def test_anonymous_answers_are_the_same_under_any_accept(self, world, site, accept): headers = {"Accept": accept} if accept else {} c = T.client(**headers) r = c.get("/", follow_redirects=False) assert r.status_code == 302 and r.headers["location"] == "/login" assert r.headers["x-robots-tag"] == "noindex, nofollow" and r.headers["cache-control"] == "no-store" r = c.get("/products/some-bottle-1", follow_redirects=False) assert r.status_code == 302 and r.headers["location"] == "/login?next=/products/some-bottle-1" assert c.get("/data", follow_redirects=False).status_code == 302 r = c.get("/api/sources/expandable") assert r.status_code == 401 and r.json()["detail"]["error_code"] == "NOT_SIGNED_IN" assert c.get("/api/health").status_code == 200 for path in ("/docs", "/openapi.json", "/sitemap.xml", "/feed.xml", "/llms.txt"): assert c.get(path, follow_redirects=False).status_code == 404, path r = c.get("/robots.txt") assert r.status_code == 200 and r.text == "User-agent: *\nDisallow: /\n" and "Sitemap" not in r.text assert r.headers["x-robots-tag"] == "noindex, nofollow" assert c.get("/login", follow_redirects=False).status_code in (200, 404) # the shell route lands with the SPA pages assert c.get("/logo.png").content == b"png-bytes" # a root file, by exact path def test_signed_in_admin_sees_the_storefront_and_client_pages_not_the_plan(self, world, site): c = T.client() T.as_user(c, "adam") assert c.get("/", follow_redirects=False).status_code == 200 assert c.get("/api/sources/expandable").status_code == 200 assert c.get("/discuss", follow_redirects=False).status_code == 200 r = c.get("/plan", follow_redirects=False) assert r.status_code == 404 and '

' in r.text r = c.get("/api/plan") assert r.status_code == 403 and r.json()["detail"]["error_code"] == "FORBIDDEN" assert c.get("/api/sources").status_code == 403 assert c.post("/api/items/x/decide", json={"text": "no"}).status_code == 403 assert c.get("/no-such-page", follow_redirects=False).status_code == 404 def test_the_owner_reaches_the_owner_routes(self, world, site): c = T.client() T.as_user(c, "rian") assert c.get("/api/plan").status_code == 200 assert c.get("/api/ops/impersonations").json() == {"impersonations": []} assert c.get("/plan", follow_redirects=False).status_code == 200 def test_a_disabled_account_is_refused_with_its_cookie(self, world, site): c = T.client() T.as_user(c, "gone") r = c.get("/api/sources/expandable") assert r.status_code == 401 and r.json()["detail"]["error_code"] == "ACCOUNT_DISABLED" assert c.get("/", follow_redirects=False).status_code == 302 def test_public_mode_opens_the_storefront_and_nothing_else(self, world, site, monkeypatch): monkeypatch.setattr(settings, "site_access", "public") c = T.client() r = c.get("/api/sources/expandable") assert r.status_code == 200 and "x-robots-tag" not in r.headers assert c.get("/no-such-page", follow_redirects=False).status_code == 404 assert c.get("/api/plan").status_code == 401 assert c.get("/discuss", follow_redirects=False).status_code == 302 assert "Disallow: /api/" in c.get("/robots.txt").text def test_every_write_needs_a_matching_origin(self, world): c = T.client() T.as_user(c, "rian") bare = {"Origin": ""} r = c.post("/api/auth/heartbeat", headers=bare) assert r.status_code == 403 and r.json()["detail"]["error_code"] == "ORIGIN_MISMATCH" r = c.post("/api/auth/heartbeat", headers={"Origin": "https://evil.example"}) assert r.status_code == 403 assert c.post("/api/auth/heartbeat", headers={"Origin": "", "Referer": "https://testserver/account"}).status_code == 200 assert c.post("/api/auth/heartbeat").status_code == 200 class TestTraversalThroughTheMiddleware: def _call(self, path: str, cookie: str | None = None) -> tuple[int, bytes, dict]: headers = [(b"host", b"testserver")] if cookie: headers.append((b"cookie", cookie.encode())) scope = {"type": "http", "http_version": "1.1", "method": "GET", "scheme": "https", "path": path, "raw_path": path.encode(), "query_string": b"", "root_path": "", "headers": headers, "client": ("127.0.0.1", 1), "server": ("testserver", 443)} status, body, hdrs = {}, bytearray(), {} async def receive(): return {"type": "http.request", "body": b"", "more_body": False} async def send(message): if message["type"] == "http.response.start": status["code"] = message["status"] hdrs.update({k.decode(): v.decode() for k, v in message["headers"]}) elif message["type"] == "http.response.body": body.extend(message.get("body", b"")) asyncio.run(main.app(scope, receive, send)) return status["code"], bytes(body), hdrs @pytest.mark.parametrize("path", ["/../../../../../../etc/hostname", "/%2e%2e/%2e%2e/outside.txt", "/assets/../../outside.txt"]) def test_probes_never_reach_a_file_in_any_mode(self, world, site, path, monkeypatch): code, body, _ = self._call(path) assert code in (302, 401, 404), (path, code) assert b"above the shell root" not in body and b"png-bytes" not in body monkeypatch.setattr(settings, "site_access", "public") code, body, _ = self._call(path) assert code == 404, (path, code) assert b"above the shell root" not in body and b"png-bytes" not in body class TestActMode: def _comment(self) -> int: with _env.TestSessionLocal() as db: adam_id = db.scalar(select(Account.id).where(Account.username == "adam")) row = DiscussionComment(feature_key="structure:urls", author="Adam", author_id=adam_id, body="original") db.add(row) db.commit() return row.id def test_an_act_mode_write_leaves_exactly_one_row_naming_both_ids(self, world): cid = self._comment() c = T.client() T.as_user(c, "rian") assert c.post("/api/bw/view-as/start", json={"target": "adam", "mode": "act"}).status_code == 200 r = c.post(f"/api/discussion/feature-comments/{cid}", json={"body": "edited as adam"}) assert r.status_code == 200, r.text with _env.TestSessionLocal() as db: rows = db.scalars(select(AuditLog).where(AuditLog.action == "act_as.write")).all() ids = {a.username: a.id for a in db.scalars(select(Account))} assert len(rows) == 1 assert rows[0].account_id == ids["rian"] and rows[0].acting_as_id == ids["adam"] assert rows[0].detail == {"key": "POST /api/discussion/feature-comments/{comment_id}", "status": 200} def test_a_read_only_view_as_write_is_refused_and_leaves_no_row(self, world): cid = self._comment() c = T.client() T.as_user(c, "rian") c.post("/api/bw/view-as/start", json={"target": "adam"}) r = c.post(f"/api/discussion/feature-comments/{cid}", json={"body": "edited"}) assert r.status_code == 403 and r.json()["detail"]["error_code"] == "VIEW_AS_READ_ONLY" assert c.post("/api/bw/view-as/stop").status_code == 200 # the exemption with _env.TestSessionLocal() as db: assert db.scalar(select(AuditLog).where(AuditLog.action == "act_as.write")) is None assert db.get(DiscussionComment, cid).body == "original"