Merge pull request #4426 from songoow/codex/public-safety-drive-qualified-paths

fix(public-safety): recognize drive-qualified, UNC and /data paths and quoted secrets

Author: huangruiteng

loopx/control_plane/runtime/public_safety.py

@@ -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(
     {

tests/control_plane/test_public_safety_path_shapes.py

@@ -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"})

tests/test_loopx_turn_executor.py

@@ -756,15 +756,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)
@@ -775,9 +787,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(