"""PBS platform, HTTPS protocol, authorization and isolated AppImage regressions.""" import ast from datetime import datetime, timedelta, timezone import hashlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import ipaddress import json from pathlib import Path import shlex import ssl import subprocess import sys import tempfile import threading from types import SimpleNamespace import unittest from unittest import mock ROOT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(ROOT / "AppImage/scripts")) from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.x509.oid import NameOID from flask import Flask import jwt import auth_manager import pbs_manager as pbs from flask_pbs_routes import pbs_bp class PlatformTests(unittest.TestCase): def test_supported_platforms_and_partial_dpkg_failure(self): module = pbs.detect_platform.__globals__ with tempfile.TemporaryDirectory() as directory: release = Path(directory) / "os-release" for package, version, suite, product in ( ("pve-manager", "8.4.15", "bookworm", "pve"), ("pve-manager", "9.2.1", "trixie", "pve"), ("proxmox-backup-server", "3.4.0-1", "bookworm", "pbs"), ("proxmox-backup-server", "4.2.5-1", "trixie", "pbs"), ): release.write_text(f'VERSION_CODENAME="{suite}"\n') output = f"{package}\tinstall ok installed\t{version}\n" with self.subTest(product=product, version=version), \ mock.patch.dict(module, {"OS_RELEASE": release}), \ mock.patch.object(subprocess, "run", return_value=SimpleNamespace(stdout=output, returncode=1)): detected = pbs.detect_platform() self.assertEqual(detected["product"], product) self.assertEqual(detected["version"], version) self.assertTrue(detected["supported"]) def test_client_only_and_removed_server_are_not_pbs(self): for output in ("proxmox-backup-client\tinstall ok installed\t4.2.5\n", "proxmox-backup-server\tdeinstall ok config-files\t4.2.5\n", ""): with self.subTest(output=output), mock.patch.object(subprocess, "run", return_value=SimpleNamespace(stdout=output)): self.assertEqual(pbs.detect_platform()["product"], "unknown") def test_pbs_utilities_preserve_sources_and_never_call_pve_policy(self): helper = ROOT / "scripts/global/repository-functions.sh" for supported in (True, False): harness = f''' source {shlex.quote(str(helper))} pmx_platform_field() {{ if [[ "$1" == product ]]; then echo pbs; else echo {str(supported).lower()}; fi; }} msg_info() {{ :; }} msg_error() {{ :; }} translate() {{ printf '%s' "$1"; }} pveversion() {{ echo 'FORBIDDEN: pveversion'; return 90; }} repository_policy() {{ echo 'FORBIDDEN: repository mutation'; return 90; }} apt-get() {{ echo 'FORBIDDEN: apt'; return 90; }} ensure_repositories ''' result = subprocess.run(["bash"], input=harness, text=True, capture_output=True) self.assertNotIn("FORBIDDEN", result.stdout + result.stderr) self.assertEqual(result.returncode, 0 if supported else 1) class PBSManagerTests(unittest.TestCase): def setUp(self): temporary = tempfile.TemporaryDirectory() self.addCleanup(temporary.cleanup) self.root = Path(temporary.name) patch = mock.patch.multiple(auth_manager, CONFIG_DIR=self.root, AUTH_CONFIG_FILE=self.root / "auth.json") patch.start(); self.addCleanup(patch.stop) auth_manager.save_auth_config({"enabled": True, "username": "admin", "password_hash": "configured", "jwt_secret": "s" * 48, "api_tokens": [], "revoked_tokens": []}) auth_manager._invalidate_revoked_cache() pbs._status_cache.clear() self.server = {"name": "Backup", "url": "https://pbs.test:8007", "token_id": "audit@pbs!monitor", "token_secret": "fixture-token-secret", "fingerprint": "a" * 64} app = Flask(__name__); app.register_blueprint(pbs_bp) self.client = app.test_client() self.headers = {"Authorization": "Bearer " + auth_manager.generate_token("admin")} def test_save_redacts_secret_and_probes_authenticated_endpoint(self): with mock.patch.object(pbs, "_remote_get", return_value=[]) as request: identifier = pbs.save_server(self.server) request.assert_called_once() self.assertEqual(request.call_args.args[1], "/status/datastore-usage") self.assertEqual((self.root / "pbs_servers.json").stat().st_mode & 0o777, 0o600) public = pbs.public_servers() self.assertEqual(public[0]["id"], identifier) self.assertNotIn("fixture-token-secret", json.dumps(public)) self.assertTrue(public[0]["has_token"]) with mock.patch.object(pbs, "_remote_get", return_value=[]): pbs.save_server({"id": identifier, "name": "Renamed"}) self.assertEqual(pbs._load_servers()[0]["token_secret"], "fixture-token-secret") pbs.delete_server(identifier) self.assertEqual(pbs.public_servers(), []) def test_failed_credentials_or_invalid_config_do_not_replace_saved_connection(self): with mock.patch.object(pbs, "_remote_get", return_value=[]): identifier = pbs.save_server(self.server) original = (self.root / "pbs_servers.json").read_bytes() with mock.patch.object(pbs, "_remote_get", side_effect=pbs.PBSError("Rejected token")): with self.assertRaises(pbs.PBSError): pbs.save_server({**self.server, "id": identifier, "token_secret": "wrong"}) for changes in ({"url": "http://pbs.test"}, {"url": "https://user:pass@pbs.test"}, {"fingerprint": "bad-pin"}, {"token_id": "not-a-token"}, {"token_secret": "line\nbreak"}): with self.subTest(changes=changes), self.assertRaises(pbs.PBSError): pbs.save_server({**self.server, **changes, "id": identifier}) self.assertEqual((self.root / "pbs_servers.json").read_bytes(), original) def test_remote_partial_failure_is_unknown_instead_of_zero(self): def remote(server, endpoint): if endpoint == "/version": return {"version": "4.2.5"} if endpoint == "/status/datastore-usage": return [{"store": "backup", "total": 100, "used": 30, "avail": 70}] if endpoint == "/admin/verify": raise pbs.PBSError("Permission denied") return [] with mock.patch.object(pbs, "_remote_get", side_effect=remote) as request: identifier = pbs.save_server(self.server) result = pbs.get_status(identifier) count = request.call_count self.assertEqual(pbs.get_status(identifier), result) self.assertEqual(request.call_count, count) self.assertEqual(result["datastores"][0]["name"], "backup") self.assertIsNone(result["verify"]) self.assertIn("verify", result["errors"]) def test_unmounted_removable_datastore_has_no_fabricated_capacity(self): with mock.patch.object(pbs, "_cli", return_value=[{"name": "removable", "path": "/not-mounted", "backing-device": "uuid"}]), \ mock.patch.object(pbs.os.path, "ismount", return_value=False), \ mock.patch.object(pbs.shutil, "disk_usage") as usage: result = pbs._local_datastores() usage.assert_not_called() self.assertIsNone(result[0]["total"]) self.assertIn("error", result[0]) def test_local_status_reads_pbs_commands_and_filesystem_capacity(self): def run(command, **kwargs): self.assertNotIn("shell", kwargs) self.assertEqual(command[0], "proxmox-backup-manager") self.assertEqual(command[-2:], ["--output-format", "json"]) data = [{"name": "backup", "path": str(self.root)}] if command[1:3] == ["datastore", "list"] else [] return SimpleNamespace(stdout=json.dumps(data), returncode=0) with mock.patch.object(pbs, "get_platform", return_value={"product": "pbs", "version": "4.2.5"}), \ mock.patch.object(pbs, "_local_services", return_value={key: "active" for key in pbs.PBS_SERVICES}), \ mock.patch.object(pbs.subprocess, "run", side_effect=run), \ mock.patch.object(pbs.shutil, "disk_usage", return_value=SimpleNamespace(total=100, used=30, free=70)): result = pbs.get_status("local") self.assertEqual(result["version"], "4.2.5") self.assertEqual(result["errors"], {}) self.assertEqual(result["datastores"][0]["used"], 30) self.assertEqual(set(result["services"]), set(pbs.PBS_SERVICES)) def test_connection_api_requires_full_admin_and_does_not_return_secret(self): self.assertEqual(self.client.get("/api/pbs/servers").status_code, 401) with mock.patch.object(pbs, "_remote_get", return_value=[]): response = self.client.post("/api/pbs/servers", headers=self.headers, json=self.server) self.assertEqual(response.status_code, 200) self.assertNotIn("fixture-token-secret", response.text) payload = jwt.decode(self.headers["Authorization"].split()[1], "s" * 48, algorithms=["HS256"], audience=auth_manager.JWT_AUDIENCE) payload["scope"] = "read_only" read_only = {"Authorization": "Bearer " + jwt.encode(payload, "s" * 48, algorithm="HS256")} self.assertEqual(self.client.post("/api/pbs/servers", headers=read_only, json=self.server).status_code, 403) self.assertEqual(self.client.get("/api/pbs/servers", headers=read_only).status_code, 200) auth_manager.save_auth_config({"enabled": False}) self.assertEqual(self.client.post("/api/pbs/servers", json=self.server).status_code, 409) def test_health_uses_pbs_services_and_skips_pve_only_checks(self): source = ast.parse((ROOT / "AppImage/scripts/health_monitor.py").read_text()) names = ("_check_pve_services", "_check_vms_cts_optimized", "_check_vms_cts_with_persistence", "_check_proxmox_storages", "_check_proxmox_storage", "_check_lxc_disk_usage", "_check_vm_disk_usage", "_check_lxc_mount_capacity", "_check_pve_storage_capacity") functions = [node for node in ast.walk(source) if isinstance(node, ast.FunctionDef) and node.name in names] persistence = mock.Mock(); persistence.is_error_active.return_value = False run = mock.Mock(return_value=SimpleNamespace(returncode=0, stdout="active\n")) namespace = {"Dict": dict, "Any": object, "Optional": __import__('typing').Optional, "PBS_SERVICES": pbs.PBS_SERVICES, "health_persistence": persistence, "subprocess": SimpleNamespace(run=run), "os": __import__('os')} exec(compile(ast.Module(body=functions, type_ignores=[]), '', 'exec'), namespace) instance = SimpleNamespace(platform_type="pbs", PVE_SERVICES=["pveproxy", "pvedaemon", "pvestatd", "pve-cluster"]) services = namespace["_check_pve_services"](instance) self.assertEqual(services["status"], "OK") self.assertEqual([call.args[0][2] for call in run.call_args_list], list(pbs.PBS_SERVICES)) for name in names[1:]: result = namespace[name](instance) self.assertTrue(result is None or result.get("not_applicable")) def test_appimage_copies_and_imports_pbs_dependencies_outside_source_tree(self): builder = (ROOT / "AppImage/scripts/build_appimage.sh").read_text() modules = ("pbs_manager", "flask_pbs_routes", "auth_manager", "jwt_middleware") copies = [] for name in modules: matches = [line for line in builder.splitlines() if line.startswith(f'cp "$SCRIPT_DIR/{name}.py"')] self.assertEqual(len(matches), 1, name) copies.extend(matches) copies.extend(line for line in builder.splitlines() if line.startswith('cp "$APPIMAGE_ROOT/../scripts/global/proxmox_platform.py"')) self.assertEqual(len(copies), len(modules) + 1) app = self.root / 'AppDir'; (app / 'usr/bin').mkdir(parents=True) q = shlex.quote shell = f'APP_DIR={q(str(app))}\nSCRIPT_DIR={q(str(ROOT / "AppImage/scripts"))}\nAPPIMAGE_ROOT={q(str(ROOT / "AppImage"))}\n' + '\n'.join(copies) subprocess.run(['bash', '-e', '-c', shell], check=True) code = f'import sys; sys.path.insert(0, {str(app / "usr/bin")!r}); import pbs_manager, flask_pbs_routes, proxmox_platform' subprocess.run([sys.executable, '-I', '-B', '-c', code], cwd=self.root, check=True, capture_output=True) class TLSTests(unittest.TestCase): @classmethod def setUpClass(cls): cls.temporary = tempfile.TemporaryDirectory() root = Path(cls.temporary.name) key = rsa.generate_private_key(public_exponent=65537, key_size=2048) subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, "localhost")]) certificate = (x509.CertificateBuilder().subject_name(subject).issuer_name(subject).public_key(key.public_key()) .serial_number(x509.random_serial_number()).not_valid_before(datetime.now(timezone.utc) - timedelta(minutes=1)) .not_valid_after(datetime.now(timezone.utc) + timedelta(hours=1)) .add_extension(x509.SubjectAlternativeName([x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]), critical=False) .sign(key, hashes.SHA256())) cert_path, key_path = root / 'server.pem', root / 'server.key' cert_path.write_bytes(certificate.public_bytes(serialization.Encoding.PEM)) key_path.write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption())) cls.fingerprint = hashlib.sha256(certificate.public_bytes(serialization.Encoding.DER)).hexdigest() cls.requests = [] cls.response_status = 200 class Handler(BaseHTTPRequestHandler): def handle(self): try: super().handle() except ConnectionResetError: pass # A rejected TLS pin closes before any HTTP request. def do_GET(self): cls.requests.append((self.path, self.headers.get("Authorization"))) body = json.dumps({"data": []}).encode() self.send_response(cls.response_status) self.send_header('Content-Length', str(len(body))) self.end_headers(); self.wfile.write(body) def log_message(self, *args): pass cls.server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER); context.load_cert_chain(cert_path, key_path) cls.server.socket = context.wrap_socket(cls.server.socket, server_side=True) cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True); cls.thread.start() @classmethod def tearDownClass(cls): cls.server.shutdown(); cls.server.server_close(); cls.thread.join(timeout=5); cls.temporary.cleanup() def setUp(self): self.requests.clear(); type(self).response_status = 200 self.connection = {"url": f'https://127.0.0.1:{self.server.server_port}', "fingerprint": self.fingerprint, "token_id": "audit@pbs!monitor", "token_secret": "fake-secret"} def test_real_tls_matching_pin_sends_correct_pbs_authorization(self): self.assertEqual(pbs._remote_get(self.connection, '/status/datastore-usage'), []) self.assertEqual(self.requests, [('/api2/json/status/datastore-usage', 'PBSAPIToken=audit@pbs!monitor:fake-secret')]) def test_bad_pin_and_untrusted_ca_send_no_http_credentials(self): for fingerprint in ('0' * 64, ''): with self.subTest(fingerprint=fingerprint), self.assertRaises(pbs.PBSError): pbs._remote_get({**self.connection, "fingerprint": fingerprint}, '/status/datastore-usage') self.assertEqual(self.requests, []) def test_bad_credentials_and_redirects_are_rejected(self): for status in (401, 403, 302): type(self).response_status = status with self.subTest(status=status), self.assertRaises(pbs.PBSError): pbs._remote_get(self.connection, '/status/datastore-usage') if __name__ == '__main__': unittest.main()