From 2b5c14385d0a3e612296cf6013caf5b7fdee1795 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Tue, 4 Aug 2026 14:01:05 +0200 Subject: [PATCH] Add Voting provider assurance contract --- src/govoplan_core/core/voting.py | 197 ++++++++++++++++++++++++++++++- tests/test_voting_contract.py | 67 ++++++++++- 2 files changed, 262 insertions(+), 2 deletions(-) diff --git a/src/govoplan_core/core/voting.py b/src/govoplan_core/core/voting.py index 04867b4..bffbc9d 100644 --- a/src/govoplan_core/core/voting.py +++ b/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", ] diff --git a/tests/test_voting_contract.py b/tests/test_voting_contract.py index bf4aaea..e0227a0 100644 --- a/tests/test_voting_contract.py +++ b/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( {