4b96524964
BackupExporter (backup_exporter.py): - Export complet (workflows, correction packs, coaching sessions, configs) - Export sélectif (workflows only, configs only, etc.) - Export modèles entraînés opt-in (embeddings, FAISS anonymisés) - Sanitisation des configs (masquage des secrets) - Statistiques de backup disponibles VersionManager (version_manager.py): - Suivi de version avec composants - Vérification des mises à jour (manifest local) - Vérification intégrité packages (SHA-256) - Création/restauration de backups pour rollback - Information système complète Ces modules supportent les fonctionnalités Dashboard: - Téléchargement sauvegardes par le client - Mise à jour du système - Rollback en cas de problème Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
98 lines
2.9 KiB
Python
98 lines
2.9 KiB
Python
"""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() |