Skills › Scientific Agent Skills › hypothesis-generation › _common.py
_common.py
437 lines · skills/hypothesis-generation/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared, dependency-free safety helpers for local hypothesis CLIs."""
4from __future__ import annotations
6import csv
7import json
8import os
9import re
10import sys
11import tempfile
12from datetime import date
13from pathlib import Path
14from typing import Any, Iterable
16MAX_INPUT_BYTES = 2 * 1024 * 1024
17MAX_ROWS = 1_000
18MAX_CELL_CHARS = 8_000
19MAX_TEXT_CHARS = 20_000
20MAX_LIST_ITEMS = 500
21MAX_JSON_DEPTH = 64
23IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._:-]{0,95}$")
24URL_SCHEME_RE = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://")
25HTTPS_URL_RE = re.compile(r"^https://[^\s]+$", re.IGNORECASE)
26PARTIAL_DATE_RE = re.compile(r"^\d{4}(?:-\d{2}(?:-\d{2})?)?$")
27ISO_DATE_RE = re.compile(r"^[0-9]{4}-[0-9]{2}-[0-9]{2}$")
30class ValidationError(ValueError):
31 """A deterministic, user-correctable validation failure."""
34def _duplicate_safe_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
35 result: dict[str, Any] = {}
36 for key, value in pairs:
37 if key in result:
38 raise ValidationError(f"JSON object contains duplicate key: {key}")
39 result[key] = value
40 return result
43def _reject_json_constant(value: str) -> None:
44 raise ValidationError(f"JSON contains a nonstandard numeric constant: {value}")
47def _reject_url_like_path(raw_path: str | Path, context: str) -> None:
48 if URL_SCHEME_RE.match(str(raw_path).strip()):
49 raise ValidationError(f"{context} must be a local file path")
52def safe_input_path(raw_path: str | Path, suffixes: Iterable[str]) -> Path:
53 """Resolve a bounded regular local file and reject symlink inputs."""
54 _reject_url_like_path(raw_path, "input")
55 path = Path(raw_path).expanduser()
56 if path.is_symlink():
57 raise ValidationError(f"symlink inputs are not allowed: {path}")
58 try:
59 resolved = path.resolve(strict=True)
60 except FileNotFoundError as exc:
61 raise ValidationError(f"input file does not exist: {path}") from exc
62 if not resolved.is_file():
63 raise ValidationError(f"input path is not a regular file: {resolved}")
64 allowed = {suffix.lower() for suffix in suffixes}
65 if resolved.suffix.lower() not in allowed:
66 choices = ", ".join(sorted(allowed))
67 raise ValidationError(f"expected one of [{choices}]: {resolved}")
68 size = resolved.stat().st_size
69 if size > MAX_INPUT_BYTES:
70 raise ValidationError(
71 f"input exceeds {MAX_INPUT_BYTES} bytes: {resolved} ({size} bytes)"
72 )
73 return resolved
76def safe_output_path(
77 raw_path: str | Path, suffix: str, *, force: bool = False
78) -> Path:
79 """Resolve an output in an existing directory without implicit overwrite."""
80 _reject_url_like_path(raw_path, "output")
81 path = Path(raw_path).expanduser()
82 if path.suffix.lower() != suffix.lower():
83 raise ValidationError(f"output must use {suffix}: {path}")
84 try:
85 parent = path.parent.resolve(strict=True)
86 except FileNotFoundError as exc:
87 raise ValidationError(f"output parent does not exist: {path.parent}") from exc
88 if not parent.is_dir():
89 raise ValidationError(f"output parent is not a directory: {parent}")
90 resolved = parent / path.name
91 if resolved.is_symlink():
92 raise ValidationError(f"symlink outputs are not allowed: {resolved}")
93 if resolved.exists():
94 if not resolved.is_file():
95 raise ValidationError(f"output is not a regular file: {resolved}")
96 if not force:
97 raise ValidationError(
98 f"output already exists; pass --force to replace it: {resolved}"
99 )
100 return resolved
103def read_json(raw_path: str | Path) -> Any:
104 """Read strict UTF-8 JSON with duplicate-key detection."""
105 path = safe_input_path(raw_path, {".json"})
106 try:
107 text = path.read_text(encoding="utf-8")
108 except UnicodeDecodeError as exc:
109 raise ValidationError(f"JSON must be UTF-8: {path}") from exc
110 if "\x00" in text:
111 raise ValidationError(f"JSON contains a NUL byte: {path}")
112 try:
113 payload = json.loads(
114 text,
115 object_pairs_hook=_duplicate_safe_object,
116 parse_constant=_reject_json_constant,
117 )
118 except json.JSONDecodeError as exc:
119 raise ValidationError(
120 f"invalid JSON at line {exc.lineno}, column {exc.colno}: {exc.msg}"
121 ) from exc
122 except RecursionError as exc:
123 raise ValidationError("JSON nesting exceeds the parser limit") from exc
124 # The decoder's recursion threshold varies between Python releases.
125 # Enforce our own bound without a second recursive walk.
126 pending = [(payload, 1)]
127 while pending:
128 value, depth = pending.pop()
129 if isinstance(value, (dict, list)):
130 if depth > MAX_JSON_DEPTH:
131 raise ValidationError(f"JSON nesting exceeds {MAX_JSON_DEPTH} levels")
132 children = value.values() if isinstance(value, dict) else value
133 pending.extend((child, depth + 1) for child in children)
134 return payload
137def read_markdown(raw_path: str | Path) -> str:
138 """Read one bounded UTF-8 Markdown file."""
139 path = safe_input_path(raw_path, {".md", ".markdown"})
140 try:
141 text = path.read_text(encoding="utf-8")
142 except UnicodeDecodeError as exc:
143 raise ValidationError(f"Markdown must be UTF-8: {path}") from exc
144 if "\x00" in text:
145 raise ValidationError(f"Markdown contains a NUL byte: {path}")
146 if any(len(line) > MAX_CELL_CHARS for line in text.splitlines()):
147 raise ValidationError(
148 f"Markdown contains a line longer than {MAX_CELL_CHARS} characters"
149 )
150 return text
153def read_csv_records(
154 raw_path: str | Path,
155 *,
156 fields: Iterable[str],
157 max_rows: int = MAX_ROWS,
158) -> list[dict[str, str]]:
159 """Read strict UTF-8 CSV with an exact ordered header and bounded cells."""
160 path = safe_input_path(raw_path, {".csv"})
161 expected = tuple(fields)
162 try:
163 with path.open("r", encoding="utf-8-sig", newline="") as handle:
164 reader = csv.DictReader(handle)
165 headers = reader.fieldnames
166 if not headers:
167 raise ValidationError(f"CSV has no header: {path}")
168 normalized = tuple(header.strip() for header in headers)
169 if any(not header for header in normalized):
170 raise ValidationError("CSV headers must not be blank")
171 if len(normalized) != len(set(normalized)):
172 raise ValidationError("CSV headers must be unique")
173 if normalized != expected:
174 raise ValidationError(
175 "CSV header must exactly match: " + ",".join(expected)
176 )
177 reader.fieldnames = list(normalized)
178 records: list[dict[str, str]] = []
179 for line_number, row in enumerate(reader, start=2):
180 if line_number - 1 > max_rows:
181 raise ValidationError(f"CSV exceeds {max_rows} data rows")
182 if None in row:
183 raise ValidationError(
184 f"row {line_number} has more cells than the header"
185 )
186 cleaned: dict[str, str] = {}
187 for key, value in row.items():
188 cell = "" if value is None else value.strip()
189 if "\x00" in cell:
190 raise ValidationError(
191 f"row {line_number}, column {key} contains a NUL byte"
192 )
193 if len(cell) > MAX_CELL_CHARS:
194 raise ValidationError(
195 f"row {line_number}, column {key} exceeds "
196 f"{MAX_CELL_CHARS} characters"
197 )
198 cleaned[key] = cell
199 if any(cleaned.values()):
200 records.append(cleaned)
201 except UnicodeDecodeError as exc:
202 raise ValidationError(f"CSV must be UTF-8: {path}") from exc
203 except csv.Error as exc:
204 raise ValidationError(f"invalid CSV: {exc}") from exc
205 if not records:
206 raise ValidationError("CSV must contain at least one data row")
207 return records
210def require_object(value: Any, context: str) -> dict[str, Any]:
211 if not isinstance(value, dict):
212 raise ValidationError(f"{context} must be a JSON object")
213 return value
216def require_exact_keys(
217 value: dict[str, Any],
218 *,
219 required: Iterable[str],
220 optional: Iterable[str] = (),
221 context: str,
222) -> None:
223 required_set = set(required)
224 allowed = required_set | set(optional)
225 missing = sorted(required_set - set(value))
226 unknown = sorted(set(value) - allowed)
227 if missing:
228 raise ValidationError(f"{context} is missing fields: {', '.join(missing)}")
229 if unknown:
230 raise ValidationError(f"{context} has unknown fields: {', '.join(unknown)}")
233def require_list(
234 value: Any,
235 context: str,
236 *,
237 minimum: int = 0,
238 maximum: int = MAX_LIST_ITEMS,
239) -> list[Any]:
240 if not isinstance(value, list):
241 raise ValidationError(f"{context} must be a JSON array")
242 if not minimum <= len(value) <= maximum:
243 raise ValidationError(
244 f"{context} must contain between {minimum} and {maximum} items"
245 )
246 return value
249def require_text(
250 value: Any,
251 context: str,
252 *,
253 allow_empty: bool = False,
254 minimum: int = 1,
255 maximum: int = MAX_TEXT_CHARS,
256) -> str:
257 if not isinstance(value, str):
258 raise ValidationError(f"{context} must be a string")
259 text = value.strip()
260 if not allow_empty and not text:
261 raise ValidationError(f"{context} must not be empty")
262 if text and len(text) < minimum:
263 raise ValidationError(f"{context} must contain at least {minimum} characters")
264 if len(text) > maximum:
265 raise ValidationError(f"{context} exceeds {maximum} characters")
266 if "\x00" in text:
267 raise ValidationError(f"{context} contains a NUL byte")
268 return text
271def require_bool(value: Any, context: str) -> bool:
272 if type(value) is not bool:
273 raise ValidationError(f"{context} must be true or false")
274 return value
277def require_enum(value: Any, choices: Iterable[str], context: str) -> str:
278 text = require_text(value, context, maximum=96)
279 allowed = set(choices)
280 if text not in allowed:
281 raise ValidationError(
282 f"{context} must be one of: {', '.join(sorted(allowed))}"
283 )
284 return text
287def require_identifier(value: Any, context: str) -> str:
288 identifier = require_text(value, context, maximum=96)
289 if not IDENTIFIER_RE.fullmatch(identifier):
290 raise ValidationError(f"{context} has an invalid identifier format")
291 return identifier
294def require_unique(values: Iterable[str], context: str) -> None:
295 seen: set[str] = set()
296 duplicates: set[str] = set()
297 for value in values:
298 if value in seen:
299 duplicates.add(value)
300 seen.add(value)
301 if duplicates:
302 raise ValidationError(
303 f"{context} contains duplicate IDs: {', '.join(sorted(duplicates))}"
304 )
307def require_text_list(
308 value: Any,
309 context: str,
310 *,
311 minimum: int = 0,
312 maximum: int = 100,
313 item_maximum: int = 1_000,
314) -> list[str]:
315 items = require_list(value, context, minimum=minimum, maximum=maximum)
316 return [
317 require_text(item, f"{context}[{index}]", maximum=item_maximum)
318 for index, item in enumerate(items)
319 ]
322def require_identifier_list(
323 value: Any,
324 context: str,
325 *,
326 minimum: int = 0,
327 maximum: int = 100,
328) -> list[str]:
329 items = require_list(value, context, minimum=minimum, maximum=maximum)
330 parsed = [
331 require_identifier(item, f"{context}[{index}]")
332 for index, item in enumerate(items)
333 ]
334 require_unique(parsed, context)
335 return parsed
338def split_identifiers(
339 value: Any, context: str, *, allow_empty: bool = False
340) -> list[str]:
341 text = require_text(value, context, allow_empty=allow_empty, maximum=MAX_CELL_CHARS)
342 if not text:
343 return []
344 parts = [part.strip() for part in text.split(";")]
345 if any(not part for part in parts):
346 raise ValidationError(f"{context} contains an empty identifier")
347 parsed = [require_identifier(part, context) for part in parts]
348 require_unique(parsed, context)
349 return parsed
352def require_iso_date(value: Any, context: str) -> str:
353 text = require_text(value, context, maximum=10)
354 if not ISO_DATE_RE.fullmatch(text):
355 raise ValidationError(f"{context} must be an ISO date (YYYY-MM-DD)")
356 try:
357 date.fromisoformat(text)
358 except ValueError as exc:
359 raise ValidationError(f"{context} must be an ISO date (YYYY-MM-DD)") from exc
360 return text
363def require_partial_date(value: Any, context: str) -> str:
364 text = require_text(value, context, maximum=10)
365 if not PARTIAL_DATE_RE.fullmatch(text):
366 raise ValidationError(f"{context} must be YYYY, YYYY-MM, or YYYY-MM-DD")
367 try:
368 if len(text) == 4:
369 date(int(text), 1, 1)
370 elif len(text) == 7:
371 date.fromisoformat(f"{text}-01")
372 else:
373 date.fromisoformat(text)
374 except ValueError as exc:
375 raise ValidationError(f"{context} contains an invalid date") from exc
376 return text
379def require_https_url(value: Any, context: str) -> str:
380 text = require_text(value, context, maximum=2_000)
381 if not HTTPS_URL_RE.fullmatch(text):
382 raise ValidationError(f"{context} must be an https URL")
383 return text
386def atomic_write_text(
387 destination: Path, text: str, *, mode: int = 0o600
388) -> Path:
389 """Atomically write private local text to a validated destination."""
390 temporary_name: str | None = None
391 try:
392 with tempfile.NamedTemporaryFile(
393 "w",
394 encoding="utf-8",
395 dir=destination.parent,
396 prefix=f".{destination.name}.",
397 suffix=".tmp",
398 delete=False,
399 ) as handle:
400 handle.write(text)
401 temporary_name = handle.name
402 os.chmod(temporary_name, mode)
403 os.replace(temporary_name, destination)
404 os.chmod(destination, mode)
405 finally:
406 if temporary_name and Path(temporary_name).exists():
407 Path(temporary_name).unlink()
408 return destination
411def write_json_report(
412 data: Any, output: str | Path | None, *, force: bool = False
413) -> None:
414 """Print JSON or atomically write it to an explicitly selected local file."""
415 serialized = json.dumps(data, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
416 if output is None:
417 print(serialized, end="")
418 return
419 destination = safe_output_path(output, ".json", force=force)
420 atomic_write_text(destination, serialized)
423def write_markdown(
424 text: str, output: str | Path, *, force: bool = False
425) -> Path:
426 destination = safe_output_path(output, ".md", force=force)
427 return atomic_write_text(destination, text)
430def issue(code: str, field: str) -> dict[str, str]:
431 """Create a content-free finding safe for reports."""
432 return {"code": code, "field": field}
435def error_exit(exc: ValidationError) -> int:
436 print(f"ERROR: {exc}", file=sys.stderr)
437 return 2