Skills › Scientific Agent Skills › matlab › _common.py
_common.py
264 lines · skills/matlab/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared bounded local-only helpers for the MATLAB skill CLIs."""
4from __future__ import annotations
6import hashlib
7import json
8import re
9import sys
10from pathlib import Path
11from typing import Any, Iterable
13MAX_INPUT_BYTES = 64 * 1024 * 1024
14MAX_TEXT_BYTES = 4 * 1024 * 1024
15MAX_JSON_BYTES = 2 * 1024 * 1024
16MAX_FILES = 500
17MAX_JSON_ITEMS = 20_000
18MAX_JSON_DEPTH = 24
19MAX_PATH_CHARS = 4096
20# Conservative helper portability subset, not modern MATLAB's namelengthmax.
21MATLAB_IDENTIFIER = re.compile(r"^[A-Za-z][A-Za-z0-9_]{0,62}$")
22MATLAB_RELEASE = re.compile(r"^R(20[0-9]{2})([ab])$")
23URL_LIKE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://")
26class CliError(ValueError):
27 """Expected, user-actionable CLI failure."""
30def _raw_path(value: str) -> Path:
31 if not value or len(value) > MAX_PATH_CHARS:
32 raise CliError("path is empty or exceeds the length limit")
33 if "\x00" in value or URL_LIKE.match(value):
34 raise CliError("only local filesystem paths are accepted")
35 return Path(value)
38def _reject_symlink_chain(path: Path) -> None:
39 """Reject every existing symlink component without following it."""
40 absolute = path.absolute()
41 current = Path(absolute.anchor)
42 for part in absolute.parts[1:]:
43 current = current / part
44 if current.exists() and current.is_symlink():
45 raise CliError(f"symlink paths are not accepted: {current}")
48def checked_root(value: str | Path) -> Path:
49 raw = _raw_path(str(value))
50 _reject_symlink_chain(raw)
51 try:
52 root = raw.resolve(strict=True)
53 except OSError as exc:
54 raise CliError(f"root does not exist or is inaccessible: {raw}") from exc
55 if not root.is_dir():
56 raise CliError(f"root is not a directory: {root}")
57 return root
60def checked_input(
61 value: str | Path,
62 *,
63 root: Path,
64 kind: str = "file",
65 suffixes: Iterable[str] | None = None,
66 max_bytes: int = MAX_INPUT_BYTES,
67) -> Path:
68 raw = _raw_path(str(value))
69 candidate = raw if raw.is_absolute() else root / raw
70 _reject_symlink_chain(candidate)
71 try:
72 path = candidate.resolve(strict=True)
73 path.relative_to(root)
74 except (OSError, ValueError) as exc:
75 raise CliError(f"input must exist within root {root}") from exc
76 if kind == "file" and not path.is_file():
77 raise CliError(f"input is not a regular file: {path}")
78 if kind == "directory" and not path.is_dir():
79 raise CliError(f"input is not a directory: {path}")
80 if kind == "any" and not (path.is_file() or path.is_dir()):
81 raise CliError(f"input is not a regular file or directory: {path}")
82 if kind not in {"file", "directory", "any"}:
83 raise CliError(f"unsupported input kind: {kind}")
84 if suffixes is not None and path.is_file():
85 allowed = {suffix.casefold() for suffix in suffixes}
86 if path.suffix.casefold() not in allowed:
87 raise CliError(
88 f"unsupported suffix {path.suffix!r}; expected one of {sorted(allowed)}"
89 )
90 if path.is_file():
91 try:
92 size = path.stat().st_size
93 except OSError as exc:
94 raise CliError(f"cannot stat input: {path}") from exc
95 if size > max_bytes:
96 raise CliError(f"input is {size} bytes; limit is {max_bytes}")
97 return path
100def checked_output(
101 value: str | Path,
102 *,
103 root: Path,
104 suffixes: Iterable[str] | None = None,
105) -> Path:
106 raw = _raw_path(str(value))
107 candidate = raw if raw.is_absolute() else root / raw
108 _reject_symlink_chain(candidate)
109 try:
110 parent = candidate.parent.resolve(strict=True)
111 parent.relative_to(root)
112 except (OSError, ValueError) as exc:
113 raise CliError(f"output parent must exist within root {root}") from exc
114 path = parent / candidate.name
115 if path.exists() or path.is_symlink():
116 raise CliError(f"refusing to overwrite existing output: {path}")
117 if suffixes is not None:
118 allowed = {suffix.casefold() for suffix in suffixes}
119 if path.suffix.casefold() not in allowed:
120 raise CliError(
121 f"unsupported output suffix {path.suffix!r}; "
122 f"expected one of {sorted(allowed)}"
123 )
124 return path
127def read_bytes(path: Path, *, max_bytes: int = MAX_INPUT_BYTES) -> bytes:
128 try:
129 size = path.stat().st_size
130 if size > max_bytes:
131 raise CliError(f"input is {size} bytes; limit is {max_bytes}")
132 return path.read_bytes()
133 except CliError:
134 raise
135 except OSError as exc:
136 raise CliError(f"cannot read input: {path}") from exc
139def read_text(path: Path, *, max_bytes: int = MAX_TEXT_BYTES) -> str:
140 payload = read_bytes(path, max_bytes=max_bytes)
141 try:
142 return payload.decode("utf-8")
143 except UnicodeDecodeError as exc:
144 raise CliError(f"input is not UTF-8 text: {path}") from exc
147def _reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
148 output: dict[str, Any] = {}
149 for key, value in pairs:
150 if key in output:
151 raise CliError(f"duplicate JSON key: {key!r}")
152 output[key] = value
153 return output
156def _validate_json_shape(value: Any, *, depth: int = 0) -> int:
157 if depth > MAX_JSON_DEPTH:
158 raise CliError(f"JSON nesting exceeds {MAX_JSON_DEPTH}")
159 if value is None or isinstance(value, (bool, int, float, str)):
160 if isinstance(value, str) and len(value) > 100_000:
161 raise CliError("JSON string exceeds 100000 characters")
162 return 1
163 if isinstance(value, list):
164 return 1 + sum(_validate_json_shape(item, depth=depth + 1) for item in value)
165 if isinstance(value, dict):
166 if not all(isinstance(key, str) for key in value):
167 raise CliError("JSON object keys must be strings")
168 return 1 + sum(
169 _validate_json_shape(item, depth=depth + 1) for item in value.values()
170 )
171 raise CliError(f"unsupported JSON value type: {type(value).__name__}")
174def parse_json_text(text: str) -> Any:
175 try:
176 value = json.loads(text, object_pairs_hook=_reject_duplicate_keys)
177 except CliError:
178 raise
179 except (json.JSONDecodeError, ValueError) as exc:
180 raise CliError(f"invalid JSON: {exc}") from exc
181 item_count = _validate_json_shape(value)
182 if item_count > MAX_JSON_ITEMS:
183 raise CliError(f"JSON has {item_count} values; limit is {MAX_JSON_ITEMS}")
184 return value
187def load_json(path: Path) -> Any:
188 return parse_json_text(read_text(path, max_bytes=MAX_JSON_BYTES))
191def bounded_int(
192 value: str | int,
193 *,
194 name: str,
195 minimum: int,
196 maximum: int,
197) -> int:
198 try:
199 number = int(value)
200 except (TypeError, ValueError) as exc:
201 raise CliError(f"{name} must be an integer") from exc
202 if number < minimum or number > maximum:
203 raise CliError(f"{name} must be between {minimum} and {maximum}")
204 return number
207def validate_identifier(value: str, *, name: str = "identifier") -> str:
208 if not MATLAB_IDENTIFIER.fullmatch(value):
209 raise CliError(
210 f"{name} must use the helper's MATLAB identifier subset (at most 63 characters)"
211 )
212 return value
215def validate_release(value: str) -> str:
216 if not MATLAB_RELEASE.fullmatch(value):
217 raise CliError("MATLAB release must look like R2026a")
218 return value
221def sha256_file(path: Path, *, max_bytes: int = MAX_INPUT_BYTES) -> str:
222 size = path.stat().st_size
223 if size > max_bytes:
224 raise CliError(f"input is {size} bytes; hash limit is {max_bytes}")
225 digest = hashlib.sha256()
226 try:
227 with path.open("rb") as handle:
228 while chunk := handle.read(64 * 1024):
229 digest.update(chunk)
230 except OSError as exc:
231 raise CliError(f"cannot hash input: {path}") from exc
232 return digest.hexdigest()
235def relative_id(path: Path, root: Path) -> str:
236 return path.relative_to(root).as_posix()
239def write_new_text(path: Path, text: str) -> None:
240 try:
241 with path.open("x", encoding="utf-8", newline="\n") as handle:
242 handle.write(text)
243 except FileExistsError as exc:
244 raise CliError(f"refusing to overwrite existing output: {path}") from exc
245 except OSError as exc:
246 raise CliError(f"cannot write output: {path}") from exc
249def emit_json(value: Any) -> None:
250 json.dump(value, sys.stdout, indent=2, sort_keys=True, ensure_ascii=False)
251 sys.stdout.write("\n")
254def fail_json(tool: str, exc: Exception) -> int:
255 emit_json(
256 {
257 "error": str(exc),
258 "executes_external_code": False,
259 "network_accessed": False,
260 "ok": False,
261 "tool": tool,
262 }
263 )
264 return 2