Merge remote-tracking branch 'origin/develop' into HEAD

AppImage/scripts/build_appimage.sh

@@ -164,6 +164,8 @@ cp "$SCRIPT_DIR/temperature_sensor_resolver.py" "$APP_DIR/usr/bin/" 2>/dev/null
 cp "$SCRIPT_DIR/proxmox_storage_monitor.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  proxmox_storage_monitor.py not found"
 cp "$SCRIPT_DIR/pbs_manager.py" "$APP_DIR/usr/bin/"
 cp "$SCRIPT_DIR/flask_pbs_routes.py" "$APP_DIR/usr/bin/"
+cp "$SCRIPT_DIR/monitor_links.py" "$APP_DIR/usr/bin/"
+cp "$SCRIPT_DIR/flask_monitor_links_routes.py" "$APP_DIR/usr/bin/"
 cp "$APPIMAGE_ROOT/../scripts/global/proxmox_platform.py" "$APP_DIR/usr/bin/"
 cp "$SCRIPT_DIR/flask_script_runner.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  flask_script_runner.py not found"
 cp "$SCRIPT_DIR/security_manager.py" "$APP_DIR/usr/bin/" 2>/dev/null || echo "⚠️  security_manager.py not found"

AppImage/scripts/flask_monitor_links_routes.py

@@ -0,0 +1,58 @@
+"""Links between ProxMenux Monitors for the node selector in the header."""
+
+from flask import Blueprint, jsonify, request
+import monitor_links
+from jwt_middleware import require_auth, require_admin_scope
+
+monitor_links_bp = Blueprint("monitor_links", __name__)
+
+
+@monitor_links_bp.after_request
+def no_cache(response):
+    response.headers["Cache-Control"] = "no-store"
+    return response
+
+
+@monitor_links_bp.route("/api/monitor-links")
+@require_auth
+def list_links():
+    return jsonify(monitor_links.cached_links(include_hidden=request.args.get("all") == "1"))
+
+
+@monitor_links_bp.route("/api/monitor-links", methods=["POST"])
+@require_admin_scope
+def add_link():
+    try:
+        key = monitor_links.add_link(request.get_json(silent=True))
+        monitor_links.refresh()
+        return jsonify(success=True, key=key)
+    except monitor_links.LinkError as exc:
+        return jsonify(error=str(exc)), 400
+    except OSError:
+        return jsonify(error="Could not save the node list."), 500
+
+
+@monitor_links_bp.route("/api/monitor-links/<path:key>", methods=["PATCH"])
+@require_admin_scope
+def update_link(key):
+    try:
+        monitor_links.update_link(key, request.get_json(silent=True))
+        monitor_links.refresh()
+        return jsonify(success=True)
+    except monitor_links.LinkError as exc:
+        return jsonify(error=str(exc)), 400
+    except OSError:
+        return jsonify(error="Could not save the node list."), 500
+
+
+@monitor_links_bp.route("/api/monitor-links/<path:key>", methods=["DELETE"])
+@require_admin_scope
+def delete_link(key):
+    try:
+        monitor_links.delete_link(key)
+        monitor_links.refresh()
+        return jsonify(success=True)
+    except monitor_links.LinkError as exc:
+        return jsonify(error=str(exc)), 400
+    except OSError:
+        return jsonify(error="Could not save the node list."), 500

AppImage/scripts/flask_notification_routes.py

@@ -1156,6 +1156,14 @@ def setup_pve_webhook_core() -> dict:
         'fallback_commands': [],
         'error': None,
     }
+    try:  # Proxmox Backup Server has no /etc/pve notification configuration
+        from pbs_manager import is_pbs_host
+        on_pbs = is_pbs_host()
+    except Exception:
+        on_pbs = False
+    if on_pbs:
+        result['not_applicable'] = True
+        return result
     
     try:
         if not _pve_cluster_fs_mounted():
@@ -1339,6 +1347,13 @@ def cleanup_pve_webhook_core() -> dict:
     Only removes blocks named 'proxmenux-webhook' / 'proxmenux-default'.
     """
     result = {'cleaned': False, 'error': None}
+    try:  # Proxmox Backup Server has no /etc/pve notification configuration
+        from pbs_manager import is_pbs_host
+        on_pbs = is_pbs_host()
+    except Exception:
+        on_pbs = False
+    if on_pbs:
+        return {'cleaned': True, 'error': None, 'not_applicable': True}
     
     try:
         if not _pve_cluster_fs_mounted():

AppImage/scripts/flask_server.py

@@ -91,6 +91,7 @@ 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_pbs_routes import pbs_bp  # noqa: E402
+from flask_monitor_links_routes import monitor_links_bp  # noqa: E402
 from pbs_manager import get_platform, is_pbs_host  # noqa: E402
 from flask_security_routes import security_bp  # noqa: E402
 from flask_notification_routes import notification_bp  # noqa: E402
@@ -198,6 +199,7 @@ app.register_blueprint(auth_bp)
 app.register_blueprint(health_bp)
 app.register_blueprint(proxmenux_bp)
 app.register_blueprint(pbs_bp)
+app.register_blueprint(monitor_links_bp)
 app.register_blueprint(security_bp)
 app.register_blueprint(notification_bp)
 app.register_blueprint(oci_bp)
@@ -12216,6 +12218,17 @@ def _get_node_metrics_payload(timeframe, max_age=_NODE_METRICS_TTL):
         lock.release()
 
 
+def _node_rrd_command(local_node, timeframe, cf=None):
+    """Node RRD query; Proxmox Backup Server keeps the same fields under its own API."""
+    if is_pbs_host():
+        return ['proxmox-backup-debug', 'api', 'get', '/nodes/localhost/rrd',
+                '--timeframe', timeframe, '--cf', cf or 'AVERAGE', '--output-format', 'json']
+    command = ['pvesh', 'get', f'/nodes/{local_node}/rrddata', '--timeframe', timeframe]
+    if cf:
+        command += ['--cf', cf]
+    return command + ['--output-format', 'json']
+
+
 def _compute_node_metrics_payload(timeframe: str) -> dict:
     """Fetch and shape one node RRD snapshot within a bounded query budget."""
     valid_timeframes = ('hour', 'day', 'week', 'month', 'year')
@@ -12239,8 +12252,7 @@ def _compute_node_metrics_payload(timeframe: str) -> dict:
 
     try:
         rrd_result = subprocess.run(
-            ['pvesh', 'get', f'/nodes/{local_node}/rrddata',
-             '--timeframe', timeframe, '--output-format', 'json'],
+            _node_rrd_command(local_node, timeframe),
             capture_output=True, text=True, timeout=_NODE_METRICS_TIMEOUT,
         )
     except subprocess.TimeoutExpired:
@@ -12334,9 +12346,7 @@ def _compute_node_metrics_payload(timeframe: str) -> dict:
             return None
         try:
             extra = subprocess.run(
-                ['pvesh', 'get', f'/nodes/{local_node}/rrddata',
-                 '--timeframe', timeframe, '--cf', cf,
-                 '--output-format', 'json'],
+                _node_rrd_command(local_node, timeframe, cf),
                 capture_output=True, text=True, timeout=remaining,
             )
             if extra.returncode == 0 and extra.stdout:
@@ -23355,8 +23365,8 @@ if __name__ == '__main__':
     # the 24 h PollingCollector cycle. Runs in a daemon thread with
     # a delay so it doesn't block startup — notification_manager needs
     # a moment to bind channels before we emit.
-    # The guest services below read VMs, LXCs and the PVE node metrics,
-    # none of which exist on a Proxmox Backup Server.
+    # The guest services below read VMs and LXCs, which do not exist on a
+    # Proxmox Backup Server.
     pve_host = not is_pbs_host()
     try:
         def _startup_emit_pending_app_updates():
@@ -23375,13 +23385,19 @@ if __name__ == '__main__':
 
     # Prewarm only the Overview's default day range; other ranges are lazy.
     try:
-        if pve_host:
-            metrics_thread = threading.Thread(target=_node_metrics_prewarmer_loop, daemon=True, name='node-metrics-prewarmer')
-            metrics_thread.start()
-            print("[ProxMenux] Node-metrics prewarmer started (day range, 90s interval)")
+        metrics_thread = threading.Thread(target=_node_metrics_prewarmer_loop, daemon=True, name='node-metrics-prewarmer')
+        metrics_thread.start()
+        print("[ProxMenux] Node-metrics prewarmer started (day range, 90s interval)")
     except Exception as e:
         print(f"[ProxMenux] Node-metrics prewarmer failed to start: {e}")
 
+    # Linked Monitors for the header node selector, renewed every 5 minutes.
+    try:
+        import monitor_links
+        monitor_links.start_prewarmer()
+    except Exception as e:
+        print(f"[ProxMenux] Monitor links prewarmer failed to start: {e}")
+
     # ── VM/CT modal-cache prewarmer ──
     # Keeps _vm_details_cache / _vm_backups_cache / app snapshots /
     # _vm_schedule_cache warm from the backend so the "Loading

AppImage/scripts/monitor_links.py

@@ -0,0 +1,321 @@
+"""Other ProxMenux Monitors reachable from this one: cluster nodes, PBS servers
+and entries added by hand.
+
+Only links are kept here. No credential is stored or sent: the browser opens
+the other Monitor, which keeps its own sign-in.
+"""
+
+import ipaddress
+import json
+import os
+import re
+import secrets
+import socket
+import ssl
+import subprocess
+import threading
+import time
+import urllib.parse
+import urllib.request
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+import pbs_manager
+
+LINKS_FILE = Path(os.environ.get("PROXMENUX_MONITOR_LINKS", "/usr/local/share/proxmenux/monitor_links.json"))
+SSL_CONFIG_FILE = Path(os.environ.get("PROXMENUX_SSL_CONFIG", "/etc/proxmenux/ssl_config.json"))
+STORAGE_CFG = Path("/etc/pve/storage.cfg")
+MONITOR_PORT = 8008
+PBS_UI_PORT = 8007
+PROBE_TIMEOUT = 3
+PROBE_TTL = 300
+MAX_MANUAL = 64
+PRODUCTS = ("pve", "pbs")
+
+_lock = threading.Lock()
+_probe_cache = {}
+_snapshot = {"at": 0.0, "data": None}
+_refresh_lock = threading.Lock()
+
+
+class LinkError(ValueError):
+    pass
+
+
+def _load():
+    try:
+        data = json.loads(LINKS_FILE.read_text())
+    except (OSError, ValueError):
+        return {"manual": [], "overrides": {}}
+    manual = data.get("manual") if isinstance(data.get("manual"), list) else []
+    overrides = data.get("overrides") if isinstance(data.get("overrides"), dict) else {}
+    return {"manual": [row for row in manual if isinstance(row, dict)], "overrides": overrides}
+
+
+def _save(data):
+    LINKS_FILE.parent.mkdir(parents=True, exist_ok=True)
+    temporary = LINKS_FILE.with_suffix(".tmp")
+    temporary.write_text(json.dumps(data, indent=2))
+    os.chmod(temporary, 0o600)
+    os.replace(temporary, LINKS_FILE)
+
+
+def normalize_url(value):
+    """Return the origin of an http(s) address, or raise LinkError."""
+    if not isinstance(value, str) or not value.strip():
+        raise LinkError("Enter the address of the Monitor.")
+    text = value.strip()
+    if "://" not in text:
+        text = "https://" + text
+    parsed = urllib.parse.urlsplit(text)
+    if parsed.scheme not in ("http", "https") or not parsed.hostname:
+        raise LinkError("The address must start with http:// or https:// and name a host.")
+    if parsed.username or parsed.password:
+        raise LinkError("The address must not contain a user or password.")
+    try:
+        port = parsed.port
+    except ValueError as exc:
+        raise LinkError("The port of the address is not valid.") from exc
+    host = parsed.hostname
+    if ":" in host:
+        host = f"[{host}]"
+    return f"{parsed.scheme}://{host}" + (f":{port}" if port else "")
+
+
+def _local_scheme():
+    try:
+        return "https" if json.loads(SSL_CONFIG_FILE.read_text()).get("enabled") else "http"
+    except (OSError, ValueError, AttributeError):
+        return "http"
+
+
+def _host_url(host, port, scheme=None):
+    if ":" in host and not host.startswith("["):
+        host = f"[{host}]"
+    return f"{scheme or _local_scheme()}://{host}:{port}"
+
+
+def _is_private(host):
+    """Only addresses in private ranges are probed without being asked to."""
+    try:
+        addresses = {info[4][0] for info in socket.getaddrinfo(host, None)}
+    except OSError:
+        return False
+    try:
+        return bool(addresses) and all(ipaddress.ip_address(a.split("%")[0]).is_private for a in addresses)
+    except ValueError:
+        return False
+
+
+def _cluster_nodes():
+    try:
+        result = subprocess.run(["pvesh", "get", "/cluster/status", "--output-format", "json"],
+                                capture_output=True, text=True, timeout=8)
+        rows = json.loads(result.stdout) if result.returncode == 0 else []
+    except (OSError, subprocess.SubprocessError, ValueError):
+        return []
+    nodes = []
+    for row in rows if isinstance(rows, list) else []:
+        if isinstance(row, dict) and row.get("type") == "node" and not row.get("local") and row.get("name") and row.get("ip"):
+            nodes.append({"key": f"cluster:{row['name']}", "name": str(row["name"]), "product": "pve",
+                          "group": "cluster", "host": str(row["ip"])})
+    return nodes
+
+
+def _storage_pbs():
+    try:
+        text = STORAGE_CFG.read_text()
+    except OSError:
+        return []
+    servers, current = [], None
+    for line in text.splitlines():
+        header = re.match(r"^(\w+):\s*(\S+)", line)
+        if header:
+            current = {"name": header.group(2), "port": PBS_UI_PORT} if header.group(1) == "pbs" else None
+            if current is not None:
+                servers.append(current)
+            continue
+        option = re.match(r"^\s+(server|port)\s+(\S+)", line)
+        if current is not None and option:
+            current[option.group(1)] = option.group(2)
+    return [row for row in servers if row.get("server")]
+
+
+def _pbs_servers():
+    found = {}
+    for row in _storage_pbs():
+        host = row["server"]
+        port = row.get("port") if str(row.get("port", "")).isdigit() else PBS_UI_PORT
+        found.setdefault(host, {"name": row["name"], "native_url": _host_url(host, port, "https")})
+    try:
+        for row in pbs_manager.public_servers():
+            host = urllib.parse.urlsplit(row.get("url", "")).hostname
+            if host:
+                found.setdefault(host, {"name": row.get("name") or host, "native_url": normalize_url(row["url"])})
+    except Exception:
+        pass
+    return [{"key": f"pbs:{host}", "name": info["name"], "product": "pbs", "group": "pbs",
+             "host": host, "native_url": info["native_url"]} for host, info in found.items()]
+
+
+def _probe(url):
+    """True when a ProxMenux Monitor answers at url; nothing is sent but the request."""
+    now = time.monotonic()
+    with _lock:
+        cached = _probe_cache.get(url)
+        if cached and cached[0] > now:
+            return cached[1]
+    context = ssl.create_default_context()
+    context.check_hostname = False
+    context.verify_mode = ssl.CERT_NONE  # liveness only: no credential or token is ever sent
+    online = False
+    try:
+        request = urllib.request.Request(url + "/api/auth/status", headers={"Accept": "application/json"})
+        with urllib.request.urlopen(request, timeout=PROBE_TIMEOUT, context=context) as response:
+            online = response.status == 200 and "auth_enabled" in json.loads(response.read(4096))
+    except Exception:
+        online = False
+    with _lock:
+        if len(_probe_cache) > 256:
+            _probe_cache.clear()
+        _probe_cache[url] = (now + PROBE_TTL, online)
+    return online
+
+
+def _resolve(entry, override):
+    """Pick the Monitor address and its state for one discovered or manual entry."""
+    if override.get("url"):
+        return override["url"], "online" if _probe(override["url"]) else "offline"
+    if entry.get("url"):
+        return entry["url"], "online" if _probe(entry["url"]) else "offline"
+    host = entry["host"]
+    if not _is_private(host):
+        return None, "unknown"
+    for scheme in (_local_scheme(), "https" if _local_scheme() == "http" else "http"):
+        url = _host_url(host, MONITOR_PORT, scheme)
+        if _probe(url):
+            return url, "online"
+    return _host_url(host, MONITOR_PORT), "offline"
+
+
+def list_links(include_hidden=False):
+    platform = pbs_manager.get_platform()
+    data = _load()
+    entries = []
+    if platform.get("product") != "pbs":
+        entries += _cluster_nodes() + _pbs_servers()
+    for row in data["manual"]:
+        entries.append({"key": f"manual:{row.get('id')}", "name": row.get("name") or row.get("url"),
+                        "product": row.get("product") if row.get("product") in PRODUCTS else "pve",
+                        "group": "pbs" if row.get("product") == "pbs" else "standalone",
+                        "url": row.get("url"), "manual": True})
+    visible = [e for e in entries if include_hidden or not data["overrides"].get(e["key"], {}).get("hidden")]
+    with ThreadPoolExecutor(max_workers=8) as executor:
+        resolved = list(executor.map(lambda e: _resolve(e, data["overrides"].get(e["key"], {})), visible))
+    links = []
+    for entry, (url, state) in zip(visible, resolved):
+        override = data["overrides"].get(entry["key"], {})
+        links.append({"key": entry["key"], "name": entry["name"], "product": entry["product"],
+                      "group": entry["group"], "url": url, "status": state,
+                      "native_url": entry.get("native_url"), "manual": bool(entry.get("manual")),
+                      "hidden": bool(override.get("hidden")), "custom_url": bool(override.get("url"))})
+    return {"self": {"name": platform.get("hostname") or socket.gethostname(), "product": platform.get("product")},
+            "links": links}
+
+
+def refresh():
+    """Recompute the list with fresh probes and keep it for the header."""
+    with _refresh_lock:
+        with _lock:
+            _probe_cache.clear()
+        data = list_links(include_hidden=True)
+        _snapshot.update(at=time.monotonic(), data=data)
+        return data
+
+
+def _refresh_quietly():
+    try:
+        refresh()
+    except Exception:
+        pass
+
+
+def cached_links(include_hidden=False):
+    """The last list at once; a stale one is renewed in the background."""
+    data = _snapshot["data"]
+    if data is None:
+        data = refresh()
+    elif time.monotonic() - _snapshot["at"] > PROBE_TTL and not _refresh_lock.locked():
+        threading.Thread(target=_refresh_quietly, daemon=True, name="monitor-links-refresh").start()
+    if include_hidden:
+        return data
+    return {**data, "links": [row for row in data["links"] if not row["hidden"]]}
+
+
+def start_prewarmer():
+    def loop():
+        time.sleep(10)
+        while True:
+            _refresh_quietly()
+            time.sleep(PROBE_TTL)
+    threading.Thread(target=loop, daemon=True, name="monitor-links-prewarmer").start()
+
+
+def add_link(payload):
+    if not isinstance(payload, dict):
+        raise LinkError("Invalid request.")
+    name = str(payload.get("name") or "").strip()[:64]
+    url = normalize_url(payload.get("url"))
+    product = payload.get("product") if payload.get("product") in PRODUCTS else "pve"
+    with _lock:
+        data = _load()
+        if len(data["manual"]) >= MAX_MANUAL:
+            raise LinkError(f"At most {MAX_MANUAL} nodes can be added.")
+        if any(row.get("url") == url for row in data["manual"]):
+            raise LinkError("This address is already in the list.")
+        identifier = secrets.token_hex(6)
+        data["manual"].append({"id": identifier, "name": name or urllib.parse.urlsplit(url).hostname,
+                               "url": url, "product": product})
+        _save(data)
+    return f"manual:{identifier}"
+
+
+def update_link(key, payload):
+    if not isinstance(payload, dict):
+        raise LinkError("Invalid request.")
+    with _lock:
+        data = _load()
+        if key.startswith("manual:"):
+            row = next((r for r in data["manual"] if f"manual:{r.get('id')}" == key), None)
+            if row is None:
+                raise LinkError("Node not found.")
+            if "name" in payload:
+                row["name"] = str(payload["name"] or "").strip()[:64] or row["name"]
+            if payload.get("url"):
+                row["url"] = normalize_url(payload["url"])
+            if payload.get("product") in PRODUCTS:
+                row["product"] = payload["product"]
+        elif re.fullmatch(r"(cluster|pbs):[\w.:\-\[\]]{1,253}", key):
+            override = data["overrides"].setdefault(key, {})
+            if "url" in payload:
+                if payload["url"]:
+                    override["url"] = normalize_url(payload["url"])
+                else:
+                    override.pop("url", None)
+            if "hidden" in payload:
+                override["hidden"] = bool(payload["hidden"])
+            if not override:
+                data["overrides"].pop(key, None)
+        else:
+            raise LinkError("Node not found.")
+        _save(data)
+
+
+def delete_link(key):
+    with _lock:
+        data = _load()
+        remaining = [r for r in data["manual"] if f"manual:{r.get('id')}" != key]
+        if len(remaining) == len(data["manual"]):
+            raise LinkError("Only nodes added by hand can be removed; detected ones can be hidden.")
+        data["manual"] = remaining
+        _save(data)

AppImage/scripts/tests/test_fwbr_udev_rule_order.py

@@ -26,18 +26,18 @@ class FwbrUdevRuleOrderTests(unittest.TestCase):
     def test_auto_post_install_uses_late_rule_and_removes_legacy_rule(self):
         source = network_optimization_block(AUTO_POST_INSTALL)
         self.assertIn('local FUNC_VERSION="1.2"', source)
-        self.assertIn(f"rm -f /etc/udev/rules.d/{OLD_RULE}", source)
-        self.assertIn(f"cat > /etc/udev/rules.d/{NEW_RULE}", source)
+        self.assertIn(f"pmx_remove_file /etc/udev/rules.d/{OLD_RULE}", source)
+        self.assertIn(f"pmx_write_file /etc/udev/rules.d/{NEW_RULE}", source)
         self.assertIn(f"chmod 0644 /etc/udev/rules.d/{NEW_RULE}", source)
-        self.assertNotIn(f"cat > /etc/udev/rules.d/{OLD_RULE}", source)
+        self.assertNotIn(f"pmx_write_file /etc/udev/rules.d/{OLD_RULE}", source)
 
     def test_customizable_post_install_uses_late_rule_and_removes_legacy_rule(self):
         source = network_optimization_block(CUSTOM_POST_INSTALL)
         self.assertIn('local FUNC_VERSION="1.2"', source)
-        self.assertIn(f"rm -f /etc/udev/rules.d/{OLD_RULE}", source)
-        self.assertIn(f"cat > /etc/udev/rules.d/{NEW_RULE}", source)
+        self.assertIn(f"pmx_remove_file /etc/udev/rules.d/{OLD_RULE}", source)
+        self.assertIn(f"pmx_write_file /etc/udev/rules.d/{NEW_RULE}", source)
         self.assertIn(f"chmod 0644 /etc/udev/rules.d/{NEW_RULE}", source)
-        self.assertNotIn(f"cat > /etc/udev/rules.d/{OLD_RULE}", source)
+        self.assertNotIn(f"pmx_write_file /etc/udev/rules.d/{OLD_RULE}", source)
 
     def test_uninstall_removes_new_and_legacy_rule_names(self):
         source = UNINSTALL_TOOLS.read_text()

AppImage/scripts/tests/test_monitor_links.py

@@ -0,0 +1,150 @@
+import json
+from pathlib import Path
+import sys
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parents[3]
+sys.path.insert(0, str(ROOT / "AppImage/scripts"))
+
+import monitor_links as links
+
+
+class MonitorLinksTests(unittest.TestCase):
+    def setUp(self):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name)
+        storage = self.root / "storage.cfg"
+        storage.write_text("dir: local\n\tpath /var/lib/vz\n\npbs: Lab\n\tdatastore store1\n\tserver 10.0.0.49\n\n"
+                           "pbs: Cloud\n\tdatastore x\n\tserver pbs.example.com\n\tport 8443\n")
+        patches = [
+            mock.patch.object(links, "LINKS_FILE", self.root / "links.json"),
+            mock.patch.object(links, "SSL_CONFIG_FILE", self.root / "ssl.json"),
+            mock.patch.object(links, "STORAGE_CFG", storage),
+            mock.patch.object(links.pbs_manager, "get_platform", return_value={"product": "pve", "hostname": "amd"}),
+            mock.patch.object(links.pbs_manager, "public_servers", return_value=[]),
+        ]
+        for patch in patches:
+            patch.start()
+            self.addCleanup(patch.stop)
+        links._probe_cache.clear()
+        cluster = [{"type": "cluster", "name": "Casa"},
+                   {"type": "node", "name": "amd", "ip": "10.0.0.55", "local": 1},
+                   {"type": "node", "name": "constructor", "ip": "10.0.0.50", "local": 0}]
+        self.run_patch = mock.patch.object(links.subprocess, "run",
+                                           return_value=SimpleNamespace(returncode=0, stdout=json.dumps(cluster)))
+        self.run_patch.start()
+        self.addCleanup(self.run_patch.stop)
+
+    def test_normalize_url_keeps_only_the_origin_and_rejects_credentials(self):
+        self.assertEqual(links.normalize_url("192.168.0.49:8008/dashboard?x=1"), "https://192.168.0.49:8008")
+        self.assertEqual(links.normalize_url("http://pve.lan"), "http://pve.lan")
+        for bad in ("", "ftp://host", "https://user:pass@host", "https://host:99999", "javascript:alert(1)"):
+            with self.subTest(bad=bad), self.assertRaises(links.LinkError):
+                links.normalize_url(bad)
+
+    def test_discovers_other_cluster_nodes_and_pbs_without_the_local_node(self):
+        probed = []
+        with mock.patch.object(links, "_probe", side_effect=lambda url: probed.append(url) or url.endswith("10.0.0.50:8008")), \
+                mock.patch.object(links, "_is_private", side_effect=lambda host: host.startswith("10.")):
+            result = links.list_links()
+        by_key = {row["key"]: row for row in result["links"]}
+        self.assertEqual(result["self"]["name"], "amd")
+        self.assertNotIn("cluster:amd", by_key)
+        self.assertEqual(by_key["cluster:constructor"]["status"], "online")
+        self.assertEqual(by_key["pbs:10.0.0.49"]["native_url"], "https://10.0.0.49:8007")
+        self.assertEqual(by_key["pbs:10.0.0.49"]["status"], "offline")
+        # A public PBS is listed with its own interface but never probed unasked.
+        self.assertEqual(by_key["pbs:pbs.example.com"]["status"], "unknown")
+        self.assertEqual(by_key["pbs:pbs.example.com"]["native_url"], "https://pbs.example.com:8443")
+        self.assertFalse(any("example.com" in url for url in probed))
+
+    def test_pbs_host_lists_only_manual_entries(self):
+        links.pbs_manager.get_platform.return_value = {"product": "pbs", "hostname": "pbs-lab"}
+        links.add_link({"name": "amd", "url": "https://10.0.0.55:8008", "product": "pve"})
+        with mock.patch.object(links, "_probe", return_value=True):
+            result = links.list_links()
+        self.assertEqual([row["name"] for row in result["links"]], ["amd"])
+        self.links_run_not_called()
+
+    def links_run_not_called(self):
+        self.assertFalse(any(call.args and call.args[0][:1] == ["pvesh"] for call in links.subprocess.run.call_args_list))
+
+    def test_manual_entries_overrides_and_hidden_detected_nodes(self):
+        key = links.add_link({"name": "lab", "url": "10.0.0.46:8008", "product": "pve"})
+        with self.assertRaises(links.LinkError):
+            links.add_link({"url": "https://10.0.0.46:8008"})
+        links.update_link("cluster:constructor", {"url": "https://constructor.lan:8008"})
+        links.update_link("pbs:10.0.0.49", {"hidden": True})
+        with mock.patch.object(links, "_probe", return_value=False), \
+                mock.patch.object(links, "_is_private", return_value=True):
+            visible = {row["key"]: row for row in links.list_links()["links"]}
+            everything = {row["key"]: row for row in links.list_links(include_hidden=True)["links"]}
+        self.assertEqual(visible[key]["url"], "https://10.0.0.46:8008")
+        self.assertEqual(visible["cluster:constructor"]["url"], "https://constructor.lan:8008")
+        self.assertNotIn("pbs:10.0.0.49", visible)
+        self.assertTrue(everything["pbs:10.0.0.49"]["hidden"])
+        with self.assertRaises(links.LinkError):
+            links.delete_link("cluster:constructor")
+        links.delete_link(key)
+        self.assertEqual(json.loads((self.root / "links.json").read_text())["manual"], [])
+        self.assertEqual((self.root / "links.json").stat().st_mode & 0o777, 0o600)
+
+    def test_cached_list_is_served_without_new_probes_and_renewed_when_stale(self):
+        links._snapshot.update(at=0.0, data=None)
+        self.addCleanup(links._snapshot.update, at=0.0, data=None)
+        links.update_link("pbs:10.0.0.49", {"hidden": True})
+        with mock.patch.object(links, "_probe", return_value=True) as probe, \
+                mock.patch.object(links, "_is_private", return_value=True):
+            first = links.cached_links()
+            calls = probe.call_count
+            again = links.cached_links()
+            self.assertEqual(probe.call_count, calls)
+            self.assertEqual(first, again)
+            self.assertNotIn("pbs:10.0.0.49", [row["key"] for row in first["links"]])
+            self.assertIn("pbs:10.0.0.49", [row["key"] for row in links.cached_links(include_hidden=True)["links"]])
+            links._snapshot["at"] -= links.PROBE_TTL + 1
+            with mock.patch.object(links.threading, "Thread") as thread:
+                stale = links.cached_links()
+            self.assertEqual(stale, first)
+            thread.assert_called_once()
+
+    def test_unknown_keys_are_rejected(self):
+        for key in ("other:x", "cluster:../../etc", "manual:nope"):
+            with self.subTest(key=key), self.assertRaises(links.LinkError):
+                links.update_link(key, {"hidden": True})
+
+    def test_probe_sends_no_credentials_and_needs_a_monitor_answer(self):
+        seen = []
+
+        class Response:
+            status = 200
+
+            def __init__(self, body):
+                self.body = body
+
+            def read(self, _size):
+                return self.body
+
+            def __enter__(self):
+                return self
+
+            def __exit__(self, *_):
+                return False
+
+        def urlopen(request, timeout, context):
+            seen.append(request)
+            return Response(b'{"auth_enabled": true}' if "monitor" in request.full_url else b"<html>")
+
+        with mock.patch.object(links.urllib.request, "urlopen", side_effect=urlopen):
+            self.assertTrue(links._probe("https://monitor:8008"))
+            self.assertFalse(links._probe("https://other:8008"))
+        self.assertTrue(all(request.full_url.endswith("/api/auth/status") for request in seen))
+        self.assertTrue(all("Authorization" not in request.headers for request in seen))
+
+
+if __name__ == "__main__":
+    unittest.main()