Skills › Scientific Agent Skills › pydicom › uid_mapping_validator.py
uid_mapping_validator.py
242 lines · skills/pydicom/scripts/uid_mapping_validator.py at 5f2bc03
1#!/usr/bin/env python3
2"""Validate bounded DICOM UID mapping consistency without printing UIDs."""
4from __future__ import annotations
6import argparse
7import os
8import stat
9import sys
10from collections import Counter
11from collections.abc import Mapping
12from pathlib import Path
13from typing import Any
15from _common import (
16 MAX_JSON_BYTES,
17 SCHEMA_VERSION,
18 ToolError,
19 atomic_write,
20 bounded_int,
21 checked_input,
22 checked_output,
23 counter_dict,
24 derive_uid,
25 emit_json,
26 fail_json,
27 json_bytes,
28 load_json,
29 paths_overlap,
30 require_text,
31 sha256_text,
32 valid_uid,
33)
35TOOL = "uid_mapping_validator"
36STANDARD_ROOT = "1.2.840.10008"
39def _load_key(path: Path) -> bytes:
40 info = path.stat()
41 if os.name != "nt":
42 if stat.S_IMODE(info.st_mode) & 0o077:
43 raise ToolError("UID key permissions must deny group and other access")
44 if hasattr(os, "getuid") and info.st_uid != os.getuid():
45 raise ToolError("UID key must be owned by the current user")
46 payload = path.read_bytes()
47 if len(payload) == 64:
48 try:
49 payload = bytes.fromhex(payload.decode("ascii"))
50 except (UnicodeError, ValueError):
51 pass
52 if len(payload) < 32:
53 raise ToolError("UID key must contain at least 32 bytes")
54 return payload[:128]
57def validate_mapping(
58 document: Any,
59 *,
60 max_entries: int,
61 require_2_25: bool,
62 key: bytes | None = None,
63 scope: str | None = None,
64) -> dict[str, Any]:
65 if not isinstance(document, Mapping):
66 raise ToolError("mapping document must be a JSON object")
67 allowed = {"entries", "schema_version", "scope_sha256", "sensitive", "tool"}
68 unknown = sorted(set(document) - allowed)
69 if unknown:
70 raise ToolError(f"mapping document has unsupported keys: {', '.join(unknown)}")
71 entries = document.get("entries")
72 if not isinstance(entries, list):
73 raise ToolError("mapping entries must be a JSON array")
74 bounded_int(
75 len(entries),
76 name="mapping entry count",
77 minimum=0,
78 maximum=max_entries,
79 )
80 if (key is None) != (scope is None):
81 raise ToolError("deterministic verification requires both key and scope")
82 if scope is not None:
83 expected_scope_hash = sha256_text(scope)
84 stored_scope_hash = document.get("scope_sha256")
85 if stored_scope_hash and stored_scope_hash != expected_scope_hash:
86 raise ToolError("mapping scope digest does not match the supplied scope")
88 errors: Counter[str] = Counter()
89 originals: dict[str, str] = {}
90 replacements: dict[str, str] = {}
91 for entry in entries:
92 if not isinstance(entry, Mapping) or set(entry) != {
93 "original",
94 "replacement",
95 }:
96 errors["entry_schema_invalid"] += 1
97 continue
98 original = entry.get("original")
99 replacement = entry.get("replacement")
100 if not valid_uid(original):
101 errors["original_uid_invalid"] += 1
102 continue
103 if not valid_uid(replacement):
104 errors["replacement_uid_invalid"] += 1
105 continue
106 if original == replacement:
107 errors["unchanged_uid"] += 1
108 if original == STANDARD_ROOT or original.startswith(STANDARD_ROOT + "."):
109 errors["standard_uid_mapped"] += 1
110 if require_2_25 and not replacement.startswith("2.25."):
111 errors["replacement_not_2_25"] += 1
112 previous = originals.get(original)
113 if previous is not None and previous != replacement:
114 errors["one_original_to_multiple_replacements"] += 1
115 originals[original] = replacement
116 previous_original = replacements.get(replacement)
117 if previous_original is not None and previous_original != original:
118 errors["replacement_collision"] += 1
119 replacements[replacement] = original
120 if key is not None and scope is not None:
121 expected = derive_uid(original, key=key, scope=scope)
122 if replacement != expected:
123 errors["deterministic_mapping_mismatch"] += 1
124 return {
125 "deterministic_mapping_verified": key is not None,
126 "duplicate_entries": len(entries) - len(originals),
127 "entries": len(entries),
128 "error_codes": counter_dict(errors),
129 "errors": sum(errors.values()),
130 "network_accessed": False,
131 "ok": not errors,
132 "original_uids_emitted": False,
133 "replacement_uids_emitted": False,
134 "schema_version": SCHEMA_VERSION,
135 "tool": TOOL,
136 "unique_originals": len(originals),
137 "unique_replacements": len(replacements),
138 "warnings": [
139 "UID maps contain identifiers and must be stored as sensitive data.",
140 "Mapping consistency does not establish de-identification or referential completeness.",
141 "Structural UIDs such as SOP Class and Transfer Syntax UIDs must not be remapped.",
142 ],
143 }
146def build_parser() -> argparse.ArgumentParser:
147 parser = argparse.ArgumentParser(
148 description=(
149 "Validate UID syntax, one-to-one mapping, collisions, standard-UID "
150 "protection, and optional keyed deterministic derivation."
151 ),
152 formatter_class=argparse.RawDescriptionHelpFormatter,
153 epilog="""
154The mapping file is sensitive because it contains original identifiers. This
155tool emits aggregate findings only and performs no network access.
157Examples:
158 python uid_mapping_validator.py uid-map.json
159 python uid_mapping_validator.py uid-map.json --uid-key-file project.key \\
160 --uid-scope study-export-v1
161""",
162 )
163 parser.add_argument("mapping", help="Local sensitive UID mapping JSON")
164 parser.add_argument("--root", default=".", help="Existing local I/O root")
165 parser.add_argument("--uid-key-file", help="Optional local deterministic key")
166 parser.add_argument("--uid-scope", help="Scope paired with --uid-key-file")
167 parser.add_argument(
168 "--allow-non-2-25",
169 action="store_true",
170 help="Allow valid replacement UIDs outside the 2.25 UUID-derived root",
171 )
172 parser.add_argument(
173 "--max-entries",
174 type=int,
175 default=100_000,
176 help="Maximum mapping entries (default: 100000)",
177 )
178 parser.add_argument("--output", "-o", help="Private aggregate JSON report")
179 parser.add_argument(
180 "--force", action="store_true", help="Overwrite an existing report only"
181 )
182 return parser
185def main(argv: list[str] | None = None) -> int:
186 args = build_parser().parse_args(argv)
187 try:
188 max_entries = bounded_int(
189 args.max_entries,
190 name="max_entries",
191 minimum=1,
192 maximum=1_000_000,
193 )
194 mapping_path = checked_input(
195 args.mapping,
196 root=args.root,
197 kind="file",
198 max_bytes=MAX_JSON_BYTES,
199 )
200 key: bytes | None = None
201 scope: str | None = None
202 if args.uid_key_file or args.uid_scope:
203 if not args.uid_key_file or not args.uid_scope:
204 raise ToolError("--uid-key-file and --uid-scope must be used together")
205 key_path = checked_input(
206 args.uid_key_file,
207 root=args.root,
208 kind="file",
209 max_bytes=4 * 1024,
210 )
211 key = _load_key(key_path)
212 scope = require_text(args.uid_scope, name="uid_scope", maximum=128)
213 report = validate_mapping(
214 load_json(mapping_path),
215 max_entries=max_entries,
216 require_2_25=not args.allow_non_2_25,
217 key=key,
218 scope=scope,
219 )
220 if args.output:
221 destination = checked_output(args.output, root=args.root, force=args.force)
222 if paths_overlap(mapping_path, destination):
223 raise ToolError("report must not overwrite the sensitive mapping")
224 if args.uid_key_file and paths_overlap(key_path, destination):
225 raise ToolError("report must not overwrite the UID key")
226 atomic_write(destination, json_bytes(report), force=args.force)
227 emit_json(
228 {
229 "ok": report["ok"],
230 "report_written": True,
231 "tool": TOOL,
232 }
233 )
234 else:
235 emit_json(report)
236 return 0 if report["ok"] else 1
237 except Exception as exc: # noqa: BLE001 - sanitize unexpected library errors
238 return fail_json(TOOL, exc)
241if __name__ == "__main__":
242 sys.exit(main())