Add Voting provider assurance contract

src/govoplan_core/core/voting.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import UTC, datetime
 import json
 from typing import Literal, Protocol, runtime_checkable
 
@@ -15,6 +15,20 @@ VOTING_ASSURANCE_CONFIDENTIAL = "confidential"
 VOTING_ASSURANCE_SECRET = "secret"
 VOTING_ASSURANCE_EXTERNAL_CERTIFIED = "external_certified"
 
+VOTING_CERTIFICATION_NOT_CERTIFIED = "not_certified"
+VOTING_CERTIFICATION_IN_EVALUATION = "in_evaluation"
+VOTING_CERTIFICATION_CERTIFIED = "certified"
+VOTING_CERTIFICATION_EXPIRED = "expired"
+VOTING_CERTIFICATION_REVOKED = "revoked"
+
+VotingProviderCertificationState = Literal[
+    "not_certified",
+    "in_evaluation",
+    "certified",
+    "expired",
+    "revoked",
+]
+
 
 class VotingCapabilityError(ValueError):
     """Stable error raised by Voting capability implementations."""
@@ -30,6 +44,119 @@ def voting_provider_capability(provider_id: str) -> str:
     return f"{CAPABILITY_VOTING_PROVIDER_PREFIX}{normalized}"
 
 
+@dataclass(frozen=True, slots=True)
+class VotingProviderAssuranceDeclaration:
+    """Pinned assurance and certification claim made by a Voting provider."""
+
+    provider_id: str
+    implementation_ref: str
+    supported_assurance_profiles: tuple[
+        Literal["confidential", "secret", "external_certified"], ...
+    ]
+    certification_state: VotingProviderCertificationState
+    protocol_ref: str
+    protocol_version: str
+    certification_authority: str | None = None
+    certification_reference: str | None = None
+    certification_evidence_ref: str | None = None
+    certification_valid_from: datetime | None = None
+    certification_valid_until: datetime | None = None
+    notes: tuple[str, ...] = ()
+
+    def __post_init__(self) -> None:
+        normalized_id = str(self.provider_id or "").strip().lower()
+        voting_provider_capability(normalized_id)
+        if normalized_id != self.provider_id:
+            raise ValueError("Voting provider assurance id must be normalized.")
+        for field_name in ("implementation_ref", "protocol_ref", "protocol_version"):
+            if not str(getattr(self, field_name) or "").strip():
+                raise ValueError(
+                    f"Voting provider assurance {field_name} is required."
+                )
+        profiles = tuple(self.supported_assurance_profiles)
+        allowed_profiles = {
+            VOTING_ASSURANCE_CONFIDENTIAL,
+            VOTING_ASSURANCE_SECRET,
+            VOTING_ASSURANCE_EXTERNAL_CERTIFIED,
+        }
+        if (
+            not profiles
+            or len(set(profiles)) != len(profiles)
+            or not set(profiles) <= allowed_profiles
+        ):
+            raise ValueError(
+                "Voting provider assurance profiles must be unique supported external profiles."
+            )
+        if self.certification_state not in {
+            VOTING_CERTIFICATION_NOT_CERTIFIED,
+            VOTING_CERTIFICATION_IN_EVALUATION,
+            VOTING_CERTIFICATION_CERTIFIED,
+            VOTING_CERTIFICATION_EXPIRED,
+            VOTING_CERTIFICATION_REVOKED,
+        }:
+            raise ValueError("Voting provider certification state is invalid.")
+        valid_from = _aware_datetime(
+            self.certification_valid_from,
+            field_name="certification_valid_from",
+        )
+        valid_until = _aware_datetime(
+            self.certification_valid_until,
+            field_name="certification_valid_until",
+        )
+        if valid_from and valid_until and valid_until <= valid_from:
+            raise ValueError(
+                "Voting provider certification validity must end after it starts."
+            )
+        if self.certification_state == VOTING_CERTIFICATION_CERTIFIED:
+            required = (
+                self.certification_authority,
+                self.certification_reference,
+                self.certification_evidence_ref,
+                valid_from,
+                valid_until,
+            )
+            if any(value is None or value == "" for value in required):
+                raise ValueError(
+                    "Certified Voting providers require authority, reference, evidence, and a validity window."
+                )
+        if len(self.notes) > 16 or any(not str(item or "").strip() for item in self.notes):
+            raise ValueError("Voting provider assurance notes must be bounded non-empty text.")
+
+    def is_currently_certified(self, *, at: datetime | None = None) -> bool:
+        if self.certification_state != VOTING_CERTIFICATION_CERTIFIED:
+            return False
+        moment = _aware_datetime(at or datetime.now(UTC), field_name="at")
+        valid_from = _aware_datetime(
+            self.certification_valid_from,
+            field_name="certification_valid_from",
+        )
+        valid_until = _aware_datetime(
+            self.certification_valid_until,
+            field_name="certification_valid_until",
+        )
+        return bool(valid_from and valid_until and valid_from <= moment < valid_until)
+
+    def to_dict(self) -> dict[str, object]:
+        return {
+            "provider_id": self.provider_id,
+            "implementation_ref": self.implementation_ref,
+            "supported_assurance_profiles": list(self.supported_assurance_profiles),
+            "certification_state": self.certification_state,
+            "protocol_ref": self.protocol_ref,
+            "protocol_version": self.protocol_version,
+            "certification_authority": self.certification_authority,
+            "certification_reference": self.certification_reference,
+            "certification_evidence_ref": self.certification_evidence_ref,
+            "certification_valid_from": _datetime_text(
+                self.certification_valid_from
+            ),
+            "certification_valid_until": _datetime_text(
+                self.certification_valid_until
+            ),
+            "notes": list(self.notes),
+        }
+
+
 @dataclass(frozen=True, slots=True)
 class VotingOption:
     key: str
@@ -184,6 +311,8 @@ class ExternalVotingProvider(Protocol):
     provider credentials must not cross this boundary.
     """
 
+    def assurance_declaration(self) -> VotingProviderAssuranceDeclaration: ...
+
     def finalize_ballot(
         self,
         session: object,
@@ -283,6 +412,45 @@ class VotingBallotProvider(Protocol):
     ) -> VotingBallotRef: ...
 
 
+def require_voting_provider_assurance(
+    provider: object,
+    *,
+    provider_id: str,
+    assurance_profile: str,
+    at: datetime | None = None,
+) -> VotingProviderAssuranceDeclaration:
+    """Validate and return the provider claim required for a frozen ballot."""
+
+    if not isinstance(provider, ExternalVotingProvider):
+        raise VotingCapabilityError("Voting provider does not implement the contract.")
+    try:
+        declaration = provider.assurance_declaration()
+    except (TypeError, ValueError) as exc:
+        raise VotingCapabilityError(
+            "Voting provider assurance declaration was rejected."
+        ) from exc
+    if not isinstance(declaration, VotingProviderAssuranceDeclaration):
+        raise VotingCapabilityError(
+            "Voting provider returned an invalid assurance declaration."
+        )
+    normalized_provider_id = str(provider_id or "").strip().lower()
+    if declaration.provider_id != normalized_provider_id:
+        raise VotingCapabilityError(
+            "Voting provider assurance declaration does not match the selected provider."
+        )
+    if assurance_profile not in declaration.supported_assurance_profiles:
+        raise VotingCapabilityError(
+            "Voting provider does not support the selected assurance profile."
+        )
+    if (
+        assurance_profile == VOTING_ASSURANCE_EXTERNAL_CERTIFIED
+        and not declaration.is_currently_certified(at=at)
+    ):
+        raise VotingCapabilityError(
+            "Externally certified Voting requires a currently valid provider certification."
+        )
+    return declaration
+
 def _validate_provider_evidence(
     evidence: Sequence[Mapping[str, object]],
 ) -> None:
@@ -330,6 +498,25 @@ def _reject_sensitive_evidence(value: object) -> None:
             _reject_sensitive_evidence(nested)
 
 
+def _aware_datetime(
+    value: datetime | None,
+    *,
+    field_name: str,
+) -> datetime | None:
+    if value is None:
+        return None
+    if value.tzinfo is None or value.utcoffset() is None:
+        raise ValueError(
+            f"Voting provider assurance {field_name} must be timezone-aware."
+        )
+    return value.astimezone(UTC)
+
+
+def _datetime_text(value: datetime | None) -> str | None:
+    aware = _aware_datetime(value, field_name="datetime")
+    return aware.isoformat() if aware is not None else None
+
+
 __all__ = [
     "CAPABILITY_VOTING_BALLOTS",
     "CAPABILITY_VOTING_PROVIDER_PREFIX",
@@ -343,6 +530,11 @@ __all__ = [
     "VOTING_ASSURANCE_EXTERNAL_CERTIFIED",
     "VOTING_ASSURANCE_RECORDED",
     "VOTING_ASSURANCE_SECRET",
+    "VOTING_CERTIFICATION_CERTIFIED",
+    "VOTING_CERTIFICATION_EXPIRED",
+    "VOTING_CERTIFICATION_IN_EVALUATION",
+    "VOTING_CERTIFICATION_NOT_CERTIFIED",
+    "VOTING_CERTIFICATION_REVOKED",
     "VotingBallotCreateCommand",
     "VotingBallotProvider",
     "VotingBallotRef",
@@ -350,7 +542,10 @@ __all__ = [
     "VotingCastCommand",
     "VotingElector",
     "VotingOption",
+    "VotingProviderAssuranceDeclaration",
+    "VotingProviderCertificationState",
     "VotingReceipt",
     "VotingResult",
+    "require_voting_provider_assurance",
     "voting_provider_capability",
 ]

tests/test_voting_contract.py

@@ -1,8 +1,16 @@
 from __future__ import annotations
 
+from datetime import UTC, datetime, timedelta
 import unittest
 
-from govoplan_core.core.voting import VotingResult
+from govoplan_core.core.voting import (
+    VOTING_CERTIFICATION_CERTIFIED,
+    VOTING_CERTIFICATION_IN_EVALUATION,
+    VotingCapabilityError,
+    VotingProviderAssuranceDeclaration,
+    VotingResult,
+    require_voting_provider_assurance,
+)
 
 
 def result_with_evidence(*evidence):
@@ -23,7 +31,64 @@ def result_with_evidence(*evidence):
     )
 
 
+class FakeProvider:
+    def __init__(self, declaration: VotingProviderAssuranceDeclaration) -> None:
+        self.declaration = declaration
+
+    def assurance_declaration(self) -> VotingProviderAssuranceDeclaration:
+        return self.declaration
+
+    def finalize_ballot(self, session, principal, *, request):
+        raise NotImplementedError
+
+
 class VotingContractTests(unittest.TestCase):
+    def test_external_certification_requires_current_evidence_backed_claim(self) -> None:
+        now = datetime.now(UTC)
+        declaration = VotingProviderAssuranceDeclaration(
+            provider_id="certified_provider",
+            implementation_ref="certified-provider/adapter@1",
+            supported_assurance_profiles=("external_certified",),
+            certification_state=VOTING_CERTIFICATION_CERTIFIED,
+            protocol_ref="vendor:certified-ballot",
+            protocol_version="3.0",
+            certification_authority="Independent authority",
+            certification_reference="certificate-2026-1",
+            certification_evidence_ref="evidence://certificate-2026-1",
+            certification_valid_from=now - timedelta(days=1),
+            certification_valid_until=now + timedelta(days=1),
+        )
+
+        selected = require_voting_provider_assurance(
+            FakeProvider(declaration),
+            provider_id="certified_provider",
+            assurance_profile="external_certified",
+            at=now,
+        )
+
+        self.assertEqual("certificate-2026-1", selected.certification_reference)
+        self.assertEqual(
+            (now - timedelta(days=1)).isoformat(),
+            selected.to_dict()["certification_valid_from"],
+        )
+
+    def test_external_certification_rejects_evaluation_only_provider(self) -> None:
+        declaration = VotingProviderAssuranceDeclaration(
+            provider_id="candidate_provider",
+            implementation_ref="candidate-provider/adapter@1",
+            supported_assurance_profiles=("external_certified",),
+            certification_state=VOTING_CERTIFICATION_IN_EVALUATION,
+            protocol_ref="vendor:candidate-ballot",
+            protocol_version="1.0",
+        )
+
+        with self.assertRaisesRegex(VotingCapabilityError, "currently valid"):
+            require_voting_provider_assurance(
+                FakeProvider(declaration),
+                provider_id="candidate_provider",
+                assurance_profile="external_certified",
+            )
+
     def test_accepts_sanitized_provider_evidence(self) -> None:
         value = result_with_evidence(
             {