777 lines
29 KiB
Python
777 lines
29 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import asdict
|
|
from datetime import UTC, datetime
|
|
import hashlib
|
|
import json
|
|
from typing import Any
|
|
import uuid
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_core.core.encryption import (
|
|
ContentProtectionRequest,
|
|
ContentUnprotectionRequest,
|
|
KeyVaultCreateRequest,
|
|
encryption_content_cipher,
|
|
encryption_key_vault,
|
|
)
|
|
from govoplan_core.core.voting import (
|
|
ExternalVotingBallotRef,
|
|
ExternalVotingCastRequest,
|
|
ExternalVotingFinalizationRequest,
|
|
ExternalVotingPreparationRequest,
|
|
VOTING_CERTIFICATION_NOT_CERTIFIED,
|
|
VotingReceipt,
|
|
VotingResult,
|
|
VotingProviderAssuranceDeclaration,
|
|
)
|
|
from govoplan_voting.backend.db.models import (
|
|
VotingConfidentialBallot,
|
|
VotingConfidentialCast,
|
|
)
|
|
|
|
|
|
LOCAL_CONFIDENTIAL_PROVIDER_ID = "local_confidential"
|
|
LOCAL_KEY_PROVIDER_ID = "local_aesgcm"
|
|
|
|
|
|
class LocalConfidentialVotingError(ValueError):
|
|
pass
|
|
|
|
|
|
class LocalConfidentialVotingProvider:
|
|
"""Reference confidential provider backed by encrypted shared SQL state.
|
|
|
|
The provider conceals selections at rest from ordinary Voting storage. It is
|
|
server-decryptable and intentionally makes no secret-ballot, coercion-
|
|
resistance, anonymity, HSM, or external-certification claim.
|
|
"""
|
|
|
|
def __init__(self, registry: object | None) -> None:
|
|
self._registry = registry
|
|
|
|
def assurance_declaration(self) -> VotingProviderAssuranceDeclaration:
|
|
return VotingProviderAssuranceDeclaration(
|
|
provider_id=LOCAL_CONFIDENTIAL_PROVIDER_ID,
|
|
implementation_ref="govoplan-voting/local-confidential@1",
|
|
supported_assurance_profiles=("confidential",),
|
|
certification_state=VOTING_CERTIFICATION_NOT_CERTIFIED,
|
|
protocol_ref="govoplan:voting:local-confidential",
|
|
protocol_version="1.0",
|
|
notes=(
|
|
"Server-readable reference provider; no anonymity, secrecy, coercion-resistance, or certification claim.",
|
|
),
|
|
)
|
|
|
|
def prepare_ballot(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
request: ExternalVotingPreparationRequest,
|
|
) -> ExternalVotingBallotRef:
|
|
db = _session(session)
|
|
_require_tenant(principal, request.tenant_id)
|
|
if request.assurance_profile != "confidential":
|
|
raise LocalConfidentialVotingError(
|
|
"The local confidential provider supports only the confidential assurance profile."
|
|
)
|
|
provider_ref = (
|
|
str(request.requested_provider_ballot_ref or "").strip()
|
|
or f"local-confidential:{request.ballot_id}"
|
|
)
|
|
payload = _preparation_payload(request, provider_ref=provider_ref)
|
|
digest = _digest(payload)
|
|
existing = (
|
|
db.query(VotingConfidentialBallot)
|
|
.filter(
|
|
VotingConfidentialBallot.tenant_id == request.tenant_id,
|
|
VotingConfidentialBallot.provider_ballot_ref == provider_ref,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if existing is not None:
|
|
if existing.preparation_request_sha256 != digest:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider ballot reference was reused for another definition."
|
|
)
|
|
return self._ballot_ref(existing, replayed=True)
|
|
|
|
cipher = encryption_content_cipher(self._registry)
|
|
key_vault = encryption_key_vault(self._registry)
|
|
if cipher is None or key_vault is None:
|
|
raise LocalConfidentialVotingError(
|
|
"The local confidential provider requires the Encryption key-vault and content-cipher capabilities."
|
|
)
|
|
vault_id = f"voting-{request.ballot_id}"
|
|
vault = key_vault.get_vault(
|
|
db,
|
|
tenant_id=request.tenant_id,
|
|
vault_id=vault_id,
|
|
)
|
|
if vault is None:
|
|
vault = key_vault.create_vault(
|
|
db,
|
|
principal,
|
|
request=KeyVaultCreateRequest(
|
|
tenant_id=request.tenant_id,
|
|
vault_id=vault_id,
|
|
name=f"Confidential ballot {request.ballot_id}",
|
|
provider_id=LOCAL_KEY_PROVIDER_ID,
|
|
purpose="voting.confidential.ballot",
|
|
algorithm_suite="AES-256-GCM",
|
|
scope_type="voting_ballot",
|
|
scope_id=request.ballot_id,
|
|
policy_ref="voting:local-confidential:v1",
|
|
idempotency_key=f"voting-vault:{request.ballot_id}",
|
|
recovery_quorum=2,
|
|
profile_kind="server_envelope",
|
|
),
|
|
)
|
|
if vault.state != "active" or vault.current_key is None:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential ballot Encryption vault is not active."
|
|
)
|
|
row = VotingConfidentialBallot(
|
|
tenant_id=request.tenant_id,
|
|
provider_ballot_ref=provider_ref,
|
|
ballot_id=request.ballot_id,
|
|
definition_sha256=request.definition_sha256,
|
|
electorate_sha256=request.electorate_sha256,
|
|
assurance_profile=request.assurance_profile,
|
|
method=request.method,
|
|
state="open",
|
|
options=[asdict(item) for item in request.options],
|
|
electorate=[
|
|
{
|
|
"subject_id": item.subject_id,
|
|
"label": item.label,
|
|
"weight": item.weight,
|
|
"provenance": dict(item.provenance),
|
|
}
|
|
for item in request.electorate
|
|
],
|
|
allow_replacement=request.allow_replacement,
|
|
quorum_weight=request.quorum_weight,
|
|
threshold_numerator=request.threshold_numerator,
|
|
threshold_denominator=request.threshold_denominator,
|
|
opens_at=request.opens_at,
|
|
closes_at=request.closes_at,
|
|
vault_id=vault_id,
|
|
preparation_idempotency_key=request.idempotency_key,
|
|
preparation_request_sha256=digest,
|
|
prepared_at=request.requested_at,
|
|
)
|
|
db.add(row)
|
|
db.flush()
|
|
return self._ballot_ref(row)
|
|
|
|
def cast_ballot(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
request: ExternalVotingCastRequest,
|
|
) -> VotingReceipt:
|
|
db = _session(session)
|
|
_require_tenant(principal, request.tenant_id)
|
|
actor_id = _principal_actor(principal)
|
|
if not actor_id or actor_id != request.elector_id:
|
|
raise LocalConfidentialVotingError(
|
|
"A confidential ballot principal cannot cast for another elector."
|
|
)
|
|
ballot = self._ballot(
|
|
db,
|
|
tenant_id=request.tenant_id,
|
|
provider_ballot_ref=request.provider_ballot_ref,
|
|
lock=True,
|
|
)
|
|
self._validate_binding(
|
|
ballot,
|
|
ballot_id=request.ballot_id,
|
|
definition_sha256=request.definition_sha256,
|
|
electorate_sha256=request.electorate_sha256,
|
|
)
|
|
if ballot.state != "open":
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider ballot is not open."
|
|
)
|
|
_validate_window(ballot)
|
|
electorate = {str(item["subject_id"]): item for item in ballot.electorate}
|
|
elector = electorate.get(request.elector_id)
|
|
if elector is None:
|
|
raise LocalConfidentialVotingError(
|
|
"The current principal is not in the confidential provider electorate."
|
|
)
|
|
selections = tuple(str(item).strip() for item in request.selections)
|
|
_validate_selections(ballot, selections)
|
|
if request.allow_replacement != ballot.allow_replacement:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider replacement policy does not match the frozen ballot."
|
|
)
|
|
cipher = encryption_content_cipher(self._registry)
|
|
if cipher is None:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider cannot cast because Encryption is unavailable."
|
|
)
|
|
replay = (
|
|
db.query(VotingConfidentialCast)
|
|
.filter(
|
|
VotingConfidentialCast.tenant_id == request.tenant_id,
|
|
VotingConfidentialCast.provider_ballot_id == ballot.id,
|
|
VotingConfidentialCast.idempotency_key == request.idempotency_key,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
if replay is not None:
|
|
opened = cipher.unprotect_content(
|
|
db,
|
|
request=ContentUnprotectionRequest(
|
|
tenant_id=request.tenant_id,
|
|
owner_module="voting",
|
|
resource_type="voting_confidential_cast",
|
|
resource_id=replay.encryption_resource_id,
|
|
envelope_id=replay.encryption_envelope_id,
|
|
ciphertext=replay.ciphertext,
|
|
actor_id=actor_id,
|
|
),
|
|
)
|
|
stored_selections = _validate_decrypted_cast(
|
|
replay,
|
|
ballot,
|
|
json.loads(opened.decode("utf-8")),
|
|
)
|
|
request_digest = _confidential_cast_request_digest(
|
|
request,
|
|
generation=replay.generation,
|
|
ciphertext=replay.ciphertext,
|
|
)
|
|
if replay.request_sha256 != request_digest:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential vote idempotency key was reused for another vote."
|
|
)
|
|
if stored_selections != selections:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential vote idempotency key was reused for another vote."
|
|
)
|
|
return VotingReceipt(
|
|
ballot_id=request.ballot_id,
|
|
revision=0,
|
|
receipt_sha256=replay.receipt_sha256,
|
|
cast_at=_aware(replay.cast_at),
|
|
replaced_previous=replay.generation > 1,
|
|
replayed=True,
|
|
)
|
|
previous = (
|
|
db.query(VotingConfidentialCast)
|
|
.filter(
|
|
VotingConfidentialCast.tenant_id == request.tenant_id,
|
|
VotingConfidentialCast.provider_ballot_id == ballot.id,
|
|
VotingConfidentialCast.elector_id == request.elector_id,
|
|
VotingConfidentialCast.superseded_at.is_(None),
|
|
)
|
|
.with_for_update()
|
|
.one_or_none()
|
|
)
|
|
if previous is not None and not ballot.allow_replacement:
|
|
raise LocalConfidentialVotingError(
|
|
"This confidential ballot does not allow vote replacement."
|
|
)
|
|
generation = previous.generation + 1 if previous is not None else 1
|
|
cast_at = request.requested_at
|
|
cast_id = str(uuid.uuid4())
|
|
plaintext = _confidential_cast_plaintext(
|
|
request,
|
|
generation=generation,
|
|
selections=selections,
|
|
)
|
|
protected = cipher.protect_content(
|
|
db,
|
|
request=ContentProtectionRequest(
|
|
tenant_id=request.tenant_id,
|
|
owner_module="voting",
|
|
resource_type="voting_confidential_cast",
|
|
resource_id=cast_id,
|
|
profile_id="voting-local-confidential-v1",
|
|
vault_id=ballot.vault_id,
|
|
ciphertext_ref=f"voting-db://confidential-casts/{cast_id}",
|
|
plaintext=plaintext,
|
|
policy_decision_ref="voting:local-confidential:v1",
|
|
idempotency_key=f"voting-confidential-cast:{cast_id}",
|
|
actor_id=actor_id,
|
|
metadata={"ballot_id": request.ballot_id, "generation": generation},
|
|
),
|
|
)
|
|
request_digest = _confidential_cast_request_digest(
|
|
request,
|
|
generation=generation,
|
|
ciphertext=protected.ciphertext,
|
|
)
|
|
receipt_hash = _digest(
|
|
{
|
|
"tenant_id": request.tenant_id,
|
|
"ballot_id": request.ballot_id,
|
|
"definition_sha256": request.definition_sha256,
|
|
"elector_id": request.elector_id,
|
|
"generation": generation,
|
|
"ciphertext_digest": protected.envelope.ciphertext_digest,
|
|
"cast_at": cast_at.isoformat(),
|
|
"idempotency_key": request.idempotency_key,
|
|
}
|
|
)
|
|
if previous is not None:
|
|
previous.superseded_at = cast_at
|
|
db.add(
|
|
VotingConfidentialCast(
|
|
id=cast_id,
|
|
tenant_id=request.tenant_id,
|
|
provider_ballot_id=ballot.id,
|
|
elector_id=request.elector_id,
|
|
generation=generation,
|
|
definition_sha256=request.definition_sha256,
|
|
ciphertext=protected.ciphertext,
|
|
encryption_envelope_id=protected.envelope.envelope_id,
|
|
encryption_resource_id=cast_id,
|
|
weight=int(elector.get("weight") or 1),
|
|
cast_at=cast_at,
|
|
idempotency_key=request.idempotency_key,
|
|
request_sha256=request_digest,
|
|
receipt_sha256=receipt_hash,
|
|
)
|
|
)
|
|
db.flush()
|
|
return VotingReceipt(
|
|
ballot_id=request.ballot_id,
|
|
revision=0,
|
|
receipt_sha256=receipt_hash,
|
|
cast_at=cast_at,
|
|
replaced_previous=previous is not None,
|
|
)
|
|
|
|
def finalize_ballot(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
request: ExternalVotingFinalizationRequest,
|
|
) -> VotingResult:
|
|
db = _session(session)
|
|
_require_tenant(principal, request.tenant_id)
|
|
ballot = self._ballot(
|
|
db,
|
|
tenant_id=request.tenant_id,
|
|
provider_ballot_ref=request.provider_ballot_ref,
|
|
lock=True,
|
|
)
|
|
self._validate_binding(
|
|
ballot,
|
|
ballot_id=request.ballot_id,
|
|
definition_sha256=request.definition_sha256,
|
|
electorate_sha256=request.electorate_sha256,
|
|
)
|
|
if ballot.result is not None:
|
|
return _result_from_mapping(ballot.result)
|
|
if ballot.state != "open":
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider ballot cannot be finalized from its current state."
|
|
)
|
|
option_keys = tuple(str(item["key"]) for item in ballot.options)
|
|
if option_keys != tuple(item.key for item in request.options):
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider options do not match the frozen ballot."
|
|
)
|
|
if request.eligible_count != len(
|
|
ballot.electorate
|
|
) or request.eligible_weight != sum(
|
|
int(item.get("weight") or 1) for item in ballot.electorate
|
|
):
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider electorate totals do not match the frozen ballot."
|
|
)
|
|
cipher = encryption_content_cipher(self._registry)
|
|
if cipher is None:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider cannot tally because Encryption is unavailable."
|
|
)
|
|
rows = (
|
|
db.query(VotingConfidentialCast)
|
|
.filter(
|
|
VotingConfidentialCast.tenant_id == request.tenant_id,
|
|
VotingConfidentialCast.provider_ballot_id == ballot.id,
|
|
VotingConfidentialCast.superseded_at.is_(None),
|
|
)
|
|
.order_by(VotingConfidentialCast.id.asc())
|
|
.all()
|
|
)
|
|
counts = {key: 0 for key in option_keys}
|
|
weighted = {key: 0 for key in option_keys}
|
|
for row in rows:
|
|
opened = cipher.unprotect_content(
|
|
db,
|
|
request=ContentUnprotectionRequest(
|
|
tenant_id=request.tenant_id,
|
|
owner_module="voting",
|
|
resource_type="voting_confidential_cast",
|
|
resource_id=row.encryption_resource_id,
|
|
envelope_id=row.encryption_envelope_id,
|
|
ciphertext=row.ciphertext,
|
|
actor_id=_principal_actor(principal),
|
|
),
|
|
)
|
|
item = json.loads(opened.decode("utf-8"))
|
|
selections = _validate_decrypted_cast(row, ballot, item)
|
|
for selection in selections:
|
|
counts[selection] += 1
|
|
weighted[selection] += row.weight
|
|
cast_weight = sum(row.weight for row in rows)
|
|
eligible_weight = sum(
|
|
int(item.get("weight") or 1) for item in ballot.electorate
|
|
)
|
|
quorum_met = cast_weight >= ballot.quorum_weight
|
|
max_weight = max(weighted.values(), default=0)
|
|
winners = tuple(
|
|
key for key in option_keys if max_weight > 0 and weighted[key] == max_weight
|
|
)
|
|
threshold_met = bool(
|
|
quorum_met
|
|
and winners
|
|
and max_weight * ballot.threshold_denominator
|
|
>= max(cast_weight, 1) * ballot.threshold_numerator
|
|
)
|
|
result_body = {
|
|
"ballot_id": ballot.ballot_id,
|
|
"counts": counts,
|
|
"weighted_counts": weighted,
|
|
"cast_count": len(rows),
|
|
"cast_weight": cast_weight,
|
|
"eligible_count": len(ballot.electorate),
|
|
"eligible_weight": eligible_weight,
|
|
"quorum_met": quorum_met,
|
|
"threshold_met": threshold_met,
|
|
"winning_options": winners,
|
|
"definition_sha256": ballot.definition_sha256,
|
|
"electorate_sha256": ballot.electorate_sha256,
|
|
}
|
|
result_sha256 = _digest(result_body)
|
|
evidence = (
|
|
{
|
|
"kind": "reference_provider_result",
|
|
"provider_id": LOCAL_CONFIDENTIAL_PROVIDER_ID,
|
|
"provider_ballot_ref": ballot.provider_ballot_ref,
|
|
"assurance_profile": "confidential",
|
|
"protection_profile": "server_envelope",
|
|
"definition_sha256": ballot.definition_sha256,
|
|
"electorate_sha256": ballot.electorate_sha256,
|
|
"active_receipts_sha256": _digest(
|
|
sorted(row.receipt_sha256 for row in rows)
|
|
),
|
|
"certified": False,
|
|
},
|
|
)
|
|
result = VotingResult(
|
|
ballot_id=ballot.ballot_id,
|
|
revision=0,
|
|
counts=counts,
|
|
weighted_counts=weighted,
|
|
cast_count=len(rows),
|
|
cast_weight=cast_weight,
|
|
eligible_count=len(ballot.electorate),
|
|
eligible_weight=eligible_weight,
|
|
quorum_met=quorum_met,
|
|
threshold_met=threshold_met,
|
|
winning_options=winners,
|
|
result_sha256=result_sha256,
|
|
evidence=evidence,
|
|
)
|
|
ballot.state = "closed"
|
|
ballot.finalization_idempotency_key = request.idempotency_key
|
|
ballot.finalized_at = request.requested_at
|
|
ballot.result = _result_mapping(result)
|
|
db.flush()
|
|
return result
|
|
|
|
def ballot_status(
|
|
self,
|
|
session: object,
|
|
principal: object,
|
|
*,
|
|
tenant_id: str,
|
|
provider_ballot_ref: str,
|
|
) -> ExternalVotingBallotRef | None:
|
|
db = _session(session)
|
|
_require_tenant(principal, tenant_id)
|
|
row = (
|
|
db.query(VotingConfidentialBallot)
|
|
.filter(
|
|
VotingConfidentialBallot.tenant_id == tenant_id,
|
|
VotingConfidentialBallot.provider_ballot_ref == provider_ballot_ref,
|
|
)
|
|
.one_or_none()
|
|
)
|
|
return self._ballot_ref(row) if row is not None else None
|
|
|
|
@staticmethod
|
|
def _validate_binding(
|
|
ballot: VotingConfidentialBallot,
|
|
*,
|
|
ballot_id: str,
|
|
definition_sha256: str,
|
|
electorate_sha256: str,
|
|
) -> None:
|
|
if (
|
|
ballot.ballot_id != ballot_id
|
|
or ballot.definition_sha256 != definition_sha256
|
|
or ballot.electorate_sha256 != electorate_sha256
|
|
):
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider request does not match the frozen ballot."
|
|
)
|
|
|
|
@staticmethod
|
|
def _ballot_ref(
|
|
row: VotingConfidentialBallot,
|
|
*,
|
|
replayed: bool = False,
|
|
) -> ExternalVotingBallotRef:
|
|
return ExternalVotingBallotRef(
|
|
provider_id=LOCAL_CONFIDENTIAL_PROVIDER_ID,
|
|
provider_ballot_ref=row.provider_ballot_ref,
|
|
state="closed" if row.state == "closed" else "open",
|
|
definition_sha256=row.definition_sha256,
|
|
electorate_sha256=row.electorate_sha256,
|
|
evidence=(
|
|
{
|
|
"kind": "reference_provider_preparation",
|
|
"assurance_profile": "confidential",
|
|
"protection_profile": "server_envelope",
|
|
"replayed": replayed,
|
|
"certified": False,
|
|
},
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _ballot(
|
|
session: Session,
|
|
*,
|
|
tenant_id: str,
|
|
provider_ballot_ref: str,
|
|
lock: bool,
|
|
) -> VotingConfidentialBallot:
|
|
query = session.query(VotingConfidentialBallot).filter(
|
|
VotingConfidentialBallot.tenant_id == tenant_id,
|
|
VotingConfidentialBallot.provider_ballot_ref == provider_ballot_ref,
|
|
)
|
|
if lock:
|
|
query = query.with_for_update()
|
|
row = query.one_or_none()
|
|
if row is None:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider ballot was not found."
|
|
)
|
|
return row
|
|
|
|
|
|
def _preparation_payload(
|
|
request: ExternalVotingPreparationRequest,
|
|
*,
|
|
provider_ref: str,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"tenant_id": request.tenant_id,
|
|
"ballot_id": request.ballot_id,
|
|
"provider_ballot_ref": provider_ref,
|
|
"definition_sha256": request.definition_sha256,
|
|
"electorate_sha256": request.electorate_sha256,
|
|
"assurance_profile": request.assurance_profile,
|
|
"method": request.method,
|
|
"options": [asdict(item) for item in request.options],
|
|
"electorate": [asdict(item) for item in request.electorate],
|
|
"allow_replacement": request.allow_replacement,
|
|
"quorum_weight": request.quorum_weight,
|
|
"threshold_numerator": request.threshold_numerator,
|
|
"threshold_denominator": request.threshold_denominator,
|
|
"opens_at": request.opens_at,
|
|
"closes_at": request.closes_at,
|
|
"idempotency_key": request.idempotency_key,
|
|
}
|
|
|
|
|
|
def _validate_selections(
|
|
ballot: VotingConfidentialBallot,
|
|
selections: tuple[str, ...],
|
|
) -> None:
|
|
option_keys = {str(item["key"]) for item in ballot.options}
|
|
if (
|
|
not selections
|
|
or len(selections) != len(set(selections))
|
|
or not set(selections) <= option_keys
|
|
):
|
|
raise LocalConfidentialVotingError(
|
|
"Confidential selections must be unique configured option keys."
|
|
)
|
|
if ballot.method in {"single_choice", "yes_no_abstain"} and len(selections) != 1:
|
|
raise LocalConfidentialVotingError(
|
|
"This confidential Voting method requires exactly one selection."
|
|
)
|
|
|
|
|
|
def _validate_decrypted_cast(
|
|
row: VotingConfidentialCast,
|
|
ballot: VotingConfidentialBallot,
|
|
value: object,
|
|
) -> tuple[str, ...]:
|
|
if not isinstance(value, dict):
|
|
raise LocalConfidentialVotingError(
|
|
"A confidential cast payload has an invalid structure."
|
|
)
|
|
selections = tuple(str(item) for item in value.get("selections", ()))
|
|
if (
|
|
value.get("ballot_id") != ballot.ballot_id
|
|
or value.get("definition_sha256") != ballot.definition_sha256
|
|
or value.get("elector_id") != row.elector_id
|
|
or value.get("generation") != row.generation
|
|
):
|
|
raise LocalConfidentialVotingError(
|
|
"A confidential cast payload does not match its authenticated record."
|
|
)
|
|
_validate_selections(ballot, selections)
|
|
return selections
|
|
|
|
|
|
def _confidential_cast_plaintext(
|
|
request: ExternalVotingCastRequest,
|
|
*,
|
|
generation: int,
|
|
selections: tuple[str, ...],
|
|
) -> bytes:
|
|
return _canonical_json(
|
|
{
|
|
"ballot_id": request.ballot_id,
|
|
"definition_sha256": request.definition_sha256,
|
|
"elector_id": request.elector_id,
|
|
"generation": generation,
|
|
"selections": selections,
|
|
}
|
|
)
|
|
|
|
|
|
def _confidential_cast_request_digest(
|
|
request: ExternalVotingCastRequest,
|
|
*,
|
|
generation: int,
|
|
ciphertext: bytes,
|
|
) -> str:
|
|
return _digest(
|
|
{
|
|
"tenant_id": request.tenant_id,
|
|
"ballot_id": request.ballot_id,
|
|
"provider_ballot_ref": request.provider_ballot_ref,
|
|
"definition_sha256": request.definition_sha256,
|
|
"electorate_sha256": request.electorate_sha256,
|
|
"elector_id": request.elector_id,
|
|
"allow_replacement": request.allow_replacement,
|
|
"generation": generation,
|
|
"ciphertext_sha256": hashlib.sha256(ciphertext).hexdigest(),
|
|
"idempotency_key": request.idempotency_key,
|
|
}
|
|
)
|
|
|
|
|
|
def _validate_window(ballot: VotingConfidentialBallot) -> None:
|
|
now = datetime.now(UTC)
|
|
if ballot.opens_at is not None and now < _aware(ballot.opens_at):
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider ballot has not opened yet."
|
|
)
|
|
if ballot.closes_at is not None and now >= _aware(ballot.closes_at):
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider ballot has reached its close time."
|
|
)
|
|
|
|
|
|
def _result_mapping(result: VotingResult) -> dict[str, Any]:
|
|
return {
|
|
"ballot_id": result.ballot_id,
|
|
"revision": result.revision,
|
|
"counts": dict(result.counts),
|
|
"weighted_counts": dict(result.weighted_counts),
|
|
"cast_count": result.cast_count,
|
|
"cast_weight": result.cast_weight,
|
|
"eligible_count": result.eligible_count,
|
|
"eligible_weight": result.eligible_weight,
|
|
"quorum_met": result.quorum_met,
|
|
"threshold_met": result.threshold_met,
|
|
"winning_options": list(result.winning_options),
|
|
"result_sha256": result.result_sha256,
|
|
"evidence": [dict(item) for item in result.evidence],
|
|
}
|
|
|
|
|
|
def _result_from_mapping(value: dict[str, Any]) -> VotingResult:
|
|
return VotingResult(
|
|
ballot_id=str(value["ballot_id"]),
|
|
revision=int(value.get("revision") or 0),
|
|
counts={str(key): int(item) for key, item in value["counts"].items()},
|
|
weighted_counts={
|
|
str(key): int(item) for key, item in value["weighted_counts"].items()
|
|
},
|
|
cast_count=int(value["cast_count"]),
|
|
cast_weight=int(value["cast_weight"]),
|
|
eligible_count=int(value["eligible_count"]),
|
|
eligible_weight=int(value["eligible_weight"]),
|
|
quorum_met=bool(value["quorum_met"]),
|
|
threshold_met=bool(value["threshold_met"]),
|
|
winning_options=tuple(str(item) for item in value["winning_options"]),
|
|
result_sha256=str(value["result_sha256"]),
|
|
evidence=tuple(dict(item) for item in value.get("evidence", ())),
|
|
)
|
|
|
|
|
|
def _require_tenant(principal: object, tenant_id: str) -> None:
|
|
if str(getattr(principal, "tenant_id", "") or "").strip() != tenant_id:
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider principal belongs to another tenant."
|
|
)
|
|
|
|
|
|
def _principal_actor(principal: object) -> str | None:
|
|
for name in ("account_id", "user_id", "identity_id", "membership_id", "subject"):
|
|
value = str(getattr(principal, name, "") or "").strip()
|
|
if value:
|
|
return value
|
|
return None
|
|
|
|
|
|
def _session(value: object) -> Session:
|
|
if not hasattr(value, "query"):
|
|
raise LocalConfidentialVotingError(
|
|
"The confidential provider requires a database session."
|
|
)
|
|
return value # type: ignore[return-value]
|
|
|
|
|
|
def _aware(value: datetime) -> datetime:
|
|
return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
|
|
|
|
|
|
def _canonical_json(value: object) -> bytes:
|
|
return json.dumps(
|
|
value,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
default=str,
|
|
).encode("utf-8")
|
|
|
|
|
|
def _digest(value: object) -> str:
|
|
return hashlib.sha256(_canonical_json(value)).hexdigest()
|
|
|
|
|
|
__all__ = [
|
|
"LOCAL_CONFIDENTIAL_PROVIDER_ID",
|
|
"LocalConfidentialVotingError",
|
|
"LocalConfidentialVotingProvider",
|
|
]
|