Add governed institutional capability contracts
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
import unittest
|
||||
|
||||
from govoplan_core.core.encryption import (
|
||||
CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX,
|
||||
ContentProtectionEnvelope,
|
||||
KeyMaterialDescriptor,
|
||||
KeyVaultCreateRequest,
|
||||
key_material_provider,
|
||||
)
|
||||
|
||||
|
||||
NOW = datetime(2026, 8, 1, 12, 0, tzinfo=UTC)
|
||||
|
||||
|
||||
class Provider:
|
||||
def provision_key(self, _session, _principal, *, request):
|
||||
return KeyMaterialDescriptor(
|
||||
provider_id="test",
|
||||
provider_key_ref=f"test:{request.vault_id}:{request.key_version}",
|
||||
algorithm_suite=request.algorithm_suite,
|
||||
state="active",
|
||||
created_at=NOW,
|
||||
)
|
||||
|
||||
def revoke_key(self, _session, _principal, **_kwargs):
|
||||
raise NotImplementedError
|
||||
|
||||
def schedule_key_destruction(self, _session, _principal, **_kwargs):
|
||||
raise NotImplementedError
|
||||
|
||||
def key_status(self, _session, **_kwargs):
|
||||
return None
|
||||
|
||||
|
||||
class Registry:
|
||||
def __init__(self, values):
|
||||
self.values = values
|
||||
|
||||
def has_capability(self, name: str) -> bool:
|
||||
return name in self.values
|
||||
|
||||
def capability(self, name: str):
|
||||
return self.values[name]
|
||||
|
||||
|
||||
class EncryptionContractTests(unittest.TestCase):
|
||||
def test_provider_resolution_is_optional_and_structural(self) -> None:
|
||||
name = f"{CAPABILITY_ENCRYPTION_KEY_MATERIAL_PROVIDER_PREFIX}test"
|
||||
provider = Provider()
|
||||
self.assertIs(
|
||||
provider, key_material_provider(Registry({name: provider}), "test")
|
||||
)
|
||||
self.assertIsNone(key_material_provider(Registry({}), "test"))
|
||||
|
||||
def test_secret_bearing_policy_and_envelope_metadata_are_rejected(self) -> None:
|
||||
with self.assertRaisesRegex(ValueError, "secret material"):
|
||||
KeyVaultCreateRequest(
|
||||
tenant_id="tenant-1",
|
||||
vault_id="vault-1",
|
||||
name="Vault",
|
||||
provider_id="test",
|
||||
purpose="content",
|
||||
algorithm_suite="AES-256-GCM",
|
||||
scope_type="tenant",
|
||||
scope_id="tenant-1",
|
||||
policy_ref="policy:1",
|
||||
idempotency_key="create-1",
|
||||
provider_policy={"password": "forbidden"},
|
||||
)
|
||||
with self.assertRaisesRegex(ValueError, "secret material"):
|
||||
ContentProtectionEnvelope(
|
||||
envelope_id="envelope-1",
|
||||
tenant_id="tenant-1",
|
||||
owner_module="files",
|
||||
resource_type="file",
|
||||
resource_id="file-1",
|
||||
profile_kind="server_envelope",
|
||||
profile_id="profile-1",
|
||||
provider_id="test",
|
||||
vault_id="vault-1",
|
||||
key_version=1,
|
||||
algorithm_suite="AES-256-GCM",
|
||||
ciphertext_ref="object://ciphertext/1",
|
||||
ciphertext_digest="sha256:ciphertext",
|
||||
authenticated_context_digest="sha256:context",
|
||||
state="active",
|
||||
created_at=NOW,
|
||||
metadata={"private_key": "forbidden"},
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user