Skills › Scientific Agent Skills › pathml › _common.py
_common.py
401 lines · skills/pathml/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared bounded-I/O helpers for the PathML skill CLIs."""
4from __future__ import annotations
6import hashlib
7import json
8import math
9import os
10import stat
11import tempfile
12from collections.abc import Iterable, Mapping
13from pathlib import Path
14from typing import Any
17PATHML_VERSION = "3.0.8"
18PINNED_INSTALL = 'uv pip install "pathml==3.0.8"'
19MAX_JSON_BYTES = 16 * 1024 * 1024
20MAX_CSV_BYTES = 64 * 1024 * 1024
21MAX_IMAGE_BYTES = 256 * 1024 * 1024
22MAX_REPORT_BYTES = 8 * 1024 * 1024
23MAX_ROWS = 1_000_000
24MAX_PIXELS = 16_000_000
25MAX_DIMENSION = 1_000_000_000
28class CliError(ValueError):
29 """An expected, concise command-line validation error."""
32def _reject_constant(value: str) -> None:
33 raise CliError(f"non-standard JSON constant is not allowed: {value}")
36def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
37 result: dict[str, Any] = {}
38 for key, value in pairs:
39 if key in result:
40 raise CliError(f"duplicate JSON key is not allowed: {key!r}")
41 result[key] = value
42 return result
45def _reject_url(value: str) -> None:
46 lowered = value.strip().lower()
47 if "://" in lowered or lowered.startswith(
48 ("http:", "https:", "ftp:", "s3:", "gs:", "file:")
49 ):
50 raise CliError("URLs are not accepted; provide a bounded local path")
51 if "\x00" in value:
52 raise CliError("paths must not contain a NUL byte")
55def _absolute_lexical(path: Path) -> Path:
56 """Make a path absolute and collapse dot segments without following links."""
58 return Path(os.path.abspath(os.fspath(path)))
61def _reject_symlink_components(path: Path) -> None:
62 """Reject an existing symlink anywhere in an absolute path."""
64 absolute = _absolute_lexical(path)
65 current = Path(absolute.anchor)
66 for part in absolute.parts[1:]:
67 current /= part
68 try:
69 if current.is_symlink():
70 raise CliError(f"symlink paths are not accepted: {current.name}")
71 except OSError as exc:
72 raise CliError(f"cannot inspect path component {current.name}: {exc}") from exc
75def checked_root(value: str | os.PathLike[str]) -> Path:
76 """Return an existing non-symlink directory used as an I/O boundary."""
78 raw = os.fspath(value)
79 _reject_url(raw)
80 supplied = _absolute_lexical(Path(raw).expanduser())
81 if supplied.is_symlink():
82 raise CliError("root directory must not itself be a symlink")
83 try:
84 root = supplied.resolve(strict=True)
85 info = root.stat()
86 except OSError as exc:
87 raise CliError(f"cannot access root directory: {exc}") from exc
88 if not stat.S_ISDIR(info.st_mode):
89 raise CliError("root must be an existing directory")
90 _reject_symlink_components(root)
91 return root
94def _within_root(candidate: Path, root: Path) -> None:
95 try:
96 candidate.relative_to(root)
97 except ValueError as exc:
98 raise CliError("path escapes the declared root directory") from exc
101def _reject_links_below_root(path: Path, root: Path) -> None:
102 """Inspect requested components before resolution hides directory links.
104 OS aliases above the declared boundary (for example macOS /tmp) are allowed.
105 """
106 current = path
107 while current != current.parent:
108 if current.is_symlink():
109 raise CliError(f"symlink paths are not accepted: {current.name}")
110 if current.resolve() == root:
111 return
112 current = current.parent
115def _suffix_matches(path: Path, suffixes: Iterable[str]) -> bool:
116 name = path.name.lower()
117 return any(name.endswith(suffix.lower()) for suffix in suffixes)
120def checked_input_file(
121 value: str | os.PathLike[str],
122 *,
123 root: str | os.PathLike[str] = ".",
124 suffixes: Iterable[str] | None = None,
125 max_bytes: int,
126) -> Path:
127 """Return a bounded regular local file within root, rejecting symlinks."""
129 raw = os.fspath(value)
130 _reject_url(raw)
131 root_path = checked_root(root)
132 path = Path(raw).expanduser()
133 if not path.is_absolute():
134 path = root_path / path
135 path = _absolute_lexical(path)
136 if path.is_symlink():
137 raise CliError(f"input must not be a symlink: {path.name!r}")
138 _reject_links_below_root(path, root_path)
139 try:
140 resolved = path.resolve(strict=True)
141 info = resolved.stat()
142 except OSError as exc:
143 raise CliError(f"cannot access input file {path.name!r}: {exc}") from exc
144 _within_root(resolved, root_path)
145 _reject_symlink_components(resolved)
146 if not stat.S_ISREG(info.st_mode):
147 raise CliError(f"input is not a regular file: {path.name!r}")
148 if info.st_size > max_bytes:
149 raise CliError(
150 f"input {path.name!r} is {info.st_size} bytes; limit is {max_bytes}"
151 )
152 if suffixes is not None and not _suffix_matches(path, suffixes):
153 allowed = ", ".join(sorted({suffix.lower() for suffix in suffixes}))
154 raise CliError(f"input suffix must be one of: {allowed}")
155 return resolved
158def checked_output_file(
159 value: str | os.PathLike[str],
160 *,
161 root: str | os.PathLike[str] = ".",
162 suffixes: Iterable[str],
163 force: bool = False,
164) -> Path:
165 """Return a local output path within root without following symlinks."""
167 raw = os.fspath(value)
168 _reject_url(raw)
169 root_path = checked_root(root)
170 path = Path(raw).expanduser()
171 if not path.is_absolute():
172 path = root_path / path
173 path = _absolute_lexical(path)
174 if path.name in {"", ".", ".."}:
175 raise CliError("output must name a file")
176 if not _suffix_matches(path, suffixes):
177 allowed = ", ".join(sorted({suffix.lower() for suffix in suffixes}))
178 raise CliError(f"output suffix must be one of: {allowed}")
179 if path.is_symlink():
180 raise CliError(f"output must not be a symlink: {path.name!r}")
181 if path.parent.is_symlink():
182 raise CliError("output parent must not be a symlink")
183 _reject_links_below_root(path.parent, root_path)
184 try:
185 resolved_parent = path.parent.resolve(strict=True)
186 parent_info = resolved_parent.stat()
187 except OSError as exc:
188 raise CliError(f"cannot access output parent: {exc}") from exc
189 _within_root(resolved_parent, root_path)
190 _reject_symlink_components(resolved_parent)
191 path = resolved_parent / path.name
192 if not stat.S_ISDIR(parent_info.st_mode):
193 raise CliError("output parent must be an existing directory")
194 if path.exists():
195 if not path.is_file():
196 raise CliError("output exists and is not a regular file")
197 if not force:
198 raise CliError(f"refusing to overwrite existing output: {path.name!r}")
199 return path
202def strict_json_bytes(document: Any) -> bytes:
203 """Serialize deterministic RFC-compatible JSON with a size cap."""
205 try:
206 payload = (
207 json.dumps(
208 document,
209 indent=2,
210 sort_keys=True,
211 ensure_ascii=False,
212 allow_nan=False,
213 )
214 + "\n"
215 ).encode("utf-8")
216 except (TypeError, ValueError) as exc:
217 raise CliError(f"report is not strict JSON: {exc}") from exc
218 if len(payload) > MAX_REPORT_BYTES:
219 raise CliError(
220 f"report is {len(payload)} bytes; limit is {MAX_REPORT_BYTES}"
221 )
222 return payload
225def atomic_write_bytes(
226 destination: Path,
227 payload: bytes,
228 *,
229 root: str | os.PathLike[str] = ".",
230 suffixes: Iterable[str],
231 force: bool = False,
232) -> Path:
233 """Atomically write a private file in an existing local directory."""
235 destination = checked_output_file(
236 destination, root=root, suffixes=suffixes, force=force
237 )
238 descriptor, temporary_name = tempfile.mkstemp(
239 prefix=f".{destination.name}.", suffix=".tmp", dir=destination.parent
240 )
241 temporary = Path(temporary_name)
242 try:
243 with os.fdopen(descriptor, "wb") as handle:
244 handle.write(payload)
245 handle.flush()
246 os.fsync(handle.fileno())
247 os.chmod(temporary, 0o600)
248 if destination.exists() and not force:
249 raise CliError(f"refusing to overwrite existing output: {destination.name!r}")
250 os.replace(temporary, destination)
251 finally:
252 temporary.unlink(missing_ok=True)
253 return destination
256def emit_json(
257 document: Any,
258 *,
259 output: str | os.PathLike[str] | None = None,
260 root: str | os.PathLike[str] = ".",
261 force: bool = False,
262) -> None:
263 """Print strict JSON or atomically write it with private permissions."""
265 payload = strict_json_bytes(document)
266 if output is None:
267 print(payload.decode("utf-8"), end="")
268 return
269 atomic_write_bytes(
270 Path(output),
271 payload,
272 root=root,
273 suffixes={".json"},
274 force=force,
275 )
278def load_json_object(
279 value: str | os.PathLike[str],
280 *,
281 root: str | os.PathLike[str] = ".",
282 max_bytes: int = MAX_JSON_BYTES,
283) -> dict[str, Any]:
284 """Load a bounded strict JSON object from a local regular file."""
286 path = checked_input_file(
287 value,
288 root=root,
289 suffixes={".json"},
290 max_bytes=max_bytes,
291 )
292 try:
293 with path.open("r", encoding="utf-8") as handle:
294 document = json.load(
295 handle,
296 parse_constant=_reject_constant,
297 object_pairs_hook=_unique_object,
298 )
299 except CliError:
300 raise
301 except (OSError, UnicodeError, json.JSONDecodeError) as exc:
302 raise CliError(f"cannot read valid JSON from {path.name!r}: {exc}") from exc
303 if not isinstance(document, dict):
304 raise CliError("JSON root must be an object")
305 return document
308def validate_keys(
309 value: Mapping[str, Any],
310 *,
311 allowed: Iterable[str],
312 required: Iterable[str] = (),
313 context: str,
314) -> None:
315 """Reject unknown keys and report required keys."""
317 allowed_set = set(allowed)
318 required_set = set(required)
319 unknown = sorted(set(value) - allowed_set)
320 missing = sorted(required_set - set(value))
321 if unknown:
322 raise CliError(f"{context} has unknown keys: {', '.join(unknown)}")
323 if missing:
324 raise CliError(f"{context} is missing keys: {', '.join(missing)}")
327def bounded_int(
328 value: Any,
329 *,
330 name: str,
331 minimum: int,
332 maximum: int,
333) -> int:
334 """Validate an integer, excluding booleans."""
336 if isinstance(value, bool) or not isinstance(value, int):
337 raise CliError(f"{name} must be an integer")
338 if not minimum <= value <= maximum:
339 raise CliError(f"{name} must be between {minimum} and {maximum}")
340 return value
343def finite_float(
344 value: Any,
345 *,
346 name: str,
347 minimum: float | None = None,
348 maximum: float | None = None,
349) -> float:
350 """Validate a finite real number, excluding booleans."""
352 if isinstance(value, bool) or not isinstance(value, (int, float)):
353 raise CliError(f"{name} must be numeric")
354 number = float(value)
355 if not math.isfinite(number):
356 raise CliError(f"{name} must be finite")
357 if minimum is not None and number < minimum:
358 raise CliError(f"{name} must be at least {minimum}")
359 if maximum is not None and number > maximum:
360 raise CliError(f"{name} must be at most {maximum}")
361 return number
364def parse_name_list(value: str | None, *, name: str) -> list[str]:
365 """Parse a comma-separated list without empty or duplicate names."""
367 if value is None:
368 return []
369 names = [item.strip() for item in value.split(",")]
370 if not names or any(not item for item in names):
371 raise CliError(f"{name} must be a comma-separated list of nonempty names")
372 if len(names) != len(set(names)):
373 raise CliError(f"{name} must not contain duplicates")
374 return names
377def sha256_file(path: Path, *, max_bytes: int) -> str:
378 """Hash a bounded regular file using fixed-size streaming reads."""
380 size = path.stat().st_size
381 if size > max_bytes:
382 raise CliError(f"file is {size} bytes; hashing limit is {max_bytes}")
383 digest = hashlib.sha256()
384 with path.open("rb") as handle:
385 for chunk in iter(lambda: handle.read(1024 * 1024), b""):
386 digest.update(chunk)
387 return digest.hexdigest()
390def run_cli(function: Any) -> int:
391 """Run a CLI body with concise expected-error handling."""
393 try:
394 function()
395 except CliError as exc:
396 print(f"error: {exc}", file=os.sys.stderr)
397 return 2
398 except KeyboardInterrupt:
399 print("error: interrupted", file=os.sys.stderr)
400 return 130
401 return 0