diff --git a/system_monitor.py b/system_monitor.py new file mode 100644 index 0000000..07b0cf8 --- /dev/null +++ b/system_monitor.py @@ -0,0 +1,1164 @@ +#!/usr/bin/env python3 +""" +Fleet Control - Node Agent (v2.0) + +Compatibile con node_config.json esistenti: tutte le chiavi nuove sono OPZIONALI. +Vedi AGENT_CHANGELOG.md per l'elenco completo delle modifiche e dei nuovi parametri. +""" +import paho.mqtt.client as mqtt +import json +import time +import psutil +import platform +import subprocess +import threading +import sys +import os +import shutil +import re +import queue +import random +import signal +import requests +from pathlib import Path +import logging +from logging.handlers import RotatingFileHandler + +AGENT_VERSION = "2.0" + +# --- CALCOLO PERCORSI DINAMICI --- +AGENT_DIR = os.path.dirname(os.path.abspath(__file__)) +LOG_FILE = os.path.join(AGENT_DIR, 'node_agent.log') +CONFIG_FILE = os.environ.get('FLEET_AGENT_CONFIG', os.path.join(AGENT_DIR, 'node_config.json')) + +# ========================================== +# 0. LOGGING & HARDWARE CONFIGURATION +# ========================================== +logging.basicConfig( + handlers=[ + RotatingFileHandler(LOG_FILE, maxBytes=2000000, backupCount=3), + logging.StreamHandler() + ], + level=logging.INFO, + format='[%(asctime)s] %(levelname)s - %(message)s', + datefmt='%Y-%m-%d %H:%M:%S' +) +logger = logging.getLogger("NodeAgent") + +try: + import RPi.GPIO as GPIO + GPIO_AVAILABLE = True +except ImportError: + GPIO_AVAILABLE = False + logger.warning("RPi.GPIO library not found. Hardware reset disabled.") +except RuntimeError as e: + # La libreria c'รจ ma non funziona su questa scheda (l'import stesso solleva RuntimeError): non deve bloccare l'agent + GPIO_AVAILABLE = False + logger.warning(f"RPi.GPIO not usable on this board ({e}). GPIO hardware reset disabled.") + +# ========================================== +# 1. CONFIGURATION LOADING (con validazione) +# ========================================== +CONFIG_PATH = Path(CONFIG_FILE) + + +def load_config(): + try: + if not CONFIG_PATH.exists(): + logger.error(f"ERROR: File {CONFIG_PATH} not found!") + sys.exit(1) + with open(CONFIG_PATH, 'r') as f: + data = json.load(f) + except Exception as e: + logger.error(f"CRITICAL JSON ERROR: {e}") + sys.exit(1) + + if not isinstance(data, dict): + logger.error("CRITICAL: node_config.json must contain a JSON object") + sys.exit(1) + + missing = [] + if not str(data.get('client_id', '')).strip(): + missing.append('client_id') + mqtt_cfg = data.get('mqtt') if isinstance(data.get('mqtt'), dict) else {} + for k in ('broker', 'port', 'user', 'password'): + if k not in mqtt_cfg: + missing.append(f'mqtt.{k}') + if missing: + logger.error(f"CRITICAL: missing required settings in node_config.json: {', '.join(missing)}") + sys.exit(1) + + # sezioni opzionali: sempre presenti come dict (niente KeyError in giro per il codice) + for section in ('paths', 'settings', 'telegram', 'profiles'): + if not isinstance(data.get(section), dict): + data[section] = {} + return data + + +cfg = load_config() + +# Identifiers and Topics +CLIENT_ID = str(cfg['client_id']).strip().lower() +MQTT_CFG = cfg['mqtt'] +BASE_TOPIC = MQTT_CFG.get('base_topic', f"servizi/{CLIENT_ID}") +if BASE_TOPIC != f"servizi/{CLIENT_ID}": + logger.warning(f"mqtt.base_topic='{BASE_TOPIC}' diverso da 'servizi/{CLIENT_ID}': " + f"il server di controllo usa SEMPRE servizi/ e non ti vedrebbe.") + +TOPIC_CMD = f"{BASE_TOPIC}/cmnd" +TOPIC_STAT = f"{BASE_TOPIC}/stat" +TOPIC_REQUEST = "devices/control/request" +TOPIC_CONTROL = f"devices/{CLIENT_ID}/control" +TOPIC_CONFIG_SET = f"devices/{CLIENT_ID}/config_set/" +TOPIC_LOGS = f"devices/{CLIENT_ID}/logs" +TOPIC_SERVICES = f"devices/{CLIENT_ID}/services" + +# Percorsi (tutti configurabili in "paths") +FILE_LIST_PATH = cfg['paths'].get('file_list') or '/opt/file_list.txt' +PROCESS_LIST_PATH = cfg['paths'].get('process_list') or '/opt/process_list.txt' +LAST_PROFILE_PATH = cfg['paths'].get('last_profile') or '/opt/last_profile.txt' +STATE_PATH = cfg['paths'].get('state_file') or '/opt/agent_state.json' + +# Global Status Variables +mqtt_client = None +current_status = "ONLINE" +auto_healing_counter = {} +stopped_by_admin = set() # servizi fermati volontariamente dalla dashboard (l'auto-healing li ignora) +ops_lock = threading.RLock() # una sola operazione "pesante" alla volta (profilo, config, reset, update...) +heal_pause_until = 0.0 +disconnected_since = None +published_sig = {} # percorso file -> (mtime_ns, size) dell'ultima versione pubblicata +_warned_units = set() + +cmd_queue = queue.Queue(maxsize=20) +tg_queue = queue.Queue(maxsize=50) + +UNIT_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9_.@\-]{0,63}$') +USB_ID_RE = re.compile(r'^[A-Za-z0-9][A-Za-z0-9:/_.\- ]{0,63}$') +SLUG_RE = re.compile(r'^[A-Za-z0-9_.\-]{1,64}$') +SENSITIVE_KEY_RE = re.compile(r'^(.*password|passwd|pass|secret|.*token|authkey|apikey)$', re.I) +MAX_CONFIG_BYTES = 65536 + + +def int_setting(name, default, minimum=0): + try: + value = int(cfg['settings'].get(name, default)) + except (TypeError, ValueError): + value = default + return max(minimum, value) + + +def get_update_interval(): + return int_setting('update_interval', 30, 10) + + +# ========================================== +# 2. HELPERS: SICUREZZA, FILE, COMANDI +# ========================================== +def redact(text): + """Toglie dai testi (log, messaggi) token Telegram e credenziali negli URL.""" + text = str(text) + token = str(cfg['telegram'].get('token') or '') + if token and token != "TOKEN ID": + text = text.replace(token, '***') + text = re.sub(r'bot\d+:[A-Za-z0-9_\-]+', 'bot***', text) + text = re.sub(r'(https?://)[^/@\s]+@', r'\1***@', text) + return text + + +def safe_write_bytes(path, data): + """Scrittura atomica (file temporaneo + fsync + replace) mantenendo permessi e proprietario. + Se la directory non รจ scrivibile (es. /etc con file concesso all'utente) ripiega sulla scrittura in-place.""" + path = str(path) + tmp = f"{path}.tmp.{os.getpid()}" + try: + st = None + try: + st = os.stat(path) + except FileNotFoundError: + pass + with open(tmp, 'wb') as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + if st is not None: + os.chmod(tmp, st.st_mode & 0o777) + try: + os.chown(tmp, st.st_uid, st.st_gid) + except (PermissionError, OSError): + pass + os.replace(tmp, path) + return + except OSError as e: + try: + os.unlink(tmp) + except OSError: + pass + logger.warning(f"Scrittura atomica di {path} non possibile ({e}): scrivo in-place.") + with open(path, 'wb') as f: + f.write(data) + f.flush() + os.fsync(f.fileno()) + + +def safe_write_text(path, text): + safe_write_bytes(path, text.encode('utf-8')) + + +def run_cmd(args, timeout=30, cwd=None): + """Esegue un comando SENZA shell e con timeout. rc=124 timeout, rc=127 comando mancante.""" + try: + return subprocess.run(args, capture_output=True, text=True, timeout=timeout, cwd=cwd) + except subprocess.TimeoutExpired: + logger.error(f"Timeout ({timeout}s) eseguendo: {' '.join(map(str, args))}") + return subprocess.CompletedProcess(args, 124, '', 'timeout') + except FileNotFoundError: + return subprocess.CompletedProcess(args, 127, '', 'command not found') + except Exception as e: + return subprocess.CompletedProcess(args, 1, '', str(e)) + + +def read_lines(path): + """Righe non vuote e non commentate di un file di testo ([] se manca).""" + try: + p = Path(str(path)) + if not str(path) or not p.is_file(): + return [] + return [l.strip() for l in p.read_text(encoding='utf-8', errors='ignore').splitlines() + if l.strip() and not l.strip().startswith('#')] + except Exception as e: + logger.error(f"Errore lettura {path}: {e}") + return [] + + +def connected(): + return mqtt_client is not None and mqtt_client.is_connected() + + +def log_to_broker(msg): + try: + if mqtt_client is not None: + mqtt_client.publish(TOPIC_LOGS, msg) + except Exception as e: + logger.error(f"Publish log error: {e}") + + +def publish_status(): + """Stato del nodo (retained). Va ripubblicato a ogni connessione: dopo un blip il LWT l'ha messo OFFLINE.""" + try: + if mqtt_client is not None: + mqtt_client.publish(TOPIC_STAT, current_status, qos=1, retain=True) + except Exception as e: + logger.error(f"Publish status error: {e}") + + +def note_ops_done(): + """Dopo un'operazione l'auto-healing resta in pausa qualche secondo (i servizi stanno ripartendo).""" + global heal_pause_until + heal_pause_until = time.time() + int_setting('heal_grace_seconds', 20, 0) + + +# ========================================== +# 3. TELEGRAM (invio in un thread dedicato) +# ========================================== +def in_quiet_hours(hour): + q = cfg['telegram'].get('quiet_hours', {"start": 23, "end": 7}) + if q is False or q is None or not isinstance(q, dict): + return False + try: + start, end = int(q.get('start', 23)), int(q.get('end', 7)) + except (TypeError, ValueError): + return False + if start == end: + return False + if start < end: + return start <= hour < end + return hour >= start or hour < end + + +def send_telegram_message(message, urgent=False): + t_cfg = cfg['telegram'] + if not t_cfg.get('enabled', False): + return + + current_hour = int(time.strftime("%H")) + if not urgent and in_quiet_hours(current_hour): + logger.info(f"๐ŸŒ™ Quiet hours ({current_hour}:00): Telegram notification skipped.") + return + + token = t_cfg.get('token') + chat_id = t_cfg.get('chat_id') + if not token or not chat_id or token == "TOKEN ID": + return + + clean_msg = str(message).replace("", "").replace("", "") + try: + tg_queue.put_nowait((token, chat_id, f"[{CLIENT_ID.upper()}]\n{clean_msg}")) + except queue.Full: + logger.warning("Telegram queue full: message dropped") + + +def telegram_worker(): + while True: + token, chat_id, text = tg_queue.get() + try: + r = requests.post(f"https://api.telegram.org/bot{token}/sendMessage", + json={"chat_id": chat_id, "text": text}, timeout=10) + if not r.ok: + logger.error(f"Telegram HTTP {r.status_code}") + except Exception as e: + # l'eccezione di requests contiene l'URL con il token: va ripulita + logger.error(f"Telegram sending error: {type(e).__name__}: {redact(e)}") + + +# ========================================== +# 4. STATO PERSISTENTE (profilo attivo, servizi fermati a mano) +# ========================================== +def norm_name(name): + n = str(name).strip().lower() + return n[:-8] if n.endswith('.service') else n + + +def load_state(): + global stopped_by_admin + try: + with open(STATE_PATH, 'r') as f: + data = json.load(f) + stopped_by_admin = {norm_name(x) for x in data.get('stopped', [])} + except FileNotFoundError: + pass + except Exception as e: + logger.warning(f"Impossibile leggere {STATE_PATH}: {e}") + + +def save_state(): + try: + safe_write_text(STATE_PATH, json.dumps({"stopped": sorted(stopped_by_admin)})) + except Exception as e: + logger.warning(f"Impossibile salvare {STATE_PATH}: {e}") + + +def get_actual_config_from_disk(): + try: + if os.path.exists(LAST_PROFILE_PATH): + with open(LAST_PROFILE_PATH, "r") as f: + label = f.read().strip() + if label: + return f"ONLINE - {label}" + except Exception as e: + logger.error(f"Errore lettura memoria profilo: {e}") + return "ONLINE" # Default se il file non esiste o รจ vuoto + + +def save_last_profile(label): + try: + safe_write_text(LAST_PROFILE_PATH, label) + except Exception as e: + logger.error(f"Impossibile salvare il profilo attivo in {LAST_PROFILE_PATH}: {e}") + + +# ========================================== +# 5. SYSTEMD: NOMI UNITร€ (whitelist) E CONTROLLI +# ========================================== +def build_unit_map(): + """nome normalizzato -> nome unitร . Solo ciรฒ che il nodo conosce: process_list, profili, settings.service_map.""" + m = {} + for name in read_lines(PROCESS_LIST_PATH): + n = norm_name(name) + if UNIT_RE.match(n): + m[n] = n + for profile in cfg['profiles'].values(): + for s in (profile.get('services', []) if isinstance(profile, dict) else []): + name = str(s.get('name', '')) + if UNIT_RE.match(name): + m[norm_name(name)] = name + for k, v in (cfg['settings'].get('service_map') or {}).items(): + if UNIT_RE.match(str(v)): + m[norm_name(k)] = str(v) + return m + + +def resolve_unit(name): + return build_unit_map().get(norm_name(name)) + + +def systemctl(action, unit, timeout=60): + return run_cmd(["sudo", "systemctl", action, unit], timeout=timeout) + + +def unit_is_active(unit): + r = run_cmd(["systemctl", "is-active", unit], timeout=10) + return r.returncode == 0 and r.stdout.strip() == 'active' + + +def unit_exists(unit): + return run_cmd(["systemctl", "cat", unit], timeout=10).returncode == 0 + + +def unit_is_disabled(unit): + r = run_cmd(["systemctl", "is-enabled", unit], timeout=10) + return r.stdout.strip() in ('disabled', 'masked') + + +def wait_units_active(units, seconds): + """Attende fino a `seconds` che tutte le unitร  siano attive. Ritorna la lista di quelle NON attive.""" + deadline = time.monotonic() + seconds + pending = list(units) + while True: + pending = [u for u in pending if not unit_is_active(u)] + if not pending or time.monotonic() >= deadline: + return pending + time.sleep(1) + + +# ========================================== +# 6. RESET HARDWARE (USB / GPIO) +# ========================================== +def gpio_pin_config(): + pin = cfg['settings'].get('gpio_reset_pin') + if pin in (None, '', 0, False): + return None # nessun default implicito: il GPIO si usa solo se configurato + try: + return int(pin) + except (TypeError, ValueError): + return None + + +def hardware_reset(auto=False): + """Reset del modem. Ritorna {"attempted", "ok", "method", "msg"}.""" + usb_id = str(cfg['settings'].get('usb_reset_id') or '').strip() + if usb_id: + if not USB_ID_RE.match(usb_id): + return {"attempted": False, "ok": False, "method": "USB", "msg": f"โš ๏ธ usb_reset_id non valido: {usb_id!r}"} + logger.info(f"Esecuzione reset USB per il dispositivo {usb_id}...") + r = run_cmd(["sudo", "usbreset", usb_id], timeout=15) + time.sleep(1.5) + if r.returncode == 0: + return {"attempted": True, "ok": True, "method": "USB", "msg": f"๐Ÿ”Œ USB Reset ({usb_id}) inviato!"} + return {"attempted": True, "ok": False, "method": "USB", + "msg": f"โš ๏ธ USB Reset ({usb_id}) fallito (rc={r.returncode}): {r.stderr.strip()[:120]}"} + + pin = gpio_pin_config() + if pin is not None and GPIO_AVAILABLE and (not auto or cfg['settings'].get('gpio_reset_auto', True)): + logger.info(f"Esecuzione reset GPIO (HAT) sul pin {pin}...") + try: + GPIO.setwarnings(False) + GPIO.setmode(GPIO.BCM) + GPIO.setup(pin, GPIO.OUT) + GPIO.output(pin, GPIO.LOW) + time.sleep(0.5) + GPIO.output(pin, GPIO.HIGH) + GPIO.cleanup(pin) + time.sleep(1.5) + return {"attempted": True, "ok": True, "method": "HAT", "msg": f"๐Ÿ”Œ GPIO Pulse (Pin {pin}) inviato!"} + except Exception as e: + logger.error(f"Errore GPIO: {e}") + return {"attempted": True, "ok": False, "method": "HAT", "msg": f"โš ๏ธ Errore GPIO (Pin {pin}): {e}"} + + return {"attempted": False, "ok": False, "method": "", + "msg": "โš ๏ธ Nessun metodo di reset (USB o GPIO) configurato o disponibile sul nodo."} + + +# ========================================== +# 7. TELEMETRIA +# ========================================== +def get_cpu_temperature(): + temp = 0.0 + try: + temps = psutil.sensors_temperatures() if hasattr(psutil, 'sensors_temperatures') else {} + if temps.get('cpu_thermal'): + temp = temps['cpu_thermal'][0].current + elif temps.get('coretemp'): + temp = temps['coretemp'][0].current + elif platform.system() == "Linux": + r = run_cmd(["vcgencmd", "measure_temp"], timeout=5) + m = re.search(r'temp=([\d.]+)', r.stdout) if r.returncode == 0 else None + if m: + temp = float(m.group(1)) + except Exception as e: + logger.debug(f"Temperature read error: {e}") + return round(temp, 1) + + +def get_system_status(): + status = { + "cpu_usage_percent": psutil.cpu_percent(interval=0.5), + "cpu_temp": get_cpu_temperature(), + "memory_usage_percent": psutil.virtual_memory().percent, + "disk_usage_percent": psutil.disk_usage('/').percent, + "processes": {}, + "timestamp": time.strftime("%H:%M:%S"), + "profiles": {k: (v.get('label', f'PROFILE {k}') if isinstance(v, dict) else f'PROFILE {k}') + for k, v in cfg['profiles'].items()}, + "update_interval": get_update_interval(), # il server ne ricava le soglie "agent muto" + "agent_version": AGENT_VERSION, + } + names = read_lines(PROCESS_LIST_PATH) + if names: + try: + running = {(p.info.get('name') or '').lower() for p in psutil.process_iter(['name'])} + for name in names: + n = name.strip().lower() + if n: + status["processes"][n] = "online" if n in running else "offline" + except Exception as e: + logger.error(f"Process check error: {e}") + return status + + +def publish_telemetry(): + status = get_system_status() + stems = [Path(f).stem for f in read_lines(FILE_LIST_PATH)] + status["config_files"] = stems + status["files"] = stems + # nomi completi con estensione (es. dstargateway.cfg): la dashboard li usa solo per l'etichetta + status["file_names"] = [Path(f).name for f in read_lines(FILE_LIST_PATH)] + if connected(): + mqtt_client.publish(TOPIC_SERVICES, json.dumps(status), qos=1) + return status + + +# ========================================== +# 8. PUBBLICAZIONE CONFIGURAZIONI (solo se cambiate) +# ========================================== +def parse_ini(text): + ini_data = {} + current_section = None + for line in text.splitlines(): + line = line.strip() + if not line or line.startswith(('#', ';')): + continue + if line.startswith('[') and line.endswith(']'): + current_section = line[1:-1].strip() + ini_data[current_section] = {} + elif '=' in line and current_section is not None: + k, v = line.split('=', 1) + k, v = k.strip(), v.strip() + if k in ini_data[current_section]: + ini_data[current_section][k] = str(ini_data[current_section][k]) + "," + v + else: + ini_data[current_section][k] = v + return ini_data + + +def mask_sensitive(ini_data): + """Nasconde Password/Secret/Token nei topic per sezione (NON tocca PassAllTG/PassAllPC, che servono al server).""" + for payload in ini_data.values(): + for k, v in payload.items(): + if SENSITIVE_KEY_RE.match(k) and str(v).strip(): + payload[k] = "********" + return ini_data + + +def profile_only_sources(): + """File usati solo come sorgente di un profilo (es. *_BM.ini): per l'editor basta full_config.""" + sources, targets = set(), set() + for profile in cfg['profiles'].values(): + for s in (profile.get('services', []) if isinstance(profile, dict) else []): + if s.get('source'): + sources.add(s['source']) + if s.get('target'): + targets.add(s['target']) + return sources - targets + + +def publish_configs(force=False): + """Pubblica full_config e sezioni INI (retained) SOLO per i file cambiati dall'ultima volta.""" + if not connected(): + return + skip_sections = profile_only_sources() + mask = bool(cfg['settings'].get('mask_passwords', True)) + for fp in read_lines(FILE_LIST_PATH): + try: + if not os.path.isfile(fp): + continue + st = os.stat(fp) + sig = (st.st_mtime_ns, st.st_size) + if not force and published_sig.get(fp) == sig: + continue + text = Path(fp).read_text(encoding='utf-8', errors='replace') + stem = Path(fp).stem + mqtt_client.publish(f"data/{CLIENT_ID}/{stem}/full_config", + json.dumps({"raw_text": text}), qos=1, retain=True) + if fp not in skip_sections: + ini_data = parse_ini(text) + if mask: + mask_sensitive(ini_data) + for section, payload in ini_data.items(): + part = re.sub(r'[/+#]', '_', section) + if part: + mqtt_client.publish(f"data/{CLIENT_ID}/{stem}/{part}", json.dumps(payload), qos=1, retain=True) + published_sig[fp] = sig + except Exception as e: + logger.error(f"Error publishing config {fp}: {e}") + + +# ========================================== +# 9. CAMBIO PROFILO +# ========================================== +def find_profile(key): + k = str(key).strip().lower() + for pkey, profile in cfg['profiles'].items(): + if str(pkey).lower() == k and isinstance(profile, dict): + return pkey, profile + return None, None + + +def switch_config(key): + """Ritorna (ok, label_o_errore). Non tocca nessun servizio se i controlli preliminari falliscono; + se qualcosa va storto ripristina la configurazione precedente e riavvia i servizi.""" + pkey, profile = find_profile(key) + if not profile: + return False, f"Profile {key} not found in JSON" + label = profile.get('label', f"Profile {pkey}") + services = profile.get('services', []) + if not services: + return False, f"No services configured for {pkey}" + + # --- CONTROLLI PRELIMINARI (prima di fermare qualsiasi cosa) --- + for s in services: + name, src, dst = s.get('name'), s.get('source'), s.get('target') + if not (name and src and dst): + return False, f"Invalid service entry in profile {pkey}" + if not UNIT_RE.match(str(name)): + return False, f"Invalid unit name {name!r}" + if not os.path.isfile(src): + return False, f"Missing source file {src}" + if not os.path.isdir(os.path.dirname(str(dst)) or '.'): + return False, f"Target directory missing for {dst}" + + verify = int_setting('verify_seconds', 8, 0) + with ops_lock: + previous = {} + for s in services: + dst = s['target'] + try: + previous[dst] = Path(dst).read_bytes() if os.path.exists(dst) else None + except Exception as e: + return False, f"Cannot read target {dst}: {e}" + + error = None + try: + for s in services: + systemctl("stop", s['name']) + for s in services: + safe_write_bytes(s['target'], Path(s['source']).read_bytes()) + except Exception as e: + error = f"Copy failed: {e}" + + if error is None: + for s in services: + systemctl("start", s['name']) + if verify > 0: + failed = wait_units_active([s['name'] for s in services], verify) + if failed: + error = f"Units not active after switch: {', '.join(failed)}" + + if error: + logger.error(f"Cambio profilo {pkey} fallito ({error}): ripristino la configurazione precedente.") + for s in services: + systemctl("stop", s['name']) + for dst, data in previous.items(): + if data is not None: + try: + safe_write_bytes(dst, data) + except Exception as e: + logger.error(f"Rollback di {dst} fallito: {e}") + for s in services: + systemctl("start", s['name']) + note_ops_done() + return False, f"{error} - rolled back to previous configuration" + + for s in services: + stopped_by_admin.discard(norm_name(s['name'])) + save_state() + save_last_profile(label) + note_ops_done() + return True, label + + +# ========================================== +# 10. SCRITTURA CONFIG DA REMOTO +# ========================================== +def service_for_file(path): + """Servizio da riavviare dopo la modifica di un file di configurazione (None = nessuno).""" + stem = Path(path).stem + unit = resolve_unit(stem) + if unit: + return unit + guess = stem.lower() + if str(path) not in profile_only_sources() and UNIT_RE.match(guess) and unit_exists(guess): + return guess + return None + + +def apply_config_set(slug, payload): + try: + data = json.loads(payload) + except Exception as e: + logger.error(f"Config {slug}: JSON non valido ({e})") + log_to_broker(f"โŒ Config {slug}: invalid JSON") + return + raw = data.get('raw_text') if isinstance(data, dict) else None + if not isinstance(raw, str) or not raw.strip(): + logger.error(f"Config {slug}: raw_text mancante o vuoto: rifiutato") + log_to_broker(f"โŒ Config {slug}: empty or missing raw_text, refused") + return + raw_bytes = raw.encode('utf-8') + if len(raw_bytes) > MAX_CONFIG_BYTES: + logger.error(f"Config {slug}: troppo grande ({len(raw_bytes)} byte)") + log_to_broker(f"โŒ Config {slug}: too large") + return + + target = None + for line in read_lines(FILE_LIST_PATH): + if Path(line).stem.lower() == slug.lower(): + target = Path(line) + break + if target is None: + logger.error(f"Config {slug}: file non presente in file_list") + log_to_broker(f"โŒ Config {slug}: not in file list") + return + + verify = int_setting('verify_seconds', 8, 0) + with ops_lock: + previous = target.read_bytes() if target.exists() else None + if previous is not None: + try: + shutil.copy2(target, str(target) + ".bak") + except Exception as e: + logger.warning(f"Backup .bak non creato per {target}: {e}") + safe_write_bytes(target, raw_bytes) + + unit = service_for_file(target) + if unit: + systemctl("restart", unit) + if verify > 0: + failed = wait_units_active([unit], verify) + if failed: + logger.error(f"Config {slug}: {unit} non riparte con la nuova configurazione: ripristino.") + if previous is not None: + safe_write_bytes(target, previous) + systemctl("restart", unit) + log_to_broker(f"โŒ Config {slug}: service did not start, previous configuration restored") + send_telegram_message(f"โŒ Config {slug.upper()} rejected: service did not start (restored).") + note_ops_done() + return + else: + logger.info(f"Config {slug} salvata: nessun servizio da riavviare.") + note_ops_done() + + send_telegram_message(f"๐Ÿ“ Config {slug.upper()} updated via Web.") + logger.info(f"Configuration {slug} updated successfully.") + + +# ========================================== +# 11. AUTO-HEALING +# ========================================== +def check_auto_healing(status): + if not cfg['settings'].get('auto_healing', False): + return + if time.time() < heal_pause_until: + return + if not ops_lock.acquire(blocking=False): # c'รจ un'operazione in corso: ci riprovo al prossimo giro + return + try: + _heal(status) + finally: + ops_lock.release() + + +def _heal(status): + state_changed = False + for proc_name, state in status["processes"].items(): + norm = norm_name(proc_name) + + if state == "online": + auto_healing_counter[proc_name] = 0 + if norm in stopped_by_admin: + stopped_by_admin.discard(norm) + state_changed = True + continue + + if norm in stopped_by_admin: + continue # fermato volontariamente dalla dashboard + unit = resolve_unit(norm) + if not unit or not unit_exists(unit) or unit_is_disabled(unit): + if norm not in _warned_units: + _warned_units.add(norm) + logger.info(f"Auto-healing: '{proc_name}' ignorato (unitร  assente o disabilitata su questo nodo).") + continue + + attempts = auto_healing_counter.get(proc_name, 0) + if attempts < 3: + auto_healing_counter[proc_name] = attempts + 1 + msg = f"๐Ÿ›  Auto-healing: {proc_name} offline. Restarting {attempts + 1}/3..." + log_to_broker(msg) + send_telegram_message(msg) + + if norm == "mmdvmhost": + res = hardware_reset(auto=True) + if res["attempted"]: + log_to_broker(res["msg"]) + if not res["ok"]: + logger.warning(res["msg"]) + + systemctl("restart", unit) + elif attempts == 3: + msg = f"๐Ÿšจ CRITICAL: {proc_name} failed!" + log_to_broker(msg) + send_telegram_message(msg, urgent=True) + auto_healing_counter[proc_name] = 4 + + if state_changed: + save_state() + + +# ========================================== +# 12. COMANDI (eseguiti da un thread dedicato, mai nella callback MQTT) +# ========================================== +def enqueue(name, fn): + try: + cmd_queue.put_nowait((name, fn)) + return True + except queue.Full: + logger.error(f"Coda comandi piena: '{name}' scartato") + log_to_broker(f"โŒ Command queue full: {name} discarded") + return False + + +def command_worker(): + while True: + name, fn = cmd_queue.get() + try: + fn() + except Exception as e: + logger.exception(f"Errore eseguendo il comando '{name}'") + log_to_broker(f"โŒ ERROR ({name}): {redact(e)}"[:300]) + publish_status() + + +def do_switch_profile(key): + global current_status + ok, text = switch_config(key) + if ok: + current_status = f"ONLINE - {text}" + send_telegram_message(f"โœ… Multiple switch completed: {text}") + else: + logger.error(f"Cambio profilo fallito: {text}") + log_to_broker(f"โŒ {text}") + send_telegram_message(f"โŒ Profile switch failed: {text}") + publish_status() # anche in caso di errore: il server esce dallo stato "Sent..." + publish_telemetry() + publish_configs() + + +def do_update_agent(): + log_to_broker("๐Ÿ”„ Avvio aggiornamento OTA (git pull)...") + logger.info("Ricevuto comando di UPDATE_AGENT. Esecuzione git pull...") + with ops_lock: + head = run_cmd(["git", "rev-parse", "HEAD"], timeout=15, cwd=AGENT_DIR) + old = head.stdout.strip() if head.returncode == 0 else None + + pull = run_cmd(["git", "pull", "--ff-only"], timeout=120, cwd=AGENT_DIR) + if pull.returncode != 0: + err = redact(pull.stderr or pull.stdout)[:500] + log_to_broker(f"โŒ Errore aggiornamento:\n{err}") + logger.error(f"Errore git pull:\n{err}") + publish_status() + return + + head = run_cmd(["git", "rev-parse", "HEAD"], timeout=15, cwd=AGENT_DIR) + new = head.stdout.strip() if head.returncode == 0 else None + if old and new and old == new: + log_to_broker("โœ… Giร  aggiornato (nessuna modifica).") + publish_status() + return + + # La nuova versione deve almeno essere sintatticamente valida, altrimenti il nodo resterebbe irraggiungibile + check = run_cmd([sys.executable, "-c", + "import ast,sys; ast.parse(open(sys.argv[1], encoding='utf-8').read())", + os.path.abspath(__file__)], timeout=30) + if check.returncode != 0: + if old: + run_cmd(["git", "reset", "--hard", old], timeout=30, cwd=AGENT_DIR) + msg = redact(check.stderr)[:300] + log_to_broker(f"โŒ Nuova versione non valida: ripristinata la precedente.\n{msg}") + logger.error(f"Nuova versione non valida, rollback a {old}: {msg}") + publish_status() + return + + log_to_broker("โœ… Git pull completato. Riavvio in corso...") + logger.info(f"Aggiornamento completato:\n{pull.stdout}") + time.sleep(1) + # Systemd (Restart=always) lo fa ripartire con il nuovo codice + os._exit(1) + + +def do_reboot(): + mqtt_client.publish(TOPIC_STAT, f"OFFLINE - Rebooting {CLIENT_ID.upper()}...", qos=1, retain=True) + logger.info("REBOOT command received. Rebooting system...") + time.sleep(1) + r = run_cmd(["sudo", "reboot"], timeout=30) + if r.returncode != 0: + logger.error(f"Reboot fallito (rc={r.returncode}): {r.stderr.strip()[:200]}") + log_to_broker("โŒ Reboot failed") + publish_status() + + +def do_reset_hat(): + with ops_lock: + res = hardware_reset(auto=False) + if not res["attempted"]: + logger.warning(res["msg"]) + log_to_broker(res["msg"]) + publish_status() + return + logger.info("Restarting MMDVMHost...") + unit = resolve_unit("mmdvmhost") or "mmdvmhost" + systemctl("restart", unit) + if res["ok"]: + mqtt_client.publish(f"fleet/{CLIENT_ID}/status", f"{res['method']} RESET + MMDVM RESTART OK") + log_to_broker(f"{res['msg']} + MMDVMHost Restarted") + else: + mqtt_client.publish(f"fleet/{CLIENT_ID}/status", f"RESET ERROR: {res['msg']}") + log_to_broker(f"{res['msg']} (MMDVMHost restarted anyway)") + note_ops_done() + publish_status() + + +def do_toggle_telegram(new_state): + try: + # si parte dal file su disco (non dalla copia in memoria) per non perdere modifiche fatte a mano + with open(CONFIG_PATH, 'r') as f: + disk = json.load(f) + if not isinstance(disk.get('telegram'), dict): + disk['telegram'] = {} + disk['telegram']['enabled'] = new_state + safe_write_text(CONFIG_PATH, json.dumps(disk, indent=4)) + cfg['telegram']['enabled'] = new_state + log_to_broker(f"{'๐Ÿ””' if new_state else '๐Ÿ”‡'} Notifications {'ON' if new_state else 'OFF'}") + if new_state: + send_telegram_message("Notifications reactivated!") + except Exception as e: + logger.error(f"Error saving Telegram status: {e}") + publish_status() + + +def do_service_control(action, service): + unit = resolve_unit(service) + if not unit: + logger.warning(f"Servizio non consentito: {service[:40]!r}") + log_to_broker(f"โŒ Service not allowed: {service[:40]}") + publish_status() + return + with ops_lock: + r = systemctl(action, unit) + if r.returncode == 0: + n = norm_name(service) + if action == "stop": + stopped_by_admin.add(n) # l'auto-healing non lo rialzerร  + else: + stopped_by_admin.discard(n) + save_state() + log_to_broker(f"โœ… {action.upper()}: {service}") + logger.info(f"Service command executed: {action.upper()} {service}") + else: + err = r.stderr.strip()[:200] + log_to_broker(f"โŒ ERROR: {action} {service}: {err}") + logger.error(f"Error executing service command {action} {service}: {err}") + note_ops_done() + publish_telemetry() + publish_status() + + +def do_global_request(): + time.sleep(random.uniform(0, 5)) # evita che tutta la flotta risponda nello stesso istante + publish_telemetry() + publish_configs(force=True) + publish_status() + + +def initial_publish(): + publish_telemetry() + publish_configs(force=True) + + +# ========================================== +# 13. MQTT CALLBACKS (leggere: validano e mettono in coda) +# ========================================== +def handle_command(payload): + cmd = payload.upper() + pkey, profile = find_profile(payload) + if profile: + enqueue(f"profile {pkey}", lambda: do_switch_profile(pkey)) + elif cmd == "UPDATE_AGENT": + enqueue("update_agent", do_update_agent) + elif cmd == "REBOOT": + enqueue("reboot", do_reboot) + elif cmd == "RESET_HAT": + enqueue("reset_hat", do_reset_hat) + elif cmd in ("TG:OFF", "TG:ON"): + enqueue("toggle_telegram", lambda: do_toggle_telegram(cmd == "TG:ON")) + else: + logger.warning(f"Comando sconosciuto ignorato: {payload[:40]!r}") + log_to_broker(f"โ“ Unknown command: {payload[:40]}") + publish_status() + + +def handle_control(payload): + action, _, service = payload.partition(":") + action, service = action.strip().lower(), service.strip() + if action not in ("restart", "stop", "start") or not service: + logger.warning(f"Comando di controllo non valido: {payload[:60]!r}") + log_to_broker(f"โŒ Invalid control command: {payload[:60]}") + return + enqueue(f"{action} {service}", lambda: do_service_control(action, service)) + + +def on_connect(client, userdata, flags, reason_code, properties=None): + if reason_code == 0: + logger.info(f"โœ… Connected to MQTT broker: {CLIENT_ID.upper()}") + client.subscribe([ + (TOPIC_CMD, 0), + (TOPIC_REQUEST, 0), + (TOPIC_CONTROL, 0), + (TOPIC_CONFIG_SET + "#", 0) + ]) + publish_status() # a ogni connessione: dopo un blip il LWT ha messo il nodo OFFLINE + enqueue("initial publish", initial_publish) + else: + logger.error(f"โŒ MQTT connection error. Code: {reason_code}") + + +def on_connect_fail(client, userdata): + logger.warning("MQTT: connessione al broker non riuscita, riprovo...") + + +def on_disconnect(client, userdata, disconnect_flags, reason_code, properties=None): + logger.warning(f"โš ๏ธ Disconnesso dal broker MQTT (codice: {reason_code}).") + if cfg['settings'].get('exit_on_disconnect', False): + logger.error("exit_on_disconnect attivo: riavvio tramite systemd...") + os._exit(1) + + +def on_message(client, userdata, msg): + try: + topic = msg.topic + payload = msg.payload.decode('utf-8', errors='replace').strip() + + if getattr(msg, 'retain', False): + logger.warning(f"Messaggio retained ignorato su {topic} (i comandi non vanno mai pubblicati retained)") + return + + if topic == TOPIC_CMD: + handle_command(payload) + elif topic == TOPIC_REQUEST: + if payload.lower() in ("status", "update"): + logger.info("๐Ÿ“ฅ Received global update command (REQ CONFIG)") + enqueue("global request", do_global_request) + elif topic == TOPIC_CONTROL: + handle_control(payload) + elif topic.startswith(TOPIC_CONFIG_SET): + slug = topic[len(TOPIC_CONFIG_SET):] + if SLUG_RE.match(slug): + enqueue(f"config {slug}", lambda: apply_config_set(slug, payload)) + else: + logger.warning(f"config_set con nome non valido: {slug[:40]!r}") + except Exception: + logger.exception("Errore in on_message") + + +# ========================================== +# 14. THREAD DI SERVIZIO: TELEMETRIA E WATCHDOG +# ========================================== +def telemetry_loop(): + while True: + try: + if connected(): + status = publish_telemetry() + publish_configs() + else: + status = get_system_status() + check_auto_healing(status) # l'auto-healing รจ locale: funziona anche senza broker + except Exception as e: + logger.error(f"โŒ Errore critico nel thread telemetria: {e}") + time.sleep(get_update_interval()) + + +def watchdog_tick(now=None): + """Se il broker resta irraggiungibile oltre max_offline_seconds il processo esce e systemd lo rilancia.""" + global disconnected_since + now = time.time() if now is None else now + if connected(): + if disconnected_since is not None: + logger.info("โœ… Connessione MQTT ripristinata.") + disconnected_since = None + return + if disconnected_since is None: + disconnected_since = now + return + limit = int_setting('max_offline_seconds', 300, 30) + if now - disconnected_since >= limit: + logger.error(f"๐Ÿšจ Broker MQTT irraggiungibile da {int(now - disconnected_since)}s. Riavvio del processo...") + os._exit(1) + + +def watchdog_loop(): + while True: + try: + watchdog_tick() + except Exception as e: + logger.error(f"Watchdog error: {e}") + time.sleep(10) + + +def shutdown_handler(signum, frame): + """systemctl stop/restart: con una disconnessione 'pulita' il LWT non parte, quindi OFFLINE lo pubblichiamo noi.""" + logger.info("Shutdown richiesto: pubblico OFFLINE e chiudo.") + try: + info = mqtt_client.publish(TOPIC_STAT, f"OFFLINE - {CLIENT_ID.upper()}", qos=1, retain=True) + info.wait_for_publish(timeout=3) + mqtt_client.disconnect() + except Exception: + pass + os._exit(0) + + +def start_service(): + global current_status, mqtt_client, heal_pause_until + load_state() + current_status = get_actual_config_from_disk() + heal_pause_until = time.time() + int_setting('startup_grace_seconds', 60, 0) + + client = mqtt.Client(callback_api_version=mqtt.CallbackAPIVersion.VERSION2, client_id=CLIENT_ID.upper()) + mqtt_client = client + # LWT retained: se il nodo muore il broker sostituisce anche il messaggio "ONLINE" conservato + client.will_set(TOPIC_STAT, payload=f"OFFLINE - {CLIENT_ID.upper()}", qos=1, retain=True) + client.username_pw_set(MQTT_CFG['user'], MQTT_CFG['password']) + if MQTT_CFG.get('tls'): + client.tls_set(ca_certs=MQTT_CFG.get('ca_cert') or None) + client.reconnect_delay_set(min_delay=1, max_delay=30) + client.on_connect = on_connect + client.on_connect_fail = on_connect_fail + client.on_disconnect = on_disconnect + client.on_message = on_message + + for target, name in ((command_worker, "commands"), (telegram_worker, "telegram"), + (telemetry_loop, "telemetry"), (watchdog_loop, "watchdog")): + threading.Thread(target=target, name=name, daemon=True).start() + + signal.signal(signal.SIGTERM, shutdown_handler) + signal.signal(signal.SIGINT, shutdown_handler) + + logger.info(f"Node agent v{AGENT_VERSION} avviato ({CLIENT_ID.upper()}). Connessione al broker...") + # connect_async + loop_start: paho gestisce da solo primo collegamento e riconnessioni + client.connect_async(MQTT_CFG['broker'], int(MQTT_CFG['port']), int(MQTT_CFG.get('keepalive', 15))) + client.loop_start() + + while True: + time.sleep(3600) + + +if __name__ == "__main__": + start_service()