La LA LA
#!/usr/bin/env python3
"""Thalora Routing Experiment v0.2 — identity/location separation.
An offline, deterministic topology simulation. No sockets, telemetry, or external
connections. Requires: pip install cryptography
Demonstrates signed endpoint location announcements, relay-derived shortest paths,
route epochs, movement, hop limits, and expiry. NOT a secure routing protocol:
relay topology/cost information is trusted in this demo; signed endpoint claims
alone cannot prevent malicious relays, Sybil attacks, or route hijacks.
"""
import hashlib
import heapq
import json
import time
from dataclasses import dataclass
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey, Ed25519PublicKey
from cryptography.hazmat.primitives import serialization
def canonical(obj):
return json.dumps(obj, sort_keys=True, separators=(',', ':')).encode()
@dataclass
class Identity:
name: str
private: Ed25519PrivateKey
@classmethod
def fresh(cls, name):
return cls(name, Ed25519PrivateKey.generate())
@property
def public(self):
return self.private.public_key().public_bytes(
serialization.Encoding.Raw, serialization.PublicFormat.Raw)
@property
def node_id(self):
return hashlib.sha256(self.public).hexdigest()
def announce(self, locator, epoch, expires):
body = {"version": 2, "kind": "location", "identity": self.node_id,
"public_key": self.public.hex(), "locator": locator,
"epoch": epoch, "expires": expires}
return {"body": body, "signature": self.private.sign(canonical(body)).hex()}
def verify_announcement(packet, now):
b = packet["body"]
if (b["version"] != 2 or b["kind"] != "location" or
not isinstance(b["epoch"], int) or b["epoch"] < 0 or
not isinstance(b["locator"], str) or len(b["locator"]) > 64 or
not isinstance(b["expires"], int) or b["expires"] <= now or
b["expires"] > now + 3600):
raise ValueError("invalid announcement fields or expiry")
public = bytes.fromhex(b["public_key"])
if hashlib.sha256(public).hexdigest() != b["identity"]:
raise ValueError("identity/public key mismatch")
Ed25519PublicKey.from_public_bytes(public).verify(
bytes.fromhex(packet["signature"]), canonical(b))
return b
class Mesh:
def __init__(self):
self.links = {} # trusted local topology for this simulation
self.announcements = {} # identity -> verified latest body
def link(self, a, b, cost=1):
if cost <= 0:
raise ValueError("cost must be positive")
self.links.setdefault(a, {})[b] = cost
self.links.setdefault(b, {})[a] = cost
def learn(self, signed, now):
body = verify_announcement(signed, now)
previous = self.announcements.get(body["identity"])
if previous and body["epoch"] <= previous["epoch"]:
return False # no stale replay or conflicting same-epoch location
self.announcements[body["identity"]] = body
return True
def route(self, origin, identity, now):
body = self.announcements.get(identity)
if not body or body["expires"] <= now:
return None
destination = body["locator"]
queue = [(0, origin, (origin,))]
best = {}
while queue:
cost, current, path = heapq.heappop(queue)
if current in best:
continue
best[current] = cost
if current == destination:
return {"identity": identity, "epoch": body["epoch"],
"cost": cost, "path": list(path)}
for neighbor, weight in self.links.get(current, {}).items():
if neighbor not in best:
heapq.heappush(queue, (cost + weight, neighbor, path + (neighbor,)))
return None
Cont in next message
