Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 12 additions & 5 deletions loopx/control_plane/goals/artifact_lifecycle.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,11 @@

from typing import Any

from ...public_safe_text import find_private_text_match
from ...public_safe_text import (
ARTIFACT_LIFECYCLE_CATEGORIES,
classify_private_text,
)
from ..runtime.public_safety import (
SECRET_LIKE_SURFACE_PATTERN,
public_safe_compact_text,
validate_public_safe_value,
)
Expand Down Expand Up @@ -60,9 +62,14 @@ def _compact_text(value: Any, *, limit: int = 240) -> str | None:
validate_public_safe_value(value)
except ValueError:
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 SECRET_LIKE_SURFACE_PATTERN.search(value):
# One policy-aware call into the shared classifier replaces OR-ing the
# text-owner detector with the credential shape detector. The named policy
# (ARTIFACT_LIFECYCLE_CATEGORIES) preserves this projection's historical
# verdict exactly: every category but a raw remote location.
if (
classify_private_text(value, categories=ARTIFACT_LIFECYCLE_CATEGORIES)
is not None
):
return None
return public_safe_compact_text(value, limit=limit)

Expand Down
43 changes: 11 additions & 32 deletions loopx/control_plane/runtime/public_safety.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,42 +4,21 @@
from collections.abc import Mapping
from typing import Any, Callable, Optional

# Refs #5136: the text-shape definitions live in one owner now
# (loopx/public_safe_text.py). This module consumes them for recursive payload
# validation and public-output policy instead of restating a competing set. The
# redundant-alias form makes each an explicit re-export (house style under
# --no-implicit-reexport), so the existing importers of these names from this
# module are unchanged.
from ...public_safe_text import (
LOCAL_PATH_SURFACE_PATTERN as LOCAL_PATH_SURFACE_PATTERN,
REMOTE_LOCATION_SURFACE_PATTERN as REMOTE_LOCATION_SURFACE_PATTERN,
SECRET_LIKE_SURFACE_PATTERN as SECRET_LIKE_SURFACE_PATTERN,
)

NormalizeText = Callable[..., str]
CompactText = Callable[..., Optional[str]]
DEFAULT_PUBLIC_SAFE_LIST_LIMIT = 4
LOCAL_PATH_SURFACE_PATTERN = re.compile(
r"(?<![:/A-Za-z0-9])(?:"
r"/(?:Users|home|Volumes|private|tmp|var|etc|opt|srv|mnt|root|data|workspace|workspaces)/"
r"[^\s`'\"<>]+|"
r"[A-Za-z]:[\\/][^\s`'\"<>]+|"
r"\\\\[A-Za-z0-9_.-]+\\[^\s`'\"<>]+"
r")",
re.IGNORECASE,
)
# Refs #5136: one definition for "this string carries a raw remote location".
# Three validators each restated the same scheme list, and the canonical
# public-safety owner had no counterpart, so a fourth caller had to invent one.
REMOTE_LOCATION_SURFACE_PATTERN = re.compile(r"(?i)\b(?:https?|file|s3|gs|tos|hdfs)://")
SECRET_LIKE_SURFACE_PATTERN = re.compile(
r"(?i)(?:\bbearer\s+[a-z0-9._~+/=-]{16,}|"
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]{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(
{
"accesskey",
Expand Down
279 changes: 265 additions & 14 deletions loopx/public_safe_text.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,17 @@
"""Canonical private-looking-text rules for public-safe control-plane fields.

Four real validator owners enforce the same contract: `feedback`,
This module is the single owner of "does this string look private?" for the
control plane. It answers two questions that used to be conflated:

* **Detection** -- recognize a credential shape, a local path, a raw remote
location, or an internal organizational marker, and return an explicit
*category* plus a *reason* so a caller can decide what to do.
* **Permission** -- a named *policy* selects which categories a given surface
rejects. Detection recognizing a value never implies every surface must
reject it: owner-private operational state and repository/PR publication have
different disclosure boundaries (Refs #5136).

Four real validator owners enforce the same text contract: `feedback`,
`authority`, `boundary_authority`, and the TypeScript Vision checkpoint. The
rule set used to be copied into each owner, and the copies drifted: one still
rejected the ordinary English word "authorization" while another accepted a
Expand All @@ -9,13 +20,39 @@
`tests/fixtures/public_safe_text_corpus.json` pins both runtimes to one
contract.

`control_plane/runtime/public_safety.py` also consumes the shape definitions
here (`SECRET_LIKE_SURFACE_PATTERN`, `LOCAL_PATH_SURFACE_PATTERN`,
`REMOTE_LOCATION_SURFACE_PATTERN`) instead of owning a competing set, so a
caller such as `artifact_lifecycle` can make one policy-aware call rather than
OR-ing independent detectors.

Each owner keeps its own error message and guidance, because those describe
the owning surface, not the shared rule.
"""

from __future__ import annotations

import re
from dataclasses import dataclass

# ---------------------------------------------------------------------------
# Explicit categories. A caller names a policy (a set of categories) rather
# than reaching for a bare regex, so "recognized" and "rejected here" stay
# separate decisions (Refs #5136, direction 2).
# ---------------------------------------------------------------------------
CATEGORY_CREDENTIAL = "credential"
CATEGORY_LOCAL_PATH = "local_path"
CATEGORY_REMOTE_LOCATION = "remote_location"
CATEGORY_ORG_MARKER = "org_marker"

ALL_CATEGORIES: frozenset[str] = frozenset(
{
CATEGORY_CREDENTIAL,
CATEGORY_LOCAL_PATH,
CATEGORY_REMOTE_LOCATION,
CATEGORY_ORG_MARKER,
}
)


# Credential shape, not the plain English word. LoopX governance prose says
Expand All @@ -38,27 +75,241 @@
r"[A-Za-z0-9+/=]{16,}",
)

PRIVATE_TEXT_PATTERNS: tuple[re.Pattern[str], ...] = (
re.compile(r"/" + r"Users/"),
re.compile(r"/" + r"ext_data/"),
re.compile("la" + "rk" + "office", re.I),
re.compile("docs" + r"\." + "internal", re.I),
re.compile(r"\bt-20\d{12}-[a-z0-9]+\b"),
re.compile(r"\b" + "Bear" + r"er\b", re.I),
_AUTHORIZATION_CREDENTIAL_SHAPE,
_BASIC_CREDENTIAL_VALUE,
re.compile(r"\b" + "tok" + r"en\s*=", re.I),
re.compile(r"\b" + "pass" + r"word\b", re.I),
re.compile(r"\b" + "sec" + r"ret\b", re.I),
# 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
# pattern is byte-identical to the one public_safety enforced before the move:
# slice A is a behavior-preserving consolidation, so no existing consumer's
# verdict changes.
LOCAL_PATH_SURFACE_PATTERN = re.compile(
r"(?<![:/A-Za-z0-9])(?:"
r"/(?:Users|home|Volumes|private|tmp|var|etc|opt|srv|mnt|root|data|workspace|workspaces)/"
r"[^\s`'\"<>]+|"
r"[A-Za-z]:[\\/][^\s`'\"<>]+|"
r"\\\\[A-Za-z0-9_.-]+\\[^\s`'\"<>]+"
r")",
re.IGNORECASE,
)
# Refs #5136, direction 3: the shared classifier can *recognize* the local-path
# shapes the legacy surface pattern misses -- a home-relative `~/...` path and a
# local path behind an explicit `path:` prefix. Recognition is opt-in
# (`include_path_gaps`) so this consolidation does not silently tighten the 30+
# consumers of LOCAL_PATH_SURFACE_PATTERN; wiring these into a surface's
# enforcement policy is the disclosed behavior change tracked separately.
# `file://` is not added to the gap set here: direction 3 does classify it as a
# local path, but acting on that means a public projection stops carrying a
# location it accepts today, which is a disclosed tightening rather than part of
# this relocation. It is applied with the enforcement policy in the follow-up.
HOME_RELATIVE_PATH_PATTERN = re.compile(r"(?<![\w~])~[\\/][^\s`'\"<>]+")
PATH_PREFIX_LOCAL_PATTERN = re.compile(
r"(?<![\w:])path:[\\/][^\s`'\"<>]+", re.IGNORECASE
)
LOCAL_PATH_GAP_PATTERNS: tuple[re.Pattern[str], ...] = (
HOME_RELATIVE_PATH_PATTERN,
PATH_PREFIX_LOCAL_PATTERN,
)
# Refs #5136: one definition for "this string carries a raw remote location".
# Three validators each restated the same scheme list, and the canonical
# public-safety owner had no counterpart, so a fourth caller had to invent one.
REMOTE_LOCATION_SURFACE_PATTERN = re.compile(r"(?i)\b(?:https?|file|s3|gs|tos|hdfs)://")
SECRET_LIKE_SURFACE_PATTERN = re.compile(
r"(?i)(?:\bbearer\s+[a-z0-9._~+/=-]{16,}|"
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]{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,})"
)


# The text-owner rule set (feedback / authority / boundary_authority / the
# TypeScript Vision checkpoint). Each entry carries an explicit category and a
# stable reason so `classify_private_text` can hand a caller a named verdict
# instead of a bare regex object. Order is significant: `find_private_text_match`
# returns the first match, and the shared corpus pins that first-match contract.
@dataclass(frozen=True)
class _CategorizedPattern:
pattern: re.Pattern[str]
category: str
reason: str


_CATEGORIZED_PRIVATE_TEXT_PATTERNS: tuple[_CategorizedPattern, ...] = (
_CategorizedPattern(
re.compile(r"/" + r"Users/"), CATEGORY_LOCAL_PATH, "absolute home-directory path"
),
_CategorizedPattern(
re.compile(r"/" + r"ext_data/"), CATEGORY_ORG_MARKER, "internal ext_data path"
),
_CategorizedPattern(
re.compile("la" + "rk" + "office", re.I),
CATEGORY_ORG_MARKER,
"internal Lark/Feishu office marker",
),
_CategorizedPattern(
re.compile("docs" + r"\." + "internal", re.I),
CATEGORY_ORG_MARKER,
"internal docs host marker",
),
_CategorizedPattern(
re.compile(r"\bt-20\d{12}-[a-z0-9]+\b"),
CATEGORY_ORG_MARKER,
"internal ticket identifier",
),
_CategorizedPattern(
re.compile(r"\b" + "Bear" + r"er\b", re.I),
CATEGORY_CREDENTIAL,
"bearer auth scheme word",
),
_CategorizedPattern(
_AUTHORIZATION_CREDENTIAL_SHAPE,
CATEGORY_CREDENTIAL,
"authorization header/assignment shape",
),
_CategorizedPattern(
_BASIC_CREDENTIAL_VALUE, CATEGORY_CREDENTIAL, "basic-auth credential value"
),
_CategorizedPattern(
re.compile(r"\b" + "tok" + r"en\s*=", re.I),
CATEGORY_CREDENTIAL,
"token assignment shape",
),
_CategorizedPattern(
re.compile(r"\b" + "pass" + r"word\b", re.I),
CATEGORY_CREDENTIAL,
"password word",
),
_CategorizedPattern(
re.compile(r"\b" + "sec" + r"ret\b", re.I),
CATEGORY_CREDENTIAL,
"secret word",
),
)

# Kept as the plain pattern tuple so `find_private_text_match` and every
# existing importer see byte-identical behavior (same patterns, same order).
PRIVATE_TEXT_PATTERNS: tuple[re.Pattern[str], ...] = tuple(
entry.pattern for entry in _CATEGORIZED_PRIVATE_TEXT_PATTERNS
)


# ---------------------------------------------------------------------------
# Named policies. A policy is the set of categories a surface rejects. Keeping
# these named (rather than inline regex ORs at each caller) is what lets one
# detection owner serve surfaces with different disclosure boundaries.
# ---------------------------------------------------------------------------
# The four text owners reject every recognized category (their historical
# behavior): credential shapes, local paths, remote locations, org markers.
TEXT_OWNER_CATEGORIES: frozenset[str] = ALL_CATEGORIES
# artifact_lifecycle historically OR-ed find_private_text_match (the text-owner
# set) with SECRET_LIKE_SURFACE_PATTERN, downstream of a validate_public_safe_value
# call that already rejected the local-path and credential shapes. That union
# covers every category *except* a raw remote location: this projection has
# always let an ordinary http(s) URL through. The policy preserves that exactly
# rather than silently tightening it; widening it to remote_location is a
# separate, disclosed decision (Refs #5136, direction 2).
ARTIFACT_LIFECYCLE_CATEGORIES: frozenset[str] = frozenset(
{CATEGORY_CREDENTIAL, CATEGORY_LOCAL_PATH, CATEGORY_ORG_MARKER}
)


@dataclass(frozen=True)
class PrivateTextMatch:
"""An explicit, categorized private-text detection result."""

category: str
reason: str
pattern: re.Pattern[str]


def find_private_text_match(value: str | None) -> re.Pattern[str] | None:
"""Return the first matching private-text pattern, or None when clean."""
"""Return the first matching private-text pattern, or None when clean.

Preserved verbatim for the four text owners and the shared corpus parity
test; `classify_private_text` is the category-aware successor.
"""

if not value:
return None
for pattern in PRIVATE_TEXT_PATTERNS:
if pattern.search(value):
return pattern
return None


# The shape-based detectors, categorized. These supplement the text-owner
# patterns so a single call can cover both owners that artifact_lifecycle used
# to OR together.
_SHAPE_DETECTORS: tuple[tuple[re.Pattern[str], str, str], ...] = (
(SECRET_LIKE_SURFACE_PATTERN, CATEGORY_CREDENTIAL, "credential-like value shape"),
(LOCAL_PATH_SURFACE_PATTERN, CATEGORY_LOCAL_PATH, "local filesystem path"),
(
REMOTE_LOCATION_SURFACE_PATTERN,
CATEGORY_REMOTE_LOCATION,
"raw remote location URL",
),
)


def classify_private_text(
value: str | None,
*,
categories: frozenset[str] = ALL_CATEGORIES,
include_path_gaps: bool = False,
) -> PrivateTextMatch | None:
"""Return the first recognized private-text match within ``categories``.

Detection only: recognizing a value does not decide whether a given surface
may publish it. Callers pass the named policy (category set) for their
destination. The text-owner patterns are checked first, in their pinned
order, then the relocated shape detectors, so a value that both owners used
to flag still resolves to a single explicit category and reason.

``include_path_gaps`` opts a surface into the direction-3 recognition of
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.
"""

if not value:
return None
for entry in _CATEGORIZED_PRIVATE_TEXT_PATTERNS:
if entry.category in categories and entry.pattern.search(value):
return PrivateTextMatch(entry.category, entry.reason, entry.pattern)
for pattern, category, reason in _SHAPE_DETECTORS:
if category in categories and pattern.search(value):
return PrivateTextMatch(category, reason, pattern)
if include_path_gaps and CATEGORY_LOCAL_PATH in categories:
for pattern in LOCAL_PATH_GAP_PATTERNS:
if pattern.search(value):
return PrivateTextMatch(
CATEGORY_LOCAL_PATH, "local path behind a relative/prefixed form", pattern
)
return None


def matches_private_text_policy(
value: str | None,
*,
categories: frozenset[str] = ALL_CATEGORIES,
include_path_gaps: 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
)
is not None
)
Loading
Loading