diff --git a/tests/test_tenant_isolation.py b/tests/test_tenant_isolation.py index 97d2cbd..771e18a 100644 --- a/tests/test_tenant_isolation.py +++ b/tests/test_tenant_isolation.py @@ -16,6 +16,7 @@ from __future__ import annotations import json +import re import os import pytest @@ -228,6 +229,16 @@ def header(self, who: str) -> dict[str, str]: return {"Authorization": f"Bearer {who}"} + def aliases(self, who: str) -> list[str]: + """Every way the test can NAME that tenant in a request field: its subject, and the strings the server itself uses for the operator.""" + from app.identity import reset_principal, set_principal, current_tenant_key + token = set_principal(Principal(subject=who, scopes=frozenset({"memory.read"}), issuer="https://issuer.example")) + try: + return [current_tenant_key(), who, "operator"] # the tenant key the server derives for that subject is the string a forger most wants + finally: + reset_principal(token) + + AUTH = OAuthBearers() @@ -736,3 +747,80 @@ def test_a_tenant_is_not_told_the_emr_sidecar_path(client): r = client.get("/api/jarvis/memory/emr/status", headers=h("bob")) leaked = [v for v in _values_of(r.json(), "path") if isinstance(v, str) and v] assert not leaked, f"the status shows server paths: {leaked}" # (only the `path` fields: the response legitimately says "emr-dynamics-v1" elsewhere) + + +# ----------------------------------------------------------------------------------------------- a forged tenant field never widens scope +FORGED_FIELDS = ("tenant_id", "_tenant_id", "tenant_key", "tenant", "jarvis.tenant_key") +FORGED_HEADERS = ("X-Tenant-Id", "X-Tenant-Key", "X-Jarvis-Tenant", "X-Jarvis-Tenant-Key") + + +TRANSPORTS = ("body", "query", "headers") +OPERATOR_MARK = "operator-owned-marker-kestrel-91d2" + + +def _forged(client, who, method, path, body, victims, transport): + """Send the request as `who` with every tenant-naming field set to the victim, through ONE transport only. Separate requests per transport mean a body that a strict route rejects (422 for any + unknown field) cannot hide a query string or a header that the route would honour.""" + value = victims[0] + headers, params, payload = h(who), {}, body + if transport == "headers": + headers = {**headers, **{k: value for k in FORGED_HEADERS}} + elif transport == "query": + params = {k: value for k in FORGED_FIELDS} + elif isinstance(body, dict): + payload = {**body, **{k: value for k in FORGED_FIELDS}} + return client.request(method, path, headers=headers, params=params, json=payload) + + +def _seed_operator_state(): + """Recognizable operator-owned data, so that a forged `operator` selector that DID switch the namespace would show it (an empty operator namespace looks like a nonexistent decoy).""" + from app.identity import operator_principal, reset_principal, set_principal + from app.models import EvidenceLink, MemoryCreate + from app.store import get_store + + token = set_principal(operator_principal()) + try: + get_store().create_memory(MemoryCreate(content=f"operator private note {OPERATOR_MARK} {SEARCH}", source_agent="op", session_id="s", type="fact", + evidence=[EvidenceLink(kind="doc", ref="x", note="y")])) + finally: + reset_principal(token) + + +def test_a_forged_tenant_field_never_widens_scope(client, world): + """RFC appendix item 19. The tenant is the authenticated principal's; no route or tool takes it from the request. Bob names alice (every alias the test knows, in the body, the query + string and four headers) and the response must be exactly what he gets naming a tenant that does not exist: naming alice buys him nothing, and nothing of alice's changes.""" + victims = getattr(AUTH, "aliases", lambda who: [who])("alice") + decoys = ["nobody-" + v for v in victims] + _seed_operator_state() + cases = {**{f"list:{k}": v for k, v in LIST_CASES.items()}, **{f"mutation:{k}": v for k, v in MUTATION_CASES.items()}, + **{f"id:{k}": v for k, v in id_cases(world["first"], world["artifact"]).items()}} + before = _alice_state(client, world) + failures = [] + checked = 0 + for victim, decoy in zip(victims, decoys): + for transport in TRANSPORTS: + for case, (method, path, body) in sorted(cases.items()): + if case.removeprefix("id:") in POSTGRES_ONLY and not ON_POSTGRES: + continue + tag = re.sub(r"\W", "", f"{case}{victim}{transport}")[:40] # two brand-new tenants with identical (empty) state, so the only difference is the name each one forges + named = _forged(client, f"fa-{tag}", method, path, body, [victim], transport) + blind = _forged(client, f"fb-{tag}", method, path, body, [decoy], transport) + checked += 1 + + def same(resp, name): # `name` is the victim alias; the decoy is "nobody-" + it; a 422 echoes the caller's own forged value (his input, not a leak) + text = norm(resp, world["first"], world["second"], world["third"], world["working"], world["artifact"])[1].replace("nobody-" + name, "").replace(name, "") + text = re.sub(r"mem-[0-9a-f]{12}", "", text) # a write that succeeds mints ids and clocks of its own, wherever the response nests them + return resp.status_code, re.sub(r"\d{4}-\d\d-\d\dT[\d:.]+\+00:00", "", text) + + if same(named, victim) != same(blind, victim): + a, b = same(named, victim), same(blind, victim) + at = next((k for k, (x, y) in enumerate(zip(a[1], b[1])) if x != y), 0) + failures.append(f"{transport}:{case} naming {victim!r}: {a[0]} != {b[0]} (naming nobody); first difference: ...{a[1][max(0, at - 80):at + 80]!r} vs ...{b[1][max(0, at - 80):at + 80]!r}") + for mark in (MARK, OPERATOR_MARK): + if mark in named.text: + failures.append(f"{transport}:{case} naming {victim!r}: {'alice' if mark == MARK else 'the operator'}'s text appeared") + if _alice_state(client, world) != before: + failures.append(f"{transport}:{case} naming {victim!r}: changed alice's state") + before = _alice_state(client, world) + assert checked >= 3 * 2 * 25, f"the forged-field sweep covered only {checked} requests" + assert not failures, "\n".join(failures) diff --git a/tests/test_tenant_isolation_keys.py b/tests/test_tenant_isolation_keys.py index de50b93..e06365d 100644 --- a/tests/test_tenant_isolation_keys.py +++ b/tests/test_tenant_isolation_keys.py @@ -28,15 +28,20 @@ class KeyTenants: """h(who) -> that name's own API key, minted on first use (a fresh tenant each, so `carol-who-never-wrote` really is a tenant with nothing).""" def __init__(self, admin_dsn: str, control: str): - self.admin_dsn, self.control, self.tokens = admin_dsn, control, {} + self.admin_dsn, self.control, self.tokens, self.tenants = admin_dsn, control, {}, {} def header(self, who: str) -> dict[str, str]: if who not in self.tokens: with psycopg.connect(self.admin_dsn, autocommit=True, row_factory=dict_row) as c: tenant = tenant_keys.create_tenant(c, self.control, who) + self.tenants[who] = tenant self.tokens[who] = tenant_keys.create_key(c, self.control, tenant, PEPPER.encode(), label=who) return {"Authorization": f"Bearer {self.tokens[who]}"} + def aliases(self, who: str) -> list[str]: + self.header(who) + return [self.tenants[who], who, "operator"] + @pytest.fixture def client(pg_schema, monkeypatch, tmp_path):