tests/test_support_bundle.py

from __future__ import annotations

import copy
import hashlib
import importlib.util
import io
import json
import os
import socket
import stat
import tempfile
import unittest
from contextlib import redirect_stderr
from pathlib import Path
from unittest import mock

from lumi_eggcracker import support_bundle
from lumi_eggcracker.jsonio import JsonInputError

ROOT = Path(__file__).resolve().parents[1]
SCRIPT_SPEC = importlib.util.spec_from_file_location(
    "support_bundle_script", ROOT / "scripts" / "support_bundle.py"
)
if SCRIPT_SPEC is None or SCRIPT_SPEC.loader is None:
    raise RuntimeError("cannot load support-bundle script")
support_bundle_script = importlib.util.module_from_spec(SCRIPT_SPEC)
SCRIPT_SPEC.loader.exec_module(support_bundle_script)


class SupportBundleTests(unittest.TestCase):
    @staticmethod
    def host() -> dict[str, object]:
        return {
            "platform": "TestOS",
            "kernel": "test-kernel",
            "machine": "test-machine",
            "python": "3.11.0",
            "cgroup_v2": False,
            "controllers": [],
            "pidfd_open": False,
            "pidfd_send_signal": False,
            "systemd": None,
        }

    @unittest.skipUnless(os.name == "posix", "installed helper is native-Linux behavior")
    def test_release_helper_delegates_to_installed_zipapp(self) -> None:
        installed = Path("/usr/local/lib/lumi-eggcracker/lumi-eggcracker.pyz")
        with (
            mock.patch.object(support_bundle_script.os, "geteuid", return_value=0),
            mock.patch.object(
                support_bundle_script,
                "validated_installed_app",
                return_value=installed,
            ),
            mock.patch.object(
                support_bundle_script.os,
                "execv",
                side_effect=OSError("sentinel"),
            ) as execute,
            self.assertRaisesRegex(OSError, "sentinel"),
        ):
            support_bundle_script.main(["--output", "/tmp/support.json"])
        execute.assert_called_once_with(
            "/usr/bin/python3",
            [
                "/usr/bin/python3",
                "-I",
                "-S",
                str(installed),
                "support-bundle",
                "--output",
                "/tmp/support.json",
            ],
        )

    @staticmethod
    def query(action: str, **_args: object) -> dict[str, object]:
        if action == "doctor":
            return {
                "result": "PASS",
                "backend": "root-supervisor",
                "version": "0.6.0",
                "workload_uid": 997,
                "autonomous_discovery": True,
                "cgroup_v2": True,
                "pidfd": True,
                "execution_boundary": {
                    "linux": True,
                    "architecture": True,
                    "fcntl": True,
                    "libc": True,
                    "user_notification": True,
                    "supported": True,
                },
                "installation": {
                    "state": "HEALTHY",
                    "journal": False,
                    "files_match": True,
                    "manifest_version": "1.0.10",
                },
                "incidents": {
                    "healthy": True,
                    "count": 2,
                    "active": 1,
                    "lockdown": True,
                },
                "network": {
                    "mode": "offline",
                    "cleanup_healthy": True,
                    "primitives": {"supported": True},
                },
                "discovery": {
                    "healthy": True,
                    "consecutive_failures": 0,
                    "last_scan_duration_ms": 2.0,
                    "last_scan_completed": True,
                    "receipt_persistence_healthy": True,
                    "private_path": "/secret",
                },
            }
        if action == "detections":
            return {
                "detections": [
                    {
                        "event_id": "SENTINEL-EVENT-ID",
                        "result": "TERMINATED",
                        "trigger": "UNAPPROVED_AI",
                        "version": "0.6.0",
                        "detector": {"profile": "content.gguf-llama", "path": "/private/model.gguf"},
                        "boundary": {
                            "address_family": "inet",
                            "mode": "offline",
                            "policy_sha256": "a" * 64,
                            "violation": "DENIED",
                        },
                        "argv": ["--secret"],
                    }
                ]
            }
        if action == "list":
            return {"runs": [{"name": "private-name", "state": "TERMINATED", "unit": "/private"}]}
        if action == "incidents":
            return {"incidents": [{"state": "ACTIVE"}, {"state": "CLOSED"}]}
        raise AssertionError(action)

    def test_collect_is_redacted_and_bounded(self) -> None:
        with mock.patch.object(support_bundle, "_host", return_value=self.host()):
            value = support_bundle.collect(self.query)
        text = str(value)
        self.assertNotIn("/private", text)
        self.assertNotIn("--secret", text)
        self.assertEqual(1, len(value["receipts"]))
        self.assertEqual({"TERMINATED": 1}, value["workloads"]["states"])
        self.assertFalse(value["privacy"]["raw_receipts"])
        self.assertEqual(1, value["incidents"]["active"])
        self.assertEqual({"ACTIVE": 1, "CLOSED": 1}, value["incidents"]["states"])

    def test_validate_returns_only_aggregate_summary_without_modifying_input(self) -> None:
        with mock.patch.object(support_bundle, "_host", return_value=self.host()):
            value = support_bundle.collect(self.query)
        with tempfile.TemporaryDirectory() as raw:
            path = Path(raw) / "bundle.json"
            path.write_text(json.dumps(value), encoding="utf-8")
            original = path.read_bytes()
            digest = hashlib.sha256(original).hexdigest()
            with (
                mock.patch.object(support_bundle, "collect") as collect,
                mock.patch.object(support_bundle, "request") as request,
                mock.patch.object(support_bundle, "_host") as host,
                mock.patch.object(support_bundle.subprocess, "run") as run,
                mock.patch.object(socket, "socket") as create_socket,
            ):
                summary = support_bundle.validate_bundle(path)
            collect.assert_not_called()
            request.assert_not_called()
            host.assert_not_called()
            run.assert_not_called()
            create_socket.assert_not_called()
            self.assertEqual(
                {
                    "result": "STRUCTURE_VALID",
                    "schema_version": "lumi-eggcracker.support-bundle.v1",
                    "receipts": {
                        "included": 1,
                        "maximum_included": 100,
                        "total_represented": False,
                    },
                    "workloads": {"run_count": 1, "reported_state_count": 1},
                    "incidents": {
                        "count": 2,
                        "active_count": 1,
                        "reported_state_count": 2,
                    },
                    "limitations": (
                        "Structural validation only; not authentication, privacy assurance, "
                        "completeness, containment, or incident truth."
                    ),
                },
                summary,
            )
            serialized = json.dumps(summary, sort_keys=True)
            for value_to_hide in (
                "SENTINEL-EVENT-ID",
                "content.gguf-llama",
                "root-supervisor",
                "/private",
                "--secret",
                "997",
            ):
                self.assertNotIn(value_to_hide, serialized)
            self.assertEqual(digest, hashlib.sha256(path.read_bytes()).hexdigest())
            self.assertEqual(original, path.read_bytes())

    def test_validate_uses_nonblocking_open_after_mocked_fifo_replacement(self) -> None:
        with tempfile.TemporaryDirectory() as raw:
            path = Path(raw) / "bundle.json"
            path.write_text("{}", encoding="utf-8")
            fake_descriptor = 937
            nonblock_flag = getattr(os, "O_NONBLOCK", 1 << 30)
            fifo_stat = os.stat_result(
                (stat.S_IFIFO | 0o600, 1, 1, 1, 1, 1, 0, 0, 0, 0)
            )
            with (
                mock.patch.object(
                    support_bundle.os, "open", return_value=fake_descriptor
                ) as open_file,
                mock.patch.object(support_bundle.os, "fstat", return_value=fifo_stat),
                mock.patch.object(support_bundle.os, "close") as close_file,
                mock.patch.object(support_bundle.os, "fdopen") as fdopen,
                mock.patch.object(
                    support_bundle.os,
                    "O_NONBLOCK",
                    nonblock_flag,
                    create=True,
                ),
                self.assertRaises(JsonInputError),
            ):
                support_bundle.validate_bundle(path)

            open_file.assert_called_once()
            self.assertEqual(path, open_file.call_args.args[0])
            self.assertTrue(open_file.call_args.args[1] & nonblock_flag)
            close_file.assert_called_once_with(fake_descriptor)
            fdopen.assert_not_called()

    def test_validate_rejects_numeric_privacy_values(self) -> None:
        with mock.patch.object(support_bundle, "_host", return_value=self.host()):
            bundle = support_bundle.collect(self.query)
        privacy_flags = ("raw_receipts", "argv", "paths", "pids", "model_data")
        substitutions = [
            (name, value)
            for name in privacy_flags + ("network",)
            for value in (0, 0.0)
        ]
        with tempfile.TemporaryDirectory() as raw:
            path = Path(raw) / "bundle.json"
            for name, value in substitutions:
                with self.subTest(name=name, value=value):
                    invalid = copy.deepcopy(bundle)
                    invalid["privacy"][name] = value
                    path.write_text(json.dumps(invalid), encoding="utf-8")
                    with self.assertRaises(JsonInputError):
                        support_bundle.validate_bundle(path)

    def test_validate_rejects_injected_fields_duplicate_keys_bad_types_and_unbounded_json(self) -> None:
        with mock.patch.object(support_bundle, "_host", return_value=self.host()):
            bundle = support_bundle.collect(self.query)
        with tempfile.TemporaryDirectory() as raw:
            path = Path(raw) / "must-not-echo-private-input.json"

            def rejected(content: str | bytes, *secrets: str) -> str:
                if isinstance(content, bytes):
                    path.write_bytes(content)
                else:
                    path.write_text(content, encoding="utf-8")
                errors = io.StringIO()
                with redirect_stderr(errors):
                    self.assertEqual(4, support_bundle.validate_main([str(path)]))
                self.assertNotIn(str(path), errors.getvalue())
                for secret in secrets:
                    self.assertNotIn(secret, errors.getvalue())
                return errors.getvalue()

            for field in ("argv", "path", "pid", "credential", "model_data"):
                injected = copy.deepcopy(bundle)
                injected[field] = "SENTINEL-PRIVATE-VALUE"
                rejected(json.dumps(injected), field, "SENTINEL-PRIVATE-VALUE")

            raw_json = json.dumps(bundle)
            duplicate = raw_json.replace(
                '"schema_version":',
                '"schema_version":"attacker", "schema_version":',
                1,
            )
            rejected(duplicate, "attacker")

            wrong_type = copy.deepcopy(bundle)
            wrong_type["health"]["workload_uid"] = True
            rejected(json.dumps(wrong_type), "true")

            nonfinite = copy.deepcopy(bundle)
            nonfinite["health"]["discovery"]["last_scan_duration_ms"] = float("inf")
            rejected(json.dumps(nonfinite), "Infinity")

            rejected("[" * 40 + "0" + "]" * 40)
            rejected(b"x" * (support_bundle.MAX_VALIDATION_BYTES + 1))

    @unittest.skipUnless(os.name == "posix", "atomic directory fsync is native-Linux behavior")
    def test_write_requires_new_output(self) -> None:
        with tempfile.TemporaryDirectory() as raw:
            destination = Path(raw) / "support.json"
            with mock.patch.object(support_bundle, "_host", return_value=self.host()):
                support_bundle.write_bundle(destination, self.query)
            with (
                self.assertRaises(JsonInputError),
                mock.patch.object(support_bundle, "_host", return_value=self.host()),
            ):
                support_bundle.write_bundle(destination, self.query)


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