Packer

xcore_agent.packer.builder — build, encrypt, and sign a .xdeploy artifact (tar → zstd → AES-256-GCM → Ed25519).

xcore_agent.packer.builder

Builds .xdeploy artifacts.

This is the build-side counterpart to agent.pipeline.DeploymentRunner, which consumes exactly the format produced here: tar the project, zstd -compress it, AES-256-GCM encrypt it, and sign the ciphertext with Ed25519. Both sides call the same crypto.compute_tree_digest to produce and to re-verify manifest.json's content_sha256, so there is no protocol drift between "what the packer hashed" and "what the agent re-hashes".

MANIFEST_FILENAME = 'manifest.json' module-attribute

INSTALL_PLAN_PATH = 'deployment/install.yaml' module-attribute

EXTENSION_MANIFEST_FILENAMES = ('service.yaml', 'extension.yaml') module-attribute

BuildError

Bases: Exception

Raised when a source tree cannot be turned into a valid .xdeploy artifact.

Source code in xcore_agent/packer/builder.py

class BuildError(Exception):
    """Raised when a source tree cannot be turned into a valid .xdeploy artifact."""

BuildResult dataclass

Source code in xcore_agent/packer/builder.py

@dataclass(frozen=True)
class BuildResult:
    output_path: Path
    manifest: ProjectManifest
    # Per-artifact DEK. The packer generates it but never stores it — the
    # caller (a future build-engine talking to XCore Hub) is responsible for
    # handing it to the Hub for KEK-wrapped storage.
    dek: bytes
    signature: bytes
    signer_public_key: bytes

output_path: Path instance-attribute

manifest: ProjectManifest instance-attribute

dek: bytes instance-attribute

signature: bytes instance-attribute

signer_public_key: bytes instance-attribute

__init__(output_path: Path, manifest: ProjectManifest, dek: bytes, signature: bytes, signer_public_key: bytes) -> None

build_artifact(source_root: Path, *, project_id: str, project_name: str, version: str, output_path: Path, signing_key: Ed25519PrivateKey | None = None) -> BuildResult

Build, encrypt, and sign a .xdeploy artifact from source_root.

source_root must already contain a plugins directory (plugins/ by default — see _read_plugins_dirname for how a project overrides that via integration.yaml's plugins.directory), integration.yaml, and deployment/install.yaml. This writes manifest.json into it (and refuses to run if one is already there — see write_manifest). Pass signing_key to sign with a specific, persisted Hub key; a fresh throwaway one is generated and returned otherwise.

A plugin/extension whose manifest declares source: gets pruned down to just its manifest file (plugin.yaml/extension.yaml) before sealing — even if the operator's local source_root happens to have the real code checked out at that path too (e.g. because they cloned it to poke around, or a prior embedded build left it there). It's resolved from git at deploy time (see plugin_resolver.py and agent.pipeline's _resolve_plugins/_resolve_extensions), so embedding it here would only bloat the artifact with a copy nothing ever reads back out of it. source_root itself is never mutated — pruning happens on a temporary copy that gets sealed and discarded.

Source code in xcore_agent/packer/builder.py

def build_artifact(
    source_root: Path,
    *,
    project_id: str,
    project_name: str,
    version: str,
    output_path: Path,
    signing_key: Ed25519PrivateKey | None = None,
) -> BuildResult:
    """Build, encrypt, and sign a `.xdeploy` artifact from `source_root`.

    `source_root` must already contain a plugins directory (`plugins/` by
    default — see `_read_plugins_dirname` for how a project overrides that
    via `integration.yaml`'s `plugins.directory`), `integration.yaml`, and
    `deployment/install.yaml`. This writes `manifest.json` into it (and
    refuses to run if one is already there — see `write_manifest`). Pass
    `signing_key` to sign with a specific, persisted Hub key; a fresh
    throwaway one is generated and returned otherwise.

    A plugin/extension whose manifest declares `source:` gets pruned down
    to just its manifest file (`plugin.yaml`/`extension.yaml`) before
    sealing — even if the operator's local `source_root` happens to have
    the real code checked out at that path too (e.g. because they cloned it
    to poke around, or a prior embedded build left it there). It's
    resolved from git at deploy time (see `plugin_resolver.py` and
    `agent.pipeline`'s `_resolve_plugins`/`_resolve_extensions`), so
    embedding it here would only bloat the artifact with a copy nothing
    ever reads back out of it. `source_root` itself is never mutated —
    pruning happens on a temporary copy that gets sealed and discarded.
    """
    plugins_dirname = _read_plugins_dirname(source_root)
    plan = _load_install_plan(source_root, project_id=project_id, version=version)
    _validate_source_tree(source_root, plan=plan, plugins_dirname=plugins_dirname)
    manifest = write_manifest(
        source_root,
        project_id=project_id,
        project_name=project_name,
        version=version,
        plugins_dirname=plugins_dirname,
        plan=plan,
    )
    with tempfile.TemporaryDirectory(prefix="xcore-agent-pack-") as tmp:
        packaging_root = Path(tmp) / "package"
        _prepare_packaging_view(source_root, packaging_root, manifest)
        ciphertext, dek, signature, signer_public_key = seal_directory(
            packaging_root, signing_key=signing_key
        )

    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_bytes(ciphertext)

    return BuildResult(
        output_path=output_path,
        manifest=manifest,
        dek=dek,
        signature=signature,
        signer_public_key=signer_public_key,
    )

write_manifest(source_root: Path, *, project_id: str, project_name: str, version: str, plugins_dirname: str = 'plugins', plan: InstallPlan | None = None) -> ProjectManifest

Compute per-plugin and whole-tree content hashes and write manifest.json into source_root. Refuses to overwrite an existing one: the manifest is always generated fresh from the current tree, never hand-edited, so a leftover one is almost certainly stale.

plan is deployment/install.yaml, already parsed — pass it when caller already has one (build_artifact does, from _load_install_ plan) to avoid re-parsing; re-read from source_root here otherwise (e.g. a test calling write_manifest directly) if the file exists, falling back to no install-plan sources if it doesn't. Its steps' source: (see InstallPluginStep/InstallExtensionStep) is checked before a plugin's own plugin.yaml source: and the xcli-written registry — see _install_plan_plugin_sources's docstring for why.

Source code in xcore_agent/packer/builder.py

def write_manifest(
    source_root: Path,
    *,
    project_id: str,
    project_name: str,
    version: str,
    plugins_dirname: str = "plugins",
    plan: InstallPlan | None = None,
) -> ProjectManifest:
    """Compute per-plugin and whole-tree content hashes and write
    `manifest.json` into `source_root`. Refuses to overwrite an existing one:
    the manifest is always generated fresh from the current tree, never
    hand-edited, so a leftover one is almost certainly stale.

    `plan` is `deployment/install.yaml`, already parsed — pass it when
    caller already has one (`build_artifact` does, from `_load_install_
    plan`) to avoid re-parsing; re-read from `source_root` here otherwise
    (e.g. a test calling `write_manifest` directly) if the file exists,
    falling back to no install-plan sources if it doesn't. Its steps'
    `source:` (see `InstallPluginStep`/`InstallExtensionStep`) is checked
    before a plugin's own plugin.yaml `source:` and the xcli-written
    registry — see `_install_plan_plugin_sources`'s docstring for why."""
    manifest_path = source_root / MANIFEST_FILENAME
    if manifest_path.exists():
        raise BuildError(
            f"{MANIFEST_FILENAME} already exists in {source_root} — remove it, "
            "the packer always regenerates it from the current tree"
        )

    if plan is None:
        install_path = source_root / INSTALL_PLAN_PATH
        if install_path.is_file():
            plan = InstallPlan.model_validate(yaml.safe_load(install_path.read_text()))
    plugin_sources = _install_plan_plugin_sources(plan)
    extension_sources = _install_plan_extension_sources(plan)

    # Files that will be pruned away by `_prepare_packaging_view` (source-
    # based plugins/extensions get reduced to just their manifest file) —
    # `content_sha256` below must exclude these too, since it's computed on
    # `source_root` before that pruning happens. See `_non_manifest_
    # relpaths`'s docstring for why this matters.
    pruned_relpaths: set[str] = set()

    plugins_dir = source_root / plugins_dirname
    plugin_refs = []
    for plugin_dir in sorted(p for p in plugins_dir.iterdir() if p.is_dir()):
        plugin_yaml = plugin_dir / "plugin.yaml"
        if not plugin_yaml.is_file():
            raise BuildError(f"plugin {plugin_dir.name!r} is missing plugin.yaml")
        source = (
            plugin_sources.get(plugin_dir.name)
            or _read_plugin_source(plugin_yaml)
            or _read_registry_source(plugin_dir)
        )
        if source is None:
            _check_env_template_present(plugin_yaml, plugin_dir)
        else:
            pruned_relpaths.update(
                _non_manifest_relpaths(plugin_dir, "plugin.yaml", source_root=source_root)
            )
        plugin_refs.append(
            PluginRef(
                id=plugin_dir.name,
                version=_read_plugin_version(plugin_yaml),
                # A source-based plugin's code isn't necessarily embedded
                # here (it may be resolved from git at deploy time — see
                # plugin_resolver.py), so there's nothing to hash at build
                # time. An embedded plugin is always hashed: sha256 is what
                # the agent re-verifies post-extraction.
                sha256=None if source is not None else crypto.compute_tree_digest(plugin_dir),
                environment=_read_plugin_environment(plugin_yaml),
                source=source,
            )
        )
    if not plugin_refs:
        raise BuildError(f"no plugins found under {plugins_dirname}/")

    extension_refs = []
    extensions_dir = source_root / "extensions"
    if extensions_dir.is_dir():
        for extension_dir in sorted(p for p in extensions_dir.iterdir() if p.is_dir()):
            # A manifest file is optional — absent means "embedded, hash
            # the whole directory" (the original, still-default behavior);
            # a source resolved (install.yaml, the manifest's own
            # `source:`, or the registry — same priority and same shared
            # .xcore-registry.json as plugins above, see _read_registry_
            # source's docstring for why registry_dir=plugins_dir is
            # required here) means "resolved at deploy time, nothing to
            # hash here".
            ext_manifest_path = _find_extension_manifest(extension_dir)
            ext_source = (
                extension_sources.get(extension_dir.name)
                or (_read_extension_source(ext_manifest_path) if ext_manifest_path else None)
                or _read_registry_source(extension_dir, registry_dir=plugins_dir)
            )
            manifest_filename = (
                ext_manifest_path.name if ext_manifest_path else EXTENSION_MANIFEST_FILENAMES[0]
            )
            if ext_source is not None:
                pruned_relpaths.update(
                    _non_manifest_relpaths(
                        extension_dir, manifest_filename, source_root=source_root
                    )
                )
            extension_refs.append(
                ExtensionRef(
                    id=extension_dir.name,
                    sha256=(
                        None
                        if ext_source is not None
                        else crypto.compute_tree_digest(extension_dir)
                    ),
                    source=ext_source,
                )
            )

    content_sha256 = crypto.compute_tree_digest(
        source_root,
        exclude=frozenset({MANIFEST_FILENAME}) | pruned_relpaths,
        skip_patterns=_PACKAGING_EXCLUDE_PATTERNS,
    )

    manifest = ProjectManifest(
        format_version="1",
        project_id=project_id,
        project_name=project_name,
        version=version,
        built_at=datetime.now(timezone.utc),
        plugins=plugin_refs,
        extensions=extension_refs,
        plugins_dirname=plugins_dirname,
        content_sha256=content_sha256,
    )
    manifest_path.write_text(manifest.model_dump_json())
    return manifest

seal_directory(source_root: Path, *, signing_key: Ed25519PrivateKey | None = None) -> tuple[bytes, bytes, bytes, bytes]

Tar, zstd-compress, AES-256-GCM encrypt, and Ed25519-sign source_root as-is. Pure packaging — does not touch or require manifest.json, so it can also be used to build a deliberately tampered artifact for tests.

Returns (ciphertext, dek, signature, signer_public_key). ciphertext is a 12-byte nonce prefix followed by the AES-256-GCM ciphertext, matching what agent.pipeline.DeploymentRunner._decrypt expects.

Source code in xcore_agent/packer/builder.py

def seal_directory(
    source_root: Path, *, signing_key: Ed25519PrivateKey | None = None
) -> tuple[bytes, bytes, bytes, bytes]:
    """Tar, zstd-compress, AES-256-GCM encrypt, and Ed25519-sign
    `source_root` as-is. Pure packaging — does not touch or require
    `manifest.json`, so it can also be used to build a deliberately
    tampered artifact for tests.

    Returns (ciphertext, dek, signature, signer_public_key). `ciphertext` is
    a 12-byte nonce prefix followed by the AES-256-GCM ciphertext, matching
    what `agent.pipeline.DeploymentRunner._decrypt` expects.
    """
    plaintext_tar = _tar_bytes(source_root)
    compressed = zstandard.ZstdCompressor(level=19).compress(plaintext_tar)

    # cryptography's own Rust-backed AESGCM.generate_key accepts bit_length
    # at runtime (verified — every build in this test suite calls this),
    # but its bundled type stub doesn't declare the kwarg, so mypy flags it
    # regardless of the installed cryptography version.
    dek = AESGCM.generate_key(bit_length=256)  # type: ignore[call-arg]
    nonce = secrets.token_bytes(12)
    ciphertext = nonce + AESGCM(dek).encrypt(nonce, compressed, None)

    key = signing_key or Ed25519PrivateKey.generate()
    signature = key.sign(ciphertext)
    signer_public_key = key.public_key().public_bytes_raw()

    return ciphertext, dek, signature, signer_public_key