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
11 changes: 6 additions & 5 deletions loopx/control_plane/runtime/public_safety.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,20 +10,21 @@
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|workspace|workspaces)/"
r"/(?:Users|home|Volumes|private|tmp|var|etc|opt|srv|mnt|root|data|workspace|workspaces)/"
r"[^\s`'\"<>]+|"
r"[A-Za-z]:[\\/](?:Users|Documents and Settings)[\\/][^\s`'\"<>]+"
r"[A-Za-z]:[\\/][^\s`'\"<>]+|"
r"\\\\[A-Za-z0-9_.-]+\\[^\s`'\"<>]+"
r")",
re.IGNORECASE,
)
SECRET_LIKE_SURFACE_PATTERN = re.compile(
r"(?i)(?:\bbearer\s+[a-z0-9._~+/=-]{16,}|"
r"\b(?:access|secret)[_-]?key\s*[=:]\s*[^\s`'\"<>]+|"
r"\b(?:ak|sk)\s*[=:]\s*[^\s`'\"<>]+|"
r"\b(?:access|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"\beyj[a-z0-9_-]{10,}\.[a-z0-9_-]{10,}\.[a-z0-9_-]{10,}\b|"
r"\btoken\s*[=:]\s*[^\s`'\"<>]{12,})"
r"\btoken[\"']?\s*[=:]\s*[\"']?[^\s`'\"<>]{12,})"
)
_CREDENTIAL_FIELD_FAMILIES = frozenset(
{
Expand Down
77 changes: 77 additions & 0 deletions tests/control_plane/test_public_safety_path_shapes.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,77 @@
"""Boundary shapes for the shared public-safety surface rules.

The local-path rule recognizes every drive-qualified path, UNC shares and
``/data`` roots; the secret rule tolerates quoted keys and values. Neither
widening may start matching URLs, clock times or ratios.
"""

import pytest

from loopx.control_plane.runtime.public_safety import (
LOCAL_PATH_SURFACE_PATTERN,
SECRET_LIKE_SURFACE_PATTERN,
validate_public_safe_value,
)

REJECTED_PATHS = [
"C:" + chr(92) + "build" + chr(92) + "evidence.txt",
"C:/workspace/private/worker.json",
"D:" + chr(92) + "Projects" + chr(92) + "loopx" + chr(92) + "state.json",
chr(92) * 2 + "server" + chr(92) + "share" + chr(92) + "evidence.txt",
"/data/reports/evidence.txt",
"C:" + chr(92) + "Users" + chr(92) + "fixture" + chr(92) + "evidence.txt",
]

ACCEPTED_TEXT = [
"https://example.org/data/report",
"s3://bucket/key",
"notion://page/1",
"12:30/45",
"ratio 3:4/5 done",
"id x:y/z",
"docs/evidence.md",
"access key rotation guide",
]

def _secret(key: str, separator: str, quote: str = "", key_quote: str = "") -> str:
"""Build a credential-shaped probe without a literal secret assignment."""

return f"{key_quote}{key}{key_quote}{separator}{quote}{'synthetic' * 4}{quote}"


REJECTED_SECRETS = [
_secret("token", ": ", quote='"'),
_secret("token", "="),
_secret("access_key", "=", quote="'"),
_secret("access_key", ": ", quote='"', key_quote='"'),
_secret("sk", " = ", quote="'"),
]


@pytest.mark.parametrize("value", REJECTED_PATHS)
def test_drive_unc_and_data_paths_are_local_paths(value):
assert LOCAL_PATH_SURFACE_PATTERN.search(value)
with pytest.raises(ValueError, match="absolute local path"):
validate_public_safe_value(value)


@pytest.mark.parametrize("value", ACCEPTED_TEXT)
def test_urls_times_and_ratios_are_not_local_paths(value):
assert not LOCAL_PATH_SURFACE_PATTERN.search(value)
assert not SECRET_LIKE_SURFACE_PATTERN.search(value)
validate_public_safe_value(value)


@pytest.mark.parametrize("value", REJECTED_SECRETS)
def test_quoted_secret_keys_and_values_are_credential_like(value):
assert SECRET_LIKE_SURFACE_PATTERN.search(value)
with pytest.raises(ValueError, match="credential-like"):
validate_public_safe_value(value)


def test_path_shaped_value_reports_local_path_before_opaque_shape():
# The turn-executor contract records this ordering: a drive-qualified
# path is a local path first, so that diagnostic wins over the
# opaque-reference shape check that runs afterwards.
with pytest.raises(ValueError, match="absolute local path"):
validate_public_safe_value({"worker_ref": "C:/workspace/private/worker.json"})
22 changes: 16 additions & 6 deletions tests/test_loopx_turn_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -735,15 +735,27 @@ def test_enabled_host_result_rejects_receipt_local_path() -> None:


@pytest.mark.parametrize(
("field", "value"),
("field", "value", "expected_error"),
[
("worker_ref", "C:/workspace/private/worker.json"),
("evidence_refs", ["file:/tmp/private-result.json"]),
# A drive-qualified path is now recognized as a local path, so the
# shared public-safety rule reports it before the opaque-shape check.
# Both rules reject the value; only the diagnostic differs.
(
"worker_ref",
"C:/workspace/private/worker.json",
"contains an absolute local path",
),
(
"evidence_refs",
["file:/tmp/private-result.json"],
"opaque 1-192 character public-safe reference",
),
],
)
def test_enabled_host_result_rejects_path_shaped_opaque_refs(
field: str,
value: object,
expected_error: str,
) -> None:
plan = _adaptive_observation_plan()
result = _host_result(plan)
Expand All @@ -754,9 +766,7 @@ def test_enabled_host_result_rejects_path_shaped_opaque_refs(
rejected = validate_loopx_turn_host_result(plan, result)

assert rejected["ok"] is False
assert "opaque 1-192 character public-safe reference" in " ".join(
rejected["errors"]
)
assert expected_error in " ".join(rejected["errors"])
assert "child_execution_receipts" not in rejected["result"]
rejected_value = value[0] if isinstance(value, list) else value
assert rejected_value not in json.dumps(
Expand Down