"""The public-link door (erp_platform.public) and the composition's public app (compose.public_app).
Over the real app (httpx's ASGI transport) and the real tables on SQLite. The races that DataK3 decides at
COMMIT are measured on a bucket by scripts/live_public_check.py.
"""
import logging
import re
import sys
import textwrap
import threading
from datetime import timedelta
import anyio
import httpx
import pytest
from sqlalchemy import create_engine, select
from erp_platform import compose, public
from erp_platform.bricks import Brick, Refused
from erp_platform.public import Download, Page
BASE = "https://links.co.example"
# ── a small contracts-like brick with public handlers ───────────────────────────────────────
CT = Brick("zzct", instructions="test")
SIGNED: list = []
SEEN_LINKS: list = []
def sign_contract(caller: dict, contract_id: str, typed_name: str) -> dict:
"""The brick's own use case. It decides what a PUBLIC caller may do: only its own subject."""
subject = public.subject_of(caller)
if subject is None or subject[:2] != ("contract", contract_id) or subject[2] != "SIGN":
raise Refused(404, f"no contract {contract_id}")
if typed_name == "out of order":
raise Refused(409, "signing waits on Sam Lee, who signs first")
SIGNED.append((contract_id, caller["email"], typed_name))
return {"signed": True}
@CT.public("SIGN", subject_kind="contract")
def signing_page(link, req):
SEEN_LINKS.append(link)
done = " (signed)" if link.used_at else ""
return Page(f"Contract {link.subject_id}{done}",
f"<p>For {public.esc(link.recipient)}</p><form method='post'><input name='name'>"
"<button>Sign</button></form>")
@signing_page.act
def sign(link, req):
SEEN_LINKS.append(link)
sign_contract(link.actor, link.subject_id, req.form.get("name", ""))
return Page("Signed", "<p>Thank you.</p>")
@CT.public("READ", subject_kind="contract")
def signed_copy(link, req):
return Download(b"%PDF-1.7 frozen bytes", "application/pdf", f"{link.subject_id}.pdf")
INTAKE_SAVES: list = []
@CT.public("SUBMIT", subject_kind="intake")
def intake_form(link, req):
return Page("Your details", "<form method='post'><input name='address'></form>")
@intake_form.act(single_use=False)
def save_intake(link, req):
INTAKE_SAVES.append(dict(req.form))
if req.form.get("final") == "yes":
link.use()
return Page("Saved", "<p>Saved.</p>")
# ── fixtures ────────────────────────────────────────────────────────────────────────────────
@pytest.fixture(autouse=True)
def database(tmp_path, monkeypatch):
engine = create_engine(f"sqlite:///{tmp_path / 'public.db'}",
connect_args={"timeout": 30, "check_same_thread": False})
public.md.create_all(engine)
public.configure(engine.begin)
monkeypatch.setenv("PUBLIC_LINK_BASE_URL", BASE)
monkeypatch.setenv("PUBLIC_REDIRECT_HOSTS", "co.example,*.shop.example")
SIGNED.clear(), SEEN_LINKS.clear(), INTAKE_SAVES.clear()
yield engine
public.configure(None)
APP = public.app(CT.public_handlers, name="zz-public")
async def _call(method, path, **kw):
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=APP, client=("203.0.113.9", 4000)),
base_url="http://door") as c:
return await c.request(method, path, **kw)
def call(method, path, **kw) -> httpx.Response:
return anyio.run(lambda: _call(method, path, **kw))
def path_of(url: str) -> str:
assert url.startswith(BASE)
return url[len(BASE):]
def token_of(url: str) -> str:
return url.rsplit("/", 1)[1]
def issue(grant="SIGN", subject=("contract", "C-17|2"), recipient="Kim@Client.example", **kw):
return CT.links.issue(subject[0], subject[1], recipient, grant, actor="owner@co.example", **kw)
def all_rows(engine) -> list:
with engine.connect() as c:
return [dict(r._mapping) for t in (public.links, public.events) for r in c.execute(select(t))]
# ── issue / resolve / seen ──────────────────────────────────────────────────────────────────
def test_issue_returns_the_link_once_and_the_page_resolves_it(database):
link_id, url = issue()
assert re.fullmatch(BASE + r"/p/[A-Za-z0-9_-]{43}", url)
r = call("GET", path_of(url))
assert r.status_code == 200 and "Contract C-17|2" in r.text and "kim@client.example" in r.text
[st] = CT.links.status("contract", "C-17|2")
assert st["link_id"] == link_id and st["recipient"] == "kim@client.example" and st["grant"] == "SIGN"
assert st["state"] == "active" and st["seen_count"] == 1 and st["used_at"] is None
def test_the_table_holds_the_hash_never_the_token(database):
_, url = issue()
token = token_of(url)
call("GET", path_of(url))
rows = all_rows(database)
assert rows, "nothing stored"
dump = repr(rows)
assert token not in dump and token[:16] not in dump and token[-16:] not in dump
[link] = [r for r in rows if "token_hash" in r]
assert link["token_hash"] == public.hash_token(token)
def test_seen_is_kept_apart_from_used_and_a_head_is_not_seen(database):
_, url = issue()
for _ in range(3):
call("GET", path_of(url))
call("HEAD", path_of(url)) # a mail scanner's probe
[st] = CT.links.status("contract", "C-17|2")
assert st["seen_count"] == 3 and st["first_seen_at"] <= st["last_seen_at"] and st["used_at"] is None
with database.connect() as c:
row = c.execute(select(public.links)).first()._mapping
assert row["seen_count"] == 3 and row["first_seen_at"] is not None # the projection agrees
def test_the_handler_gets_the_resolved_link_never_the_raw_token(database):
_, url = issue()
token = token_of(url)
call("GET", path_of(url))
call("POST", path_of(url), data={"name": "Kim Lee"})
assert len(SEEN_LINKS) == 2
for link in SEEN_LINKS:
assert token not in repr(link) and token not in repr(link.actor)
assert not any(token in str(getattr(link, f)) for f in link.__dataclass_fields__)
actor = SEEN_LINKS[0].actor
assert actor["auth_source"] == "public" and actor["permissions"] == [] and actor["roles"] == []
assert actor["sub"] == f"public:{SEEN_LINKS[0].link_id}" and actor["email"] == "kim@client.example"
def test_the_act_calls_the_use_case_as_the_public_actor_once(database):
_, url = issue()
r = call("POST", path_of(url), data={"name": "Kim Lee"})
assert r.status_code == 200 and SIGNED == [("C-17|2", "kim@client.example", "Kim Lee")]
again = call("POST", path_of(url), data={"name": "Kim Lee"})
assert again.status_code == 404 and len(SIGNED) == 1 # a second act is refused
assert call("GET", path_of(url)).status_code == 200 # READ until expiry: it says signed
assert CT.links.status("contract", "C-17|2")[0]["state"] == "used"
def test_a_refused_act_does_not_use_up_the_link(database):
_, url = issue()
r = call("POST", path_of(url), data={"name": "out of order"})
assert r.status_code == 409 and "Sam Lee" in r.text
assert call("POST", path_of(url), data={"name": "Kim Lee"}).status_code == 200
assert len(SIGNED) == 1
def test_a_repeatable_act_runs_until_the_handler_uses_the_link(database):
_, url = issue("SUBMIT", ("intake", "p-812|1"), "starter@home.example")
assert call("POST", path_of(url), data={"address": "1 Road"}).status_code == 200
assert call("POST", path_of(url), data={"address": "1 Road", "final": "yes"}).status_code == 200
assert call("POST", path_of(url), data={"address": "2 Road"}).status_code == 404
assert len(INTAKE_SAVES) == 2
def test_a_read_link_downloads_the_frozen_bytes(database):
_, url = issue("READ")
r = call("GET", path_of(url))
assert r.status_code == 200 and r.content.startswith(b"%PDF") and "attachment" in r.headers["content-disposition"]
assert call("POST", path_of(url)).status_code == 404 # READ never acts
# ── single use under concurrency ────────────────────────────────────────────────────────────
def test_single_use_under_ten_threads(database):
_, url = issue()
link = public._resolve(token_of(url), None)
wins, errors, gate = [], [], threading.Barrier(10)
def go():
try:
gate.wait()
wins.append(link.use())
except Exception as exc: # noqa: BLE001 — reported below
errors.append(exc)
threads = [threading.Thread(target=go) for _ in range(10)]
[t.start() for t in threads]
[t.join(30) for t in threads]
assert not errors, errors
assert sorted(wins) == [False] * 9 + [True]
with database.connect() as c:
used = [r for r in c.execute(select(public.events.c.kind)) if r[0] == "used"]
assert len(used) == 1
# ── expire / revoke / reissue, and the one 404 ──────────────────────────────────────────────
def test_expiry_revocation_and_reissue(database):
_, expired = issue(expires_in=timedelta(seconds=-1))
assert call("GET", path_of(expired)).status_code == 404
link_id, url = issue(recipient="other@client.example")
assert CT.links.revoke(link_id, actor="owner@co.example", reason="sent to the wrong person")
assert not CT.links.revoke(link_id) # already revoked: says so, changes nothing
assert call("GET", path_of(url)).status_code == 404
first_id, first = issue(recipient="kim@client.example")
new_id, new = CT.links.reissue(first_id, actor="owner@co.example")
assert call("GET", path_of(first)).status_code == 404 and call("GET", path_of(new)).status_code == 200
by_id = {s["link_id"]: s for s in CT.links.status("contract", "C-17|2")}
assert by_id[first_id]["state"] == "revoked" and by_id[new_id]["reissued_from"] == first_id
assert by_id[link_id]["revoke_reason"] == "sent to the wrong person"
with database.connect() as c:
kinds = {r[0] for r in c.execute(select(public.events.c.kind))}
assert {"issued", "revoked", "reissued", "refused", "seen"} <= kinds
def test_one_person_one_grant_one_live_link(database):
a_id, a = issue()
b_id, b = issue() # the same person, again
assert call("GET", path_of(a)).status_code == 404 and call("GET", path_of(b)).status_code == 200
assert {s["link_id"]: s["state"] for s in CT.links.status("contract", "C-17|2")} == {a_id: "revoked", b_id: "active"}
_, read = issue("READ") # another grant for the same person: both live
assert call("GET", path_of(b)).status_code == 200 and call("GET", path_of(read)).status_code == 200
def test_every_failure_is_the_same_404(database):
_, expired = issue(recipient="e@x.example", expires_in=timedelta(seconds=-1))
rid, revoked = issue(recipient="r@x.example")
CT.links.revoke(rid)
_, used = issue(recipient="u@x.example")
call("POST", path_of(used), data={"name": "U"})
_, read = issue("READ", recipient="g@x.example")
_, track = issue("TRACK", recipient="t@x.example")
other = Brick("zzother")
_, foreign = other.links.issue("contract", "C-1", "f@x.example", "SIGN") # no handler in this app
attempts = [("GET", "/p/" + "A" * 43), ("GET", "/p/short"), ("GET", path_of(expired)),
("GET", path_of(revoked)), ("POST", path_of(used)), ("POST", path_of(read)),
("GET", "/p/" + token_of(track)), ("GET", path_of(foreign)), ("GET", "/p/x/y/z"),
("GET", "/admin"), ("DELETE", path_of(read)), ("GET", "/c/" + "B" * 43)]
replies = [call(m, p, data={"name": "x"} if m == "POST" else None) for m, p in attempts]
first = replies[0]
assert first.status_code == 404 and "This link no longer works" in first.text
def shape(r):
return r.status_code, r.content, sorted((k, v) for k, v in r.headers.items() if k != "x-request-id")
for (m, p), r in zip(attempts, replies):
assert shape(r) == shape(first), f"{m} {p} is distinguishable"
# ── the headers, and what the public app is NOT ─────────────────────────────────────────────
def test_strict_headers_on_every_response(database):
_, url = issue()
for r in (call("GET", path_of(url)), call("GET", "/p/nope"), call("GET", "/healthz")):
h = r.headers
assert h["content-security-policy"] == public.CSP
assert "default-src 'none'" in h["content-security-policy"] and "script-src" not in h["content-security-policy"]
assert "frame-ancestors 'none'" in h["content-security-policy"]
assert h["x-robots-tag"].startswith("noindex") and h["cache-control"] == "no-store"
assert h["referrer-policy"] == "no-referrer"
assert "<script" not in call("GET", path_of(url)).text.lower()
def make_brick(root, package, name):
pkg = root / package
pkg.mkdir()
(pkg / "__init__.py").write_text("")
(pkg / "schema.sql").write_text(f"CREATE TABLE IF NOT EXISTS {name}_rows (id INTEGER NOT NULL, PRIMARY KEY (id));")
(pkg / "use_cases.py").write_text(textwrap.dedent(f"""
from erp_platform.bricks import Brick
from erp_platform.public import Page
B = Brick({name!r})
@B.use_case(kind="query", route="GET /secrets")
def list_secrets_{name}(caller):
'''Staff only.'''
return {{}}
@B.use_case(kind="command", route="POST /admin/wipe", permission="{name}:admin:wipe")
def wipe_{name}(caller):
'''Staff only.'''
return {{}}
@B.public("APPROVE", subject_kind="proof")
def proof(link, req):
return Page("Proof", "")
"""))
@pytest.fixture
def composition(tmp_path, monkeypatch):
monkeypatch.syspath_prepend(str(tmp_path))
make_brick(tmp_path, "zz_pa", "pa")
make_brick(tmp_path, "zz_pb", "pb")
path = tmp_path / "compose.yaml"
path.write_text("name: shop\nversion: 0.1.0\ntier: 1\nbricks:\n"
" - { name: pa, brick: 'zz_pa.use_cases:B', version: 0.1.0 }\n"
" - { name: pb, brick: 'zz_pb.use_cases:B', version: 0.1.0 }\n")
yield path
for name in list(sys.modules):
if name.startswith("zz_p"):
del sys.modules[name]
def test_the_public_app_has_no_admin_route(composition):
app = compose.public_app(composition)
# Every route object, flat. Anything without a path of its own — an included router, a mount — shows up
# as its type and fails the comparison: a sub-app could carry routes this list would not see.
paths = sorted(getattr(r, "path", f"<{type(r).__name__}>") for r in app.router.routes)
assert paths == sorted(public.ROUTES), paths # enumerated: nothing else is there
staff = compose.app(composition)
staff_paths = set(staff.openapi()["paths"]) | {getattr(r, "path", "") for r in staff.router.routes}
assert {"/pa/secrets", "/pa/admin/wipe", "/mcp", "/composition"} <= staff_paths
assert not staff_paths & {"/p/{token}", "/o/{token}", "/c/{token}"} # and the staff app has no door
async def probe():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://door") as c:
return [await c.request(m, p) for m, p in (("GET", "/pa/secrets"), ("POST", "/pa/admin/wipe"),
("POST", "/mcp"), ("GET", "/composition"),
("GET", "/openapi.json"), ("GET", "/docs"))]
assert {r.status_code for r in anyio.run(probe)} == {404}
report = compose.check(compose.load(composition))
assert {(h["brick"], h["grant"], h["subject_kind"]) for h in report["public"]} == {
("pa", "APPROVE", "proof"), ("pb", "APPROVE", "proof")}
def test_plan_says_the_public_app_needs_its_own_account_and_no_pool(composition, capsys):
assert compose.main(["plan", str(composition), "--unsigned", "wiring only: no customer design here"]) == 0
out = capsys.readouterr().out
assert "public app (2 handlers)" in out and "no user pool" in out and "never migrates" in out
assert "its OWN service account" in out and "PUBLIC_LINK_BASE_URL" in out
assert "never set DODIL_USER_POOL" in out
def test_serve_public_runs_with_the_access_log_off(composition, monkeypatch):
seen = {}
import uvicorn
monkeypatch.setattr(uvicorn, "run", lambda app, **kw: seen.update(kw, app=app))
assert compose.main(["serve", "--public", str(composition)]) == 0
assert seen["access_log"] is False
assert sorted(getattr(r, "path", "?") for r in seen["app"].router.routes) == sorted(public.ROUTES)
assert compose.main(["migrate", "--public", str(composition)]) == 2 # the public app never migrates
def test_a_brick_table_may_not_take_a_platform_name(tmp_path, monkeypatch):
monkeypatch.syspath_prepend(str(tmp_path))
pkg = tmp_path / "zz_pc"
pkg.mkdir()
(pkg / "__init__.py").write_text("")
(pkg / "schema.sql").write_text("CREATE TABLE IF NOT EXISTS erp_public_link (id INTEGER NOT NULL, PRIMARY KEY (id));")
(pkg / "use_cases.py").write_text("from erp_platform.bricks import Brick\nB = Brick('pc')\n")
path = tmp_path / "compose.yaml"
path.write_text("name: t\nversion: 0.1.0\nbricks:\n - { name: pc, brick: 'zz_pc.use_cases:B' }\n")
r = compose.check(compose.load(path))
assert any("reserved for erp-platform" in p for p in r["problems"])
del sys.modules["zz_pc.use_cases"], sys.modules["zz_pc"]
# ── TRACK: opens and clicks ─────────────────────────────────────────────────────────────────
def test_track_click_goes_only_to_an_allow_listed_destination(database, monkeypatch):
with pytest.raises(ValueError, match="PUBLIC_REDIRECT_HOSTS"):
issue("TRACK", destination="https://evil.example/phish")
with pytest.raises(ValueError):
issue("TRACK", destination="http://co.example/plain-http")
with pytest.raises(ValueError):
issue("TRACK", destination="https://co.example.evil.example/")
link_id, url = issue("TRACK", destination="https://www.shop.example/offer?x=1")
assert "/c/" in url
r = call("GET", path_of(url) + "?url=https://evil.example/", follow_redirects=False)
assert r.status_code == 303 and r.headers["location"] == "https://www.shop.example/offer?x=1"
call("GET", path_of(url), follow_redirects=False)
[st] = [s for s in CT.links.status("contract", "C-17|2") if s["link_id"] == link_id]
assert st["clicked"] and st["seen_count"] == 2
with database.connect() as c:
assert [r[0] for r in c.execute(select(public.events.c.kind))].count("clicked") == 1 # once per event
monkeypatch.setenv("PUBLIC_REDIRECT_HOSTS", "co.example") # the list narrowed after issue
assert call("GET", path_of(url), follow_redirects=False).status_code == 404
def test_track_open_is_a_pixel_counted_once(database):
link_id, url = issue("TRACK")
assert "/o/" in url
for _ in range(3):
r = call("GET", path_of(url))
assert r.status_code == 200 and r.headers["content-type"] == "image/gif" and r.content == public.PIXEL_GIF
assert call("GET", "/o/" + "Z" * 43).content == public.PIXEL_GIF # unknown: the same pixel, nothing recorded
[st] = CT.links.status("contract", "C-17|2")
assert st["opened"] and st["seen_count"] == 3
with database.connect() as c:
assert [r[0] for r in c.execute(select(public.events.c.kind))].count("opened") == 1
# ── the token never reaches a log ───────────────────────────────────────────────────────────
def test_the_token_is_absent_from_every_log_line(database, caplog):
_, url = issue()
token = token_of(url)
with caplog.at_level(logging.DEBUG):
call("GET", path_of(url))
call("POST", path_of(url), data={"name": "Kim"})
call("POST", path_of(url), data={"name": "Kim"}) # refused: the 404
call("GET", path_of(url) + "/extra") # unmatched: the raw path
call("GET", "/p/" + token[::-1])
# Every server-side line. httpx's own INFO line is THIS TEST's client logging the URL it requested — the
# holder's browser, in production — and is not the app's.
lines = [r.getMessage() for r in caplog.records if r.name != "httpx"]
assert any('"path": "/p/{token}"' in l for l in lines), lines # the route template is logged
assert any("{redacted}" in l for l in lines) # and an unmatched path redacted
assert not any(token in l or token[::-1] in l for l in lines)
def test_issue_needs_the_public_address_and_a_real_grant(database, monkeypatch):
with pytest.raises(ValueError, match="grant"):
issue("DELETE")
with pytest.raises(ValueError, match="one person"):
issue(recipient=" ")
monkeypatch.delenv("PUBLIC_LINK_BASE_URL")
with pytest.raises(RuntimeError, match="PUBLIC_LINK_BASE_URL"):
issue()
assert all_rows(database) == [] # refused before any write
with pytest.raises(ValueError, match="does not act"):
CT.public("READ", subject_kind="memo")(lambda l, r: None).act(lambda l, r: None)
with pytest.raises(ValueError, match="TRACK"):
CT.public("TRACK", subject_kind="mail")
def test_platform_sql_is_the_schema_the_code_uses(monkeypatch):
"""platform.sql (what migrate applies) and the tables the code reads and writes (public.md, and since 0.5
idempotency.md) must agree."""
monkeypatch.setenv("BUCKET", "not-used")
from erp_platform import idempotency, migrate
sql = migrate.platform_schema().read_text()
expected = migrate.expected_columns(sql)
assert expected == {t.name: [c.name for c in t.c]
for t in (public.links, public.events, idempotency.claims)}
assert all(migrate.classify(s) == "SAFE" for s in migrate.statements(sql))
def test_readyz_says_whether_the_link_table_answers(database, tmp_path):
assert call("GET", "/readyz").status_code == 200
public.configure(create_engine(f"sqlite:///{tmp_path / 'empty.db'}").begin) # never migrated
r = call("GET", "/readyz")
assert r.status_code == 503 and r.json()["ok"] is False
assert call("GET", "/healthz").status_code == 200 # liveness needs no database