3"""AST inventory of checker scope authorities and their concrete values."""
5from __future__
import annotations
13from dataclasses
import dataclass
14from pathlib
import Path, PurePosixPath
16from suppression_catalog
import ownership
17from suppression_checker_census
import (
19 classification_digest,
21 repository_mutation_findings,
25from suppression_model
import Finding, Suppression
26from suppression_nonauth_registry
import NON_AUTHORITIES
27from suppression_scope_registry
import (
29 NON_AUTHORITY_CATEGORIES,
33CANDIDATE_PREFIXES = (
"scripts/checks/",
"scripts/ci/")
34EXPECTED_AUTHORITIES = 1014
36MIN_CENSUS_CONSTANTS = 1700
39EXPECTED_AUTHORITY_VALUE_SHA256 =
"685059fc7c4ac4f8cbb832f6fe8fa6444ef5c4a2db4c01aefd6a445997e052f8"
40EXPECTED_AUTHORITY_REASON_SHA256 = (
41 "a0ea0bb802f42b69372805001a8b9144b58937d5569a84646f0c4ae9d80aab6c"
43EXPECTED_CLASSIFICATION_SHA256 =
"e306ca46f50ce3146310deda9caa7705d10f0d51867eba7efec11edffc48ccf3"
44_EMPTY_AUTHORITIES = frozenset(
46 "scripts/checks/stack_usage_check.py:FIRST_PARTY_EXEMPTIONS",
47 "scripts/checks/suppression_governance.py:GLOBAL_EXCLUSION_AUTHORITIES",
51 "positive-scope":
"Explicit positive boundary determines which repository inputs are checked.",
53 "Explicit path exclusion removes reviewed non-first-party or out-of-domain input."
56 "Self-reference exemption prevents the policy checker from flagging its own grammar."
59 "Named token is an explicitly reviewed semantic exception to the checker rule."
61 "vendor-exemption":
"Vendored or SOUP input is governed by its upstream validation boundary.",
62 "generated-classification": (
63 "Generated input is governed by a distinct reproducible generator boundary."
65 "host-exemption":
"Hosted-only input is outside the embedded target semantic boundary.",
66 "ignored-literal":
"Named semantic literal is an explicit checker exception.",
67 "regex-exclusion":
"Compiled expression defines an exact reviewed exclusion boundary.",
68 "known-gap":
"Named gap remains explicit and measurable until its checker support lands.",
69 "stack-exemption":
"Named frame exception binds an exact function, ceiling, and rationale.",
70 "suppression-control-plane": (
71 "Scanner control-plane value defines ownership, syntax, or census scope."
73 "anti-vacuity-floor": (
74 "Numeric floor or audited count keeps a collapsed scan from reporting clean."
76 "waiver-recognizer": (
77 "Pattern recognizing an explicit in-source waiver marker; it defines how policy is waived."
82@dataclass(frozen=True)
84 """One safely resolved concrete value and an optional source rationale."""
90@dataclass(frozen=True)
91class ResolvedAuthority:
92 """One final module assignment bound to its explicit registry schema."""
97 schema: AuthoritySchema
98 values: tuple[PolicyValue, ...]
101def _call_name(node: ast.Call) -> str:
102 """Return a simple/attribute call name without executing it."""
104 if isinstance(func, ast.Name):
106 if isinstance(func, ast.Attribute):
111def _dedupe(values: list[PolicyValue]) -> list[PolicyValue]:
112 """Preserve declaration order while rejecting duplicate concrete leaves."""
113 result: list[PolicyValue] = []
114 seen: set[str] = set()
116 if item.value
in seen:
123def _pair_value(node: ast.AST, env: dict[str, list[PolicyValue]], path: str) -> PolicyValue |
None:
124 """Resolve one explicitly schema-bound ``(value, reason)`` policy tuple."""
125 if not isinstance(node, (ast.Tuple, ast.List))
or len(node.elts) != PAIR_SIZE:
127 second = node.elts[1]
128 if not isinstance(second, ast.Constant)
or not isinstance(second.value, str):
130 reason = second.value.strip()
131 if not any(character.isalnum()
for character
in reason):
133 first = _resolve(node.elts[0], env, path, reasoned_pairs=
False)
136 return PolicyValue(first[0].value, reason)
139def _path_join(left: str, right: str) -> str:
140 """Join repo-relative path AST operands without host-path semantics."""
143 return str(PurePosixPath(left) / right)
146def _is_own_file(node: ast.AST) -> bool:
147 """Recognize a ``Path(__file__)`` expression, optionally ``.resolve()``d."""
148 if isinstance(node, ast.Call)
and _call_name(node) ==
"resolve":
149 return isinstance(node.func, ast.Attribute)
and _is_own_file(node.func.value)
150 if isinstance(node, ast.Call)
and _call_name(node) ==
"Path" and node.args:
152 return isinstance(first, ast.Name)
and first.id ==
"__file__"
156def _is_repo_root_chain(node: ast.AST) -> bool:
157 """Recognize ``repo_root()`` calls and parent-chains over ``__file__``."""
158 if isinstance(node, ast.Call)
and _call_name(node) ==
"repo_root":
160 if isinstance(node, ast.Attribute)
and node.attr
in {
"parent",
"parents"}:
161 return _is_own_file(node.value)
or _is_repo_root_chain(node.value)
162 if isinstance(node, ast.Subscript):
163 return _is_repo_root_chain(node.value)
167def _resolve_comprehension(
168 node: ast.GeneratorExp | ast.ListComp | ast.SetComp,
169 env: dict[str, list[PolicyValue]],
171) -> list[PolicyValue]:
172 """Statically evaluate a single-variable, condition-free comprehension."""
173 if len(node.generators) != 1:
175 generator = node.generators[0]
176 if generator.ifs
or generator.is_async
or not isinstance(generator.target, ast.Name):
178 items = _resolve(generator.iter, env, path)
181 loop_name = generator.target.id
182 result: list[PolicyValue] = []
185 scoped[loop_name] = [item]
186 element = _resolve(node.elt, scoped, path)
187 if len(element) != 1:
189 result.append(element[0])
190 return _dedupe(result)
194 node: ast.JoinedStr, env: dict[str, list[PolicyValue]], path: str
195) -> list[PolicyValue]:
196 """Concatenate an f-string whose every part resolves to one value."""
197 parts: list[str] = []
198 for value
in node.values:
199 if isinstance(value, ast.Constant)
and isinstance(value.value, str):
200 parts.append(value.value)
201 elif isinstance(value, ast.FormattedValue):
202 inner = _resolve(value.value, env, path)
205 parts.append(inner[0].value)
208 return [PolicyValue(
"".join(parts))]
213 env: dict[str, list[PolicyValue]],
216 reasoned_pairs: bool,
217) -> list[PolicyValue]:
218 """Resolve the closed call vocabulary used by scope authorities."""
219 name = _call_name(node)
220 result: list[PolicyValue] = []
221 if _is_repo_root_chain(node):
222 result = [PolicyValue(
"")]
223 elif _is_own_file(node):
224 result = [PolicyValue(path)]
225 elif name
in {
"frozenset",
"set",
"tuple",
"list"}:
227 _resolve(node.args[0], env, path, reasoned_pairs=reasoned_pairs)
if node.args
else []
229 elif name
in {
"compile",
"Path",
"PurePosixPath",
"escape"}
and node.args:
232 isinstance(first, ast.Call)
233 and _call_name(first) ==
"get"
234 and len(first.args) == PAIR_SIZE
236 result = _resolve(first.args[1], env, path)
238 result = _resolve(first, env, path)
240 result = _resolve_call_tail(name, node, env, path)
244def _resolve_call_tail(
247 env: dict[str, list[PolicyValue]],
249) -> list[PolicyValue]:
250 """Resolve the join, fromkeys, own-path, and Gap call forms."""
251 result: list[PolicyValue] = []
252 if name ==
"resolve" and isinstance(node.func, ast.Attribute):
253 result = _resolve(node.func.value, env, path)
254 elif name ==
"fromkeys" and len(node.args) == PAIR_SIZE:
255 keys = _resolve(node.args[0], env, path)
256 label = _resolve(node.args[1], env, path)
257 if keys
and len(label) == 1:
258 result = [PolicyValue(f
"{key.value}:{label[0].value}")
for key
in keys]
259 elif name ==
"join" and isinstance(node.func, ast.Attribute)
and node.args:
260 separator = _resolve(node.func.value, env, path)
261 joined = _resolve(node.args[0], env, path)
262 if len(separator) == 1
and joined:
263 result = [PolicyValue(separator[0].value.join(item.value
for item
in joined))]
264 elif name ==
"Gap" and len(node.args) >= GAP_MIN_ARGS:
265 resolved = _resolve(node.args[0], env, path)
266 if len(resolved) == 1:
269 for argument
in node.args[1:GAP_MIN_ARGS]
270 for item
in _resolve(argument, env, path)
272 result = [PolicyValue(resolved[0].value,
" ".join(reason_bits))]
276def _resolve_collection(
277 node: ast.Tuple | ast.List | ast.Set,
278 env: dict[str, list[PolicyValue]],
281 reasoned_pairs: bool,
282) -> list[PolicyValue]:
283 """Resolve a literal container under one explicit authority schema."""
285 pairs = [_pair_value(child, env, path)
for child
in node.elts]
286 return _dedupe([item
for item
in pairs
if item
is not None])
if all(pairs)
else []
288 [item
for child
in node.elts
for item
in _resolve(child, env, path, reasoned_pairs=
False)]
293 node: ast.Dict, env: dict[str, list[PolicyValue]], path: str
294) -> list[PolicyValue]:
295 """Flatten a mapping authority as key:value concrete leaves."""
296 values: list[PolicyValue] = []
297 for key_node, value_node
in zip(node.keys, node.values, strict=
True):
299 values.extend(_resolve(value_node, env, path))
301 keys = _resolve(key_node, env, path)
302 children = _resolve(value_node, env, path)
305 PolicyValue(f
"{key.value}:{child.value}", child.reason)
for child
in children
307 return _dedupe(values)
311 node: ast.BinOp, env: dict[str, list[PolicyValue]], path: str
312) -> list[PolicyValue]:
313 """Resolve union, string concatenation, and repo-path joins."""
314 left = _resolve(node.left, env, path)
315 right = _resolve(node.right, env, path)
316 if isinstance(node.op, ast.BitOr):
317 return _dedupe(left + right)
318 if len(left) != 1
or len(right) != 1:
320 if isinstance(node.op, ast.Add):
321 return [PolicyValue(left[0].value + right[0].value)]
322 if isinstance(node.op, ast.Div):
323 return [PolicyValue(_path_join(left[0].value, right[0].value))]
327def _resolve_atom(node: ast.AST, env: dict[str, list[PolicyValue]]) -> list[PolicyValue]:
328 """Resolve constants, names, and repository-root chains."""
329 if isinstance(node, ast.Constant):
330 if isinstance(node.value, (str, int, float, bool)):
331 return [PolicyValue(str(node.value))]
333 if isinstance(node, (ast.Subscript, ast.Attribute)):
334 return [PolicyValue(
"")]
if _is_repo_root_chain(node)
else []
335 if isinstance(node, ast.Name):
336 return [PolicyValue(
"")]
if node.id ==
"REPO_ROOT" else list(env.get(node.id, []))
342 env: dict[str, list[PolicyValue]],
345 reasoned_pairs: bool =
False,
346) -> list[PolicyValue]:
347 """Safely resolve the literal/computed expression subset used by checkers."""
348 result: list[PolicyValue] = []
349 if isinstance(node, (ast.Constant, ast.Subscript, ast.Attribute, ast.Name)):
350 result = _resolve_atom(node, env)
351 elif isinstance(node, (ast.Tuple, ast.List, ast.Set)):
352 result = _resolve_collection(node, env, path, reasoned_pairs=reasoned_pairs)
353 elif isinstance(node, ast.Dict):
354 result = _resolve_dict(node, env, path)
355 elif isinstance(node, ast.Call):
356 result = _dedupe(_resolve_call(node, env, path, reasoned_pairs=reasoned_pairs))
357 elif isinstance(node, ast.BinOp):
358 result = _resolve_binary(node, env, path)
359 elif isinstance(node, (ast.Starred, ast.GeneratorExp, ast.ListComp, ast.SetComp)):
360 result = _resolve_spread(node, env, path, reasoned_pairs=reasoned_pairs)
361 elif isinstance(node, ast.JoinedStr):
362 result = _resolve_joined(node, env, path)
363 elif isinstance(node, ast.IfExp):
365 _resolve(node.body, env, path, reasoned_pairs=reasoned_pairs)
366 + _resolve(node.orelse, env, path, reasoned_pairs=reasoned_pairs)
373 env: dict[str, list[PolicyValue]],
376 reasoned_pairs: bool,
377) -> list[PolicyValue]:
378 """Resolve starred spreads and statically-evaluable comprehensions."""
379 if isinstance(node, ast.Starred):
380 return _resolve(node.value, env, path, reasoned_pairs=reasoned_pairs)
381 return _resolve_comprehension(node, env, path)
384def _string_leaves(node: ast.AST) -> list[PolicyValue]:
385 """Bind every string constant inside a structured table, sorted exactly."""
389 for child
in ast.walk(node)
390 if isinstance(child, ast.Constant)
and isinstance(child.value, str)
393 return [PolicyValue(value)
for value
in leaves]
396def _callable_table(node: ast.AST) -> list[PolicyValue]:
397 """Bind a dispatch table by the exact names of the callables it holds."""
399 {child.id
for child
in ast.walk(node)
if isinstance(child, ast.Name)}
400 | {child.attr
for child
in ast.walk(node)
if isinstance(child, ast.Attribute)}
402 return [PolicyValue(value)
for value
in names]
405def _is_empty_literal(node: ast.AST) -> bool:
406 """Return whether a value is an explicitly empty literal container."""
407 if isinstance(node, (ast.Tuple, ast.List, ast.Set))
and not node.elts:
409 if isinstance(node, ast.Dict)
and not node.keys:
411 if isinstance(node, ast.Call)
and _call_name(node)
in {
"frozenset",
"set",
"tuple",
"dict"}:
412 return not node.args
or _is_empty_literal(node.args[0])
416def _assignment(node: ast.AST) -> tuple[str, ast.AST] |
None:
417 """Return a single-name module assignment."""
419 isinstance(node, ast.Assign)
420 and len(node.targets) == 1
421 and isinstance(node.targets[0], ast.Name)
423 return node.targets[0].id, node.value
425 isinstance(node, ast.AnnAssign)
426 and isinstance(node.target, ast.Name)
427 and node.value
is not None
429 return node.target.id, node.value
433def _authority_records(
437 resolved: list[PolicyValue],
438 schema: AuthoritySchema,
439) -> tuple[list[Suppression], list[Finding]]:
440 """Build all concrete rows for one classified authority."""
441 identity = f
"{rel}:{name}"
442 if not resolved
and identity
not in _EMPTY_AUTHORITIES:
444 "unresolved-checker-scope-authority",
445 f
"{name} uses an unsupported expression",
450 shared_reason = _SUBTYPE_REASONS[schema.subtype]
456 "checker-scope-control",
457 "repository-checker",
460 f
"value:{value.value}",
461 value.reason
or shared_reason,
462 "module-ast-authority",
465 evidence=(f
"authority:{rel}:{name}:{value.value}",),
467 for value
in resolved
472def _stack_exemptions(
473 node: ast.AST, env: dict[str, list[PolicyValue]], path: str
474) -> list[PolicyValue]:
475 """Resolve exact ``(translation-unit, function, ceiling, reason)`` rows."""
476 if not isinstance(node, (ast.Tuple, ast.List)):
478 values: list[PolicyValue] = []
479 for child
in node.elts:
480 if not isinstance(child, (ast.Tuple, ast.List))
or len(child.elts) != GAP_MIN_ARGS:
482 fields = [_resolve(field, env, path)
for field
in child.elts[:3]]
483 reason = _resolve(child.elts[3], env, path)
484 if any(len(field) != 1
for field
in fields)
or len(reason) != 1:
486 if not any(character.isalnum()
for character
in reason[0].value):
488 identity =
":".join(field[0].value
for field
in fields)
489 values.append(PolicyValue(identity, reason[0].value))
493def _resolve_registered(
495 env: dict[str, list[PolicyValue]],
497 schema: AuthoritySchema,
498) -> list[PolicyValue]:
499 """Resolve one closed AST shape selected by its explicit registry entry."""
500 if schema.mode ==
"expression-digest":
501 expression = ast.dump(node, annotate_fields=
True, include_attributes=
False)
502 digest = hashlib.sha256(expression.encode(
"utf-8")).hexdigest()
503 resolved = [PolicyValue(f
"sha256:{digest}")]
504 elif schema.mode ==
"reasoned-pairs":
505 resolved = _resolve(node, env, path, reasoned_pairs=
True)
506 elif schema.mode ==
"stack-exemptions":
507 resolved = _stack_exemptions(node, env, path)
508 elif schema.mode ==
"string-leaves":
509 resolved = _string_leaves(node)
510 elif schema.mode ==
"callable-table":
511 resolved = _callable_table(node)
512 elif schema.mode.startswith(
"derived-"):
515 resolved = _resolve(node, env, path)
522 census: list[tuple[str, int, ast.AST |
None]],
523 module_authorities: dict[str, frozenset[str]],
525 """Enforce classification, shape, immutability, and sink rules for one file."""
526 findings: list[Finding] = []
528 isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and "selftest" in node.name
529 for node
in ast.walk(tree)
531 for name, line, value_node
in census:
532 identity = f
"{rel}:{name}"
533 if identity
in AUTHORITY_SCHEMAS:
535 category = NON_AUTHORITIES.get(identity)
539 "unclassified-checker-constant",
540 f
"{name} is neither a registered authority nor a classified non-authority",
545 elif category
not in NON_AUTHORITY_CATEGORIES:
548 "non-authority-shape-mismatch",
549 f
"{name}: unknown non-authority category {category}",
554 elif value_node
is not None:
555 problem = shape_problem(rel, category, value_node, has_selftest=has_selftest)
556 if problem
is not None:
558 Finding(
"non-authority-shape-mismatch", f
"{name}: {problem}", rel, line)
560 local_authorities = {
561 name
for name, _line, _value
in census
if f
"{rel}:{name}" in AUTHORITY_SCHEMAS
563 findings.extend(mutation_findings(rel, tree, local_authorities, module_authorities))
564 findings.extend(sink_findings(rel, tree))
568def _module_authorities() -> dict[str, frozenset[str]]:
569 """Map checker module basenames to their registered authority names."""
570 modules: dict[str, set[str]] = {}
571 for identity
in AUTHORITY_SCHEMAS:
572 path, _colon, name = identity.rpartition(
":")
573 modules.setdefault(Path(path).stem, set()).add(name)
574 return {module: frozenset(names)
for module, names
in modules.items()}
577def _condition_dependent(value_node: ast.AST) -> bool:
578 """Return whether an authority value depends on a runtime condition."""
579 return any(isinstance(node, ast.IfExp)
for node
in ast.walk(value_node))
583 root: Path, rel: str, module_authorities: dict[str, frozenset[str]]
584) -> tuple[dict[str, ResolvedAuthority], list[Finding], set[str], set[str]]:
585 """Parse one checker, resolve authorities, and enforce the constant census."""
587 tree = ast.parse((root / rel).read_text(encoding=
"utf-8"), filename=rel)
588 except (OSError, SyntaxError)
as exc:
589 return {}, [Finding(
"checker-scope-ast", str(exc), rel)], set(), set()
590 findings: list[Finding] = []
591 authorities: dict[str, ResolvedAuthority] = {}
592 diagnosed: set[str] = set()
593 env: dict[str, list[PolicyValue]] = {}
594 census = census_bindings(tree)
595 findings.extend(_census_findings(rel, tree, census, module_authorities))
596 for node
in tree.body:
597 item = _assignment(node)
600 name, value_node = item
601 identity = f
"{rel}:{name}"
602 schema = AUTHORITY_SCHEMAS.get(identity)
605 and schema.mode !=
"expression-digest"
606 and _condition_dependent(value_node)
610 "condition-dependent-authority",
611 f
"{name} is built from a runtime condition and cannot be authenticated",
616 diagnosed.add(identity)
617 authorities[identity] = ResolvedAuthority(rel, node.lineno, name, schema, ())
620 _resolve_registered(value_node, env, rel, schema)
621 if schema
is not None
622 else _resolve(value_node, env, rel)
624 if resolved
or isinstance(value_node, (ast.Tuple, ast.List, ast.Set, ast.Dict)):
626 if schema
is not None:
627 if not resolved
and _is_empty_literal(value_node):
628 diagnosed.add(identity)
629 authorities[identity] = ResolvedAuthority(
630 rel, node.lineno, name, schema, tuple(resolved)
632 census_identities = {f
"{rel}:{name}" for name, _line, _value
in census}
633 return authorities, findings, diagnosed, census_identities
637 values: tuple[PolicyValue, ...], *, reason_terms: tuple[str, ...]
638) -> tuple[PolicyValue, ...]:
639 """Select reasoned prefixes through one explicit imported-authority contract."""
641 PolicyValue(item.value)
643 if any(term
in item.reason.lower()
for term
in reason_terms)
647def _classified_values(
648 values: tuple[PolicyValue, ...], classification: str
649) -> tuple[PolicyValue, ...]:
650 """Select mapping keys carrying one exact semantic classification."""
651 suffix = f
":{classification}"
653 PolicyValue(item.value.removesuffix(suffix))
655 if item.value.endswith(suffix)
659def _schema_digest() -> str:
660 """Authenticate every registered identity with its subtype and value mode."""
662 identity: f
"{schema.subtype}/{schema.mode}"
663 for identity, schema
in sorted(AUTHORITY_SCHEMAS.items())
665 encoded = json.dumps(payload, sort_keys=
True, separators=(
",",
":")).encode()
666 return hashlib.sha256(encoded).hexdigest()
669def _live_registry_values() -> dict[str, tuple[PolicyValue, ...]]:
670 """Bind the classification registries' exact live content as authority values."""
672 PolicyValue(f
"entries:{len(AUTHORITY_SCHEMAS)}"),
673 PolicyValue(f
"sha256:{_schema_digest()}"),
676 PolicyValue(f
"entries:{len(NON_AUTHORITIES)}"),
677 PolicyValue(f
"sha256:{classification_digest(AUTHORITY_SCHEMAS, NON_AUTHORITIES)}"),
680 "scripts/checks/suppression_scope_registry.py:AUTHORITY_SCHEMAS": schema_binding,
681 "scripts/checks/suppression_scope_registry.py:_SCHEMA_GROUPS": schema_binding,
682 "scripts/checks/suppression_nonauth_registry.py:_GROUPS": nonauth_binding,
683 "scripts/checks/suppression_scope_registry.py:NON_AUTHORITY_CATEGORIES": tuple(
684 PolicyValue(category, reason)
685 for category, reason
in sorted(NON_AUTHORITY_CATEGORIES.items())
687 "scripts/checks/suppression_nonauth_registry.py:NON_AUTHORITIES": nonauth_binding,
691def _apply_derived(authorities: dict[str, ResolvedAuthority]) ->
None:
692 """Resolve registered cross-module comprehensions from their authenticated source."""
693 exemptions = authorities[
"scripts/checks/lint_coverage_rules.py:EXEMPT_PREFIXES"].values
694 path_classes = authorities[
"scripts/checks/lint_coverage_rules.py:PATH_CLASS"].values
695 extension_classes = authorities[
"scripts/checks/lint_coverage_rules.py:EXT_CLASS"].values
697 **_live_registry_values(),
698 "scripts/checks/suppression_catalog.py:VENDOR_PREFIXES": _filtered_values(
699 exemptions, reason_terms=(
"vendored",
"soup")
701 "scripts/checks/suppression_catalog.py:GENERATED_PREFIXES": _filtered_values(
702 exemptions, reason_terms=(
"generated",
"emitted")
704 "scripts/checks/suppression_catalog.py:BINARY_SUFFIXES": _classified_values(
705 extension_classes,
"binary"
707 "scripts/checks/check_no_null.py:GENERATED_SOURCE_PATHS": _classified_values(
708 path_classes,
"generated-source"
710 "scripts/checks/check_no_stdio_streams.py:GENERATED_SOURCE_PATHS": _classified_values(
711 path_classes,
"generated-source"
714 for identity, values
in replacements.items():
715 authority = authorities[identity]
716 authorities[identity] = ResolvedAuthority(
725SELF_PIN_IDENTITIES = frozenset(
727 "scripts/checks/suppression_checker_scope.py:EXPECTED_VALUES",
728 "scripts/checks/suppression_checker_scope.py:EXPECTED_AUTHORITY_VALUE_SHA256",
729 "scripts/checks/suppression_checker_scope.py:EXPECTED_AUTHORITY_REASON_SHA256",
730 "scripts/checks/suppression_checker_scope.py:EXPECTED_CLASSIFICATION_SHA256",
735def _authority_value_digest(authorities: dict[str, ResolvedAuthority]) -> str:
736 """Authenticate each authority name and its exact concrete values.
738 The digest pins themselves are excluded from the domain: a digest that
739 hashed its own committed value could never reach a fixed point.
742 key: sorted(item.value
for item
in authority.values)
743 for key, authority
in sorted(authorities.items())
744 if key
not in SELF_PIN_IDENTITIES
746 encoded = json.dumps(payload, sort_keys=
True, separators=(
",",
":")).encode()
747 return hashlib.sha256(encoded).hexdigest()
750def _authority_reason_digest(records: list[Suppression]) -> str:
751 """Authenticate rationales separately from authority/value parsing."""
752 payload: dict[str, list[tuple[str, str]]] = {}
754 if f
"{item.path}:{item.directive}" in SELF_PIN_IDENTITIES:
756 key = f
"{item.path}:{item.directive}"
757 payload.setdefault(key, []).append((item.scope.removeprefix(
"value:"), item.reason))
758 encoded = json.dumps(
759 {key: sorted(values)
for key, values
in sorted(payload.items())},
761 separators=(
",",
":"),
763 return hashlib.sha256(encoded).hexdigest()
766def _collect_authorities(
767 root: Path, paths: list[str]
768) -> tuple[dict[str, ResolvedAuthority], list[Finding], set[str]]:
769 """Collect every registered assignment and enforce the exhaustive census."""
770 findings: list[Finding] = []
771 authorities: dict[str, ResolvedAuthority] = {}
772 diagnosed: set[str] = set()
773 module_authorities = _module_authorities()
775 rel
for rel
in paths
if rel.startswith(CANDIDATE_PREFIXES)
and rel.endswith(
".py")
778 seen_identities: set[str] = set()
779 for rel
in candidates:
780 file_authorities, problems, file_diagnosed, census_identities = _scan_scope_file(
781 root, rel, module_authorities
783 findings.extend(problems)
784 authorities.update(file_authorities)
785 diagnosed.update(file_diagnosed)
786 census_total += len(census_identities)
787 seen_identities.update(census_identities)
788 if census_total < MIN_CENSUS_CONSTANTS:
791 "checker-census-floor",
792 f
"only {census_total} module constants seen; floor is {MIN_CENSUS_CONSTANTS}",
795 stale = sorted(identity
for identity
in NON_AUTHORITIES
if identity
not in seen_identities)
797 Finding(
"stale-checker-classification", f
"{identity} no longer exists")
798 for identity
in stale
800 missing = sorted(set(AUTHORITY_SCHEMAS) - set(authorities))
801 findings.extend(Finding(
"missing-checker-scope-authority", identity)
for identity
in missing)
803 _apply_derived(authorities)
805 repository_mutation_findings(root, paths, module_authorities, frozenset(candidates))
807 return authorities, findings, diagnosed
810def _records_for_authorities(
811 authorities: dict[str, ResolvedAuthority],
813) -> tuple[list[Suppression], list[Finding]]:
814 """Render source-located inventory rows after cross-module derivation."""
815 records: list[Suppression] = []
816 findings: list[Finding] = []
817 for identity, authority
in sorted(authorities.items()):
818 rows, problems = _authority_records(
822 list(authority.values),
826 if identity
not in diagnosed:
827 findings.extend(problems)
828 return records, findings
831def scan_checker_scope_controls(
832 root: Path, paths: list[str]
833) -> tuple[list[Suppression], list[Finding]]:
834 """Inventory every reviewed checker authority and fail unknown/unresolved shapes."""
835 authorities, findings, diagnosed = _collect_authorities(root, paths)
836 records, record_findings = _records_for_authorities(authorities, diagnosed)
837 findings.extend(record_findings)
838 live_classification = classification_digest(AUTHORITY_SCHEMAS, NON_AUTHORITIES)
839 if live_classification != EXPECTED_CLASSIFICATION_SHA256:
842 "checker-classification-digest",
843 f
"found {live_classification}; audited contract is "
844 f
"{EXPECTED_CLASSIFICATION_SHA256}",
847 if len(authorities) != EXPECTED_AUTHORITIES:
850 "checker-scope-authority-count",
851 f
"found {len(authorities)}; audited contract is {EXPECTED_AUTHORITIES}",
854 if EXPECTED_VALUES
and len(records) != EXPECTED_VALUES:
857 "checker-scope-value-count",
858 f
"found {len(records)}; audited contract is {EXPECTED_VALUES}",
861 value_digest = _authority_value_digest(authorities)
862 if value_digest != EXPECTED_AUTHORITY_VALUE_SHA256:
865 "checker-scope-value-digest",
866 f
"found {value_digest}; audited contract is {EXPECTED_AUTHORITY_VALUE_SHA256}",
869 reason_digest = _authority_reason_digest(records)
870 if reason_digest != EXPECTED_AUTHORITY_REASON_SHA256:
873 "checker-scope-reason-digest",
874 f
"found {reason_digest}; audited contract is {EXPECTED_AUTHORITY_REASON_SHA256}",
877 return records, findings
886 """Rewrite the audited EXPECTED_* constants in this module's source."""
887 text = path.read_text(encoding=
"utf-8")
889 r"EXPECTED_VALUES = \d+",
890 f
"EXPECTED_VALUES = {values}",
895 r'EXPECTED_AUTHORITY_VALUE_SHA256 = "[0-9a-f]{64}"',
896 f
'EXPECTED_AUTHORITY_VALUE_SHA256 = "{value_digest}"',
901 r'EXPECTED_AUTHORITY_REASON_SHA256 = \(\n "[0-9a-f]{64}"\n\)',
902 f
'EXPECTED_AUTHORITY_REASON_SHA256 = (\n "{reason_digest}"\n)',
906 path.write_text(text, encoding=
"utf-8")
909def _update(root: Path, paths: list[str]) -> int:
910 """Re-freeze the audited scope constants from the live census (idempotent)."""
911 authorities, findings, diagnosed = _collect_authorities(root, paths)
912 records, record_findings = _records_for_authorities(authorities, diagnosed)
913 findings.extend(record_findings)
914 if any(f.code ==
"checker-census-floor" for f
in findings):
916 if f.code ==
"checker-census-floor":
917 print(f
"scope update refused: {f.message}", file=sys.stderr)
919 live_values = len(records)
920 live_value_digest = _authority_value_digest(authorities)
921 live_reason_digest = _authority_reason_digest(records)
922 module = Path(__file__).resolve()
923 _write_constants(module, live_values, live_value_digest, live_reason_digest)
925 f
"blessed scope constants: values={live_values} "
926 f
"value-digest={live_value_digest[:12]} reason-digest={live_reason_digest[:12]}"
931def main(argv: list[str] |
None =
None) -> int:
932 """CLI: check (default), --update to re-freeze the blessed scope constants."""
934 parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
935 parser.add_argument(
"--root", default=str(root))
936 parser.add_argument(
"--update", action=
"store_true", help=
"re-freeze the audited constants")
937 args = parser.parse_args(argv)
938 root = Path(args.root)
939 from suppression_scan
import (
943 raw_candidates, git_findings = git_paths(root)
945 for finding
in git_findings:
946 print(f
"{finding.code}: {finding.message}", file=sys.stderr)
949 return _update(root, raw_candidates)
950 records, findings = scan_checker_scope_controls(root, raw_candidates)
951 for finding
in findings:
952 print(f
"{finding.code}: {finding.message}", file=sys.stderr)
954 print(f
"suppression_checker_scope.py: FAIL -- {len(findings)} finding(s)", file=sys.stderr)
956 print(f
"suppression_checker_scope.py: PASS -- {len(records)} scope value(s) audited")
960if __name__ ==
"__main__":
void main(void)
The application entry point Reset_Handler hands control to.