Skills › Scientific Agent Skills › peer-review › _common.py
_common.py
398 lines · skills/peer-review/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared, dependency-free safety helpers for local peer-review CLIs."""
4from __future__ import annotations
6import csv
7import json
8import os
9import re
10import tempfile
11from datetime import date
12from pathlib import Path
13from typing import Any, Iterable
15MAX_INPUT_BYTES = 4 * 1024 * 1024
16MAX_ROWS = 5_000
17MAX_CELL_CHARS = 12_000
18MAX_TEXT_CHARS = 50_000
19MAX_LIST_ITEMS = 2_000
21IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9._:-]{0,95}$")
22DOI_RE = re.compile(r"^10\.\d{4,9}/\S+$", re.IGNORECASE)
23URL_RE = re.compile(r"^https?://[^\s]+$", re.IGNORECASE)
26class ValidationError(ValueError):
27 """A deterministic, user-correctable validation failure."""
30def _duplicate_safe_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
31 result: dict[str, Any] = {}
32 for key, value in pairs:
33 if key in result:
34 raise ValidationError(f"JSON object contains duplicate key: {key}")
35 result[key] = value
36 return result
39def safe_input_path(raw_path: str | Path, suffixes: Iterable[str]) -> Path:
40 """Resolve a bounded regular local file and reject symlink inputs."""
41 path = Path(raw_path).expanduser()
42 if path.is_symlink():
43 raise ValidationError(f"symlink inputs are not allowed: {path}")
44 try:
45 resolved = path.resolve(strict=True)
46 except FileNotFoundError as exc:
47 raise ValidationError(f"input file does not exist: {path}") from exc
48 if not resolved.is_file():
49 raise ValidationError(f"input path is not a regular file: {resolved}")
50 allowed = {suffix.lower() for suffix in suffixes}
51 if resolved.suffix.lower() not in allowed:
52 choices = ", ".join(sorted(allowed))
53 raise ValidationError(f"expected one of [{choices}]: {resolved}")
54 size = resolved.stat().st_size
55 if size > MAX_INPUT_BYTES:
56 raise ValidationError(
57 f"input exceeds {MAX_INPUT_BYTES} bytes: {resolved} ({size} bytes)"
58 )
59 return resolved
62def safe_output_path(
63 raw_path: str | Path, suffix: str, *, force: bool = False
64) -> Path:
65 """Resolve an output in an existing directory without implicit overwrite."""
66 path = Path(raw_path).expanduser()
67 if path.suffix.lower() != suffix.lower():
68 raise ValidationError(f"output must use {suffix}: {path}")
69 try:
70 parent = path.parent.resolve(strict=True)
71 except FileNotFoundError as exc:
72 raise ValidationError(f"output parent does not exist: {path.parent}") from exc
73 if not parent.is_dir():
74 raise ValidationError(f"output parent is not a directory: {parent}")
75 resolved = parent / path.name
76 if resolved.is_symlink():
77 raise ValidationError(f"symlink outputs are not allowed: {resolved}")
78 if resolved.exists():
79 if not resolved.is_file():
80 raise ValidationError(f"output is not a regular file: {resolved}")
81 if not force:
82 raise ValidationError(
83 f"output already exists; pass --force to replace it: {resolved}"
84 )
85 return resolved
88def read_json(raw_path: str | Path) -> Any:
89 """Read strict UTF-8 JSON with duplicate-key detection."""
90 path = safe_input_path(raw_path, {".json"})
91 try:
92 text = path.read_text(encoding="utf-8")
93 except UnicodeDecodeError as exc:
94 raise ValidationError(f"JSON must be UTF-8: {path}") from exc
95 if "\x00" in text:
96 raise ValidationError(f"JSON contains a NUL byte: {path}")
97 try:
98 return json.loads(text, object_pairs_hook=_duplicate_safe_object)
99 except json.JSONDecodeError as exc:
100 raise ValidationError(
101 f"invalid JSON at line {exc.lineno}, column {exc.colno}: {exc.msg}"
102 ) from exc
105def read_markdown(raw_path: str | Path) -> str:
106 """Read one bounded UTF-8 Markdown file."""
107 path = safe_input_path(raw_path, {".md", ".markdown"})
108 try:
109 text = path.read_text(encoding="utf-8")
110 except UnicodeDecodeError as exc:
111 raise ValidationError(f"Markdown must be UTF-8: {path}") from exc
112 if "\x00" in text:
113 raise ValidationError(f"Markdown contains a NUL byte: {path}")
114 return text
117def read_csv_records(
118 raw_path: str | Path,
119 *,
120 required_fields: Iterable[str],
121 allowed_fields: Iterable[str] | None = None,
122 max_rows: int = MAX_ROWS,
123) -> list[dict[str, str]]:
124 """Read strict UTF-8 CSV records with bounded, unique cells and headers."""
125 path = safe_input_path(raw_path, {".csv"})
126 required = tuple(required_fields)
127 allowed = set(allowed_fields) if allowed_fields is not None else None
128 try:
129 with path.open("r", encoding="utf-8-sig", newline="") as handle:
130 reader = csv.DictReader(handle)
131 headers = reader.fieldnames
132 if not headers:
133 raise ValidationError(f"CSV has no header: {path}")
134 normalized = [header.strip() for header in headers]
135 if any(not header for header in normalized):
136 raise ValidationError("CSV headers must not be blank")
137 if len(normalized) != len(set(normalized)):
138 raise ValidationError("CSV headers must be unique")
139 missing = sorted(set(required) - set(normalized))
140 if missing:
141 raise ValidationError(
142 f"CSV is missing required columns: {', '.join(missing)}"
143 )
144 if allowed is not None:
145 unknown = sorted(set(normalized) - allowed)
146 if unknown:
147 raise ValidationError(
148 f"CSV contains unknown columns: {', '.join(unknown)}"
149 )
150 reader.fieldnames = normalized
151 records: list[dict[str, str]] = []
152 for line_number, row in enumerate(reader, start=2):
153 if line_number - 1 > max_rows:
154 raise ValidationError(f"CSV exceeds {max_rows} data rows")
155 if None in row:
156 raise ValidationError(
157 f"row {line_number} has more cells than the header"
158 )
159 cleaned: dict[str, str] = {}
160 for key, value in row.items():
161 cell = "" if value is None else value.strip()
162 if "\x00" in cell:
163 raise ValidationError(
164 f"row {line_number}, column {key} contains a NUL byte"
165 )
166 if len(cell) > MAX_CELL_CHARS:
167 raise ValidationError(
168 f"row {line_number}, column {key} exceeds "
169 f"{MAX_CELL_CHARS} characters"
170 )
171 cleaned[key] = cell
172 if any(cleaned.values()):
173 records.append(cleaned)
174 except UnicodeDecodeError as exc:
175 raise ValidationError(f"CSV must be UTF-8: {path}") from exc
176 except csv.Error as exc:
177 raise ValidationError(f"invalid CSV: {exc}") from exc
178 if not records:
179 raise ValidationError("CSV must contain at least one data row")
180 return records
183def require_object(value: Any, context: str) -> dict[str, Any]:
184 if not isinstance(value, dict):
185 raise ValidationError(f"{context} must be a JSON object")
186 return value
189def require_exact_keys(
190 value: dict[str, Any],
191 *,
192 required: Iterable[str],
193 optional: Iterable[str] = (),
194 context: str,
195) -> None:
196 required_set = set(required)
197 allowed = required_set | set(optional)
198 missing = sorted(required_set - set(value))
199 unknown = sorted(set(value) - allowed)
200 if missing:
201 raise ValidationError(f"{context} is missing fields: {', '.join(missing)}")
202 if unknown:
203 raise ValidationError(f"{context} has unknown fields: {', '.join(unknown)}")
206def require_list(
207 value: Any,
208 context: str,
209 *,
210 minimum: int = 0,
211 maximum: int = MAX_LIST_ITEMS,
212) -> list[Any]:
213 if not isinstance(value, list):
214 raise ValidationError(f"{context} must be a JSON array")
215 if not minimum <= len(value) <= maximum:
216 raise ValidationError(
217 f"{context} must contain between {minimum} and {maximum} items"
218 )
219 return value
222def require_text(
223 value: Any,
224 context: str,
225 *,
226 allow_empty: bool = False,
227 minimum: int = 1,
228 maximum: int = MAX_TEXT_CHARS,
229) -> str:
230 if not isinstance(value, str):
231 raise ValidationError(f"{context} must be a string")
232 text = value.strip()
233 if not allow_empty and not text:
234 raise ValidationError(f"{context} must not be empty")
235 if text and len(text) < minimum:
236 raise ValidationError(f"{context} must contain at least {minimum} characters")
237 if len(text) > maximum:
238 raise ValidationError(f"{context} exceeds {maximum} characters")
239 if "\x00" in text:
240 raise ValidationError(f"{context} contains a NUL byte")
241 return text
244def require_bool(value: Any, context: str) -> bool:
245 if type(value) is not bool:
246 raise ValidationError(f"{context} must be true or false")
247 return value
250def require_enum(value: Any, choices: Iterable[str], context: str) -> str:
251 text = require_text(value, context, maximum=96)
252 allowed = set(choices)
253 if text not in allowed:
254 raise ValidationError(
255 f"{context} must be one of: {', '.join(sorted(allowed))}"
256 )
257 return text
260def require_identifier(value: Any, context: str) -> str:
261 identifier = require_text(value, context, maximum=96)
262 if not IDENTIFIER_RE.fullmatch(identifier):
263 raise ValidationError(f"{context} has an invalid identifier format")
264 return identifier
267def require_unique(values: Iterable[str], context: str) -> None:
268 seen: set[str] = set()
269 duplicates: set[str] = set()
270 for value in values:
271 if value in seen:
272 duplicates.add(value)
273 seen.add(value)
274 if duplicates:
275 raise ValidationError(
276 f"{context} contains duplicate IDs: {', '.join(sorted(duplicates))}"
277 )
280def require_text_list(
281 value: Any,
282 context: str,
283 *,
284 minimum: int = 0,
285 maximum: int = 100,
286) -> list[str]:
287 items = require_list(value, context, minimum=minimum, maximum=maximum)
288 return [
289 require_text(item, f"{context}[{index}]", maximum=500)
290 for index, item in enumerate(items)
291 ]
294def require_identifier_list(
295 value: Any,
296 context: str,
297 *,
298 minimum: int = 0,
299 maximum: int = 100,
300) -> list[str]:
301 items = require_list(value, context, minimum=minimum, maximum=maximum)
302 parsed = [
303 require_identifier(item, f"{context}[{index}]")
304 for index, item in enumerate(items)
305 ]
306 require_unique(parsed, context)
307 return parsed
310def split_identifiers(
311 value: Any, context: str, *, allow_empty: bool = False
312) -> list[str]:
313 text = require_text(value, context, allow_empty=allow_empty, maximum=MAX_CELL_CHARS)
314 if not text:
315 return []
316 parts = [part.strip() for part in text.split(";")]
317 if any(not part for part in parts):
318 raise ValidationError(f"{context} contains an empty identifier")
319 parsed = [require_identifier(part, context) for part in parts]
320 require_unique(parsed, context)
321 return parsed
324def require_iso_date(value: Any, context: str) -> str:
325 text = require_text(value, context, maximum=10)
326 try:
327 date.fromisoformat(text)
328 except ValueError as exc:
329 raise ValidationError(f"{context} must be an ISO date (YYYY-MM-DD)") from exc
330 return text
333def require_url(value: Any, context: str, *, allow_empty: bool = False) -> str:
334 text = require_text(value, context, allow_empty=allow_empty, maximum=2_000)
335 if text and not URL_RE.fullmatch(text):
336 raise ValidationError(f"{context} must be an http or https URL")
337 return text
340def require_doi(value: Any, context: str, *, allow_empty: bool = False) -> str:
341 text = require_text(value, context, allow_empty=allow_empty, maximum=500)
342 if text and not DOI_RE.fullmatch(text):
343 raise ValidationError(f"{context} has an invalid DOI format")
344 return text
347def atomic_write_text(
348 destination: Path, text: str, *, mode: int = 0o600
349) -> Path:
350 """Atomically write private local text to a validated destination."""
351 temporary_name: str | None = None
352 try:
353 with tempfile.NamedTemporaryFile(
354 "w",
355 encoding="utf-8",
356 dir=destination.parent,
357 prefix=f".{destination.name}.",
358 suffix=".tmp",
359 delete=False,
360 ) as handle:
361 handle.write(text)
362 temporary_name = handle.name
363 os.chmod(temporary_name, mode)
364 os.replace(temporary_name, destination)
365 os.chmod(destination, mode)
366 finally:
367 if temporary_name and Path(temporary_name).exists():
368 Path(temporary_name).unlink()
369 return destination
372def write_json_report(
373 data: Any, output: str | Path | None, *, force: bool = False
374) -> None:
375 """Print JSON or atomically write it to an explicitly selected local file."""
376 serialized = json.dumps(data, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
377 if output is None:
378 print(serialized, end="")
379 return
380 destination = safe_output_path(output, ".json", force=force)
381 atomic_write_text(destination, serialized)
384def write_markdown(
385 text: str, output: str | Path, *, force: bool = False
386) -> Path:
387 destination = safe_output_path(output, ".md", force=force)
388 return atomic_write_text(destination, text)
391def issue(code: str, field: str) -> dict[str, str]:
392 """Create a content-free finding safe for confidential reports."""
393 return {"code": code, "field": field}
396def error_exit(exc: ValidationError) -> int:
397 print(f"ERROR: {exc}", file=os.sys.stderr)
398 return 2