Retain verified tenant provenance and reconcile production blockers

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a1046e-fdbf-7b93-a267-5f72b9b24526

tests/test_human_control.py

@@ -144,7 +144,7 @@ def test_legacy_v4_rows_are_not_reclassified_from_human_entries(tmp_path):
     assert survivor.entries[0]["principal_type"] == "human"
     assert survivor.human_control is False
     assert upgraded.claim(obj.id)["binding"]["human_control"] is False
-    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 5
+    assert upgraded.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 6
     upgraded.close()
 
 

tests/test_tenant_provenance.py

@@ -0,0 +1,97 @@
+"""Verified tenant origin remains evidence, including absence and legacy rows."""
+import json
+import sqlite3
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric import rsa
+
+from approval_engine.api import App, call
+from approval_engine.auth import JWTAuthenticator
+from approval_engine.errors import StoreUnavailable
+from approval_engine.store import Engine, LATEST_SCHEMA_VERSION
+from tests.conftest import FROZEN, binding, validity
+from tests.test_auth import _JWKS, _jwt
+
+
+def authenticator():
+    private = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    auth = JWTAuthenticator(
+        issuer="https://keycape.example", audience="approval-engine",
+        jwks_url="https://keycape.example/jwks", jwks_client=_JWKS(private.public_key()),
+    )
+    return private, auth
+
+
+@pytest.mark.parametrize("source", ["directory", "registration", "default", None])
+def test_verified_provenance_survives_http_entry_restart_and_use(tmp_path, source):
+    private, auth = authenticator()
+    claims = dict(sub="human", tenant="tenant:platform", principal_type="human",
+                  scope="approval:approve")
+    if source is not None:
+        claims["tenant_source"] = source
+    token = _jwt(private, **claims)
+    path = tmp_path / "approval.sqlite"
+    engine = Engine(path, clock=lambda: FROZEN)
+    obj = engine.create(binding(), validity(), human_control=True)
+    digest = obj.binding_digest
+    status, approved = call(App(engine, auth), "POST", f"/v1/approvals/{obj.id}/entries",
+                            {"tenant_source": "spoofed"}, authorization="Bearer " + token)
+    assert status == 200
+    assert approved["entries"][0]["tenant_source"] == source
+    engine.close()
+    engine = Engine(path, clock=lambda: FROZEN, auto_migrate=False)
+    assert engine.get(obj.id).entries[0]["tenant_source"] == source
+    assert engine.get(obj.id).binding_digest == digest
+    assert "tenant_source" not in json.dumps(engine.claim(obj.id))
+    engine.consume(obj.id, digest)
+    rows = engine._conn().execute("SELECT class, payload_json FROM outbox").fetchall()
+    for row in rows:
+        payload = json.loads(row["payload_json"])
+        assert payload["details"]["threshold"]["approvers"][0]["tenant_source"] == source
+    assert {row["class"] for row in rows} == {"issuance", "use"}
+    engine.close()
+
+
+@pytest.mark.parametrize("source", [None, "unknown", "", {}, [], 1])
+def test_explicit_invalid_provenance_refuses_http_bind_without_mutation(engine, source):
+    private, auth = authenticator()
+    token = _jwt(private, sub="human", tenant="tenant:platform", principal_type="human",
+                 scope="approval:approve", tenant_source=source)
+    obj = engine.create(binding(), validity(), human_control=True)
+    status, _ = call(App(engine, auth), "POST", f"/v1/approvals/{obj.id}/entries", {},
+                     authorization="Bearer " + token)
+    assert status == 401
+    assert engine.get(obj.id).entries == []
+    assert engine._conn().execute("SELECT COUNT(*) FROM outbox").fetchone()[0] == 0
+
+
+def test_v5_migration_preserves_unknown_provenance_and_existing_evidence(tmp_path):
+    path = tmp_path / "approval.sqlite"
+    engine = Engine(path, clock=lambda: FROZEN)
+    obj = engine.create(binding(), validity(), human_control=True)
+    engine.add_entry(obj.id, "human", principal_type="human")
+    digest = obj.binding_digest
+    evidence = engine._conn().execute("SELECT payload_json FROM outbox").fetchall()[0][0]
+    engine.close()
+    with sqlite3.connect(path) as conn:
+        conn.execute("ALTER TABLE entries DROP COLUMN tenant_source")
+        conn.execute("PRAGMA user_version=5")
+    with pytest.raises(StoreUnavailable, match="migrate"):
+        Engine(path, auto_migrate=False)
+    engine = Engine(path, clock=lambda: FROZEN)
+    assert engine.storage_status()["schema_version"] == LATEST_SCHEMA_VERSION == 6
+    assert engine.get(obj.id).entries[0]["tenant_source"] is None
+    assert engine.get(obj.id).binding_digest == digest
+    assert engine.get(obj.id).human_control is True
+    assert engine._conn().execute("SELECT payload_json FROM outbox").fetchone()[0] == evidence
+    engine.close()
+
+
+def test_provenance_rolls_back_with_failed_outbox(engine):
+    obj = engine.create(binding(), validity(), human_control=True)
+    engine.fail_outbox = True
+    with pytest.raises(StoreUnavailable):
+        engine.add_entry(obj.id, "human", principal_type="human", tenant_source="directory")
+    engine.fail_outbox = False
+    assert engine.get(obj.id).entries == []
+    assert engine.get(obj.id).status == "requested"