aaronrene / gabriel-muse public
archive.py python
340 lines 10.9 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """``muse archive`` — export a snapshot as a portable archive.
2
3 Creates a ``tar.gz`` or ``zip`` archive from any historical snapshot —
4 HEAD by default. The archive contains only the tracked files (the contents
5 of ``state/`` at that point in time), making it the canonical way to share
6 a specific version without exposing the ``.muse/`` internals.
7
8 Usage::
9
10 muse archive # HEAD snapshot → archive.tar.gz
11 muse archive --ref feat/audio # branch tip
12 muse archive --ref a1b2c3d4 # specific commit SHA prefix
13 muse archive --format zip # zip instead of tar.gz
14 muse archive --output release-v1.0.zip # custom output path
15 muse archive --prefix myproject/ # add a directory prefix inside the archive
16 muse archive --json # machine-readable output on stdout
17
18 The archive is purely content — no Muse metadata (``.muse/``) is included.
19 This is intentional: archives are for *distribution*, not collaboration.
20 Use ``muse push`` / ``muse clone`` for distribution with full history.
21
22 JSON output (``--json``) schema::
23
24 {
25 "path": "<output file path>",
26 "format": "tar.gz" | "zip",
27 "file_count": <int>,
28 "bytes": <int>,
29 "commit_id": "<sha256>",
30 "message": "<commit message>",
31 "branch": "<branch>",
32 "ref": "<ref used or null>"
33 }
34
35 Exit codes::
36
37 0 — success
38 1 — bad arguments (bad format, missing commits, bad prefix)
39 2 — internal error (missing snapshot or object)
40 """
41
42 from __future__ import annotations
43
44 import argparse
45 import json
46 import logging
47 import pathlib
48 import sys
49 import tarfile
50 import zipfile
51 from typing import TypedDict
52
53 from muse.core.errors import ExitCode
54 from muse.core.object_store import object_path
55 from muse.core.repo import read_repo_id, require_repo
56 from muse.core.store import (
57 get_head_commit_id,
58 read_commit,
59 read_current_branch,
60 read_snapshot,
61 resolve_commit_ref,
62 )
63 from muse.core.validation import sanitize_display
64 from muse.core._types import Manifest
65
66 logger = logging.getLogger(__name__)
67
68 _FORMAT_CHOICES = {"tar.gz", "zip"}
69
70
71 # ---------------------------------------------------------------------------
72 # Typed JSON schema
73 # ---------------------------------------------------------------------------
74
75
76 class _ArchiveJson(TypedDict):
77 """Machine-readable output of ``muse archive --json``."""
78
79 path: str
80 format: str
81 file_count: int
82 bytes: int
83 commit_id: str
84 message: str
85 branch: str
86 ref: str | None
87
88
89 # ---------------------------------------------------------------------------
90 # Path safety
91 # ---------------------------------------------------------------------------
92
93
94 def _safe_arcname(prefix: str, rel_path: str) -> str | None:
95 """Build a safe archive entry name, guarding against zip-slip/tar-slip.
96
97 Returns ``None`` if either *prefix* or *rel_path* is unsafe or empty;
98 the caller must skip those entries.
99
100 Safety rules enforced:
101
102 - *rel_path* must be non-empty and must not be ``"."`` after normalisation.
103 - *rel_path* must not be absolute.
104 - Neither *prefix* nor *rel_path* may contain ``..`` path segments.
105 - Null bytes are rejected (they can confuse archive readers).
106 """
107 if not rel_path or "\x00" in rel_path or "\x00" in prefix:
108 return None
109
110 clean_prefix = prefix.rstrip("/").strip()
111 if clean_prefix and ".." in clean_prefix.split("/"):
112 return None
113
114 resolved = pathlib.PurePosixPath(rel_path)
115 if resolved.is_absolute() or ".." in resolved.parts:
116 return None
117
118 safe_rel = str(resolved)
119 # PurePosixPath("") normalises to "." — reject it.
120 if not safe_rel or safe_rel == ".":
121 return None
122
123 return (clean_prefix + "/" + safe_rel) if clean_prefix else safe_rel
124
125
126 # ---------------------------------------------------------------------------
127 # Archive builders
128 # ---------------------------------------------------------------------------
129
130
131 def _build_tar(
132 root: pathlib.Path,
133 manifest: Manifest,
134 output_path: pathlib.Path,
135 prefix: str,
136 ) -> int:
137 """Write a ``tar.gz`` archive from *manifest*; return the file count.
138
139 Each entry's archive name is validated by ``_safe_arcname`` to prevent
140 tar-slip path traversal attacks. Unsafe entries are logged and skipped.
141 """
142 count = 0
143 with tarfile.open(output_path, "w:gz") as tar:
144 for rel_path, object_id in sorted(manifest.items()):
145 arcname = _safe_arcname(prefix, rel_path)
146 if arcname is None:
147 logger.warning(
148 "⚠️ Skipping unsafe archive path: %s",
149 sanitize_display(rel_path),
150 )
151 continue
152 obj = object_path(root, object_id)
153 if not obj.exists():
154 logger.warning(
155 "⚠️ Missing object %s for %s — skipping",
156 object_id[:12],
157 sanitize_display(rel_path),
158 )
159 continue
160 tar.add(str(obj), arcname=arcname, recursive=False)
161 count += 1
162 return count
163
164
165 def _build_zip(
166 root: pathlib.Path,
167 manifest: Manifest,
168 output_path: pathlib.Path,
169 prefix: str,
170 ) -> int:
171 """Write a ``zip`` archive from *manifest*; return the file count.
172
173 Each entry's archive name is validated by ``_safe_arcname`` to prevent
174 zip-slip path traversal attacks. Unsafe entries are logged and skipped.
175 """
176 count = 0
177 with zipfile.ZipFile(output_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
178 for rel_path, object_id in sorted(manifest.items()):
179 arcname = _safe_arcname(prefix, rel_path)
180 if arcname is None:
181 logger.warning(
182 "⚠️ Skipping unsafe archive path: %s",
183 sanitize_display(rel_path),
184 )
185 continue
186 obj = object_path(root, object_id)
187 if not obj.exists():
188 logger.warning(
189 "⚠️ Missing object %s for %s — skipping",
190 object_id[:12],
191 sanitize_display(rel_path),
192 )
193 continue
194 zf.write(str(obj), arcname=arcname)
195 count += 1
196 return count
197
198
199 # ---------------------------------------------------------------------------
200 # Registration
201 # ---------------------------------------------------------------------------
202
203
204 def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None:
205 """Register the ``archive`` subcommand."""
206 parser = subparsers.add_parser(
207 "archive",
208 help="Export any historical snapshot as a portable archive.",
209 description=__doc__,
210 formatter_class=argparse.RawDescriptionHelpFormatter,
211 )
212 parser.add_argument(
213 "--ref",
214 default=None,
215 help="Branch, tag, or commit SHA to archive (default: HEAD).",
216 )
217 parser.add_argument(
218 "--format", "-f",
219 default="tar.gz",
220 dest="fmt",
221 choices=sorted(_FORMAT_CHOICES),
222 help="Archive format: tar.gz or zip (default: tar.gz).",
223 )
224 parser.add_argument(
225 "--output", "-o",
226 default=None,
227 help="Output file path (default: <sha12>.<format>).",
228 )
229 parser.add_argument(
230 "--prefix",
231 default="",
232 help="Directory prefix inside the archive (e.g. myproject/).",
233 )
234 parser.add_argument(
235 "--json",
236 action="store_true",
237 dest="output_json",
238 default=False,
239 help="Emit machine-readable JSON to stdout instead of human text.",
240 )
241 parser.set_defaults(func=run)
242
243
244 # ---------------------------------------------------------------------------
245 # Command implementation
246 # ---------------------------------------------------------------------------
247
248
249 def run(args: argparse.Namespace) -> None:
250 """Export any historical snapshot as a portable archive.
251
252 The archive contains only tracked files — no ``.muse/`` metadata. It is
253 the canonical distribution format for a specific version.
254
255 Agents should pass ``--json`` to receive a machine-readable result with
256 ``path``, ``format``, ``file_count``, ``bytes``, ``commit_id``,
257 ``message``, ``branch``, and ``ref``.
258
259 Examples::
260
261 muse archive # HEAD → <sha12>.tar.gz
262 muse archive --ref v1.0.0 # tag → <sha12>.tar.gz
263 muse archive --format zip --output dist/release.zip
264 muse archive --prefix myproject/ # all files under myproject/
265 muse archive --json # machine-readable
266 """
267 ref: str | None = args.ref
268 fmt: str = args.fmt
269 output: str | None = args.output
270 prefix: str = args.prefix
271 output_json: bool = args.output_json
272
273 # Validate prefix against traversal — _safe_arcname will also check
274 # per-entry, but an early explicit check gives the user a clear message.
275 clean_prefix = prefix.rstrip("/").strip()
276 if clean_prefix and ".." in clean_prefix.split("/"):
277 print(
278 f"❌ --prefix must not contain '..' segments: {sanitize_display(prefix)}",
279 file=sys.stderr,
280 )
281 raise SystemExit(ExitCode.USER_ERROR)
282
283 root = require_repo()
284 repo_id = read_repo_id(root)
285 branch = read_current_branch(root)
286
287 if ref is None:
288 commit_id = get_head_commit_id(root, branch)
289 if not commit_id:
290 print("❌ No commits yet on this branch.", file=sys.stderr)
291 raise SystemExit(ExitCode.USER_ERROR)
292 commit = read_commit(root, commit_id)
293 else:
294 commit = resolve_commit_ref(root, repo_id, branch, ref)
295
296 if commit is None:
297 print(
298 f"❌ Ref '{sanitize_display(ref or 'HEAD')}' not found.",
299 file=sys.stderr,
300 )
301 raise SystemExit(ExitCode.USER_ERROR)
302
303 snapshot = read_snapshot(root, commit.snapshot_id)
304 if snapshot is None:
305 print(
306 f"❌ Snapshot {commit.snapshot_id[:8]} not found.",
307 file=sys.stderr,
308 )
309 raise SystemExit(ExitCode.INTERNAL_ERROR)
310
311 short = commit.commit_id[:12]
312 out_name = output or f"{short}.{fmt}"
313 out_path = pathlib.Path(out_name)
314
315 if fmt == "tar.gz":
316 count = _build_tar(root, snapshot.manifest, out_path, prefix)
317 else:
318 count = _build_zip(root, snapshot.manifest, out_path, prefix)
319
320 archive_bytes = out_path.stat().st_size if out_path.exists() else 0
321
322 if output_json:
323 payload = _ArchiveJson(
324 path=str(out_path),
325 format=fmt,
326 file_count=count,
327 bytes=archive_bytes,
328 commit_id=commit.commit_id,
329 message=commit.message,
330 branch=branch,
331 ref=ref,
332 )
333 print(json.dumps(payload))
334 return
335
336 size_kb = archive_bytes / 1024
337 print(
338 f"✅ Archive: {out_path} ({count} file(s), {size_kb:.1f} KiB)\n"
339 f" Commit: {short} {sanitize_display(commit.message)}"
340 )
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago