#!/usr/bin/env python3
from __future__ import annotations

import copy
import importlib.util
import json
import os
import sqlite3
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

MODULE_PATH = Path(__file__).resolve().parents[1] / "remote/phase1-server.py"
FIXTURE = Path(__file__).resolve().parent / "fixtures/debian3-20260811-server-state.json"
SPEC = importlib.util.spec_from_file_location("phase1_server", MODULE_PATH)
assert SPEC and SPEC.loader
server = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(server)


def live_state() -> dict:
    return json.loads(FIXTURE.read_text())


class Phase1ServerTest(unittest.TestCase):
    def test_command_lookup_uses_approved_path_not_inherited_sudo_path(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            local_bin = Path(tmp) / "usr/local/bin"
            local_bin.mkdir(parents=True)
            binary = local_bin / "k3s"
            binary.write_text("#!/bin/sh\nexit 0\n")
            binary.chmod(0o755)
            old_path = server.SAFE_EXECUTABLE_PATH
            server.SAFE_EXECUTABLE_PATH = str(local_bin)
            try:
                with patch.dict(os.environ, {"PATH": "/usr/bin:/bin"}):
                    self.assertEqual(server.command_path("k3s"), str(binary))
            finally:
                server.SAFE_EXECUTABLE_PATH = old_path

    def test_child_commands_receive_deterministic_safe_path(self) -> None:
        completed = subprocess.CompletedProcess(["true"], 0, "", "")
        with patch.object(server.subprocess, "run", return_value=completed) as mocked:
            server.run(["true"])
        self.assertEqual(mocked.call_args.kwargs["env"]["PATH"], server.SAFE_EXECUTABLE_PATH)

    @staticmethod
    def make_launcher(root: Path, version: str = "v1.36.3+k3s1") -> Path:
        binary = root / "k3s"
        binary.write_text(f"#!/bin/sh\nprintf 'k3s version {version} (fixture)\\n'\n")
        binary.chmod(0o755)
        return binary

    def test_launcher_uses_canonical_path_when_sudo_path_omits_it(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            binary = self.make_launcher(Path(tmp))
            with patch.object(server, "K3S_CANONICAL_BINARY_PATHS", (binary,)), patch.object(
                server, "systemd_k3s_launcher_candidates", return_value=[]
            ), patch.object(server, "command_path", return_value=None), patch.object(
                server.os, "geteuid", return_value=1000
            ):
                inventory = server.discover_k3s_launcher()
            self.assertEqual(inventory["invocation_path"], str(binary))
            self.assertEqual(inventory["resolved_path"], str(binary.resolve()))
            self.assertEqual(inventory["source"], "canonical")

    def test_launcher_can_be_discovered_from_systemd_execstart(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            binary = self.make_launcher(Path(tmp))
            with patch.object(server, "K3S_CANONICAL_BINARY_PATHS", ()), patch.object(
                server, "systemd_k3s_launcher_candidates", return_value=[binary]
            ), patch.object(server, "command_path", return_value=None), patch.object(
                server.os, "geteuid", return_value=1000
            ):
                inventory = server.discover_k3s_launcher()
            self.assertEqual(inventory["invocation_path"], str(binary))
            self.assertEqual(inventory["source"], "systemd-execstart")

    def test_systemd_execstart_parser_never_uses_runtime_pid_or_arguments(self) -> None:
        completed = subprocess.CompletedProcess(
            ["systemctl"],
            0,
            "{ path=/usr/local/bin/k3s ; argv[]=/usr/local/bin/k3s server --token SUPER-SECRET ; ignore_errors=no ; }\n",
            "",
        )
        with patch.object(server, "run", return_value=completed):
            candidates = server.systemd_k3s_launcher_candidates()
        self.assertEqual(candidates, [Path("/usr/local/bin/k3s")])
        self.assertNotIn("SUPER-SECRET", repr(candidates))

    def test_running_runtime_is_evidence_not_command_launcher(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            (root / "launcher").mkdir()
            launcher = self.make_launcher(root / "launcher")
            runtime_dir = root / "runtime"
            runtime_dir.mkdir()
            runtime = self.make_launcher(runtime_dir)
            runtime.write_text("#!/bin/sh\ncase \"$1\" in --version) echo 'k3s version v1.36.3+k3s1 (runtime)' ;; *) echo 'unknown command kubectl' >&2; exit 1;; esac\n")
            runtime.chmod(0o755)
            with patch.object(server, "K3S_CANONICAL_BINARY_PATHS", (launcher,)), patch.object(
                server, "systemd_k3s_launcher_candidates", return_value=[launcher]
            ), patch.object(server, "runtime_k3s_candidate", return_value=(runtime, "123")), patch.object(
                server, "command_path", return_value=None
            ), patch.object(server.os, "geteuid", return_value=1000):
                inventory = server.k3s_executable_inventory()
            self.assertEqual(inventory["launcher"]["invocation_path"], str(launcher))
            self.assertEqual(inventory["runtime"]["resolved_path"], str(runtime.resolve()))
            self.assertNotEqual(inventory["launcher"]["sha256"], inventory["runtime"]["sha256"])

    def test_wait_ready_invokes_outer_launcher_not_runtime(self) -> None:
        launcher = Path("/usr/local/bin/k3s")
        calls: list[list[str]] = []

        def fake_run(argv, **kwargs):
            calls.append(list(argv))
            self.assertEqual(argv[0], str(launcher))
            return subprocess.CompletedProcess(argv, 0, "ok\n", "")

        with patch.object(server, "service_state", return_value="active"), patch.object(server, "run", side_effect=fake_run):
            ready = server.wait_ready(1, launcher)
        self.assertTrue(ready["ready"])
        self.assertIn("kubectl", calls[0])

    def test_node_inventory_never_silently_turns_probe_failure_into_empty_success(self) -> None:
        failed = subprocess.CompletedProcess(["k3s"], 1, "", "connection refused")
        with patch.object(server, "run", return_value=failed):
            probe = server.node_inventory_probe(Path("/usr/local/bin/k3s"))
        self.assertFalse(probe["ok"])
        self.assertEqual(probe["classification"], "api-unreachable")
        with patch.object(server, "node_inventory_probe", return_value=probe):
            with self.assertRaises(server.RemoteError):
                server.node_inventory(Path("/usr/local/bin/k3s"))

    def test_missing_launcher_includes_sanitized_service_diagnostics(self) -> None:
        service = {"returncode": 0, "LoadState": "loaded", "ActiveState": "active", "SubState": "running", "FragmentPath": "/etc/systemd/system/k3s.service"}
        with patch.object(server, "K3S_CANONICAL_BINARY_PATHS", (Path("/missing/k3s"),)), patch.object(
            server, "systemd_k3s_launcher_candidates", return_value=[]
        ), patch.object(server, "command_path", return_value=None), patch.object(
            server, "safe_k3s_service_diagnostics", return_value=service
        ):
            with self.assertRaises(server.K3sDiscoveryError) as caught:
                server.discover_k3s_launcher()
        self.assertEqual(caught.exception.diagnostics["service"]["ActiveState"], "active")
        self.assertFalse(caught.exception.diagnostics["secret_values_recorded"])
        self.assertNotIn("SUPER-SECRET", json.dumps(caught.exception.diagnostics))

    def test_privileged_launcher_discovery_rejects_writable_executable(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            binary = self.make_launcher(Path(tmp))
            binary.chmod(0o775)
            with patch.object(server.os, "geteuid", return_value=0):
                with self.assertRaises(server.RemoteError):
                    server.validated_root_executable(binary)

    def test_agent_token_symlink_to_server_token_is_supported_and_reconstructed_by_descriptor(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            token = root / "token"
            token.write_text("K10fixture::0123456789abcdef0123456789abcdef\n")
            token.chmod(0o600)
            agent = root / "agent-token"
            agent.symlink_to("token")
            old = (server.SERVER_TOKEN, server.AGENT_TOKEN)
            server.SERVER_TOKEN, server.AGENT_TOKEN = token, agent
            try:
                classified = server.classify_agent_token()
                self.assertTrue(classified["supported"])
                self.assertEqual(classified["classification"], "symlink-to-server-token")
                payload = root / "payload"
                # This test exercises the agent-token layout, not host ownership.
                # Production still requires the server token to be root-owned; mock
                # that independently so the suite is valid under a non-root runner.
                with patch.object(
                    server,
                    "classify_server_token",
                    return_value={"supported": True, "classification": "regular"},
                ):
                    descriptor = server.stage_token_material(payload)
            finally:
                server.SERVER_TOKEN, server.AGENT_TOKEN = old
            self.assertEqual(descriptor["restore"], "symlink-to-token")
            self.assertFalse((payload / "server/agent-token").exists())
            self.assertEqual((payload / "server/token").read_text(), token.read_text())

    def test_agent_token_symlink_to_unrelated_file_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            token = root / "token"; token.write_text("x" * 32); token.chmod(0o600)
            other = root / "other"; other.write_text("y" * 32); other.chmod(0o600)
            agent = root / "agent-token"; agent.symlink_to(other)
            old = (server.SERVER_TOKEN, server.AGENT_TOKEN)
            server.SERVER_TOKEN, server.AGENT_TOKEN = token, agent
            try:
                classified = server.classify_agent_token()
            finally:
                server.SERVER_TOKEN, server.AGENT_TOKEN = old
            self.assertFalse(classified["supported"])
            self.assertEqual(classified["classification"], "symlink-target-not-server-token")

    def test_backup_manifest_never_contains_token_value_and_keeps_launcher_runtime_separate(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            stage = Path(tmp)
            token = stage / "payload/server/token"
            token.parent.mkdir(parents=True)
            token.write_text("SECRET-TOKEN-VALUE")
            token.chmod(0o600)
            state = live_state()
            descriptor = {"classification": "symlink-to-server-token", "restore": "symlink-to-token", "link_target": "token", "payload_path": None}
            manifest = server.backup_manifest(stage, "tx", "prechange", {"type": "sqlite"}, state, descriptor)
            rendered = json.dumps(manifest)
            self.assertNotIn("SECRET-TOKEN-VALUE", rendered)
            self.assertEqual(manifest["purpose"], "prechange")
            self.assertEqual(manifest["k3s"]["launcher"]["sha256"], "a" * 64)
            self.assertNotEqual(manifest["k3s"]["launcher"]["sha256"], manifest["k3s"]["runtime"]["sha256"])

    def test_live_receipt_fixture_plans_schema2_without_conflating_binaries(self) -> None:
        state = live_state()
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp)
            old_paths = (server.MANAGED_CONFIG, server.CONTRACT_FILE, server.VERSION_LOCK_FILE)
            server.MANAGED_CONFIG = root / "config.yaml.d/90-overdeck-control-plane.yaml"
            server.CONTRACT_FILE = root / "overdeck/control-plane.json"
            server.VERSION_LOCK_FILE = root / "overdeck/version-lock.json"
            try:
                with patch.object(server, "inspect_state", return_value=state):
                    config, contract_bytes, lock_bytes, _ = server.planned_documents(
                        "https://100.101.104.41:6443",
                        ["100.101.104.41", "debian3", "debian3.taild2daa0.ts.net"],
                    )
                contract = json.loads(contract_bytes)
                lock = json.loads(lock_bytes)
                self.assertEqual(contract["schema_version"], 2)
                self.assertEqual(contract["qualification"]["agent_token_layout"], "symlink-to-server-token")
                self.assertEqual(lock["launcher"]["invocation_path"], "/usr/local/bin/k3s")
                self.assertIn("/var/lib/rancher/k3s/data/", lock["runtime"]["resolved_path"])
                self.assertEqual(lock["launcher"]["release_url"], "https://github.com/k3s-io/k3s/releases/download/v1.36.3+k3s1/k3s")
                self.assertIn(b"tls-san+:", config)
                for path, payload in zip([server.MANAGED_CONFIG, server.CONTRACT_FILE, server.VERSION_LOCK_FILE], [config, contract_bytes, lock_bytes], strict=True):
                    path.parent.mkdir(parents=True, exist_ok=True); path.write_bytes(payload)
                with patch.object(server, "inspect_state", return_value=state):
                    second = server.planned_documents("https://100.101.104.41:6443", ["100.101.104.41", "debian3", "debian3.taild2daa0.ts.net"])
                self.assertEqual((config, contract_bytes, lock_bytes), second[:3])
                drift = copy.deepcopy(state)
                drift["k3s"]["launcher"]["sha256"] = "d" * 64
                with patch.object(server, "inspect_state", return_value=drift):
                    with self.assertRaisesRegex(server.RemoteError, "version/launcher"):
                        server.planned_documents("https://100.101.104.41:6443", ["100.101.104.41"])
            finally:
                server.MANAGED_CONFIG, server.CONTRACT_FILE, server.VERSION_LOCK_FILE = old_paths

    def test_managed_config_is_deterministic_and_validates_inputs(self) -> None:
        payload = server.managed_config_content(["100.0.0.1", "Node.ts.net.", "100.0.0.1"]).decode()
        self.assertEqual(payload.count('"100.0.0.1"'), 1)
        self.assertIn('"node.ts.net"', payload)
        with self.assertRaises(server.RemoteError):
            server.normalize_endpoint("https://node:6443/readyz")
        with self.assertRaises(server.RemoteError):
            server.normalize_san("node;touch-/tmp/pwn")

    def test_sqlite_backup_creates_integrity_checked_copy(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); db_dir = root / "db"; db_dir.mkdir(); source = db_dir / "state.db"
            connection = sqlite3.connect(source); connection.execute("create table proof(value text not null)"); connection.execute("insert into proof values ('ok')"); connection.commit(); connection.close()
            old = server.DB_DIR; server.DB_DIR = db_dir
            try:
                destination = root / "backup"; result = server.sqlite_backup(destination)
            finally:
                server.DB_DIR = old
            self.assertEqual(result["integrity_check"], "ok")
            restored = sqlite3.connect(destination / "state.db")
            try: self.assertEqual(restored.execute("select value from proof").fetchone()[0], "ok")
            finally: restored.close()

    def test_detect_datastore_fails_closed_for_external_endpoint(self) -> None:
        self.assertEqual(server.detect_datastore(["k3s", "server", "--datastore-endpoint", "<redacted>"], []), "external")
        self.assertEqual(server.detect_datastore([], ["K3S_DATASTORE_ENDPOINT"]), "external")

    def test_secret_file_inventory_never_fingerprints_contents(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            token = Path(tmp) / "token"; token.write_text("SECRET-TOKEN-VALUE")
            metadata = server.secret_file_metadata(token)
            self.assertNotIn("sha256", metadata)
            self.assertFalse(metadata["content_hash_recorded"])

    def test_overlapping_control_plane_transaction_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            old_root = server.TRANSACTION_ROOT; server.TRANSACTION_ROOT = Path(tmp)
            try:
                tx = server.TRANSACTION_ROOT / "existing"; tx.mkdir(); (tx / "transaction.json").write_text(json.dumps({"schema_version": 1, "transaction_id": "existing", "status": "converged"}))
                with self.assertRaises(server.RemoteError): server.create_transaction("new", [], {})
            finally: server.TRANSACTION_ROOT = old_root

    def test_certificate_san_gate_fails_closed(self) -> None:
        state = {"serving_certificate": {"present": True, "sans": ["100.0.0.3", "debian3.example.ts.net"]}}
        server.require_certificate_sans(state, ["100.0.0.3"])
        with self.assertRaises(server.RemoteError): server.require_certificate_sans(state, ["missing.example.ts.net"])

    def test_canonical_json_schema_and_managed_symlink_fail_closed(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); canonical = root / "canonical.json"; canonical.write_text(json.dumps({"schema_version": 3}))
            with self.assertRaises(server.RemoteError): server.read_canonical_json(canonical)
            target = root / "target"; target.write_text("managed"); link = root / "managed-link"; link.symlink_to(target)
            with self.assertRaises(server.RemoteError): server.content_status(link, b"managed")

    def test_control_plane_transaction_finalize_is_resumable_and_irreversible(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); old_root = server.TRANSACTION_ROOT; server.TRANSACTION_ROOT = root / "transactions"; managed = root / "managed"; managed.write_text("before")
            try:
                tx_dir = server.create_transaction("phase1-test", [managed], {"proof": True}); _, document = server.load_transaction("phase1-test"); server.update_transaction(tx_dir, document, "converged")
                self.assertEqual(server.finalize_transaction("phase1-test")["status"], "finalized")
                self.assertFalse((tx_dir / "rollback").exists())
                self.assertTrue(server.finalize_transaction("phase1-test")["idempotent"])
                with self.assertRaises(server.RemoteError): server.rollback_transaction("phase1-test", "too late")
            finally: server.TRANSACTION_ROOT = old_root

    def test_finalizing_transaction_can_resume_but_not_rollback(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); old_root = server.TRANSACTION_ROOT; server.TRANSACTION_ROOT = root / "transactions"; managed = root / "managed"; managed.write_text("before")
            try:
                tx_dir = server.create_transaction("phase1-resume", [managed], {}); _, document = server.load_transaction("phase1-resume"); server.update_transaction(tx_dir, document, "finalizing")
                with self.assertRaises(server.RemoteError): server.rollback_transaction("phase1-resume", "unsafe")
                self.assertEqual(server.finalize_transaction("phase1-resume")["status"], "finalized")
            finally: server.TRANSACTION_ROOT = old_root

    def test_stale_backup_cleanup_only_removes_phase_owned_artifacts(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); old_root = server.BACKUP_ROOT; server.BACKUP_ROOT = root
            try:
                stale_dir = root / "backup-old"; stale_dir.mkdir(); stale_tar = root / "backup-old.tar"; stale_tar.write_bytes(b"old"); unrelated = root / "operator-evidence.tar"; unrelated.write_bytes(b"keep"); recent = root / "backup-recent.tar"; recent.write_bytes(b"recent")
                os.utime(stale_dir, (1, 1)); os.utime(stale_tar, (1, 1))
                self.assertEqual(sorted(server.cleanup_stale_backup_artifacts(max_age_seconds=1)), ["backup-old", "backup-old.tar"])
                self.assertTrue(unrelated.exists()); self.assertTrue(recent.exists())
            finally: server.BACKUP_ROOT = old_root

    def test_backup_tree_rejects_all_symlinks(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            root = Path(tmp); source = root / "source"; source.mkdir(); (source / "regular").write_text("ok"); (source / "link").symlink_to(source / "regular")
            with self.assertRaises(server.RemoteError): server.copy_regular_tree(source, root / "destination")


if __name__ == "__main__":
    unittest.main()