tegwick
aaa2b4c50b
_read gated on may_read alone and never called permits_tenant, so any reader credential could read every tenant through /v1/events, /v1/events/<id>, /v1/dead-letters and /v1/secret-findings. Deployment bounded the exposure -- the only production sender holds may_read: false -- but the boundary was not in the code, which is the difference between E2 and E1 on the tenancy posture enforcement ladder. Two rules, because the surfaces divide cleanly. Event reads are filtered to the tenants the credential may act for. Surfaces with no tenant key to filter on -- stats, integrity, dead letters, secret findings -- require full tenant scope and are refused rather than served instance-wide facts to a scoped reader. A cross-tenant fetch returns 404 rather than 403. A distinguishable forbidden would confirm that an event id exists and which tenant holds it, turning the read surface into an existence oracle. Correlation lookup is filtered rather than refused, since a correlation id legitimately spans tenants. _readable_by fails closed: a record with no tenant is readable only at full scope. Three existing tests read instance-wide surfaces with a scoped credential, which this makes a 403; bound_app now carries an unrestricted operator identity and those reads use it. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
594 lines
23 KiB
Python
594 lines
23 KiB
Python
import io
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from audit_core.ingestion import IngestionApplication
|
|
from audit_core.interface import (
|
|
BackendUnavailableError,
|
|
RetentionPolicy,
|
|
custody_class_satisfies,
|
|
)
|
|
from audit_core.mock_file_backend import MockFileAuditBackend
|
|
from audit_core.sqlite_backend import SQLiteAuditBackend
|
|
|
|
|
|
def invoke(app, payload, *, token="opaque", key="evt-1", method="POST",
|
|
path="/v1/events", body=None, length=None):
|
|
raw = body if body is not None else json.dumps(payload).encode()
|
|
environ = {
|
|
"PATH_INFO": path,
|
|
"REQUEST_METHOD": method,
|
|
"CONTENT_LENGTH": str(len(raw)) if length is None else length,
|
|
"wsgi.input": io.BytesIO(raw),
|
|
"HTTP_AUTHORIZATION": f"Bearer {token}",
|
|
}
|
|
if key is not None:
|
|
environ["HTTP_IDEMPOTENCY_KEY"] = key
|
|
result = {}
|
|
out = b"".join(app(environ, lambda status, headers: result.update(status=status)))
|
|
return result["status"], (json.loads(out) if out else {})
|
|
|
|
|
|
def event(**overrides):
|
|
base = {
|
|
"id": "evt-1",
|
|
"type": "membership.added",
|
|
"source": "user-engine",
|
|
"subject": "membership-1",
|
|
"tenant": "tenant:friendly:binky",
|
|
"correlation_id": "corr-1",
|
|
"occurred_at": "2026-08-09T00:00:00+00:00",
|
|
"data": {"membership_id": "membership-1"},
|
|
}
|
|
base.update(overrides)
|
|
return base
|
|
|
|
|
|
@pytest.fixture
|
|
def app(tmp_path):
|
|
return IngestionApplication(SQLiteAuditBackend(str(tmp_path / "events.db")), "opaque")
|
|
|
|
|
|
# --- backend contract (T01) -------------------------------------------------
|
|
|
|
def test_refuses_a_non_durable_backend():
|
|
"""The development file backend must never become the production sink."""
|
|
with pytest.raises(ValueError, match="not durable"):
|
|
IngestionApplication(MockFileAuditBackend(base_dir="/tmp/unused"), "opaque")
|
|
|
|
|
|
def test_readiness_reports_custody_class(app):
|
|
status, body = invoke(app, None, path="/readyz", method="GET", body=b"")
|
|
assert status.startswith("200")
|
|
assert body["durable"] is True
|
|
assert body["custody_class"] == "development"
|
|
|
|
|
|
def test_accepted_events_are_durable_across_reopen(tmp_path):
|
|
path = str(tmp_path / "events.db")
|
|
first = IngestionApplication(SQLiteAuditBackend(path), "opaque")
|
|
assert invoke(first, event())[0].startswith("202")
|
|
|
|
reopened = IngestionApplication(SQLiteAuditBackend(path), "opaque")
|
|
status, body = invoke(reopened, event())
|
|
assert status.startswith("200") and body["status"] == "duplicate"
|
|
|
|
|
|
# --- idempotency and conflict (T01, T02) ------------------------------------
|
|
|
|
def test_accepts_once_and_replays_idempotently(app):
|
|
assert invoke(app, event())[0].startswith("202")
|
|
status, body = invoke(app, event())
|
|
assert status.startswith("200") and body["status"] == "duplicate"
|
|
|
|
|
|
def test_same_id_different_payload_is_a_conflict(app):
|
|
assert invoke(app, event())[0].startswith("202")
|
|
status, body = invoke(app, event(subject="membership-2"))
|
|
assert status.startswith("409")
|
|
assert body["error"] == "event_id_conflict"
|
|
|
|
|
|
# --- authentication (T02) ---------------------------------------------------
|
|
|
|
def test_rejects_wrong_credential(app):
|
|
assert invoke(app, event(), token="wrong")[0].startswith("401")
|
|
|
|
|
|
def test_non_ascii_credential_is_unauthorized_not_a_crash(app):
|
|
"""compare_digest raises TypeError on non-ASCII str; that is a 401."""
|
|
assert invoke(app, event(), token="wröng")[0].startswith("401")
|
|
|
|
|
|
# --- validation (T02, T07) --------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("payload,expected", [
|
|
(event(source="somewhere-else"), "source_not_allowed"),
|
|
(event(occurred_at="2026-08-09T00:00:00"), "timestamp_missing_timezone"),
|
|
(event(occurred_at="not-a-date"), "invalid_timestamp"),
|
|
(event(tenant=""), "invalid_event"),
|
|
])
|
|
def test_rejects_bad_events(app, payload, expected):
|
|
status, body = invoke(app, payload)
|
|
assert status.startswith("400")
|
|
assert body["error"] == expected
|
|
|
|
|
|
def test_rejects_mismatched_idempotency_key(app):
|
|
status, body = invoke(app, event(), key="other")
|
|
assert status.startswith("400")
|
|
assert body["error"] == "idempotency_key_mismatch"
|
|
|
|
|
|
def test_rejects_malformed_json(app):
|
|
assert invoke(app, None, body=b"{not json")[0].startswith("400")
|
|
|
|
|
|
def test_rejects_empty_body(app):
|
|
status, body = invoke(app, None, body=b"")
|
|
assert status.startswith("400")
|
|
assert body["error"] == "empty_body"
|
|
|
|
|
|
def test_rejects_oversized_body(app):
|
|
status, body = invoke(app, None, body=b"x", length=str(512 * 1024))
|
|
assert status.startswith("400")
|
|
assert body["error"] == "payload_too_large"
|
|
|
|
|
|
def test_rejects_truncated_body(app):
|
|
status, body = invoke(app, None, body=b"{}", length="500")
|
|
assert status.startswith("400")
|
|
assert body["error"] == "truncated_body"
|
|
|
|
|
|
# --- routing (T07) ----------------------------------------------------------
|
|
|
|
@pytest.mark.parametrize("path,method", [
|
|
("/nope", "POST"),
|
|
("/nope", "GET"),
|
|
("/v1/events", "DELETE"),
|
|
])
|
|
def test_unknown_routes_are_not_found(app, path, method):
|
|
assert invoke(app, event(), path=path, method=method)[0].startswith("404")
|
|
|
|
|
|
def test_healthz_needs_no_credential(app):
|
|
status, _ = invoke(app, None, path="/healthz", method="GET", body=b"", token="wrong")
|
|
assert status.startswith("200")
|
|
|
|
|
|
# --- failure handling (T02) -------------------------------------------------
|
|
|
|
class _BrokenBackend:
|
|
@property
|
|
def retention_policy(self):
|
|
return RetentionPolicy("archive", 3650, True, True, durable=True)
|
|
|
|
def emit(self, event):
|
|
raise BackendUnavailableError("down")
|
|
|
|
def accept(self, event, payload_hash):
|
|
raise BackendUnavailableError("down")
|
|
|
|
|
|
class _ExplodingBackend(_BrokenBackend):
|
|
def accept(self, event, payload_hash):
|
|
raise RuntimeError("unexpected")
|
|
|
|
|
|
def test_backend_unavailable_is_retryable_503():
|
|
status, body = invoke(IngestionApplication(_BrokenBackend(), "opaque"), event())
|
|
assert status.startswith("503")
|
|
assert body["error"] == "backend_unavailable"
|
|
|
|
|
|
def test_unexpected_backend_error_still_returns_a_response():
|
|
"""No request path may terminate without calling start_response."""
|
|
status, body = invoke(IngestionApplication(_ExplodingBackend(), "opaque"), event())
|
|
assert status.startswith("500")
|
|
assert body["error"] == "internal_error"
|
|
|
|
|
|
# --- concurrency (T01) ------------------------------------------------------
|
|
|
|
def test_concurrent_duplicates_produce_exactly_one_record(tmp_path):
|
|
"""Racing submissions of one event: one acceptance, one custody record.
|
|
|
|
This is the assertion the whole service rests on, so it is exercised rather
|
|
than assumed. An earlier single-connection implementation passed every
|
|
serial test while letting two callers both be told they were first.
|
|
"""
|
|
import threading
|
|
|
|
backend = SQLiteAuditBackend(str(tmp_path / "race.db"))
|
|
app = IngestionApplication(backend, "opaque")
|
|
outcomes: list[str] = []
|
|
lock = threading.Lock()
|
|
barrier = threading.Barrier(16)
|
|
|
|
def submit():
|
|
barrier.wait()
|
|
status, body = invoke(app, event())
|
|
with lock:
|
|
outcomes.append(body.get("status", status))
|
|
|
|
threads = [threading.Thread(target=submit) for _ in range(16)]
|
|
for t in threads:
|
|
t.start()
|
|
for t in threads:
|
|
t.join()
|
|
|
|
assert outcomes.count("accepted") == 1, outcomes
|
|
assert outcomes.count("duplicate") == 15, outcomes
|
|
stored = backend.db.execute("SELECT COUNT(*) FROM events").fetchone()[0]
|
|
assert stored == 1
|
|
|
|
|
|
# --- sender identity binding (T03) ------------------------------------------
|
|
|
|
from audit_core.senders import SenderIdentity, SenderRegistry # noqa: E402
|
|
|
|
|
|
def bound_app(tmp_path, **kw):
|
|
identity = SenderIdentity(
|
|
name="user-engine",
|
|
tokens=kw.get("tokens", ("opaque",)),
|
|
sources=frozenset(kw.get("sources", {"user-engine"})),
|
|
tenants=frozenset(kw.get("tenants", {"tenant:friendly:binky"})),
|
|
may_read=kw.get("may_read", False),
|
|
secret_policy=kw.get("secret_policy", "redact"),
|
|
)
|
|
# An unrestricted operator sits alongside the scoped sender. The
|
|
# instance-wide read surfaces — stats, dead letters, secret findings,
|
|
# integrity — carry no tenant key and so require full scope
|
|
# (AUDIT-WP-0008-T04); reading them as the scoped sender is what that task
|
|
# made a 403.
|
|
operator = SenderIdentity(
|
|
name="operator",
|
|
tokens=("operator",),
|
|
sources=frozenset({"user-engine"}),
|
|
tenants=frozenset({"*"}),
|
|
may_write=False,
|
|
may_read=True,
|
|
)
|
|
backend = SQLiteAuditBackend(str(tmp_path / "bound.db"))
|
|
return IngestionApplication(backend, SenderRegistry([identity, operator])), backend
|
|
|
|
|
|
def test_credential_may_not_claim_another_tenant(tmp_path):
|
|
"""The property WP-0003 recorded as done but never implemented."""
|
|
app, _ = bound_app(tmp_path)
|
|
assert invoke(app, event())[0].startswith("202")
|
|
status, body = invoke(app, event(tenant="tenant:coulomb"))
|
|
assert status.startswith("400")
|
|
assert body["error"] == "tenant_not_allowed"
|
|
|
|
|
|
def test_credential_may_not_claim_another_source(tmp_path):
|
|
app, _ = bound_app(tmp_path)
|
|
status, body = invoke(app, event(source="issue-core"))
|
|
assert status.startswith("400")
|
|
assert body["error"] == "source_not_allowed"
|
|
|
|
|
|
def test_rotation_accepts_both_tokens(tmp_path):
|
|
"""Rotation must not need a delivery gap."""
|
|
app, _ = bound_app(tmp_path, tokens=("current", "next"))
|
|
assert invoke(app, event(), token="current")[0].startswith("202")
|
|
assert invoke(app, event(id="evt-2"), key="evt-2", token="next")[0].startswith("202")
|
|
assert invoke(app, event(id="evt-3"), key="evt-3", token="retired")[0].startswith("401")
|
|
|
|
|
|
def test_sender_credential_cannot_read_the_trail_back(tmp_path):
|
|
app, _ = bound_app(tmp_path, may_read=False)
|
|
assert invoke(app, event())[0].startswith("202")
|
|
status, body = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"")
|
|
assert status.startswith("403")
|
|
assert body["error"] == "read_forbidden"
|
|
|
|
|
|
# --- operator read surface (T05) --------------------------------------------
|
|
|
|
def test_lookup_by_event_id_and_correlation(app):
|
|
assert invoke(app, event())[0].startswith("202")
|
|
assert invoke(app, event(id="evt-2", correlation_id="corr-1"), key="evt-2")[0].startswith("202")
|
|
|
|
status, body = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"")
|
|
assert status.startswith("200")
|
|
assert body["event_id"] == "evt-1"
|
|
assert body["tenant"] == "tenant:friendly:binky"
|
|
|
|
status, body = invoke_query(app, "correlation_id=corr-1")
|
|
assert status.startswith("200")
|
|
assert {e["event_id"] for e in body["events"]} == {"evt-1", "evt-2"}
|
|
|
|
|
|
def invoke_query(app, query, token="opaque"):
|
|
environ = {
|
|
"PATH_INFO": "/v1/events",
|
|
"REQUEST_METHOD": "GET",
|
|
"QUERY_STRING": query,
|
|
"CONTENT_LENGTH": "0",
|
|
"wsgi.input": io.BytesIO(b""),
|
|
"HTTP_AUTHORIZATION": f"Bearer {token}",
|
|
}
|
|
result = {}
|
|
out = b"".join(app(environ, lambda status, headers: result.update(status=status)))
|
|
return result["status"], (json.loads(out) if out else {})
|
|
|
|
|
|
def test_unknown_event_id_is_not_found(app):
|
|
status, _ = invoke(app, None, path="/v1/events/nope", method="GET", body=b"")
|
|
assert status.startswith("404")
|
|
|
|
|
|
def test_correlation_lookup_requires_a_correlation_id(app):
|
|
status, body = invoke_query(app, "")
|
|
assert status.startswith("400")
|
|
assert body["error"] == "correlation_id_required"
|
|
|
|
|
|
def test_rejected_events_appear_as_dead_letters(app):
|
|
assert invoke(app, event(source="issue-core"))[0].startswith("400")
|
|
status, body = invoke(app, None, path="/v1/dead-letters", method="GET", body=b"")
|
|
assert status.startswith("200")
|
|
entry = body["dead_letters"][0]
|
|
assert entry["reason"] == "source_not_allowed"
|
|
assert entry["event_id"] == "evt-1"
|
|
assert entry["payload"] is not None
|
|
|
|
|
|
def test_secret_rejection_withholds_the_payload(tmp_path):
|
|
"""Storing the body of an event rejected for carrying secret-shaped
|
|
material would write that material into the audit store."""
|
|
app, _ = bound_app(tmp_path, secret_policy="reject", may_read=True)
|
|
assert invoke(app, event(data={"password": "hunter2"}))[0].startswith("400")
|
|
_, body = invoke(app, None, path="/v1/dead-letters", method="GET", body=b"", token="operator")
|
|
entry = body["dead_letters"][0]
|
|
assert entry["reason"] == "secret_shaped_field"
|
|
assert entry["payload_withheld"] is True
|
|
assert entry["payload"] is None
|
|
assert entry["payload_hash"]
|
|
|
|
|
|
# --- redaction policy (T04) -------------------------------------------------
|
|
|
|
def test_default_policy_redacts_and_accepts(app):
|
|
"""Default is redact: losing the whole audit record over one field is
|
|
worse than storing it with that field masked."""
|
|
status, _ = invoke(app, event(data={"membership_id": "m-1", "auth_token": "s3cret"}))
|
|
assert status.startswith("202")
|
|
|
|
_, record = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"")
|
|
assert record["details"]["data"]["auth_token"] == "[redacted]"
|
|
assert record["details"]["data"]["membership_id"] == "m-1"
|
|
# The record must admit it was modified.
|
|
assert record["details"]["redaction"]["policy"] == "redact"
|
|
assert record["details"]["redaction"]["paths"] == ["data.auth_token"]
|
|
|
|
|
|
def test_reject_policy_is_available_per_sender(tmp_path):
|
|
app, _ = bound_app(tmp_path, secret_policy="reject")
|
|
status, body = invoke(app, event(data={"password": "x"}))
|
|
assert status.startswith("400")
|
|
assert body["error"] == "secret_shaped_field"
|
|
|
|
|
|
def test_nested_and_listed_secrets_are_redacted(app):
|
|
payload = event(data={"items": [{"api_secret": "a"}, {"ok": 1}], "n": {"private_key": "k"}})
|
|
assert invoke(app, payload)[0].startswith("202")
|
|
_, record = invoke(app, None, path="/v1/events/evt-1", method="GET", body=b"")
|
|
data = record["details"]["data"]
|
|
assert data["items"][0]["api_secret"] == "[redacted]"
|
|
assert data["items"][1]["ok"] == 1
|
|
assert data["n"]["private_key"] == "[redacted]"
|
|
assert set(record["details"]["redaction"]["paths"]) == {
|
|
"data.items[0].api_secret", "data.n.private_key",
|
|
}
|
|
|
|
|
|
def test_findings_are_counted_by_path_for_both_outcomes(tmp_path):
|
|
"""Counters name the field to fix, not just a total."""
|
|
app, backend = bound_app(tmp_path, may_read=True)
|
|
for i in range(3):
|
|
invoke(app, event(id=f"e{i}", data={"auth_token": "x"}), key=f"e{i}")
|
|
|
|
strict, _ = bound_app(tmp_path, secret_policy="reject")
|
|
invoke(strict, event(id="r1", data={"auth_token": "x"}), key="r1")
|
|
|
|
_, body = invoke(app, None, path="/v1/secret-findings", method="GET", body=b"", token="operator")
|
|
rows = {(r["outcome"], r["field_path"]): r for r in body["secret_findings"]}
|
|
assert rows[("redacted", "data.auth_token")]["occurrences"] == 3
|
|
assert rows[("redacted", "data.auth_token")]["action"] == "membership.added"
|
|
assert rows[("redacted", "data.auth_token")]["source"] == "user-engine"
|
|
assert rows[("redacted", "data.auth_token")]["persisted"] is True
|
|
assert rows[("rejected", "data.auth_token")]["occurrences"] == 1
|
|
|
|
|
|
def test_counters_survive_restart(tmp_path):
|
|
"""The counters drive a fix in the sending service; that work outlives a
|
|
pod restart, so they are durable rather than in-memory."""
|
|
path = str(tmp_path / "counters.db")
|
|
first = IngestionApplication(SQLiteAuditBackend(path), "opaque")
|
|
invoke(first, event(data={"auth_token": "x"}))
|
|
|
|
reopened = IngestionApplication(SQLiteAuditBackend(path), "opaque")
|
|
_, body = invoke(reopened, None, path="/v1/secret-findings", method="GET", body=b"")
|
|
assert body["secret_findings"][0]["occurrences"] == 1
|
|
|
|
|
|
# --- deployment guards and counters (WP-0005-T03) ---------------------------
|
|
|
|
def test_required_custody_class_refuses_a_development_backend(tmp_path):
|
|
"""Losing AUDIT_CORE_DATABASE_URL must fail to start, not silently
|
|
downgrade custody to the development store."""
|
|
backend = SQLiteAuditBackend(str(tmp_path / "dev.db"))
|
|
with pytest.raises(ValueError, match="does not meet the required"):
|
|
IngestionApplication(backend, "opaque", require_custody_class="operational")
|
|
with pytest.raises(ValueError, match="does not meet the required"):
|
|
IngestionApplication(backend, "opaque", require_custody_class="archive")
|
|
|
|
|
|
def test_operational_and_archive_alias_for_one_deploy():
|
|
"""A mixed rollout must start: new backend + old require, and the reverse."""
|
|
|
|
class _Operational(_BrokenBackend):
|
|
@property
|
|
def retention_policy(self):
|
|
return RetentionPolicy(
|
|
custody_class="operational",
|
|
retention_days=None,
|
|
immutable=True,
|
|
tamper_evidence=False,
|
|
durable=True,
|
|
recoverable_days=30,
|
|
recoverable_source="cited",
|
|
recoverable_basis="measured",
|
|
)
|
|
|
|
IngestionApplication(_Operational(), "opaque", require_custody_class="archive")
|
|
IngestionApplication(_Operational(), "opaque", require_custody_class="operational")
|
|
|
|
|
|
def test_custody_class_alias_is_not_development():
|
|
assert custody_class_satisfies("operational", "archive")
|
|
assert custody_class_satisfies("archive", "operational")
|
|
assert not custody_class_satisfies("development", "operational")
|
|
assert not custody_class_satisfies("development", "archive")
|
|
assert custody_class_satisfies("development", "development")
|
|
|
|
|
|
def test_readiness_reports_recovery_fields_for_operational_backend():
|
|
class _Operational(_BrokenBackend):
|
|
@property
|
|
def retention_policy(self):
|
|
return RetentionPolicy(
|
|
custody_class="operational",
|
|
retention_days=None,
|
|
immutable=True,
|
|
tamper_evidence=False,
|
|
durable=True,
|
|
recoverable_days=30,
|
|
recoverable_source="resource-control/data/capability/platform-audit-storage.json",
|
|
recoverable_basis="measured",
|
|
)
|
|
|
|
def health(self):
|
|
return None
|
|
|
|
status, body = invoke(
|
|
IngestionApplication(_Operational(), "opaque"),
|
|
None, path="/readyz", method="GET", body=b"",
|
|
)
|
|
assert status.startswith("200")
|
|
assert body["custody_class"] == "operational"
|
|
assert body["durable"] is True
|
|
assert body["recoverable_days"] == 30
|
|
assert body["recoverable_basis"] == "measured"
|
|
assert "platform-audit-storage" in body["recoverable_source"]
|
|
|
|
|
|
def test_counters_track_each_outcome(tmp_path):
|
|
app, _ = bound_app(tmp_path, may_read=True)
|
|
invoke(app, event()) # accepted
|
|
invoke(app, event()) # duplicate
|
|
invoke(app, event(subject="other")) # conflict
|
|
invoke(app, event(id="e2", tenant="tenant:coulomb"), key="e2") # rejected
|
|
invoke(app, event(), token="nope") # unauthorized
|
|
|
|
_, body = invoke(app, None, path="/v1/stats", method="GET", body=b"", token="operator")
|
|
counts = body["counts"]
|
|
assert counts["accepted"] == 1
|
|
assert counts["duplicate"] == 1
|
|
assert counts["conflict"] == 1
|
|
assert counts["rejected"] == 1
|
|
assert counts["unauthorized"] == 1
|
|
assert body["since"]
|
|
|
|
|
|
def test_stats_require_the_read_privilege(tmp_path):
|
|
app, _ = bound_app(tmp_path, may_read=False)
|
|
status, _ = invoke(app, None, path="/v1/stats", method="GET", body=b"")
|
|
assert status.startswith("403")
|
|
|
|
|
|
# --- read is tenant-scoped (AUDIT-WP-0008-T04) ------------------------------
|
|
|
|
def two_tenant_app(tmp_path):
|
|
"""A seeded store, plus a reader scoped to one of the two tenants.
|
|
|
|
The writer is unrestricted so both tenants exist; the reader is bound to
|
|
``binky`` only. Both applications share one backend, which is the point —
|
|
the boundary has to hold in the read path, not in the store.
|
|
"""
|
|
backend = SQLiteAuditBackend(str(tmp_path / "scoped.db"))
|
|
writer = SenderIdentity(
|
|
name="seeder", tokens=("seed",), sources=frozenset({"user-engine"}),
|
|
tenants=frozenset({"*"}), may_read=True,
|
|
)
|
|
reader = SenderIdentity(
|
|
name="scoped-reader", tokens=("scoped",), sources=frozenset({"user-engine"}),
|
|
tenants=frozenset({"tenant:friendly:binky"}), may_read=True,
|
|
)
|
|
seed = IngestionApplication(backend, SenderRegistry([writer]))
|
|
assert invoke(seed, event(), token="seed")[0].startswith("202")
|
|
assert invoke(
|
|
seed, event(id="evt-2", tenant="tenant:coulomb"), key="evt-2", token="seed"
|
|
)[0].startswith("202")
|
|
return IngestionApplication(backend, SenderRegistry([reader])), seed
|
|
|
|
|
|
def test_scoped_reader_cannot_fetch_another_tenants_event(tmp_path):
|
|
"""The defect AUDIT-WP-0008 found: may_read was the only gate."""
|
|
scoped, _ = two_tenant_app(tmp_path)
|
|
|
|
status, body = invoke(scoped, None, path="/v1/events/evt-1", method="GET",
|
|
body=b"", token="scoped")
|
|
assert status.startswith("200")
|
|
assert body["tenant"] == "tenant:friendly:binky"
|
|
|
|
status, body = invoke(scoped, None, path="/v1/events/evt-2", method="GET",
|
|
body=b"", token="scoped")
|
|
assert status.startswith("404")
|
|
assert body["error"] == "not_found"
|
|
|
|
|
|
def test_cross_tenant_refusal_is_indistinguishable_from_absence(tmp_path):
|
|
"""403 here would confirm the event exists and is someone else's."""
|
|
scoped, _ = two_tenant_app(tmp_path)
|
|
present = invoke(scoped, None, path="/v1/events/evt-2", method="GET",
|
|
body=b"", token="scoped")
|
|
absent = invoke(scoped, None, path="/v1/events/evt-nope", method="GET",
|
|
body=b"", token="scoped")
|
|
assert present == absent
|
|
|
|
|
|
def test_correlation_lookup_is_filtered_not_refused(tmp_path):
|
|
"""One correlation legitimately spans tenants; serve the readable slice."""
|
|
scoped, seed = two_tenant_app(tmp_path)
|
|
|
|
status, body = invoke_query(scoped, "correlation_id=corr-1", token="scoped")
|
|
assert status.startswith("200")
|
|
assert {e["event_id"] for e in body["events"]} == {"evt-1"}
|
|
|
|
status, body = invoke_query(seed, "correlation_id=corr-1", token="seed")
|
|
assert {e["event_id"] for e in body["events"]} == {"evt-1", "evt-2"}
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path", ["/v1/dead-letters", "/v1/stats", "/v1/secret-findings", "/v1/integrity"]
|
|
)
|
|
def test_unscoped_surfaces_require_full_tenant_scope(tmp_path, path):
|
|
"""Not tenant-keyed, so they cannot be filtered — refuse instead."""
|
|
scoped, seed = two_tenant_app(tmp_path)
|
|
|
|
status, body = invoke(scoped, None, path=path, method="GET", body=b"",
|
|
token="scoped", key=None)
|
|
assert status.startswith("403")
|
|
assert body["error"] == "full_tenant_scope_required"
|
|
|
|
status, _ = invoke(seed, None, path=path, method="GET", body=b"",
|
|
token="seed", key=None)
|
|
assert status.startswith("200")
|