Add identity trust administration surfaces

tests/test_identity_trust.py

@@ -19,6 +19,7 @@ from govoplan_identity_trust.backend.db.models import (
     TrustKeyEpoch,
 )
 from govoplan_identity_trust.backend.service import (
+    IdentityTrustAccessDenied,
     IdentityTrustError,
     SqlIdentityTrustService,
     record_assurance_evidence,
@@ -36,9 +37,18 @@ class Principal:
         return scope in {
             "identity_trust:device:admin",
             "identity_trust:device:read_all",
+            "identity_trust:key_access:approve",
         }
 
 
+class RestrictedPrincipal:
+    tenant_id = "tenant-1"
+    account_id = "account-2"
+
+    def has(self, scope: str) -> bool:
+        return False
+
+
 def registration(**changes) -> DeviceKeyRegistration:
     values = {
         "tenant_id": "tenant-1",
@@ -220,6 +230,89 @@ class IdentityTrustTests(unittest.TestCase):
         self.assertTrue(allowed.allowed)
         self.assertFalse(stale.allowed)
 
+    def test_bounded_projections_enforce_subject_access_and_keep_provenance(self) -> None:
+        self.service.register_device_key(
+            self.session,
+            self.principal,
+            request=registration(),
+        )
+        record_assurance_evidence(
+            self.session,
+            self.principal,
+            tenant_id="tenant-1",
+            account_id="account-1",
+            evidence_ref="webauthn:assertion-2",
+            assurance_level="hardware",
+            provider_id="webauthn",
+            verified_at=NOW,
+            expires_at=NOW + timedelta(minutes=10),
+            device_key_id="key-1",
+            provenance={"ceremony": "uv", "policy_ref": "policy:assurance:1"},
+        )
+        epoch = self.service.rotate_epoch(
+            self.session,
+            self.principal,
+            request=KeyEpochRotationRequest(
+                tenant_id="tenant-1",
+                subject_kind="postbox",
+                subject_id="postbox-1",
+                reason="Initial epoch.",
+                access_decision_ref="access:grant-1",
+                idempotency_key="epoch-list-1",
+                previous_epoch=None,
+                history_policy="all_retained",
+            ),
+        )
+        self.service.decide_key_access(
+            self.session,
+            self.principal,
+            request=KeyAccessRequest(
+                tenant_id="tenant-1",
+                account_id="account-1",
+                device_key_id="key-1",
+                subject_kind="postbox",
+                subject_id="postbox-1",
+                key_epoch=epoch.epoch,
+                access_decision_ref="access:grant-1",
+                purpose="postbox.message.read",
+                requested_at=NOW,
+            ),
+        )
+
+        evidence = self.service.list_assurance_evidence(
+            self.session,
+            self.principal,
+            tenant_id="tenant-1",
+            account_id="account-1",
+        )
+        epochs = self.service.list_epochs(
+            self.session,
+            self.principal,
+            tenant_id="tenant-1",
+            subject_kind="postbox",
+            subject_id="postbox-1",
+        )
+        decisions = self.service.list_key_access_decisions(
+            self.session,
+            self.principal,
+            tenant_id="tenant-1",
+            account_id="account-1",
+        )
+
+        self.assertEqual(1, len(evidence))
+        self.assertEqual("policy:assurance:1", evidence[0].provenance["policy_ref"])
+        self.assertEqual((1,), tuple(item.epoch for item in epochs))
+        self.assertEqual(1, len(decisions))
+        self.assertFalse(decisions[0].provenance["cryptographic_material_released"])
+
+        with self.assertRaises(IdentityTrustAccessDenied):
+            self.service.list_assurance_evidence(
+                self.session,
+                RestrictedPrincipal(),
+                tenant_id="tenant-1",
+                account_id="account-1",
+            )
+
 
 if __name__ == "__main__":
     unittest.main()