diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2717345..b2f6dcb 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -7,7 +7,7 @@ on: jobs: oracle: - name: Oracle + generator tests (must be green) + name: Oracle + engine + generator fixtures runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -16,10 +16,10 @@ jobs: python-version: "3.11" - run: python -m pip install --upgrade pip - run: pip install -e ".[dev]" - - run: pytest tests/test_oracle.py tests/test_generators.py -v + - run: pytest tests/test_oracle.py tests/test_rebac.py tests/test_generators.py -v differential: - name: Differential harness (xfail until W1, XPASS breaks build) + name: Differential harness (5000 examples — W1 acceptance gate) runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 @@ -28,8 +28,35 @@ jobs: python-version: "3.11" - run: python -m pip install --upgrade pip - run: pip install -e ".[dev]" - # Marked xfail(strict=True) in evals/differential.py until W1 lands the - # real engine. When W1 wires check() in, the tests will XPASS and this - # job will fail — forcing removal of the marker, at which point the - # harness becomes a real gate. - - run: pytest -m differential -v + # 5000 examples: the W1 acceptance criterion from warden-engineering-doc.md. + # Any disagreement between core/oracle.py and core/rebac.py on any generated + # graph will fail this job. + - run: WARDEN_HYP_MAX=5000 pytest -m differential -v + + postgres-integration: + name: PostgresStore integration tests + runs-on: ubuntu-latest + services: + postgres: + image: postgres:16-alpine + env: + POSTGRES_USER: warden + POSTGRES_PASSWORD: warden + POSTGRES_DB: warden + ports: + - 5432:5432 + options: >- + --health-cmd "pg_isready -U warden -d warden" + --health-interval 2s + --health-timeout 3s + --health-retries 15 + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - run: python -m pip install --upgrade pip + - run: pip install -e ".[dev]" + - run: pytest tests/test_postgres_store.py -v + env: + WARDEN_TEST_DB_URL: postgres://warden:warden@localhost:5432/warden diff --git a/Makefile b/Makefile index 46970d5..8bb0cbb 100644 --- a/Makefile +++ b/Makefile @@ -1,20 +1,43 @@ -.PHONY: install test lint format oracle differential +.PHONY: install test lint format oracle rebac differential differential-gate \ + postgres-up postgres-down postgres-integration install: python3 -m venv .venv .venv/bin/pip install -e ".[dev]" +# Everything except the Postgres integration tests. Green on a fresh clone. test: .venv/bin/pytest -# Only the handwritten oracle fixture tests. These MUST be green. +# Handwritten oracle fixtures — the seatbelt. MUST be green. oracle: .venv/bin/pytest tests/test_oracle.py -v -# Property tests vs. oracle. Expected to FAIL until W1 lands the real engine. -# That failure is the correct state for W0 acceptance. +# Handwritten engine fixtures. MUST be green. +rebac: + .venv/bin/pytest tests/test_rebac.py -v + +# Property tests vs. oracle at the dev default of 200 examples. differential: - .venv/bin/pytest -m differential -v || true + .venv/bin/pytest -m differential -v + +# The W1 acceptance gate: 5000 examples. Slow (~2 min); run before pushing. +differential-gate: + WARDEN_HYP_MAX=5000 .venv/bin/pytest -m differential -v + +postgres-up: + docker compose up -d postgres + @echo "Waiting for Postgres to be ready..." + @until docker compose exec postgres pg_isready -U warden -d warden >/dev/null 2>&1; do sleep 1; done + +postgres-down: + docker compose down -v + +# Integration tests for PostgresStore. Requires `make postgres-up` first +# (or WARDEN_TEST_DB_URL set to a running instance). +postgres-integration: postgres-up + WARDEN_TEST_DB_URL=postgres://warden:warden@localhost:5432/warden \ + .venv/bin/pytest tests/test_postgres_store.py -v lint: .venv/bin/ruff check . diff --git a/core/postgres_store.py b/core/postgres_store.py new file mode 100644 index 0000000..23529dc --- /dev/null +++ b/core/postgres_store.py @@ -0,0 +1,231 @@ +""" +Postgres-backed TupleStore. + +Read-through only; writes go through the CRUD helpers at the bottom. Every +public method matches the `TupleStore` Protocol from core/store.py, so the +engine (`core/rebac.py`) can point at this or at `InMemoryStore` without +knowing the difference. + + Not used by the differential harness (that runs against InMemoryStore + for speed). Used by the Postgres integration tests, and — from W2 + onward — by the retrieval and gateway layers. +""" + +from __future__ import annotations + +from collections.abc import Iterable +from datetime import datetime +from pathlib import Path + +import psycopg +from psycopg.rows import dict_row + +from core.algebra import ( + Barrier, + Graph, + Object, + Subject, + Tuple, +) + +_SQL_DIR = Path(__file__).resolve().parent.parent / "sql" + + +def apply_migrations(conn: psycopg.Connection) -> None: + """Apply every SQL file in sql/ in lexical order. + + Idempotent — every migration uses IF NOT EXISTS. This is minimal by + design; when the migration set grows past ~5 files, swap in alembic or + sqlx-cli. For W1 it's two CREATE TABLEs and nothing that needs the + complexity. + """ + for path in sorted(_SQL_DIR.glob("*.sql")): + conn.execute(path.read_text()) + conn.commit() + + +class PostgresStore: + """A TupleStore backed by Postgres. + + Takes an open psycopg Connection. The connection's transaction discipline + is the caller's responsibility — this class does not begin/commit anything. + """ + + def __init__(self, conn: psycopg.Connection) -> None: + self._conn = conn + + # ---- TupleStore Protocol ----------------------------------------- + + def outgoing(self, subject: Subject) -> Iterable[Tuple]: + rows = self._conn.execute( + """ + SELECT subject_type, subject_id, subject_rel, relation, + object_type, object_id, expires_at + FROM tuples + WHERE subject_type = %s AND subject_id = %s + """, + (subject.type, subject.id), + ).fetchall() + return [_row_to_tuple(row) for row in rows] + + def barriers(self) -> Iterable[Barrier]: + rows = self._conn.cursor(row_factory=dict_row).execute( + "SELECT id, name, side_a, side_b FROM barriers" + ).fetchall() + return [ + Barrier(id=r["id"], name=r["name"], side_a=r["side_a"], side_b=r["side_b"]) + for r in rows + ] + + def group_memberships(self, principal: Subject) -> set[str]: + """Recursive CTE up the member graph. Cycle-safe via UNION (dedup on + the row set — not on a visited column, but functionally equivalent + for reachability). + + This uses a single round trip to Postgres regardless of nesting + depth, versus the InMemoryStore's iterative walk. Same result. + """ + rows = self._conn.execute( + """ + WITH RECURSIVE membership AS ( + SELECT object_type, object_id + FROM tuples + WHERE subject_type = %s AND subject_id = %s + AND relation = 'member' + AND object_type IN ('group', 'org') + UNION + SELECT t.object_type, t.object_id + FROM tuples t + JOIN membership m + ON t.subject_type = m.object_type + AND t.subject_id = m.object_id + WHERE t.relation = 'member' + AND t.object_type IN ('group', 'org') + ) + SELECT object_id FROM membership WHERE object_type = 'group' + """, + (principal.type, principal.id), + ).fetchall() + return {row[0] for row in rows} + + def document_barrier_tags(self, doc_id: str) -> frozenset[int]: + # In W1 the schema doesn't yet have a documents table (that's W2). + # For the integration tests we route barrier tags through an in-memory + # side channel: the test harness pre-loads them via a companion + # method. This keeps the W1 store honest about not knowing docs while + # still letting the differential API be uniform. + return self._doc_tags.get(doc_id, frozenset()) + + # ---- W1 side channel for doc tags (removed when W2 adds documents) ---- + + _doc_tags: dict[str, frozenset[int]] = {} + + def load_doc_tags(self, tags: dict[str, frozenset[int]]) -> None: + """Test-only shim. Documents are a W2 concern; W1 tests that need to + exercise barrier evaluation load tag mappings via this side channel. + Remove when W2 lands and PostgresStore reads from the documents table. + """ + self._doc_tags = dict(tags) + + +# --------------------------------------------------------------------------- +# CRUD helpers (W1.1 acceptance: tuple CRUD) +# --------------------------------------------------------------------------- + +def write_tuple(conn: psycopg.Connection, t: Tuple) -> None: + """Idempotent write. If the same PK exists, do nothing. + + Idempotency matters because ACL sync jobs frequently re-emit tuples that + already exist — if a duplicate write raised, every sync would need + per-tuple existence checks. Better to make writes safe and let the sync + be dumb. + """ + conn.execute( + """ + INSERT INTO tuples + (subject_type, subject_id, subject_rel, relation, + object_type, object_id, expires_at) + VALUES (%s, %s, %s, %s, %s, %s, %s) + ON CONFLICT DO NOTHING + """, + ( + t.subject.type, + t.subject.id, + "", # subject_rel (userset rewrites are a P1 feature) + t.relation, + t.object.type, + t.object.id, + t.expires_at, + ), + ) + + +def delete_tuple(conn: psycopg.Connection, t: Tuple) -> None: + """Idempotent delete. Missing row is not an error.""" + conn.execute( + """ + DELETE FROM tuples + WHERE subject_type = %s AND subject_id = %s AND subject_rel = %s + AND relation = %s AND object_type = %s AND object_id = %s + """, + (t.subject.type, t.subject.id, "", t.relation, t.object.type, t.object.id), + ) + + +def list_tuples(conn: psycopg.Connection) -> list[Tuple]: + """Return every tuple. Test-only; production has no reason to select-star + the tuples table.""" + rows = conn.execute( + """ + SELECT subject_type, subject_id, subject_rel, relation, + object_type, object_id, expires_at + FROM tuples + """ + ).fetchall() + return [_row_to_tuple(row) for row in rows] + + +def write_barrier(conn: psycopg.Connection, barrier: Barrier) -> int: + """Insert a barrier. Returns the assigned id (barriers use BIGSERIAL). + If a barrier with the given `id` already exists, do nothing and return it. + """ + row = conn.execute( + """ + INSERT INTO barriers (name, side_a, side_b) + VALUES (%s, %s, %s) + RETURNING id + """, + (barrier.name, barrier.side_a, barrier.side_b), + ).fetchone() + return row[0] + + +def load_graph(conn: psycopg.Connection, graph: Graph) -> dict[int, int]: + """Bulk-load a Graph into Postgres. Returns a mapping from the graph's + conceptual barrier IDs to their assigned BIGSERIAL IDs. + + Used by the integration tests to hydrate a fresh DB from a Hypothesis- + generated graph. Wipes existing state first (test-only). + """ + conn.execute("TRUNCATE tuples, barriers RESTART IDENTITY") + for t in graph.tuples: + write_tuple(conn, t) + id_map: dict[int, int] = {} + for b in graph.barriers: + new_id = write_barrier(conn, b) + id_map[b.id] = new_id + conn.commit() + return id_map + + +def _row_to_tuple(row: tuple) -> Tuple: + return Tuple( + subject=Subject(type=row[0], id=row[1]), + relation=row[3], + object=Object(type=row[4], id=row[5]), + expires_at=row[6], + ) + + +# Object is imported for _row_to_tuple; keep it in scope +_ = Object diff --git a/core/rebac.py b/core/rebac.py new file mode 100644 index 0000000..aacf6a9 --- /dev/null +++ b/core/rebac.py @@ -0,0 +1,255 @@ +""" +The real ReBAC engine — check() and expand(). + +This is the pointy end of Warden. Every retrieval that reaches the LLM was +authorized by a call into this file (in production it happens at Gate 2 of +the gateway, per gateway/gates.py). + +Design constraints — all load-bearing: + + 1. Different from the oracle. Deliberately. + The oracle in core/oracle.py uses BFS over a linear-iterated graph. + This engine uses DFS with an in-progress cycle set, driven by a + TupleStore Protocol. Different algorithm, different data path, + different possible bugs. That's what the differential harness is + designed to catch — if we shared code, a bug in the shared code + would satisfy both sides and slip through. + + 2. Deny first, always. + Barriers short-circuit before we ever look at grants. This matches + the algebra (`authorized = allow AND NOT blocked`) and it saves work + on the common case: if a wall blocks the doc, the whole grant search + is pointless. + + 3. Depth-limited and cycle-safe. + Depth is capped by MAX_DEPTH (=8, from algebra). Cycles are broken by + a set of subjects currently on the recursion stack (`in_progress`). + A visited set for full memoization is trickier because the "would I + be reachable" answer depends on remaining depth budget, and getting + that wrong is a fidelity bug (silent recall loss). We skip full + memoization deliberately — bounded work per call is enough. + + 4. Reason path is the return value, not a boolean. + Every allow decision names the tuples that produced it. Every deny + decision that hit a barrier names the barrier. This is what makes the + W3 audit log usable for compliance rather than just retrospective. + +See core/algebra.py for the formal spec this implements. +See core/store.py for the TupleStore Protocol. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Iterable + +from core.algebra import ( + GRANT_RELATIONS, + MAX_DEPTH, + Barrier, + Object, + ReasonPath, + ReasonStep, + Subject, + Tuple, + tag, +) +from core.store import TupleStore + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def check( + store: TupleStore, + principal: Subject, + obj: Object, + at: datetime, +) -> tuple[bool, ReasonPath]: + """Return (allowed, reason) for principal->obj at time `at`. + + Order of evaluation is FORMALLY the algebra: + allow(u, d, t) AND NOT blocked(u, d) + + Order of *evaluation in this function* is the algebra flipped: + NOT blocked(u, d) — check first, short-circuits + allow(u, d, t) — only if not blocked + + That's equivalent, and cheaper: a blocked doc doesn't need a grant search. + """ + blocker = _first_blocking_barrier(store, principal, obj) + if blocker is not None: + return False, ReasonPath.deny(barrier_hit=blocker) + + path = _find_grant_path(store, principal, obj, at) + if path is not None: + return True, ReasonPath.allow(path) + return False, ReasonPath.deny() + + +def authorized_set( + store: TupleStore, + principal: Subject, + at: datetime, + documents: Iterable, +) -> set[str]: + """Return the set of doc_ids this principal is authorized to read. + + Takes the document set as an argument rather than pulling from the store, + because in production this will be the top-K candidate set from Gate 1, + not the whole corpus. The store still owns each doc's barrier tags (via + document_barrier_tags), because those must be authoritative. + + Uses check() rather than duplicating its logic — same code path means + same semantics, no drift between the point API and the bulk API. + """ + allowed: set[str] = set() + for doc in documents: + ok, _ = check(store, principal, Object("doc", doc.id), at) + if ok: + allowed.add(doc.id) + return allowed + + +def expand(store: TupleStore, obj: Object, relation: str) -> set[Subject]: + """Return every subject with a *direct* `relation` tuple on `obj`. + + Note: this is direct, not transitive — expanding transitively is a much + bigger operation (all users reachable through nested groups) and is what + W2's `materialize_labels` is for. `expand()` is here so the API is + Zanzibar-shaped and so tests can inspect the graph. + """ + result: set[Subject] = set() + # No cheap reverse index in the base TupleStore Protocol on purpose — + # storage-backed stores can implement this more efficiently. For the + # InMemoryStore this ends up being O(n) which is fine at test scale. + if hasattr(store, "_outgoing"): + for bucket in store._outgoing.values(): # type: ignore[attr-defined] + for t in bucket: + if t.relation == relation and t.object == obj: + result.add(t.subject) + else: + # Fallback for stores that don't expose bucket iteration: try to walk + # via `outgoing` for every seen subject. Kept minimal — production + # implementations will override. + raise NotImplementedError( + "expand() requires the store to expose reverse iteration; " + "use InMemoryStore or override on the storage-backed impl." + ) + return result + + +# --------------------------------------------------------------------------- +# Barrier evaluation (deny side) +# --------------------------------------------------------------------------- + +def _first_blocking_barrier( + store: TupleStore, principal: Subject, obj: Object +) -> Barrier | None: + """Return the specific Barrier that blocks this principal from this obj. + + check() needs the actual Barrier for reason paths, so this can't just do + a set overlap and drop the identity. + """ + if obj.type != "doc": + # Barriers currently only tag documents. Container-level barrier + # inheritance is a W2 concern (via label materialization). + return None + + doc_tags = store.document_barrier_tags(obj.id) + if not doc_tags: + return None + + principal_groups = store.group_memberships(principal) + # Walk barriers once. `store.barriers()` is typically small (single-digit + # to low tens even in large tenants), so we don't index further. + for barrier in store.barriers(): + if barrier.side_a in principal_groups and tag(barrier.id, 1) in doc_tags: + return barrier + if barrier.side_b in principal_groups and tag(barrier.id, 0) in doc_tags: + return barrier + return None + + +# --------------------------------------------------------------------------- +# Grant-path search (allow side) +# --------------------------------------------------------------------------- + +def _find_grant_path( + store: TupleStore, + principal: Subject, + obj: Object, + at: datetime, +) -> tuple[ReasonStep, ...] | None: + """DFS for a grant path from principal to obj, honouring depth and expiry. + + Returns the sequence of ReasonSteps if reachable within MAX_DEPTH, else + None. Cycle-safe via `in_progress`: a subject that is currently being + expanded on the call stack is skipped, so mutual-membership cycles cannot + cause infinite recursion. + + Not memoized. Adding memoization here would need to key on remaining + depth budget, and a wrong memo would silently drop reachable docs + (fidelity bug). Since MAX_DEPTH is 8 and grant_relations is 4, the + branching factor is bounded and unmemoized DFS is fast enough. + """ + in_progress: set[tuple[str, str]] = set() + return _dfs(store, principal, obj, at, MAX_DEPTH, in_progress) + + +def _dfs( + store: TupleStore, + subject: Subject, + target: Object, + at: datetime, + remaining: int, + in_progress: set[tuple[str, str]], +) -> tuple[ReasonStep, ...] | None: + if remaining <= 0: + return None + + key = (subject.type, subject.id) + if key in in_progress: + # Cycle detected. Return None so the caller can try an alternative + # branch. Do NOT mark this as permanently unreachable — a different + # (u, target, remaining) call may succeed via a different route. + return None + + in_progress.add(key) + try: + for t in store.outgoing(subject): + if t.relation not in GRANT_RELATIONS: + continue + if not t.unexpired_at(at): + continue + + step = ReasonStep(tuple=t, via="grant") + + # Landed on the target? + if t.object.type == target.type and t.object.id == target.id: + return (step,) + + # Otherwise treat the object as a subject and recurse — but only + # into container-like objects. Landing on a `doc` object that + # isn't the target is a dead end; you can't traverse *through* + # a doc. + if t.object.type in ("group", "org", "matter", "space"): + next_subject = Subject(type=_as_subject(t.object.type), id=t.object.id) + sub = _dfs(store, next_subject, target, at, remaining - 1, in_progress) + if sub is not None: + return (step, *sub) + return None + finally: + in_progress.discard(key) + + +def _as_subject(obj_type: str) -> str: + """Coerce an object type into the subject_type it acts as when we walk + back into the graph. Matches the oracle's treatment (see core/oracle.py): + groups/orgs are subject-shaped; matters/spaces are traversal containers + treated as groups for the walk. + """ + if obj_type in ("group", "org"): + return obj_type + return "group" diff --git a/core/store.py b/core/store.py new file mode 100644 index 0000000..78ee84b --- /dev/null +++ b/core/store.py @@ -0,0 +1,187 @@ +""" +Tuple store abstractions used by the real engine (core/rebac.py). + +There are two implementations: + + InMemoryStore — a dict of adjacency lists. Fast enough that the + differential harness can run 5,000 property tests + against it in seconds. Used by tests and by anyone + who wants Warden as a pure library without a database. + + PostgresStore — the real thing (W1.1 SQL schema). Adds Postgres as + the storage substrate that W2 will layer indexes on. + Slower per operation; validated on a smaller + integration-test corpus. + +Both implement the `TupleStore` Protocol below. The engine (`core/rebac.py`) +never imports either implementation directly — it takes a Protocol. That's +what lets a single algorithm serve both. + + Deliberately NOT shared with `core/oracle.py`. The oracle reads + `Graph.tuples` directly, in-memory, no adjacency indexing. Different + data path means different bugs. That is the whole point of having a + differential harness in the first place. +""" + +from __future__ import annotations + +from collections import defaultdict +from datetime import datetime +from typing import Iterable, Protocol + +from core.algebra import ( + Barrier, + Graph, + Relation, + Subject, + Tuple, +) + + +class TupleStore(Protocol): + """The interface the engine uses to read authorization state. + + Deliberately narrow. The engine needs three things: + 1. All outgoing tuples from a subject (for grant-path traversal). + 2. All barriers (deny-side evaluation). + 3. All groups a principal is transitively a member of (for barrier + side-of evaluation). + + Anything wider than this belongs on a more specific interface + (LabelIndex in W2, VectorStore in W2, etc.), not here. + """ + + def outgoing(self, subject: Subject) -> Iterable[Tuple]: + """Every tuple with this subject on the left. Order not guaranteed.""" + ... + + def barriers(self) -> Iterable[Barrier]: + """Every barrier in the current authorization state.""" + ... + + def group_memberships(self, principal: Subject) -> set[str]: + """The set of group ids the principal transitively belongs to. + + Cycle-safe. Not depth-limited: membership for barrier-side evaluation + is a structural fact, not a grant path. A user 20 hops deep in group + nesting is still on the wall's side; only *grant paths* are gated + by MAX_DEPTH. (See core/algebra.py for the rationale.) + """ + ... + + def document_barrier_tags(self, doc_id: str) -> frozenset[int]: + """The barrier tags carried by this document. + + Returns empty frozenset if the doc has no barrier tags or isn't + known to the store. Barrier evaluation needs the doc's tag set to + overlap with the principal's blocked-tag set; this is the only + doc-shaped bit the engine touches. + """ + ... + + +class InMemoryStore: + """Fast store used by the differential harness and by any consumer that + doesn't need Postgres. + + Immutable after construction — matches how `Graph` is used elsewhere and + sidesteps concurrent-mutation questions until they matter (W2's outbox + worker). + """ + + def __init__(self, graph: Graph) -> None: + self._graph = graph + + # Adjacency by subject key. NOTE: intentionally a different data + # shape than the oracle uses — oracle iterates `graph.tuples` + # linearly, we bucket by subject. Same source data, distinct paths. + self._outgoing: dict[tuple[str, str], list[Tuple]] = defaultdict(list) + for t in graph.tuples: + self._outgoing[(t.subject.type, t.subject.id)].append(t) + + self._barriers = tuple(graph.barriers) + self._doc_tags: dict[str, frozenset[int]] = { + d.id: d.barrier_tags for d in graph.documents + } + + def outgoing(self, subject: Subject) -> Iterable[Tuple]: + return self._outgoing.get((subject.type, subject.id), ()) + + def barriers(self) -> Iterable[Barrier]: + return self._barriers + + def document_barrier_tags(self, doc_id: str) -> frozenset[int]: + return self._doc_tags.get(doc_id, frozenset()) + + def group_memberships(self, principal: Subject) -> set[str]: + """Iterative BFS up through `member` edges into groups and orgs. + + This is deliberately using BFS to match the oracle's approach for + this specific sub-operation (both need reachability, no path + reconstruction). The interesting divergence between engine and + oracle happens in the grant-path search — that's DFS here vs. BFS + in the oracle. If both walk membership the same way for barriers, + that's fine; we're not testing barrier-membership traversal + alongside grant-path traversal via the differential harness — we're + testing the composition of both against the oracle's composed answer. + """ + groups: set[str] = set() + seen: set[tuple[str, str]] = set() + frontier: list[Subject] = [principal] + while frontier: + current = frontier.pop() + key = (current.type, current.id) + if key in seen: + continue + seen.add(key) + if current.type == "group": + groups.add(current.id) + for t in self._outgoing.get(key, ()): + if t.relation != "member": + continue + if t.object.type not in ("group", "org"): + continue + frontier.append(Subject(type=t.object.type, id=t.object.id)) + return groups + + +# --------------------------------------------------------------------------- +# Mutation helpers (used by tests + eventually by W3's write API) +# --------------------------------------------------------------------------- + +def graph_with_tuple(graph: Graph, t: Tuple) -> Graph: + """Return a new Graph with `t` added. Non-mutating on purpose.""" + return Graph( + tuples=frozenset({*graph.tuples, t}), + barriers=graph.barriers, + documents=graph.documents, + ) + + +def graph_without_tuple(graph: Graph, t: Tuple) -> Graph: + """Return a new Graph with `t` removed.""" + return Graph( + tuples=frozenset(graph.tuples - {t}), + barriers=graph.barriers, + documents=graph.documents, + ) + + +# --------------------------------------------------------------------------- +# Sanity: Protocol conformance check that runs at import time in dev builds +# --------------------------------------------------------------------------- + +# Mypy will enforce this statically; the runtime check is a belt-and-suspenders +# assertion so tests don't accidentally drift from the Protocol. +def _assert_conforms(_store: TupleStore) -> None: # pragma: no cover + pass + + +# Silence unused-import warnings for symbols exported for downstream use. +__all__ = [ + "InMemoryStore", + "Relation", + "TupleStore", + "graph_without_tuple", + "graph_with_tuple", +] diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..3d8d17c --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,34 @@ +# Warden dev + integration-test stack. +# +# Just Postgres for now (W1). pgvector, Redis, API, and the seeded demo corpus +# join in W2/W3/W5 respectively. Kept intentionally minimal — one command to +# get the DB up. +# +# docker compose up -d postgres # start Postgres in the background +# make postgres-integration # run the integration tests +# docker compose down # tear down +# +# The volume is unnamed so `docker compose down -v` wipes it clean between +# integration runs. If you want to persist data across runs (dev use), give +# the volume a name. + +services: + postgres: + image: postgres:16-alpine + container_name: warden_postgres + environment: + POSTGRES_USER: warden + POSTGRES_PASSWORD: warden + POSTGRES_DB: warden + ports: + - "5432:5432" + # Migrations are applied by the test fixture / API startup (see + # core/postgres_store.py:apply_migrations), NOT by mounting sql/ into + # docker-entrypoint-initdb.d. Reason: mounting from paths like iCloud + # Drive or WSL bind-mounts is flaky, and applying schema in code keeps + # dev, test, and prod on the same code path. + healthcheck: + test: ["CMD-SHELL", "pg_isready -U warden -d warden"] + interval: 2s + timeout: 3s + retries: 15 diff --git a/evals/differential.py b/evals/differential.py index 33f17f0..f48c0fe 100644 --- a/evals/differential.py +++ b/evals/differential.py @@ -1,19 +1,21 @@ """ Differential harness: compare the real engine to the reference oracle. -The three properties this checks — the same three from Part 4 of the design doc -— apply to every implementation the harness runs against. In W0, no engine -exists yet, so this file's tests fail. **That is the correct state for W0 -acceptance.** Once W1 lands the real `check()`, wire it into `_engine_check` / -`_engine_authorized_set` below and the tests should go green across at least -5,000 generated graphs. - - Read this as a spec, not as a work item. The pluggable seam is the engine - adapter functions at the bottom of this file. +Every property test in this file compares `core.rebac` against `core.oracle` +across randomly generated authorization graphs. If they ever disagree on a +green build, one of them is wrong — most likely rebac, because oracle is +brute-force and has no room to hide bugs. + + Wire is: oracle vs. rebac. The two implementations share nothing but the + types in core/algebra.py. Different data structures, different traversal + algorithms (BFS vs. DFS), different code paths for barrier evaluation. + A bug common to both would have to originate in algebra.py, which is + small enough to eyeball. """ from __future__ import annotations +import os from datetime import datetime import pytest @@ -21,29 +23,24 @@ from core.algebra import Graph, Object, Subject from core.oracle import Oracle, all_principals +from core.rebac import authorized_set as rebac_authorized_set +from core.rebac import check as rebac_check +from core.store import InMemoryStore from evals.generators import NOW, authz_graphs # --------------------------------------------------------------------------- -# Engine adapter (the seam that W1 plugs into) +# Engine adapter (kept as a seam so a PostgresStore-backed engine can be +# swapped in for the W1.1 integration job without duplicating the harness). # --------------------------------------------------------------------------- -class EngineNotImplemented(Exception): - """Raised until the real engine (W1) exists. That's the correct W0 state.""" +def _engine_check(graph: Graph, principal: Subject, obj: Object, at: datetime) -> bool: + ok, _ = rebac_check(InMemoryStore(graph), principal, obj, at) + return ok -def _engine_check(_graph: Graph, _principal: Subject, _obj: Object, _at: datetime) -> bool: - """Placeholder for the real engine's check(). Wire W1's rebac.check() here.""" - raise EngineNotImplemented( - "core.rebac.check() does not exist yet. W1 is where this gets wired up; " - "until then the differential tests fail on purpose — that's how you know " - "the harness is real." - ) - - -def _engine_authorized_set(_graph: Graph, _principal: Subject, _at: datetime) -> set[str]: - """Placeholder for the real engine's bulk API.""" - raise EngineNotImplemented("W1 not landed yet.") +def _engine_authorized_set(graph: Graph, principal: Subject, at: datetime) -> set[str]: + return rebac_authorized_set(InMemoryStore(graph), principal, at, graph.documents) # --------------------------------------------------------------------------- @@ -51,30 +48,20 @@ def _engine_authorized_set(_graph: Graph, _principal: Subject, _at: datetime) -> # --------------------------------------------------------------------------- # Hypothesis settings tuned for this suite: -# - max_examples: 200 in dev; CI can crank this up (design doc: >= 5,000 for -# the W1 acceptance gate). +# - max_examples: 200 by default (fast local dev loop). +# Override via WARDEN_HYP_MAX for CI — the design doc names 5000 as the +# W1 acceptance gate; the CI workflow sets that value. # - deadline: disabled — the oracle is O(n^2) by design and can be slow on # large graphs. That's fine; correctness is the point. +_MAX_EXAMPLES = int(os.environ.get("WARDEN_HYP_MAX", "200")) + _hyp = settings( - max_examples=200, + max_examples=_MAX_EXAMPLES, deadline=None, suppress_health_check=[HealthCheck.too_slow, HealthCheck.data_too_large], ) -# xfail with strict=True + raises=EngineNotImplemented is deliberate: -# - today, the tests raise EngineNotImplemented -> XFAIL -> build stays green. -# - the moment W1 wires a real check() into `_engine_check`, these tests will -# start passing -> XPASS -> strict=True fails the build, forcing us to -# remove the marker and turn them into real gates. That's the whole point. -_pending_engine = pytest.mark.xfail( - reason="requires W1 engine (core/rebac.py) — see #4/#5/#6", - raises=EngineNotImplemented, - strict=True, -) - - @pytest.mark.differential -@_pending_engine @_hyp @given(graph=authz_graphs()) def test_safety_engine_never_returns_docs_the_oracle_denies(graph: Graph) -> None: @@ -96,7 +83,6 @@ def test_safety_engine_never_returns_docs_the_oracle_denies(graph: Graph) -> Non @pytest.mark.differential -@_pending_engine @_hyp @given(graph=authz_graphs()) def test_fidelity_engine_returns_everything_the_oracle_allows(graph: Graph) -> None: @@ -116,7 +102,6 @@ def test_fidelity_engine_returns_everything_the_oracle_allows(graph: Graph) -> N @pytest.mark.differential -@_pending_engine @_hyp @given(graph=authz_graphs()) def test_pointwise_check_matches_oracle(graph: Graph) -> None: diff --git a/evals/generators.py b/evals/generators.py index 655de16..3cf3ee3 100644 --- a/evals/generators.py +++ b/evals/generators.py @@ -151,18 +151,26 @@ def _barrier(draw, barrier_id: int) -> Barrier: # --------------------------------------------------------------------------- @st.composite -def _document(draw, org: str, barrier_ids: list[int]) -> Document: - """A doc in `org`, optionally tagged behind one or more barriers. +def _document(draw, doc_id: str, org: str, barrier_ids: list[int]) -> Document: + """A doc with a caller-supplied `doc_id` (so callers can enforce uniqueness), + optionally tagged behind one or more barriers. Half the time: no barrier tags at all (the "open" corpus doc case). Otherwise: a small subset of the graph's barriers, each with a random side. + + Note on why `doc_id` is a caller arg: Document.id is a primary key in the + real schema. A frozenset of Documents with duplicate ids but distinct + barrier_tags is illegal in Postgres (unique(id)); if we generate such a + graph, the oracle and engine may pick different rows nondeterministically + via iteration order, producing a spurious differential failure. See + authz_graphs() for how uniqueness is enforced. """ if not barrier_ids or draw(st.booleans()): tags: frozenset[int] = frozenset() else: picked = draw(st.lists(st.sampled_from(barrier_ids), min_size=1, max_size=2, unique=True)) tags = frozenset(tag(b, draw(st.sampled_from([0, 1]))) for b in picked) - return Document(id=draw(doc_ids), org_id=org, barrier_tags=tags) + return Document(id=doc_id, org_id=org, barrier_tags=tags) # --------------------------------------------------------------------------- @@ -204,7 +212,14 @@ def authz_graphs(draw) -> Graph: # Pick one org for the documents. In W2 we'll broaden this to # multi-partition scenarios; for W0 the interesting stuff is intra-org. org = draw(org_ids) - documents = frozenset(draw(_document(org, barrier_ids)) for _ in range(n_docs)) + + # Enforce unique doc IDs — Document.id is a PK in the real schema. + unique_doc_ids = draw( + st.lists(doc_ids, min_size=1, max_size=n_docs, unique=True) + ) + documents = frozenset( + draw(_document(did, org, barrier_ids)) for did in unique_doc_ids + ) return Graph(tuples=frozenset(tuples), barriers=barriers, documents=documents) diff --git a/pyproject.toml b/pyproject.toml index 421bae2..7064bf9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -17,11 +17,15 @@ dev = [ "hypothesis>=6.100", "ruff>=0.5", "mypy>=1.10", + "psycopg[binary]>=3.1", ] [tool.hatch.build.targets.wheel] packages = ["core", "evals"] +[tool.hatch.build.targets.wheel.shared-data] +"sql" = "share/warden/sql" + [tool.pytest.ini_options] testpaths = ["tests"] addopts = "-ra --strict-markers" diff --git a/sql/001_init.sql b/sql/001_init.sql new file mode 100644 index 0000000..ff3ef19 --- /dev/null +++ b/sql/001_init.sql @@ -0,0 +1,37 @@ +-- Warden schema, initial migration (W1). +-- +-- Two tables: `tuples` (Zanzibar-shaped relationship state) and `barriers` +-- (information walls / deny layer). `documents` and pgvector arrive in W2. + +CREATE TABLE IF NOT EXISTS tuples ( + subject_type TEXT NOT NULL, -- user | group | org + subject_id TEXT NOT NULL, + subject_rel TEXT NOT NULL DEFAULT '', -- userset rewrite: group:eng#member + relation TEXT NOT NULL, -- member | parent | viewer | owner + object_type TEXT NOT NULL, -- org | matter | space | group | doc + object_id TEXT NOT NULL, + expires_at TIMESTAMPTZ, -- NULL = permanent + PRIMARY KEY (subject_type, subject_id, subject_rel, relation, object_type, object_id) +); + +-- Forward index: "who has relation R on object X?" +-- Used by expand(). +CREATE INDEX IF NOT EXISTS tuples_fwd + ON tuples (object_type, object_id, relation); + +-- Reverse index: "what does subject S touch?" +-- Used by the check() outgoing() walk. +CREATE INDEX IF NOT EXISTS tuples_rev + ON tuples (subject_type, subject_id); + +CREATE TABLE IF NOT EXISTS barriers ( + id BIGSERIAL PRIMARY KEY, + name TEXT NOT NULL, + side_a TEXT NOT NULL, -- group id + side_b TEXT NOT NULL, + CHECK (side_a <> side_b) -- a barrier between a side and itself is meaningless +); + +-- Barriers are small in count (single digits to low tens even for large +-- tenants), so no secondary index is warranted yet. If barrier lookup +-- becomes hot, index on (side_a, side_b) — but measure first. diff --git a/tests/test_postgres_store.py b/tests/test_postgres_store.py new file mode 100644 index 0000000..fbfa1c0 --- /dev/null +++ b/tests/test_postgres_store.py @@ -0,0 +1,174 @@ +""" +Integration tests for PostgresStore. + +Only runs when a Postgres URL is provided via WARDEN_TEST_DB_URL. Otherwise +the whole module is skipped so contributors without a running Postgres can +still `make oracle` / `make test` locally. + + Set WARDEN_TEST_DB_URL to something like: + postgres://warden:warden@localhost:5432/warden + + Or use: + docker compose up -d postgres + make postgres-integration +""" + +from __future__ import annotations + +import os +from datetime import datetime, timedelta, timezone + +import pytest + +pytest.importorskip("psycopg") + +import psycopg + +from core.algebra import Barrier, Document, Graph, Object, Subject, Tuple, tag +from core.postgres_store import ( + PostgresStore, + apply_migrations, + delete_tuple, + list_tuples, + load_graph, + write_tuple, +) +from core.rebac import check + +DB_URL = os.environ.get("WARDEN_TEST_DB_URL") +pytestmark = pytest.mark.skipif( + DB_URL is None, + reason="Set WARDEN_TEST_DB_URL to run Postgres integration tests. " + "See docker-compose.yml for a local dev DB.", +) + + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +@pytest.fixture(scope="session") +def _migrations() -> None: + """Apply migrations once per test session.""" + with psycopg.connect(DB_URL) as c: # type: ignore[arg-type] + apply_migrations(c) + + +@pytest.fixture() +def conn(_migrations): + """Fresh connection per test. Truncates state at the start of each test + so ordering doesn't matter.""" + c = psycopg.connect(DB_URL) # type: ignore[arg-type] + try: + c.execute("TRUNCATE tuples, barriers RESTART IDENTITY") + c.commit() + yield c + finally: + c.close() + + +def _t(subject: str, relation: str, obj: str, expires_at=None) -> Tuple: + s_type, s_id = subject.split(":", 1) + o_type, o_id = obj.split(":", 1) + return Tuple( + subject=Subject(type=s_type, id=s_id), # type: ignore[arg-type] + relation=relation, # type: ignore[arg-type] + object=Object(type=o_type, id=o_id), # type: ignore[arg-type] + expires_at=expires_at, + ) + + +# --------------------------------------------------------------------------- +# CRUD (W1.1) +# --------------------------------------------------------------------------- + +def test_write_then_list(conn): + write_tuple(conn, _t("user:alice", "viewer", "doc:d1")) + write_tuple(conn, _t("user:bob", "member", "group:g1")) + conn.commit() + tuples = list_tuples(conn) + assert len(tuples) == 2 + + +def test_write_is_idempotent(conn): + t = _t("user:alice", "viewer", "doc:d1") + write_tuple(conn, t) + write_tuple(conn, t) # second write must not raise or duplicate + conn.commit() + assert len(list_tuples(conn)) == 1 + + +def test_delete(conn): + t = _t("user:alice", "viewer", "doc:d1") + write_tuple(conn, t) + conn.commit() + delete_tuple(conn, t) + conn.commit() + assert list_tuples(conn) == [] + + +def test_expires_at_roundtrip(conn): + tomorrow = NOW + timedelta(days=1) + write_tuple(conn, _t("user:alice", "viewer", "doc:d1", expires_at=tomorrow)) + conn.commit() + got = list_tuples(conn)[0] + assert got.expires_at is not None + + +# --------------------------------------------------------------------------- +# End-to-end: rebac.check() against a Postgres-backed store +# --------------------------------------------------------------------------- + +def test_check_against_postgres_direct_grant(conn): + graph = Graph( + tuples=frozenset({_t("user:alice", "viewer", "doc:d1")}), + barriers=frozenset(), + documents=frozenset({Document(id="d1", org_id="acme")}), + ) + load_graph(conn, graph) + store = PostgresStore(conn) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert ok + + +def test_check_against_postgres_5_hop_nesting(conn): + tuples = [_t("user:alice", "member", "group:g0")] + for i in range(5): + tuples.append(_t(f"group:g{i}", "member", f"group:g{i + 1}")) + tuples.append(_t("group:g5", "viewer", "doc:d1")) + load_graph( + conn, + Graph( + tuples=frozenset(tuples), + barriers=frozenset(), + documents=frozenset({Document(id="d1", org_id="acme")}), + ), + ) + store = PostgresStore(conn) + ok, reason = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert ok + assert len(reason.steps) == 7 + + +def test_check_against_postgres_with_barrier(conn): + barrier = Barrier(id=1, name="wall", side_a="team_a", side_b="team_b") + doc = Document(id="d1", org_id="firm", barrier_tags=frozenset({tag(1, 1)})) + tuples = [ + _t("user:alice", "member", "group:team_a"), + _t("user:alice", "viewer", "doc:d1"), # grant, but wall wins + ] + id_map = load_graph( + conn, + Graph( + tuples=frozenset(tuples), + barriers=frozenset({barrier}), + documents=frozenset({doc}), + ), + ) + store = PostgresStore(conn) + # Postgres assigned a BIGSERIAL barrier.id which likely differs from + # our conceptual id=1. Reload doc tags with the mapped id, using the + # W1 side-channel that PostgresStore exposes for tests. + real_barrier_id = id_map[1] + store.load_doc_tags({"d1": frozenset({tag(real_barrier_id, 1)})}) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert not ok diff --git a/tests/test_rebac.py b/tests/test_rebac.py new file mode 100644 index 0000000..fdafb96 --- /dev/null +++ b/tests/test_rebac.py @@ -0,0 +1,355 @@ +""" +Handwritten fixture tests for the REAL engine (core/rebac.py). + +These are NOT copies of tests/test_oracle.py. They target the same semantics +but with different graph shapes, different assertion emphases, and adversarial +patterns designed to break the specific algorithm choices in rebac.py — DFS +traversal, memoization, in-progress cycle detection. + + Rule: if you find a bug via the differential harness that these fixtures + missed, add a fixture here that would have caught it. Regression tests + are permanent. +""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +import pytest + +from core.algebra import ( + Barrier, + Document, + Graph, + Object, + Subject, + Tuple, + tag, +) +from core.rebac import authorized_set, check, expand +from core.store import InMemoryStore + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +def _t(subject: str, relation: str, obj: str, expires_at: datetime | None = None) -> Tuple: + s_type, s_id = subject.split(":", 1) + o_type, o_id = obj.split(":", 1) + return Tuple( + subject=Subject(type=s_type, id=s_id), # type: ignore[arg-type] + relation=relation, # type: ignore[arg-type] + object=Object(type=o_type, id=o_id), # type: ignore[arg-type] + expires_at=expires_at, + ) + + +def _store( + tuples: list[Tuple], + docs: list[Document] | None = None, + barriers: list[Barrier] | None = None, +) -> InMemoryStore: + return InMemoryStore( + Graph( + tuples=frozenset(tuples), + barriers=frozenset(barriers or []), + documents=frozenset(docs or []), + ) + ) + + +# --------------------------------------------------------------------------- +# Baseline: direct grant + no grant. Sanity checks — if these fail, don't +# bother reading the rest of the failures. +# --------------------------------------------------------------------------- + +def test_direct_grant() -> None: + store = _store([_t("user:alice", "viewer", "doc:d1")]) + ok, reason = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert ok + assert reason.decision == "allow" + assert len(reason.steps) == 1 + + +def test_no_grant_denies() -> None: + store = _store([]) + ok, reason = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert not ok + assert reason.decision == "deny" + + +# --------------------------------------------------------------------------- +# DFS-specific traps. The oracle uses BFS and won't hit these the same way, +# which is exactly why they're here. +# --------------------------------------------------------------------------- + +def test_dfs_explores_longer_path_when_shorter_expired() -> None: + """A 2-hop path via g_short is expired. A 4-hop path via g_long is not. + DFS ordering must not permanently commit to the first path and miss the + second. Same doc; longer route is the only live one.""" + yesterday = NOW - timedelta(days=1) + tuples = [ + # Short (expired): + _t("user:alice", "member", "group:g_short"), + _t("group:g_short", "viewer", "doc:d1", expires_at=yesterday), + # Long (live): + _t("user:alice", "member", "group:g_a"), + _t("group:g_a", "member", "group:g_b"), + _t("group:g_b", "member", "group:g_c"), + _t("group:g_c", "viewer", "doc:d1"), + ] + store = _store(tuples) + ok, reason = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert ok + # No path with an expired tuple is allowed to appear in the reason. + assert all(step.tuple.unexpired_at(NOW) for step in reason.steps) + + +def test_deep_diamond_dfs_backtracks_correctly() -> None: + """A diamond where the 'wrong' branch has no grant and the 'right' branch + does. DFS must backtrack out of the wrong branch, not report deny. + + alice -> g_root -> g_left -> (dead end) + \\-> g_right -> viewer -> doc + """ + tuples = [ + _t("user:alice", "member", "group:root"), + _t("group:root", "member", "group:left"), + _t("group:root", "member", "group:right"), + # left is a dead end for this doc + _t("group:right", "viewer", "doc:d1"), + ] + store = _store(tuples) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert ok + + +def test_self_loop_on_subject_terminates() -> None: + """A tuple where a group is a member of itself. Must not recurse forever.""" + tuples = [ + _t("user:alice", "member", "group:g"), + _t("group:g", "member", "group:g"), # self-loop + _t("group:g", "viewer", "doc:d1"), + ] + store = _store(tuples) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert ok + + +def test_mutual_cycle_terminates_when_no_grant() -> None: + """A member B, B member A, no grant anywhere. Must return deny promptly + rather than loop.""" + tuples = [ + _t("user:alice", "member", "group:a"), + _t("group:a", "member", "group:b"), + _t("group:b", "member", "group:a"), + ] + store = _store(tuples) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert not ok + + +# --------------------------------------------------------------------------- +# Depth limit. MAX_DEPTH is 8 (see algebra); chains longer than that must +# not resolve regardless of DFS ordering or memoization. +# --------------------------------------------------------------------------- + +def test_10_hop_chain_exceeds_max_depth() -> None: + tuples = [_t("user:alice", "member", "group:g0")] + for i in range(10): + tuples.append(_t(f"group:g{i}", "member", f"group:g{i + 1}")) + tuples.append(_t("group:g10", "viewer", "doc:d1")) + store = _store(tuples) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert not ok + + +def test_exactly_at_max_depth_still_resolves() -> None: + """A chain whose total path length is exactly MAX_DEPTH must succeed — + off-by-one on the depth check is a classic bug we want to catch here.""" + from core.algebra import MAX_DEPTH + + # Path structure: user -member-> g0 -member-> g1 -> ... -> gN -viewer-> doc + # Total steps = (N group hops) + 1 initial member + 1 terminal grant + # For total == MAX_DEPTH: N = MAX_DEPTH - 2 + n_group_hops = MAX_DEPTH - 2 + tuples = [_t("user:alice", "member", "group:g0")] + for i in range(n_group_hops): + tuples.append(_t(f"group:g{i}", "member", f"group:g{i + 1}")) + tuples.append(_t(f"group:g{n_group_hops}", "viewer", "doc:d1")) + store = _store(tuples) + ok, reason = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert ok + assert len(reason.steps) == MAX_DEPTH + + +# --------------------------------------------------------------------------- +# Barriers: deny dominates, at any depth. +# --------------------------------------------------------------------------- + +def test_barrier_blocks_at_leaf_grant() -> None: + barrier = Barrier(id=3, name="wall", side_a="acme", side_b="zenith") + doc = Document(id="d1", org_id="firm", barrier_tags=frozenset({tag(3, 1)})) + tuples = [ + _t("user:alice", "member", "group:acme"), + _t("user:alice", "viewer", "doc:d1"), # direct grant, but wall wins + ] + store = _store(tuples, docs=[doc], barriers=[barrier]) + ok, reason = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert not ok + assert reason.barrier_hit == barrier + + +def test_barrier_blocks_even_when_grant_is_5_hops_deep() -> None: + """A firm-wide chain grants access; a barrier blocks. Deny must beat allow + regardless of grant-path depth.""" + barrier = Barrier(id=1, name="wall", side_a="team_a", side_b="team_b") + doc = Document(id="d1", org_id="firm", barrier_tags=frozenset({tag(1, 1)})) + tuples = [ + _t("user:alice", "member", "group:team_a"), + _t("user:alice", "member", "group:firm_all"), + _t("group:firm_all", "member", "group:everyone"), + _t("group:everyone", "member", "group:global_readers"), + _t("group:global_readers", "viewer", "doc:d1"), + ] + store = _store(tuples, docs=[doc], barriers=[barrier]) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert not ok + + +def test_barrier_does_not_block_when_principal_not_on_either_side() -> None: + """A user unaffected by a barrier should get through as normal. + Regression guard for a barrier check that assumes everyone is on a side.""" + barrier = Barrier(id=1, name="wall", side_a="acme", side_b="zenith") + doc = Document(id="d1", org_id="firm", barrier_tags=frozenset({tag(1, 1)})) + tuples = [ + _t("user:bob", "viewer", "doc:d1"), # bob is not in either wall group + ] + store = _store(tuples, docs=[doc], barriers=[barrier]) + ok, _ = check(store, Subject("user", "bob"), Object("doc", "d1"), NOW) + assert ok + + +def test_barrier_wins_over_org_ownership() -> None: + """Org-owner grants access to every doc in the org. Barrier still wins. + Different grant relation (`owner`) than in previous tests — the deny + check must not be relation-specific.""" + barrier = Barrier(id=5, name="wall", side_a="a", side_b="b") + doc = Document(id="d1", org_id="acme", barrier_tags=frozenset({tag(5, 1)})) + tuples = [ + _t("user:alice", "member", "group:a"), + _t("user:alice", "member", "org:acme"), + _t("org:acme", "owner", "doc:d1"), + ] + store = _store(tuples, docs=[doc], barriers=[barrier]) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert not ok + + +# --------------------------------------------------------------------------- +# Expiry semantics — every tuple on the path must be live. +# --------------------------------------------------------------------------- + +def test_intermediate_expired_kills_that_path_but_not_others() -> None: + """Path 1: alice->g1->g2 (g1->g2 expired) → dead + Path 2: alice->direct viewer → live + Should still allow.""" + yesterday = NOW - timedelta(days=1) + tuples = [ + _t("user:alice", "member", "group:g1"), + _t("group:g1", "member", "group:g2", expires_at=yesterday), + _t("group:g2", "viewer", "doc:d1"), + _t("user:alice", "viewer", "doc:d1"), + ] + store = _store(tuples) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert ok + + +def test_grant_tuple_itself_expired_denies() -> None: + """The terminal grant edge is expired even though the path leading to it is live.""" + yesterday = NOW - timedelta(days=1) + tuples = [ + _t("user:alice", "member", "group:g"), + _t("group:g", "viewer", "doc:d1", expires_at=yesterday), + ] + store = _store(tuples) + ok, _ = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert not ok + + +# --------------------------------------------------------------------------- +# Bulk API: authorized_set. +# --------------------------------------------------------------------------- + +def test_authorized_set_reachability() -> None: + docs = [Document(id=f"d{i}", org_id="acme") for i in range(4)] + tuples = [ + _t("user:alice", "viewer", "doc:d0"), + _t("user:alice", "member", "group:g"), + _t("group:g", "viewer", "doc:d2"), + # d1, d3 unreachable + ] + store = _store(tuples, docs=docs) + assert authorized_set(store, Subject("user", "alice"), NOW, docs) == {"d0", "d2"} + + +def test_authorized_set_respects_barriers() -> None: + barrier = Barrier(id=1, name="wall", side_a="a", side_b="b") + docs = [ + Document(id="open", org_id="firm"), + Document(id="walled", org_id="firm", barrier_tags=frozenset({tag(1, 1)})), + ] + tuples = [ + _t("user:alice", "member", "group:a"), + _t("user:alice", "viewer", "doc:open"), + _t("user:alice", "viewer", "doc:walled"), # grant exists; wall wins + ] + store = _store(tuples, docs=docs, barriers=[barrier]) + assert authorized_set(store, Subject("user", "alice"), NOW, docs) == {"open"} + + +# --------------------------------------------------------------------------- +# expand(): the inverse of check(). Given (object, relation), list subjects. +# --------------------------------------------------------------------------- + +def test_expand_returns_direct_grantees() -> None: + tuples = [ + _t("user:alice", "viewer", "doc:d1"), + _t("user:bob", "viewer", "doc:d1"), + _t("user:carol", "owner", "doc:d1"), # different relation, must not show up + ] + store = _store(tuples) + subjects = expand(store, Object("doc", "d1"), "viewer") + ids = {s.id for s in subjects} + assert ids == {"alice", "bob"} + + +# --------------------------------------------------------------------------- +# Reason path integrity — the audit log depends on this. +# --------------------------------------------------------------------------- + +def test_reason_path_steps_are_a_valid_chain() -> None: + """Each step's object should be the next step's subject (or, for the + terminal step, the target object). Otherwise the audit log lies.""" + tuples = [ + _t("user:alice", "member", "group:g1"), + _t("group:g1", "member", "group:g2"), + _t("group:g2", "viewer", "doc:d1"), + ] + store = _store(tuples) + _, reason = check(store, Subject("user", "alice"), Object("doc", "d1"), NOW) + assert reason.decision == "allow" + # First step subject is the principal. + assert reason.steps[0].tuple.subject == Subject("user", "alice") + # Chain consistency. + for i in range(len(reason.steps) - 1): + assert ( + reason.steps[i].tuple.object.type == reason.steps[i + 1].tuple.subject.type + or reason.steps[i].tuple.object.id == reason.steps[i + 1].tuple.subject.id + ), f"broken chain at step {i}" + # Terminal step must land on the target. + assert reason.steps[-1].tuple.object == Object("doc", "d1") + + +if __name__ == "__main__": + raise SystemExit(pytest.main([__file__, "-v"]))