"""public — the door outsiders reach with no login, built once for every brick.
A counterparty signs a contract, a starter fills in their details, a client approves a proof: none of them
has an account, and none should need one. They hold a LINK. This module is everything about that link —
issuing it, resolving it, recording that it was seen, letting it act once — and the separate app that
serves it (`compose.public_app`, `erp-compose serve --public`). A brick never writes a token table or a
public route of its own: the established pattern is (a second app, one
token table, seen kept apart from acted), and the reason it is platform code is that a second
implementation is a second place the token rules drift.
ONE TABLE, platform-owned: `erp_public_link` (+ its event log `erp_public_link_event`), created by
`erp-compose migrate` / `erp-migrate --platform` from erp_platform/platform.sql — so a brick issues links
without owning a door, and two bricks in one bucket never both claim the table. A row is one link:
token_hash sha256 hex of the secret (secrets.token_urlsafe(32)). THE SECRET IS NEVER STORED: it is
returned once, by issue/reissue, and a database dump opens nothing.
link_id the handle bricks use (revoke, reissue, status); not derivable from or to the secret
subject_kind/_id what it opens (contract / C-17|2); issued_by_brick: whose handler serves it
recipient the PERSON it was sent to — one token per person per grant, never one per document,
because a shared link cannot say who acted
grant_type READ | APPROVE | SIGN | SUBMIT | TRACK ("grant" is an SQL keyword)
expires_at nullable; revoked_at/_by/revoke_reason: revocation is a column, never a delete
first_seen_at, last_seen_at, seen_count — SEEN, kept apart from used_at — the ACT (signed, submitted)
reissued_from the link this one replaced
EVENTS, decided here rather than left to each brick: one row per issue, seen, use, refusal, revoke and
reissue, with time, IP and user agent (hr's HR-53 and contracts' CT-53 both ask the door to audit every
use). The rows are the record; the seen/used columns are a projection kept for a quick SQL read — status()
computes from the events, so a projection update lost to a concurrent write (40001) loses nothing. A brick
still writes its OWN domain evidence (contracts' hash-chained signature events) from its use case.
SINGLE USE, by the measured property. `use()` claims the act by a plain INSERT of the event (link_id,
"used"): a second claim fails 23505 (at the statement, or at COMMIT on DataK3 — scripts/live_public_check.py
measures 10 concurrent claims → 1). The same property the capability claim and the optimistic version stand
on; no UPDATE … WHERE used_at IS NULL, whose rowcount DataK3 reports before COMMIT decides.
NO ORACLE. Unknown, malformed, expired, revoked, wrong grant, an act on a used link, a path that matches
nothing: the SAME 404 page, byte for byte, with the same headers. A holder of one valid token learns nothing
about any other. A HEAD (mail scanners use it) is never counted as seen; a GET never acts.
HANDLERS. A brick registers what a link of a grant and subject kind shows and does:
@CT.public("SIGN", subject_kind="contract")
def signing_page(link: Link, req: Request) -> Page: ... # GET: render
@signing_page.act # POST: act (single use by default)
def sign(link: Link, req: Request) -> Page:
consent_and_sign(link.actor, link.subject_id, typed_name=req.form["name"])
The handler gets the RESOLVED link — subject, recipient, grant, link_id — never the raw token, and calls
the brick's own use cases with `link.actor`: a PUBLIC identity (auth_source "public", no roles, no
permissions, sub "public:<link_id>"), not a pool user. The use case decides what a public actor may do
(`public.subject_of(caller)`); no staff permission is ever satisfied by one.
TRACK. An open pixel (`/o/<token>`, a 1×1 GIF) and a click (`/c/<token>`, a 303 to the destination STORED on
the link at issue, allow-listed by host in PUBLIC_REDIRECT_HOSTS at issue AND at click — never a URL from
the request, so never an open redirect). Each counts once per event ("opened", "clicked" are once-only
event rows). NOISE, stated: Apple Mail Privacy Protection and Gmail's image proxy fetch every pixel, so an
open means "a mail client or proxy fetched it"; corporate scanners (Safe Links, Mimecast) follow links
before a person does, so a click minutes after delivery from a datacentre address may be a robot. The
events keep IP and user agent so a report can discount them. A GET on a signing page is seen, never acted.
RATE LIMITING belongs in front of the door (the Ignite gateway / an ingress), per IP and per path prefix:
the app scales to zero and holds no shared counter, and a counter per replica is a limit an attacker
multiplies by the replica count. A 256-bit token is not guessable; the limit is for cost, not secrecy.
"""
import hashlib
import hmac
import html
import logging
import os
import re
import secrets
import uuid
from contextlib import nullcontext
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from typing import Any, Callable, NamedTuple, Optional
from urllib.parse import parse_qsl, urlparse
from sqlalchemy import (Column, Integer, MetaData, String, Table, TIMESTAMP, and_, delete, func, insert, select,
update)
from erp_platform import requestlog
log = logging.getLogger("erp_platform.public")
GRANTS = ("READ", "APPROVE", "SIGN", "SUBMIT", "TRACK")
ACTS = ("APPROVE", "SIGN", "SUBMIT") # grants whose POST is an act; READ renders, TRACK is the door's own
DEFAULT_EXPIRY = timedelta(days=14)
_TOKEN = re.compile(r"[A-Za-z0-9_-]{43}") # secrets.token_urlsafe(32)
UA_CHARS = 300
md = MetaData()
links = Table(
"erp_public_link", md,
Column("token_hash", String, primary_key=True),
Column("link_id", String, nullable=False),
Column("subject_kind", String, nullable=False),
Column("subject_id", String, nullable=False),
Column("issued_by_brick", String, nullable=False),
Column("recipient", String, nullable=False),
Column("grant_type", String, nullable=False),
Column("destination", String),
Column("issued_at", TIMESTAMP, nullable=False),
Column("issued_by", String),
Column("expires_at", TIMESTAMP),
Column("revoked_at", TIMESTAMP),
Column("revoked_by", String),
Column("revoke_reason", String),
Column("first_seen_at", TIMESTAMP),
Column("last_seen_at", TIMESTAMP),
Column("seen_count", Integer),
Column("used_at", TIMESTAMP),
Column("reissued_from", String))
events = Table(
"erp_public_link_event", md,
Column("link_id", String, primary_key=True),
Column("event_id", String, primary_key=True), # a uuid; or the name of a once-only event — its claim
Column("kind", String, nullable=False), # issued seen used released refused revoked reissued opened clicked
Column("event_at", TIMESTAMP, nullable=False),
Column("actor", String),
Column("ip", String),
Column("user_agent", String),
Column("detail", String))
ONCE = ("used", "opened", "clicked")
# ── storage ─────────────────────────────────────────────────────────────────────────────────
CONNECT: Optional[Callable[[], Any]] = None # a callable returning a connection context (engine.begin)
def configure(connect: Optional[Callable[[], Any]]) -> None:
"""Where the link tables live. None = erp_platform.db's engine (the app's bucket); tests pass a SQLite
engine's `begin`."""
global CONNECT
CONNECT = connect
def _connect():
if CONNECT is not None:
return CONNECT()
from erp_platform import db
return db.engine.begin()
def _now() -> datetime:
"""Real time, naive UTC (what a TIMESTAMP column returns). Never work.now(): expiry is about the link, not
the business clock a demo moves."""
return datetime.now(timezone.utc).replace(tzinfo=None)
def _utc(v: Any) -> Any:
"""A stored time as naive UTC. SQLite hands back naive; DataK3's pg wire hands back an AWARE datetime for a
TIMESTAMP column (measured 2026-10-17) — and Python refuses to compare the two."""
if isinstance(v, datetime) and v.tzinfo is not None:
return v.astimezone(timezone.utc).replace(tzinfo=None)
return v
def _plain(row: Any) -> dict:
return {k: _utc(v) for k, v in dict(row._mapping).items()}
def hash_token(token: str) -> str:
return hashlib.sha256(token.encode()).hexdigest()
def _lost_claim(exc: BaseException) -> bool:
"""A once-only event already claimed: 23505 (or SQLite's UNIQUE failure), or 40001 — the engine refusing
the second of two concurrent writers at COMMIT."""
from erp_platform import retry
code = retry.sqlstate(exc)
text = str(exc).lower()
return code in ("23505", "40001") or "unique constraint failed" in text or "duplicate key" in text
def _event(conn, link_id: str, kind: str, *, event_id: Optional[str] = None, actor: str = "",
meta: Optional["Request"] = None, detail: str = "", at: Optional[datetime] = None) -> None:
conn.execute(insert(events).values(
link_id=link_id, event_id=event_id or uuid.uuid4().hex, kind=kind, event_at=at or _now(),
actor=actor or None, ip=meta.ip if meta else None,
user_agent=(meta.user_agent or "")[:UA_CHARS] if meta else None, detail=detail or None))
def _project(stmt) -> None:
"""A best-effort write of the seen/used columns. Their truth is the event rows; a concurrent writer on the
same row (40001 at COMMIT) must not fail a page a person is waiting for."""
for _ in range(3):
try:
with _connect() as c:
c.execute(stmt)
return
except Exception as exc: # noqa: BLE001 — logged by type, never by message
last = exc
log.warning('{"event": "projection_skipped", "error": "%s"}', type(last).__name__)
# ── the resolved link, and the identity a public caller acts as ─────────────────────────────
@dataclass(frozen=True)
class Link:
"""What a handler gets: who the link was for and what it opens. NEVER the raw token."""
link_id: str
subject_kind: str
subject_id: str
issued_by_brick: str
recipient: str
grant: str
issued_at: datetime
expires_at: Optional[datetime]
used_at: Optional[datetime]
seen_count: int
_hash: str = field(repr=False, compare=False, default="")
@property
def actor(self) -> dict:
"""The identity to call a use case with: a recipient, not a pool user. No roles, no permissions — so no
staff permission check can pass for it — and the subject it may act on."""
return public_actor(self)
def use(self, conn: Any = None) -> bool:
"""Claim this link's act, once: True for the first caller, False for every other (a plain INSERT of the
"used" event — the measured property). Pass `conn` to claim inside YOUR transaction, so the act and the
claim commit together (tier 1: same bucket); then a claim lost at COMMIT surfaces from your commit
(an IntegrityError naming erp_public_link_event), and the door answers it with the 404 page."""
now = _now()
try:
with (nullcontext(conn) if conn is not None else _connect()) as c:
_event(c, self.link_id, "used", event_id="used", actor=self.recipient, at=now)
except Exception as exc: # noqa: BLE001 — a lost claim is an answer, anything else re-raised
if conn is None and _lost_claim(exc):
return False
raise
_project(update(links).where(links.c.token_hash == self._hash).values(used_at=now))
return True
def release(self, reason: str = "") -> None:
"""Undo this call's claim after its act failed, so the person can try again."""
with _connect() as c:
c.execute(delete(events).where(and_(events.c.link_id == self.link_id, events.c.event_id == "used")))
_event(c, self.link_id, "released", detail=reason)
_project(update(links).where(links.c.token_hash == self._hash).values(used_at=None))
def public_actor(link: Link) -> dict:
return {"sub": f"public:{link.link_id}", "email": link.recipient, "roles": [], "permissions": [],
"connection": "public-link", "tenant": None, "auth_source": "public",
"public_link": {"link_id": link.link_id, "subject_kind": link.subject_kind,
"subject_id": link.subject_id, "grant": link.grant,
"issued_by_brick": link.issued_by_brick}}
def is_public(caller: dict) -> bool:
return caller.get("auth_source") == "public"
def subject_of(caller: dict) -> Optional[tuple]:
"""(subject_kind, subject_id, grant) for a public caller, else None. A use case that serves the public
door checks it matches the record asked for — and answers 404 when it does not."""
p = caller.get("public_link") if is_public(caller) else None
return (p["subject_kind"], p["subject_id"], p["grant"]) if p else None
# ── the issuing side, for bricks ────────────────────────────────────────────────────────────
class Issued(NamedTuple):
link_id: str
url: str # the raw link — returned ONCE; put it in the mail and forget it
def _allowed_destination(url: str) -> bool:
"""https, to a host in PUBLIC_REDIRECT_HOSTS (exact, or `*.example.com` for subdomains)."""
try:
u = urlparse(url)
except ValueError:
return False
if u.scheme != "https" or not u.hostname or u.username or u.password:
return False
host = u.hostname.lower()
for pattern in (h.strip().lower() for h in os.environ.get("PUBLIC_REDIRECT_HOSTS", "").split(",")):
if not pattern:
continue
if pattern.startswith("*.") and host.endswith(pattern[1:]):
return True
if host == pattern:
return True
return False
def _base_url() -> str:
base = os.environ.get("PUBLIC_LINK_BASE_URL", "").strip().rstrip("/")
if not base:
raise RuntimeError("set PUBLIC_LINK_BASE_URL — the public app's address, on a host the mail is "
"signed by — before issuing links")
return base
def _url(token: str, grant: str, destination: Optional[str]) -> str:
path = "p" if grant != "TRACK" else ("c" if destination else "o")
return f"{_base_url()}/{path}/{token}"
class Links:
"""One brick's links: `BRICK.links.issue(...)`. Every operation is scoped to links this brick issued."""
def __init__(self, brick: str):
self.brick = brick
def _ctx(self, conn):
return nullcontext(conn) if conn is not None else _connect()
def issue(self, subject_kind: str, subject_id: str, recipient: str, grant: str, *,
expires_in: Optional[timedelta] = DEFAULT_EXPIRY, actor: str = "",
destination: Optional[str] = None, conn: Any = None, _reissued_from: Optional[str] = None) -> Issued:
"""A new link for one person to one subject with one grant: (link_id, url), the url ONCE.
One person, one grant, one live link: a still-active link for the same (subject, recipient, grant) is
revoked by this issue ("reissued") and named in reissued_from. `expires_in=None` never expires.
TRACK with `destination` is a click link to that (allow-listed) address; TRACK without, an open pixel.
Pass `conn` to issue inside your own transaction (the send and its links commit together)."""
grant = (grant or "").upper()
recipient = (recipient or "").strip().lower()
if grant not in GRANTS:
raise ValueError(f"grant is one of {', '.join(GRANTS)}, not {grant!r}")
if not (subject_kind and str(subject_id) and recipient):
raise ValueError("a link needs a subject_kind, a subject_id and a recipient (one person)")
if destination is not None and grant != "TRACK":
raise ValueError("only a TRACK link has a destination")
if destination is not None and not _allowed_destination(destination):
raise ValueError("the destination is not an https address on a host in PUBLIC_REDIRECT_HOSTS — "
"a click link never redirects anywhere else")
token = secrets.token_urlsafe(32)
url = _url(token, grant, destination) # fails before any write when the base is unset
now = _now()
link_id = uuid.uuid4().hex
with self._ctx(conn) as c:
previous = _reissued_from
for row in c.execute(select(links.c.link_id, links.c.token_hash).where(and_(
links.c.issued_by_brick == self.brick, links.c.subject_kind == subject_kind,
links.c.subject_id == str(subject_id), links.c.recipient == recipient,
links.c.grant_type == grant, links.c.revoked_at.is_(None),
links.c.destination.is_(None) if destination is None else links.c.destination == destination))):
c.execute(update(links).where(links.c.token_hash == row[1]).values(
revoked_at=now, revoked_by=actor or None, revoke_reason="reissued"))
_event(c, row[0], "reissued", actor=actor, detail=f"replaced by {link_id}", at=now)
previous = previous or row[0]
c.execute(insert(links).values(
token_hash=hash_token(token), link_id=link_id, subject_kind=subject_kind,
subject_id=str(subject_id), issued_by_brick=self.brick, recipient=recipient, grant_type=grant,
destination=destination, issued_at=now, issued_by=actor or None,
expires_at=(now + expires_in) if expires_in is not None else None, seen_count=0,
reissued_from=previous))
_event(c, link_id, "issued", actor=actor, at=now,
detail=f"{grant} {subject_kind}/{subject_id}" + (f" reissues {previous}" if previous else ""))
return Issued(link_id, url)
def _row(self, c, link_id: str):
r = c.execute(select(links).where(and_(links.c.link_id == link_id,
links.c.issued_by_brick == self.brick))).first()
return _plain(r) if r is not None else None
def revoke(self, link_id: str, *, actor: str = "", reason: str = "", conn: Any = None) -> bool:
"""Stop a link working. True if it was live; False if it was already revoked or is not this brick's."""
now = _now()
with self._ctx(conn) as c:
row = self._row(c, link_id)
if row is None or row["revoked_at"] is not None:
return False
c.execute(update(links).where(links.c.token_hash == row["token_hash"]).values(
revoked_at=now, revoked_by=actor or None, revoke_reason=reason or None))
_event(c, link_id, "revoked", actor=actor, detail=reason, at=now)
return True
def reissue(self, link_id: str, *, actor: str = "", expires_in: Any = "same", conn: Any = None) -> Issued:
"""Revoke a link and issue its replacement — same subject, person, grant (and destination) — returned
ONCE. `expires_in` defaults to the old link's lifetime."""
with self._ctx(conn) as c:
row = self._row(c, link_id)
if row is None:
raise LookupError(f"no link {link_id} issued by {self.brick}")
if expires_in == "same":
expires_in = (row["expires_at"] - row["issued_at"]) if row["expires_at"] else None
return self.issue(row["subject_kind"], row["subject_id"], row["recipient"], row["grant_type"],
expires_in=expires_in, actor=actor, destination=row["destination"], conn=conn,
_reissued_from=link_id)
def status(self, subject_kind: str, subject_id: str, *, conn: Any = None) -> list:
"""Every link to a subject, per person: seen (first/last/count) apart from used, and its state —
active, used, expired or revoked. Computed from the event rows, which are the record."""
now = _now()
with self._ctx(conn) as c:
rows = [_plain(r) for r in c.execute(select(links).where(and_(
links.c.issued_by_brick == self.brick, links.c.subject_kind == subject_kind,
links.c.subject_id == str(subject_id))))]
ids = [r["link_id"] for r in rows]
evs = [_plain(r) for r in c.execute(
select(events.c.link_id, events.c.kind, events.c.event_at).where(events.c.link_id.in_(ids)))] if ids else []
out = []
for r in sorted(rows, key=lambda r: (r["issued_at"], r["link_id"])):
mine = [e for e in evs if e["link_id"] == r["link_id"]]
seen = sorted(e["event_at"] for e in mine if e["kind"] == "seen")
used = [e["event_at"] for e in mine if e["kind"] == "used"]
used_at = used[0] if used else None
if r["revoked_at"] is not None:
state = "revoked"
elif used_at is not None and r["grant_type"] in ACTS:
state = "used"
elif r["expires_at"] is not None and r["expires_at"] <= now:
state = "expired"
else:
state = "active"
out.append({"link_id": r["link_id"], "recipient": r["recipient"], "grant": r["grant_type"],
"state": state, "issued_at": r["issued_at"], "expires_at": r["expires_at"],
"revoked_at": r["revoked_at"], "revoke_reason": r["revoke_reason"],
"first_seen_at": seen[0] if seen else None, "last_seen_at": seen[-1] if seen else None,
"seen_count": len(seen), "used_at": used_at,
"opened": any(e["kind"] == "opened" for e in mine),
"clicked": any(e["kind"] == "clicked" for e in mine),
"reissued_from": r["reissued_from"]})
return out
# ── what a handler returns ──────────────────────────────────────────────────────────────────
esc = html.escape
@dataclass
class Page:
"""A server-rendered page. `body` is HTML you built — escape every value with public.esc. Forms post to
the same address: <form method="post">. No script runs (the CSP forbids it); inline styles do."""
title: str
body: str
status: int = 200
@dataclass
class Download:
"""A file to hand to the holder — the frozen bytes of a signed copy, never an object URL."""
content: bytes
content_type: str
filename: str
@dataclass
class Request:
"""What a handler may know about the request. Never the token, never the headers wholesale."""
method: str
ip: Optional[str]
user_agent: Optional[str]
form: dict = field(default_factory=dict)
files: dict = field(default_factory=dict) # name -> (filename, content_type, bytes); needs python-multipart
@dataclass
class Handler:
brick: str
grant: str
subject_kind: str
render: Callable[[Link, Request], Any]
action: Optional[Callable[[Link, Request], Any]] = None
single_use: bool = True
@property
def name(self) -> str:
return self.render.__name__
def act(self, fn: Optional[Callable] = None, *, single_use: bool = True):
"""Declare the POST. single_use (the default) has the door claim the act before running it and
release it if the act fails; False leaves the act repeatable until the handler calls link.use()
(an intake form saved in parts, submitted at the end)."""
def register(f):
if self.grant not in ACTS:
raise ValueError(f"a {self.grant} link does not act — only {', '.join(ACTS)} do")
self.action, self.single_use = f, single_use
return f
return register(fn) if fn is not None else register
def __call__(self, link: Link, req: Request):
return self.render(link, req)
def handler(brick: Any, grant: str, subject_kind: str):
"""What `@BRICK.public(grant, subject_kind=...)` does: register the GET of a link of this grant and kind."""
grant = grant.upper()
if grant not in GRANTS or grant == "TRACK":
raise ValueError(f"a public handler's grant is one of {', '.join(g for g in GRANTS if g != 'TRACK')} "
f"(TRACK is served by the door itself), not {grant!r}")
def register(fn: Callable) -> Handler:
if any(h.grant == grant and h.subject_kind == subject_kind for h in brick.public_handlers):
raise ValueError(f"{brick.name} already has a {grant} handler for {subject_kind}")
h = Handler(brick.name, grant, subject_kind, fn)
brick.public_handlers.append(h)
return h
return register
# ── the door: resolve, record, dispatch ─────────────────────────────────────────────────────
NOT_FOUND_TEXT = os.environ.get(
"PUBLIC_NOT_FOUND_TEXT",
"This link no longer works. It may have expired or been replaced — ask the person who sent it for a new one.")
def _resolve(token: str, meta: Optional[Request]) -> Optional[Link]:
"""The live link for a token, or None — the same None for every reason, so the caller cannot differ."""
if not _TOKEN.fullmatch(token or ""):
return None
digest = hash_token(token)
with _connect() as c:
row = c.execute(select(links).where(links.c.token_hash == digest)).first()
if row is None:
return None
r = _plain(row)
if not hmac.compare_digest(r["token_hash"], digest):
return None
now = _now()
if r["revoked_at"] is not None or (r["expires_at"] is not None and r["expires_at"] <= now):
_event(c, r["link_id"], "refused", meta=meta, at=now,
detail="revoked" if r["revoked_at"] is not None else "expired")
return None
used = c.execute(select(events.c.event_at).where(and_(
events.c.link_id == r["link_id"], events.c.event_id == "used"))).first()
return Link(r["link_id"], r["subject_kind"], r["subject_id"], r["issued_by_brick"], r["recipient"],
r["grant_type"], r["issued_at"], r["expires_at"], _utc(used[0]) if used else None,
int(r["seen_count"] or 0), _hash=digest)
def _seen(link: Link, meta: Request) -> None:
now = _now()
with _connect() as c:
_event(c, link.link_id, "seen", meta=meta, at=now)
_project(update(links).where(links.c.token_hash == link._hash).values(
first_seen_at=func.coalesce(links.c.first_seen_at, now), last_seen_at=now,
seen_count=func.coalesce(links.c.seen_count, 0) + 1))
def _once(link: Link, kind: str, meta: Request) -> bool:
"""A once-only event (opened, clicked): True the first time."""
try:
with _connect() as c:
_event(c, link.link_id, kind, event_id=kind, meta=meta)
return True
except Exception as exc: # noqa: BLE001
if _lost_claim(exc):
return False
raise
class Door:
"""The resolved-link dispatcher behind the public app's routes. Pure functions of (token, request) →
a response tuple, run on a worker thread (the tables are reached through sync SQLAlchemy)."""
def __init__(self, handlers: list):
self.handlers = {(h.brick, h.grant, h.subject_kind): h for h in handlers}
def page(self, token: str, meta: Request):
from erp_platform.bricks import Refused
link = _resolve(token, meta)
if link is None or link.grant == "TRACK":
return NOT_FOUND
h = self.handlers.get((link.issued_by_brick, link.grant, link.subject_kind))
if h is None:
return NOT_FOUND
requestlog.note(door="public", use_case=f"public:{h.brick}:{h.name}", sub=f"public:{link.link_id}")
if meta.method in ("GET", "HEAD"):
if meta.method == "GET":
_seen(link, meta)
try:
return h.render(link, meta)
except Refused as r:
return NOT_FOUND if r.status == 404 else Page("Not available", f"<p>{esc(r.reason)}</p>", r.status)
if h.action is None or link.used_at is not None:
return NOT_FOUND
if h.single_use and not link.use():
return NOT_FOUND
try:
return h.action(link, meta)
except Refused as r:
if h.single_use:
link.release(f"refused {r.status}")
return NOT_FOUND if r.status == 404 else Page("Not done", f"<p>{esc(r.reason)}</p>", r.status)
except Exception as exc:
if h.single_use:
link.release(type(exc).__name__)
if _lost_claim(exc) and "erp_public_link_event" in str(exc):
return NOT_FOUND # a use(conn) claim lost at the handler's own COMMIT
raise
def pixel(self, token: str, meta: Request):
link = _resolve(token, meta)
if link is not None and link.grant == "TRACK" and meta.method == "GET":
requestlog.note(door="public", use_case="public:open", sub=f"public:{link.link_id}")
_once(link, "opened", meta)
_seen(link, meta)
return PIXEL # always the pixel: a broken image helps nobody
def click(self, token: str, meta: Request):
link = _resolve(token, meta)
if link is None or link.grant != "TRACK":
return NOT_FOUND
with _connect() as c:
dest = c.execute(select(links.c.destination).where(links.c.token_hash == link._hash)).scalar()
if not dest or not _allowed_destination(dest): # checked again: the allow-list may have narrowed
return NOT_FOUND
requestlog.note(door="public", use_case="public:click", sub=f"public:{link.link_id}")
if meta.method == "GET":
if _once(link, "clicked", meta):
_project(update(links).where(links.c.token_hash == link._hash).values(used_at=_now()))
_seen(link, meta)
return Redirect(dest)
@dataclass
class Redirect:
location: str
PIXEL_GIF = (b"GIF89a\x01\x00\x01\x00\x80\x00\x00\x00\x00\x00\xff\xff\xff!\xf9\x04\x01\x00\x00\x00\x00"
b",\x00\x00\x00\x00\x01\x00\x01\x00\x00\x02\x02D\x01\x00;")
PIXEL = Download(PIXEL_GIF, "image/gif", "")
NOT_FOUND = Page("Link not available", f"<p>{esc(NOT_FOUND_TEXT)}</p>", 404)
# ── the app ─────────────────────────────────────────────────────────────────────────────────
# No script, no frames, no fetches, nothing from elsewhere; inline style only; a form may post only back here.
CSP = ("default-src 'none'; style-src 'unsafe-inline'; img-src 'self' data:; form-action 'self'; "
"frame-ancestors 'none'; base-uri 'none'")
HEADERS = {
"content-security-policy": CSP,
"x-robots-tag": "noindex, nofollow",
"cache-control": "no-store",
"referrer-policy": "no-referrer", # the token is in the path: never hand it to the next site
"x-content-type-options": "nosniff",
"x-frame-options": "DENY",
}
ROUTES = ("/healthz", "/readyz", "/p/{token}", "/o/{token}", "/c/{token}")
STYLE = ("body{font:16px/1.5 system-ui,sans-serif;max-width:40rem;margin:2rem auto;padding:0 1rem;color:#1a1a1a}"
"h1{font-size:1.4rem}input,textarea,button{font:inherit}button{padding:.5rem 1rem}")
MAX_FORM_BYTES = int(os.environ.get("PUBLIC_MAX_FORM_BYTES", str(64 * 1024)))
MAX_UPLOAD_BYTES = int(os.environ.get("PUBLIC_MAX_UPLOAD_BYTES", str(10 * 1024 * 1024)))
def document(page: Page) -> bytes:
return (f'<!doctype html><html lang="en"><head><meta charset="utf-8">'
f'<meta name="viewport" content="width=device-width,initial-scale=1">'
f'<meta name="robots" content="noindex,nofollow"><title>{esc(page.title)}</title>'
f"<style>{STYLE}</style></head><body><main><h1>{esc(page.title)}</h1>{page.body}</main></body></html>"
).encode()
class _SecurityHeaders:
"""Pure ASGI: the strict headers on EVERY response the public app sends — pages, 404s, health, a 500."""
def __init__(self, app):
self.app = app
async def __call__(self, scope, receive, send):
if scope["type"] != "http":
return await self.app(scope, receive, send)
async def send_strict(message):
if message["type"] == "http.response.start":
names = {k.encode() for k in HEADERS}
headers = [(k, v) for k, v in message.get("headers", []) if k.lower() not in names]
message = {**message, "headers": headers + [(k.encode(), v.encode()) for k, v in HEADERS.items()]}
await send(message)
await self.app(scope, receive, send_strict)
def app(handlers: list, *, name: str = "public", version: str = "0.0.0"):
"""The public app: ONLY the link routes, /healthz and /readyz. No admin route, no MCP, no OpenAPI, no
migration. Serve it with access logging OFF (erp-compose serve --public does): an access log line carries
the raw path, and the raw path carries the token."""
import anyio
from fastapi import FastAPI, Request as FRequest
from fastapi.responses import JSONResponse, RedirectResponse, Response
from starlette.exceptions import HTTPException as StarletteHTTPException
door = Door(handlers)
api = FastAPI(title=name, version=version, docs_url=None, redoc_url=None, openapi_url=None)
requestlog.configure()
def respond(out) -> Response:
if isinstance(out, Redirect):
return RedirectResponse(out.location, status_code=303)
if isinstance(out, Download):
headers = {"content-disposition": f'attachment; filename="{out.filename}"'} if out.filename else {}
return Response(out.content, media_type=out.content_type, headers=headers)
if not isinstance(out, Page):
raise TypeError(f"a public handler returns Page or Download, not {type(out).__name__}")
return Response(document(out), status_code=out.status, media_type="text/html; charset=utf-8")
def meta(request: FRequest, form=None, files=None) -> Request:
fwd = request.headers.get("x-forwarded-for", "")
ip = fwd.split(",")[0].strip() if fwd else (request.client.host if request.client else None)
return Request(request.method, ip or None, request.headers.get("user-agent"), form or {}, files or {})
async def read_form(request: FRequest):
ctype = request.headers.get("content-type", "")
size = int(request.headers.get("content-length") or 0)
if ctype.startswith("multipart/form-data"):
if size > MAX_UPLOAD_BYTES:
return None, None, Page("Too large", "<p>That upload is too large.</p>", 413)
try:
data = await request.form(max_part_size=MAX_UPLOAD_BYTES)
except AssertionError: # starlette without python-multipart
return None, None, Page("Not supported", "<p>Uploads are not enabled here.</p>", 415)
form, files = {}, {}
for k, v in data.multi_items():
if hasattr(v, "read"):
files[k] = (v.filename, v.content_type, await v.read())
else:
form[k] = v
return form, files, None
body = await request.body()
if len(body) > MAX_FORM_BYTES:
return None, None, Page("Too large", "<p>That form is too large.</p>", 413)
return dict(parse_qsl(body.decode("utf-8", "replace"), keep_blank_values=True)), {}, None
async def link_page(request: FRequest, token: str):
form = files = None
if request.method == "POST":
form, files, problem = await read_form(request)
if problem is not None:
return respond(problem)
m = meta(request, form, files)
return respond(await anyio.to_thread.run_sync(lambda: door.page(token, m)))
async def open_pixel(request: FRequest, token: str):
m = meta(request)
return respond(await anyio.to_thread.run_sync(lambda: door.pixel(token, m)))
async def click(request: FRequest, token: str):
m = meta(request)
return respond(await anyio.to_thread.run_sync(lambda: door.click(token, m)))
api.add_api_route("/p/{token}", link_page, methods=["GET", "POST"], include_in_schema=False)
api.add_api_route("/o/{token}", open_pixel, methods=["GET"], include_in_schema=False)
api.add_api_route("/c/{token}", click, methods=["GET"], include_in_schema=False)
@api.get("/healthz", include_in_schema=False)
def healthz():
"""Liveness: the process is up. No identity, no database."""
return {"ok": True, "name": name, "version": version, "door": "public"}
@api.get("/readyz", include_in_schema=False)
def readyz():
"""Ready: the link table answers. 503 (not a crash) when it does not — a missing table means the
composition was not migrated, which this app must never do itself."""
try:
with _connect() as c:
c.execute(select(links.c.link_id).limit(1)).all()
return {"ok": True}
except Exception as exc: # noqa: BLE001 — reported by type only
return JSONResponse({"ok": False, "error": type(exc).__name__}, status_code=503)
@api.exception_handler(StarletteHTTPException)
async def same_404(_request, exc):
# A path that matches nothing, or a method a route does not take: the same page as a dead link.
return respond(NOT_FOUND)
# Added last = outermost: the strict headers land even on the 500 requestlog answers for a crash.
api.add_middleware(requestlog.RequestLog)
api.add_middleware(_SecurityHeaders)
return api