From b5f5be15f68a1d670acabf88d35337ffc52d5f6e Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Thu, 6 Aug 2026 12:42:20 +0200 Subject: [PATCH] Add governed form evidence contract --- docs/RECORDS_FILING_CONTRACT.md | 10 ++ src/govoplan_core/core/form_evidence.py | 227 ++++++++++++++++++++++++ tests/test_form_evidence_contract.py | 122 +++++++++++++ 3 files changed, 359 insertions(+) create mode 100644 src/govoplan_core/core/form_evidence.py create mode 100644 tests/test_form_evidence_contract.py diff --git a/docs/RECORDS_FILING_CONTRACT.md b/docs/RECORDS_FILING_CONTRACT.md index 7c82406..276ae33 100644 --- a/docs/RECORDS_FILING_CONTRACT.md +++ b/docs/RECORDS_FILING_CONTRACT.md @@ -82,3 +82,13 @@ handling. The simulation is explicitly non-conformant, transfers no custody, and cannot be used as evidence of an archive handoff. A real provider requires a selected target/profile, provider-specific recovery declaration, and target test evidence. + +## Form Evidence Boundary + +Form attachments use the separate provider-neutral contract in +`govoplan_core.core.form_evidence`. Forms Runtime requests short-lived, +purpose-bound upload grants and re-inspects the exact provider-owned evidence +before final submission. The provider keeps byte storage, quarantine, +classification, and retention ownership; Forms Runtime stores only immutable +evidence references and bounded verification results. This contract is not an +alternative path for Records filing or archive custody. diff --git a/src/govoplan_core/core/form_evidence.py b/src/govoplan_core/core/form_evidence.py new file mode 100644 index 0000000..ae68edc --- /dev/null +++ b/src/govoplan_core/core/form_evidence.py @@ -0,0 +1,227 @@ +from __future__ import annotations + +from collections.abc import Mapping, Sequence +from dataclasses import dataclass, field +from datetime import datetime +from typing import Literal, Protocol, runtime_checkable + +from govoplan_core.core.institutional import EvidenceReference, InstitutionalReference + + +CAPABILITY_FORM_EVIDENCE_PREFIX = "forms_runtime.evidence." + +FormEvidenceState = Literal[ + "accepted", + "pending", + "rejected", + "expired", + "revoked", + "unavailable", +] + +_FORM_EVIDENCE_STATES = { + "accepted", + "pending", + "rejected", + "expired", + "revoked", + "unavailable", +} + + +class FormEvidenceContractError(ValueError): + """Stable error for provider-neutral Form evidence operations.""" + + +@dataclass(frozen=True, slots=True) +class FormEvidenceGrantRequest: + """Request a short-lived, purpose-bound grant from an evidence owner.""" + + tenant_id: str + instance_id: str + definition_ref: InstitutionalReference + evidence_kind: str + purpose: str + idempotency_key: str + expires_at: datetime + custodian_ref: str | None = None + max_size_bytes: int | None = None + allowed_content_types: tuple[str, ...] = () + metadata: Mapping[str, object] = field(default_factory=dict) + + def __post_init__(self) -> None: + _require_text(self.tenant_id, "Form evidence tenant") + _require_text(self.instance_id, "Form evidence instance") + _require_text(self.evidence_kind, "Form evidence kind") + _require_text(self.purpose, "Form evidence purpose") + _require_text(self.idempotency_key, "Form evidence idempotency key") + if ( + self.definition_ref.kind != "form" + or self.definition_ref.tenant_id != self.tenant_id + or not self.definition_ref.version + ): + raise FormEvidenceContractError( + "Form evidence grants require an exact same-tenant Form definition." + ) + if self.expires_at.tzinfo is None or self.expires_at.utcoffset() is None: + raise FormEvidenceContractError( + "Form evidence grant expiry must include a timezone." + ) + if self.max_size_bytes is not None and self.max_size_bytes <= 0: + raise FormEvidenceContractError( + "Form evidence grant size limits must be positive." + ) + if any(not item.strip() for item in self.allowed_content_types): + raise FormEvidenceContractError( + "Form evidence content types cannot be empty." + ) + + +@dataclass(frozen=True, slots=True) +class FormEvidenceGrant: + provider_id: str + grant_id: str + upload_token: str | None + upload_url: str + expires_at: datetime + max_size_bytes: int + allowed_content_types: tuple[str, ...] = () + replayed: bool = False + + def __post_init__(self) -> None: + for value, label in ( + (self.provider_id, "Form evidence provider"), + (self.grant_id, "Form evidence grant"), + (self.upload_url, "Form evidence upload URL"), + ): + _require_text(value, label) + if self.upload_token is not None: + _require_text(self.upload_token, "Form evidence upload token") + if self.max_size_bytes <= 0: + raise FormEvidenceContractError( + "Form evidence grant size limits must be positive." + ) + if self.expires_at.tzinfo is None or self.expires_at.utcoffset() is None: + raise FormEvidenceContractError( + "Form evidence grant expiry must include a timezone." + ) + + +@dataclass(frozen=True, slots=True) +class FormEvidenceInspectionRequest: + tenant_id: str + instance_id: str + definition_ref: InstitutionalReference + evidence: EvidenceReference + purpose: str + final: bool + + def __post_init__(self) -> None: + _require_text(self.tenant_id, "Form evidence tenant") + _require_text(self.instance_id, "Form evidence instance") + _require_text(self.purpose, "Form evidence purpose") + if self.definition_ref.tenant_id != self.tenant_id: + raise FormEvidenceContractError( + "Form evidence inspection cannot cross tenants." + ) + if self.evidence.tenant_id != self.tenant_id: + raise FormEvidenceContractError( + "Form evidence inspection cannot cross tenants." + ) + + +@dataclass(frozen=True, slots=True) +class FormEvidenceInspection: + provider_id: str + reference: EvidenceReference + state: FormEvidenceState + observed_at: datetime + retryable: bool = False + reason: str | None = None + metadata: Mapping[str, object] = field(default_factory=dict) + + def __post_init__(self) -> None: + _require_text(self.provider_id, "Form evidence provider") + if self.state not in _FORM_EVIDENCE_STATES: + raise FormEvidenceContractError( + f"Unsupported Form evidence state: {self.state!r}." + ) + if self.observed_at.tzinfo is None or self.observed_at.utcoffset() is None: + raise FormEvidenceContractError( + "Form evidence inspection time must include a timezone." + ) + if self.state == "accepted" and self.retryable: + raise FormEvidenceContractError( + "Accepted Form evidence cannot require a retry." + ) + + @property + def accepted(self) -> bool: + return self.state == "accepted" + + +@runtime_checkable +class FormEvidenceProvider(Protocol): + provider_id: str + + def supported_kinds(self) -> Sequence[str]: ... + + def create_upload_grant( + self, + session: object, + principal: object, + *, + request: FormEvidenceGrantRequest, + ) -> FormEvidenceGrant: ... + + def inspect_evidence( + self, + session: object, + principal: object, + *, + request: FormEvidenceInspectionRequest, + ) -> FormEvidenceInspection: ... + + +def form_evidence_capability(provider_id: str) -> str: + normalized = str(provider_id or "").strip().lower().replace("-", "_") + if not normalized or not normalized.replace("_", "").isalnum(): + raise FormEvidenceContractError("Invalid Form evidence provider id.") + return f"{CAPABILITY_FORM_EVIDENCE_PREFIX}{normalized}" + + +def form_evidence_provider( + registry: object | None, + provider_id: str, +) -> FormEvidenceProvider | None: + capability_name = form_evidence_capability(provider_id) + if registry is None or not hasattr(registry, "has_capability"): + return None + if not registry.has_capability(capability_name): + return None + if hasattr(registry, "require_capability"): + provider = registry.require_capability(capability_name) + elif hasattr(registry, "capability"): + provider = registry.capability(capability_name) + else: + return None + return provider if isinstance(provider, FormEvidenceProvider) else None + + +def _require_text(value: str, label: str) -> None: + if not isinstance(value, str) or not value.strip(): + raise FormEvidenceContractError(f"{label} is required.") + + +__all__ = [ + "CAPABILITY_FORM_EVIDENCE_PREFIX", + "FormEvidenceContractError", + "FormEvidenceGrant", + "FormEvidenceGrantRequest", + "FormEvidenceInspection", + "FormEvidenceInspectionRequest", + "FormEvidenceProvider", + "FormEvidenceState", + "form_evidence_capability", + "form_evidence_provider", +] diff --git a/tests/test_form_evidence_contract.py b/tests/test_form_evidence_contract.py new file mode 100644 index 0000000..780d2ae --- /dev/null +++ b/tests/test_form_evidence_contract.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +import unittest + +from govoplan_core.core.form_evidence import ( + FormEvidenceContractError, + FormEvidenceGrantRequest, + FormEvidenceInspection, + FormEvidenceInspectionRequest, + form_evidence_capability, + form_evidence_provider, +) +from govoplan_core.core.institutional import EvidenceReference, InstitutionalReference + + +NOW = datetime(2026, 8, 7, 9, 0, tzinfo=UTC) + + +class Provider: + provider_id = "files" + + def supported_kinds(self): + return ("document",) + + def create_upload_grant(self, session, principal, *, request): + raise NotImplementedError + + def inspect_evidence(self, session, principal, *, request): + return FormEvidenceInspection( + provider_id=self.provider_id, + reference=request.evidence, + state="accepted", + observed_at=NOW, + ) + + +class Registry: + def __init__(self) -> None: + self.provider = Provider() + + def has_capability(self, name: str) -> bool: + return name == "forms_runtime.evidence.files" + + def require_capability(self, name: str): + return self.provider + + +class FormEvidenceContractTests(unittest.TestCase): + def setUp(self) -> None: + self.definition = InstitutionalReference( + kind="form", + owner_module="forms", + object_id="permit", + tenant_id="tenant-1", + version="2", + ) + self.evidence = EvidenceReference( + kind="document", + owner_module="files", + evidence_id="asset-1", + tenant_id="tenant-1", + version="version-3", + checksum="a" * 64, + ) + + def test_grants_and_inspections_are_exact_and_tenant_bound(self) -> None: + grant = FormEvidenceGrantRequest( + tenant_id="tenant-1", + instance_id="form-1", + definition_ref=self.definition, + evidence_kind="document", + purpose="attach application evidence", + idempotency_key="grant-1", + expires_at=NOW + timedelta(minutes=10), + max_size_bytes=1024, + ) + inspection = FormEvidenceInspectionRequest( + tenant_id="tenant-1", + instance_id="form-1", + definition_ref=self.definition, + evidence=self.evidence, + purpose="submit application", + final=True, + ) + + self.assertEqual("2", grant.definition_ref.version) + self.assertEqual("version-3", inspection.evidence.version) + + def test_cross_tenant_evidence_is_rejected(self) -> None: + with self.assertRaisesRegex(FormEvidenceContractError, "cross tenants"): + FormEvidenceInspectionRequest( + tenant_id="tenant-2", + instance_id="form-1", + definition_ref=self.definition, + evidence=self.evidence, + purpose="submit", + final=True, + ) + + def test_provider_resolution_is_optional_and_runtime_checked(self) -> None: + registry = Registry() + self.assertEqual( + "forms_runtime.evidence.files", + form_evidence_capability("files"), + ) + self.assertIs(registry.provider, form_evidence_provider(registry, "files")) + self.assertIsNone(form_evidence_provider(registry, "missing")) + + def test_accepted_evidence_is_not_retryable(self) -> None: + with self.assertRaisesRegex(FormEvidenceContractError, "cannot require"): + FormEvidenceInspection( + provider_id="files", + reference=self.evidence, + state="accepted", + observed_at=NOW, + retryable=True, + ) + + +if __name__ == "__main__": + unittest.main()