Skills › Scientific Agent Skills › labarchive-integration › notebook_operations.py
notebook_operations.py
451 lines · skills/labarchive-integration/scripts/notebook_operations.py at 5f2bc03
1#!/usr/bin/env python3
2"""Safely inspect a local LabArchives LA container ZIP without extracting it.
4An LA container is an attachment packaging format with lamanifest.xml. It is
5not a notebook backup. This script performs no network or remote write.
6"""
8from __future__ import annotations
10import argparse
11import json
12import os
13import re
14import stat
15import sys
16import zipfile
17from collections import Counter
18from collections.abc import Mapping, Sequence
19from pathlib import Path, PurePosixPath
20from typing import Any
21from xml.etree import ElementTree
24MANIFEST_NAME = "lamanifest.xml"
25DEFAULT_MAX_MEMBERS = 10_000
26DEFAULT_MAX_TOTAL_BYTES = 4 * 1024 * 1024 * 1024
27DEFAULT_MAX_MANIFEST_BYTES = 1024 * 1024
28DEFAULT_MAX_INDEX_BYTES = 8 * 1024 * 1024
29DEFAULT_MAX_COMPRESSION_RATIO = 1_000.0
30_WINDOWS_DRIVE = re.compile(r"^[A-Za-z]:")
33class InspectionError(ValueError):
34 """Raised for invalid CLI inputs or unsafe local output paths."""
37def _local_name(tag: str) -> str:
38 return tag.rsplit("}", 1)[-1]
41def _member_path_error(name: str) -> str | None:
42 if not name or "\x00" in name:
43 return "empty or NUL-containing member name"
44 if "\\" in name:
45 return "backslash in member name"
46 if name.startswith("/") or _WINDOWS_DRIVE.match(name):
47 return "absolute member path"
48 parts = PurePosixPath(name).parts
49 if any(part in {"", ".", ".."} for part in parts):
50 return "ambiguous or traversing member path"
51 return None
54def _is_symlink(info: zipfile.ZipInfo) -> bool:
55 mode = info.external_attr >> 16
56 return stat.S_ISLNK(mode)
59def _find_child(parent: ElementTree.Element, name: str) -> ElementTree.Element | None:
60 for child in parent:
61 if _local_name(child.tag) == name:
62 return child
63 return None
66def _manifest_reference(
67 entry_info: ElementTree.Element,
68 element_name: str,
69 archive_names: set[str],
70 errors: list[str],
71) -> str | None:
72 element = _find_child(entry_info, element_name)
73 if element is None:
74 errors.append(f"manifest is missing required {element_name} element")
75 return None
76 reference = (element.get("name") or "").strip()
77 if not reference:
78 errors.append(f"manifest {element_name} has no name attribute")
79 return None
80 path_error = _member_path_error(reference)
81 if path_error:
82 errors.append(
83 f"manifest {element_name} reference {reference!r} is unsafe: {path_error}"
84 )
85 return reference
86 if reference not in archive_names:
87 errors.append(
88 f"manifest {element_name} reference is absent from archive: {reference!r}"
89 )
90 return reference
93def _parse_manifest(
94 archive: zipfile.ZipFile,
95 info: zipfile.ZipInfo,
96 archive_names: set[str],
97 max_manifest_bytes: int,
98 max_index_bytes: int,
99 errors: list[str],
100 warnings: list[str],
101) -> dict[str, Any]:
102 manifest: dict[str, Any] = {
103 "present": True,
104 "parsed": False,
105 "application_file": None,
106 "preview_file": None,
107 "index_file": None,
108 "index_utf8_valid": None,
109 }
110 if info.file_size > max_manifest_bytes:
111 errors.append(
112 "lamanifest.xml exceeds the configured manifest-size limit "
113 f"({info.file_size} > {max_manifest_bytes} bytes)"
114 )
115 return manifest
116 if info.flag_bits & 0x1:
117 errors.append("lamanifest.xml is encrypted and cannot be safely inspected")
118 return manifest
120 raw = archive.read(info)
121 upper = raw.upper()
122 if b"<!DOCTYPE" in upper or b"<!ENTITY" in upper:
123 errors.append("lamanifest.xml contains a forbidden DTD or entity declaration")
124 return manifest
126 try:
127 root = ElementTree.fromstring(raw)
128 except ElementTree.ParseError as exc:
129 errors.append(f"lamanifest.xml is not well-formed XML: {exc}")
130 return manifest
132 manifest["root_element"] = _local_name(root.tag)
133 if _local_name(root.tag) != "la_manifest":
134 errors.append("lamanifest.xml root element must be la_manifest")
135 return manifest
136 entry_info = _find_child(root, "entry_info")
137 if entry_info is None:
138 errors.append("lamanifest.xml is missing entry_info")
139 return manifest
141 manifest["parsed"] = True
142 for field in ("application_file", "preview_file", "index_file"):
143 manifest[field] = _manifest_reference(entry_info, field, archive_names, errors)
145 index_name = manifest["index_file"]
146 if isinstance(index_name, str) and index_name in archive_names:
147 index_info = archive.getinfo(index_name)
148 if index_info.flag_bits & 0x1:
149 errors.append("manifest index_file is encrypted")
150 elif index_info.file_size > max_index_bytes:
151 warnings.append(
152 "index_file UTF-8 validation skipped because it exceeds the "
153 f"configured limit ({index_info.file_size} > {max_index_bytes} bytes)"
154 )
155 else:
156 try:
157 archive.read(index_info).decode("utf-8")
158 except UnicodeDecodeError as exc:
159 errors.append(f"index_file is not valid UTF-8: {exc}")
160 manifest["index_utf8_valid"] = False
161 else:
162 manifest["index_utf8_valid"] = True
163 return manifest
166def inspect_container(
167 input_path: Path,
168 *,
169 max_members: int = DEFAULT_MAX_MEMBERS,
170 max_total_bytes: int = DEFAULT_MAX_TOTAL_BYTES,
171 max_manifest_bytes: int = DEFAULT_MAX_MANIFEST_BYTES,
172 max_index_bytes: int = DEFAULT_MAX_INDEX_BYTES,
173 max_compression_ratio: float = DEFAULT_MAX_COMPRESSION_RATIO,
174) -> dict[str, Any]:
175 """Inspect an LA container and return a JSON-serializable report."""
177 if (
178 min(
179 max_members,
180 max_total_bytes,
181 max_manifest_bytes,
182 max_index_bytes,
183 )
184 <= 0
185 ):
186 raise InspectionError("all size/member limits must be positive")
187 if max_compression_ratio <= 0:
188 raise InspectionError("compression-ratio limit must be positive")
190 resolved = input_path.expanduser().resolve(strict=True)
191 if not resolved.is_file():
192 raise InspectionError(f"input is not a regular file: {resolved}")
194 errors: list[str] = []
195 warnings: list[str] = []
196 with zipfile.ZipFile(resolved, "r") as archive:
197 infos = archive.infolist()
198 names = [info.filename for info in infos]
199 name_counts = Counter(names)
200 duplicates = sorted(name for name, count in name_counts.items() if count > 1)
201 if duplicates:
202 errors.append(f"duplicate archive member names: {duplicates!r}")
204 if len(infos) > max_members:
205 errors.append(
206 f"archive member count exceeds limit ({len(infos)} > {max_members})"
207 )
209 total_uncompressed = sum(info.file_size for info in infos)
210 total_compressed = sum(info.compress_size for info in infos)
211 if total_uncompressed > max_total_bytes:
212 errors.append(
213 "total uncompressed size exceeds limit "
214 f"({total_uncompressed} > {max_total_bytes} bytes)"
215 )
217 unsafe_members: list[dict[str, str]] = []
218 encrypted_members: list[str] = []
219 symlink_members: list[str] = []
220 high_ratio_members: list[dict[str, Any]] = []
221 for info in infos:
222 path_error = _member_path_error(info.filename)
223 if path_error:
224 unsafe_members.append({"name": info.filename, "reason": path_error})
225 if info.flag_bits & 0x1:
226 encrypted_members.append(info.filename)
227 if _is_symlink(info):
228 symlink_members.append(info.filename)
230 if info.file_size:
231 ratio = (
232 float("inf")
233 if info.compress_size == 0
234 else info.file_size / info.compress_size
235 )
236 if ratio > max_compression_ratio:
237 high_ratio_members.append(
238 {
239 "name": info.filename,
240 "ratio": None if ratio == float("inf") else round(ratio, 2),
241 }
242 )
244 if unsafe_members:
245 errors.append("archive contains unsafe member paths")
246 if encrypted_members:
247 errors.append("archive contains encrypted members")
248 if symlink_members:
249 errors.append("archive contains symbolic-link members")
250 if high_ratio_members:
251 errors.append("archive contains members above the compression-ratio limit")
253 manifest_infos = [info for info in infos if info.filename == MANIFEST_NAME]
254 if not manifest_infos:
255 errors.append(f"archive is missing required {MANIFEST_NAME}")
256 manifest: dict[str, Any] = {"present": False, "parsed": False}
257 elif len(manifest_infos) > 1:
258 errors.append(f"archive contains multiple {MANIFEST_NAME} members")
259 manifest = {"present": True, "parsed": False}
260 else:
261 manifest = _parse_manifest(
262 archive,
263 manifest_infos[0],
264 set(names),
265 max_manifest_bytes,
266 max_index_bytes,
267 errors,
268 warnings,
269 )
271 referenced = {
272 value
273 for field in ("application_file", "preview_file", "index_file")
274 if isinstance((value := manifest.get(field)), str)
275 }
276 unreferenced = sorted(
277 name for name in names if name != MANIFEST_NAME and name not in referenced
278 )
279 if unreferenced:
280 warnings.append(
281 "archive has members not referenced by lamanifest.xml; review them"
282 )
284 return {
285 "valid": not errors,
286 "input": str(resolved),
287 "format": "LabArchives LA container ZIP",
288 "network_request_performed": False,
289 "archive": {
290 "member_count": len(infos),
291 "total_compressed_bytes": total_compressed,
292 "total_uncompressed_bytes": total_uncompressed,
293 "duplicate_members": duplicates,
294 "unsafe_members": unsafe_members,
295 "encrypted_members": encrypted_members,
296 "symlink_members": symlink_members,
297 "high_compression_ratio_members": high_ratio_members,
298 "unreferenced_members": unreferenced,
299 },
300 "manifest": manifest,
301 "limits": {
302 "max_members": max_members,
303 "max_total_uncompressed_bytes": max_total_bytes,
304 "max_manifest_bytes": max_manifest_bytes,
305 "max_index_bytes": max_index_bytes,
306 "max_compression_ratio": max_compression_ratio,
307 },
308 "errors": errors,
309 "warnings": warnings,
310 }
313def _write_json_safely(
314 output_path: Path,
315 payload: Mapping[str, Any],
316 *,
317 input_path: Path,
318 force: bool,
319 compact: bool,
320) -> None:
321 parent = output_path.expanduser().parent.resolve(strict=True)
322 if not parent.is_dir():
323 raise InspectionError(f"output parent is not a directory: {parent}")
324 destination = parent / output_path.name
325 if destination.resolve(strict=False) == input_path.resolve(strict=True):
326 raise InspectionError("output path must not overwrite the input archive")
327 if destination.is_symlink():
328 raise InspectionError("refusing to write through an output symlink")
329 if destination.exists() and not force:
330 raise InspectionError(
331 f"output already exists (use --force for a regular file): {destination}"
332 )
333 if destination.exists() and not destination.is_file():
334 raise InspectionError("output destination is not a regular file")
336 flags = os.O_WRONLY | os.O_CREAT
337 flags |= os.O_TRUNC if force else os.O_EXCL
338 if hasattr(os, "O_NOFOLLOW"):
339 flags |= os.O_NOFOLLOW
340 descriptor = os.open(destination, flags, 0o600)
341 os.fchmod(descriptor, 0o600)
342 with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
343 if compact:
344 json.dump(payload, stream, sort_keys=True, separators=(",", ":"))
345 else:
346 json.dump(payload, stream, indent=2, sort_keys=True)
347 stream.write("\n")
350def _emit_json(payload: Mapping[str, Any], *, compact: bool) -> None:
351 if compact:
352 json.dump(payload, sys.stdout, sort_keys=True, separators=(",", ":"))
353 else:
354 json.dump(payload, sys.stdout, indent=2, sort_keys=True)
355 sys.stdout.write("\n")
358def command_inspect(args: argparse.Namespace) -> int:
359 input_path = Path(args.input)
360 report = inspect_container(
361 input_path,
362 max_members=args.max_members,
363 max_total_bytes=args.max_total_bytes,
364 max_manifest_bytes=args.max_manifest_bytes,
365 max_index_bytes=args.max_index_bytes,
366 max_compression_ratio=args.max_compression_ratio,
367 )
368 if args.output:
369 _write_json_safely(
370 Path(args.output),
371 report,
372 input_path=input_path,
373 force=args.force,
374 compact=args.compact,
375 )
376 else:
377 _emit_json(report, compact=args.compact)
378 return 0 if report["valid"] else 1
381def build_parser() -> argparse.ArgumentParser:
382 parser = argparse.ArgumentParser(
383 description=(
384 "Inspect a local LabArchives LA container ZIP without extracting it. "
385 "This is not a notebook-backup client."
386 )
387 )
388 subparsers = parser.add_subparsers(dest="command", required=True)
389 inspect_parser = subparsers.add_parser(
390 "inspect", help="validate archive safety and lamanifest.xml references"
391 )
392 inspect_parser.add_argument("input", help="path to a local LA container ZIP")
393 inspect_parser.add_argument(
394 "--output",
395 help="write JSON to this existing-parent path with mode 0600",
396 )
397 inspect_parser.add_argument(
398 "--force",
399 action="store_true",
400 help="replace an existing regular output file; never follow symlinks",
401 )
402 inspect_parser.add_argument(
403 "--compact", action="store_true", help="emit compact JSON"
404 )
405 inspect_parser.add_argument("--max-members", type=int, default=DEFAULT_MAX_MEMBERS)
406 inspect_parser.add_argument(
407 "--max-total-bytes", type=int, default=DEFAULT_MAX_TOTAL_BYTES
408 )
409 inspect_parser.add_argument(
410 "--max-manifest-bytes", type=int, default=DEFAULT_MAX_MANIFEST_BYTES
411 )
412 inspect_parser.add_argument(
413 "--max-index-bytes", type=int, default=DEFAULT_MAX_INDEX_BYTES
414 )
415 inspect_parser.add_argument(
416 "--max-compression-ratio",
417 type=float,
418 default=DEFAULT_MAX_COMPRESSION_RATIO,
419 )
420 inspect_parser.set_defaults(handler=command_inspect)
421 return parser
424def main(argv: Sequence[str] | None = None) -> int:
425 parser = build_parser()
426 args = parser.parse_args(argv)
427 try:
428 return int(args.handler(args))
429 except (
430 FileNotFoundError,
431 InspectionError,
432 OSError,
433 zipfile.BadZipFile,
434 zipfile.LargeZipFile,
435 ) as exc:
436 _emit_json(
437 {
438 "valid": False,
439 "error": str(exc),
440 "network_request_performed": False,
441 },
442 compact=getattr(args, "compact", False),
443 )
444 return 2
445 except KeyboardInterrupt:
446 print('{"valid":false,"error":"cancelled"}', file=sys.stderr)
447 return 130
450if __name__ == "__main__":
451 raise SystemExit(main())