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, live=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_a_development_surface_does_not_exist_on_live(self): """SITE_ROLE=live: the surfaces built for the build answer 404 before any sign-in redirect or permission answer, to the owner too, so live never discloses them.""" owner = dict(signed_in=True, active=True, holds=lambda p: True, live=True) # A page: the not-found shell, by status, not by what the shell renders. for path in sorted(access.DEVELOPMENT_SPA): v = dec(key=access.CATCH_ALL, page=path, wants_html=True, **owner) assert (v.status, v.code, v.summary) == (404, "NOT_FOUND", "No such page."), path assert dec(key=access.CATCH_ALL, page=path + "/", wants_html=True, **owner).status == 404, path # A read: JSON, never the shell (`_render` picks the shell off the summary). v = dec(key="GET /api/plan", **owner) assert (v.status, v.summary) == (404, "No such route.") assert dec(key="GET /api/items", **owner).status == 404 assert dec(key="POST /api/todos/{todo_id}/complete", method="POST", **owner).status == 404 v = dec(key="GET /{page}.html", page="/privacy.html", wants_html=True, **owner) assert (v.status, v.summary) == (404, "No such page.") # Anonymous on the public site: the same 404, not a 302 to the sign-in page. v = dec(key=access.CATCH_ALL, page="/plan", wants_html=True, site_open=True, live=True) assert (v.status, v.code) == (404, "NOT_FOUND") def test_the_live_features_and_the_storefront_are_untouched_by_the_environment(self): owner = dict(signed_in=True, active=True, holds=lambda p: True, live=True) for path in ("/collectors", "/review", "/sources", "/admin"): assert dec(key=access.CATCH_ALL, page=path, wants_html=True, **owner) is ADMITTED, path for key in ("GET /api/collectors", "GET /api/review/sheets", "GET /api/sources", "GET /api/pages/status", "GET /api/discussion/threads"): assert dec(key=key, **owner) is ADMITTED, key assert dec(key="GET /api/products", site_open=True, live=True) is ADMITTED assert dec(key=access.CATCH_ALL, page="/", wants_html=True, site_open=True, live=True) is ADMITTED for path in sorted(access.SPA_PUBLIC_ALWAYS): assert dec(key=access.CATCH_ALL, page=path, wants_html=True, live=True) is ADMITTED, path assert dec(key="POST /api/auth/login", method="POST", live=True) is ADMITTED def test_staging_answers_a_development_surface_exactly_as_its_class_says(self): """live=False is the whole of the old behaviour: the class decides and nothing else.""" owner = dict(signed_in=True, active=True, holds=lambda p: True) assert dec(key="GET /api/plan", **owner) is ADMITTED assert dec(key=access.CATCH_ALL, page="/discuss", wants_html=True, **owner) is ADMITTED v = dec(key=access.CATCH_ALL, page="/plan", wants_html=True, signed_in=True, active=True) assert (v.status, v.code) == (404, "NOT_FOUND") # the plan_view shell, not the environment assert dec(key="GET /api/plan", signed_in=True, active=True).code == "FORBIDDEN" assert dec(key="GET /api/plan").code == "NOT_SIGNED_IN" 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 TestTheEnvironmentLineThroughTheMiddleware: """`SITE_ROLE=live` through the real app: the owner gets a 404 shell for a development page and a 404 JSON body for its read, and the live features still open.""" def test_the_owner_loses_the_development_surfaces_and_keeps_the_rest(self, world, site, monkeypatch): c = T.client() T.as_user(c, "rian") assert c.get("/plan", follow_redirects=False).status_code == 200 monkeypatch.setattr(settings, "site_role", "live") r = c.get("/plan", follow_redirects=False) assert r.status_code == 404 and '

' in r.text assert c.get("/discuss", follow_redirects=False).status_code == 404 r = c.get("/api/plan") assert r.status_code == 404 and r.json()["detail"]["error_code"] == "NOT_FOUND" assert r.headers["content-type"].startswith("application/json"), "a read gets JSON, not the shell" assert c.get("/api/items").status_code == 404 # The live features and the account's own pages are unmoved (the rest of the live set # is walked in `decide()` above; these two are the ones that need no catalogue tables). assert c.get("/api/ops/impersonations").status_code == 200 assert c.get("/api/pages/status", params={"path": "/"}).status_code == 200 assert c.get("/account", follow_redirects=False).status_code == 200 assert c.get("/api/health").json()["role"] == "live" def test_only_the_literal_live_is_live_and_health_says_which(self, world, site, monkeypatch): """An unset or misspelt line reads as staging, which shows MORE, so the launch checks the environment by behaviour rather than by trusting the file.""" assert settings.is_live is False c = T.client() assert c.get("/api/health").json()["role"] == "staging" for value, expected in (("", False), ("production", False), ("Live", True), (" live ", True)): monkeypatch.setattr(settings, "site_role", value) assert settings.is_live is expected, value assert c.get("/api/health").json()["role"] == ("live" if expected else "staging"), value class TestThePageStatusRead: """`GET /api/pages/status` is the owner's and the admin level's, in both site modes: the badge tells a person which pages search engines are shown, which is nobody else's business and would be a map of the site to anyone who could read it anonymously.""" def test_anonymous_is_refused_in_both_modes(self, world, site, monkeypatch): c = T.client() r = c.get("/api/pages/status", params={"path": "/"}) assert r.status_code == 401 and r.json()["detail"]["error_code"] == "NOT_SIGNED_IN" monkeypatch.setattr(settings, "site_access", "public") r = c.get("/api/pages/status", params={"path": "/"}) assert r.status_code == 401, "a public storefront does not make the owner's marks public" def test_a_member_level_account_is_refused_and_an_admin_is_not(self, world, site): T.person("consumer", level="member") c = T.client() T.as_user(c, "consumer") r = c.get("/api/pages/status", params={"path": "/"}) assert r.status_code == 403 and r.json()["detail"]["error_code"] == "FORBIDDEN" T.as_user(c, "adam") r = c.get("/api/pages/status", params={"path": "/"}) assert r.status_code == 200 and r.json()["status"] == "indexable" def test_the_owner_reads_it_with_the_site_mode_beside_the_word(self, world, site): c = T.client() T.as_user(c, "rian") body = c.get("/api/pages/status", params={"path": "/discuss"}).json() assert body["status"] == "unlisted" and body["in_sitemap"] is False assert body["site_open"] is False, "members-only: the badge says so ahead of the word" assert c.get("/api/pages/status", params={"path": "not-a-path"}).status_code == 422 assert c.get("/api/pages/status", params={"path": "/" + "x" * 500}).status_code == 422 def test_the_capability_rides_on_the_me_payload(self, world, site): c = T.client() T.as_user(c, "adam") caps = c.get("/api/bw/me").json()["capabilities"] assert caps["pages_status"] is True T.person("consumer2", level="member") T.as_user(c, "consumer2") assert c.get("/api/bw/me").json()["capabilities"]["pages_status"] is False 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"