#!/usr/bin/env python3 import errno import json import logging import os import shutil import socket import ssl import sys import threading import time import urllib.error import urllib.request from datetime import datetime, timedelta from html import escape from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer # Setup logging logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) VERSION = "0.0.1" WORKDIR = os.path.dirname(os.path.abspath(__file__)) CONFIG_FILE = os.path.join(WORKDIR, 'config.json') DEFAULT_DATA_ROOT = '/opt/limristem-monitor' HISTORY_DIRECTORY_NAME = 'history' HOSTS_DIRECTORY_NAME = 'hosts' SITES_DIRECTORY_NAME = 'sites' MASTER_SITE_ID = 'master' MONTHLY_SUMMARY_FILENAME = 'summary.json' YEARLY_SUMMARY_FILENAME = 'summary.json' HISTORY_RETENTION_PREVIOUS_YEARS = 5 data_lock = threading.Lock() storage_lock = threading.Lock() SYSTEM_STATUS = {} SSL_CONTEXTS = {} SSL_CONTEXTS_LOCK = threading.Lock() HTTP_CLIENT_SSL_CONTEXT = ssl.create_default_context() HTTP_CLIENT_SSL_CONTEXT.check_hostname = False HTTP_CLIENT_SSL_CONTEXT.verify_mode = ssl.CERT_NONE HTTP_CHECK_DEFAULT_HEADERS = { 'User-Agent': ( 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 ' '(KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36' ), 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8', 'Accept-Language': 'it-IT,it;q=0.9,en-US;q=0.8,en;q=0.7', 'Cache-Control': 'no-cache', 'Pragma': 'no-cache', } TRANSIENT_NETWORK_ERRNOS = { errno.ECONNABORTED, errno.ECONNRESET, errno.ETIMEDOUT, errno.EHOSTUNREACH, errno.ENETUNREACH, errno.EPIPE, } HTTP_CHECK_TRANSIENT_ATTEMPTS = 2 HTTP_CHECK_TRANSIENT_DELAY_SECONDS = 0.25 CLIENT_DISCONNECT_EXCEPTIONS = ( BrokenPipeError, ConnectionResetError, TimeoutError, socket.timeout, ssl.SSLError, ) SERVER_CLIENT_TIMEOUT_SECONDS = 15 SERVER_REQUEST_QUEUE_SIZE = 128 SERVER_RESTART_DELAY_SECONDS = 3 SERVER_SSL_HANDSHAKE_TIMEOUT_SECONDS = 5 HTTPS_CERT_RELOAD_INTERVAL_SECONDS = 300 # Config caching _cached_config = None _last_config_load = 0 _config_cache_duration = 5 # seconds def load_config(): global _cached_config, _last_config_load now = time.time() if _cached_config and (now - _last_config_load < _config_cache_duration): return _cached_config try: with open(CONFIG_FILE, 'r', encoding='utf-8') as f: _cached_config = json.load(f) _last_config_load = now return _cached_config except Exception as e: logging.error(f"Errore lettura config: {e}") return _cached_config if _cached_config else None def _as_positive_int(value, default, minimum=1): try: parsed = int(value) except (TypeError, ValueError): return default return parsed if parsed >= minimum else default def _as_float(value): try: return float(value) except (TypeError, ValueError): return None def _round_float(value, digits=2): parsed = _as_float(value) if parsed is None: return None return round(parsed, digits) def _safe_slug(value): text = str(value or '').strip().lower() if not text: return 'unknown' cleaned = [] for char in text: if char.isalnum() or char in ('-', '_', '.'): cleaned.append(char) else: cleaned.append('_') slug = ''.join(cleaned).strip('._-') return slug or 'unknown' def _normalize_domain(value): return str(value or '').strip().lower() def _normalize_http_path(path): raw_path = str(path or '/').strip() if not raw_path: return '/' if raw_path.startswith('/'): return raw_path return f'/{raw_path}' def _normalize_http_headers(headers): if not isinstance(headers, dict): return {} normalized = {} for key, value in headers.items(): name = str(key or '').strip() if not name or ':' in name or '\r' in name or '\n' in name: continue text_value = str(value or '').strip() if not text_value or '\r' in text_value or '\n' in text_value: continue normalized[name] = text_value return normalized def _build_http_check_headers(custom_headers=None): headers = dict(HTTP_CHECK_DEFAULT_HEADERS) headers.update(_normalize_http_headers(custom_headers)) return headers def _unwrap_network_exception(exc): current = exc seen = set() while isinstance(current, urllib.error.URLError): reason = getattr(current, 'reason', None) if reason is None or id(current) in seen: break seen.add(id(current)) current = reason return current def _format_network_error(exc): root = _unwrap_network_exception(exc) message = str(root or exc).strip() return message or exc.__class__.__name__ def _is_retryable_http_exception(exc): root = _unwrap_network_exception(exc) if isinstance(root, (ConnectionResetError, TimeoutError, socket.timeout, ssl.SSLEOFError, ssl.SSLZeroReturnError)): return True if isinstance(root, OSError): return root.errno in TRANSIENT_NETWORK_ERRNOS return isinstance(exc, urllib.error.URLError) def _is_expected_client_disconnect(exc): root = _unwrap_network_exception(exc) if isinstance(root, CLIENT_DISCONNECT_EXCEPTIONS): return True if isinstance(root, OSError): return root.errno in TRANSIENT_NETWORK_ERRNOS return False def _format_client_endpoint(client_address): client_host = client_address[0] if isinstance(client_address, tuple) and client_address else 'sconosciuto' client_port = client_address[1] if isinstance(client_address, tuple) and len(client_address) > 1 else '?' return f'{client_host}:{client_port}' def _close_socket_quietly(sock): if not sock: return try: sock.shutdown(socket.SHUT_RDWR) except OSError: pass try: sock.close() except OSError: pass def _get_path_state_signature(path): normalized_path = str(path or '').strip() if not normalized_path: return ('missing', '') try: stats = os.stat(normalized_path) except OSError as e: return ('error', normalized_path, e.errno) return ( 'ok', normalized_path, stats.st_mtime_ns, stats.st_size, stats.st_ino, ) def _build_https_reload_signature(config): settings = config.get('settings', {}) signature = [ ( 'master', _get_path_state_signature(settings.get('master_ssl_cert')), _get_path_state_signature(settings.get('master_ssl_key')), ) ] for site in config.get('sites', []): if not isinstance(site, dict): continue domain = _normalize_domain(site.get('domain')) or 'site-senza-dominio' signature.append( ( domain, _get_path_state_signature(site.get('ssl_cert')), _get_path_state_signature(site.get('ssl_key')), ) ) return tuple(signature) def _compose_target(host_conf): target = str(host_conf.get('target', '')).strip() port = _as_positive_int(host_conf.get('port'), 0, minimum=1) return f"{target}:{port}" if target and port else target or 'N/D' def _get_data_root(config=None): env_override = str(os.environ.get('LIMRISTEM_MONITOR_DATA_DIR', '')).strip() if env_override: return env_override if isinstance(config, dict): configured = str(config.get('settings', {}).get('data_dir', '')).strip() if configured: return configured return DEFAULT_DATA_ROOT def get_history_root(config=None): return os.path.join(_get_data_root(config), HISTORY_DIRECTORY_NAME) def get_hosts_history_root(config=None): return os.path.join(get_history_root(config), HOSTS_DIRECTORY_NAME) def get_sites_summary_root(config=None): return os.path.join(get_history_root(config), SITES_DIRECTORY_NAME) def ensure_directory(path): os.makedirs(path, exist_ok=True) return path def ensure_storage_roots(config=None): ensure_directory(get_hosts_history_root(config)) ensure_directory(get_sites_summary_root(config)) def get_host_root(host_id, config=None): return os.path.join(get_hosts_history_root(config), _safe_slug(host_id)) def get_host_year_dir(host_id, year, config=None): return os.path.join(get_host_root(host_id, config), f"{int(year):04d}") def get_host_month_dir(host_id, year, month, config=None): return os.path.join(get_host_year_dir(host_id, year, config), f"{int(month):02d}") def get_host_daily_log_path(host_id, date_like, config=None): month_dir = get_host_month_dir(host_id, date_like.year, date_like.month, config) return os.path.join(month_dir, f"{date_like.year:04d}-{date_like.month:02d}-{date_like.day:02d}.jsonl") def get_host_month_summary_path(host_id, year, month, config=None): return os.path.join(get_host_month_dir(host_id, year, month, config), MONTHLY_SUMMARY_FILENAME) def get_host_year_summary_path(host_id, year, config=None): return os.path.join(get_host_year_dir(host_id, year, config), YEARLY_SUMMARY_FILENAME) def get_site_identifier(site=None): if not site: return MASTER_SITE_ID domain = _normalize_domain(site.get('domain')) if domain: return domain title = str(site.get('title', '')).strip() return title or MASTER_SITE_ID def get_site_summary_path(site_identifier, config=None): return os.path.join(get_sites_summary_root(config), f"{_safe_slug(site_identifier)}.json") def atomic_write_json(path, payload): ensure_directory(os.path.dirname(path)) temp_path = f"{path}.tmp-{os.getpid()}-{threading.get_ident()}" with storage_lock: try: with open(temp_path, 'w', encoding='utf-8') as handle: json.dump(payload, handle, ensure_ascii=False, indent=2) handle.write('\n') os.replace(temp_path, path) finally: if os.path.exists(temp_path): try: os.remove(temp_path) except OSError: pass def append_json_line(path, payload): ensure_directory(os.path.dirname(path)) line = json.dumps(payload, ensure_ascii=False) with storage_lock: with open(path, 'a', encoding='utf-8') as handle: handle.write(line) handle.write('\n') def read_json_file(path): if not os.path.exists(path): return None with open(path, 'r', encoding='utf-8') as handle: return json.load(handle) def iter_daily_log_entries(path): if not os.path.exists(path): return with open(path, 'r', encoding='utf-8') as handle: for line_number, line in enumerate(handle, start=1): line = line.strip() if not line: continue try: yield json.loads(line) except json.JSONDecodeError as e: logging.warning(f"Riga JSON non valida in {path}:{line_number}: {e}") def parse_timestamp(value): if not value: return None try: return datetime.fromisoformat(str(value)) except ValueError: return None def _pick_latest_timestamp(current_value, candidate_value): current_dt = parse_timestamp(current_value) candidate_dt = parse_timestamp(candidate_value) if candidate_dt and (current_dt is None or candidate_dt > current_dt): return candidate_value return current_value def _empty_hourly_timeline(now_dt=None): now_dt = now_dt or datetime.now() start_hour = now_dt.replace(minute=0, second=0, microsecond=0) - timedelta(hours=23) timeline = [] for offset in range(24): hour_dt = start_hour + timedelta(hours=offset) timeline.append({ 'hour': hour_dt.isoformat(), 'status': 'unknown' }) return timeline def _normalize_string_list(values): if isinstance(values, str): values = [values] if not isinstance(values, list): return [] result = [] seen = set() for item in values: value = str(item).strip() if not value: continue key = value.lower() if key in seen: continue seen.add(key) result.append(value) return result def get_site_label_groups(site): labels = site.get('labels', []) if not isinstance(labels, list): return [] groups = [] for idx, label in enumerate(labels, start=1): if not isinstance(label, dict): continue name = str(label.get('name', label.get('label', ''))).strip() or f"Etichetta {idx}" host_ids = _normalize_string_list(label.get('host_ids', label.get('visible_host_ids', []))) groups.append({'name': name, 'host_ids': host_ids}) return groups def get_site_visible_host_ids(site): label_groups = get_site_label_groups(site) if label_groups: visible_ids = [] for group in label_groups: for host_id in group['host_ids']: if host_id not in visible_ids: visible_ids.append(host_id) return visible_ids return _normalize_string_list(site.get('visible_host_ids', [])) def get_notification_recipients(config, host_id=None): recipients = [] seen = set() def add_email(addr): if not addr: return email = str(addr).strip() if not email: return key = email.lower() if key in seen: return seen.add(key) recipients.append(email) add_email(config.get('settings', {}).get('admin_email')) if host_id: for site in config.get('sites', []): visible_ids = get_site_visible_host_ids(site) if host_id not in visible_ids: continue site_emails = site.get('email', []) if isinstance(site_emails, str): site_emails = [site_emails] if isinstance(site_emails, list): for site_email in site_emails: add_email(site_email) return recipients def _allowed_alert_type(value): alert_type = str(value or 'info').strip().lower() if alert_type in ('warning', 'info', 'error', 'success'): return alert_type return 'info' def _host_name(host_conf): h_id = str(host_conf.get('id', '')).strip() return str(host_conf.get('name', h_id or 'Unknown')).strip() or 'Unknown' def create_default_status_entry(host_conf, now_dt=None): now_dt = now_dt or datetime.now() h_id = str(host_conf.get('id', '')).strip() return { 'id': h_id, 'status': 'PENDING', 'fails': 0, 'msg': 'Inizializzazione...', 'name': _host_name(host_conf), 'target': _compose_target(host_conf), 'last_check': 'Attendi...', 'last_check_at': None, 'status_since': now_dt.isoformat(), 'ping_ms': None, 'uptime_percentage': None, 'uptime_percentage_24h': None, 'average_latency_ms': None, 'timeline_24h': _empty_hourly_timeline(now_dt) } def sync_status_entries(config): hosts = config.get('hosts', []) if not isinstance(hosts, list): return [] current_ids = [] now_dt = datetime.now() with data_lock: for host_conf in hosts: if not isinstance(host_conf, dict): continue h_id = str(host_conf.get('id', '')).strip() if not h_id: continue current_ids.append(h_id) target = _compose_target(host_conf) if h_id not in SYSTEM_STATUS: SYSTEM_STATUS[h_id] = create_default_status_entry(host_conf, now_dt) else: SYSTEM_STATUS[h_id]['id'] = h_id SYSTEM_STATUS[h_id]['name'] = _host_name(host_conf) SYSTEM_STATUS[h_id]['target'] = target keys_to_remove = [key for key in SYSTEM_STATUS if key not in current_ids] for key in keys_to_remove: del SYSTEM_STATUS[key] return current_ids def _empty_monthly_summary(host_id, host_name, year, month): return { 'generated_at': datetime.now().isoformat(), 'host_id': host_id, 'host_name': host_name, 'year': int(year), 'month': int(month), 'period': f"{int(year):04d}-{int(month):02d}", 'daily_log_files': 0, 'days_with_data': 0, 'source_files': [], 'total_checks': 0, 'up_checks': 0, 'down_checks': 0, 'uptime_percentage': None, 'average_latency_ms': None, 'min_latency_ms': None, 'max_latency_ms': None, 'latency_samples': 0, 'online_events': 0, 'offline_events': 0, 'last_online_at': None, 'last_offline_at': None, } def _empty_yearly_summary(host_id, host_name, year): return { 'generated_at': datetime.now().isoformat(), 'host_id': host_id, 'host_name': host_name, 'year': int(year), 'months': [], 'months_with_data': 0, 'total_checks': 0, 'up_checks': 0, 'down_checks': 0, 'uptime_percentage': None, 'average_latency_ms': None, 'min_latency_ms': None, 'max_latency_ms': None, 'latency_samples': 0, 'online_events': 0, 'offline_events': 0, 'last_online_at': None, 'last_offline_at': None, } def list_month_daily_log_paths(host_id, year, month, config=None): month_dir = get_host_month_dir(host_id, year, month, config) if not os.path.isdir(month_dir): return [] return sorted( os.path.join(month_dir, entry) for entry in os.listdir(month_dir) if entry.endswith('.jsonl') ) def build_monthly_summary(host_id, host_name, year, month, config=None): summary = _empty_monthly_summary(host_id, host_name, year, month) daily_paths = list_month_daily_log_paths(host_id, year, month, config) days_with_data = set() latency_sum = 0.0 for path in daily_paths: file_has_entries = False for entry in iter_daily_log_entries(path): file_has_entries = True status = str(entry.get('status', '')).upper() timestamp = entry.get('timestamp') event = entry.get('event') latency_ms = _as_float(entry.get('latency_ms')) summary['total_checks'] += 1 if status == 'UP': summary['up_checks'] += 1 if latency_ms is not None: latency_sum += latency_ms summary['latency_samples'] += 1 summary['min_latency_ms'] = latency_ms if summary['min_latency_ms'] is None else min(summary['min_latency_ms'], latency_ms) summary['max_latency_ms'] = latency_ms if summary['max_latency_ms'] is None else max(summary['max_latency_ms'], latency_ms) elif status == 'DOWN': summary['down_checks'] += 1 if event == 'ONLINE': summary['online_events'] += 1 summary['last_online_at'] = _pick_latest_timestamp(summary['last_online_at'], timestamp) elif event == 'OFFLINE': summary['offline_events'] += 1 summary['last_offline_at'] = _pick_latest_timestamp(summary['last_offline_at'], timestamp) if file_has_entries: days_with_data.add(os.path.basename(path)) summary['source_files'] = [os.path.basename(path) for path in daily_paths] summary['daily_log_files'] = len(daily_paths) summary['days_with_data'] = len(days_with_data) if summary['total_checks']: summary['uptime_percentage'] = round((summary['up_checks'] / summary['total_checks']) * 100, 2) if summary['latency_samples']: summary['average_latency_ms'] = round(latency_sum / summary['latency_samples'], 2) summary['min_latency_ms'] = round(summary['min_latency_ms'], 2) summary['max_latency_ms'] = round(summary['max_latency_ms'], 2) summary_path = get_host_month_summary_path(host_id, year, month, config) atomic_write_json(summary_path, summary) return summary def build_yearly_summary(host_id, host_name, year, config=None): summary = _empty_yearly_summary(host_id, host_name, year) latency_weighted_sum = 0.0 for month in range(1, 13): month_path = get_host_month_summary_path(host_id, year, month, config) month_summary = read_json_file(month_path) if not month_summary: continue summary['months'].append({ 'month': month_summary.get('month', month), 'period': month_summary.get('period', f"{int(year):04d}-{month:02d}"), 'total_checks': month_summary.get('total_checks', 0), 'up_checks': month_summary.get('up_checks', 0), 'down_checks': month_summary.get('down_checks', 0), 'uptime_percentage': month_summary.get('uptime_percentage'), 'average_latency_ms': month_summary.get('average_latency_ms'), 'latency_samples': month_summary.get('latency_samples', 0), 'online_events': month_summary.get('online_events', 0), 'offline_events': month_summary.get('offline_events', 0), }) summary['total_checks'] += month_summary.get('total_checks', 0) summary['up_checks'] += month_summary.get('up_checks', 0) summary['down_checks'] += month_summary.get('down_checks', 0) summary['online_events'] += month_summary.get('online_events', 0) summary['offline_events'] += month_summary.get('offline_events', 0) summary['last_online_at'] = _pick_latest_timestamp(summary['last_online_at'], month_summary.get('last_online_at')) summary['last_offline_at'] = _pick_latest_timestamp(summary['last_offline_at'], month_summary.get('last_offline_at')) month_latency_samples = month_summary.get('latency_samples', 0) month_average_latency = _as_float(month_summary.get('average_latency_ms')) month_min_latency = _as_float(month_summary.get('min_latency_ms')) month_max_latency = _as_float(month_summary.get('max_latency_ms')) if month_latency_samples and month_average_latency is not None: latency_weighted_sum += month_average_latency * month_latency_samples summary['latency_samples'] += month_latency_samples if month_min_latency is not None: summary['min_latency_ms'] = month_min_latency if summary['min_latency_ms'] is None else min(summary['min_latency_ms'], month_min_latency) if month_max_latency is not None: summary['max_latency_ms'] = month_max_latency if summary['max_latency_ms'] is None else max(summary['max_latency_ms'], month_max_latency) summary['months_with_data'] = len(summary['months']) if summary['total_checks']: summary['uptime_percentage'] = round((summary['up_checks'] / summary['total_checks']) * 100, 2) if summary['latency_samples']: summary['average_latency_ms'] = round(latency_weighted_sum / summary['latency_samples'], 2) summary['min_latency_ms'] = round(summary['min_latency_ms'], 2) summary['max_latency_ms'] = round(summary['max_latency_ms'], 2) summary_path = get_host_year_summary_path(host_id, year, config) atomic_write_json(summary_path, summary) return summary def build_hourly_uptime_timeline(host_id, now_dt=None, config=None): now_dt = now_dt or datetime.now() start_hour = now_dt.replace(minute=0, second=0, microsecond=0) - timedelta(hours=23) day_cursor = start_hour.date() end_day = now_dt.date() buckets = {} total_checks = 0 total_up_checks = 0 while day_cursor <= end_day: hour_dt = datetime(day_cursor.year, day_cursor.month, day_cursor.day) day_path = get_host_daily_log_path(host_id, day_cursor, config) if os.path.exists(day_path): for entry in iter_daily_log_entries(day_path): timestamp_dt = parse_timestamp(entry.get('timestamp')) if not timestamp_dt or timestamp_dt < start_hour or timestamp_dt > now_dt: continue bucket_key = timestamp_dt.replace(minute=0, second=0, microsecond=0).isoformat() bucket = buckets.setdefault(bucket_key, {'checks': 0, 'up_checks': 0, 'has_down': False}) bucket['checks'] += 1 total_checks += 1 if str(entry.get('status', '')).upper() == 'UP': bucket['up_checks'] += 1 total_up_checks += 1 else: bucket['has_down'] = True day_cursor += timedelta(days=1) timeline = [] for offset in range(24): hour_dt = start_hour + timedelta(hours=offset) hour_key = hour_dt.isoformat() bucket = buckets.get(hour_key) status = 'unknown' if bucket: if bucket['has_down'] or bucket['up_checks'] != bucket['checks']: status = 'down' else: status = 'up' timeline.append({'hour': hour_key, 'status': status}) uptime_24h = None if total_checks: uptime_24h = round((total_up_checks / total_checks) * 100, 2) return timeline, uptime_24h def cleanup_host_history(host_id, current_year, config=None): host_root = get_host_root(host_id, config) if not os.path.isdir(host_root): return for entry in os.listdir(host_root): year_path = os.path.join(host_root, entry) if not os.path.isdir(year_path): continue try: year = int(entry) except ValueError: continue if year < current_year - HISTORY_RETENTION_PREVIOUS_YEARS: shutil.rmtree(year_path, ignore_errors=False) logging.info(f"🧹 Storico host {host_id}: rimosso anno fuori retention {year}") continue if year >= current_year: continue for child in os.listdir(year_path): child_path = os.path.join(year_path, child) if child == YEARLY_SUMMARY_FILENAME and os.path.isfile(child_path): continue if os.path.isdir(child_path): shutil.rmtree(child_path, ignore_errors=False) else: os.remove(child_path) def record_host_check(config, host_conf, status_snapshot, checked_at, status_changed): ensure_storage_roots(config) host_id = status_snapshot['id'] host_name = status_snapshot['name'] daily_log_path = get_host_daily_log_path(host_id, checked_at, config) event = None if status_changed: event = 'ONLINE' if status_snapshot['status'] == 'UP' else 'OFFLINE' payload = { 'timestamp': checked_at.isoformat(), 'host_id': host_id, 'host_name': host_name, 'status': status_snapshot['status'], 'message': status_snapshot.get('msg', ''), 'latency_ms': _round_float(status_snapshot.get('ping_ms')), 'event': event, 'status_changed': bool(status_changed), 'status_since': status_snapshot.get('status_since'), 'fails': status_snapshot.get('fails', 0), 'check_type': str(host_conf.get('type', 'tcp')).strip().lower(), 'target': str(host_conf.get('target', '')).strip(), 'port': _as_positive_int(host_conf.get('port'), 0, minimum=1), 'path': _normalize_http_path(host_conf.get('path', '/')), } append_json_line(daily_log_path, payload) monthly_summary = build_monthly_summary(host_id, host_name, checked_at.year, checked_at.month, config) yearly_summary = build_yearly_summary(host_id, host_name, checked_at.year, config) timeline_24h, uptime_percentage_24h = build_hourly_uptime_timeline(host_id, checked_at, config) cleanup_host_history(host_id, checked_at.year, config) return { 'uptime_percentage': monthly_summary.get('uptime_percentage'), 'uptime_percentage_24h': uptime_percentage_24h, 'average_latency_ms': monthly_summary.get('average_latency_ms'), 'timeline_24h': timeline_24h, 'last_online_at': monthly_summary.get('last_online_at'), 'last_offline_at': monthly_summary.get('last_offline_at'), 'current_month_summary': monthly_summary, 'current_year_summary': yearly_summary, } def build_host_site_payload(host_conf, status_data, now_dt=None): now_dt = now_dt or datetime.now() merged = create_default_status_entry(host_conf, now_dt) if isinstance(status_data, dict): merged.update(status_data) timeline = merged.get('timeline_24h') if not isinstance(timeline, list) or len(timeline) != 24: timeline = _empty_hourly_timeline(now_dt) return { 'id': merged.get('id'), 'name': merged.get('name'), 'status': merged.get('status', 'PENDING'), 'message': merged.get('msg', ''), 'target': merged.get('target', _compose_target(host_conf)), 'last_check': merged.get('last_check'), 'last_check_at': merged.get('last_check_at'), 'status_since': merged.get('status_since'), 'ping_ms': merged.get('ping_ms'), 'fails': merged.get('fails', 0), 'uptime_percentage': merged.get('uptime_percentage'), 'uptime_percentage_24h': merged.get('uptime_percentage_24h'), 'average_latency_ms': merged.get('average_latency_ms'), 'timeline_24h': timeline, } def build_site_summary_payload(config, site=None): now_dt = datetime.now() hosts = config.get('hosts', []) if not isinstance(hosts, list): hosts = [] manual_alerts = config.get('manual_alerts', []) label_groups = [] ordered_host_ids = [] domain = None title = 'Limristem Monitor Master Console' is_filtered = False if site: domain = _normalize_domain(site.get('domain')) title = str(site.get('title', 'System Status')).strip() or 'System Status' label_groups = get_site_label_groups(site) ordered_host_ids = get_site_visible_host_ids(site) manual_alerts = site.get('manual_alerts', []) is_filtered = True else: ordered_host_ids = [str(host.get('id', '')).strip() for host in hosts if isinstance(host, dict) and str(host.get('id', '')).strip()] if not isinstance(manual_alerts, list): manual_alerts = [] with data_lock: status_snapshot = {host_id: dict(data) for host_id, data in SYSTEM_STATUS.items()} payload_hosts = {} for host_conf in hosts: if not isinstance(host_conf, dict): continue host_id = str(host_conf.get('id', '')).strip() if not host_id: continue if is_filtered and host_id not in ordered_host_ids: continue payload_hosts[host_id] = build_host_site_payload(host_conf, status_snapshot.get(host_id), now_dt) return { 'generated_at': now_dt.isoformat(), 'site': { 'id': get_site_identifier(site), 'domain': domain, 'title': title, 'is_filtered': is_filtered, }, 'label_groups': label_groups, 'ordered_host_ids': ordered_host_ids, 'manual_alerts': manual_alerts, 'hosts': payload_hosts, } def update_site_summaries(config): ensure_storage_roots(config) master_payload = build_site_summary_payload(config) atomic_write_json(get_site_summary_path(MASTER_SITE_ID, config), master_payload) for site in config.get('sites', []): if not isinstance(site, dict): continue payload = build_site_summary_payload(config, site) atomic_write_json(get_site_summary_path(get_site_identifier(site), config), payload) def get_current_site(config, host_header): normalized_host = _normalize_domain(host_header) for site in config.get('sites', []): if not isinstance(site, dict): continue if _normalize_domain(site.get('domain')) == normalized_host: return site return None def load_site_summary_payload(config, site=None): summary_path = get_site_summary_path(get_site_identifier(site), config) try: summary = read_json_file(summary_path) except (OSError, ValueError, json.JSONDecodeError) as e: logging.error(f"Errore lettura site summary {summary_path}: {e}") summary = None if summary: return summary return build_site_summary_payload(config, site) class HardenedThreadingHTTPServer(ThreadingHTTPServer): daemon_threads = True allow_reuse_address = True request_queue_size = SERVER_REQUEST_QUEUE_SIZE def get_request(self): request, client_address = super().get_request() request.settimeout(SERVER_CLIENT_TIMEOUT_SECONDS) return request, client_address def handle_error(self, request, client_address): _, exc, _ = sys.exc_info() if exc and _is_expected_client_disconnect(exc): logging.debug( f"Connessione client interrotta da {_format_client_endpoint(client_address)}: " f"{_format_network_error(exc)}" ) return super().handle_error(request, client_address) class HardenedThreadingHTTPSServer(HardenedThreadingHTTPServer): def __init__(self, server_address, request_handler_class, ssl_context): self.ssl_context = ssl_context super().__init__(server_address, request_handler_class) def get_request(self): request, client_address = ThreadingHTTPServer.get_request(self) request.settimeout(SERVER_SSL_HANDSHAKE_TIMEOUT_SECONDS) return request, client_address def process_request_thread(self, request, client_address): secure_request = None try: secure_request = self.ssl_context.wrap_socket( request, server_side=True, do_handshake_on_connect=False, ) secure_request.settimeout(SERVER_SSL_HANDSHAKE_TIMEOUT_SECONDS) secure_request.do_handshake() secure_request.settimeout(SERVER_CLIENT_TIMEOUT_SECONDS) super().process_request_thread(secure_request, client_address) except Exception as e: self._log_handshake_error(e, client_address) _close_socket_quietly(secure_request or request) def _log_handshake_error(self, exc, client_address): endpoint = _format_client_endpoint(client_address) message = _format_network_error(exc) if _is_expected_client_disconnect(exc): logging.debug(f"Handshake TLS interrotto da {endpoint}: {message}") return root = _unwrap_network_exception(exc) if isinstance(root, ssl.SSLError): logging.warning(f"⚠️ Handshake TLS fallito da {endpoint}: {message}") return logging.warning(f"⚠️ Errore HTTPS da {endpoint}: {message}") # --- CORE MONITORING --- def check_tcp(host, port, timeout): started = time.perf_counter() try: with socket.create_connection((host, port), timeout=timeout): latency_ms = (time.perf_counter() - started) * 1000 return True, 'OK', latency_ms except Exception as e: latency_ms = (time.perf_counter() - started) * 1000 return False, str(e), latency_ms def check_http(host, port, path, timeout, protocol_override=None, headers=None): protocol = protocol_override if protocol_override in ('http', 'https') else ('https' if port == 443 else 'http') url = f"{protocol}://{host}:{port}{_normalize_http_path(path)}" request_headers = _build_http_check_headers(headers) started = time.perf_counter() for attempt in range(HTTP_CHECK_TRANSIENT_ATTEMPTS): try: req = urllib.request.Request(url, headers=request_headers) open_kwargs = {'timeout': timeout} if protocol == 'https': open_kwargs['context'] = HTTP_CLIENT_SSL_CONTEXT with urllib.request.urlopen(req, **open_kwargs) as response: code = response.getcode() latency_ms = (time.perf_counter() - started) * 1000 if 200 <= code < 400: return True, f"HTTP {code}", latency_ms return False, f"HTTP {code}", latency_ms except urllib.error.HTTPError as e: latency_ms = (time.perf_counter() - started) * 1000 return False, f"HTTP {e.code}", latency_ms except Exception as e: latency_ms = (time.perf_counter() - started) * 1000 if attempt < HTTP_CHECK_TRANSIENT_ATTEMPTS - 1 and _is_retryable_http_exception(e): time.sleep(HTTP_CHECK_TRANSIENT_DELAY_SECONDS * (attempt + 1)) continue return False, _format_network_error(e), latency_ms def check_udp(host, port, timeout): # Nota: UDP è connectionless. Questo check verifica solo se riusciamo a inviare il pacchetto. # Non conferma che il servizio dall'altra parte sia attivo o in ascolto. started = time.perf_counter() try: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock: sock.settimeout(timeout) sock.sendto(b'Ping', (host, port)) latency_ms = (time.perf_counter() - started) * 1000 return True, 'UDP Packet Sent', latency_ms except Exception as e: latency_ms = (time.perf_counter() - started) * 1000 return False, str(e), latency_ms def send_notification(service_name, status, details, config, host_id=None): msg_text = f"ALERT: {service_name} is {status}. Info: {details}" logging.info(f"NOTIFICA: {msg_text}") alerts_config = config.get('alerts', {}) # 1. SMTP Email smtp_conf = alerts_config.get('smtp', {}) if smtp_conf.get('enabled'): recipients = get_notification_recipients(config, host_id) if recipients: try: import smtplib from email.mime.text import MIMEText msg = MIMEText(msg_text) msg['Subject'] = f"Limristem Monitor Alert: {service_name} is {status}" msg['From'] = smtp_conf.get('from_email', 'noreply@limristem-monitor.local') msg['To'] = ', '.join(recipients) s_host = smtp_conf.get('host', 'localhost') s_port = smtp_conf.get('port', 25) s_user = smtp_conf.get('username') s_pass = smtp_conf.get('password') with smtplib.SMTP(s_host, s_port) as server: if smtp_conf.get('tls', False): server.starttls() if s_user and s_pass: server.login(s_user, s_pass) server.send_message(msg, to_addrs=recipients) logging.info(f"📧 Email inviata a {', '.join(recipients)}") except Exception as e: logging.error(f"❌ Errore invio email: {e}") else: logging.warning('⚠️ SMTP attivo ma nessun destinatario email disponibile.') # 2. Webhook webhook_url = alerts_config.get('webhook_url') if webhook_url: try: payload = json.dumps({ 'service': service_name, 'status': status, 'details': details, 'time': datetime.now().isoformat() }).encode('utf-8') req = urllib.request.Request(webhook_url, data=payload, headers={ 'Content-Type': 'application/json', 'User-Agent': 'Limristem Monitor/1.0' }) with urllib.request.urlopen(req, timeout=5) as response: logging.info(f"🔗 Webhook inviato: HTTP {response.getcode()}") except Exception as e: logging.error(f"❌ Errore invio webhook: {e}") def monitoring_loop(): logging.info('Monitor loop avviato.') while True: config = load_config() if not config: time.sleep(10) continue hosts = config.get('hosts', []) if not isinstance(hosts, list): logging.error('Configurazione hosts non valida: atteso elenco.') time.sleep(10) continue sync_status_entries(config) for host_conf in hosts: if not isinstance(host_conf, dict): logging.warning('Host ignorato: voce hosts non valida (atteso oggetto).') continue h_id = str(host_conf.get('id', '')).strip() h_type = str(host_conf.get('type', 'tcp')).strip().lower() h_name = _host_name(host_conf) h_target = str(host_conf.get('target', '')).strip() h_port = _as_positive_int(host_conf.get('port'), 0, minimum=1) h_path = _normalize_http_path(host_conf.get('path', '/')) h_http_headers = _normalize_http_headers(host_conf.get('http_headers')) timeout = _as_positive_int(host_conf.get('timeout', 5), 5, minimum=1) retries = _as_positive_int(host_conf.get('retries', 1), 1, minimum=1) if not h_id or not h_target or not h_port: logging.warning(f"Host ignorato per config incompleta: id={h_id!r}, target={h_target!r}, port={h_port!r}") continue with data_lock: entry = SYSTEM_STATUS.get(h_id) if entry is None: entry = create_default_status_entry(host_conf) SYSTEM_STATUS[h_id] = entry prev_status = entry.get('status', 'PENDING') prev_status_since = entry.get('status_since') fails = entry.get('fails', 0) is_up = False msg = '' latency_ms = None for attempt in range(retries): try: if h_type == 'http': is_up, msg, latency_ms = check_http(h_target, h_port, h_path, timeout, headers=h_http_headers) elif h_type == 'https': is_up, msg, latency_ms = check_http(h_target, h_port, h_path, timeout, 'https', headers=h_http_headers) elif h_type == 'udp': is_up, msg, latency_ms = check_udp(h_target, h_port, timeout) else: is_up, msg, latency_ms = check_tcp(h_target, h_port, timeout) except Exception as e: is_up = False msg = f"Errore interno check: {_format_network_error(e)}" latency_ms = None logging.exception(f"❌ Errore inatteso durante il check host {h_id}: {e}") if is_up: break if attempt < retries - 1: time.sleep(1) checked_at = datetime.now() new_status = 'UP' if is_up else 'DOWN' fails = 0 if new_status == 'UP' else fails + 1 status_changed = prev_status != new_status status_since = checked_at.isoformat() if status_changed or not prev_status_since else prev_status_since ping_ms = _round_float(latency_ms) with data_lock: SYSTEM_STATUS[h_id].update({ 'id': h_id, 'name': h_name, 'target': f"{h_target}:{h_port}", 'status': new_status, 'msg': msg, 'last_check': checked_at.strftime('%H:%M:%S'), 'last_check_at': checked_at.isoformat(), 'status_since': status_since, 'fails': fails, 'ping_ms': ping_ms, }) status_snapshot = dict(SYSTEM_STATUS[h_id]) if status_changed: send_notification(h_name, new_status, msg, config, h_id) try: history_updates = record_host_check(config, host_conf, status_snapshot, checked_at, status_changed) except (OSError, ValueError, json.JSONDecodeError) as e: logging.error(f"❌ Errore aggiornamento storico host {h_id}: {e}") else: with data_lock: SYSTEM_STATUS[h_id].update(history_updates) try: update_site_summaries(config) except (OSError, ValueError, json.JSONDecodeError) as e: logging.error(f"❌ Errore aggiornamento site summaries: {e}") check_interval = _as_positive_int(config.get('settings', {}).get('check_interval_seconds', 60), 60, minimum=1) time.sleep(check_interval) def format_percentage(value): parsed = _as_float(value) if parsed is None: return 'N/D' return f"{parsed:.2f}%" def render_manual_alerts(alerts): blocks = [] for alert in alerts: if not isinstance(alert, dict): continue alert_type = _allowed_alert_type(alert.get('type', 'info')) title = escape(str(alert.get('title', '')).strip()) message = escape(str(alert.get('message', '')).strip()) title_html = f"{title} " if title else '' blocks.append(f"
{title_html}{message}
") return ''.join(blocks) def render_uptime_timeline(timeline): if not isinstance(timeline, list) or not timeline: timeline = _empty_hourly_timeline() cells = [] for item in timeline: hour_label = escape(str(item.get('hour', ''))) status = str(item.get('status', 'unknown')).lower() if status not in ('up', 'down', 'unknown'): status = 'unknown' title = { 'up': 'Attivo per tutta l’ora', 'down': 'Rilevato almeno un downtime nell’ora', 'unknown': 'Nessun dato disponibile per l’ora' }[status] cells.append( f"" ) return ( "
" "
Uptime ultime 24 ore
" f"
{''.join(cells)}
" "
" ) def render_status_card(host_payload): host_id = str(host_payload.get('id', '')).strip() or 'host' name = escape(str(host_payload.get('name', host_id))) status = str(host_payload.get('status', 'PENDING')).upper() status_class = status.lower() if status.lower() in ('up', 'down', 'pending') else 'pending' message = escape(str(host_payload.get('message', '')).strip()) uptime_percentage = format_percentage(host_payload.get('uptime_percentage')) timeline_html = render_uptime_timeline(host_payload.get('timeline_24h', [])) down_message_html = f"{message}" if status == 'DOWN' and message else '' return f"""
{name} {timeline_html} {down_message_html}
{escape(status)}
Uptime mese {uptime_percentage}
""" def render_site_html(summary): site_info = summary.get('site', {}) if isinstance(summary, dict) else {} title = escape(str(site_info.get('title', 'System Status')).strip() or 'System Status') manual_alerts = summary.get('manual_alerts', []) if isinstance(summary, dict) else [] label_groups = summary.get('label_groups', []) if isinstance(summary, dict) else [] ordered_host_ids = summary.get('ordered_host_ids', []) if isinstance(summary, dict) else [] host_map = summary.get('hosts', {}) if isinstance(summary, dict) else {} html = f"""{title}

{title}

{render_manual_alerts(manual_alerts)} """ rendered_any = False if label_groups: for group in label_groups: group_name = escape(str(group.get('name', 'Etichetta')).strip() or 'Etichetta') html += f"

{group_name}

" rendered_group = False for host_id in group.get('host_ids', []): host_payload = host_map.get(host_id) if not host_payload: continue rendered_group = True rendered_any = True html += render_status_card(host_payload) if not rendered_group: html += """
Nessun host assegnato a questa etichetta.
N/A
""" else: for host_id in ordered_host_ids: host_payload = host_map.get(host_id) if not host_payload: continue rendered_any = True html += render_status_card(host_payload) if not rendered_any: html += """
Nessun host disponibile per questo sito.
N/A
""" html += "
Powered by Limristem Monitor
" return html # --- REQUEST HANDLER CONDIVISO --- class UniversalRequestHandler(BaseHTTPRequestHandler): def log_message(self, format, *args): return def _send_bytes_response(self, status_code, body=b'', content_type=None): try: self.send_response(status_code) if content_type: self.send_header('Content-type', content_type) self.send_header('Content-Length', str(len(body))) self.end_headers() if body: self.wfile.write(body) return True except CLIENT_DISCONNECT_EXCEPTIONS as e: logging.debug(f"Connessione client chiusa durante la risposta: {_format_network_error(e)}") return False def do_GET(self): if self.path == '/health': self._send_bytes_response(200, b'OK', 'text/plain; charset=utf-8') return if self.path != '/': self._send_bytes_response(404) return config = load_config() if not config: self._send_bytes_response(500, b'Error loading configuration', 'text/plain; charset=utf-8') return host_header = self.headers.get('Host', '').split(':')[0] current_site = get_current_site(config, host_header) summary = load_site_summary_payload(config, current_site) html = render_site_html(summary) self._send_bytes_response(200, html.encode('utf-8'), 'text/html; charset=utf-8') # --- SNI CALLBACK (SSL MULTIPLO) --- def sni_callback(sock, req_hostname, original_context): if not req_hostname: return with SSL_CONTEXTS_LOCK: site_context = SSL_CONTEXTS.get(req_hostname.lower()) if site_context: sock.context = site_context def get_ssl_access_issues(certfile, keyfile, label): issues = [] files_to_check = ( ('certificato', str(certfile or '').strip()), ('chiave', str(keyfile or '').strip()), ) for file_label, path in files_to_check: if not path: issues.append(f"{label}: {file_label} non configurato") continue if not os.path.exists(path): issues.append(f"{label}: file non trovato: {path}") continue if not os.path.isfile(path): issues.append(f"{label}: percorso non valido: {path}") continue if not os.access(path, os.R_OK): issues.append( f"{label}: permessi insufficienti su {path}. " "Assicurati che l'utente del servizio possa leggere il file." ) return issues def load_https_contexts(config): settings = config.get('settings', {}) master_cert = settings.get('master_ssl_cert') master_key = settings.get('master_ssl_key') if not master_cert or not master_key: logging.error('❌ ERRORE: Certificati Master mancanti! HTTPS non partirà.') return None access_issues = get_ssl_access_issues(master_cert, master_key, 'Certificato Master') if access_issues: logging.error('❌ ' + '; '.join(access_issues)) return None try: default_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) default_ctx.load_cert_chain(certfile=master_cert, keyfile=master_key) default_ctx.sni_callback = sni_callback except Exception as e: logging.error(f"❌ Errore certificato Master: {e}") return None site_contexts = {} for site in config.get('sites', []): if not isinstance(site, dict): continue domain = _normalize_domain(site.get('domain')) certfile = site.get('ssl_cert') keyfile = site.get('ssl_key') if domain and certfile and keyfile: access_issues = get_ssl_access_issues(certfile, keyfile, f"SSL sito {domain}") if access_issues: logging.warning('⚠️ ' + '; '.join(access_issues)) continue try: ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ctx.load_cert_chain(certfile=certfile, keyfile=keyfile) site_contexts[domain] = ctx logging.info(f"🔐 SSL caricato per: {domain}") except Exception as e: logging.warning(f"⚠️ Errore SSL sito {domain}: {e}") return default_ctx, site_contexts # --- SERVER RUNNERS --- def run_http_server(port): while True: server = None try: server = HardenedThreadingHTTPServer(('0.0.0.0', port), UniversalRequestHandler) logging.info(f"✅ HTTP attivo su porta {port}") server.serve_forever(poll_interval=0.5) except Exception as e: logging.exception(f"❌ HTTP server arrestato in modo anomalo: {e}. Riavvio in corso...") finally: if server: try: server.server_close() except Exception as e: logging.warning(f"⚠️ Errore chiusura socket HTTP: {e}") time.sleep(SERVER_RESTART_DELAY_SECONDS) def run_https_server(port): while True: server = None try: config = load_config() if not config: logging.error('Configurazione non disponibile: impossibile avviare HTTPS.') time.sleep(SERVER_RESTART_DELAY_SECONDS) continue contexts = load_https_contexts(config) if not contexts: time.sleep(SERVER_RESTART_DELAY_SECONDS) continue default_ctx, site_contexts = contexts with SSL_CONTEXTS_LOCK: SSL_CONTEXTS.clear() SSL_CONTEXTS.update(site_contexts) current_ssl_signature = _build_https_reload_signature(config) server = HardenedThreadingHTTPSServer(('0.0.0.0', port), UniversalRequestHandler, default_ctx) server.timeout = 1 logging.info(f"✅ HTTPS attivo su porta {port}") last_reload = time.time() while True: server.handle_request() if time.time() - last_reload < HTTPS_CERT_RELOAD_INTERVAL_SECONDS: continue refreshed_config = load_config() if not refreshed_config: logging.warning('⚠️ Reload SSL saltato: configurazione non disponibile.') last_reload = time.time() continue refreshed_contexts = load_https_contexts(refreshed_config) if not refreshed_contexts: logging.warning('⚠️ Reload SSL fallito: mantengo i certificati correnti.') last_reload = time.time() continue refreshed_signature = _build_https_reload_signature(refreshed_config) if refreshed_signature == current_ssl_signature: last_reload = time.time() continue refreshed_default_ctx, refreshed_site_contexts = refreshed_contexts with SSL_CONTEXTS_LOCK: SSL_CONTEXTS.clear() SSL_CONTEXTS.update(refreshed_site_contexts) server.ssl_context = refreshed_default_ctx current_ssl_signature = refreshed_signature logging.info('🔄 Certificati HTTPS ricaricati senza riavvio listener.') last_reload = time.time() except Exception as e: logging.exception(f"❌ HTTPS server arrestato in modo anomalo: {e}. Riavvio in corso...") finally: if server: try: server.server_close() except Exception as e: logging.warning(f"⚠️ Errore chiusura socket HTTPS: {e}") time.sleep(SERVER_RESTART_DELAY_SECONDS) def check_for_updates(): while True: try: config = load_config() or {} req = urllib.request.Request("https://get.limristem.eu/monitor/version.json", headers={'User-Agent': f'LimristemMonitor/{VERSION}'}) with urllib.request.urlopen(req, timeout=10) as response: data = json.loads(response.read().decode('utf-8')) remote_version = data.get('version') if remote_version and remote_version != VERSION: logging.info(f"Nuova versione disponibile: {remote_version} (attuale: {VERSION})") alerts_config = config.get('alerts', {}) smtp_conf = alerts_config.get('smtp', {}) if smtp_conf.get('enabled'): recipients = get_notification_recipients(config) if recipients: import smtplib from email.mime.text import MIMEText msg_text = f"È disponibile una nuova versione di Limristem Monitor: {remote_version}.\nScaricala da {data.get('archive_url', 'get.limristem.eu/monitor')}" msg = MIMEText(msg_text) msg['Subject'] = f"Limristem Monitor Update Available: {remote_version}" msg['From'] = smtp_conf.get('from_email', 'noreply@limristem-monitor.local') msg['To'] = ', '.join(recipients) s_host = smtp_conf.get('host', 'localhost') s_port = smtp_conf.get('port', 25) s_user = smtp_conf.get('username') s_pass = smtp_conf.get('password') with smtplib.SMTP(s_host, s_port) as server: if smtp_conf.get('tls', False): server.starttls() if s_user and s_pass: server.login(s_user, s_pass) server.send_message(msg, to_addrs=recipients) logging.info(f"📧 Alert aggiornamento inviato a {', '.join(recipients)}") except Exception as e: logging.warning(f"Errore durante il controllo aggiornamenti: {e}") time.sleep(86400) # Controlla ogni 24 ore if __name__ == '__main__': conf = load_config() if not conf: raise SystemExit(1) settings = conf.get('settings', {}) sync_status_entries(conf) try: ensure_storage_roots(conf) update_site_summaries(conf) except (OSError, ValueError, json.JSONDecodeError) as e: logging.error(f"❌ Errore inizializzazione storage Limristem Monitor: {e}") # 1. Thread Monitoraggio threading.Thread(target=monitoring_loop, daemon=True).start() # Thread Controllo Aggiornamenti threading.Thread(target=check_for_updates, daemon=True).start() # 2. Thread HTTP http_port = _as_positive_int(settings.get('http_port', 8080), 8080, minimum=1) t_http = threading.Thread(target=run_http_server, args=(http_port,), daemon=True) t_http.start() # 3. Thread HTTPS (se configurato) https_port = _as_positive_int(settings.get('https_port', 8443), 8443, minimum=1) t_https = threading.Thread(target=run_https_server, args=(https_port,), daemon=True) t_https.start() # Loop principale per tenere vivo lo script try: while True: time.sleep(1) except KeyboardInterrupt: logging.info('🛑 Arresto richiesto. Uscita...')