refactor(public-safety): decide credential shapes in one owner

loopx/capabilities/decision_context/packets.py

@@ -10,6 +10,8 @@
 from datetime import datetime
 from typing import Any
 
+from ...control_plane.runtime.public_safety import SECRET_LIKE_SURFACE_PATTERN
+
 DECISION_EVIDENCE_PACKET_SCHEMA_VERSION = "decision_evidence_packet_v0"
 DECISION_PROPOSAL_SCHEMA_VERSION = "decision_proposal_v0"
 DECISION_REVIEW_RECEIPT_SCHEMA_VERSION = "decision_review_receipt_v0"
@@ -37,6 +39,8 @@
 _TOKEN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
 _LOCAL_PATH_RE = re.compile(r"(^|[\s:=])(?:/Users/|/private/|/tmp/|~/)")
 _RAW_LOCATION_RE = re.compile(r"(?i)\b(?:https?|file|s3|gs|tos|hdfs)://")
+# 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(
@@ -75,7 +79,7 @@ def _compact_text(value: Any, *, field: str, max_len: int = 320) -> str:
         raise ValueError(f"{field} must not contain a local path")
     if _RAW_LOCATION_RE.search(text):
         raise ValueError(f"{field} must use an opaque source reference, not a raw URL")
-    if _CREDENTIAL_RE.search(text):
+    if SECRET_LIKE_SURFACE_PATTERN.search(text) or _CREDENTIAL_RE.search(text):
         raise ValueError(f"{field} contains a credential-like value")
     return text
 

loopx/capabilities/material_lifecycle/_validation.py

@@ -9,9 +9,13 @@
 from datetime import datetime
 from typing import Any
 
+from ...control_plane.runtime.public_safety import SECRET_LIKE_SURFACE_PATTERN
+
 _TOKEN_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$")
 _LOCAL_PATH_RE = re.compile(r"(^|[\s:=])(?:/Users/|/private/|/tmp/|~/)")
 _RAW_LOCATION_RE = re.compile(r"(?i)\b(?:https?|file|s3|gs|tos|hdfs)://")
+# 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(
@@ -51,7 +55,7 @@ def compact_text(value: Any, *, field: str, max_len: int = 320) -> str:
         raise ValueError(f"{field} must not contain a local path")
     if _RAW_LOCATION_RE.search(text):
         raise ValueError(f"{field} must use an opaque reference, not a raw URL")
-    if _CREDENTIAL_RE.search(text):
+    if SECRET_LIKE_SURFACE_PATTERN.search(text) or _CREDENTIAL_RE.search(text):
         raise ValueError(f"{field} contains a credential-like value")
     return text
 

loopx/capabilities/periodic_report/core.py

@@ -7,6 +7,8 @@
 from datetime import datetime, timezone
 from typing import Any
 
+from ...control_plane.runtime.public_safety import SECRET_LIKE_SURFACE_PATTERN
+
 
 REQUEST_SCHEMA = "periodic_report_run_request_v0"
 RUN_SCHEMA = "periodic_report_v0"
@@ -36,11 +38,6 @@
 _LOCAL_PATH_SURFACE_PATTERN = re.compile(
     r"(?<!<)/(?:Users|Volumes|home|var/folders|tmp|private/tmp)/[^\s`'\"<>]+"
 )
-_SECRET_LIKE_SURFACE_PATTERN = re.compile(
-    r"(?i)(?:\bbearer\s+[a-z0-9._~+/=-]{16,}|"
-    r"(?<![a-z0-9_])(?:ak|sk)[-_=:][a-z0-9_=-]{10,}|"
-    r"\b(?:api[_-]?key|password|secret|token)\s*[=:]\s*[^\s`'\"<>]{12,})"
-)
 _FORBIDDEN_RAW_KEYS = {
     "credential",
     "credentials",
@@ -140,7 +137,7 @@ def _reject_raw_keys(value: object, label: str) -> None:
             _reject_raw_keys(item, f"{label}[{index}]")
     elif isinstance(value, str) and (
         _LOCAL_PATH_SURFACE_PATTERN.search(value)
-        or _SECRET_LIKE_SURFACE_PATTERN.search(value)
+        or SECRET_LIKE_SURFACE_PATTERN.search(value)
     ):
         raise ValueError(f"{label} contains a private path or credential-like value")
 

loopx/control_plane/goals/artifact_lifecycle.py

@@ -17,11 +17,14 @@
 
 from __future__ import annotations
 
-import re
 from typing import Any
 
 from ...public_safe_text import find_private_text_match
-from ..runtime.public_safety import public_safe_compact_text, validate_public_safe_value
+from ..runtime.public_safety import (
+    SECRET_LIKE_SURFACE_PATTERN,
+    public_safe_compact_text,
+    validate_public_safe_value,
+)
 from ..runtime.session_runtime import session_runtime_work_observation
 from .acceptance_observation import build_goal_acceptance_observation
 from ..work_items.work_lane import WorkLaneObservation
@@ -48,12 +51,6 @@
 
 _TERMINAL_GOAL_STATUSES = {"closed", "retired", "archived", "done", "complete"}
 
-# Provider token shapes the shared private-text rules do not cover. A run
-# history reference is free text, so a leaked token there must never reach a
-# public projection just because the shared corpus did not list its prefix.
-_TOKEN_SHAPES = re.compile(
-    r"\b(?:gh[pousr]_[A-Za-z0-9]{16,}|sk-[A-Za-z0-9_-]{16,}|AKIA[0-9A-Z]{16})\b"
-)
 
 def _compact_text(value: Any, *, limit: int = 240) -> str | None:
     """Validate the complete source before bounding any public label/ref."""
@@ -65,7 +62,7 @@ def _compact_text(value: Any, *, limit: int = 240) -> str | None:
         return None
     # Preserve the stricter existing private-text/provider-token contract too;
     # these checks supplement, never replace, the shared public-safety owner.
-    if find_private_text_match(value) or _TOKEN_SHAPES.search(value):
+    if find_private_text_match(value) or SECRET_LIKE_SURFACE_PATTERN.search(value):
         return None
     return public_safe_compact_text(value, limit=limit)
 

loopx/control_plane/runtime/public_safety.py

@@ -19,11 +19,21 @@
 )
 SECRET_LIKE_SURFACE_PATTERN = re.compile(
     r"(?i)(?:\bbearer\s+[a-z0-9._~+/=-]{16,}|"
-    r"\b(?:access|secret)[_-]?key[\"']?\s*[=:]\s*[\"']?[^\s`'\"<>]+|"
+    r"\b(?:access|api|secret)[_-]?key[\"']?\s*[=:]\s*[\"']?[^\s`'\"<>]+|"
     r"\b(?:ak|sk)[\"']?\s*[=:]\s*[\"']?[^\s`'\"<>]+|"
     r"(?<![a-z0-9_])(?:ak|sk)[-_=:][a-z0-9_=-]{10,}|"
-    r"\bgh[pousr]_[a-z0-9]{20,}\b|"
+    r"\bgh[pousr]_[a-z0-9]{16,}\b|"
+    r"\bgithub_pat_[a-z0-9_]{20,}|"
+    r"\b(?:akia|asia)[a-z0-9]{16}\b|"
+    r"\bxox[baprs]-[a-z0-9-]{10,}|"
+    r"\baiza[a-z0-9_-]{20,}|"
+    r"\b(?:sk|rk)_(?:live|test)_[a-z0-9]{12,}|"
+    r"\bnpm_[a-z0-9]{20,}|"
+    r"\bpypi-[a-z0-9_-]{20,}|"
     r"\beyj[a-z0-9_-]{10,}\.[a-z0-9_-]{10,}\.[a-z0-9_-]{10,}\b|"
+    r"\b(?:access|refresh)[_-]?token[\"']?\s*[=:]\s*[\"']?[^\s`'\"<>]{12,}|"
+    r"\b(?:password|secret)[\"']?\s*[=:]\s*[\"']?[^\s`'\"<>]{12,}|"
+    r"-{3,}\s*BEGIN (?:[A-Z]+ )?PRIVATE KEY|"
     r"\btoken[\"']?\s*[=:]\s*[\"']?[^\s`'\"<>]{12,})"
 )
 _CREDENTIAL_FIELD_FAMILIES = frozenset(

loopx/extensions/presentation.py

@@ -11,6 +11,8 @@
 import re
 import tempfile
 from typing import Any
+
+from ..control_plane.runtime.public_safety import SECRET_LIKE_SURFACE_PATTERN
 from urllib.parse import parse_qsl, urlparse
 
 from ..file_lock import exclusive_file_lock
@@ -52,6 +54,8 @@
     r"(?:^|[\s:=('/\\])\.(?:codex|git|local)(?:[/\\]|$)",
     re.IGNORECASE,
 )
+# 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(
     r"(?:bearer\s+[A-Za-z0-9._~+/=-]{8,}|"
     r"(?:api[_ -]?key|access[_ -]?token|secret|password)\s*[:=]\s*\S+)",
@@ -153,7 +157,7 @@ def _plain_text(
         raise ValueError(f"{context} must be plain text without markup")
     if _LOCAL_PATH_RE.search(text) or _PRIVATE_RELATIVE_PATH_RE.search(text):
         raise ValueError(f"{context} must not contain a local path")
-    if _CREDENTIAL_RE.search(text):
+    if SECRET_LIKE_SURFACE_PATTERN.search(text) or _CREDENTIAL_RE.search(text):
         raise ValueError(f"{context} must not contain credential material")
     if _SENSITIVE_TEXT_RE.search(text):
         raise ValueError(f"{context} must not contain sensitive material")

tests/control_plane/test_public_safety_credential_shape_owner.py

@@ -0,0 +1,157 @@
+"""One owner decides what a credential-shaped value looks like.
+
+Before this change five surfaces each compiled their own credential test, and no
+two agreed: `control_plane.runtime.public_safety` recognized a GitHub token or a
+JWT while `decision_context`, `material_lifecycle`, `periodic_report` and
+`extensions.presentation` did not, and two of those four carried the same
+alternation list copied byte for byte. A value that one boundary treated as a
+secret therefore reached a published surface through another boundary that never
+heard of that shape.
+
+`SECRET_LIKE_SURFACE_PATTERN` is now the single shape owner, widened to the union
+of the shapes every site already guarded. Each site keeps only its own threshold
+policy (how short a value still counts), which is a per-surface judgement, and
+consults the owner for the shapes. These tests pin the wiring, not just the
+pattern, so a site that quietly stops consulting the owner goes red.
+"""
+
+import pathlib
+from collections.abc import 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.capabilities.periodic_report.core import (
+    _reject_raw_keys as reject_report_keys,
+)
+from loopx.control_plane.goals.artifact_lifecycle import (
+    _compact_text as compact_goal_reference,
+)
+from loopx.control_plane.runtime.public_safety import (
+    SECRET_LIKE_SURFACE_PATTERN,
+    validate_public_safe_value,
+)
+from loopx.extensions.presentation import _plain_text as presentation_text
+
+REPOSITORY_ROOT = pathlib.Path(__file__).resolve().parents[2]
+OWNER_MODULE = "loopx/control_plane/runtime/public_safety.py"
+# One consumer decides a different question: it rejects a goal id whose *whole*
+# value is a provider token, so its list is anchored and cannot be reused as a
+# surface scan. Named here so a new surface copy still fails this test.
+ANCHORED_WHOLE_VALUE_ID_RULES = frozenset(
+    {"loopx/extensions/openviking_semantic_preference/history_export.py"}
+)
+
+CREDENTIAL_SHAPES = {
+    "github token": "ghp_" + "a" * 36,
+    "jwt": "eyJ" + "h" * 12 + "." + "a" * 12 + "." + "s" * 12,
+    "password assignment": "password=hunter2hunter2",
+    "api key colon": "api_key: qwertyuiopasdfghjkl",
+    "secret assignment": "secret=abcdefghijklmn",
+    "token assignment": "token=abcdefghijklmn",
+    "access token assignment": "access_token=abcdefghijklmn",
+    "bearer value": "Bearer " + "z" * 20,
+    "signed key pair": "sk-" + "a" * 30,
+    "private key material": "-----BEGIN RSA PRIVATE KEY-----",
+    "openssh key material": "---- BEGIN OPENSSH PRIVATE KEY ----",
+    "fine-grained pat": "github_pat_" + "a1B2" * 8,
+    "aws access id": "AKIA" + "A1B2C3D4E5F6G7H8",
+    "slack token": "xoxb-" + "a1B2c3D4e5F6g7H8",
+    "google api key": "AIza" + "a1B2c3D4e5F6g7H8i9J0K1L2",
+    "stripe live key": "sk_live_" + "a1B2c3D4e5F6g7H8",
+    "npm token": "npm_" + "a1B2c3D4e5F6g7H8i9J0K1L2",
+}
+
+TEXT_SITES: list[tuple[str, Callable[[str], object], str]] = [
+    (
+        "decision_context",
+        lambda text: decision_text(text, field="summary"),
+        "contains a credential-like value",
+    ),
+    (
+        "material_lifecycle",
+        lambda text: material_text(text, field="summary"),
+        "contains a credential-like value",
+    ),
+    (
+        "extensions.presentation",
+        lambda text: presentation_text(text, context="label"),
+        "credential material",
+    ),
+]
+
+
+@pytest.mark.parametrize("shape", sorted(CREDENTIAL_SHAPES))
+def test_the_owner_recognizes_every_shape_a_site_used_to_guard(shape: str) -> None:
+    assert SECRET_LIKE_SURFACE_PATTERN.search(CREDENTIAL_SHAPES[shape])
+
+
+@pytest.mark.parametrize(
+    "site,call,reason", TEXT_SITES, ids=[site for site, _, _ in TEXT_SITES]
+)
+@pytest.mark.parametrize("shape", sorted(CREDENTIAL_SHAPES))
+def test_every_text_site_rejects_a_shape_the_owner_recognizes(
+    site: str, call: Callable[[str], object], reason: str, shape: str
+) -> None:
+    with pytest.raises(ValueError, match=reason):
+        call(CREDENTIAL_SHAPES[shape])
+
+
+@pytest.mark.parametrize("shape", sorted(CREDENTIAL_SHAPES))
+def test_public_payload_values_cannot_carry_a_credential_shape(shape: str) -> None:
+    with pytest.raises(ValueError, match="credential-like value"):
+        validate_public_safe_value({"note": CREDENTIAL_SHAPES[shape]})
+
+
+@pytest.mark.parametrize("shape", sorted(CREDENTIAL_SHAPES))
+def test_periodic_report_rejects_a_shape_the_owner_recognizes(shape: str) -> None:
+    with pytest.raises(ValueError, match="credential-like value"):
+        reject_report_keys({"summary": CREDENTIAL_SHAPES[shape]}, "report")
+
+
+def test_benign_text_still_passes_the_owner() -> None:
+    for text in (
+        "weekly cadence digest rendered for goal_42",
+        "token budget left for this stage: 1200",
+        "the operator rotated the deploy credentials yesterday",
+    ):
+        assert not SECRET_LIKE_SURFACE_PATTERN.search(text), text
+
+
+def test_goal_artifact_projection_keeps_both_owners_and_the_positive_case() -> None:
+    # A GitHub token is invisible to the private-text corpus, so this row fails
+    # the moment this surface stops consulting the credential-shape owner.
+    assert compact_goal_reference("ghp_" + "a" * 36) is None
+    # The bare word is invisible to the shape owner, so this row fails the
+    # moment this surface stops consulting the private-text corpus.
+    assert compact_goal_reference("the Bearer token expired") is None
+    # Positive control: without this row the two assertions above could pass on
+    # a surface that refused every value.
+    assert (
+        compact_goal_reference("weekly digest for goal_42")
+        == "weekly digest for goal_42"
+    )
+
+
+@pytest.mark.parametrize(
+    "identity_marker",
+    [r"gh[pousr]_", r"\beyj"],
+)
+def test_an_identity_shape_is_declared_in_one_owner_only(identity_marker: str) -> None:
+    sources = [
+        path
+        for path in REPOSITORY_ROOT.glob("loopx/**/*.py")
+        if path.relative_to(REPOSITORY_ROOT).as_posix() != OWNER_MODULE
+        and path.relative_to(REPOSITORY_ROOT).as_posix()
+        not in ANCHORED_WHOLE_VALUE_ID_RULES
+        and identity_marker in path.read_text(encoding="utf-8")
+    ]
+
+    assert [path.relative_to(REPOSITORY_ROOT).as_posix() for path in sources] == []
+    # Guard the guard: the owner must still declare the marker literally, or this
+    # test would pass even after the owner lost the shape entirely.
+    owner_source = (REPOSITORY_ROOT / OWNER_MODULE).read_text(encoding="utf-8")
+    assert identity_marker in owner_source