JWeBot/tests/test_kernel_release.py

#!/usr/bin/env python3

import gzip
import hashlib
import io
import os
import sys
import tarfile
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
    ReleaseError,
    ReleaseRecord,
    build_release,
    parse_manifest,
    verify_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) -> Path:
    root.mkdir()
    (root / "src").mkdir()
    (root / "plugin.toml").write_text(
        'id = "hello-command"\n',
        encoding="utf-8",
    )
    (root / "README.md").write_text("# hello\n", encoding="utf-8")
    (root / "src" / "hello_command.py").write_text(
        "\n".join(
            (
                "from jwebot_sdk import ExtensionResult, ReplyPlan",
                "",
                "def handle(event, arguments):",
                "    return ExtensionResult(reply=ReplyPlan.text('hello'))",
            )
        ),
        encoding="utf-8",
    )
    return root


class ReleaseTests(unittest.TestCase):
    def test_release_is_deterministic_across_mtime_changes(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            source = extension(base / "extension")
            first, _, first_record = build_release(
                source,
                base / "first",
                manifest=manifest(),
            )
            os.utime(source / "src" / "hello_command.py", (123456789, 123456789))
            second, _, second_record = build_release(
                source,
                base / "second",
                manifest=manifest(),
            )
            self.assertEqual(first.read_bytes(), second.read_bytes())
            self.assertEqual(first_record.source_digest, second_record.source_digest)
            self.assertEqual(
                first_record.artifact_sha256,
                second_record.artifact_sha256,
            )

    def test_built_release_verifies_without_extraction(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            artifact, _, record = build_release(
                extension(base / "extension"),
                base / "output",
                manifest=manifest(),
            )
            verify_release(artifact, record)
            self.assertEqual(record.author, "tester")
            self.assertEqual(
                record.files,
                ("README.md", "plugin.toml", "src/hello_command.py"),
            )

    def test_artifact_tampering_is_rejected(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            artifact, _, record = build_release(
                extension(base / "extension"),
                base / "output",
                manifest=manifest(),
            )
            content = bytearray(artifact.read_bytes())
            content[-1] ^= 0x01
            artifact.write_bytes(content)
            with self.assertRaisesRegex(ReleaseError, "digest mismatch"):
                verify_release(artifact, record)

    def test_release_record_rejects_invalid_author(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            _, _, record = build_release(
                extension(base / "extension"),
                base / "output",
                manifest=manifest(),
            )
            value = record.to_mapping()
            value["author"] = "../tester"
            with self.assertRaisesRegex(ReleaseError, "invalid release record"):
                ReleaseRecord.from_mapping(value)

    def test_unsupported_source_file_is_rejected(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            source = extension(base / "extension")
            (source / "src" / "payload.bin").write_bytes(b"payload")
            with self.assertRaisesRegex(ReleaseError, "unsupported release file"):
                build_release(
                    source,
                    base / "output",
                    manifest=manifest(),
                )

    def test_interpreter_cache_is_excluded_from_release(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            source = extension(base / "extension")
            cache = source / "src" / "__pycache__"
            cache.mkdir()
            (cache / "hello_command.cpython-313.pyc").write_bytes(b"cache")

            _, _, record = build_release(
                source,
                base / "output",
                manifest=manifest(),
            )

            self.assertEqual(
                record.files,
                ("README.md", "plugin.toml", "src/hello_command.py"),
            )

    def test_immutable_artifact_cannot_be_replaced(self):
        with tempfile.TemporaryDirectory() as directory:
            base = Path(directory)
            source = extension(base / "extension")
            artifact, _, _ = build_release(
                source,
                base / "output",
                manifest=manifest(),
            )
            artifact.write_bytes(b"tampered")
            with self.assertRaisesRegex(ReleaseError, "immutable release"):
                build_release(
                    source,
                    base / "output",
                    manifest=manifest(),
                )

    def test_unsafe_archive_member_is_rejected(self):
        with tempfile.TemporaryDirectory() as directory:
            artifact = Path(directory) / "unsafe.tar.gz"
            with artifact.open("wb") as raw:
                with gzip.GzipFile(
                    filename="",
                    mode="wb",
                    fileobj=raw,
                    mtime=0,
                ) as compressed:
                    with tarfile.open(fileobj=compressed, mode="w") as archive:
                        content = b"unsafe"
                        info = tarfile.TarInfo("plugin/../outside.py")
                        info.size = len(content)
                        archive.addfile(info, io.BytesIO(content))
            artifact_bytes = artifact.read_bytes()
            record = ReleaseRecord(
                plugin_id="hello-command",
                author="tester",
                plugin_version="1.0.0",
                api_version="1",
                source_digest="0" * 64,
                artifact_name=artifact.name,
                artifact_sha256=hashlib.sha256(artifact_bytes).hexdigest(),
                artifact_size=len(artifact_bytes),
                files=("outside.py",),
            )
            with self.assertRaisesRegex(ReleaseError, "unsafe member"):
                verify_release(artifact, record)


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