diff --git a/loopx/capabilities/decision_context/packets.py b/loopx/capabilities/decision_context/packets.py index b13eef77b..1bec8af9a 100644 --- a/loopx/capabilities/decision_context/packets.py +++ b/loopx/capabilities/decision_context/packets.py @@ -13,6 +13,7 @@ from ...control_plane.runtime.public_safety import ( REMOTE_LOCATION_SURFACE_PATTERN, SECRET_LIKE_SURFACE_PATTERN, + find_public_safe_local_path, ) DECISION_EVIDENCE_PACKET_SCHEMA_VERSION = "decision_evidence_packet_v0" @@ -40,7 +41,10 @@ } _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/|~/)") +# 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( @@ -77,7 +81,7 @@ def _compact_text(value: Any, *, field: str, max_len: int = 320) -> str: raise ValueError(f"{field} must be non-empty") if len(text) > max_len: raise ValueError(f"{field} must be at most {max_len} characters") - if _LOCAL_PATH_RE.search(text): + if find_public_safe_local_path(text) is not None: 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") diff --git a/loopx/capabilities/material_lifecycle/_validation.py b/loopx/capabilities/material_lifecycle/_validation.py index c8ec6d58c..544c7b634 100644 --- a/loopx/capabilities/material_lifecycle/_validation.py +++ b/loopx/capabilities/material_lifecycle/_validation.py @@ -12,10 +12,14 @@ from ...control_plane.runtime.public_safety import ( REMOTE_LOCATION_SURFACE_PATTERN, SECRET_LIKE_SURFACE_PATTERN, + find_public_safe_local_path, ) _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/|~/)") +# 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( @@ -53,7 +57,7 @@ def compact_text(value: Any, *, field: str, max_len: int = 320) -> str: raise ValueError(f"{field} must be non-empty") if len(text) > max_len: raise ValueError(f"{field} must be at most {max_len} characters") - if _LOCAL_PATH_RE.search(text): + if find_public_safe_local_path(text) is not None: 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") diff --git a/loopx/control_plane/runtime/public_safety.py b/loopx/control_plane/runtime/public_safety.py index 211057667..c229e1011 100644 --- a/loopx/control_plane/runtime/public_safety.py +++ b/loopx/control_plane/runtime/public_safety.py @@ -12,8 +12,10 @@ # module are unchanged. from ...public_safe_text import ( LOCAL_PATH_SURFACE_PATTERN as LOCAL_PATH_SURFACE_PATTERN, + PUBLIC_SAFE_LOCAL_PATH_PATTERNS as PUBLIC_SAFE_LOCAL_PATH_PATTERNS, REMOTE_LOCATION_SURFACE_PATTERN as REMOTE_LOCATION_SURFACE_PATTERN, SECRET_LIKE_SURFACE_PATTERN as SECRET_LIKE_SURFACE_PATTERN, + find_public_safe_local_path as find_public_safe_local_path, ) NormalizeText = Callable[..., str] diff --git a/loopx/public_safe_text.py b/loopx/public_safe_text.py index 47f3836f5..db3fbbf99 100644 --- a/loopx/public_safe_text.py +++ b/loopx/public_safe_text.py @@ -90,16 +90,16 @@ 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. +# Refs #5136, direction 3: the local-path shapes this owner recognizes. +# `LOCAL_PATH_SURFACE_PATTERN` carries the absolute roots, Windows drive letters +# and UNC shares; the gap pair adds the home-relative `~/...` form and a local +# path behind an explicit `path:` prefix. Whether a surface *rejects* what this +# owner recognizes stays the caller's named policy, so a surface can still opt +# into the narrower legacy set by asking for `LOCAL_PATH_SURFACE_PATTERN` alone. +# `file://` is deliberately not in this set: direction 3 does classify it as a +# local path, but every surface that has to stop carrying one already rejects it +# here as a raw remote location, and the surfaces that keep ordinary URLs would +# need a per-surface decision rather than a shared-pattern change. HOME_RELATIVE_PATH_PATTERN = re.compile(r"(?]+") PATH_PREFIX_LOCAL_PATTERN = re.compile( r"(?]+", re.IGNORECASE @@ -108,6 +108,21 @@ HOME_RELATIVE_PATH_PATTERN, PATH_PREFIX_LOCAL_PATTERN, ) +# The boundary form one pair of migrating surfaces already enforced: a local +# reference introduced by `:` or `=`. `LOCAL_PATH_SURFACE_PATTERN`'s lookbehind +# deliberately skips a preceding colon, so without this arm a surface moving +# onto the shared decision would start accepting `:/Users/...` -- a loosening +# no migration is allowed to introduce. +LOCAL_PATH_BOUNDARY_REFERENCE_PATTERN = re.compile( + r"(?:^|[\s:=])(?:/Users/|/private/|/tmp/|~[/\\])", + re.IGNORECASE, +) +PUBLIC_SAFE_LOCAL_PATH_PATTERNS: tuple[re.Pattern[str], ...] = ( + LOCAL_PATH_SURFACE_PATTERN, + HOME_RELATIVE_PATH_PATTERN, + PATH_PREFIX_LOCAL_PATTERN, + LOCAL_PATH_BOUNDARY_REFERENCE_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. @@ -262,6 +277,27 @@ def find_private_text_match(value: str | None) -> re.Pattern[str] | None: ) +def find_public_safe_local_path(value: str | None) -> re.Pattern[str] | None: + """Return the local-path shape ``value`` carries, or None when it carries none. + + This is the single answer to "is there a local path in this text" for + surfaces that publish outside the runtime (Refs #5136, direction 3): the + absolute roots, the two gap shapes `classify_private_text` reaches only when + a caller opts into `include_path_gaps`, and the colon/equals boundary form + the migrated surfaces already enforced. Recognition is still not permission: + a caller that must keep a narrower historical verdict asks for + `LOCAL_PATH_SURFACE_PATTERN` directly, and each surface keeps its own + rejection message and length limit. + """ + + if not value: + return None + for pattern in PUBLIC_SAFE_LOCAL_PATH_PATTERNS: + if pattern.search(value): + return pattern + return None + + def classify_private_text( value: str | None, *, diff --git a/tests/control_plane/test_public_safe_local_path_owner.py b/tests/control_plane/test_public_safe_local_path_owner.py new file mode 100644 index 000000000..3d932d32f --- /dev/null +++ b/tests/control_plane/test_public_safe_local_path_owner.py @@ -0,0 +1,322 @@ +"""Refs #5136 direction 3: one owner decides whether text carries a local path. + +`loopx/capabilities/decision_context/packets.py` and +`loopx/capabilities/material_lifecycle/_validation.py` each carried the same +private copy, `(^|[\\s:=])(?:/Users/|/private/|/tmp/|~/)`, next to credential and +remote-location rules they already imported from the shared owner. These tests +pin the migration and its boundaries: + +* both sites answer through `loopx/public_safe_text.find_public_safe_local_path` + and no longer declare a local-path regex of their own; +* the union the owner decides is `LOCAL_PATH_SURFACE_PATTERN`, the two + direction-3 gap shapes and the historical `:`/`=` boundary form, in that + order, so dropping any arm is caught; +* nothing the sites rejected before is accepted now (no loosening), and the + widening is exactly the direction-3 shapes -- both halves are measured against + the shared corpus, so "no silent loosening" is enforced rather than claimed; +* each site keeps its own rejection message and length limits; +* `file://` is unchanged here (both sites still reject it through the + raw-remote-location rule), and the general runtime export gate is *not* + tightened by this PR, which `test_runtime_export_gate_is_unchanged` pins. +""" + +from __future__ import annotations + +import ast +import json +import re +from pathlib import Path + +import pytest + +from loopx import public_safe_text as owner +from loopx.capabilities.decision_context import build_decision_evidence_packet +from loopx.capabilities.decision_context import packets as decision_packets +from loopx.capabilities.material_lifecycle import _validation as material_validation +from loopx.capabilities.material_lifecycle.intake import ( + build_material_candidate_intake_proposal, +) +from loopx.control_plane.runtime import public_safety + +REPOSITORY_ROOT = Path(__file__).resolve().parents[2] +CORPUS_PATH = REPOSITORY_ROOT / "tests" / "fixtures" / "public_safe_text_corpus.json" +_PLACEHOLDER_RE = re.compile(r"\{([A-Z][A-Z_]*)\}") + +# The rule these two sites shipped before this PR, kept here only so the +# widening and the preserved half are both machine-checked. +LEGACY_SITE_COPY = re.compile(r"(^|[\s:=])(?:/Users/|/private/|/tmp/|~/)") + +OBSERVED_AT = "2026-07-25T13:30:00+00:00" + +LOCAL_PATH_MESSAGE = "must not contain a local path" + + +def _shapes_the_owner_recognizes() -> list[str]: + """Every home-relative, prefixed, drive, UNC and absolute shape direction 3 names.""" + return [ + "~/notes.md", + "~\\notes.md", + "path:/srv/data/goal.json", + "PATH:\\\\fileserver\\share", + "/Users/alex/notes.md", + "/home/alex/.codex/auth.json", + "/var/folders/9x/private/notes", + "/private/tmp/x/lease.log", + "/tmp/build/out.log", + "/etc/loopx/registry.json", + "/opt/secrets/token", + "/mnt/nas/private.log", + "/root/.ssh/id_rsa", + "/workspace/private/notes.md", + "/data/goal-evidence/dump", + "C:\\Users\\alex\\notes.md", + "D:/notes/private.md", + "\\\\fileserver\\share\\notes.md", + # The boundary form: introduced by a colon or equals sign, which the + # absolute-root pattern's lookbehind intentionally skips. + "lookup :/Users/alex/notes.md", + "dump=/private/tmp/x/lease.log", + ] + + +def _corpus_samples() -> list[str]: + corpus = json.loads(CORPUS_PATH.read_text(encoding="utf-8")) + + def render(template: str) -> str: + def substitute(match: re.Match[str]) -> str: + name = match.group(1) + if name.endswith("_LOWER"): + return "".join(corpus["tokens"][name[: -len("_LOWER")]]).lower() + return "".join(corpus["tokens"][name]) + + return _PLACEHOLDER_RE.sub(substitute, template) + + return [ + render(sample["template"]) + for kind in ("public_safe", "private_looking") + for sample in corpus[kind] + ] + + +def test_both_sites_answer_through_the_owner_object() -> None: + assert ( + public_safety.find_public_safe_local_path is owner.find_public_safe_local_path + ) + assert ( + decision_packets.find_public_safe_local_path + is owner.find_public_safe_local_path + ) + assert ( + material_validation.find_public_safe_local_path + is owner.find_public_safe_local_path + ) + + +def test_neither_site_declares_a_local_path_regex_any_more() -> None: + assert not hasattr(decision_packets, "_LOCAL_PATH_RE") + assert not hasattr(material_validation, "_LOCAL_PATH_RE") + for module in (decision_packets, material_validation): + tree = ast.parse(Path(module.__file__).read_text(encoding="utf-8")) + for node in ast.walk(tree): + if not ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and node.func.attr == "compile" + ): + continue + literal = " ".join( + part.value for part in node.args if isinstance(part, ast.Constant) + ) + assert "/Users/" not in literal and "~/)" not in literal, ( + f"{module.__name__} declares a second local-path rule: {literal!r}" + ) + + +def test_owner_recognizes_the_absolute_roots_the_gaps_and_the_boundary_form() -> None: + assert owner.PUBLIC_SAFE_LOCAL_PATH_PATTERNS == ( + owner.LOCAL_PATH_SURFACE_PATTERN, + owner.HOME_RELATIVE_PATH_PATTERN, + owner.PATH_PREFIX_LOCAL_PATTERN, + owner.LOCAL_PATH_BOUNDARY_REFERENCE_PATTERN, + ) + + +@pytest.mark.parametrize("value", _shapes_the_owner_recognizes()) +def test_owner_recognizes_every_local_path_shape(value: str) -> None: + assert owner.find_public_safe_local_path(value) is not None + + +@pytest.mark.parametrize("value", _corpus_samples()) +def test_owner_never_loosens_what_the_shared_corpus_already_established( + value: str, +) -> None: + # Anything the replaced private copy flagged must still be flagged. This is + # the enforced half of "no silent loosening" over the corpus both runtimes + # are pinned to. + if LEGACY_SITE_COPY.search(value) is not None: + assert owner.find_public_safe_local_path(value) is not None + + +def test_owner_covers_the_whole_legacy_equivalence_class() -> None: + # Enumerated rather than hand-picked: every boundary character x every root + # the copies knew or direction 3 names x a trailing segment. The claim this + # PR makes is one-directional -- the owner must reject everything the copies + # rejected -- so the whole product is checked, not a sample of it. + boundaries = ["", " ", ":", "=", "(", ",", '"', "'", "/", "x", "-", "@", "\t", "\n"] + roots = [ + "/Users/", + "/private/", + "/tmp/", + "~/", + "~\\", + "/home/", + "/var/folders/", + "/etc/", + "/opt/", + "/srv/", + "/mnt/", + "/root/", + "/data/", + "/workspace/", + "/workspaces/", + "/Volumes/", + "C:\\", + "D:/", + "\\\\fileserver\\share\\", + "path:/", + "PATH:\\", + "//Users/", + ] + loosened: list[str] = [] + widened: set[str] = set() + for boundary in boundaries: + for root in roots: + for value in ( + f"note {boundary}{root}segment.md", + f"{boundary}{root}segment.md", + ): + legacy = LEGACY_SITE_COPY.search(value) is not None + current = owner.find_public_safe_local_path(value) is not None + if legacy and not current: + loosened.append(repr(value)) + elif current and not legacy: + widened.add(root) + assert loosened == [] + # The widening half: roots the copies simply did not know. + assert "/home/" in widened + assert "path:/" in widened + assert "\\\\fileserver\\share\\" in widened + assert "/var/folders/" in widened + + +def test_the_widening_is_the_direction_three_shapes_not_the_corpus() -> None: + newly_recognized = [ + value + for value in _shapes_the_owner_recognizes() + if LEGACY_SITE_COPY.search(value) is None + ] + # Home-relative Windows form, `path:` prefix, drive letter and UNC share were + # all accepted by the private copies; ordinary absolute roots are unchanged. + assert newly_recognized == [ + "~\\notes.md", + "path:/srv/data/goal.json", + "PATH:\\\\fileserver\\share", + "/home/alex/.codex/auth.json", + "/var/folders/9x/private/notes", + "/etc/loopx/registry.json", + "/opt/secrets/token", + "/mnt/nas/private.log", + "/root/.ssh/id_rsa", + "/workspace/private/notes.md", + "/data/goal-evidence/dump", + "C:\\Users\\alex\\notes.md", + "D:/notes/private.md", + "\\\\fileserver\\share\\notes.md", + ] + assert not [ + value + for value in _corpus_samples() + if LEGACY_SITE_COPY.search(value) is None + and owner.find_public_safe_local_path(value) is not None + ] + + +def test_decision_context_rejects_a_nested_local_path_with_its_own_message() -> None: + # `path:`-prefixed was outside the private copy's alternation, so this value + # only reaches the site through the owner. + with pytest.raises(ValueError, match=LOCAL_PATH_MESSAGE) as matched: + build_decision_evidence_packet( + goal_id="goal:decision-advisor", + decision_id="decision:20260725:priority", + observed_at=OBSERVED_AT, + changed_facts=[ + { + "fact_id": "fact:adoption-stage", + "summary": "deploy notes at path:/srv/data/goal.json", + "source_ref": "authority:collaboration-ledger", + "source_revision": "revision:42", + "observed_at": OBSERVED_AT, + "freshness": "current", + "authority": "first_party_receipt", + } + ], + ) + assert "changed_facts[0].summary" in str(matched.value) + + +def test_material_lifecycle_rejects_a_local_path_with_its_own_message() -> None: + # A home-relative path with a Windows separator: the private copy only knew + # `~/`, so rejecting this one proves the site reads the owner. + with pytest.raises(ValueError, match=LOCAL_PATH_MESSAGE) as matched: + build_material_candidate_intake_proposal( + goal_id="goal:material-intake", + proposal_id="proposal:20260725:1", + store_id="store:project", + source_authority_revision="revision:7", + material_ref="~\\notes.md", + source_ref="authority:exact-read", + source_revision="revision:8", + exact_read_ref="exact:read-1", + content_digest="sha256:0" * 8, + content_size_bytes=128, + observed_at=OBSERVED_AT, + ) + assert "material_ref" in str(matched.value) + + +def test_both_sites_keep_their_length_limits_and_clean_verdicts() -> None: + assert decision_packets._compact_text("public alias v3", field="summary") == ( + "public alias v3" + ) + assert material_validation.compact_text("public alias v3", field="summary") == ( + "public alias v3" + ) + with pytest.raises(ValueError, match="must be at most 320 characters"): + decision_packets._compact_text("x" * 321, field="summary") + with pytest.raises(ValueError, match="must be at most 320 characters"): + material_validation.compact_text("x" * 321, field="summary") + + +@pytest.mark.parametrize( + "value", + ["file:///Users/alex/notes.md", "file://./notes.md"], +) +def test_file_url_verdicts_are_unchanged_and_still_named_as_raw_urls( + value: str, +) -> None: + # The direction-3 `file://` decision is deferred, so these must keep failing + # through the raw-remote-location rule with the site's own wording. + with pytest.raises(ValueError, match="raw URL"): + material_validation.compact_text(value, field="source_ref") + with pytest.raises(ValueError, match="raw URL"): + decision_packets._compact_text(value, field="source_ref") + + +def test_runtime_export_gate_is_unchanged() -> None: + # Disclosed non-change, pinned so the later tightening has to be a deliberate + # edit here: `validate_public_safe_value` still accepts a home-relative + # reference. Direction 3 decides that per destination, and how many of the + # gate's call sites would newly reject was not measured in this PR. + public_safety.validate_public_safe_value( + {"target_layout": "~/.agents/skills", "note": "path: is fine without a path"} + )