Skills › Scientific Agent Skills › clinical-reports › _common.py
_common.py
267 lines · skills/clinical-reports/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Bounded local-file and validation helpers for clinical-reports scripts."""
4from __future__ import annotations
6import json
7import math
8import re
9from datetime import date, datetime
10from pathlib import Path
11from typing import Any, Iterable
13MAX_JSON_BYTES = 1_000_000
14MAX_CSV_BYTES = 5_000_000
15MAX_CSV_ROWS = 10_000
16MAX_NODES = 25_000
17MAX_DEPTH = 24
18MAX_TEXT_LENGTH = 2_000
20ALLOWED_DATA_CLASSES = {"synthetic", "deidentified", "aggregate"}
21IDENTIFIER_RE = re.compile(r"^[A-Za-z][A-Za-z0-9_.:-]{0,127}$")
22SHA256_RE = re.compile(r"^[a-f0-9]{64}$")
25class ValidationError(ValueError):
26 """Raised for fail-closed input validation errors."""
29def _reject_nonlocal(raw: str) -> None:
30 if not raw or "\x00" in raw:
31 raise ValidationError("path must be a non-empty local filesystem path")
32 if "://" in raw or raw.startswith(("file:", "\\\\")):
33 raise ValidationError("URLs, URI schemes, and network paths are not allowed")
36def local_input_path(
37 raw: str,
38 *,
39 suffixes: Iterable[str],
40 max_bytes: int,
41) -> Path:
42 """Resolve a bounded, regular, non-symlink local input file."""
43 _reject_nonlocal(raw)
44 path = Path(raw).expanduser()
45 if path.is_symlink():
46 raise ValidationError("symbolic-link inputs are not allowed")
47 try:
48 resolved = path.resolve(strict=True)
49 except FileNotFoundError as exc:
50 raise ValidationError(f"input file does not exist: {path}") from exc
51 if not resolved.is_file():
52 raise ValidationError("input must be a regular file")
53 allowed = {suffix.lower() for suffix in suffixes}
54 if resolved.suffix.lower() not in allowed:
55 raise ValidationError(f"input suffix must be one of: {sorted(allowed)}")
56 size = resolved.stat().st_size
57 if size <= 0:
58 raise ValidationError("input file is empty")
59 if size > max_bytes:
60 raise ValidationError(f"input exceeds {max_bytes} bytes")
61 return resolved
64def local_output_path(
65 raw: str,
66 *,
67 suffixes: Iterable[str],
68 overwrite: bool,
69) -> Path:
70 """Resolve a local output whose existing parent directory is trusted."""
71 _reject_nonlocal(raw)
72 path = Path(raw).expanduser()
73 allowed = {suffix.lower() for suffix in suffixes}
74 if path.suffix.lower() not in allowed:
75 raise ValidationError(f"output suffix must be one of: {sorted(allowed)}")
76 if path.exists() or path.is_symlink():
77 if path.is_symlink() or not path.is_file():
78 raise ValidationError("existing output must be a regular non-symlink file")
79 if not overwrite:
80 raise ValidationError("output exists; pass --overwrite to replace it")
81 parent = path.parent
82 if parent.is_symlink():
83 raise ValidationError("symbolic-link output directories are not allowed")
84 try:
85 resolved_parent = parent.resolve(strict=True)
86 except FileNotFoundError as exc:
87 raise ValidationError("output parent directory must already exist") from exc
88 if not resolved_parent.is_dir():
89 raise ValidationError("output parent is not a directory")
90 return resolved_parent / path.name
93def _object_without_duplicates(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
94 result: dict[str, Any] = {}
95 for key, value in pairs:
96 if key in result:
97 raise ValidationError(f"duplicate JSON key: {key}")
98 result[key] = value
99 return result
102def _check_tree(value: Any, *, depth: int = 0, counter: list[int] | None = None) -> None:
103 if counter is None:
104 counter = [0]
105 counter[0] += 1
106 if counter[0] > MAX_NODES:
107 raise ValidationError(f"JSON exceeds {MAX_NODES} nodes")
108 if depth > MAX_DEPTH:
109 raise ValidationError(f"JSON exceeds maximum depth {MAX_DEPTH}")
110 if isinstance(value, dict):
111 for key, child in value.items():
112 if not isinstance(key, str) or len(key) > 128:
113 raise ValidationError("JSON object keys must be strings of at most 128 characters")
114 _check_tree(child, depth=depth + 1, counter=counter)
115 elif isinstance(value, list):
116 for child in value:
117 _check_tree(child, depth=depth + 1, counter=counter)
118 elif isinstance(value, str):
119 if len(value) > MAX_TEXT_LENGTH:
120 raise ValidationError(
121 f"JSON strings may not exceed {MAX_TEXT_LENGTH} characters"
122 )
123 if any(ord(char) < 32 and char not in "\t\n\r" for char in value):
124 raise ValidationError("JSON strings contain disallowed control characters")
125 elif isinstance(value, float) and not math.isfinite(value):
126 raise ValidationError("non-finite numbers are not allowed")
129def load_json_object(raw_path: str) -> tuple[Path, dict[str, Any]]:
130 """Load a bounded JSON object with duplicate-key and depth checks."""
131 path = local_input_path(
132 raw_path,
133 suffixes={".json"},
134 max_bytes=MAX_JSON_BYTES,
135 )
136 try:
137 value = json.loads(
138 path.read_text(encoding="utf-8"),
139 object_pairs_hook=_object_without_duplicates,
140 parse_constant=lambda token: (_ for _ in ()).throw(
141 ValidationError(f"invalid JSON number: {token}")
142 ),
143 )
144 except UnicodeDecodeError as exc:
145 raise ValidationError("JSON input must be UTF-8") from exc
146 except RecursionError as exc:
147 raise ValidationError("JSON nesting exceeds the parser limit") from exc
148 except json.JSONDecodeError as exc:
149 raise ValidationError(f"invalid JSON: {exc.msg}") from exc
150 if not isinstance(value, dict):
151 raise ValidationError("top-level JSON value must be an object")
152 _check_tree(value)
153 return path, value
156def write_json_report(
157 report: dict[str, Any],
158 raw_output: str | None,
159 *,
160 overwrite: bool,
161) -> None:
162 """Print JSON or write it to an explicitly bounded local path."""
163 rendered = json.dumps(report, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
164 if raw_output is None:
165 print(rendered, end="")
166 return
167 output = local_output_path(
168 raw_output,
169 suffixes={".json"},
170 overwrite=overwrite,
171 )
172 output.write_text(rendered, encoding="utf-8")
175def require_data_class(value: Any) -> str:
176 """Require an allowed, explicitly declared data class."""
177 if not isinstance(value, str) or value not in ALLOWED_DATA_CLASSES:
178 raise ValidationError(
179 f"data classification must be one of: {sorted(ALLOWED_DATA_CLASSES)}"
180 )
181 return str(value)
184def require_exact_keys(
185 value: Any,
186 expected: Iterable[str],
187 field: str,
188) -> dict[str, Any]:
189 """Reject missing and unknown fields in a structured object."""
190 if not isinstance(value, dict):
191 raise ValidationError(f"{field} must be an object")
192 expected_set = set(expected)
193 missing = sorted(expected_set - set(value))
194 extra = sorted(set(value) - expected_set)
195 if missing or extra:
196 details = []
197 if missing:
198 details.append(f"missing={missing}")
199 if extra:
200 details.append(f"unknown={extra}")
201 raise ValidationError(f"{field} fields are invalid ({'; '.join(details)})")
202 return value
205def require_identifier(value: Any, field: str) -> str:
206 """Require a bounded machine identifier."""
207 if not isinstance(value, str) or not IDENTIFIER_RE.fullmatch(value):
208 raise ValidationError(f"{field} must match {IDENTIFIER_RE.pattern}")
209 return value
212def require_string(value: Any, field: str, *, max_length: int = 256) -> str:
213 """Require a non-empty bounded string."""
214 if not isinstance(value, str):
215 raise ValidationError(f"{field} must be a string")
216 normalized = value.strip()
217 if not normalized or len(normalized) > max_length:
218 raise ValidationError(f"{field} must contain 1-{max_length} characters")
219 return normalized
222def require_bool(value: Any, field: str) -> bool:
223 """Require an actual JSON boolean."""
224 if not isinstance(value, bool):
225 raise ValidationError(f"{field} must be a boolean")
226 return value
229def require_nonnegative_int(value: Any, field: str) -> int:
230 """Require an integer count without accepting booleans."""
231 if isinstance(value, bool) or not isinstance(value, int) or value < 0:
232 raise ValidationError(f"{field} must be a non-negative integer")
233 return value
236def parse_iso_date(value: Any, field: str) -> date:
237 """Parse an ISO calendar date without inferring missing precision."""
238 text = require_string(value, field, max_length=10)
239 if not re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", text):
240 raise ValidationError(f"{field} must be YYYY-MM-DD")
241 try:
242 return date.fromisoformat(text)
243 except ValueError as exc:
244 raise ValidationError(f"{field} must be YYYY-MM-DD") from exc
247def parse_iso_datetime(value: Any, field: str) -> datetime:
248 """Parse an ISO datetime and require an explicit timezone."""
249 text = require_string(value, field, max_length=40)
250 try:
251 parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
252 except ValueError as exc:
253 raise ValidationError(f"{field} must be an ISO 8601 datetime") from exc
254 if parsed.tzinfo is None:
255 raise ValidationError(f"{field} must include a timezone")
256 return parsed
259def error_report(tool: str, exc: Exception) -> dict[str, Any]:
260 """Return a non-sensitive machine-readable error."""
261 return {
262 "tool": tool,
263 "status": "BLOCKED_INVALID_INPUT",
264 "errors": [str(exc)],
265 "review_required": True,
266 "authorizes_clinical_use_or_submission": False,
267 }