"""``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 ``state/`` at that point in time), making it the canonical way to share a specific version without exposing the ``.muse/`` internals. Usage:: muse archive # HEAD snapshot → archive.tar.gz muse archive --ref feat/audio # branch tip muse archive --ref a1b2c3d4 # specific 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/ # add a directory prefix inside the archive muse archive --json # machine-readable output on stdout The archive is purely content — no Muse metadata (``.muse/``) is included. This is intentional: archives are for *distribution*, not collaboration. Use ``muse push`` / ``muse clone`` for distribution with full history. JSON output (``--json``) schema:: { "path": "", "format": "tar.gz" | "zip", "file_count": , "bytes": , "commit_id": "", "message": "", "branch": "", "ref": "" } Exit codes:: 0 — success 1 — bad arguments (bad format, missing commits, bad prefix) 2 — internal error (missing snapshot or object) """ from __future__ import annotations import argparse import json import logging import pathlib import sys import tarfile import zipfile from typing import TypedDict from muse.core.errors import ExitCode from muse.core.object_store import object_path from muse.core.repo import read_repo_id, require_repo from muse.core.store import ( get_head_commit_id, read_commit, read_current_branch, read_snapshot, resolve_commit_ref, ) from muse.core.validation import sanitize_display from muse.core._types import Manifest logger = logging.getLogger(__name__) _FORMAT_CHOICES = {"tar.gz", "zip"} # --------------------------------------------------------------------------- # Typed JSON schema # --------------------------------------------------------------------------- class _ArchiveJson(TypedDict): """Machine-readable output of ``muse archive --json``.""" path: str format: str file_count: int bytes: int commit_id: str message: str branch: str ref: str | None # --------------------------------------------------------------------------- # Path safety # --------------------------------------------------------------------------- def _safe_arcname(prefix: str, rel_path: str) -> str | None: """Build a safe archive entry name, guarding against zip-slip/tar-slip. Returns ``None`` if either *prefix* or *rel_path* is unsafe or empty; the caller must skip those entries. Safety rules enforced: - *rel_path* must be non-empty and must not be ``"."`` after normalisation. - *rel_path* must not be absolute. - Neither *prefix* nor *rel_path* may contain ``..`` path segments. - Null bytes are rejected (they can confuse archive readers). """ 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 (clean_prefix + "/" + safe_rel) if clean_prefix else safe_rel # --------------------------------------------------------------------------- # Archive builders # --------------------------------------------------------------------------- def _build_tar( root: pathlib.Path, manifest: Manifest, output_path: pathlib.Path, prefix: str, ) -> int: """Write a ``tar.gz`` archive from *manifest*; return the file count. Each entry's archive name is validated by ``_safe_arcname`` to prevent tar-slip path traversal attacks. Unsafe entries are logged and skipped. """ count = 0 with tarfile.open(output_path, "w:gz") as tar: for rel_path, object_id in sorted(manifest.items()): arcname = _safe_arcname(prefix, rel_path) if arcname is None: logger.warning( "⚠️ Skipping unsafe archive path: %s", sanitize_display(rel_path), ) continue obj = object_path(root, object_id) if not obj.exists(): logger.warning( "⚠️ Missing object %s for %s — skipping", object_id[:12], sanitize_display(rel_path), ) continue tar.add(str(obj), arcname=arcname, recursive=False) count += 1 return count def _build_zip( root: pathlib.Path, manifest: Manifest, output_path: pathlib.Path, prefix: str, ) -> int: """Write a ``zip`` archive from *manifest*; return the file count. Each entry's archive name is validated by ``_safe_arcname`` to prevent zip-slip path traversal attacks. Unsafe entries are logged and skipped. """ count = 0 with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as zf: for rel_path, object_id in sorted(manifest.items()): arcname = _safe_arcname(prefix, rel_path) if arcname is None: logger.warning( "⚠️ Skipping unsafe archive path: %s", sanitize_display(rel_path), ) continue obj = object_path(root, object_id) if not obj.exists(): logger.warning( "⚠️ Missing object %s for %s — skipping", object_id[:12], sanitize_display(rel_path), ) continue zf.write(str(obj), arcname=arcname) count += 1 return count # --------------------------------------------------------------------------- # Registration # --------------------------------------------------------------------------- def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: """Register the ``archive`` subcommand.""" 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: .).", ) parser.add_argument( "--prefix", default="", help="Directory prefix inside the archive (e.g. myproject/).", ) parser.add_argument( "--json", action="store_true", dest="output_json", default=False, 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. The archive contains only tracked files — no ``.muse/`` metadata. It is the canonical distribution format for a specific version. Agents should pass ``--json`` to receive a machine-readable result with ``path``, ``format``, ``file_count``, ``bytes``, ``commit_id``, ``message``, ``branch``, and ``ref``. Examples:: muse archive # HEAD → .tar.gz muse archive --ref v1.0.0 # tag → .tar.gz muse archive --format zip --output dist/release.zip muse archive --prefix myproject/ # all files under myproject/ muse archive --json # machine-readable """ ref: str | None = args.ref fmt: str = args.fmt output: str | None = args.output prefix: str = args.prefix output_json: bool = args.output_json # Validate prefix against traversal — _safe_arcname will also check # per-entry, but an early explicit check gives the user a clear message. 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() repo_id = read_repo_id(root) branch = read_current_branch(root) 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: commit = resolve_commit_ref(root, repo_id, branch, ref) if commit is None: print( f"❌ Ref '{sanitize_display(ref or 'HEAD')}' 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[:8]} not found.", file=sys.stderr, ) raise SystemExit(ExitCode.INTERNAL_ERROR) short = commit.commit_id[:12] out_name = output or f"{short}.{fmt}" out_path = pathlib.Path(out_name) if fmt == "tar.gz": count = _build_tar(root, snapshot.manifest, out_path, prefix) else: count = _build_zip(root, snapshot.manifest, out_path, prefix) archive_bytes = out_path.stat().st_size if out_path.exists() else 0 if output_json: payload = _ArchiveJson( path=str(out_path), format=fmt, file_count=count, bytes=archive_bytes, commit_id=commit.commit_id, message=commit.message, branch=branch, ref=ref, ) print(json.dumps(payload)) return size_kb = archive_bytes / 1024 print( f"✅ Archive: {out_path} ({count} file(s), {size_kb:.1f} KiB)\n" f" Commit: {short} {sanitize_display(commit.message)}" )