(repo_root: pathlib.Path, allowlist_path: pathlib.Path)
| 262 | |
| 263 | |
| 264 | def _audit(repo_root: pathlib.Path, allowlist_path: pathlib.Path) -> int: |
| 265 | allowances, errors = _read_allowances(allowlist_path) |
| 266 | by_key: dict[tuple[str, str], list[Allowance]] = {} |
| 267 | for allowance in allowances: |
| 268 | by_key.setdefault((allowance.file, allowance.function), []).append(allowance) |
| 269 | consumed: set[tuple[str, str]] = set() |
| 270 | for source_path in sorted((repo_root / "src").rglob("*.c")): |
| 271 | relative = source_path.relative_to(repo_root).as_posix() |
| 272 | if relative == "src/ui/httpd.c": |
| 273 | continue |
| 274 | masked = _mask_non_code(source_path.read_text(encoding="utf-8", errors="replace")) |
| 275 | all_calls = _scan_calls(masked) |
| 276 | for function in RAW_FUNCTIONS: |
| 277 | calls = [call for call in all_calls if call.name == function] |
| 278 | if not calls: |
| 279 | continue |
| 280 | key = (relative, function) |
| 281 | entries = by_key.get(key, []) |
| 282 | consumed.add(key) |
| 283 | if len(entries) != 1: |
| 284 | print(f"BLOCKED: {relative}: unexpected raw {function}() surface") |
| 285 | print(f" expected=none actual={len(calls)} transport=none semantic_ok=false") |
| 286 | for call in calls: |
| 287 | print(f" {call.line}: {call.text}") |
| 288 | errors.append(f"{relative}:{function}: allowance count is {len(entries)}") |
| 289 | continue |
| 290 | allowance = entries[0] |
| 291 | semantic_ok = _semantic_ok(masked, calls, allowance) |
| 292 | if len(calls) != allowance.expected or not semantic_ok: |
| 293 | print(f"BLOCKED: {relative}: unexpected raw {function}() surface") |
| 294 | print( |
| 295 | f" expected={allowance.expected} actual={len(calls)} " |
| 296 | f"transport={allowance.transport} scope={allowance.scope} " |
| 297 | f"semantic_ok={str(semantic_ok).lower()}" |
| 298 | ) |
| 299 | for call in calls: |
| 300 | print(f" {call.line}: {call.text}") |
| 301 | errors.append(f"{relative}:{function}: reviewed shape changed") |
| 302 | else: |
| 303 | print( |
| 304 | f"REVIEWED: {relative}: {len(calls)} count-bounded " |
| 305 | f"{allowance.transport} {function}() call(s) in {allowance.scope}" |
| 306 | ) |
| 307 | for key, entries in sorted(by_key.items()): |
| 308 | if key not in consumed: |
| 309 | errors.append(f"{key[0]}:{key[1]}: stale NETWORK allowance ({len(entries)} entry)") |
| 310 | for error in errors: |
| 311 | print(f" policy-error: {error}", file=sys.stderr) |
| 312 | return 1 if errors else 0 |
| 313 | |
| 314 | |
| 315 | class SelfTestFailure(RuntimeError): |
no test coverage detected