Unkerien/PLL-Sensor-Cyber-Guard
0
1"""2PLL-Sensor-Cyber-Guard-v1 -- Environment3=========================================4Discrete-time 2nd-order PLL state-space model with three cyber-attack5scenarios. Implements the OpenEnv 2026 Sync Protocol (player_id +6session_id turn validation).7 8Physics9-------10State vector x = [phase_error, freq_error]11 dx/dt = A·x + B·u + w (continuous)12 x[k+1] = Ad·x[k] + Bd·u[k] + w[k] (discrete, Ts = 0.001 s)13 14Natural frequency ωn = 2π·50 rad/s15Damping ratio ζ = 0.70716Measurement noise σ = 0.0217 18Attack Scenarios19----------------20task_1 (Easy) : Step Attack — sudden +2.0 spike in phase error21task_2 (Medium) : Frequency Ramp — +0.1 Hz/sec gradual offset22task_3 (Hard) : Stealthy FDI — within 1σ drift, long-term bias23"""24from __future__ import annotations25 26import math27import random28from typing import Optional29 30import numpy as np31 32from models import PLLObservation, PLLState33 34# ---------------------------------------------------------------------------35# PLL Constants36# ---------------------------------------------------------------------------37 38TS = 0.001 # sample period (1 kHz)39OMEGA_N = 2 * math.pi * 50 # natural frequency (rad/s)40ZETA = 0.707 # damping ratio41NOISE_STD = 0.02 # measurement noise σ42MAX_STEPS = 100 # episode length43BUFFER_LEN = 10 # rolling observation buffer size44VCO_GAIN = 0.5 # VCO voltage scale factor45 46# Discrete-time state-space (zero-order hold approximation)47_a11 = 1.048_a12 = TS49_a21 = -(OMEGA_N ** 2) * TS50_a22 = 1.0 - 2.0 * ZETA * OMEGA_N * TS51 52Ad = np.array([[_a11, _a12],53 [_a21, _a22]])54 55Bd = np.array([[0.0],56 [OMEGA_N ** 2 * TS]])57 58# ---------------------------------------------------------------------------59# Task Definitions60# ---------------------------------------------------------------------------61 62TASKS: dict[str, dict] = {63 "task_1": {64 "difficulty": "easy",65 "attack_type": "step_attack",66 "description": (67 "Step Attack — a sudden, large jump (+2.0 rad) in phase error "68 "injected at a random step in the episode."69 ),70 "attack_start_range": (15, 35), # step range for injection71 },72 "task_2": {73 "difficulty": "medium",74 "attack_type": "freq_ramp",75 "description": (76 "Frequency Ramp — a gradual +0.1 Hz/sec increase in frequency "77 "offset starting at a random step."78 ),79 "attack_start_range": (10, 30),80 },81 "task_3": {82 "difficulty": "hard",83 "attack_type": "stealthy_fdi",84 "description": (85 "Stealthy False Data Injection — subtle bias (≤ 1σ of normal "86 "noise) causing long-term drift without obvious spikes."87 ),88 "attack_start_range": (5, 25),89 },90}91 92 93# ---------------------------------------------------------------------------94# Grading95# ---------------------------------------------------------------------------96 97def compute_score(98 attack_type_true: str,99 attack_start_step: int,100 detection_step: Optional[int],101 detected_type: Optional[str],102 false_positives: int,103 max_steps: int,104) -> tuple[float, float, str]:105 """106 Reward formula:107 R = (A × Success) - (β × Latency) - (F × FalsePositive)108 109 Returns (reward, score, feedback).110 """111 parts: list[str] = []112 113 # --- Success component (A = 1.0) ---114 if detection_step is None:115 success = 0.0116 parts.append("[MISS] Attack not detected")117 elif detected_type == attack_type_true:118 success = 1.0119 parts.append(f"[OK] Correct classification: {detected_type}")120 else:121 success = 0.5122 parts.append(123 f"[PARTIAL] Detected but wrong type: "124 f"predicted '{detected_type}', actual '{attack_type_true}'"125 )126 127 # --- Latency component (β = 0.3) ---128 beta = 0.3129 if detection_step is not None and detection_step >= attack_start_step:130 latency = (detection_step - attack_start_step) / max_steps131 else:132 latency = 1.0 # max penalty if missed or false-early133 latency_pen = beta * latency134 parts.append(f"Latency penalty: -{latency_pen:.3f}")135 136 # --- False-positive component (F = 0.2 each) ---137 fp_pen = 0.2 * false_positives138 if false_positives > 0:139 parts.append(f"[FP] {false_positives} false positive(s): -{fp_pen:.2f}")140 141 reward = success - latency_pen - fp_pen142 score = max(0.0, min(1.0, reward))143 144 return reward, score, " | ".join(parts)145 146 147# ---------------------------------------------------------------------------148# Environment149# ---------------------------------------------------------------------------150 151class PLLEnvironment:152 """OpenEnv 2026 Sync environment for PLL cyber-attack detection."""153 154 def __init__(self) -> None:155 self._task_id: Optional[str] = None156 self._task: Optional[dict] = None157 self._player_id: Optional[str] = None158 self._session_id: Optional[str] = None159 self._step_count: int = 0160 self._done: bool = True161 self._last_score: Optional[float] = None162 self._last_reward: Optional[float] = None163 164 # PLL state165 self._x: np.ndarray = np.zeros(2) # [phase_error, freq_error]166 self._buffer: list[float] = []167 self._vco_voltage: float = 0.0168 169 # Attack bookkeeping170 self._attack_start: int = 0171 self._attack_active: bool = False172 self._detection_step: Optional[int] = None173 self._detected_type: Optional[str] = None174 self._false_positives: int = 0175 176 # RNG177 self._rng = np.random.default_rng()178 179 # -- validation --------------------------------------------------------180 181 def _validate_turn(self, player_id: str, session_id: str) -> None:182 if self._player_id and player_id != self._player_id:183 raise PermissionError(184 f"Player mismatch: expected '{self._player_id}', "185 f"got '{player_id}'."186 )187 if self._session_id and session_id != self._session_id:188 raise PermissionError(189 f"Session mismatch: expected '{self._session_id}', "190 f"got '{session_id}'."191 )192 193 # -- PLL physics -------------------------------------------------------194 195 def _pll_step(self) -> None:196 """Advance the PLL state by one discrete time step."""197 u = np.array([[0.0]]) # no external reference change198 noise = self._rng.normal(0.0, NOISE_STD, size=2)199 self._x = Ad @ self._x + (Bd @ u).flatten() + noise200 self._vco_voltage = float(self._x[1]) * VCO_GAIN201 202 def _inject_attack(self) -> None:203 """Inject the task-specific attack into the PLL state."""204 if not self._task or self._step_count < self._attack_start:205 return206 207 self._attack_active = True208 attack_type = self._task["attack_type"]209 elapsed = self._step_count - self._attack_start210 211 if attack_type == "step_attack":212 # One-shot spike at the attack-start step213 if elapsed == 0:214 self._x[0] += 2.0215 elif attack_type == "freq_ramp":216 # +0.1 Hz/sec => ramp = 0.1 * elapsed * Ts217 self._x[1] += 0.1 * TS218 elif attack_type == "stealthy_fdi":219 # Bias ≤ 1σ injected every step (cumulative drift)220 bias = NOISE_STD * 0.8 # 80 % of 1-sigma221 self._x[0] += bias * TS222 223 # -- observation -------------------------------------------------------224 225 def _make_obs(self, score=None, reward=None, feedback=None) -> PLLObservation:226 phase_err = float(self._x[0])227 freq_dev = float(self._x[1])228 229 # Update rolling buffer230 self._buffer.append(phase_err)231 if len(self._buffer) > BUFFER_LEN:232 self._buffer = self._buffer[-BUFFER_LEN:]233 234 return PLLObservation(235 task_id=self._task_id or "",236 step_count=self._step_count,237 phase_error=round(phase_err, 6),238 freq_deviation=round(freq_dev, 6),239 vco_voltage=round(self._vco_voltage, 6),240 buffer=[round(v, 6) for v in self._buffer],241 done=self._done,242 score=score,243 reward=reward,244 feedback=feedback,245 )246 247 # -- reset -------------------------------------------------------------248 249 def reset(250 self,251 player_id: str,252 session_id: str,253 task_id: Optional[str] = None,254 ) -> PLLObservation:255 if task_id is None:256 task_id = random.choice(list(TASKS.keys()))257 if task_id not in TASKS:258 raise ValueError(259 f"Unknown task_id '{task_id}'. Available: {list(TASKS.keys())}"260 )261 262 self._player_id = player_id263 self._session_id = session_id264 self._task_id = task_id265 self._task = TASKS[task_id]266 self._step_count = 0267 self._done = False268 self._last_score = None269 self._last_reward = None270 271 # Reset PLL state272 self._x = np.zeros(2)273 self._buffer = []274 self._vco_voltage = 0.0275 276 # Randomise attack injection step277 lo, hi = self._task["attack_start_range"]278 self._attack_start = random.randint(lo, hi)279 self._attack_active = False280 self._detection_step = None281 self._detected_type = None282 self._false_positives = 0283 284 # Seed the RNG for reproducibility within an episode285 self._rng = np.random.default_rng()286 287 return self._make_obs()288 289 # -- step --------------------------------------------------------------290 291 def step(292 self,293 player_id: str,294 session_id: str,295 attack_detected: bool,296 attack_type: str,297 confidence: float,298 ) -> PLLObservation:299 if self._done or self._task is None:300 raise RuntimeError("Episode not active. Call /reset first.")301 self._validate_turn(player_id, session_id)302 303 self._step_count += 1304 305 # 1. Advance physics306 self._pll_step()307 308 # 2. Inject attack (if applicable)309 self._inject_attack()310 311 # 3. Process agent classification312 if attack_detected:313 if self._attack_active and self._detection_step is None:314 # First correct detection315 self._detection_step = self._step_count316 self._detected_type = attack_type317 elif not self._attack_active:318 # False positive (attack hasn't started yet)319 self._false_positives += 1320 321 # 4. Check episode termination322 if self._step_count >= MAX_STEPS:323 self._done = True324 reward, score, feedback = compute_score(325 attack_type_true=self._task["attack_type"],326 attack_start_step=self._attack_start,327 detection_step=self._detection_step,328 detected_type=self._detected_type,329 false_positives=self._false_positives,330 max_steps=MAX_STEPS,331 )332 self._last_score = score333 self._last_reward = reward334 return self._make_obs(score=score, reward=reward, feedback=feedback)335 336 # Not done yet — return observation without score337 return self._make_obs()338 339 # -- state -------------------------------------------------------------340 341 def state(self) -> PLLState:342 return PLLState(343 current_task_id=self._task_id,344 player_id=self._player_id,345 session_id=self._session_id,346 step_count=self._step_count,347 max_steps=MAX_STEPS,348 done=self._done,349 last_score=self._last_score,350 last_reward=self._last_reward,351 phase_error=round(float(self._x[0]), 6),352 freq_deviation=round(float(self._x[1]), 6),353 vco_voltage=round(self._vco_voltage, 6),354 buffer=[round(v, 6) for v in self._buffer],355 attack_active=self._attack_active,356 )357 358