#!/usr/bin/env python3 from __future__ import annotations import hashlib import importlib.util import json import os import sqlite3 import tarfile import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[3] MODULE_PATH = ROOT / "tools/k3s/verify-backup.py" SPEC = importlib.util.spec_from_file_location("verify_backup", MODULE_PATH) assert SPEC and SPEC.loader verify_backup = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(verify_backup) class VerifyBackupTest(unittest.TestCase): def build_sqlite_archive( self, root: Path, *, purpose: str = "postchange", token_mode: int = 0o600, agent_classification: str = "symlink-to-server-token", canonical: bool = True, unsafe_agent_descriptor: bool = False, ) -> tuple[Path, str]: payload = root / "fixture/payload" db = payload / "datastore/db/state.db" db.parent.mkdir(parents=True) connection = sqlite3.connect(db) connection.execute("create table proof(value text not null)") connection.execute("insert into proof values ('restored')") connection.commit() connection.close() db.chmod(0o600) token_value = "K10fixture-secret-value::0123456789abcdef0123456789abcdef\n" token = payload / "server/token" token.parent.mkdir(parents=True) token.write_text(token_value) token.chmod(token_mode) config_root = payload / "config/etc-rancher-k3s" config_root.mkdir(parents=True) ordinary = config_root / "config.yaml" ordinary.write_text("write-kubeconfig-mode: '0600'\n") ordinary.chmod(0o600) if agent_classification == "regular": agent = payload / "server/agent-token" agent.write_text("K10agent::0123456789abcdef0123456789abcdef\n") agent.chmod(0o600) agent_descriptor = { "classification": "regular", "restore": "regular-file", "payload_path": "payload/server/agent-token", } elif agent_classification == "symlink-to-server-token": agent_descriptor = { "classification": "symlink-to-server-token", "restore": "symlink-to-token", "link_target": "../outside" if unsafe_agent_descriptor else "token", "payload_path": None, } elif agent_classification == "absent": agent_descriptor = {"classification": "absent", "restore": "absent", "payload_path": None} else: agent_descriptor = {"classification": agent_classification, "restore": "unknown", "payload_path": None} launcher_sha = "a" * 64 runtime_sha = "d" * 64 if canonical: managed = config_root / "config.yaml.d/90-overdeck-control-plane.yaml" managed.parent.mkdir(parents=True) managed.write_text('write-kubeconfig-mode: "0600"\ntls-san+:\n - "100.0.0.3"\n') managed.chmod(0o600) contract = config_root / "overdeck/control-plane.json" contract.parent.mkdir(parents=True) contract.write_text( json.dumps( { "schema_version": 2, "managed_by": "overdeck-k3s-phase1", "server": {"machine_id": "machine-3", "tailscale_ipv4": "100.0.0.3"}, "api": {"endpoint": "https://100.0.0.3:6443"}, } ) ) contract.chmod(0o600) lock = config_root / "overdeck/version-lock.json" lock.write_text( json.dumps( { "schema_version": 2, "managed_by": "overdeck-k3s-phase1", "version": "v1.36.3+k3s1", "launcher": { "invocation_path": "/usr/local/bin/k3s", "resolved_path": "/usr/local/bin/k3s", "sha256": launcher_sha, }, "runtime": {"present": True, "sha256": runtime_sha}, } ) ) lock.chmod(0o600) files = [] for path in sorted(payload.rglob("*")): if path.is_file() and not path.is_symlink(): files.append( { "path": path.relative_to(payload.parent).as_posix(), "size": path.stat().st_size, "sha256": hashlib.sha256(path.read_bytes()).hexdigest(), "mode": f"{path.stat().st_mode & 0o777:04o}", "uid": 0, "gid": 0, } ) manifest = { "schema_version": 2, "created_utc": "2026-08-11T00:00:00Z", "transaction_id": "fixture", "purpose": purpose, "server": {"hostname": "debian3", "machine_id": "machine-3", "tailscale_ipv4": "100.0.0.3"}, "k3s": { "version": "v1.36.3+k3s1", "launcher": {"invocation_path": "/usr/local/bin/k3s", "resolved_path": "/usr/local/bin/k3s", "sha256": launcher_sha}, "runtime": {"present": True, "resolved_path": "/var/lib/rancher/k3s/data/hash/bin/k3s", "sha256": runtime_sha}, "cacerts_sha256": "b" * 64, }, "datastore": {"type": "sqlite"}, "server_token_path": "payload/server/token", "agent_token": agent_descriptor, "systemd": {}, "files": files, } fixture = payload.parent (fixture / "manifest.json").write_text(json.dumps(manifest)) archive = root / f"backup-{purpose}.tar" with tarfile.open(archive, "w") as handle: handle.add(fixture / "manifest.json", arcname="manifest.json") handle.add(payload, arcname="payload") return archive, token_value def test_postchange_backup_materializes_safe_agent_token_symlink(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) archive, token_value = self.build_sqlite_archive(root) restore_root = root / "restore" result = verify_backup.verify_archive(archive, restore_root=restore_root) self.assertEqual(result["status"], "verified") self.assertTrue(result["canonical_config"]["present"]) agent = restore_root / "var/lib/rancher/k3s/server/agent-token" self.assertTrue(agent.is_symlink()) self.assertEqual(os.readlink(agent), "token") self.assertEqual(agent.resolve().read_text(), token_value) self.assertNotIn(token_value.strip(), json.dumps(result)) self.assertFalse(result["isolated_restore"]["live_root_modified"]) def test_regular_agent_token_is_restored_as_regular_file(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) archive, _ = self.build_sqlite_archive(root, agent_classification="regular") restore = root / "restore" result = verify_backup.verify_archive(archive, restore_root=restore) agent = restore / "var/lib/rancher/k3s/server/agent-token" self.assertTrue(agent.is_file()) self.assertFalse(agent.is_symlink()) self.assertEqual(result["isolated_restore"]["agent_token"]["classification"], "regular") def test_prechange_backup_can_precede_canonical_file_publication(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) archive, _ = self.build_sqlite_archive(root, purpose="prechange", canonical=False) result = verify_backup.verify_archive(archive, restore_root=root / "restore") self.assertFalse(result["canonical_config"]["present"]) self.assertEqual(result["canonical_config"]["reason"], "prechange-baseline") def test_postchange_backup_requires_complete_canonical_files(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) archive, _ = self.build_sqlite_archive(root, purpose="postchange", canonical=False) with self.assertRaisesRegex(verify_backup.Phase1Error, "missing canonical"): verify_backup.verify_archive(archive) def test_unsafe_agent_symlink_descriptor_is_rejected_before_extract_restore(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) archive, _ = self.build_sqlite_archive(root, unsafe_agent_descriptor=True) with self.assertRaisesRegex(verify_backup.Phase1Error, "descriptor is unsafe"): verify_backup.verify_archive(archive, restore_root=root / "restore") def test_restore_root_must_be_empty(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp); archive, _ = self.build_sqlite_archive(root); restore = root / "restore"; restore.mkdir(); (restore / "collision").write_text("x") with self.assertRaises(verify_backup.Phase1Error): verify_backup.verify_archive(archive, restore_root=restore) def test_backup_rejects_broad_server_token_permissions(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp); archive, _ = self.build_sqlite_archive(root, token_mode=0o644) with self.assertRaises(verify_backup.Phase1Error): verify_backup.verify_archive(archive) if __name__ == "__main__": unittest.main()