Skills › Scientific Agent Skills › iso-standards-readiness › _common.py
_common.py
567 lines · skills/iso-standards-readiness/scripts/_common.py at 5f2bc03
1#!/usr/bin/env python3
2"""Shared, bounded helpers for local standards-readiness evidence checks.
4These helpers validate structure and evidence metadata only. They never determine
5regulatory applicability, conformity, compliance, certification, accreditation, or
6audit outcome.
7"""
9from __future__ import annotations
11import argparse
12import json
13import math
14import re
15import sys
16from dataclasses import dataclass
17from datetime import date
18from pathlib import Path
19from typing import Any, Iterable
21from _catalog import DEFAULT_STANDARD, STANDARD_KEYS, STANDARDS, StandardProfile
23MAX_INPUT_BYTES = 2_000_000
24MAX_ITEMS = 5_000
25MAX_DEPTH = 30
26MAX_TEXT_CHARS = 20_000
27RESEARCH_BASIS = "2026-10-01"
28EXIT_OK = 0
29EXIT_FINDINGS = 1
30EXIT_INPUT_ERROR = 2
32ALLOWED_STATUSES = {
33 "draft",
34 "in-review",
35 "approved",
36 "implemented",
37 "verified",
38 "closed",
39 "not-applicable",
40}
41APPROVAL_STATUSES = {"pending", "approved", "rejected"}
42EVIDENCE_TYPES = {
43 "document",
44 "record",
45 "report",
46 "log",
47 "training",
48 "validation",
49 "approval",
50 "other",
51}
52PLACEHOLDER_RE = re.compile(
53 r"^(?:tbd|todo|unknown|replace[-_ ]?me|n/?a\s+without\s+rationale|"
54 r"<[^>]+>|\[[^\]]+\])$",
55 re.IGNORECASE,
56)
58DISCLAIMER = (
59 "Structural evidence check only. A zero-finding result does not establish "
60 "legal applicability, regulatory compliance, conformity to any standard, "
61 "certification, accreditation, MDSAP acceptability, EU conformity, or "
62 "inspection readiness."
63)
66class InputError(ValueError):
67 """Raised for unsafe, unreadable, or invalid input."""
70@dataclass(frozen=True)
71class Finding:
72 """One deterministic structural or evidence finding."""
74 code: str
75 path: str
76 message: str
77 severity: str = "gap"
79 def as_dict(self) -> dict[str, str]:
80 return {
81 "code": self.code,
82 "message": self.message,
83 "path": self.path,
84 "severity": self.severity,
85 }
88def _reject_constant(value: str) -> None:
89 raise InputError(f"non-finite JSON number is not permitted: {value}")
92def _unique_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
93 result: dict[str, Any] = {}
94 for key, value in pairs:
95 if key in result:
96 raise InputError(f"duplicate JSON key is not permitted: {key}")
97 result[key] = value
98 return result
101def _check_depth_and_size(value: Any) -> None:
102 stack: list[tuple[Any, int]] = [(value, 1)]
103 items = 0
104 while stack:
105 current, depth = stack.pop()
106 if depth > MAX_DEPTH:
107 raise InputError(f"JSON nesting exceeds {MAX_DEPTH} levels")
108 if isinstance(current, dict):
109 items += len(current)
110 stack.extend((item, depth + 1) for item in current.values())
111 elif isinstance(current, list):
112 items += len(current)
113 stack.extend((item, depth + 1) for item in current)
114 elif isinstance(current, str) and len(current) > MAX_TEXT_CHARS:
115 raise InputError(
116 f"JSON string exceeds {MAX_TEXT_CHARS} characters"
117 )
118 elif isinstance(current, float) and not math.isfinite(current):
119 # parse_constant rejects NaN/Infinity tokens, but a JSON number such
120 # as 1e999 also overflows to infinity with the standard parser.
121 raise InputError("non-finite JSON number is not permitted")
122 if items > MAX_ITEMS:
123 raise InputError(f"JSON contains more than {MAX_ITEMS} items")
126def _safe_input_file(path_text: str, suffixes: set[str]) -> Path:
127 path = Path(path_text)
128 if path.is_symlink():
129 raise InputError(f"symbolic-link input is refused: {path}")
130 try:
131 stat = path.stat()
132 except OSError as exc:
133 raise InputError(f"cannot stat input {path}: {exc}") from exc
134 if not path.is_file():
135 raise InputError(f"input is not a regular file: {path}")
136 if path.suffix.lower() not in suffixes:
137 expected = ", ".join(sorted(suffixes))
138 raise InputError(f"input suffix must be one of: {expected}")
139 if stat.st_size > MAX_INPUT_BYTES:
140 raise InputError(
141 f"input exceeds {MAX_INPUT_BYTES} bytes: {stat.st_size}"
142 )
143 return path
146def load_json(path_text: str) -> Any:
147 """Load bounded UTF-8 JSON, rejecting links, duplicates, and non-finite values."""
149 path = _safe_input_file(path_text, {".json"})
150 try:
151 text = path.read_text(encoding="utf-8")
152 except (OSError, UnicodeError) as exc:
153 raise InputError(f"cannot read UTF-8 JSON {path}: {exc}") from exc
154 try:
155 value = json.loads(
156 text,
157 object_pairs_hook=_unique_object,
158 parse_constant=_reject_constant,
159 )
160 except InputError:
161 raise
162 except (ValueError, RecursionError) as exc:
163 raise InputError(f"invalid JSON in {path}: {exc}") from exc
164 _check_depth_and_size(value)
165 return value
168def load_markdown(path_text: str) -> str:
169 """Load bounded local Markdown without interpreting embedded content."""
171 path = _safe_input_file(path_text, {".md", ".markdown"})
172 try:
173 text = path.read_text(encoding="utf-8")
174 except (OSError, UnicodeError) as exc:
175 raise InputError(f"cannot read UTF-8 Markdown {path}: {exc}") from exc
176 if "\x00" in text:
177 raise InputError(f"NUL byte is not permitted in Markdown: {path}")
178 return text
181def require_root_object(value: Any) -> dict[str, Any]:
182 if not isinstance(value, dict):
183 raise InputError("JSON root must be an object")
184 return value
187def is_placeholder(value: Any) -> bool:
188 return (
189 not isinstance(value, str)
190 or not value.strip()
191 or bool(PLACEHOLDER_RE.fullmatch(value.strip()))
192 )
195def valid_iso_date(value: Any) -> bool:
196 if (
197 not isinstance(value, str)
198 or re.fullmatch(r"[0-9]{4}-[0-9]{2}-[0-9]{2}", value) is None
199 ):
200 return False
201 try:
202 date.fromisoformat(value)
203 except ValueError:
204 return False
205 return True
208class Review:
209 """Accumulates fail-closed findings while preserving deterministic order."""
211 def __init__(self) -> None:
212 self.findings: list[Finding] = []
214 def add(
215 self,
216 code: str,
217 path: str,
218 message: str,
219 severity: str = "gap",
220 ) -> None:
221 self.findings.append(Finding(code, path, message, severity))
223 def object(self, value: Any, path: str) -> dict[str, Any] | None:
224 if not isinstance(value, dict):
225 self.add("TYPE_OBJECT", path, "must be an object")
226 return None
227 return value
229 def list(
230 self,
231 value: Any,
232 path: str,
233 *,
234 min_items: int = 1,
235 max_items: int = MAX_ITEMS,
236 ) -> list[Any] | None:
237 if not isinstance(value, list):
238 self.add("TYPE_LIST", path, "must be an array")
239 return None
240 if len(value) < min_items:
241 self.add("LIST_EMPTY", path, f"must contain at least {min_items} item(s)")
242 if len(value) > max_items:
243 self.add("LIST_BOUNDED", path, f"must contain at most {max_items} items")
244 return value[:max_items]
245 return value
247 def text(
248 self,
249 obj: dict[str, Any],
250 key: str,
251 path: str,
252 *,
253 max_chars: int = 500,
254 ) -> str | None:
255 value = obj.get(key)
256 field_path = f"{path}.{key}"
257 if is_placeholder(value):
258 self.add("TEXT_REQUIRED", field_path, "requires non-placeholder text")
259 return None
260 assert isinstance(value, str)
261 if len(value) > max_chars:
262 self.add(
263 "TEXT_BOUNDED",
264 field_path,
265 f"must not exceed {max_chars} characters",
266 )
267 return value.strip()
269 def choice(
270 self,
271 obj: dict[str, Any],
272 key: str,
273 allowed: set[str],
274 path: str,
275 ) -> str | None:
276 value = obj.get(key)
277 field_path = f"{path}.{key}"
278 if not isinstance(value, str) or value not in allowed:
279 self.add(
280 "CHOICE_INVALID",
281 field_path,
282 f"must be one of: {', '.join(sorted(allowed))}",
283 )
284 return None
285 return value
287 def date(
288 self,
289 obj: dict[str, Any],
290 key: str,
291 path: str,
292 *,
293 required: bool = True,
294 ) -> str | None:
295 value = obj.get(key)
296 field_path = f"{path}.{key}"
297 if value in (None, "") and not required:
298 return None
299 if not valid_iso_date(value):
300 self.add("DATE_INVALID", field_path, "must be an ISO date (YYYY-MM-DD)")
301 return None
302 return str(value)
304 def unique_ids(
305 self,
306 rows: Iterable[Any],
307 path: str,
308 *,
309 key: str = "id",
310 ) -> None:
311 seen: set[str] = set()
312 for index, row in enumerate(rows):
313 if not isinstance(row, dict):
314 continue
315 value = row.get(key)
316 if not isinstance(value, str) or not value.strip():
317 continue
318 if value in seen:
319 self.add(
320 "ID_DUPLICATE",
321 f"{path}[{index}].{key}",
322 f"duplicate identifier: {value}",
323 )
324 seen.add(value)
326 def evidence(
327 self,
328 obj: dict[str, Any],
329 path: str,
330 *,
331 min_items: int = 1,
332 ) -> list[dict[str, Any]]:
333 raw = self.list(obj.get("evidence"), f"{path}.evidence", min_items=min_items)
334 if raw is None:
335 return []
336 evidence: list[dict[str, Any]] = []
337 for index, item in enumerate(raw):
338 item_path = f"{path}.evidence[{index}]"
339 record = self.object(item, item_path)
340 if record is None:
341 continue
342 self.text(record, "id", item_path, max_chars=120)
343 self.choice(record, "type", EVIDENCE_TYPES, item_path)
344 self.text(record, "location", item_path, max_chars=1_000)
345 self.text(record, "revision_or_date", item_path, max_chars=120)
346 evidence.append(record)
347 self.unique_ids(evidence, f"{path}.evidence")
348 return evidence
350 def approval(
351 self,
352 obj: dict[str, Any],
353 path: str,
354 *,
355 require_approved: bool,
356 ) -> dict[str, Any] | None:
357 approval_path = f"{path}.approval"
358 approval = self.object(obj.get("approval"), approval_path)
359 if approval is None:
360 return None
361 status = self.choice(
362 approval,
363 "status",
364 APPROVAL_STATUSES,
365 approval_path,
366 )
367 if require_approved and status != "approved":
368 self.add(
369 "APPROVAL_PENDING",
370 f"{approval_path}.status",
371 "human approval must be recorded as approved",
372 )
373 if status == "approved":
374 self.text(approval, "by", approval_path, max_chars=200)
375 self.date(approval, "date", approval_path)
376 return approval
378 def source_refs(
379 self,
380 obj: dict[str, Any],
381 path: str,
382 *,
383 min_items: int = 1,
384 ) -> list[dict[str, Any]]:
385 raw = self.list(
386 obj.get("source_refs"),
387 f"{path}.source_refs",
388 min_items=min_items,
389 )
390 if raw is None:
391 return []
392 sources: list[dict[str, Any]] = []
393 for index, item in enumerate(raw):
394 item_path = f"{path}.source_refs[{index}]"
395 source = self.object(item, item_path)
396 if source is None:
397 continue
398 self.text(source, "id", item_path, max_chars=120)
399 self.text(source, "title", item_path, max_chars=300)
400 self.text(source, "version_or_date", item_path, max_chars=120)
401 url = self.text(source, "url", item_path, max_chars=1_000)
402 if url is not None and not url.startswith("https://"):
403 self.add(
404 "SOURCE_URL",
405 f"{item_path}.url",
406 "must use an https official-source URL",
407 )
408 self.date(source, "accessed", item_path)
409 sources.append(source)
410 self.unique_ids(sources, f"{path}.source_refs")
411 return sources
413 def controlled_item(
414 self,
415 obj: dict[str, Any],
416 path: str,
417 *,
418 require_approved: bool = True,
419 require_sources: bool = True,
420 ) -> None:
421 self.text(obj, "owner", path, max_chars=200)
422 self.choice(obj, "status", ALLOWED_STATUSES, path)
423 self.evidence(obj, path)
424 self.approval(obj, path, require_approved=require_approved)
425 if require_sources:
426 self.source_refs(obj, path)
429def standard_parser(
430 description: str,
431 input_help: str,
432 *,
433 with_standard: bool = False,
434) -> argparse.ArgumentParser:
435 parser = argparse.ArgumentParser(
436 description=description,
437 epilog=DISCLAIMER,
438 )
439 parser.add_argument("input", help=input_help)
440 if with_standard:
441 parser.add_argument(
442 "--standard",
443 choices=STANDARD_KEYS,
444 default=DEFAULT_STANDARD,
445 help=(
446 "Declared standard profile supplying the process-domain and scope "
447 "vocabulary. Selecting a profile does not decide that the standard "
448 "applies. An unlisted value is refused."
449 ),
450 )
451 parser.add_argument(
452 "--output",
453 help="Write the JSON report locally instead of stdout",
454 )
455 parser.add_argument(
456 "--force",
457 action="store_true",
458 help="Allow replacing an existing regular output file",
459 )
460 parser.add_argument(
461 "--compact",
462 action="store_true",
463 help="Emit compact JSON",
464 )
465 return parser
468def profile_for(args: argparse.Namespace) -> StandardProfile:
469 """Return the declared standard profile, defaulting to the device QMS set.
471 argparse refuses an unlisted --standard value with exit code 2, so this never
472 silently falls back to the default for an unknown standard.
473 """
475 return STANDARDS[getattr(args, "standard", DEFAULT_STANDARD)]
478def build_report(
479 tool: str,
480 findings: Iterable[Finding],
481 *,
482 metrics: dict[str, Any] | None = None,
483 basis: str = RESEARCH_BASIS,
484 standard: str | None = None,
485) -> dict[str, Any]:
486 ordered = sorted(
487 findings,
488 key=lambda item: (item.path, item.code, item.message, item.severity),
489 )
490 report = {
491 "basis_date": basis,
492 "disclaimer": DISCLAIMER,
493 "findings": [item.as_dict() for item in ordered],
494 "metrics": metrics or {},
495 "result": "gaps-found" if ordered else "complete-for-human-review",
496 "tool": tool,
497 }
498 if standard is not None:
499 report["standard"] = standard
500 return report
503def emit_report(
504 report: dict[str, Any],
505 output: str | None,
506 *,
507 force: bool,
508 compact: bool,
509) -> None:
510 text = json.dumps(
511 report,
512 ensure_ascii=False,
513 indent=None if compact else 2,
514 sort_keys=True,
515 allow_nan=False,
516 )
517 if not compact:
518 text += "\n"
519 if output is None:
520 sys.stdout.write(text)
521 return
522 path = Path(output)
523 if path.is_symlink():
524 raise InputError(f"symbolic-link output is refused: {path}")
525 if path.exists():
526 if not path.is_file():
527 raise InputError(f"output is not a regular file: {path}")
528 if not force:
529 raise InputError(
530 f"output exists; pass --force to replace it: {path}"
531 )
532 try:
533 path.write_text(text, encoding="utf-8")
534 except OSError as exc:
535 raise InputError(f"cannot write output {path}: {exc}") from exc
538def finish(
539 tool: str,
540 review: Review,
541 args: argparse.Namespace,
542 *,
543 metrics: dict[str, Any] | None = None,
544 standard: str | None = None,
545) -> int:
546 report = build_report(
547 tool,
548 review.findings,
549 metrics=metrics,
550 standard=standard,
551 )
552 emit_report(
553 report,
554 args.output,
555 force=args.force,
556 compact=args.compact,
557 )
558 return EXIT_FINDINGS if review.findings else EXIT_OK
561def guarded_main(main_function: Any) -> None:
562 try:
563 code = int(main_function())
564 except InputError as exc:
565 print(f"INPUT_ERROR: {exc}", file=sys.stderr)
566 code = EXIT_INPUT_ERROR
567 raise SystemExit(code)