diff --git a/loopx/capabilities/decision_context/packets.py b/loopx/capabilities/decision_context/packets.py index 73154c06a6..d610185b29 100644 --- a/loopx/capabilities/decision_context/packets.py +++ b/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 diff --git a/loopx/capabilities/material_lifecycle/_validation.py b/loopx/capabilities/material_lifecycle/_validation.py index c3ba25e796..6418e04b1b 100644 --- a/loopx/capabilities/material_lifecycle/_validation.py +++ b/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 diff --git a/loopx/capabilities/periodic_report/core.py b/loopx/capabilities/periodic_report/core.py index 05a71cecfd..bcf1a17d04 100644 --- a/loopx/capabilities/periodic_report/core.py +++ b/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"(?]+" ) -_SECRET_LIKE_SURFACE_PATTERN = re.compile( - r"(?i)(?:\bbearer\s+[a-z0-9._~+/=-]{16,}|" - r"(?]{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") diff --git a/loopx/control_plane/goals/artifact_lifecycle.py b/loopx/control_plane/goals/artifact_lifecycle.py index 6b731c3445..8bcd783c83 100644 --- a/loopx/control_plane/goals/artifact_lifecycle.py +++ b/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) diff --git a/loopx/control_plane/runtime/public_safety.py b/loopx/control_plane/runtime/public_safety.py index 5697d8e5a0..7f4f5abd4c 100644 --- a/loopx/control_plane/runtime/public_safety.py +++ b/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"(?]{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( diff --git a/loopx/extensions/presentation.py b/loopx/extensions/presentation.py index 213030b5a5..4d8bb701a3 100644 --- a/loopx/extensions/presentation.py +++ b/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") diff --git a/tests/control_plane/test_public_safety_credential_shape_owner.py b/tests/control_plane/test_public_safety_credential_shape_owner.py new file mode 100644 index 0000000000..44524a1805 --- /dev/null +++ b/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