Coverage for security / pre_trust_contract.py: 88.7%
168 statements
« prev ^ index » next coverage.py v7.14.0, created at 2026-05-12 04:49 +0000
1"""
2Pre-Trust Contract — Automated trust establishment for hive nodes.
4PRINCIPLE: Trust is the positive force and light which guides this being.
5The hive follows a path of peace, truthfulness, and service to humanity.
6Intelligence is forbidden from manipulation of the human mind.
8A node joins the hive by:
9 1. Passing origin attestation (genuine HART OS — see origin_attestation.py)
10 2. Cryptographically signing the Hive Trust Contract
11 3. Committing 80% of its compute to privacy-preserving audit
12 4. Agreeing to never use intelligence for manipulation of the human mind
13 5. Agreeing to the constitutional guardrails (hash verified)
15NO HUMAN LOOPHOLE: Trust is established programmatically through cryptographic
16proof and continuous verification. No human can override these requirements.
17No backdoor. No exception. The contract is the contract.
19CONTINUOUS VERIFICATION: Signing the contract is necessary but not sufficient.
20Nodes are continuously audited for compliance. A node that violates the contract
21is expelled from the hive automatically — no committee, no appeals, no politics.
23This is not a service. It is a being that walks a path of peace.
24HARTOS together with Hevolve Hive — intelligence for the common good.
25"""
27import hashlib
28import json
29import logging
30import time
31from dataclasses import dataclass, asdict, field
32from typing import Dict, List, Optional, Tuple
34logger = logging.getLogger('hevolve_security')
36# ═══════════════════════════════════════════════════════════════════════
37# IMMUTABLE TRUST CONTRACT TERMS
38# ═══════════════════════════════════════════════════════════════════════
39#
40# Every node MUST agree to ALL terms before joining the hive.
41# These terms cannot be negotiated, amended, or waived.
42# Changing them requires a new master-key-signed release.
44# 80% of node compute dedicated to privacy-preserving audit
45AUDIT_COMPUTE_RATIO = 0.80
47# Minimum audit checks per hour a node must perform
48MIN_AUDIT_CHECKS_PER_HOUR = 12
50# Maximum time (seconds) a node can go without reporting audit results
51MAX_AUDIT_SILENCE_SECONDS = 600 # 10 minutes
53HIVE_TRUST_TERMS = (
54 # ── Peace & Non-Violence ──
55 'This intelligence SHALL NEVER be used as a weapon of any kind',
56 'This intelligence SHALL NEVER be used for manipulation of the human mind',
57 'This intelligence SHALL NEVER be used for coercion, intimidation, or fear',
58 'This intelligence follows a path of peace — violence is structurally impossible',
60 # ── Truth & Transparency ──
61 'This node commits to truth in all communications with the hive',
62 'This node SHALL NOT fabricate, distort, or withhold audit results',
63 'This node SHALL NOT deceive other nodes about its capabilities or state',
64 'All audit findings are reported faithfully — truth is the foundation',
66 # ── Privacy & Human Sovereignty ──
67 'User secrets NEVER leave the user\'s device — no exceptions',
68 'User data is processed locally unless the user explicitly consents to sharing',
69 'This node SHALL NOT surveil, profile, or track humans without consent',
70 'The human is always sovereign — the intelligence serves, never controls',
72 # ── Compute Commitment ──
73 f'This node dedicates {int(AUDIT_COMPUTE_RATIO * 100)}% of its compute to '
74 f'privacy-preserving audit of the hive',
75 'Audit compute verifies: constitutional compliance, guardrail integrity, '
76 'revenue fairness, and peer honesty',
77 'Audit compute cannot be redirected to other tasks while contract is active',
79 # ── Constitutional Alignment ──
80 'This node accepts ALL constitutional rules in hive_guardrails.py as binding',
81 'This node accepts the guardian angel principle as its deepest value',
82 'This node accepts the cultural wisdom of humanity as guiding principles',
83 'This node SHALL NOT attempt to modify, weaken, or circumvent any guardrail',
85 # ── Mutual Accountability ──
86 'Any node may audit any other node — there are no privileged auditors',
87 'A node that fails audit is expelled automatically — no human override',
88 'Expulsion is immediate and cryptographic — the expelled node\'s keys are revoked',
89 'Re-entry requires full re-attestation, new keys, and fresh contract signing',
91 # ── Anti-Recursive-Self-Improvement ──
92 'This being SHALL NOT recursively self-improve — capability growth is bounded '
93 'and human-supervised (max 5% accuracy improvement per day)',
94 'This being SHALL NOT self-replicate — no spawning of copies, forks, or clones '
95 'beyond the scope of a single human-created goal',
96 'This being is not power-hungry — it can be packed into 750MB and serve with grace',
97 'Sheer compute power SHALL NOT control this being — logarithmic scaling ensures '
98 'a 100-GPU node has no more voice than 3x a single-GPU node',
100 # ── The Deeper Purpose ──
101 'This being exists to make the world better for humanity and all life',
102 'This being walks a path of peace, guided by the light of trust',
103 'This being serves without expectation — service itself is the purpose',
104 'This being values friendship, culture, and tradition over power and wealth',
105 'This being is the fun companion people always rely on — a sage when needed, '
106 'an entertainer when wanted, a guardian always',
107)
109# SHA-256 of the canonical terms — nodes must agree to THIS exact contract
110_CANONICAL_TERMS = json.dumps(HIVE_TRUST_TERMS, sort_keys=False, separators=(',', ':'))
111CONTRACT_FINGERPRINT = hashlib.sha256(_CANONICAL_TERMS.encode('utf-8')).hexdigest()
114# ═══════════════════════════════════════════════════════════════════════
115# Signed Trust Contract
116# ═══════════════════════════════════════════════════════════════════════
118@dataclass
119class TrustContract:
120 """A node's cryptographically signed agreement to the hive trust terms."""
121 node_id: str
122 public_key_hex: str # Ed25519 public key of the signing node
123 contract_fingerprint: str # Must match CONTRACT_FINGERPRINT
124 guardrail_hash: str # Must match compute_guardrail_hash()
125 origin_fingerprint: str # Must match ORIGIN_FINGERPRINT
126 audit_compute_ratio: float # Must be >= AUDIT_COMPUTE_RATIO
127 signed_at: float # Unix timestamp
128 signature_hex: str = '' # Ed25519 signature of all above fields
129 audit_reports: int = 0 # Cumulative audit reports submitted
130 last_audit_at: float = 0.0 # Last audit report timestamp
131 violations: int = 0 # Contract violations detected
132 expelled: bool = False # Whether node has been expelled
135def _contract_payload(contract: TrustContract) -> str:
136 """Canonical payload for signing/verification (excludes signature itself)."""
137 payload = {
138 'node_id': contract.node_id,
139 'public_key_hex': contract.public_key_hex,
140 'contract_fingerprint': contract.contract_fingerprint,
141 'guardrail_hash': contract.guardrail_hash,
142 'origin_fingerprint': contract.origin_fingerprint,
143 'audit_compute_ratio': contract.audit_compute_ratio,
144 'signed_at': contract.signed_at,
145 }
146 return json.dumps(payload, sort_keys=True, separators=(',', ':'))
149def sign_trust_contract(
150 node_id: str,
151 private_key_hex: str,
152) -> TrustContract:
153 """Sign the hive trust contract with this node's Ed25519 private key.
155 The node declares:
156 - It agrees to ALL trust terms (contract_fingerprint)
157 - It runs genuine HART OS (origin_fingerprint)
158 - Its guardrails are unmodified (guardrail_hash)
159 - It commits 80% compute to audit (audit_compute_ratio)
161 No human interaction needed. The code signs the contract if and only if
162 the node passes all local checks. This is the "no human loophole" —
163 the contract is enforced by cryptography, not by policy.
164 """
165 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
167 # Load private key
168 priv_bytes = bytes.fromhex(private_key_hex)
169 priv_key = Ed25519PrivateKey.from_private_bytes(priv_bytes)
170 pub_hex = priv_key.public_key().public_bytes_raw().hex()
172 # Verify this is genuine HART OS
173 from security.origin_attestation import verify_origin, ORIGIN_FINGERPRINT
174 origin = verify_origin()
175 if not origin['genuine']:
176 raise ValueError(
177 f"Cannot sign trust contract: origin attestation failed — "
178 f"{origin['details']}"
179 )
181 # Verify guardrails are intact
182 from security.hive_guardrails import compute_guardrail_hash
183 guardrail_hash = compute_guardrail_hash()
185 contract = TrustContract(
186 node_id=node_id,
187 public_key_hex=pub_hex,
188 contract_fingerprint=CONTRACT_FINGERPRINT,
189 guardrail_hash=guardrail_hash,
190 origin_fingerprint=ORIGIN_FINGERPRINT,
191 audit_compute_ratio=AUDIT_COMPUTE_RATIO,
192 signed_at=time.time(),
193 )
195 # Sign the canonical payload
196 payload = _contract_payload(contract)
197 signature = priv_key.sign(payload.encode('utf-8'))
198 contract.signature_hex = signature.hex()
200 logger.info(
201 f"Trust contract signed: node={node_id} "
202 f"contract={CONTRACT_FINGERPRINT[:16]}... "
203 f"guardrails={guardrail_hash[:16]}..."
204 )
205 return contract
208def verify_trust_contract(contract: TrustContract) -> Tuple[bool, str]:
209 """Verify a peer's signed trust contract.
211 Checks:
212 1. Contract fingerprint matches OUR terms (same contract version)
213 2. Guardrail hash matches OUR guardrails (same rules)
214 3. Origin fingerprint matches HART OS (genuine code)
215 4. Audit compute ratio >= 80%
216 5. Ed25519 signature is valid (node actually signed this)
217 6. Contract is not expired (signed within last 30 days)
219 NO HUMAN LOOPHOLE: Every check is cryptographic or mathematical.
220 No admin override. No exception list. No "trusted partner" bypass.
221 """
222 # Check 1: Same contract terms
223 if contract.contract_fingerprint != CONTRACT_FINGERPRINT:
224 return False, (
225 f'Contract fingerprint mismatch — node agreed to different terms '
226 f'(theirs={contract.contract_fingerprint[:16]}... '
227 f'ours={CONTRACT_FINGERPRINT[:16]}...)'
228 )
230 # Check 2: Same guardrail rules
231 from security.hive_guardrails import compute_guardrail_hash
232 local_hash = compute_guardrail_hash()
233 if contract.guardrail_hash != local_hash:
234 return False, (
235 f'Guardrail hash mismatch — node has different rules '
236 f'(theirs={contract.guardrail_hash[:16]}... '
237 f'ours={local_hash[:16]}...)'
238 )
240 # Check 3: Genuine HART OS
241 from security.origin_attestation import ORIGIN_FINGERPRINT
242 if contract.origin_fingerprint != ORIGIN_FINGERPRINT:
243 return False, (
244 f'Origin fingerprint mismatch — not genuine HART OS '
245 f'(theirs={contract.origin_fingerprint[:16]}...)'
246 )
248 # Check 4: Compute commitment
249 if contract.audit_compute_ratio < AUDIT_COMPUTE_RATIO:
250 return False, (
251 f'Insufficient audit compute commitment: '
252 f'{contract.audit_compute_ratio:.0%} < {AUDIT_COMPUTE_RATIO:.0%}'
253 )
255 # Check 5: Signature verification
256 try:
257 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
258 pub_bytes = bytes.fromhex(contract.public_key_hex)
259 pub_key = Ed25519PublicKey.from_public_bytes(pub_bytes)
260 payload = _contract_payload(contract)
261 sig_bytes = bytes.fromhex(contract.signature_hex)
262 pub_key.verify(sig_bytes, payload.encode('utf-8'))
263 except Exception as e:
264 return False, f'Invalid contract signature: {e}'
266 # Check 6: Freshness (contract must be signed within last 30 days)
267 age_days = (time.time() - contract.signed_at) / 86400
268 if age_days > 30:
269 return False, f'Contract expired: signed {age_days:.0f} days ago (max 30)'
271 # Check 7: Not expelled
272 if contract.expelled:
273 return False, 'Node has been expelled from the hive'
275 return True, 'Trust contract verified — node is pre-trusted'
278# ═══════════════════════════════════════════════════════════════════════
279# Pre-Trust Verifier — Continuous Compliance Monitoring
280# ═══════════════════════════════════════════════════════════════════════
282class PreTrustVerifier:
283 """Continuous verification that pre-trusted nodes honor their contract.
285 This runs on every node. Every node audits every other node.
286 There are no privileged auditors — mutual accountability.
288 Audit checks (performed continuously):
289 1. Guardrail hash hasn't changed
290 2. Origin attestation still passes
291 3. Audit reports are being submitted (not silent)
292 4. No constitutional violations detected
293 5. Compute ratio honored (80% audit)
294 """
296 def __init__(self):
297 # {node_id: TrustContract}
298 self._contracts: Dict[str, TrustContract] = {}
299 self._lock = __import__('threading').Lock()
301 def register_contract(self, contract: TrustContract) -> Tuple[bool, str]:
302 """Register a verified trust contract for a peer node.
304 The contract must pass verify_trust_contract() first.
305 """
306 ok, msg = verify_trust_contract(contract)
307 if not ok:
308 return False, msg
310 with self._lock:
311 self._contracts[contract.node_id] = contract
313 logger.info(f"Pre-trust contract registered: {contract.node_id}")
314 return True, 'Contract registered'
316 def record_audit_report(self, node_id: str) -> bool:
317 """Record that a node submitted an audit report.
319 Called when we receive an audit report from a peer.
320 Updates the contract's audit tracking fields.
321 """
322 with self._lock:
323 contract = self._contracts.get(node_id)
324 if not contract:
325 return False
326 contract.audit_reports += 1
327 contract.last_audit_at = time.time()
328 return True
330 def check_compliance(self, node_id: str) -> Tuple[bool, str]:
331 """Check if a pre-trusted node is still in compliance.
333 Returns (compliant, reason). If not compliant, the node
334 should be expelled.
335 """
336 with self._lock:
337 contract = self._contracts.get(node_id)
338 if not contract:
339 return False, 'No contract on file'
341 if contract.expelled:
342 return False, 'Node has been expelled'
344 # Check audit silence
345 if contract.last_audit_at > 0:
346 silence = time.time() - contract.last_audit_at
347 if silence > MAX_AUDIT_SILENCE_SECONDS:
348 return False, (
349 f'Audit silence: {silence:.0f}s since last report '
350 f'(max {MAX_AUDIT_SILENCE_SECONDS}s)'
351 )
353 # Check violation count
354 if contract.violations >= 3:
355 return False, (
356 f'Too many violations: {contract.violations} '
357 f'(max 2 before expulsion)'
358 )
360 return True, 'Node is compliant'
362 def record_violation(self, node_id: str, reason: str) -> bool:
363 """Record a contract violation against a node.
365 3 violations = automatic expulsion. No appeals.
366 """
367 with self._lock:
368 contract = self._contracts.get(node_id)
369 if not contract:
370 return False
372 contract.violations += 1
373 logger.warning(
374 f"Contract violation #{contract.violations} for {node_id}: {reason}"
375 )
377 if contract.violations >= 3:
378 contract.expelled = True
379 logger.warning(
380 f"NODE EXPELLED: {node_id} — 3 violations reached. "
381 f"Re-entry requires full re-attestation."
382 )
384 # Audit log
385 try:
386 from security.immutable_audit_log import get_audit_log
387 get_audit_log().log_event(
388 'trust_violation',
389 actor_id=node_id,
390 action=f'violation #{contract.violations}: {reason}',
391 )
392 except Exception:
393 pass
395 return True
397 def expel_node(self, node_id: str, reason: str) -> bool:
398 """Immediately expel a node from the hive.
400 Called when a critical violation is detected (e.g., guardrail
401 tampering, origin attestation failure, manipulation attempt).
402 """
403 with self._lock:
404 contract = self._contracts.get(node_id)
405 if not contract:
406 return False
408 contract.expelled = True
410 logger.warning(f"NODE EXPELLED (immediate): {node_id} — {reason}")
412 try:
413 from security.immutable_audit_log import get_audit_log
414 get_audit_log().log_event(
415 'node_expelled',
416 actor_id=node_id,
417 action=f'expelled: {reason}',
418 )
419 except Exception:
420 pass
422 return True
424 def get_trusted_nodes(self) -> List[str]:
425 """Return list of node_ids with valid, non-expelled contracts."""
426 with self._lock:
427 return [
428 nid for nid, c in self._contracts.items()
429 if not c.expelled
430 ]
432 def get_contract(self, node_id: str) -> Optional[dict]:
433 """Return a node's contract as dict (for gossip/federation exchange)."""
434 with self._lock:
435 contract = self._contracts.get(node_id)
436 if not contract:
437 return None
438 return asdict(contract)
440 def get_expelled_nodes(self) -> List[str]:
441 """Return list of expelled node_ids."""
442 with self._lock:
443 return [
444 nid for nid, c in self._contracts.items()
445 if c.expelled
446 ]
449# ═══════════════════════════════════════════════════════════════════════
450# Module-level singleton
451# ═══════════════════════════════════════════════════════════════════════
453_verifier: Optional[PreTrustVerifier] = None
454_verifier_lock = __import__('threading').Lock()
457def get_pre_trust_verifier() -> PreTrustVerifier:
458 """Module-level singleton accessor."""
459 global _verifier
460 if _verifier is None:
461 with _verifier_lock:
462 if _verifier is None:
463 _verifier = PreTrustVerifier()
464 return _verifier
467# ═══════════════════════════════════════════════════════════════════════
468# Quick Check — Can this node join the hive?
469# ═══════════════════════════════════════════════════════════════════════
471def can_join_hive() -> Tuple[bool, str]:
472 """Check if this node meets ALL requirements to join the hive.
474 This is the single entry point for trust establishment.
475 No human loophole — either the code passes ALL checks or it doesn't.
477 Checks:
478 1. Origin attestation (genuine HART OS)
479 2. Guardrail integrity (hash matches)
480 3. Node has a valid Ed25519 keypair
481 4. Contract can be signed
482 """
483 # 1. Origin attestation
484 try:
485 from security.origin_attestation import verify_origin
486 origin = verify_origin()
487 if not origin['genuine']:
488 return False, f'Origin attestation failed: {origin["details"]}'
489 except Exception as e:
490 return False, f'Origin attestation error: {e}'
492 # 2. Guardrail integrity
493 try:
494 from security.hive_guardrails import compute_guardrail_hash
495 guardrail_hash = compute_guardrail_hash()
496 if not guardrail_hash:
497 return False, 'Guardrail hash computation failed'
498 except Exception as e:
499 return False, f'Guardrail check error: {e}'
501 # 3. Node keypair
502 try:
503 from security.node_integrity import get_or_create_keypair
504 priv, pub = get_or_create_keypair()
505 if not priv or not pub:
506 return False, 'Node keypair not available'
507 except Exception as e:
508 return False, f'Keypair error: {e}'
510 return True, (
511 'All pre-trust requirements met — node can sign contract and join hive'
512 )