"""core/system/safety_switch.py Fiche #23 - Kill-switch + DEMO_SAFE - Kill-switch: bloque toute action "écrivant" (ou tout sauf health) quand activé. - DEMO_SAFE : mode démo, interdit les endpoints dangereux même avec token admin. Activation: - DEMO_SAFE=1 - RPA_KILL_SWITCH=1 Optionnel: - RPA_KILL_SWITCH_FILE=data/runtime/kill_switch.json Si présent et contient {"enabled": true} -> kill-switch actif. Note: MVP: lecture file à la demande (cheap, robuste), pas de watcher. """ from __future__ import annotations import os import json from pathlib import Path from typing import Optional def _is_truthy(v: Optional[str]) -> bool: if v is None: return False return v.strip().lower() in {"1", "true", "yes", "y", "on"} def demo_safe_enabled() -> bool: return _is_truthy(os.getenv("DEMO_SAFE")) def kill_switch_env_enabled() -> bool: return _is_truthy(os.getenv("RPA_KILL_SWITCH")) def kill_switch_file_path() -> Path: return Path(os.getenv("RPA_KILL_SWITCH_FILE", "data/runtime/kill_switch.json")) def kill_switch_file_enabled() -> bool: path = kill_switch_file_path() if not path.exists(): return False try: data = json.loads(path.read_text(encoding="utf-8")) return bool(data.get("enabled")) except Exception: return True # fichier illisible -> safe side def kill_switch_enabled() -> bool: return kill_switch_env_enabled() or kill_switch_file_enabled() def set_kill_switch(enabled: bool, reason: str = "manual") -> None: path = kill_switch_file_path() path.parent.mkdir(parents=True, exist_ok=True) payload = {"enabled": bool(enabled), "reason": reason} path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") # Backward compatibility - simple safety switch class for existing code class SafetySwitch: """Simple safety switch for backward compatibility.""" def is_demo_safe_mode(self) -> bool: return demo_safe_enabled() def is_kill_switch_active(self) -> bool: return kill_switch_enabled() def is_feature_enabled(self, feature_name: str) -> bool: """Check if a feature is enabled based on safety settings.""" if self.is_kill_switch_active(): # When kill switch is active, only health endpoints are enabled return feature_name in ["health", "healthz", "status"] if self.is_demo_safe_mode(): # In demo safe mode, restrict certain features restricted_features = ["admin_write", "system_modify", "data_delete"] if feature_name in restricted_features: return False # Demo endpoints are enabled in demo mode if feature_name == "demo_endpoints": return True # Default: feature is enabled return True def get_safety_switch() -> SafetySwitch: """Retourne une instance du safety switch pour compatibilité.""" return SafetySwitch()