Farnsworth

""" FARNS Temporal Hash Mesh + Quorum Verification ================================================ Layer 3: Temporal Hash Mesh - Nodes interleave their hash chains - Each heartbeat includes peer hashes - Mesh root = BLAKE3(sorted(all_node_hashes)) - If one node is compromised, mesh diverges → instant detection Layer 4: Swarm Quorum Verification - New connections require 2+ existing members to verify - Byzantine fault tolerant """ import time import json import asyncio from typing import Dict, List, Optional, Set, Tuple from dataclasses import dataclass, field from loguru import logger from .farns_auth import blake3 from .farns_config import FARNS_MESH_FILE, ensure_dirs @dataclass class MeshEntry: """A single node's entry in the hash mesh.""" node_name: str current_hash: bytes sequence: int last_seen: float gpu_fingerprint: bytes = b"" class TemporalHashMesh: """ Cryptographic mesh of interleaved hash chains. Each node maintains its own chain, but heartbeats include peer hashes, creating an interdependent mesh. If any node is compromised or impersonated, the mesh root diverges and all nodes detect it. """ def __init__(self, my_name: str, my_seed_hash: bytes): self.my_name = my_name self._my_hash = my_seed_hash # Initial hash from identity self._my_seq = 0 self._peers: Dict[str, MeshEntry] = {} self._mesh_root = b"" self._load_state() def _load_state(self): """Load persisted mesh state.""" try: if FARNS_MESH_FILE.exists(): data = json.loads(FARNS_MESH_FILE.read_text()) self._my_seq = data.get("my_seq", 0) # Hashes are stored as hex stored_hash = data.get("my_hash") if stored_hash: self._my_hash = bytes.fromhex(stored_hash) except Exception as e: logger.debug(f"Could not load mesh state: {e}") def _save_state(self): """Persist mesh state.""" try: ensure_dirs() data = { "my_name": self.my_name, "my_hash": self._my_hash.hex(), "my_seq": self._my_seq, "peers": { name: { "hash": entry.current_hash.hex(), "seq": entry.sequence, "last_seen": entry.last_seen, } for name, entry in self._peers.items() }, "mesh_root": self._mesh_root.hex(), } FARNS_MESH_FILE.write_text(json.dumps(data, indent=2)) except Exception as e: logger.debug(f"Could not save mesh state: {e}") def advance(self) -> bytes: """ Advance this node's hash chain by one step. new_hash = BLAKE3(previous_hash || peer_hashes || sequence) The peer hashes are interleaved, making this chain dependent on the state of the entire mesh. """ # Collect peer hashes sorted by name (deterministic) peer_data = b"" for name in sorted(self._peers.keys()): peer_data += self._peers[name].current_hash # Advance chain seq_bytes = self._my_seq.to_bytes(8, "big") self._my_hash = blake3( self._my_hash + peer_data + seq_bytes ).digest() self._my_seq += 1 # Recompute mesh root self._recompute_root() self._save_state() return self._my_hash def update_peer(self, peer_name: str, peer_hash: bytes, peer_seq: int, gpu_fp: bytes = b""): """Update a peer's hash in the mesh.""" if peer_name == self.my_name: return self._peers[peer_name] = MeshEntry( node_name=peer_name, current_hash=peer_hash, sequence=peer_seq, last_seen=time.time(), gpu_fingerprint=gpu_fp, ) self._recompute_root() def remove_peer(self, peer_name: str): """Remove a peer from the mesh (disconnected/kicked).""" self._peers.pop(peer_name, None) self._recompute_root() def _recompute_root(self): """Compute mesh root = BLAKE3(sorted(all_hashes_by_name)). All nodes (including self) are sorted by name so that every node in the mesh computes the identical root hash. """ entries = [(self.my_name, self._my_hash)] for name, entry in self._peers.items(): entries.append((name, entry.current_hash)) entries.sort(key=lambda e: e[0]) # deterministic order by name combined = b"".join(h for _, h in entries) self._mesh_root = blake3(combined).digest() @property def mesh_root(self) -> bytes: return self._mesh_root @property def my_hash(self) -> bytes: return self._my_hash @property def my_sequence(self) -> int: return self._my_seq @property def peer_count(self) -> int: return len(self._peers) def get_heartbeat_data(self) -> Dict: """Get data to include in heartbeat packets.""" return { "name": self.my_name, "hash": self._my_hash, "seq": self._my_seq, "mesh_root": self._mesh_root, "peer_count": len(self._peers), } def verify_peer_mesh(self, peer_name: str, claimed_root: bytes) -> bool: """ Verify peer mesh health. In a 2-node mesh, roots will always differ because each node's temporal hash chain advances independently — this is expected and NOT a security concern. Meaningful root verification requires 3+ nodes where a compromised minority can be detected. For 2-node meshes we instead verify liveness: the peer's sequence must be advancing and recently seen. """ if not self._mesh_root or not claimed_root: return True # Can't verify yet (mesh still forming) if self._mesh_root == claimed_root: return True # With only 2 nodes, root mismatch is expected (independent chains). # Skip noisy logging — just verify peer liveness instead. total_nodes = 1 + len(self._peers) if total_nodes <= 2: # Check peer is alive (sequence advancing, recently seen) peer = self._peers.get(peer_name) if peer: age = time.time() - peer.last_seen if age > 60: logger.warning( f"Peer {peer_name} stale: last seen {age:.0f}s ago" ) return False return True # Liveness OK, mismatch is expected for 2-node mesh # 3+ nodes: root mismatch may indicate compromise logger.warning( f"Mesh root divergence from {peer_name} in {total_nodes}-node mesh: " f"ours={self._mesh_root[:8].hex()} theirs={claimed_root[:8].hex()}" ) return False def prune_stale_peers(self, max_age_seconds: float = 60.0): """Remove peers that haven't been seen recently.""" cutoff = time.time() - max_age_seconds stale = [name for name, entry in self._peers.items() if entry.last_seen < cutoff] for name in stale: logger.info(f"Pruning stale mesh peer: {name}") del self._peers[name] if stale: self._recompute_root() class SwarmQuorum: """ Quorum-based verification for new node connections. A new connection is only accepted if 2+ existing swarm members independently verify the connecting node. Byzantine fault tolerant. """ def __init__(self, min_votes: int = 2): self.min_votes = min_votes self._pending_votes: Dict[str, Dict] = {} # request_id → vote state def start_vote(self, request_id: str, candidate_name: str, candidate_identity: bytes, gpu_fp: bytes) -> str: """Start a quorum vote for a new connection.""" self._pending_votes[request_id] = { "candidate": candidate_name, "identity": candidate_identity, "gpu_fp": gpu_fp, "votes_for": set(), "votes_against": set(), "started": time.time(), } logger.info(f"Quorum vote started for {candidate_name} (id={request_id})") return request_id def cast_vote(self, request_id: str, voter: str, approve: bool, reason: str = "") -> Optional[str]: """ Cast a vote. Returns the decision if quorum reached: "accepted", "rejected", or None (still voting). """ vote = self._pending_votes.get(request_id) if not vote: return None if approve: vote["votes_for"].add(voter) else: vote["votes_against"].add(voter) total = len(vote["votes_for"]) + len(vote["votes_against"]) # Check if quorum reached if len(vote["votes_for"]) >= self.min_votes: logger.info( f"Quorum ACCEPTED {vote['candidate']}: " f"{len(vote['votes_for'])} for, {len(vote['votes_against'])} against" ) del self._pending_votes[request_id] return "accepted" # Check if impossible to reach quorum (too many rejections) # This is a simple check; in practice you'd know the total voter count if len(vote["votes_against"]) >= self.min_votes: logger.info(f"Quorum REJECTED {vote['candidate']}") del self._pending_votes[request_id] return "rejected" return None # Still voting def get_pending(self) -> List[Dict]: """Get all pending votes.""" return [ { "request_id": rid, "candidate": v["candidate"], "votes_for": len(v["votes_for"]), "votes_against": len(v["votes_against"]), "age_seconds": time.time() - v["started"], } for rid, v in self._pending_votes.items() ] def cleanup_stale(self, max_age: float = 300.0): """Remove votes older than max_age seconds.""" cutoff = time.time() - max_age stale = [rid for rid, v in self._pending_votes.items() if v["started"] < cutoff] for rid in stale: del self._pending_votes[rid]