examples/byzantine_detection.py
| 1 | """Example: Byzantine double-voting is detected at tally time. |
| 2 | |
| 3 | One node (Eve) attempts to double-vote with *conflicting* decisions on the |
| 4 | same proposal. The ``VoteTally`` raises ``ByzantineDetectedError`` the moment |
| 5 | the second conflicting vote is submitted, and the audit log captures the |
| 6 | violation. |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | from quantumshield.identity.agent import AgentIdentity |
| 12 | |
| 13 | from pqc_ai_governance import ( |
| 14 | ByzantineDetectedError, |
| 15 | GovernanceAuditLog, |
| 16 | GovernanceNode, |
| 17 | GovernanceProposal, |
| 18 | NodeRegistry, |
| 19 | ProposalKind, |
| 20 | VoteDecision, |
| 21 | VoteTally, |
| 22 | ) |
| 23 | |
| 24 | |
| 25 | def main() -> None: |
| 26 | print("== PQC AI Governance: Byzantine Double-Vote Detection ==\n") |
| 27 | |
| 28 | alice = GovernanceNode(identity=AgentIdentity.create("alice"), name="alice") |
| 29 | bob = GovernanceNode(identity=AgentIdentity.create("bob"), name="bob") |
| 30 | eve = GovernanceNode(identity=AgentIdentity.create("eve"), name="eve") |
| 31 | |
| 32 | registry = NodeRegistry() |
| 33 | for n in (alice, bob, eve): |
| 34 | registry.register(n) |
| 35 | |
| 36 | proposal = GovernanceProposal.create( |
| 37 | kind=ProposalKind.UPDATE_POLICY, |
| 38 | subject_id="policy:rate-limit", |
| 39 | title="Raise rate limit to 100 QPS", |
| 40 | proposer_did=alice.did, |
| 41 | ) |
| 42 | alice.sign_proposal(proposal) |
| 43 | |
| 44 | audit = GovernanceAuditLog() |
| 45 | audit.log_proposal_created(proposal) |
| 46 | |
| 47 | tally = VoteTally(proposal=proposal, registry=registry) |
| 48 | |
| 49 | # Honest votes first. |
| 50 | alice_vote = alice.cast_vote(proposal, VoteDecision.APPROVE) |
| 51 | bob_vote = bob.cast_vote(proposal, VoteDecision.APPROVE) |
| 52 | tally.add(alice_vote) |
| 53 | audit.log_vote_cast(alice_vote) |
| 54 | tally.add(bob_vote) |
| 55 | audit.log_vote_cast(bob_vote) |
| 56 | print(f"alice -> {alice_vote.vote.decision.value}") |
| 57 | print(f"bob -> {bob_vote.vote.decision.value}") |
| 58 | |
| 59 | # Eve votes once. |
| 60 | eve_first = eve.cast_vote(proposal, VoteDecision.APPROVE) |
| 61 | tally.add(eve_first) |
| 62 | audit.log_vote_cast(eve_first) |
| 63 | print(f"eve -> {eve_first.vote.decision.value} (first vote)") |
| 64 | |
| 65 | # Eve now tries to flip her vote with a brand-new signed ballot. |
| 66 | eve_conflict = eve.cast_vote(proposal, VoteDecision.REJECT) |
| 67 | print(f"eve -> {eve_conflict.vote.decision.value} (conflicting second vote)") |
| 68 | |
| 69 | try: |
| 70 | tally.add(eve_conflict) |
| 71 | except ByzantineDetectedError as exc: |
| 72 | print(f"\n[!] ByzantineDetectedError raised: {exc}") |
| 73 | audit.log_byzantine_detected( |
| 74 | voter_did=eve.did, |
| 75 | proposal_id=proposal.proposal_id, |
| 76 | prior=VoteDecision.APPROVE.value, |
| 77 | now=VoteDecision.REJECT.value, |
| 78 | ) |
| 79 | else: # pragma: no cover - should not happen |
| 80 | print("unexpected: no exception raised") |
| 81 | return |
| 82 | |
| 83 | print("\n-- Audit trail --") |
| 84 | for entry in audit.entries(): |
| 85 | extra = f" decision={entry.decision}" if entry.decision else "" |
| 86 | print(f" {entry.operation:<22s} actor={entry.actor_did[:22]}...{extra}") |
| 87 | |
| 88 | |
| 89 | if __name__ == "__main__": |
| 90 | main() |
| 91 | |