#!/usr/bin/env python3 """ ProxMenux Flask Server - Provides REST API endpoints for Proxmox monitoring (system, storage, network, VMs, etc.) - Serves the Next.js dashboard as static files - Integrates a web terminal powered by xterm.js """ import glob import json import logging import math import os import platform import re import select import shutil import socket import sqlite3 import subprocess import sys import time import threading import urllib.parse import hardware_monitor from health_persistence import health_persistence import xml.etree.ElementTree as ET from datetime import datetime, timedelta from functools import wraps from pathlib import Path import jwt import psutil from flask import Flask, jsonify, request, send_file, send_from_directory, Response from flask_cors import CORS # Ensure local imports work even if working directory changes BASE_DIR = os.path.dirname(os.path.abspath(__file__)) if BASE_DIR not in sys.path: sys.path.insert(0, BASE_DIR) from flask_script_runner import script_runner import threading from proxmox_storage_monitor import proxmox_storage_monitor from flask_terminal_routes import terminal_bp, init_terminal_routes # noqa: E402 from flask_health_routes import health_bp # noqa: E402 from flask_auth_routes import auth_bp # noqa: E402 from flask_proxmenux_routes import proxmenux_bp # noqa: E402 from flask_security_routes import security_bp # noqa: E402 from flask_notification_routes import notification_bp # noqa: E402 from flask_oci_routes import oci_bp # noqa: E402 from notification_manager import notification_manager # noqa: E402 import post_install_versions # noqa: E402 — Sprint 12A: detect post-install function updates from jwt_middleware import require_auth # noqa: E402 import auth_manager # noqa: E402 # ------------------------------------------------------------------- # Logging # ------------------------------------------------------------------- logger = logging.getLogger("proxmenux.flask") logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) # ------------------------------------------------------------------- # Proxmox node name cache # ------------------------------------------------------------------- _PROXMOX_NODE_CACHE = {"name": None, "timestamp": 0.0} _PROXMOX_NODE_CACHE_TTL = 300 # seconds (5 minutes) def get_proxmox_node_name() -> str: """ Retrieve the real Proxmox node name for the LOCAL node. - First tries reading from: `pvesh get /nodes` - In a cluster, matches the local hostname against the node list - Uses an in-memory cache to avoid repeated API calls - Falls back to the short hostname if the API call fails """ now = time.time() cached_name = _PROXMOX_NODE_CACHE.get("name") cached_ts = _PROXMOX_NODE_CACHE.get("timestamp", 0.0) # Cache hit if cached_name and (now - float(cached_ts)) < _PROXMOX_NODE_CACHE_TTL: return str(cached_name) # Get local hostname for matching local_hostname = socket.gethostname().split(".", 1)[0].lower() # Try Proxmox API try: result = subprocess.run( ["pvesh", "get", "/nodes", "--output-format", "json"], capture_output=True, text=True, timeout=5, check=False, ) if result.returncode == 0 and result.stdout: nodes = json.loads(result.stdout) if isinstance(nodes, list) and nodes: # In a cluster, find the node that matches local hostname # Node names in Proxmox typically match the hostname for node_info in nodes: node_name = node_info.get("node", "") if node_name.lower() == local_hostname: _PROXMOX_NODE_CACHE["name"] = node_name _PROXMOX_NODE_CACHE["timestamp"] = now return node_name # If no exact match, try partial match (hostname might be truncated) for node_info in nodes: node_name = node_info.get("node", "") if local_hostname.startswith(node_name.lower()) or node_name.lower().startswith(local_hostname): _PROXMOX_NODE_CACHE["name"] = node_name _PROXMOX_NODE_CACHE["timestamp"] = now return node_name # Last resort: if single node cluster, use that node if len(nodes) == 1: node_name = nodes[0].get("node") if node_name: _PROXMOX_NODE_CACHE["name"] = node_name _PROXMOX_NODE_CACHE["timestamp"] = now return node_name except Exception as exc: logger.warning("Failed to get Proxmox node name from API: %s", exc) # Fallback: short hostname (without domain) return local_hostname # ------------------------------------------------------------------- # Flask application and Blueprints # ------------------------------------------------------------------- app = Flask(__name__) # DoS / cost-amplification cap (audit Tier 3.1 — sin body-size cap en POSTs). # Without this an authenticated client could POST a 100 MB body to /api/notifications/test-ai # (or any other AI endpoint) and stall the dispatch thread plus rack up real # money on the configured AI provider. 1 MB is generous for any legitimate # request — settings payloads top out around 50 KB, AI prompts at 8 KB. app.config['MAX_CONTENT_LENGTH'] = 1 * 1024 * 1024 # 1 MB # Restrict CORS to known origins (audit Tier 3 #18). The previous wide-open # `CORS(app)` enabled drive-by attacks: any malicious site visited by an # admin's browser could call the API directly. # # In production the AppImage serves both Flask AND the Next.js static export # from the same origin — CORS isn't actually needed for the dashboard itself # in that case. We keep the dev origin (`localhost:3000` for `npm run dev`) # always allowed, and let operators add their own deployment URL via the # PROXMENUX_CORS_ORIGINS env var (comma-separated). _cors_origins = [ "http://localhost:3000", "http://127.0.0.1:3000", "http://localhost:8008", "http://127.0.0.1:8008", "https://localhost:8008", "https://127.0.0.1:8008", ] _extra_origins = os.environ.get("PROXMENUX_CORS_ORIGINS", "").strip() if _extra_origins: _cors_origins.extend(o.strip() for o in _extra_origins.split(",") if o.strip()) CORS(app, origins=_cors_origins, supports_credentials=True) # Sprint 12A: scan for post-install function updates BEFORE the rest of # the update-check pipeline (Proxmox upgrade poll, ProxMenux self-update # check) kicks in. The scan parses the on-disk auto/customizable post- # install scripts and compares each declared `# version:` against the # version recorded in installed_tools.json — anything bumped is cached # in memory + written to updates_available.json so the bash menu and # the notification poller can read it without re-parsing. post_install_versions.scan_at_startup() # Register Blueprints app.register_blueprint(auth_bp) app.register_blueprint(health_bp) app.register_blueprint(proxmenux_bp) app.register_blueprint(security_bp) app.register_blueprint(notification_bp) app.register_blueprint(oci_bp) # Initialize terminal / WebSocket routes init_terminal_routes(app) # ------------------------------------------------------------------- # Security headers — defense-in-depth (audit Tier 2 #16) # ------------------------------------------------------------------- # Defense-in-depth header policy. The XSS sinks in the React frontend are # closed (audit Tier 2 #13/#14/#17b — VM notes, Lynis report, terminal # interaction message). This header layer is the second line of defense. # # `script-src 'self' 'unsafe-inline' 'unsafe-eval'`: # - `'unsafe-inline'` is REQUIRED because Next.js static export injects # `