"""Dona V3 — read-only, optional Proxmox dashboard renderer. No model, no MCP mutation, no writes to Proxmox. A failure must be handled by build_dashboard's existing Hermes fallback. Server-side credentials only. """ from __future__ import annotations from concurrent.futures import ThreadPoolExecutor import json import os from pathlib import Path import re import ssl import time from urllib.error import HTTPError, URLError from urllib.parse import urlsplit, urlunsplit from urllib.request import Request, urlopen OFF = {'0', 'false', 'off', 'no', 'disabled'} ENV_PATH = Path('/opt/hermes/MCP_proxmox/.env') MAX_DATA = 3 * 1024 * 1024 class ProxmoxDirectUnavailable(Exception): """Only short, non-sensitive public reason codes should be logged.""" def enabled(): return os.environ.get('DONA_PROXMOX_DIRECT', 'on').strip().lower() not in OFF def is_standard_proxmox(prompt): """Fail conservative. Let Hermes handle analyses, commands and mixed topics.""" if not enabled() or not isinstance(prompt, str): return False p = prompt.casefold().strip() if not (re.search(r'\bproxmox\b', p) and re.search(r'\b(tableau|dashboard|vue d.ensemble|synth[eè]se)\b', p)): return False if re.search(r'\b(wazuh|spotify|home\s*assistant|comfyui|paperless|gmail|youtube|courriel)\b', p): return False if re.search(r'\b(analy[sz]\w*|pourquoi|compar\w*|recommand\w*|optimis\w*|pr[eé]di\w*|histor\w*|tendance\w*|\bavant\b|\bapr[eè]s\b|migr\w*|sauvegard\w*|backup\w*|restaur\w*|red[eé]marr\w*|arr[eê]t\w*|allum\w*|cr[eé]\w*|supprim\w*|modifi\w*|ceph|stockage|r[eé]seau|sur\s+\d+\s+(jours|heures)|d[eé]taill\w*)\b', p): return False return True def _read_env_file(path): env = {} try: if not path.is_file() or path.stat().st_size > 65536: return env for line in path.read_text(encoding='utf-8').splitlines(): line = line.strip() if line.startswith('export '): line = line[7:].strip() if not line or line.startswith('#') or '=' not in line: continue key, val = line.split('=', 1) key = key.strip() if not re.fullmatch(r'[A-Z][A-Z0-9_]*', key): continue val = val.strip() if val and val[0] in ("'", '"') and val[-1:] == val[0]: val = val[1:-1] else: val = val.split(' #', 1)[0].strip() env[key] = val except (OSError, UnicodeError): return {} return env def _config(): env = {} path = os.environ.get('DONA_PROXMOX_ENV_FILE') if path: env.update(_read_env_file(Path(path).expanduser())) else: env.update(_read_env_file(ENV_PATH)) env.update(os.environ) # Live service settings override .env values. def get(*names): return next((str(env[n]).strip() for n in names if env.get(n)), '') url = get('DONA_PROXMOX_URL', 'DONA_PROXMOX_API_URL', 'PROXMOX_URL', 'PROXMOX_API_URL', 'PROXMOX_HOST', 'PVE_HOST', 'PVE_URL', 'PVE_API_URL') tid = get('DONA_PROXMOX_TOKEN_ID', 'PROXMOX_TOKEN_ID', 'PROXMOX_API_TOKEN_ID', 'PVE_TOKEN_ID', 'PVE_API_TOKEN_ID') secret = get('DONA_PROXMOX_TOKEN_SECRET', 'PROXMOX_TOKEN_SECRET', 'PROXMOX_API_TOKEN_SECRET', 'PROXMOX_TOKEN_VALUE', 'PROXMOX_API_TOKEN', 'PVE_TOKEN_SECRET', 'PVE_API_TOKEN_SECRET') if not tid: user = get('PROXMOX_USER', 'PROXMOX_USERNAME', 'PVE_USER') name = get('PROXMOX_TOKEN_NAME', 'PROXMOX_TOKEN', 'PVE_TOKEN_NAME') if user and name: tid = user + '!' + name if secret.startswith('PVEAPIToken='): secret = secret.split('=', 1)[1] # Combined 'user@realm!name=secret' format is accepted if provided as one value. if not tid and '=' in secret and '!' in secret: tid, secret = secret.split('=', 1) if not url or not tid or not secret: raise ProxmoxDirectUnavailable('configuration_absente') if not re.fullmatch(r'[^\s!=]+@[^\s!=]+![^\s!=]+', tid): raise ProxmoxDirectUnavailable('token_id_invalide') if any(c in secret for c in '\r\n'): raise ProxmoxDirectUnavailable('token_invalide') if '://' not in url: url = 'https://' + url parsed = urlsplit(url) if parsed.scheme not in ('https', 'http') or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment: raise ProxmoxDirectUnavailable('url_invalide') if parsed.scheme == 'http' and get('DONA_PROXMOX_ALLOW_HTTP') not in ('1', 'true', 'yes'): raise ProxmoxDirectUnavailable('http_non_autorise') try: port = parsed.port except ValueError as exc: raise ProxmoxDirectUnavailable('port_invalide') from exc netloc = parsed.netloc if port is not None else parsed.netloc + ':8006' path = parsed.path.rstrip('/') if path not in ('', '/api2/json'): raise ProxmoxDirectUnavailable('chemin_url_invalide') base = urlunsplit((parsed.scheme, netloc, '/api2/json', '', '')) verify = get('DONA_PROXMOX_VERIFY_SSL', 'PROXMOX_VERIFY_SSL', 'PVE_VERIFY_SSL').lower() not in OFF ca_file = get('DONA_PROXMOX_CA_FILE') timeout_str = get('DONA_PROXMOX_TIMEOUT') or '3' try: timeout = float(timeout_str) except ValueError as exc: raise ProxmoxDirectUnavailable('timeout_invalide') from exc timeout = max(1.0, min(timeout, 10.0)) return base, f'PVEAPIToken={tid}={secret}', verify, ca_file, timeout def _fetch(base, token, verify, ca_file, timeout, endpoint): if verify: try: ctx = ssl.create_default_context(cafile=ca_file or None) except (OSError, ssl.SSLError) as exc: raise ProxmoxDirectUnavailable('certificat_invalide') from exc else: ctx = ssl._create_unverified_context() # Only when explicitly selected in config. req = Request(base + endpoint, headers={'Authorization': token, 'Accept': 'application/json'}, method='GET') try: with urlopen(req, timeout=timeout, context=ctx) as r: raw = r.read(MAX_DATA + 1) if len(raw) > MAX_DATA: raise ProxmoxDirectUnavailable('reponse_trop_volumineuse') data = json.loads(raw) if not isinstance(data, dict) or not isinstance(data.get('data'), list): raise ProxmoxDirectUnavailable('schema_inattendu') return data['data'] except HTTPError as exc: raise ProxmoxDirectUnavailable(f'http_{exc.code}') from None except (URLError, TimeoutError, ssl.SSLError, OSError): raise ProxmoxDirectUnavailable('connexion_impossible') from None except (ValueError, UnicodeError): raise ProxmoxDirectUnavailable('json_invalide') from None def _percent(value): try: number = float(value) except (ValueError, TypeError): return None return round(max(0, min(100, number * 100)), 1) def _text(value, limit=80): return str(value if value is not None else '—').replace('\r', ' ').replace('\n', ' ')[:limit] def _mk_dashboard(prompt, cluster, nodes, resources): if not isinstance(cluster, list) or not isinstance(nodes, list) or not isinstance(resources, list): raise ProxmoxDirectUnavailable('schema_inattendu') nodes = [n for n in nodes if isinstance(n, dict) and n.get('node')] resources = [r for r in resources if isinstance(r, dict) and r.get('type') in ('qemu', 'lxc')] if not nodes: raise ProxmoxDirectUnavailable('aucun_noeud_visible') cards = [] count_up = sum(str(n.get('status', '')).lower() == 'online' for n in nodes) cards.append({'type': 'metric', 'title': 'Nœuds actifs', 'value': f'{count_up}/{len(nodes)}'}) qemu = [r for r in resources if r['type'] == 'qemu' and not r.get('template')] lxc = [r for r in resources if r['type'] == 'lxc' and not r.get('template')] cards.append({'type': 'metric', 'title': 'VM actives', 'value': f"{sum(r.get('status') == 'running' for r in qemu)}/{len(qemu)}"}) cards.append({'type': 'metric', 'title': 'Conteneurs LXC actifs', 'value': f"{sum(r.get('status') == 'running' for r in lxc)}/{len(lxc)}"}) c = next((x for x in cluster if isinstance(x, dict) and x.get('type') == 'cluster'), None) if c is not None and c.get('quorate') is not None: cards.append({'type': 'metric', 'title': 'Quorum', 'value': 'Oui' if str(c['quorate']) in ('1', 'True', 'true') else 'Non'}) status_nodes = [f"{_text(n['node'])} — {_text(n.get('status', 'inconnu'))}" for n in sorted(nodes, key=lambda n: str(n['node']))][:24] cards.append({'type': 'list', 'title': 'État des nœuds', 'items': status_nodes or ['Aucun nœud visible']}) cpus = [(_text(n['node']), _percent(n.get('cpu'))) for n in nodes] bars = [{'label': name, 'value': value, 'unit': '%'} for name, value in cpus if value is not None] if bars: cards.append({'type': 'bars', 'title': 'CPU par nœud', 'items': bars[:16]}) workloads = sorted(qemu + lxc, key=lambda r: (str(r.get('node', '')), str(r.get('name', '')), str(r.get('vmid', '')))) rows = [[_text(r.get('vmid')), _text(r.get('name') or f"{r.get('type')} {r.get('vmid', '')}"), 'VM' if r['type'] == 'qemu' else 'LXC', _text(r.get('node')), _text(r.get('status', 'inconnu'))] for r in workloads[:20]] if rows: cards.append({'type': 'table', 'title': 'VM et conteneurs (20 premiers au maximum)', 'columns': ['ID', 'Nom', 'Type', 'Nœud', 'État'], 'rows': rows}) if len(workloads) > 20: cards.append({'type': 'text', 'title': 'Affichage', 'text': f"{len(workloads)} VM/conteneurs visibles ; 20 affichés dans le tableau."}) return {'title': 'Tableau de bord Proxmox', 'subtitle': 'Données directes de l’API Proxmox', 'cards': cards} def build_standard_dashboard(prompt): if not enabled(): raise ProxmoxDirectUnavailable('desactive') base, token, verify, ca_file, timeout = _config() endpoints = ['/cluster/status', '/nodes', '/cluster/resources?type=vm'] with ThreadPoolExecutor(max_workers=3, thread_name_prefix='dona-proxmox-ro') as ex: results = list(ex.map(lambda e: _fetch(base, token, verify, ca_file, timeout, e), endpoints)) return _mk_dashboard(prompt, *results) def diagnostic(): start = time.monotonic() try: board = build_standard_dashboard('Tableau de bord Proxmox') except ProxmoxDirectUnavailable as exc: print('Proxmox direct : INDISPONIBLE (' + str(exc) + ')') print('Dona continuera à utiliser Hermes/MCP pour les tableaux Proxmox.') return 1 print('Proxmox direct : OK') print('Cartes construites :', len(board['cards'])) print('Temps de lecture et assemblage : %.0f ms' % ((time.monotonic()-start)*1000)) return 0 if __name__ == '__main__': raise SystemExit(diagnostic())