Skills › Scientific Agent Skills › scientific-brainstorming › _common.py
_common.py
317 lines · skills/scientific-brainstorming/scripts/_common.py at 5f2bc03
1"""Shared standard-library safety helpers for brainstorming CLIs."""
3from __future__ import annotations
5import argparse
6import json
7import math
8import os
9import re
10import stat
11import tempfile
12from collections.abc import Iterable
13from datetime import date
14from pathlib import Path
15from typing import Any
17MAX_INPUT_BYTES = 5 * 1024 * 1024
18MAX_TEXT_CHARS = 10_000
19MAX_SHORT_TEXT_CHARS = 500
20MAX_ITEMS = 5_000
21IDENTIFIER_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
24class CliError(ValueError):
25 """A deterministic, user-facing CLI validation error."""
28def bounded_float(minimum: float, maximum: float):
29 """Return an argparse converter for a finite bounded float."""
31 def convert(raw: str) -> float:
32 try:
33 value = float(raw)
34 except ValueError as exc:
35 raise argparse.ArgumentTypeError(f"expected a number, got {raw!r}") from exc
36 if not math.isfinite(value) or not minimum <= value <= maximum:
37 raise argparse.ArgumentTypeError(
38 f"expected a finite number from {minimum} through {maximum}"
39 )
40 return value
42 return convert
45def require_identifier(value: Any, label: str) -> str:
46 """Validate a stable, bounded identifier."""
48 if not isinstance(value, str) or not IDENTIFIER_PATTERN.fullmatch(value):
49 raise CliError(
50 f"{label} must match {IDENTIFIER_PATTERN.pattern} and be at most "
51 "64 characters"
52 )
53 return value
56def require_text(
57 value: Any,
58 label: str,
59 *,
60 maximum: int = MAX_TEXT_CHARS,
61 allow_empty: bool = False,
62) -> str:
63 """Validate and normalize a bounded text field."""
65 if not isinstance(value, str):
66 raise CliError(f"{label} must be a string")
67 normalized = value.strip()
68 if not normalized and not allow_empty:
69 raise CliError(f"{label} must not be empty")
70 if len(normalized) > maximum:
71 raise CliError(f"{label} is {len(normalized)} characters; limit is {maximum}")
72 return normalized
75def bounded_strings(
76 values: Iterable[str],
77 label: str,
78 *,
79 maximum_items: int = 100,
80 maximum_length: int = MAX_SHORT_TEXT_CHARS,
81) -> list[str]:
82 """Validate a bounded sequence of strings."""
84 result = list(values)
85 if len(result) > maximum_items:
86 raise CliError(f"{label} has {len(result)} items; limit is {maximum_items}")
87 return [
88 require_text(
89 value,
90 f"{label}[{index}]",
91 maximum=maximum_length,
92 )
93 for index, value in enumerate(result)
94 ]
97def require_iso_date(value: str, label: str = "date") -> str:
98 """Validate an ISO calendar date without inventing a current date."""
100 try:
101 parsed = date.fromisoformat(value)
102 except (TypeError, ValueError) as exc:
103 raise CliError(f"{label} must be an ISO date (YYYY-MM-DD)") from exc
104 return parsed.isoformat()
107def finite_number(value: Any, label: str) -> float:
108 """Convert a JSON/CSV value to a finite float, rejecting booleans."""
110 if isinstance(value, bool):
111 raise CliError(f"{label} must be numeric, not boolean")
112 try:
113 result = float(value)
114 except (TypeError, ValueError, OverflowError) as exc:
115 raise CliError(f"{label} must be numeric") from exc
116 if not math.isfinite(result):
117 raise CliError(f"{label} must be finite")
118 return result
121def read_bounded_text(
122 raw_path: str,
123 *,
124 label: str,
125 suffixes: set[str],
126 encoding: str = "utf-8",
127) -> tuple[Path, str]:
128 """Read a bounded regular file without following a final symlink."""
130 path = Path(raw_path).expanduser()
131 if path.name in {"", ".", ".."}:
132 raise CliError(f"{label} must name a file")
133 if path.suffix.lower() not in suffixes:
134 expected = ", ".join(sorted(suffixes))
135 raise CliError(f"{label} must use one of these suffixes: {expected}")
136 if path.is_symlink():
137 raise CliError(f"refusing to read {label} through a symlink")
139 flags = os.O_RDONLY
140 if hasattr(os, "O_NOFOLLOW"):
141 flags |= os.O_NOFOLLOW
142 try:
143 descriptor = os.open(path, flags)
144 except OSError as exc:
145 raise CliError(f"cannot open {label} as a regular file: {path}") from exc
147 try:
148 metadata = os.fstat(descriptor)
149 if not stat.S_ISREG(metadata.st_mode):
150 raise CliError(f"{label} is not a regular file: {path}")
151 if metadata.st_size > MAX_INPUT_BYTES:
152 raise CliError(
153 f"{label} is {metadata.st_size} bytes; limit is {MAX_INPUT_BYTES}"
154 )
155 with os.fdopen(descriptor, "r", encoding=encoding, newline="") as handle:
156 descriptor = -1
157 return path, handle.read()
158 except UnicodeDecodeError as exc:
159 raise CliError(f"{label} is not valid {encoding} text") from exc
160 finally:
161 if descriptor >= 0:
162 os.close(descriptor)
165def read_json(raw_path: str, *, label: str = "JSON input") -> tuple[Path, Any]:
166 """Read strict, bounded JSON and reject non-standard constants."""
168 path, text = read_bounded_text(
169 raw_path,
170 label=label,
171 suffixes={".json"},
172 )
174 def reject_constant(value: str) -> None:
175 raise CliError(f"non-standard JSON constant is not allowed: {value}")
177 def unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
178 result: dict[str, Any] = {}
179 for key, value in pairs:
180 if key in result:
181 raise CliError(f"duplicate JSON key is not allowed: {key!r}")
182 result[key] = value
183 return result
185 try:
186 return path, json.loads(
187 text, parse_constant=reject_constant, object_pairs_hook=unique_object
188 )
189 except (json.JSONDecodeError, RecursionError) as exc:
190 raise CliError(f"invalid JSON in {path}: {exc}") from exc
193def _safe_output_target(raw_path: str, suffix: str) -> Path:
194 """Resolve a user-selected output in an existing directory."""
196 requested = Path(raw_path).expanduser()
197 if requested.name in {"", ".", ".."}:
198 raise CliError("output must name a file")
199 if requested.suffix.lower() != suffix:
200 raise CliError(f"output filename must end in {suffix}")
201 if requested.is_symlink():
202 raise CliError("refusing to write through an output symlink")
203 try:
204 parent = requested.parent.resolve(strict=True)
205 except FileNotFoundError as exc:
206 raise CliError("output parent directory does not exist") from exc
207 if not parent.is_dir():
208 raise CliError("output parent is not a directory")
209 target = parent / requested.name
210 if target.is_symlink():
211 raise CliError("refusing to replace an output symlink")
212 if target.exists() and not target.is_file():
213 raise CliError("output exists and is not a regular file")
214 return target
217def write_json(
218 raw_path: str,
219 payload: Any,
220 *,
221 force: bool = False,
222) -> Path:
223 """Write deterministic JSON with private permissions and safe overwrite."""
225 target = _safe_output_target(raw_path, ".json")
226 try:
227 text = (
228 json.dumps(
229 payload,
230 allow_nan=False,
231 ensure_ascii=False,
232 indent=2,
233 sort_keys=True,
234 )
235 + "\n"
236 )
237 except (TypeError, ValueError, RecursionError) as exc:
238 raise CliError(f"payload cannot be serialized as strict JSON: {exc}") from exc
240 if not force:
241 flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
242 if hasattr(os, "O_NOFOLLOW"):
243 flags |= os.O_NOFOLLOW
244 descriptor = -1
245 created = False
246 try:
247 descriptor = os.open(target, flags, 0o600)
248 created = True
249 os.fchmod(descriptor, 0o600)
250 with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle:
251 descriptor = -1
252 handle.write(text)
253 except FileExistsError as exc:
254 raise CliError(
255 f"output already exists: {target}; pass --force to replace it"
256 ) from exc
257 except OSError as exc:
258 if created:
259 target.unlink(missing_ok=True)
260 raise CliError(f"could not write output: {target}") from exc
261 finally:
262 if descriptor >= 0:
263 os.close(descriptor)
264 return target
266 if target.exists() and not target.is_file():
267 raise CliError("output exists and is not a regular file")
269 descriptor, temporary_name = tempfile.mkstemp(
270 prefix=f".{target.name}.",
271 suffix=".tmp",
272 dir=target.parent,
273 )
274 temporary = Path(temporary_name)
275 try:
276 os.fchmod(descriptor, 0o600)
277 with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as handle:
278 descriptor = -1
279 handle.write(text)
280 if target.is_symlink():
281 raise CliError("refusing to replace an output symlink")
282 os.replace(temporary, target)
283 except OSError as exc:
284 raise CliError(f"could not replace output: {target}") from exc
285 finally:
286 if descriptor >= 0:
287 os.close(descriptor)
288 temporary.unlink(missing_ok=True)
289 return target
292def emit_json(
293 payload: Any,
294 output: str | None,
295 *,
296 force: bool = False,
297) -> Path | None:
298 """Print strict JSON or write it to a safe output path."""
300 if output is None:
301 print(
302 json.dumps(
303 payload,
304 allow_nan=False,
305 ensure_ascii=False,
306 indent=2,
307 sort_keys=True,
308 )
309 )
310 return None
311 return write_json(output, payload, force=force)
314def rounded(value: float) -> float:
315 """Return stable display precision without changing the documented formula."""
317 return round(value, 6)