AUDIT-WP-0013-T06: independent full-scope readers may count any source (NK-WP-0035-T04)

GET /v1/reconciliation now admits may_read + tenants ["*"] identities for
any source, marked basis: independent_reader. Scoped readers get 403
full_tenant_scope_required; writers keep 403 source_not_allowed. Counts only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: opus
Assistant-Process: 308701@bnt-lap001
Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56

tests/test_independent_reconciliation.py

@@ -0,0 +1,144 @@
+"""AUDIT-WP-0013-T06 — independent cross-source reconciliation counts.
+
+NK-WP-0035-T04: an operator reader registered for one source must be able to
+count another source's window, without payloads, and only with full tenant
+scope. A writer still may not count another source.
+"""
+
+import io
+import json
+
+import pytest
+
+from audit_core.ingestion import IngestionApplication
+from audit_core.senders import SenderIdentity, SenderRegistry
+from audit_core.sqlite_backend import SQLiteAuditBackend
+
+
+def invoke(app, path, *, token, method="GET", payload=None, key=None):
+    raw = json.dumps(payload).encode() if payload is not None else b""
+    environ = {
+        "PATH_INFO": path.split("?")[0],
+        "QUERY_STRING": path.split("?")[1] if "?" in path else "",
+        "REQUEST_METHOD": method,
+        "CONTENT_LENGTH": str(len(raw)),
+        "wsgi.input": io.BytesIO(raw),
+        "HTTP_AUTHORIZATION": f"Bearer {token}",
+    }
+    if key:
+        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 {})
+
+
+TENANT = "tenant:friendly:binky"
+INSTANCE = "b351fa6b-e0d6-4f88-b7c9-6bacb1a112b7"
+WINDOW = "&since=2026-09-01T00:00:00%2B00:00&until=2027-01-01T00:00:00%2B00:00"
+
+QONTO = SenderIdentity(
+    name="qonto-assistant", tokens=("qonto",), sources=frozenset({"qonto-assistant"}),
+    tenants=frozenset({TENANT}), may_write=True, may_read=False,
+)
+USER_ENGINE = SenderIdentity(
+    name="user-engine", tokens=("writer",), sources=frozenset({"user-engine"}),
+    tenants=frozenset({"*"}), may_write=True, may_read=False,
+)
+OPERATOR = SenderIdentity(
+    name="operator", tokens=("operator",), sources=frozenset({"user-engine"}),
+    tenants=frozenset({"*"}), may_write=False, may_read=True,
+)
+SCOPED_READER = SenderIdentity(
+    name="scoped-reader", tokens=("scoped",), sources=frozenset({"user-engine"}),
+    tenants=frozenset({TENANT}), may_write=False, may_read=True,
+)
+
+
+@pytest.fixture
+def app(tmp_path):
+    backend = SQLiteAuditBackend(str(tmp_path / "s.db"))
+    return IngestionApplication(
+        backend, SenderRegistry([QONTO, USER_ENGINE, OPERATOR, SCOPED_READER])
+    )
+
+
+def emit(app, event_id, action, *, instance=INSTANCE, secret=None):
+    data = {"stream_instance_id": instance, "sequence": 1}
+    if secret:
+        data["binding"] = secret
+    payload = {
+        "id": event_id, "type": action, "source": "qonto-assistant",
+        "subject": "s:1", "tenant": TENANT, "correlation_id": "c-1",
+        "occurred_at": "2026-10-15T19:40:00+00:00", "data": data,
+    }
+    status, _ = invoke(app, "/v1/events", token="qonto", method="POST",
+                       payload=payload, key=event_id)
+    assert status.startswith("202")
+
+
+def count(app, token, query):
+    return invoke(app, "/v1/reconciliation?" + query + WINDOW, token=token)
+
+
+def test_full_scope_reader_counts_another_source(app):
+    emit(app, "h-1", "audit.heartbeat")
+    emit(app, "h-2", "audit.heartbeat")
+    emit(app, "d-1", "audit.deny")
+    emit(app, "x-1", "audit.deny", instance="other-instance")
+    status, body = count(
+        app, "operator",
+        f"source=qonto-assistant&tenant={TENANT}&stream_instance_id={INSTANCE}",
+    )
+    assert status.startswith("200")
+    assert body["counts"] == [
+        {"class": "audit.deny", "count": 1},
+        {"class": "audit.heartbeat", "count": 2},
+    ]
+    assert body["basis"] == "independent_reader"
+    assert body["stream_instance_id"] == INSTANCE
+
+
+def test_full_scope_reader_may_omit_tenant(app):
+    emit(app, "d-1", "audit.deny")
+    status, body = count(app, "operator", "source=qonto-assistant")
+    assert status.startswith("200")
+    assert body["tenant"] is None
+    assert body["counts"] == [{"class": "audit.deny", "count": 1}]
+
+
+def test_independent_counts_never_carry_payloads(app):
+    emit(app, "d-1", "audit.deny", secret="value-not-to-leak")
+    _, body = count(app, "operator", f"source=qonto-assistant&tenant={TENANT}")
+    assert set(body) == {
+        "source", "tenant", "since", "until", "counts", "means", "basis",
+    }
+    assert "value-not-to-leak" not in json.dumps(body)
+
+
+def test_tenant_scoped_reader_may_not_count_another_source(app):
+    emit(app, "d-1", "audit.deny")
+    status, body = count(app, "scoped", f"source=qonto-assistant&tenant={TENANT}")
+    assert status.startswith("403")
+    assert body["error"] == "full_tenant_scope_required"
+
+
+def test_writer_still_may_not_count_another_source(app):
+    status, body = count(app, "writer", f"source=qonto-assistant&tenant={TENANT}")
+    assert status.startswith("403")
+    assert body["error"] == "source_not_allowed"
+
+
+def test_own_source_reconciliation_is_unchanged(app):
+    emit(app, "d-1", "audit.deny")
+    status, body = count(app, "qonto", f"source=qonto-assistant&tenant={TENANT}")
+    assert status.startswith("200")
+    assert "basis" not in body
+    assert body["counts"] == [{"class": "audit.deny", "count": 1}]
+
+
+def test_independent_reader_still_needs_a_bounded_window(app):
+    status, body = invoke(
+        app, "/v1/reconciliation?source=qonto-assistant", token="operator"
+    )
+    assert status.startswith("400")
+    assert body["error"] == "since_and_until_required"