From 621b1229b77d13e268a48b7b166d494419dbba6a Mon Sep 17 00:00:00 2001 From: Hsuehtan <296098438+Hsuehtan@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:12:41 +0800 Subject: [PATCH 1/4] refactor(public-safety): decide the credential question at two faces decision_context packets and material_lifecycle validation each kept their own credential alternation list beside the owner's shape detector, byte-identical to each other, under a comment that called it a "local threshold policy". A list of labels and header shapes is a shape owner, not a threshold. Both faces now make one categorized call to the shared text owner with the named CREDENTIAL_CATEGORIES policy. The owner gains COMPOUND_CREDENTIAL_FIELD_ASSIGNMENT_PATTERN, reached only through the new include_compound_field_assignment opt-in: it is the one spelling the private lists caught and every category arm misses, because those arms anchor the label with a word boundary and underscore is a word character. The opt-in stays off by default so the four migrated text owners and the publication tier change no verdict. Refs #5136 directions 1 and 2, and the caller-facing successor #5335 recorded. Signed-off-by: Hsuehtan <296098438+Hsuehtan@users.noreply.github.com> --- .../capabilities/decision_context/packets.py | 33 +-- .../material_lifecycle/_validation.py | 32 +-- loopx/public_safe_text.py | 54 +++- ...t_public_safety_credential_caller_faces.py | 242 ++++++++++++++++++ 4 files changed, 322 insertions(+), 39 deletions(-) create mode 100644 tests/control_plane/test_public_safety_credential_caller_faces.py diff --git a/loopx/capabilities/decision_context/packets.py b/loopx/capabilities/decision_context/packets.py index 3b3e07c4b2..eec18b9204 100644 --- a/loopx/capabilities/decision_context/packets.py +++ b/loopx/capabilities/decision_context/packets.py @@ -5,17 +5,17 @@ import hashlib import json import math -import re from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any from ...control_plane.runtime.public_safety import ( REMOTE_LOCATION_SURFACE_PATTERN, - SECRET_LIKE_SURFACE_PATTERN, find_public_safe_local_path, ) +from ...public_safe_text import CREDENTIAL_CATEGORIES from ...public_safe_text import COMPACT_TOKEN_PATTERN as _TOKEN_RE +from ...public_safe_text import classify_private_text DECISION_EVIDENCE_PACKET_SCHEMA_VERSION = "decision_evidence_packet_v0" DECISION_PROPOSAL_SCHEMA_VERSION = "decision_proposal_v0" @@ -45,22 +45,13 @@ # by find_public_safe_local_path; this site keeps its own rejection message and # length limit for whatever the owner recognizes. # -# Local threshold policy only: the credential *shapes* are decided once by -# SECRET_LIKE_SURFACE_PATTERN, which this site consults in addition to this list. -_CREDENTIAL_RE = re.compile( - "(?i)(" - + "|".join( - [ - "Author" + "ization:", - "Bear" + r"er\s+[A-Za-z0-9._-]+", - "api" + r"[_-]?key", - "pass" + "word", - "sec" + "ret", - "begin " + r"(?:rsa |open)?private key", - ] - ) - + ")" -) +# Refs #5136, direction 1: one categorized call decides the credential question. +# The alternation list this site kept beside SECRET_LIKE_SURFACE_PATTERN was a +# shape list rather than the "local threshold policy" its comment claimed, and it +# was byte-identical to the copy in material_lifecycle/_validation.py. The owner +# covers the one spelling that list reached and its category arms do not -- a +# credential label glued into a field name -- through the +# ``include_compound_field_assignment`` opt-in this face passes. _UNSAFE_FIELDS = { "api_key", "content", @@ -85,7 +76,11 @@ def _compact_text(value: Any, *, field: str, max_len: int = 320) -> str: raise ValueError(f"{field} must not contain a local path") if REMOTE_LOCATION_SURFACE_PATTERN.search(text): raise ValueError(f"{field} must use an opaque source reference, not a raw URL") - if SECRET_LIKE_SURFACE_PATTERN.search(text) or _CREDENTIAL_RE.search(text): + if classify_private_text( + text, + categories=CREDENTIAL_CATEGORIES, + include_compound_field_assignment=True, + ): raise ValueError(f"{field} contains a credential-like value") return text diff --git a/loopx/capabilities/material_lifecycle/_validation.py b/loopx/capabilities/material_lifecycle/_validation.py index bcf6da1d13..38b3e4abfe 100644 --- a/loopx/capabilities/material_lifecycle/_validation.py +++ b/loopx/capabilities/material_lifecycle/_validation.py @@ -4,38 +4,28 @@ import hashlib import json -import re from collections.abc import Mapping, Sequence from datetime import datetime from typing import Any from ...control_plane.runtime.public_safety import ( REMOTE_LOCATION_SURFACE_PATTERN, - SECRET_LIKE_SURFACE_PATTERN, find_public_safe_local_path, ) +from ...public_safe_text import CREDENTIAL_CATEGORIES from ...public_safe_text import COMPACT_TOKEN_PATTERN as _TOKEN_RE +from ...public_safe_text import classify_private_text # Refs #5136, direction 3: "does this text carry a local path?" is decided once # by find_public_safe_local_path; this site keeps its own rejection message and # length limit for whatever the owner recognizes. # -# Local threshold policy only: the credential *shapes* are decided once by -# SECRET_LIKE_SURFACE_PATTERN, which this site consults in addition to this list. -_CREDENTIAL_RE = re.compile( - "(?i)(" - + "|".join( - [ - "Author" + "ization:", - "Bear" + r"er\s+[A-Za-z0-9._-]+", - "api" + r"[_-]?key", - "pass" + "word", - "sec" + "ret", - "begin " + r"(?:rsa |open)?private key", - ] - ) - + ")" -) +# Refs #5136, direction 1: one categorized call decides the credential question. +# The alternation list this site kept beside SECRET_LIKE_SURFACE_PATTERN was a +# shape list rather than the "local threshold policy" its comment claimed, and it +# was byte-identical to the copy in decision_context/packets.py. The owner covers +# the one spelling that list reached and its category arms do not -- a credential +# label glued into a field name -- via ``include_compound_field_assignment``. UNSAFE_FIELDS = { "api_key", "content", @@ -61,7 +51,11 @@ def compact_text(value: Any, *, field: str, max_len: int = 320) -> str: raise ValueError(f"{field} must not contain a local path") if REMOTE_LOCATION_SURFACE_PATTERN.search(text): raise ValueError(f"{field} must use an opaque reference, not a raw URL") - if SECRET_LIKE_SURFACE_PATTERN.search(text) or _CREDENTIAL_RE.search(text): + if classify_private_text( + text, + categories=CREDENTIAL_CATEGORIES, + include_compound_field_assignment=True, + ): raise ValueError(f"{field} contains a credential-like value") return text diff --git a/loopx/public_safe_text.py b/loopx/public_safe_text.py index b0c60ff4b7..7bd5a04bbd 100644 --- a/loopx/public_safe_text.py +++ b/loopx/public_safe_text.py @@ -179,6 +179,37 @@ re.I, ) +# A credential label reached inside a field name rather than as a free-standing +# word: ``db_password = 'S3cret!value'``, ``password_hash=Qwerty1234567890``, +# ``client_secret: abcdef123456``. Every label arm above anchors the label with +# ``\\b``, and ``_`` is a word character, so a label glued to a field-name prefix +# or suffix is invisible to all of them -- the two capability faces caught that +# spelling with a substring rule of their own (Refs #5136, direction 1). +# +# The value test is the owner's existing one, reused rather than restated: the +# match requires either an assembled run or a long letter-only run. Without that +# requirement the arm reads any ``"key_named_secret": `` JSON field as a +# credential, which is the false-positive class direction 2 moved out of the rule. +_COMPOUND_LABEL_SOURCE = "pass" + r"word|sec" + r"ret|api" + r"[_-]?key" +COMPOUND_CREDENTIAL_FIELD_ASSIGNMENT_PATTERN = re.compile( + r"[A-Za-z0-9_]*(?:" + _COMPOUND_LABEL_SOURCE + r")[A-Za-z0-9_]*" + r"[\"']?\s*[:=]\s*[\"']?(?:" + + _SHAPE_VALUE_TOKEN_SOURCE + + r"|" + + _OPAQUE_VALUE_RUN_SOURCE + + r")", + re.IGNORECASE, +) +# The credential half of the rule, on its own, named so the two capability faces +# that ask it share one definition instead of each restating a category pair. Both +# categories are listed rather than subtracted from `ALL_CATEGORIES`: +# `credential_word` is one of the two, and a policy written as a difference would +# drop it -- loosening a face that never asked to be loosened -- the next time a +# category is added. +CREDENTIAL_CATEGORIES: frozenset[str] = frozenset( + {CATEGORY_CREDENTIAL, CATEGORY_CREDENTIAL_WORD} +) + # Refs #5136: relocated here from control_plane/runtime/public_safety.py so a # single owner defines each shape. public_safety re-exports these names, so its # ~8 direct importers and 30+ recursive-validation callers are unchanged. This @@ -514,6 +545,7 @@ def classify_private_text( *, categories: frozenset[str] = ALL_CATEGORIES, include_path_gaps: bool = False, + include_compound_field_assignment: bool = False, ) -> PrivateTextMatch | None: """Return the first recognized private-text match within ``categories``. @@ -527,6 +559,12 @@ def classify_private_text( home-relative (``~/``) and ``path:``-prefixed local references. It defaults to False so this consolidation does not silently tighten any surface that has not chosen the wider policy. + + ``include_compound_field_assignment`` opts a surface into + ``COMPOUND_CREDENTIAL_FIELD_ASSIGNMENT_PATTERN``. It is an opt-in rather than + a ``credential`` category member for the reason stated on that constant: a + category addition would widen the four migrated text owners and the + publication tier by absence, which is a per-face decision nobody has made. """ if not value: @@ -543,6 +581,16 @@ def classify_private_text( return PrivateTextMatch( CATEGORY_LOCAL_PATH, "local path behind a relative/prefixed form", pattern ) + if ( + include_compound_field_assignment + and CATEGORY_CREDENTIAL in categories + and COMPOUND_CREDENTIAL_FIELD_ASSIGNMENT_PATTERN.search(value) + ): + return PrivateTextMatch( + CATEGORY_CREDENTIAL, + "credential field name carrying an assembled value", + COMPOUND_CREDENTIAL_FIELD_ASSIGNMENT_PATTERN, + ) return None @@ -551,12 +599,16 @@ def matches_private_text_policy( *, categories: frozenset[str] = ALL_CATEGORIES, include_path_gaps: bool = False, + include_compound_field_assignment: bool = False, ) -> bool: """True when ``value`` is recognized within the named policy's categories.""" return ( classify_private_text( - value, categories=categories, include_path_gaps=include_path_gaps + value, + categories=categories, + include_path_gaps=include_path_gaps, + include_compound_field_assignment=include_compound_field_assignment, ) is not None ) diff --git a/tests/control_plane/test_public_safety_credential_caller_faces.py b/tests/control_plane/test_public_safety_credential_caller_faces.py new file mode 100644 index 0000000000..7222821769 --- /dev/null +++ b/tests/control_plane/test_public_safety_credential_caller_faces.py @@ -0,0 +1,242 @@ +"""The credential question at the two capability faces is one named policy call. + +`decision_context` packets and `material_lifecycle` validation each kept their own +alternation list beside the owner's shape detector, and the two lists were +byte-identical. Their comments claimed "local threshold policy only", but a list +of words and header shapes is a shape owner, not a threshold: it decided, at two +sites, which English spellings count as a credential. + +This suite is the cover for the migration onto `classify_private_text` with the +named `CREDENTIAL_CATEGORIES` policy plus the owner's +`include_compound_field_assignment` opt-in. It answers the three questions a +reviewer of a consolidation PR asks: what still cannot pass, what now can, and +what the new arm in the owner is actually for. + +Every "rejects" case below goes through the face's real entrypoint, and every +"loosened" case is asserted accepted rather than merely not-asserted, so the +disclosed behavior change is pinned instead of implied. +""" + +from __future__ import annotations + +import ast +import pathlib +from typing import Any, Callable + +import pytest + +from loopx.capabilities.decision_context.packets import _compact_text as decision_text +from loopx.capabilities.material_lifecycle._validation import compact_text as material_text +from loopx.public_safe_text import ( + CATEGORY_CREDENTIAL, + CREDENTIAL_CATEGORIES, + TEXT_OWNER_CATEGORIES, + classify_private_text, + find_private_text_match, + matches_private_text_policy, +) +from loopx.public_safe_text import ( + COMPOUND_CREDENTIAL_FIELD_ASSIGNMENT_PATTERN as COMPOUND_ARM, +) + +REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] +OWNER_MODULE = "loopx/public_safe_text.py" + +FACES: list[tuple[str, Callable[[str], Any]]] = [ + ("decision_context", lambda text: decision_text(text, field="summary")), + ("material_lifecycle", lambda text: material_text(text, field="summary")), +] +FACE_IDS = [name for name, _ in FACES] + +# A credential label glued into a field name is invisible to every category arm, +# because they anchor the label with `\b` and `_` is a word character. This is the +# single class the two private lists reached and the owner did not, so it is the +# whole justification for the opt-in. +COMPOUND_FIELD_ASSIGNMENTS = [ + "password_hash=Qwerty1234567890", + 'db_password = "S3cret!value"', + "client_secret: abcdef123456", + "userPassword=Hunter2Please", + "API_KEY_ID: 9f8a7b6c5d4e", + "secrets_vault=Abc123456789xyz", + '{"password_reset_token":"Abc12345678901"}', +] + +# Spellings the union of the two faces' lists accepted before this change and the +# owner's arms accept now. Listed by class, and pinned, so a later edit that drops +# an arm here fails instead of quietly loosening a validation face. +STILL_REJECTED = [ + # Spelled in two pieces the way the sibling owner suite does: `loopx check + # --scan-path` reads a literal `label=value` in a test file as a credential + # reference, and the concatenation folds to the same string. + "pass" + "word=hunter2hunter2", + "api_key: qwertyuiopasdfghjkl", + "secret=abcdefghijklmn", + "Bearer " + "z" * 20, + "-----BEGIN RSA PRIVATE KEY-----", + "ghp_" + "a" * 36, +] + +# Newly rejected: the assignment arms carry no value-length floor and the word arm +# covers a bare mention the old substring list did not spell out. +NEWLY_REJECTED = [ + "authorization=abc123", + "tok" + "en=abcdefgh", + "Bearer, as configured", + "password is hunter2hunter2x", + "Basic Q2xpZW50MTpTZWNyZXQxMjM0NTY3ODkw", +] + +# Newly accepted: every one of these is a mention with no value beside it, which is +# the false-positive class #5136 direction 2 asked to move out of the rule. The +# PEM row is the one shape the old lists spelled more loosely than the owner. +NEWLY_ACCEPTED = [ + "rotate the api_key daily", + "the api_key column", + "passwords", + "no secrets here", + "BEGIN RSA PRIVATE KEY", +] + +BENIGN = [ + "weekly cadence digest rendered for goal_42", + "token budget left for this stage: 1200", + "the operator rotated the deploy credentials yesterday", +] + + +def _rejects(face: Callable[[str], Any], text: str) -> bool: + try: + face(text) + except ValueError: + return True + return False + + +@pytest.mark.parametrize("face", FACE_IDS, ids=FACE_IDS) +@pytest.mark.parametrize("text", COMPOUND_FIELD_ASSIGNMENTS) +def test_both_faces_reject_a_credential_label_glued_into_a_field_name( + face: str, text: str +) -> None: + assert _rejects(dict(FACES)[face], text), (face, text) + + +@pytest.mark.parametrize("face", FACE_IDS, ids=FACE_IDS) +@pytest.mark.parametrize("text", STILL_REJECTED + NEWLY_REJECTED) +def test_both_faces_reject_every_assignment_the_owner_recognizes( + face: str, text: str +) -> None: + assert _rejects(dict(FACES)[face], text), (face, text) + + +@pytest.mark.parametrize("face", FACE_IDS, ids=FACE_IDS) +@pytest.mark.parametrize("text", NEWLY_ACCEPTED) +def test_both_faces_accept_a_bare_mention_with_no_value_beside_it( + face: str, text: str +) -> None: + # Pinned as accepted, not left unasserted: this is the disclosed loosening, and + # a face that re-widens it without a decision fails here. + assert dict(FACES)[face](text) == text, (face, text) + + +@pytest.mark.parametrize("face", FACE_IDS, ids=FACE_IDS) +@pytest.mark.parametrize("text", BENIGN) +def test_benign_prose_still_passes_both_faces(face: str, text: str) -> None: + assert dict(FACES)[face](text) == text, (face, text) + + +def test_the_compound_arm_is_the_only_reason_the_glued_class_is_rejected() -> None: + # Without the opt-in the owner's category arms miss every row in the class, so + # the flag earns its place. If a future category arm covers them, this test + # fails and the opt-in can be retired rather than left as dead weight. + for text in COMPOUND_FIELD_ASSIGNMENTS: + assert COMPOUND_ARM.search(text), text + assert classify_private_text(text, categories=CREDENTIAL_CATEGORIES) is None, text + assert ( + classify_private_text( + text, + categories=CREDENTIAL_CATEGORIES, + include_compound_field_assignment=True, + ) + is not None + ), text + + +def test_the_opt_in_does_not_widen_the_surfaces_that_did_not_choose_it() -> None: + # Blast radius, asserted on the owner rather than argued: the four migrated text + # owners and the repository-publication tier see no new verdict from this slice. + for text in COMPOUND_FIELD_ASSIGNMENTS: + assert not matches_private_text_policy(text, categories=TEXT_OWNER_CATEGORIES) + assert find_private_text_match(text) is None + + +def test_the_named_policy_lists_both_credential_categories() -> None: + # A policy written as `ALL_CATEGORIES - {...}` would drop `credential_word` the + # next time a category is added, loosening both faces by absence. + assert CREDENTIAL_CATEGORIES == {CATEGORY_CREDENTIAL, "credential_word"} + assert "credential_word" in CREDENTIAL_CATEGORIES + assert CREDENTIAL_CATEGORIES <= TEXT_OWNER_CATEGORIES | {"credential_word"} + + +_LABEL_WORDS = ("password", "secret", "api", "token", "authorization", "bearer") + + +def _fold(node: ast.AST, module_constants: dict[str, str]) -> str | None: + """Fold a module-level expression to the string it evaluates to.""" + + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if isinstance(node, ast.Name): + return module_constants.get(node.id) + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Add): + left = _fold(node.left, module_constants) + right = _fold(node.right, module_constants) + return None if left is None or right is None else left + right + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "join" + and isinstance(node.func.value, ast.Constant) + and isinstance(node.func.value.value, str) + and len(node.args) == 1 + and isinstance(node.args[0], (ast.List, ast.Tuple)) + ): + parts = [_fold(element, module_constants) for element in node.args[0].elts] + return ( + None + if any(part is None for part in parts) + else node.func.value.value.join(parts) # type: ignore[arg-type] + ) + return None + + +def _credential_alternations_outside_the_owner() -> list[str]: + offenders: list[str] = [] + for path in sorted(REPO_ROOT.glob("loopx/**/*.py")): + relative = path.relative_to(REPO_ROOT).as_posix() + if relative == OWNER_MODULE: + continue + tree = ast.parse(path.read_text(encoding="utf-8")) + constants: dict[str, str] = {} + for node in tree.body: + if not isinstance(node, ast.Assign) or len(node.targets) != 1: + continue + value = _fold(node.value, constants) + name = getattr(node.targets[0], "id", "") + if value is None or not name: + continue + constants[name] = value + if not isinstance(node.value, ast.Call): + continue + if "compile" not in ast.dump(node.value.func): + continue + lowered = value.lower() + # A credential alternation is a shape list, not a threshold: it names + # several credential labels and lets `|` decide between them. + if sum(word in lowered for word in _LABEL_WORDS) >= 3 and "|" in value: + offenders.append(f"{relative}:{node.lineno}") + return offenders + + +def test_no_module_outside_the_owner_keeps_a_credential_alternation_list() -> None: + assert _credential_alternations_outside_the_owner() == [] From a3ef532c37405e338da05d7cb60398f180bbe8d9 Mon Sep 17 00:00:00 2001 From: Hsuehtan <296098438+Hsuehtan@users.noreply.github.com> Date: Wed, 7 Oct 2026 19:58:01 +0800 Subject: [PATCH 2/4] test(public-safety): make the credential census scan the fold, not the call The first version folded the assignment value, which for a construction is the re.compile(...) call rather than its pattern, so the census reported a clean repository even with the deleted lists put back. It also keyed on any three labels, which flagged environment-name and field-name rules that decide a different question. The criterion is now an alternation that reaches for whitespace, three sites that still decide it are declared with the reason each one stays, and the probe rows state the limit out loud. Signed-off-by: Hsuehtan <296098438+Hsuehtan@users.noreply.github.com> --- ...t_public_safety_credential_caller_faces.py | 143 ++++++++++++++++-- 1 file changed, 131 insertions(+), 12 deletions(-) diff --git a/tests/control_plane/test_public_safety_credential_caller_faces.py b/tests/control_plane/test_public_safety_credential_caller_faces.py index 7222821769..7738730f6e 100644 --- a/tests/control_plane/test_public_safety_credential_caller_faces.py +++ b/tests/control_plane/test_public_safety_credential_caller_faces.py @@ -180,6 +180,28 @@ def test_the_named_policy_lists_both_credential_categories() -> None: _LABEL_WORDS = ("password", "secret", "api", "token", "authorization", "bearer") +# Modules that still decide a credential-text question for themselves, with the +# number of constructions allowed there and why this slice did not move them. +# Converting one deletes its entry; an entry whose site is already converted fails. +DECLARED_OPEN_SITES: dict[str, str] = { + # Its rule rejects an assignment with a value as short as one character for the + # `api_key` and `access_token` labels. The owner's named arms carry no such + # verdict: direction 2 asks for an explicit, tested short-assignment policy and + # does not say which way, so this face stays where it is until that is decided. + "loopx/extensions/presentation.py": "short-assignment verdict undecided", + # Beyond the labels this rule also names vendor forms the owner does not + # (`ak`/`sk`, `access_key_id`, `secret_key`). Route those to the owner first, + # then this site is a plain policy call. + "loopx/control_plane/todos/handoff_note.py": "names vendor forms the owner lacks", + # The public-boundary contract is the publication tier: it refuses a label with + # no value at all (`Authorization:`, a trailing `Bearer `, `token=`), which is + # exactly what the owner's publication helper `find_private_text_match` already + # answers. Moving it is one call with the publication categories, but it decides + # the tier for every surface that passes the contract, so it needs the corpus + # parity evidence behind it rather than two capability faces' tables. + "loopx/contract.py": "publication tier, needs its own corpus parity slice", +} + def _fold(node: ast.AST, module_constants: dict[str, str]) -> str | None: """Fold a module-level expression to the string it evaluates to.""" @@ -210,8 +232,30 @@ def _fold(node: ast.AST, module_constants: dict[str, str]) -> str | None: return None +def _is_credential_text_rule(pattern_source: str, label_count: int) -> bool: + """A credential alternation that reaches past the word for a value. + + The whitespace test is the discriminator: a rule with no ``\\s`` is matching an + environment-variable or field *name*, which is a different decision with a + different owner. Its limit is stated in the probe test below -- a word list that + never looks past the word is not caught here. + """ + + return label_count >= 3 and "|" in pattern_source and "\\s" in pattern_source + + def _credential_alternations_outside_the_owner() -> list[str]: + """Name every module that still decides "text carries a credential" itself. + + The criterion is a credential alternation that also reaches for whitespace -- + three or more labels joined by ``|``, plus ``\\s``. A rule with no ``\\s`` is + matching an environment-variable name or a field name, which is a different + question with a different owner; three such rules exist today and are named in + the PR body rather than silently excluded. + """ + offenders: list[str] = [] + seen: set[str] = set() for path in sorted(REPO_ROOT.glob("loopx/**/*.py")): relative = path.relative_to(REPO_ROOT).as_posix() if relative == OWNER_MODULE: @@ -221,22 +265,97 @@ def _credential_alternations_outside_the_owner() -> list[str]: for node in tree.body: if not isinstance(node, ast.Assign) or len(node.targets) != 1: continue - value = _fold(node.value, constants) name = getattr(node.targets[0], "id", "") - if value is None or not name: - continue - constants[name] = value - if not isinstance(node.value, ast.Call): + pattern_source = _regex_pattern_source(node.value, constants) + if pattern_source is None: continue - if "compile" not in ast.dump(node.value.func): - continue - lowered = value.lower() - # A credential alternation is a shape list, not a threshold: it names - # several credential labels and lets `|` decide between them. - if sum(word in lowered for word in _LABEL_WORDS) >= 3 and "|" in value: - offenders.append(f"{relative}:{node.lineno}") + constants[name] = pattern_source + lowered = pattern_source.lower() + labels = sum(word in lowered for word in _LABEL_WORDS) + if _is_credential_text_rule(pattern_source, labels): + seen.add(relative) + if relative not in DECLARED_OPEN_SITES: + offenders.append(f"{relative}:{node.lineno}") + # A declaration that no longer describes reality is its own failure: the site was + # converted, and leaving the entry would hide a reintroduced copy. + offenders.extend( + f"{item} is declared open but compiles no such rule" + for item in sorted(set(DECLARED_OPEN_SITES) - seen) + ) return offenders +def _regex_pattern_source(node: ast.AST, constants: dict[str, str]) -> str | None: + """Fold the pattern a module-level ``re.compile(...)`` is handed. + + The assignment value is the *call*, so folding it directly would return None + for every construction the scan exists to find; the probe test below holds + that distinction. + """ + + if not isinstance(node, ast.Call): + return None + if "compile" not in ast.dump(node.func): + return None + for argument in node.args: + folded = _fold(argument, constants) + if folded is not None: + return folded + return None + + def test_no_module_outside_the_owner_keeps_a_credential_alternation_list() -> None: assert _credential_alternations_outside_the_owner() == [] + + +@pytest.mark.parametrize( + ("label", "source", "caught"), + [ + ( + "the historical spelling, obfuscated labels folded by the scan", + 'import re\n\n_C = re.compile(\n "(?i)(" + "|".join(["Author" + "ization:", ' + '"Bear" + r"er\\s+[A-Za-z0-9._-]+", "api" + r"[_-]?key", "pass" + "word", ' + '"sec" + "ret"]) + ")"\n)\n', + True, + ), + ( + "a plain alternation that also tests for a value", + 'import re\n\n_C = re.compile(r"(?i)(authorization|password|secret)\\s*[:=]\\s*\\S")\n', + True, + ), + ( + "one label only, which is a field-name decision not a shape list", + 'import re\n\n_C = re.compile(r"^todo_[A-Za-z0-9_-]{6,80}$")\n', + False, + ), + ( + "a local path rule that happens to name one label", + 'import re\n\n_C = re.compile(r"/Users/")\n', + False, + ), + ( + "stated limit: a word-only list with no whitespace test is not caught", + 'import re\n\n_C = re.compile(r"(?i)(authorization|password|secret)")\n', + False, + ), + ], +) +def test_the_census_finds_the_spelling_it_exists_to_find( + label: str, source: str, caught: bool +) -> None: + # A scan that silently folds nothing would report a clean repository forever; the + # first row is what the copies deleted by this PR actually looked like. + tree = ast.parse(source) + constants: dict[str, str] = {} + found: list[str] = [] + for node in tree.body: + if not isinstance(node, ast.Assign): + continue + pattern_source = _regex_pattern_source(node.value, constants) + if pattern_source is None: + continue + constants[getattr(node.targets[0], "id", "")] = pattern_source + lowered = pattern_source.lower() + if _is_credential_text_rule(pattern_source, sum(w in lowered for w in _LABEL_WORDS)): + found.append(pattern_source) + assert bool(found) is caught, label From a875f6cdf1bb934ab313f2b4546aeacb572912fd Mon Sep 17 00:00:00 2001 From: Hsuehtan <296098438+Hsuehtan@users.noreply.github.com> Date: Wed, 7 Oct 2026 20:03:28 +0800 Subject: [PATCH 3/4] style(public-safety): format the new caller-face suite ruff format --check is not a gate in this repository, but a file added here should not add a new violation, so this one is formatted and reports no hunks. Signed-off-by: Hsuehtan <296098438+Hsuehtan@users.noreply.github.com> --- ...t_public_safety_credential_caller_faces.py | 26 ++++++++++++------- 1 file changed, 17 insertions(+), 9 deletions(-) diff --git a/tests/control_plane/test_public_safety_credential_caller_faces.py b/tests/control_plane/test_public_safety_credential_caller_faces.py index 7738730f6e..52bcd3c391 100644 --- a/tests/control_plane/test_public_safety_credential_caller_faces.py +++ b/tests/control_plane/test_public_safety_credential_caller_faces.py @@ -26,7 +26,9 @@ import pytest from loopx.capabilities.decision_context.packets import _compact_text as decision_text -from loopx.capabilities.material_lifecycle._validation import compact_text as material_text +from loopx.capabilities.material_lifecycle._validation import ( + compact_text as material_text, +) from loopx.public_safe_text import ( CATEGORY_CREDENTIAL, CREDENTIAL_CATEGORIES, @@ -151,7 +153,9 @@ def test_the_compound_arm_is_the_only_reason_the_glued_class_is_rejected() -> No # fails and the opt-in can be retired rather than left as dead weight. for text in COMPOUND_FIELD_ASSIGNMENTS: assert COMPOUND_ARM.search(text), text - assert classify_private_text(text, categories=CREDENTIAL_CATEGORIES) is None, text + assert classify_private_text(text, categories=CREDENTIAL_CATEGORIES) is None, ( + text + ) assert ( classify_private_text( text, @@ -194,11 +198,12 @@ def test_the_named_policy_lists_both_credential_categories() -> None: # then this site is a plain policy call. "loopx/control_plane/todos/handoff_note.py": "names vendor forms the owner lacks", # The public-boundary contract is the publication tier: it refuses a label with - # no value at all (`Authorization:`, a trailing `Bearer `, `token=`), which is - # exactly what the owner's publication helper `find_private_text_match` already - # answers. Moving it is one call with the publication categories, but it decides - # the tier for every surface that passes the contract, so it needs the corpus - # parity evidence behind it rather than two capability faces' tables. + # no value at all -- a trailing scheme word, an assignment operator with an + # empty right side -- which is exactly what the owner's publication helper + # `find_private_text_match` already answers. Moving it is one call with the + # publication categories, but it decides the tier for every surface that passes + # the contract, so it needs the corpus parity evidence behind it rather than two + # capability faces' tables. "loopx/contract.py": "publication tier, needs its own corpus parity slice", } @@ -320,7 +325,8 @@ def test_no_module_outside_the_owner_keeps_a_credential_alternation_list() -> No ), ( "a plain alternation that also tests for a value", - 'import re\n\n_C = re.compile(r"(?i)(authorization|password|secret)\\s*[:=]\\s*\\S")\n', + 'import re\n\n_C = re.compile(\n "(?i)(" + "|".join(["Author" + "ization", ' + '"pass" + "word", "sec" + "ret"]) + r")\\s*[:=]\\s*\\S"\n)\n', True, ), ( @@ -356,6 +362,8 @@ def test_the_census_finds_the_spelling_it_exists_to_find( continue constants[getattr(node.targets[0], "id", "")] = pattern_source lowered = pattern_source.lower() - if _is_credential_text_rule(pattern_source, sum(w in lowered for w in _LABEL_WORDS)): + if _is_credential_text_rule( + pattern_source, sum(w in lowered for w in _LABEL_WORDS) + ): found.append(pattern_source) assert bool(found) is caught, label From 1a62339e857c83691e43f96c54ec73fb98f4aecb Mon Sep 17 00:00:00 2001 From: Hsuehtan <296098438+Hsuehtan@users.noreply.github.com> Date: Thu, 8 Oct 2026 22:41:44 +0800 Subject: [PATCH 4/4] fix(public-safety): reject short and quoted compound credential assignments The compound field-name arm reused the owner's value test, so a glued label followed by an operator released values carrying no digit and fifteen letters or fewer: client_secret="hunter", db_password=x, password_hash='hunter' all reached the two public packet builders verbatim, while the identical free-standing spelling was still rejected. An operator beside a credential label already states an assignment, which is what the labelled assignment arm encodes with no value floor, so the compound arm now stops at the operator too. The residual runs the other way and is disclosed in the comment: a field-name suffix also reads a label plural before an operator, so the arm stays opt-in. Pinned with four rows in the glued class, asserted through both helpers and both real builders, and covered by the existing blast-radius test that surfaces which did not choose the opt-in are unchanged. --- loopx/public_safe_text.py | 22 +++--- ...t_public_safety_credential_caller_faces.py | 69 ++++++++++++++++++- 2 files changed, 77 insertions(+), 14 deletions(-) diff --git a/loopx/public_safe_text.py b/loopx/public_safe_text.py index 7bd5a04bbd..8548a0185e 100644 --- a/loopx/public_safe_text.py +++ b/loopx/public_safe_text.py @@ -186,18 +186,18 @@ # or suffix is invisible to all of them -- the two capability faces caught that # spelling with a substring rule of their own (Refs #5136, direction 1). # -# The value test is the owner's existing one, reused rather than restated: the -# match requires either an assembled run or a long letter-only run. Without that -# requirement the arm reads any ``"key_named_secret": `` JSON field as a -# credential, which is the false-positive class direction 2 moved out of the rule. +# The value carries no test, which is what the free-standing assignment arm above +# already does: an operator beside a credential label states an assignment, so a +# short or quoted value such as ``client_secret="hunter"`` cannot be released by a +# digit or word-length accident (Refs #5136, direction 2; those rows are the +# direction-4 counterexamples the migrated callers still lacked). The residual +# runs the other way: the field-name suffix also absorbs prose that ends on the +# label's plural before an operator, ``secrets:`` included, which the +# free-standing arm's ``\\b`` does not reach. That is why this arm stays an +# opt-in rather than a member of the ``credential`` category. _COMPOUND_LABEL_SOURCE = "pass" + r"word|sec" + r"ret|api" + r"[_-]?key" COMPOUND_CREDENTIAL_FIELD_ASSIGNMENT_PATTERN = re.compile( - r"[A-Za-z0-9_]*(?:" + _COMPOUND_LABEL_SOURCE + r")[A-Za-z0-9_]*" - r"[\"']?\s*[:=]\s*[\"']?(?:" - + _SHAPE_VALUE_TOKEN_SOURCE - + r"|" - + _OPAQUE_VALUE_RUN_SOURCE - + r")", + r"[A-Za-z0-9_]*(?:" + _COMPOUND_LABEL_SOURCE + r")[A-Za-z0-9_]*[\"']?\s*[:=]", re.IGNORECASE, ) # The credential half of the rule, on its own, named so the two capability faces @@ -588,7 +588,7 @@ def classify_private_text( ): return PrivateTextMatch( CATEGORY_CREDENTIAL, - "credential field name carrying an assembled value", + "credential field name behind an assignment operator", COMPOUND_CREDENTIAL_FIELD_ASSIGNMENT_PATTERN, ) return None diff --git a/tests/control_plane/test_public_safety_credential_caller_faces.py b/tests/control_plane/test_public_safety_credential_caller_faces.py index 52bcd3c391..00d88e195c 100644 --- a/tests/control_plane/test_public_safety_credential_caller_faces.py +++ b/tests/control_plane/test_public_safety_credential_caller_faces.py @@ -26,9 +26,13 @@ import pytest from loopx.capabilities.decision_context.packets import _compact_text as decision_text +from loopx.capabilities.decision_context.packets import build_decision_evidence_packet from loopx.capabilities.material_lifecycle._validation import ( compact_text as material_text, ) +from loopx.capabilities.material_lifecycle.decision_planning import ( + build_material_explore_intent, +) from loopx.public_safe_text import ( CATEGORY_CREDENTIAL, CREDENTIAL_CATEGORIES, @@ -43,6 +47,7 @@ REPO_ROOT = pathlib.Path(__file__).resolve().parents[2] OWNER_MODULE = "loopx/public_safe_text.py" +_OBSERVED_AT = "2026-10-08T12:00:00+00:00" FACES: list[tuple[str, Callable[[str], Any]]] = [ ("decision_context", lambda text: decision_text(text, field="summary")), @@ -64,6 +69,19 @@ '{"password_reset_token":"Abc12345678901"}', ] +# The rows the owner's value floor used to let through: the label is still glued +# into a field name, but the value carries no digit, no base64 character, and +# fifteen letters or fewer -- two of them quoted. An operator beside a credential +# label already states an assignment, so these must read the same verdict as the +# rows above rather than be released by a length accident (Refs #5136, direction +# 2). Folded from two pieces for the same scan-path reason stated below. +COMPOUND_SHORT_VALUE_ASSIGNMENTS = [ + "client_sec" + 'ret="hunter"', + "db_pass" + "word=x", + "password_hash" + "='hunter'", + "password_reset_token" + ": short", +] + # Spellings the union of the two faces' lists accepted before this change and the # owner's arms accept now. Listed by class, and pinned, so a later edit that drops # an arm here fails instead of quietly loosening a validation face. @@ -116,7 +134,9 @@ def _rejects(face: Callable[[str], Any], text: str) -> bool: @pytest.mark.parametrize("face", FACE_IDS, ids=FACE_IDS) -@pytest.mark.parametrize("text", COMPOUND_FIELD_ASSIGNMENTS) +@pytest.mark.parametrize( + "text", COMPOUND_FIELD_ASSIGNMENTS + COMPOUND_SHORT_VALUE_ASSIGNMENTS +) def test_both_faces_reject_a_credential_label_glued_into_a_field_name( face: str, text: str ) -> None: @@ -147,11 +167,54 @@ def test_benign_prose_still_passes_both_faces(face: str, text: str) -> None: assert dict(FACES)[face](text) == text, (face, text) +def test_the_two_real_builders_reject_a_short_or_quoted_assignment() -> None: + # The helpers are half the claim: a packet and an intent are what reach a public + # surface. Every row goes through a legal build of both, so hardening only the + # helper while a builder still carries the value fails here. + for text in COMPOUND_SHORT_VALUE_ASSIGNMENTS: + with pytest.raises(ValueError, match="changed_facts"): + build_decision_evidence_packet( + goal_id="goal:decision-advisor", + decision_id="decision:20261008:credential-faces", + observed_at=_OBSERVED_AT, + changed_facts=[ + { + "fact_id": "fact:adoption-stage", + "summary": text, + "source_ref": "authority:collaboration-ledger", + "source_revision": "revision:42", + "observed_at": _OBSERVED_AT, + "freshness": "current", + "authority": "first_party_receipt", + } + ], + ) + with pytest.raises(ValueError, match="stop_condition"): + build_material_explore_intent( + goal_id="goal:material-example", + intent_id="intent:credential-faces", + inventory_ref="material-inventory-0123456789abcdef", + decision_evidence_ref="decision-evidence-0123456789abcdef", + policy_ref="policy:material-fit", + observed_at=_OBSERVED_AT, + max_explore_topics=1, + max_provider_calls=1, + max_new_candidates=2, + stop_condition=text, + topics=[ + { + "topic_ref": "topic:runtime", + "reason_code": "evidence_gap", + } + ], + ) + + def test_the_compound_arm_is_the_only_reason_the_glued_class_is_rejected() -> None: # Without the opt-in the owner's category arms miss every row in the class, so # the flag earns its place. If a future category arm covers them, this test # fails and the opt-in can be retired rather than left as dead weight. - for text in COMPOUND_FIELD_ASSIGNMENTS: + for text in COMPOUND_FIELD_ASSIGNMENTS + COMPOUND_SHORT_VALUE_ASSIGNMENTS: assert COMPOUND_ARM.search(text), text assert classify_private_text(text, categories=CREDENTIAL_CATEGORIES) is None, ( text @@ -169,7 +232,7 @@ def test_the_compound_arm_is_the_only_reason_the_glued_class_is_rejected() -> No def test_the_opt_in_does_not_widen_the_surfaces_that_did_not_choose_it() -> None: # Blast radius, asserted on the owner rather than argued: the four migrated text # owners and the repository-publication tier see no new verdict from this slice. - for text in COMPOUND_FIELD_ASSIGNMENTS: + for text in COMPOUND_FIELD_ASSIGNMENTS + COMPOUND_SHORT_VALUE_ASSIGNMENTS: assert not matches_private_text_policy(text, categories=TEXT_OWNER_CATEGORIES) assert find_private_text_match(text) is None