JWeBot/tests/test_kernel_activation.py

#!/usr/bin/env python3

import os
import sys
import tempfile
import unittest
from pathlib import Path


PROJECT_ROOT = Path(__file__).resolve().parents[1]
SDK_SRC = PROJECT_ROOT / "packages" / "jwebot-sdk" / "src"
KERNEL_SRC = PROJECT_ROOT / "packages" / "jwebot-kernel" / "src"
sys.path[:0] = [str(SDK_SRC), str(KERNEL_SRC)]

from jwebot_kernel import (  # noqa: E402
    ActivationError,
    ReleaseError,
    build_release,
    install_release,
    parse_manifest,
    rollback_release,
)


def manifest():
    return parse_manifest(
        {
            "id": "hello-command",
            "author": "tester",
            "description": "Say hello",
            "help": ["/hello"],
            "version": "1.0.0",
            "api_version": "1",
            "entrypoint": "hello_command:plugin",
            "type": "command",
            "capabilities": {
                "commands": ["/hello"],
                "storage": False,
                "network_profiles": [],
            },
        }
    )


def extension(root: Path, reply: str) -> Path:
    root.mkdir()
    (root / "src").mkdir()
    (root / "plugin.toml").write_text(
        'id = "hello-command"\n',
        encoding="utf-8",
    )
    (root / "src" / "hello_command.py").write_text(
        "\n".join(
            (
                "from jwebot_sdk import ExtensionResult, ReplyPlan",
                "",
                "def handle(event, arguments):",
                f"    return ExtensionResult(reply=ReplyPlan.text({reply!r}))",
            )
        ),
        encoding="utf-8",
    )
    return root


class ActivationTests(unittest.TestCase):
    def build(self, base: Path, name: str, reply: str):
        return build_release(
            extension(base / name, reply),
            base / f"{name}-dist",
            manifest=manifest(),
        )

    def test_install_creates_verified_release_and_current_pointer(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            artifact, record_path, record = self.build(base, "first", "one")
            release_dir = install_release(
                artifact,
                record_path,
                base / "state",
                allow_unsigned=True,
            )
            self.assertEqual(release_dir.name, record.source_digest)
            self.assertEqual(
                os.readlink(base / "state" / "current"),
                f"releases/{record.source_digest}",
            )
            self.assertTrue((release_dir / ".release.json").is_file())
            self.assertEqual(
                (release_dir / "src" / "hello_command.py").stat().st_mode & 0o777,
                0o600,
            )

    def test_install_rejects_unsigned_release_by_default(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            artifact, record_path, _ = self.build(base, "first", "one")
            state = base / "state"
            with self.assertRaisesRegex(
                ActivationError,
                "signed release activation is required",
            ):
                install_release(artifact, record_path, state)
            self.assertFalse(state.exists())

    def test_unsigned_local_install_rejects_publisher_policy(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            artifact, record_path, _ = self.build(base, "first", "one")
            with self.assertRaisesRegex(
                ActivationError,
                "cannot include signature inputs",
            ):
                install_release(
                    artifact,
                    record_path,
                    base / "state",
                    expected_publisher="test-publisher",
                    allow_unsigned=True,
                )

    def test_second_install_preserves_previous_and_rollback_swaps(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            first_artifact, first_record_path, first_record = self.build(
                base,
                "first",
                "one",
            )
            second_artifact, second_record_path, second_record = self.build(
                base,
                "second",
                "two",
            )
            state = base / "state"
            install_release(
                first_artifact,
                first_record_path,
                state,
                allow_unsigned=True,
            )
            install_release(
                second_artifact,
                second_record_path,
                state,
                allow_unsigned=True,
            )
            self.assertEqual(
                os.readlink(state / "previous"),
                f"releases/{first_record.source_digest}",
            )
            with self.assertRaisesRegex(
                ActivationError,
                "signed release rollback is required",
            ):
                rollback_release(state)
            rollback_release(state, allow_unsigned=True)
            self.assertEqual(
                os.readlink(state / "current"),
                f"releases/{first_record.source_digest}",
            )
            self.assertEqual(
                os.readlink(state / "previous"),
                f"releases/{second_record.source_digest}",
            )

    def test_tampered_artifact_does_not_create_activation_state(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            artifact, record_path, _ = self.build(base, "first", "one")
            artifact.write_bytes(artifact.read_bytes() + b"tampered")
            state = base / "state"
            with self.assertRaises(ReleaseError):
                install_release(
                    artifact,
                    record_path,
                    state,
                    allow_unsigned=True,
                )
            self.assertFalse((state / "current").exists())

    def test_existing_release_content_is_reverified(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            artifact, record_path, _ = self.build(base, "first", "one")
            state = base / "state"
            release_dir = install_release(
                artifact,
                record_path,
                state,
                allow_unsigned=True,
            )
            (release_dir / "src" / "hello_command.py").write_text(
                "tampered\n",
                encoding="utf-8",
            )
            with self.assertRaisesRegex(ActivationError, "source digest mismatch"):
                install_release(
                    artifact,
                    record_path,
                    state,
                    allow_unsigned=True,
                )

    def test_rollback_requires_a_previous_verified_release(self):
        with tempfile.TemporaryDirectory() as directory:
            state = Path(directory) / "state"
            with self.assertRaisesRegex(ActivationError, "no verified previous"):
                rollback_release(state)

    def test_unsafe_current_pointer_is_rejected(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            artifact, record_path, _ = self.build(base, "first", "one")
            state = base / "state"
            state.mkdir()
            (state / "current").symlink_to("/tmp")
            with self.assertRaisesRegex(ActivationError, "unsafe target"):
                install_release(
                    artifact,
                    record_path,
                    state,
                    allow_unsigned=True,
                )


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