src/pqc_ai_governance/proposal.py
4.5 KB · 136 lines · python Raw
1 """Governance proposals - the thing nodes vote on."""
2
3 from __future__ import annotations
4
5 import hashlib
6 import json
7 import uuid
8 from dataclasses import asdict, dataclass, field
9 from datetime import datetime, timedelta, timezone
10 from enum import Enum
11 from typing import Any
12
13
14 class ProposalKind(str, Enum):
15 """Types of things enterprise AI governance votes on."""
16
17 AUTHORIZE_MODEL = "authorize-model" # grant model permission to run
18 REVOKE_MODEL = "revoke-model"
19 AUTHORIZE_AGENT = "authorize-agent" # grant agent permission to act
20 REVOKE_AGENT = "revoke-agent"
21 UPDATE_POLICY = "update-policy" # change a runtime policy
22 ADD_NODE = "add-node" # admit a new governance node
23 REMOVE_NODE = "remove-node"
24 EMERGENCY_FREEZE = "emergency-freeze" # halt all agent action
25 DELEGATION = "delegation" # allow agent X to delegate to Y
26
27
28 class ProposalStatus(str, Enum):
29 OPEN = "open"
30 PASSED = "passed"
31 REJECTED = "rejected"
32 EXPIRED = "expired"
33
34
35 @dataclass
36 class GovernanceProposal:
37 """A proposal nodes vote on.
38
39 The ``subject_id`` identifies what the proposal is about (model DID, agent DID,
40 policy id, etc.). ``parameters`` is an arbitrary dict of rule-specific fields.
41 """
42
43 proposal_id: str
44 kind: ProposalKind
45 subject_id: str # e.g. "did:pqaid:..." for model/agent
46 title: str
47 description: str
48 proposer_did: str
49 parameters: dict[str, Any] = field(default_factory=dict)
50 created_at: str = ""
51 expires_at: str = ""
52 status: ProposalStatus = ProposalStatus.OPEN
53
54 # Populated by proposer signature
55 signer_did: str = ""
56 algorithm: str = ""
57 signature: str = "" # hex
58 public_key: str = "" # hex
59
60 @classmethod
61 def create(
62 cls,
63 kind: ProposalKind,
64 subject_id: str,
65 title: str,
66 proposer_did: str,
67 description: str = "",
68 parameters: dict[str, Any] | None = None,
69 ttl_seconds: int = 86400,
70 ) -> GovernanceProposal:
71 now = datetime.now(timezone.utc)
72 return cls(
73 proposal_id=f"urn:pqc-gov-prop:{uuid.uuid4().hex}",
74 kind=kind,
75 subject_id=subject_id,
76 title=title,
77 description=description,
78 proposer_did=proposer_did,
79 parameters=dict(parameters or {}),
80 created_at=now.isoformat(),
81 expires_at=(now + timedelta(seconds=ttl_seconds)).isoformat(),
82 status=ProposalStatus.OPEN,
83 )
84
85 def canonical_bytes(self) -> bytes:
86 """Bytes covered by the proposer's signature."""
87 payload = {
88 "proposal_id": self.proposal_id,
89 "kind": self.kind.value,
90 "subject_id": self.subject_id,
91 "title": self.title,
92 "description": self.description,
93 "proposer_did": self.proposer_did,
94 "parameters": self.parameters,
95 "created_at": self.created_at,
96 "expires_at": self.expires_at,
97 }
98 return json.dumps(
99 payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False
100 ).encode("utf-8")
101
102 def proposal_hash(self) -> str:
103 return hashlib.sha3_256(self.canonical_bytes()).hexdigest()
104
105 def is_expired(self) -> bool:
106 try:
107 exp = datetime.fromisoformat(self.expires_at)
108 return datetime.now(timezone.utc) > exp
109 except ValueError:
110 return False
111
112 def to_dict(self) -> dict[str, Any]:
113 d = asdict(self)
114 d["kind"] = self.kind.value
115 d["status"] = self.status.value
116 return d
117
118 @classmethod
119 def from_dict(cls, data: dict[str, Any]) -> GovernanceProposal:
120 return cls(
121 proposal_id=data["proposal_id"],
122 kind=ProposalKind(data["kind"]),
123 subject_id=data["subject_id"],
124 title=data["title"],
125 description=data.get("description", ""),
126 proposer_did=data["proposer_did"],
127 parameters=dict(data.get("parameters", {})),
128 created_at=data.get("created_at", ""),
129 expires_at=data.get("expires_at", ""),
130 status=ProposalStatus(data.get("status", "open")),
131 signer_did=data.get("signer_did", ""),
132 algorithm=data.get("algorithm", ""),
133 signature=data.get("signature", ""),
134 public_key=data.get("public_key", ""),
135 )
136