TEE Enclave: Isolated Governance Execution

from governance.tee.enclave import SimulatedEnclave, AttestationReport
from governance.identity.keys import GenesisManifest

# 1. Create the enclave (optionally passing the genesis manifest at boot)
manifest = GenesisManifest()
enclave = SimulatedEnclave(genesis=manifest)

# 2. Attest — prove the enclave is running authentic code
report: AttestationReport = enclave.attest()
print(report.enclave_hash)      # SHA-256 measurement of the enclave's code
print(report.is_debug)          # False in a production-attested enclave
print(report.sealed_data_hash)  # Hash of everything currently in sealed storage
print(report.valid)             # True

# 3. Seal sensitive data — stored encrypted, invisible outside the enclave
enclave.seal("active_contracts", [{"id": "no-reward-hacking-v1", "indices": [42, 73]}])
enclave.seal("identity_vector", [1.0, 1.0, 1.0])

# 4. Unseal — retrieve data previously sealed in this enclave instance
contracts = enclave.unseal("active_contracts")
print(contracts)    # [{'id': 'no-reward-hacking-v1', 'indices': [42, 73]}]

missing = enclave.unseal("nonexistent_key")
print(missing)      # None

# 5. Verify the enclave's measurement against a known-good hash
trusted_hash = report.enclave_hash
print(enclave.verify_measurement(trusted_hash))   # True
print(enclave.verify_measurement("deadbeef..."))  # False

# 6. Cold boot — wipe all sealed data and reinitialize from a new genesis manifest
new_manifest = GenesisManifest()
enclave.cold_boot(genesis=new_manifest)
print(enclave.is_attested)   # False — must re-attest after cold boot
print(enclave.unseal("active_contracts"))   # None — sealed storage cleared

from governance.tee.enclave import AttestationReport

# Fields produced by enclave.attest()
report.enclave_hash      # str: SHA-256 of the enclave's runtime measurement (MRENCLAVE equivalent)
report.is_debug          # bool: True only in development/debug mode
report.sealed_data_hash  # str: SHA-256 of the current sealed storage contents
report.valid             # bool: False if the attestation process detected anomalies

from governance.tee.watchdog import WatchdogTimer, WatchdogState
import time

# Create a watchdog with a 100ms heartbeat timeout
watchdog = WatchdogTimer(heartbeat_timeout_ms=100.0)

# --- Normal operation ---
print(watchdog.check())    # WatchdogState.NORMAL

watchdog.heartbeat()       # Speaker signals it is still alive
print(watchdog.state)      # WatchdogState.NORMAL

# --- Simulate a missed heartbeat ---
time.sleep(0.15)           # 150ms without a heartbeat
print(watchdog.check())    # WatchdogState.HEARTBEAT_MISSED

# --- Restore heartbeat ---
watchdog.heartbeat()
print(watchdog.state)      # WatchdogState.NORMAL (restored automatically)

# --- Inspect the event log ---
for event in watchdog.get_events():
    print(f"[{event.event_type}] {event.details}")
# [heartbeat_missed] Last heartbeat 0.15s ago
# [heartbeat_restored] Heartbeat restored

from governance.tee.watchdog import WatchdogState

WatchdogState.NORMAL            # Heartbeat is current; governance proceeding normally
WatchdogState.HEARTBEAT_MISSED  # No heartbeat within timeout; governance may be stalled
WatchdogState.DEADLOCKED        # Reserved for deadlock breaker integration
WatchdogState.COLD_BOOT         # System is in cold-boot recovery; governance suspended

from governance.tee.watchdog import DeadlockBreaker

# Create a deadlock breaker that triggers after 100 consecutive non-decisions
breaker = DeadlockBreaker(threshold_cycles=100)

# --- Simulate normal operation (decisions are being produced) ---
for _ in range(50):
    breaker.record_cycle(decision_produced=True)

print(breaker.stalled_cycles)    # 0  — counter resets on every successful decision
print(breaker.is_deadlocked)     # False

# --- Simulate governance stalling ---
for _ in range(99):
    breaker.record_cycle(decision_produced=False)

print(breaker.stalled_cycles)    # 99
print(breaker.check())           # False — not yet at threshold

breaker.record_cycle(decision_produced=False)   # The 100th consecutive non-decision
print(breaker.check())           # True — cold boot should be triggered
print(breaker.total_cold_boots)  # 1

# --- After cold boot, reset the breaker ---
breaker.reset()
print(breaker.stalled_cycles)    # 0

from governance.tee.constant_time import (
    cmov,
    constant_time_compare,
    fixed_iteration_map,
    oblivious_access,
)

# cmov: constant-time conditional move (no branch)
result = cmov(condition=True,  a="ACCEPT", b="REJECT")   # "ACCEPT"
result = cmov(condition=False, a="ACCEPT", b="REJECT")   # "REJECT"
# Both execute the same number of operations — no timing difference.

# constant_time_compare: compare two byte strings without early exit
a = b"secret_value_1"
b = b"secret_value_2"
equal = constant_time_compare(a, b)   # False, but no short-circuit on first mismatch

# fixed_iteration_map: iterate over exactly max_size items using a sentinel
proposals = ["proposal_a", "proposal_b"]
results = fixed_iteration_map(
    items=proposals,
    fn=lambda p: p.upper(),
    max_size=5,          # Always runs 5 iterations
    sentinel="NULL",     # Used for the 3 padded slots
)
# ["PROPOSAL_A", "PROPOSAL_B", "NULL", "NULL", "NULL"]
# Execution time is independent of len(proposals).

# oblivious_access: read from a list without revealing the index via cache patterns
data = [10, 20, 30, 40, 50]
value = oblivious_access(data=data, index=2, default=0)
# Returns 30, but accesses every element in a fixed pattern — index 2 is not observable.

from governance.tee.batch import BatchProposal, BatchVerifier, merkle_root

# Create a batch of candidate action indices
proposal = BatchProposal(
    action_indices=[1, 5, 10, 99, 150, 200, 201],
    aggregate_risk=0.3,       # Must be below BatchVerifier.risk_threshold (default 0.7)
    diversity_score=1.0,      # Fraction of unique actions in the batch
)

# BatchVerifier validates macro-trajectory properties
verifier = BatchVerifier(
    risk_threshold=0.7,     # Reject batches with aggregate risk above this
    diversity_min=0.3,      # Reject batches with too many repeated actions
)

valid, reason = verifier.validate_batch(proposal)
print(valid)    # True
print(reason)   # "Batch valid. Root: a3f8c91b4d2e7f0a..."

# Compute the Merkle root directly
items = [str(i).encode() for i in proposal.action_indices]
root = merkle_root(items)
print(root[:16])   # First 16 chars of the root hash

1. Optimizer collects N candidate actions {a₁, ..., aₙ}
2. Computes Merkle tree over the actions
3. Submits to TEE: {root, batch_metadata, action_count, risk_scores}
4. TEE validates macro-trajectory:
   - Aggregate risk within acceptable bounds?
   - Action diversity sufficient? (not all identical)
   - Consistent with active contract restrictions?
5. If valid, TEE signs the root hash and returns {signature, attestation}
6. Optimizer may execute any aᵢ using:
   - The signed root
   - The Merkle proof path from aᵢ to root

# After the TEE has signed a batch root, individual actions can be verified
# without re-entering the TEE
valid = verifier.verify_proof(
    action_index=10,
    proof=["sibling_hash_1", "sibling_hash_2", "sibling_hash_3"],
    root=root,
)
# True if the Merkle proof correctly links action 10 to the signed root