Skills › Scientific Agent Skills › gtars › artifact_inspector.py
artifact_inspector.py
328 lines · skills/gtars/scripts/artifact_inspector.py at 5f2bc03
1#!/usr/bin/env python3
2"""Hash and classify local Gtars artifacts without loading or executing them."""
4from __future__ import annotations
6import argparse
7import os
8import re
9import stat
10import sys
11from email.parser import Parser
12from pathlib import Path
14from _common import (
15 HARD_MAX_BYTES,
16 HARD_MAX_FILES,
17 SafetyError,
18 add_path_mode_argument,
19 display_path,
20 fail_json,
21 int_type,
22 iter_text_lines,
23 local_path,
24 print_json,
25 sha256_file,
26)
29TOOL = "gtars-artifact-inspector"
30_WHEEL = re.compile(r"^gtars-(\d+\.\d+\.\d+)-.+\.whl$")
31_CRATE = re.compile(r"^(gtars|gtars-cli)-(\d+\.\d+\.\d+)\.crate$")
32_CHECKSUM = re.compile(r"^([0-9a-fA-F]{64})[ \t]+[* ]?([^/\\]+)$")
33_ARCHIVES = (".tar.gz", ".tgz", ".zip", ".crate", ".whl")
36def build_parser() -> argparse.ArgumentParser:
37 parser = argparse.ArgumentParser(
38 description=(
39 "Inventory, hash, and version-screen local Gtars wheels, crates, "
40 "release archives, native extensions, binaries, or METADATA files. "
41 "Archives are not extracted and code is never loaded or executed."
42 )
43 )
44 parser.add_argument(
45 "--artifact",
46 action="append",
47 required=True,
48 help="Local regular file; repeat for multiple artifacts.",
49 )
50 parser.add_argument("--checksum-manifest", help="Local SHA256SUMS-style file.")
51 parser.add_argument("--expected-python-version", default="0.10.0")
52 parser.add_argument("--expected-rust-version", default="0.10.0")
53 parser.add_argument("--expected-cli-version", default="0.10.0")
54 parser.add_argument(
55 "--max-files",
56 type=int_type(minimum=1, maximum=HARD_MAX_FILES, label="max-files"),
57 default=1_000,
58 )
59 parser.add_argument(
60 "--max-file-bytes",
61 type=int_type(
62 minimum=1,
63 maximum=HARD_MAX_BYTES,
64 label="max-file-bytes",
65 ),
66 default=2 * 1024**3,
67 )
68 parser.add_argument(
69 "--max-total-bytes",
70 type=int_type(
71 minimum=1,
72 maximum=HARD_MAX_BYTES,
73 label="max-total-bytes",
74 ),
75 default=4 * 1024**3,
76 )
77 add_path_mode_argument(parser)
78 return parser
81def _read_prefix(path: Path, length: int = 16) -> bytes:
82 flags = os.O_RDONLY
83 if hasattr(os, "O_NOFOLLOW"):
84 flags |= os.O_NOFOLLOW
85 descriptor = os.open(path, flags)
86 try:
87 if not stat.S_ISREG(os.fstat(descriptor).st_mode):
88 raise SafetyError("artifact must be a regular file")
89 return os.read(descriptor, length)
90 finally:
91 os.close(descriptor)
94def _native_format(prefix: bytes) -> str | None:
95 if prefix.startswith(b"\x7fELF"):
96 return "elf"
97 if prefix.startswith(b"MZ"):
98 return "portable-executable"
99 if prefix[:4] in {
100 b"\xfe\xed\xfa\xce",
101 b"\xce\xfa\xed\xfe",
102 b"\xfe\xed\xfa\xcf",
103 b"\xcf\xfa\xed\xfe",
104 b"\xca\xfe\xba\xbe",
105 b"\xbe\xba\xfe\xca",
106 }:
107 return "mach-o"
108 return None
111def _parse_metadata(path: Path, max_bytes: int) -> dict[str, str | None]:
112 text = "\n".join(
113 line
114 for _, line in iter_text_lines(
115 path,
116 max_bytes=max_bytes,
117 max_records=100_000,
118 )
119 )
120 message = Parser().parsestr(text)
121 return {
122 "name": message.get("Name"),
123 "version": message.get("Version"),
124 "requires_python": message.get("Requires-Python"),
125 }
128def _read_checksums(
129 path: Path,
130 *,
131 max_bytes: int,
132 max_files: int,
133) -> dict[str, str]:
134 expected: dict[str, str] = {}
135 for line_number, line in iter_text_lines(
136 path,
137 max_bytes=max_bytes,
138 max_records=max_files + 100,
139 ):
140 if not line or line.startswith("#"):
141 continue
142 match = _CHECKSUM.fullmatch(line)
143 if match is None:
144 raise SafetyError(f"invalid checksum line {line_number}")
145 digest, name = match.groups()
146 if name in expected:
147 raise SafetyError(f"duplicate checksum entry at line {line_number}")
148 expected[name] = digest.lower()
149 if len(expected) > max_files:
150 raise SafetyError("checksum manifest exceeds max-files")
151 return expected
154def inspect(args: argparse.Namespace) -> tuple[dict, int]:
155 if len(args.artifact) > args.max_files:
156 raise SafetyError("artifact count exceeds max-files")
157 paths = [local_path(raw, kind="file") for raw in args.artifact]
158 if len({str(path) for path in paths}) != len(paths):
159 raise SafetyError("duplicate artifact path")
161 checksum_path = None
162 checksums: dict[str, str] = {}
163 checksum_report = None
164 if args.checksum_manifest:
165 checksum_path = local_path(args.checksum_manifest, kind="file")
166 checksums = _read_checksums(
167 checksum_path,
168 max_bytes=min(args.max_file_bytes, 64 * 1024**2),
169 max_files=args.max_files,
170 )
172 errors: dict[str, int] = {}
173 warnings: dict[str, int] = {}
174 summaries: list[dict] = []
175 total = 0
176 verified = 0
178 for index, path in enumerate(paths, start=1):
179 digest, size = sha256_file(path, max_bytes=args.max_file_bytes)
180 total += size
181 if total > args.max_total_bytes:
182 raise SafetyError("artifacts exceed max-total-bytes")
183 name = path.name
184 prefix = _read_prefix(path)
185 native = _native_format(prefix)
186 kind = "data-or-metadata"
187 version = None
188 package = None
189 metadata = None
191 wheel_match = _WHEEL.fullmatch(name)
192 crate_match = _CRATE.fullmatch(name)
193 if wheel_match:
194 kind = "python-wheel-native-code"
195 package = "gtars"
196 version = wheel_match.group(1)
197 if version != args.expected_python_version:
198 errors["python_version_mismatch"] = (
199 errors.get("python_version_mismatch", 0) + 1
200 )
201 elif crate_match:
202 kind = "cargo-crate-archive"
203 package, version = crate_match.groups()
204 expected = (
205 args.expected_cli_version
206 if package == "gtars-cli"
207 else args.expected_rust_version
208 )
209 if version != expected:
210 errors[f"{package}_version_mismatch"] = (
211 errors.get(f"{package}_version_mismatch", 0) + 1
212 )
213 elif name == "METADATA":
214 kind = "python-distribution-metadata"
215 metadata = _parse_metadata(path, min(args.max_file_bytes, 16 * 1024**2))
216 package = metadata["name"]
217 version = metadata["version"]
218 if package != "gtars":
219 errors["metadata_package_name_mismatch"] = 1
220 if version != args.expected_python_version:
221 errors["metadata_python_version_mismatch"] = 1
222 if metadata["requires_python"] != ">=3.10":
223 warnings["metadata_requires_python_differs_from_snapshot"] = 1
224 elif native is not None:
225 kind = "native-executable-code"
226 warnings["native_version_not_verified_without_execution"] = (
227 warnings.get("native_version_not_verified_without_execution", 0) + 1
228 )
229 elif name.endswith(_ARCHIVES):
230 kind = "archive-not-extracted"
231 warnings["archive_contents_not_inspected"] = (
232 warnings.get("archive_contents_not_inspected", 0) + 1
233 )
235 expected_digest = checksums.get(name)
236 checksum_ok = expected_digest == digest if expected_digest is not None else None
237 if checksum_ok:
238 verified += 1
239 elif checksum_ok is False:
240 errors["checksum_mismatch"] = errors.get("checksum_mismatch", 0) + 1
241 elif checksums:
242 warnings["artifact_missing_from_checksum_manifest"] = (
243 warnings.get("artifact_missing_from_checksum_manifest", 0) + 1
244 )
246 summaries.append(
247 {
248 "path": display_path(path, index, args.path_mode),
249 "basename": name if args.path_mode != "redacted" else None,
250 "sha256": digest,
251 "size_bytes": size,
252 "kind": kind,
253 "native_format": native,
254 "package": package,
255 "version": version,
256 "metadata": metadata,
257 "checksum_verified": checksum_ok,
258 "loaded": False,
259 "executed": False,
260 "extracted": False,
261 }
262 )
264 if checksum_path is not None:
265 manifest_digest, manifest_size = sha256_file(
266 checksum_path,
267 max_bytes=min(args.max_file_bytes, 64 * 1024**2),
268 )
269 checksum_report = {
270 "path": display_path(
271 checksum_path,
272 len(paths) + 1,
273 args.path_mode,
274 ),
275 "sha256": manifest_digest,
276 "size_bytes": manifest_size,
277 "entries": len(checksums),
278 "verified_artifacts": verified,
279 }
281 report = {
282 "ok": not errors,
283 "tool": TOOL,
284 "contract": {
285 "network_used": False,
286 "packages_imported": False,
287 "native_code_loaded": False,
288 "artifacts_executed": False,
289 "archives_extracted": False,
290 "symlinks_allowed": False,
291 },
292 "expected_versions": {
293 "python": args.expected_python_version,
294 "rust_meta_crate": args.expected_rust_version,
295 "cli": args.expected_cli_version,
296 },
297 "summary": {
298 "artifact_count": len(paths),
299 "total_bytes": total,
300 "checksum_verified_count": verified,
301 },
302 "checksum_manifest": checksum_report,
303 "artifacts": summaries,
304 "errors": dict(sorted(errors.items())),
305 "warnings": dict(sorted(warnings.items())),
306 "trust_gate": [
307 "accept only an approved official source and immutable release",
308 "verify the artifact SHA-256 against an independently obtained value",
309 "do not run untrusted binaries or Cargo build scripts",
310 "load or execute only in isolation with CPU, RAM, disk, and time limits",
311 "after trust approval, verify gtars --version or gtars.__version__",
312 ],
313 }
314 return report, 0 if not errors else 2
317def main() -> int:
318 args = build_parser().parse_args()
319 try:
320 report, status = inspect(args)
321 print_json(report)
322 return status
323 except (OSError, SafetyError, UnicodeError) as exc:
324 return fail_json(TOOL, exc)
327if __name__ == "__main__":
328 sys.exit(main())