"""``muse archive`` — export a snapshot as a portable archive. Creates a ``tar.gz`` or ``zip`` archive from any historical snapshot — HEAD by default. The archive contains only the tracked files (the contents of the snapshot at that point in time), making it the canonical way to distribute a specific version without exposing ``.muse/`` internals. Commit reference ---------------- ``--ref`` accepts any reference understood by ``resolve_commit_ref``: - Omitted or ``HEAD`` — the most recent commit on the current branch. - A branch name — the tip commit of that branch. - ``HEAD~N`` — *N* first-parent steps back from HEAD. - A full or abbreviated commit SHA. Formats ------- - ``tar.gz`` (default) — gzip-compressed POSIX tar. - ``zip`` — Deflate-compressed ZIP. Security model -------------- - Every archive entry name is validated by ``_safe_arcname`` before being written. Entries with ``..`` path segments, absolute paths, or null bytes are silently skipped with a warning — this prevents both zip-slip and tar-slip path-traversal attacks regardless of what is stored in a snapshot. - ``--prefix`` is validated up-front for ``..`` segments so users get a clear error before any I/O begins. - ``--output`` paths that would write outside the current directory are permitted (agents often write to ``/tmp/`` or explicit destinations), but the directory must already exist. - All user-supplied strings are sanitized via ``sanitize_display()`` before appearing in human-readable terminal output. - All error messages go to **stderr**; **stdout** carries only data. Agent UX -------- Pass ``--json`` for a machine-readable result. Pass ``--list`` to preview what would be archived without writing anything to disk — useful for agents that need to reason about snapshot contents before committing to a file. Usage:: muse archive # HEAD → .tar.gz muse archive --ref feat/audio # branch tip muse archive --ref a1b2c3d4 # commit SHA prefix muse archive --format zip # zip instead of tar.gz muse archive --output release-v1.0.zip # custom output path muse archive --prefix myproject/ # directory prefix inside archive muse archive --list # preview without writing muse archive --list --json # agent-readable manifest muse archive --json # machine-readable result JSON schema — normal output (``--json``):: { "path": "", "format": "tar.gz" | "zip", "file_count": , "bytes": , "commit_id": "", "snapshot_id": "", "message": "", "branch": "", "author": "", "agent_id": "", "model_id": "", "committed_at": "", "ref": "", "prefix": "" } JSON schema — list mode (``--list --json``):: { "commit_id": "", "snapshot_id": "", "message": "", "branch": "", "author": "", "committed_at": "", "ref": "", "prefix": "", "file_count": , "entries": [ {"path": "", "object_id": ""}, ... ] } Exit codes ---------- - 0 — success - 1 — bad arguments (bad format, bad prefix, missing commit, output dir missing) - 2 — not inside a Muse repository - 3 — internal error (snapshot or object data missing) """ import argparse import json import logging import pathlib import sys import tarfile import zipfile from typing import TypedDict from muse.core.types import split_id from muse.core.envelope import EnvelopeJson, make_envelope from muse.core.errors import ExitCode from muse.core.object_store import object_path, read_object from muse.core.repo import require_repo from muse.core.timing import start_timer from muse.core.refs import ( get_head_commit_id, read_current_branch, ) from muse.core.commits import ( read_commit, resolve_commit_ref, ) from muse.core.snapshots import read_snapshot from muse.core.validation import sanitize_display from muse.core.types import Manifest logger = logging.getLogger(__name__) _FORMAT_CHOICES = {"tar.gz", "zip"} # --------------------------------------------------------------------------- # Typed JSON schemas # --------------------------------------------------------------------------- class _ArchiveJson(EnvelopeJson): """Machine-readable output of ``muse archive --json`` (write mode). Fields ------ path Absolute or relative path to the archive file that was written. format ``"tar.gz"`` or ``"zip"``. file_count Number of files successfully written into the archive. bytes Size of the archive file on disk in bytes. commit_id Full ``sha256:…`` commit ID that was archived. snapshot_id Full ``sha256:…`` snapshot ID — the content-addressed tree at that commit. message Commit message. branch Branch that was current when the archive was created. author Author field from the commit record. agent_id Agent identity string (empty for human commits). model_id Model identifier (empty for human commits). committed_at ISO-8601 commit timestamp. ref The ``--ref`` value passed by the caller, or ``null`` for HEAD. prefix The ``--prefix`` value used (empty string if none). """ path: str format: str file_count: int bytes: int commit_id: str snapshot_id: str message: str branch: str author: str agent_id: str model_id: str committed_at: str ref: str | None prefix: str class _ListEntryJson(TypedDict): """One file entry in the ``--list --json`` output.""" path: str object_id: str class _ListJson(EnvelopeJson): """Machine-readable output of ``muse archive --list --json``. Fields ------ commit_id Full ``sha256:…`` commit ID. snapshot_id Full ``sha256:…`` snapshot ID. message Commit message. branch Current branch name. author Author field from the commit record. committed_at ISO-8601 commit timestamp. ref ``--ref`` value passed by the caller, or ``null`` for HEAD. prefix ``--prefix`` value used (empty string if none). file_count Total number of entries that would be written. entries Ordered list of ``{"path": , "object_id": }`` dicts — one per file, sorted by archive path. """ commit_id: str snapshot_id: str message: str branch: str author: str committed_at: str ref: str | None prefix: str file_count: int entries: list[_ListEntryJson] # --------------------------------------------------------------------------- # Path safety # --------------------------------------------------------------------------- def _safe_arcname(prefix: str, rel_path: str) -> str | None: """Build a safe archive entry name, guarding against zip-slip and tar-slip. Validates both the caller-supplied *prefix* and the per-file *rel_path* from the snapshot manifest. Returns the combined archive path string on success, or ``None`` if either component is unsafe — the caller must skip ``None`` entries and log a warning. Safety rules enforced --------------------- - *rel_path* must be non-empty and must not normalise to ``"."``. - *rel_path* must not be an absolute path. - Neither *prefix* nor *rel_path* may contain ``..`` path components. - Null bytes in either argument are rejected (they confuse archive readers and some OS path APIs). Args: prefix: Directory prefix to prepend inside the archive (may be empty). rel_path: Relative file path from the snapshot manifest. Returns: The safe archive entry name, or ``None`` if the entry should be skipped. """ if not rel_path or "\x00" in rel_path or "\x00" in prefix: return None clean_prefix = prefix.rstrip("/").strip() if clean_prefix and ".." in clean_prefix.split("/"): return None resolved = pathlib.PurePosixPath(rel_path) if resolved.is_absolute() or ".." in resolved.parts: return None safe_rel = str(resolved) # PurePosixPath("") normalises to "." — reject it. if not safe_rel or safe_rel == ".": return None return f"{clean_prefix}/{safe_rel}" if clean_prefix else safe_rel # --------------------------------------------------------------------------- # Manifest helpers # --------------------------------------------------------------------------- def _build_entries( root: pathlib.Path, manifest: Manifest, prefix: str, ) -> tuple[list[tuple[str, str, pathlib.Path]], list[str]]: """Resolve manifest entries into (arcname, object_id, obj_path) triples. Validates every entry through ``_safe_arcname`` and checks object existence. Returns a tuple of: - ``entries`` — safe ``(arcname, object_id, obj_path)`` triples, sorted by arcname. - ``skipped`` — display-safe descriptions of any entries that were skipped. Args: root: Repository root. manifest: Snapshot manifest mapping relative path → object ID. prefix: Directory prefix to prepend inside the archive. Returns: ``(entries, skipped)`` where *entries* are ready to write and *skipped* are human-readable descriptions of skipped paths for logging/warnings. """ entries: list[tuple[str, str, pathlib.Path]] = [] skipped: list[str] = [] for rel_path, object_id in sorted(manifest.items()): arcname = _safe_arcname(prefix, rel_path) if arcname is None: skipped.append(f"unsafe path: {sanitize_display(rel_path)}") continue obj = object_path(root, object_id) if not obj.exists(): skipped.append(f"missing object {object_id} for {sanitize_display(rel_path)}") continue entries.append((arcname, object_id, obj)) return entries, skipped # --------------------------------------------------------------------------- # Archive builders # --------------------------------------------------------------------------- def _build_tar( entries: list[tuple[str, str, pathlib.Path]], output_path: pathlib.Path, root: pathlib.Path | None = None, ) -> int: """Write a ``tar.gz`` archive from pre-validated *entries*. Each entry is a ``(arcname, object_id, obj_path)`` triple produced by ``_build_entries`` — every path has already been validated for safety. Args: entries: Validated ``(arcname, object_id, obj_path)`` triples. output_path: Destination file path for the archive. root: Repository root (used to resolve object content). Returns: Number of files written into the archive. """ import io count = 0 with tarfile.open(output_path, "w:gz") as tar: for arcname, object_id, obj_path in entries: if root is not None: content = read_object(root, object_id) if content is None: continue info = tarfile.TarInfo(name=arcname) info.size = len(content) tar.addfile(info, io.BytesIO(content)) else: tar.add(str(obj_path), arcname=arcname, recursive=False) count += 1 return count def _build_zip( entries: list[tuple[str, str, pathlib.Path]], output_path: pathlib.Path, root: pathlib.Path | None = None, ) -> int: """Write a ``zip`` archive from pre-validated *entries*. Each entry is a ``(arcname, object_id, obj_path)`` triple produced by ``_build_entries`` — every path has already been validated for safety. Args: entries: Validated ``(arcname, object_id, obj_path)`` triples. output_path: Destination file path for the archive. root: Repository root (used to resolve object content). Returns: Number of files written into the archive. """ count = 0 with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: for arcname, object_id, obj_path in entries: if root is not None: content = read_object(root, object_id) if content is None: continue zf.writestr(arcname, content) else: zf.write(str(obj_path), arcname=arcname) count += 1 return count # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``archive`` subcommand with its argument parser. Flags ----- --ref REF Branch, tag, or commit SHA to archive (default: HEAD). --format / -f {tar.gz,zip} Archive format. Default is ``tar.gz``. --output / -o PATH Output file path. Default: ``.`` in the current directory. The destination directory must already exist. --prefix DIR Directory prefix prepended to every entry inside the archive (e.g. ``myproject/``). Must not contain ``..`` segments. --list Preview mode — print what would be archived without writing a file. Compatible with ``--ref``, ``--prefix``, and ``--json``. --json Emit a machine-readable JSON object to stdout instead of human text. In list mode the schema is ``_ListJson``; otherwise ``_ArchiveJson``. """ parser = subparsers.add_parser( "archive", help="Export any historical snapshot as a portable archive.", description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument( "--ref", default=None, help="Branch, tag, or commit SHA to archive (default: HEAD).", ) parser.add_argument( "--format", "-f", default="tar.gz", dest="fmt", choices=sorted(_FORMAT_CHOICES), help="Archive format: tar.gz or zip (default: tar.gz).", ) parser.add_argument( "--output", "-o", default=None, help=( "Output file path (default: .). " "The destination directory must already exist." ), ) parser.add_argument( "--prefix", default="", help="Directory prefix inside the archive (e.g. myproject/).", ) parser.add_argument( "--list", action="store_true", dest="list_mode", default=False, help=( "Preview what would be archived without writing a file. " "Compatible with --ref, --prefix, and --json." ), ) parser.add_argument( "--json", "-j", action="store_true", dest="json_out", help="Emit machine-readable JSON to stdout instead of human text.", ) parser.set_defaults(func=run) # --------------------------------------------------------------------------- # Command implementation # --------------------------------------------------------------------------- def run(args: argparse.Namespace) -> None: """Export any historical snapshot as a portable archive. Resolves the commit ref, loads the snapshot manifest, validates every path for traversal safety, then writes a ``tar.gz`` or ``zip`` archive containing only tracked files (no ``.muse/`` internals). Use ``--list`` to preview entries without writing anything to disk. Agent quickstart ---------------- :: muse archive --json # HEAD → tar.gz muse archive --ref feat/audio --json # branch tip muse archive --format zip --output out.zip --json muse archive --list --json # preview without writing JSON fields ----------- path Output file path written. format ``"tar.gz"`` or ``"zip"``. file_count Number of files in the archive. bytes Archive size on disk in bytes. commit_id Full ``sha256:…`` commit ID archived. snapshot_id Full ``sha256:…`` snapshot ID. message Commit message. branch Branch name at archive time. author Author field from the commit record. agent_id Agent identity (empty for human commits). model_id Model identifier (empty for human commits). committed_at ISO-8601 commit timestamp. ref ``--ref`` value passed, or ``null`` for HEAD. prefix ``--prefix`` value used (empty string if none). With ``--list``, ``path``/``format``/``bytes`` are absent and an ``entries`` list is added — each entry: ``path`` (archive path), ``object_id`` (sha256). Exit codes ---------- 0 Archive written (or list preview complete). 1 Invalid arguments, bad prefix, output directory missing, ref not found. 2 Not inside a Muse repository. 3 Snapshot or object data missing. """ elapsed = start_timer() ref: str | None = args.ref fmt: str = args.fmt output: str | None = args.output prefix: str = args.prefix list_mode: bool = args.list_mode json_out: bool = args.json_out # Validate prefix up-front so the user gets a clear error before any I/O. clean_prefix = prefix.rstrip("/").strip() if clean_prefix and ".." in clean_prefix.split("/"): print( f"❌ --prefix must not contain '..' segments: {sanitize_display(prefix)}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) root = require_repo() branch = read_current_branch(root) # Resolve the commit reference. if ref is None: commit_id = get_head_commit_id(root, branch) if not commit_id: print("❌ No commits yet on this branch.", file=sys.stderr) raise SystemExit(ExitCode.USER_ERROR) commit = read_commit(root, commit_id) else: # Try as a branch name first (e.g. "main", "feat/audio"), then fall # through to resolve_commit_ref for SHA prefixes and HEAD~N syntax. commit = None try: branch_tip_id = get_head_commit_id(root, ref) if branch_tip_id: commit = read_commit(root, branch_tip_id) except Exception: pass if commit is None: commit = resolve_commit_ref(root, branch, ref) if commit is None: print( f"❌ Ref {sanitize_display(ref or 'HEAD')!r} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) snapshot = read_snapshot(root, commit.snapshot_id) if snapshot is None: print( f"❌ Snapshot {commit.snapshot_id} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) # Build and validate the entry list (shared between list and write modes). entries, skipped = _build_entries(root, snapshot.manifest, clean_prefix) for desc in skipped: logger.warning("⚠️ Skipping %s", desc) # --- List mode: preview without writing --- if list_mode: list_entries: list[_ListEntryJson] = [ _ListEntryJson(path=arcname, object_id=object_id) for arcname, object_id, _ in entries ] if json_out: print(json.dumps(_ListJson( **make_envelope(elapsed), commit_id=commit.commit_id, snapshot_id=commit.snapshot_id, message=commit.message, branch=branch, author=commit.author, committed_at=commit.committed_at.isoformat(), ref=ref, prefix=clean_prefix, file_count=len(entries), entries=list_entries, ))) return print( f"ℹ️ Snapshot {commit.commit_id} {sanitize_display(commit.message)}\n" f" {len(entries)} file(s) would be archived:" ) for entry in list_entries: print(f" {entry['path']}") if skipped: print(f"\n ⚠️ {len(skipped)} entry/entries skipped (unsafe or missing).") return # --- Write mode: build the archive --- # Use bare hex for the default filename — colons are invalid on Windows. _, _commit_hex = split_id(commit.commit_id) out_name = output or f"{_commit_hex}.{fmt}" out_path = pathlib.Path(out_name) # Validate that the destination directory exists before doing any work. if out_path.parent != pathlib.Path(".") and not out_path.parent.exists(): print( f"❌ Output directory does not exist: {sanitize_display(str(out_path.parent))}", file=sys.stderr, ) raise SystemExit(ExitCode.USER_ERROR) if fmt == "tar.gz": count = _build_tar(entries, out_path, root=root) else: count = _build_zip(entries, out_path, root=root) archive_bytes = out_path.stat().st_size if out_path.exists() else 0 if json_out: print(json.dumps(_ArchiveJson( **make_envelope(elapsed), path=str(out_path), format=fmt, file_count=count, bytes=archive_bytes, commit_id=commit.commit_id, snapshot_id=commit.snapshot_id, message=commit.message, branch=branch, author=commit.author, agent_id=commit.agent_id, model_id=commit.model_id, committed_at=commit.committed_at.isoformat(), ref=ref, prefix=clean_prefix, ))) return size_kb = archive_bytes / 1024 print( f"✅ Archive: {out_path} ({count} file(s), {size_kb:.1f} KiB)\n" f" Commit: {commit.commit_id} {sanitize_display(commit.message)}" )