Upload phantom_shard/spiking/anomaly.py

phantom_shard/spiking/anomaly.py

@@ -0,0 +1,329 @@
+"""
+Phantom Shard Protocol — Neuromorphic Spiking Layer
+====================================================
+Hunt for Zero-Day vulnerabilities by sensing 'logic-flow anomalies' rather
+than scanning for code patterns. Implements Leaky Integrate-and-Fire (LIF)
+neurons with Spike-Timing-Dependent Plasticity (STDP).
+
+Architecture:
+  - LIF neurons model shard activity as spike trains
+  - STDP learns normal logic-flow patterns
+  - Anomaly detection via spike-timing divergence
+  - No pattern matching — purely spike-timing based
+"""
+
+import hashlib
+import time
+import uuid
+from collections import deque
+from dataclasses import dataclass, field
+from typing import Optional
+
+import numpy as np
+
+
+@dataclass
+class LIFNeuron:
+    """Leaky Integrate-and-Fire neuron.
+
+    dv/dt = ( -(v - v_rest) + I_syn ) / tau_m
+    When v > v_threshold: spike, then v = v_reset
+    """
+    neuron_id: str
+    v_rest: float = -70.0     # Resting potential (mV)
+    v_threshold: float = -50.0 # Firing threshold (mV)
+    v_reset: float = -65.0     # Reset potential (mV)
+    tau_m: float = 20.0        # Membrane time constant (ms)
+    refractory: float = 2.0    # Refractory period (ms)
+
+    # State
+    v: float = -70.0           # Current membrane potential
+    last_spike: float = -100.0 # Time of last spike
+    spike_count: int = 0
+
+    def update(self, I_syn: float, dt: float, t: float) -> bool:
+        """Update neuron for one timestep. Returns True if spike fired."""
+        # Refractory check
+        if t - self.last_spike < self.refractory:
+            return False
+
+        # LIF dynamics
+        dv = (-(self.v - self.v_rest) + I_syn) / self.tau_m * dt
+        self.v += dv
+
+        # Spike check
+        if self.v >= self.v_threshold:
+            self.v = self.v_reset
+            self.last_spike = t
+            self.spike_count += 1
+            return True
+
+        return False
+
+
+@dataclass
+class Synapse:
+    """Plastic synapse with STDP."""
+    pre_id: str
+    post_id: str
+    weight: float = 0.5
+    w_max: float = 1.0
+    w_min: float = 0.0
+
+    # STDP traces
+    Apre: float = 0.0
+    Apost: float = 0.0
+    tau_pre: float = 20.0   # ms
+    tau_post: float = 20.0  # ms
+    delta_Apre: float = 0.01
+    delta_Apost: float = -0.012
+
+    def pre_spike(self):
+        """Called when pre-synaptic neuron fires."""
+        self.Apre += self.delta_Apre
+        self.weight = np.clip(self.weight + self.Apost, self.w_min, self.w_max)
+
+    def post_spike(self):
+        """Called when post-synaptic neuron fires."""
+        self.Apost += self.delta_Apost
+        self.weight = np.clip(self.weight + self.Apre, self.w_min, self.w_max)
+
+    def decay_traces(self, dt: float):
+        """Decay STDP eligibility traces."""
+        self.Apre *= np.exp(-dt / self.tau_pre)
+        self.Apost *= np.exp(-dt / self.tau_post)
+
+
+class SpikingNetwork:
+    """A network of LIF neurons with STDP synapses for logic-flow anomaly detection."""
+
+    def __init__(self, n_neurons: int = 100):
+        self.n_neurons = n_neurons
+        self.neurons: list[LIFNeuron] = []
+        self.synapses: list[Synapse] = []
+        self.spike_history: deque = deque(maxlen=10000)
+        self.baseline_weights: Optional[np.ndarray] = None
+        self._t: float = 0.0
+        self._dt: float = 0.1  # ms
+
+        self._initialize_network(n_neurons)
+
+    def _initialize_network(self, n: int):
+        """Initialize random recurrent spiking network."""
+        rng = np.random.RandomState(42)
+
+        # Create neurons with varied parameters for heterogeneity
+        for i in range(n):
+            self.neurons.append(LIFNeuron(
+                neuron_id=f"n{i:04d}",
+                v_rest=-70.0 + rng.normal(0, 2),
+                v_threshold=-50.0 + rng.normal(0, 3),
+                tau_m=20.0 + rng.normal(0, 5),
+            ))
+
+        # Create sparse recurrent connections
+        for i in range(n):
+            n_conns = rng.randint(5, 20)
+            targets = rng.choice(n, n_conns, replace=False)
+            for j in targets:
+                if i != j:
+                    self.synapses.append(Synapse(
+                        pre_id=f"n{i:04d}",
+                        post_id=f"n{j:04d}",
+                        weight=float(np.abs(rng.normal(0.5, 0.2))),
+                    ))
+
+        # Build index maps for fast lookup
+        self._synapse_index: dict[tuple, int] = {
+            (s.pre_id, s.post_id): idx for idx, s in enumerate(self.synapses)
+        }
+
+    def train_baseline(self, n_steps: int = 5000):
+        """Train the network on normal logic-flow patterns to establish baseline."""
+        rng = np.random.RandomState(12345)
+        for step in range(n_steps):
+            # Inject Poisson-like input to simulate normal shard activity
+            input_spikes = rng.random(self.n_neurons) < 0.05  # 5 Hz background
+
+            self._step(input_spikes)
+            self._t += self._dt
+
+        # Store baseline weights
+        self.baseline_weights = np.array([s.weight for s in self.synapses])
+        print(f"[Spiking] Baseline trained: {n_steps} steps, {len(self.synapses)} synapses")
+
+    def _step(self, input_spikes: np.ndarray):
+        """Single simulation step."""
+        # Apply input spikes
+        for i, spiked in enumerate(input_spikes):
+            if spiked:
+                self.neurons[i].v += 5.0  # Quick depolarization
+
+        # Update neurons and collect spikes
+        neuron_spiked = []
+        for i, neuron in enumerate(self.neurons):
+            # Compute synaptic input
+            I_syn = 0.0
+            for j, other in enumerate(self.neurons):
+                if i != j:
+                    key = (f"n{j:04d}", f"n{i:04d}")
+                    if key in self._synapse_index:
+                        syn = self.synapses[self._synapse_index[key]]
+                        if other.last_spike > self._t - 5:  # Recent spike
+                            I_syn += syn.weight * 2.0
+
+            if neuron.update(I_syn, self._dt, self._t):
+                neuron_spiked.append(i)
+
+        # STDP updates
+        for i in neuron_spiked:
+            # Pre-spike STDP
+            for j, neuron in enumerate(self.neurons):
+                key = (f"n{i:04d}", f"n{j:04d}")
+                if key in self._synapse_index:
+                    self.synapses[self._synapse_index[key]].pre_spike()
+
+            # Post-spike STDP
+            for j, neuron in enumerate(self.neurons):
+                key = (f"n{j:04d}", f"n{i:04d}")
+                if key in self._synapse_index:
+                    self.synapses[self._synapse_index[key]].post_spike()
+
+        # Decay all traces
+        for syn in self.synapses:
+            syn.decay_traces(self._dt)
+
+        # Record spike history
+        if neuron_spiked:
+            self.spike_history.append({
+                "t": self._t,
+                "neurons": neuron_spiked,
+                "n_spikes": len(neuron_spiked),
+            })
+
+    def detect_anomaly(self, input_pattern: np.ndarray) -> float:
+        """Detect logic-flow anomaly via spike-timing divergence from baseline.
+        
+        Returns anomaly score [0, 1] — higher = more anomalous.
+        """
+        if self.baseline_weights is None:
+            raise RuntimeError("Must train baseline before anomaly detection")
+
+        # Run input through network
+        self._t += 10.0  # Advance time
+        n_steps = 20
+        pre_spikes = sum(n.spike_count for n in self.neurons)
+
+        for _ in range(n_steps):
+            self._step(input_pattern)
+            self._t += self._dt
+
+        post_spikes = sum(n.spike_count for n in self.neurons)
+        new_spikes = post_spikes - pre_spikes
+
+        # Weight divergence from baseline
+        current_weights = np.array([s.weight for s in self.synapses])
+        weight_divergence = np.mean(np.abs(current_weights - self.baseline_weights))
+
+        # Spike-timing irregularity
+        if len(self.spike_history) > 10:
+            recent_isi = []
+            recent_times = [h["t"] for h in list(self.spike_history)[-20:] if h["n_spikes"] > 0]
+            for i in range(1, len(recent_times)):
+                recent_isi.append(recent_times[i] - recent_times[i-1])
+            isi_cv = np.std(recent_isi) / np.mean(recent_isi) if recent_isi and np.mean(recent_isi) > 0 else 0.0
+        else:
+            isi_cv = 0.0
+
+        # Combine: weight divergence (60%) + spike irregularity (40%)
+        anomaly_score = 0.6 * min(weight_divergence * 10.0, 1.0) + 0.4 * min(isi_cv * 0.5, 1.0)
+
+        return float(np.clip(anomaly_score, 0.0, 1.0))
+
+
+class ZeroDayHunter:
+    """Hunt Zero-Day vulnerabilities by sensing logic-flow anomalies.
+    
+    Does NOT scan for code patterns. Instead:
+    1. Encodes shard activity as spike trains
+    2. Learns normal logic flow via STDP
+    3. Flags spike-timing anomalies as potential zero-days
+    """
+
+    def __init__(self, n_shards: int = 1000):
+        self.n_shards = n_shards
+        self.network = SpikingNetwork(n_neurons=min(n_shards, 200))
+        self.network.train_baseline(n_steps=5000)
+        self.detections: list[dict] = []
+        self.baseline_trained = True
+
+    def encode_shard_activity(self, shard_events: list[dict]) -> np.ndarray:
+        """Encode shard computation events as spike input to the SNN.
+        
+        Each shard event is converted to a spike pattern based on:
+        - Computation intensity
+        - Memory access pattern
+        - Logic gate switching frequency
+        """
+        n_neurons = self.network.n_neurons
+        input_pattern = np.zeros(n_neurons)
+
+        for event in shard_events:
+            # Encode event properties into spike probability
+            intensity = event.get("intensity", 0.5)
+            gate_switches = event.get("gate_switches", 0)
+            memory_touches = event.get("memory_touches", 0)
+            shard_id = event.get("shard_id", 0)
+
+            # Map to neuron index
+            neuron_idx = shard_id % n_neurons
+
+            # Spike probability based on logic-flow intensity
+            spike_prob = min(
+                (intensity * 0.3 + gate_switches * 0.05 + memory_touches * 0.02),
+                0.95
+            )
+            input_pattern[neuron_idx] = 1.0 if np.random.random() < spike_prob else 0.0
+
+        return input_pattern
+
+    def hunt(self, shard_activity: list[dict]) -> dict:
+        """Hunt for zero-day vulnerabilities in shard logic flow."""
+        input_pattern = self.encode_shard_activity(shard_activity)
+        anomaly_score = self.network.detect_anomaly(input_pattern)
+
+        detection = {
+            "timestamp": time.time(),
+            "anomaly_score": float(anomaly_score),
+            "n_events": len(shard_activity),
+            "threat_level": self._assess_threat(anomaly_score),
+            "signature": hashlib.sha256(
+                str(anomaly_score).encode() + str(time.time()).encode()
+            ).hexdigest()[:16],
+        }
+
+        if anomaly_score > 0.5:
+            self.detections.append(detection)
+
+        return detection
+
+    def _assess_threat(self, score: float) -> str:
+        if score > 0.8:
+            return "CRITICAL — Likely zero-day logic-flow anomaly"
+        elif score > 0.6:
+            return "HIGH — Significant deviation from normal spike pattern"
+        elif score > 0.4:
+            return "MEDIUM — Unusual logic flow detected"
+        elif score > 0.2:
+            return "LOW — Minor spike-timing variance"
+        else:
+            return "NORMAL — Standard logic flow"
+
+    def get_hunt_summary(self) -> dict:
+        return {
+            "total_scans": len(self.detections),
+            "critical": sum(1 for d in self.detections if d["threat_level"].startswith("CRITICAL")),
+            "high": sum(1 for d in self.detections if d["threat_level"].startswith("HIGH")),
+            "medium": sum(1 for d in self.detections if d["threat_level"].startswith("MEDIUM")),
+            "latest_score": self.detections[-1]["anomaly_score"] if self.detections else 0.0,
+        }