Skills › Scientific Agent Skills › matlab › inventory_mat_file.py
inventory_mat_file.py
355 lines · skills/matlab/scripts/inventory_mat_file.py at 5f2bc03
1#!/usr/bin/env python3
2"""Inventory bounded MAT/HDF5 metadata without deserializing object values."""
4from __future__ import annotations
6import argparse
7import hashlib
8import sys
9from collections import Counter
10from pathlib import Path
11from typing import Any
13from _common import (
14 CliError,
15 bounded_int,
16 checked_input,
17 checked_root,
18 emit_json,
19 fail_json,
20 relative_id,
21 sha256_file,
22)
24TOOL = "inventory_mat_file"
25HDF5_SIGNATURE = b"\x89HDF\r\n\x1a\n"
26OBJECT_LIKE_CLASSES = {
27 "cell",
28 "function",
29 "java",
30 "object",
31 "opaque",
32 "struct",
33 "table",
34 "timetable",
35}
38def identify_header(path: Path) -> dict[str, Any]:
39 try:
40 with path.open("rb") as handle:
41 prefix = handle.read(8192)
42 except OSError as exc:
43 raise CliError(f"cannot read MAT header: {path}") from exc
44 signature_offset = next(
45 (
46 offset
47 for offset in (0, 512, 1024, 2048, 4096)
48 if prefix[offset : offset + 8] == HDF5_SIGNATURE
49 ),
50 None,
51 )
52 header = prefix[:128]
53 if header.startswith(b"MATLAB 7.3 MAT-file"):
54 kind = "matlab_v7_3_hdf5"
55 elif header.startswith(b"MATLAB 5.0 MAT-file"):
56 kind = "matlab_level5_v6_or_v7"
57 elif signature_offset is not None:
58 kind = "hdf5_not_confirmed_matlab_v7_3"
59 elif prefix.startswith(b"\x80"):
60 kind = "python_pickle_signature_refused"
61 else:
62 kind = "unknown_or_mat_v4"
63 return {
64 "detected_kind": kind,
65 "hdf5_signature_offset": signature_offset,
66 "matlab_text_header_present": header.startswith(b"MATLAB "),
67 }
70def _redacted_name(name: str, index: int, kind: str) -> dict[str, Any]:
71 return {
72 "id": f"{kind}-{index:06d}",
73 "name_emitted": False,
74 "name_sha256": hashlib.sha256(name.encode("utf-8", "surrogatepass")).hexdigest(),
75 }
78def scipy_inventory(
79 path: Path, *, max_nodes: int
80) -> tuple[list[dict[str, Any]], list[str]]:
81 try:
82 from scipy import io as scipy_io
83 except ImportError as exc:
84 raise CliError(
85 "SciPy is optional and not installed; use --backend header or "
86 "install a pinned scipy in an approved environment"
87 ) from exc
88 try:
89 entries = scipy_io.whosmat(str(path), appendmat=False)
90 except Exception as exc:
91 raise CliError(f"SciPy could not inventory MAT metadata: {exc}") from exc
92 if len(entries) > max_nodes:
93 raise CliError(f"MAT file has more than {max_nodes} variables")
94 output: list[dict[str, Any]] = []
95 warnings: list[str] = []
96 for index, (name, shape, class_name) in enumerate(entries, start=1):
97 class_text = str(class_name)
98 record = {
99 **_redacted_name(str(name), index, "variable"),
100 "class": class_text,
101 "object_like": class_text.casefold() in OBJECT_LIKE_CLASSES,
102 "shape": [int(value) for value in shape],
103 "values_loaded": False,
104 }
105 if record["object_like"]:
106 warnings.append(
107 f"{record['id']} has object-like class {class_text!r}; "
108 "do not load without expert review"
109 )
110 output.append(record)
111 return output, warnings
114def hdf5_inventory(
115 path: Path, *, max_nodes: int, max_depth: int
116) -> tuple[list[dict[str, Any]], list[str], dict[str, int]]:
117 try:
118 import h5py
119 except ImportError as exc:
120 raise CliError(
121 "h5py is optional and not installed; use --backend header or "
122 "install a pinned h5py in an approved environment"
123 ) from exc
125 records: list[dict[str, Any]] = []
126 warnings: list[str] = []
127 link_counts: Counter[str] = Counter()
128 seen_objects: set[int] = set()
130 def append_record(path_name: str, kind: str, details: dict[str, Any]) -> None:
131 if len(records) >= max_nodes:
132 raise CliError(f"HDF5 object/link count exceeds {max_nodes}")
133 records.append(
134 {
135 **_redacted_name(path_name, len(records) + 1, "hdf5-node"),
136 "kind": kind,
137 **details,
138 }
139 )
141 def walk(group: Any, prefix: str, depth: int) -> None:
142 if depth > max_depth:
143 raise CliError(f"HDF5 group depth exceeds {max_depth}")
144 try:
145 names = sorted(group.keys())
146 except Exception as exc:
147 raise CliError(f"cannot enumerate HDF5 group metadata: {exc}") from exc
148 for name in names:
149 full_name = f"{prefix}/{name}" if prefix else f"/{name}"
150 try:
151 link = group.get(name, getlink=True)
152 except Exception as exc:
153 raise CliError(f"cannot inspect HDF5 link metadata: {exc}") from exc
154 if isinstance(link, h5py.SoftLink):
155 link_counts["soft"] += 1
156 append_record(
157 full_name,
158 "soft_link",
159 {"followed": False, "target_emitted": False},
160 )
161 warnings.append("soft HDF5 link found and not followed")
162 continue
163 if isinstance(link, h5py.ExternalLink):
164 link_counts["external"] += 1
165 append_record(
166 full_name,
167 "external_link",
168 {"followed": False, "target_emitted": False},
169 )
170 warnings.append("external HDF5 link found and not followed")
171 continue
172 link_counts["hard"] += 1
173 try:
174 obj = group[name]
175 object_key = hash(obj.id)
176 except Exception as exc:
177 raise CliError(f"cannot inspect HDF5 object metadata: {exc}") from exc
178 attribute_names = sorted(str(key) for key in obj.attrs.keys())
179 if len(attribute_names) > 1000:
180 raise CliError("HDF5 object has more than 1000 attributes")
181 base = {
182 "attribute_count": len(attribute_names),
183 "attribute_names_emitted": False,
184 "attribute_name_hashes": [
185 hashlib.sha256(name.encode("utf-8")).hexdigest()
186 for name in attribute_names
187 ],
188 "hard_link_revisited": object_key in seen_objects,
189 "values_loaded": False,
190 }
191 if isinstance(obj, h5py.Dataset):
192 dtype_text = str(obj.dtype)
193 if len(dtype_text) > 500:
194 dtype_text = dtype_text[:500] + "..."
195 reference_dtype = h5py.check_dtype(ref=obj.dtype) is not None
196 object_like = bool(reference_dtype or obj.dtype.kind == "O")
197 append_record(
198 full_name,
199 "dataset",
200 {
201 **base,
202 "chunks": list(obj.chunks) if obj.chunks is not None else None,
203 "compression": obj.compression,
204 "dtype": dtype_text,
205 "object_like": object_like,
206 "shape": (
207 [int(value) for value in obj.shape]
208 if obj.shape is not None else None
209 ),
210 "null_dataspace": obj.shape is None,
211 },
212 )
213 if object_like:
214 warnings.append(
215 "HDF5 object/reference dtype found; values were not read"
216 )
217 seen_objects.add(object_key)
218 elif isinstance(obj, h5py.Group):
219 append_record(full_name, "group", base)
220 if object_key not in seen_objects:
221 seen_objects.add(object_key)
222 walk(obj, full_name, depth + 1)
223 else:
224 append_record(full_name, "unknown_hdf5_object", base)
225 warnings.append("unknown HDF5 object type found")
227 try:
228 with h5py.File(path, "r") as handle:
229 walk(handle, "", 0)
230 except CliError:
231 raise
232 except Exception as exc:
233 raise CliError(f"h5py could not inventory HDF5 metadata: {exc}") from exc
234 return records, warnings, dict(sorted(link_counts.items()))
237def inventory(args: argparse.Namespace) -> dict[str, Any]:
238 root = checked_root(args.root)
239 max_file_bytes = bounded_int(
240 args.max_file_bytes,
241 name="max_file_bytes",
242 minimum=1,
243 maximum=512 * 1024 * 1024,
244 )
245 max_nodes = bounded_int(
246 args.max_nodes, name="max_nodes", minimum=1, maximum=100_000
247 )
248 max_depth = bounded_int(
249 args.max_depth, name="max_depth", minimum=1, maximum=128
250 )
251 path = checked_input(
252 args.input,
253 root=root,
254 suffixes={".mat"},
255 max_bytes=max_file_bytes,
256 )
257 header = identify_header(path)
258 warnings = [
259 "Inventory is metadata triage, not a safety certificate; do not load "
260 "untrusted MAT/HDF5 content."
261 ]
262 backend_used = "header"
263 records: list[dict[str, Any]] = []
264 link_counts: dict[str, int] = {}
265 backend = args.backend
266 if backend == "auto":
267 backend = (
268 "hdf5"
269 if header["detected_kind"]
270 in {"matlab_v7_3_hdf5", "hdf5_not_confirmed_matlab_v7_3"}
271 else "scipy"
272 )
273 if backend == "scipy":
274 if header["detected_kind"] in {
275 "matlab_v7_3_hdf5",
276 "hdf5_not_confirmed_matlab_v7_3",
277 }:
278 raise CliError("SciPy metadata backend is not used for HDF5/v7.3")
279 records, extra = scipy_inventory(path, max_nodes=max_nodes)
280 warnings.extend(extra)
281 backend_used = "scipy.io.whosmat"
282 elif backend == "hdf5":
283 if header["hdf5_signature_offset"] is None:
284 raise CliError("HDF5 signature was not found at a recognized userblock offset")
285 records, extra, link_counts = hdf5_inventory(
286 path, max_nodes=max_nodes, max_depth=max_depth
287 )
288 warnings.extend(extra)
289 backend_used = "h5py-metadata"
290 elif backend != "header":
291 raise CliError(f"unsupported backend: {backend}")
292 if header["detected_kind"] == "python_pickle_signature_refused":
293 warnings.append(
294 "Python pickle signature found under .mat suffix; never deserialize it"
295 )
296 object_like_count = sum(bool(record.get("object_like")) for record in records)
297 return {
298 "backend": backend_used,
299 "deserializes_objects": False,
300 "executes": False,
301 "file": relative_id(path, root),
302 "file_name_emitted": True,
303 "file_sha256": sha256_file(path, max_bytes=max_file_bytes),
304 "file_size_bytes": path.stat().st_size,
305 "header": header,
306 "hdf5_links": link_counts,
307 "name_values_emitted": False,
308 "network_accessed": False,
309 "object_like_count": object_like_count,
310 "ok": (
311 header["detected_kind"] != "python_pickle_signature_refused"
312 and object_like_count == 0
313 and link_counts.get("external", 0) == 0
314 ),
315 "records": records,
316 "records_count": len(records),
317 "safe_to_load": False,
318 "tool": TOOL,
319 "values_loaded": False,
320 "warnings": sorted(set(warnings)),
321 }
324def parser() -> argparse.ArgumentParser:
325 result = argparse.ArgumentParser(
326 description=(
327 "Inventory local MAT technical metadata without MATLAB, loadmat, "
328 "pickle, dataset reads, or object deserialization. Header-only is default."
329 )
330 )
331 result.add_argument("input", help="local .mat file")
332 result.add_argument("--root", default=".", help="allowed local root")
333 result.add_argument(
334 "--backend",
335 choices=("header", "auto", "scipy", "hdf5"),
336 default="header",
337 help="optional metadata backend; header is dependency-free and safest",
338 )
339 result.add_argument("--max-file-bytes", default=64 * 1024 * 1024)
340 result.add_argument("--max-nodes", default=5000)
341 result.add_argument("--max-depth", default=24)
342 return result
345def main() -> int:
346 try:
347 report = inventory(parser().parse_args())
348 emit_json(report)
349 return 0 if report["ok"] else 1
350 except CliError as exc:
351 return fail_json(TOOL, exc)
354if __name__ == "__main__":
355 sys.exit(main())