tegwick
abd22fa0a6
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a025c2-407a-7a32-b40a-f37a52f03f62
606 lines
23 KiB
Python
606 lines
23 KiB
Python
import io
|
|
import json
|
|
from datetime import datetime, timezone
|
|
|
|
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"),
|
|
expires_at=kw.get("expires_at"),
|
|
)
|
|
# 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_expired_sender_is_unauthorized_at_http_boundary(tmp_path):
|
|
app, _ = bound_app(
|
|
tmp_path,
|
|
expires_at=datetime(2000, 1, 1, tzinfo=timezone.utc),
|
|
)
|
|
status, body = invoke(app, event())
|
|
assert status.startswith("401")
|
|
assert body["error"] == "unauthorized"
|
|
|
|
|
|
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")
|