Skip to content
Closed
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: 10 additions & 1 deletion loopx/capabilities/periodic_report/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,17 @@
"primary_goal_outcome",
"vision_closed",
}
# The private roots take whatever separator the host or a serialized string
# chose: matching the separator at every junction covers the native Windows
# spelling (C:\Users\<name>\...), the UNC spelling (\\server\Users\<name>\...)
# and the escaped spelling a JSON string produces (C:\\Users\\<name>\\...),
# the class the boundary scan (#6161) and the public-safe-outbound scanner
# (#6223) reject. The lead is a single separator so a long non-root separator
# run cannot make the raw-value scan quadratic. Paths outside these roots
# stay untouched.
_LOCAL_PATH_SURFACE_PATTERN = re.compile(
r"(?<!<)/(?:Users|Volumes|home|var/folders|tmp|private/tmp)/[^\s`'\"<>]+"
r"(?<!<)[\\/](?:Users|Volumes|home|var[\\/]+folders|tmp|private[\\/]+tmp)"
r"[\\/]+[^\s`'\"<>]+"
)
_FORBIDDEN_RAW_KEYS = {
"credential",
Expand Down
82 changes: 82 additions & 0 deletions tests/capabilities/test_periodic_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@
import json
import subprocess
import sys
import time
from pathlib import Path

import pytest
Expand Down Expand Up @@ -215,6 +216,61 @@ def test_raw_content_and_incomplete_sink_roles_are_rejected() -> None:
build_periodic_report_run(missing_delivery)


def _request_with_snapshot_ref(value: str) -> dict[str, object]:
request = _request()
request["source_snapshots"][0]["snapshot_ref"] = value # type: ignore[index]
return request


def test_private_path_spelling_variants_are_rejected() -> None:
# The same private path spelled the way a Windows host or a serialized
# string writes it is refused like the POSIX spelling above.
separator = "\\"
doubled = separator * 2
parts = ["Users", "example-user", "notes.md"]
denied = {
"native-drive": "C:" + separator + separator.join(parts),
"unc": doubled + separator.join(["fileserver", *parts]),
"rooted-native": separator + separator.join(parts),
"escaped-drive": "C:" + doubled + doubled.join(parts),
"escaped-unc": doubled * 2 + doubled.join(["fileserver", *parts]),
"native-tmp": "C:" + separator + separator.join(["tmp", "scratch.txt"]),
"native-private-tmp": "C:" + separator + separator.join(["private", "tmp", "scratch.log"]),
"native-var-folders": "C:" + separator + separator.join(["var", "folders", "app-cache"]),
}
for label, private_ref in denied.items():
try:
build_periodic_report_run(_request_with_snapshot_ref(private_ref))
except ValueError as exc:
assert "private path or credential-like value" in str(exc), label
else:
raise AssertionError(f"{label} accepted private path {private_ref!r}")


def test_public_windows_paths_and_urls_stay_accepted() -> None:
# Ordinary drive/UNC paths outside the private roots, and URLs, keep
# passing: the wider rule must not refuse every Windows path.
separator = "\\"
accepted = (
separator.join(["C:", "Projects", "public-demo", "build"]),
separator * 2 + separator.join(["fileserver", "share", "public-notes.md"]),
"https://example.com/report",
)
for public_ref in accepted:
build_periodic_report_run(_request_with_snapshot_ref(public_ref))


def test_long_separator_run_stays_bounded() -> None:
# The raw-value scan runs before field length validation, so a long
# non-root separator run must stay linear; a quadratic rescan would
# stall the caller instead of failing fast on the length error.
request = _request_with_snapshot_ref("\\" * 32_768)
started = time.perf_counter()
with pytest.raises(ValueError, match="exceeds 500 characters"):
build_periodic_report_run(request)
assert time.perf_counter() - started < 1.0


def test_periodic_report_cli_composes_json(tmp_path, capsys) -> None:
request_path = tmp_path / "request.json"
request_path.write_text(json.dumps(_request()), encoding="utf-8")
Expand All @@ -237,6 +293,32 @@ def test_periodic_report_cli_composes_json(tmp_path, capsys) -> None:
assert payload["run_state"]["status"] == "pending"


def test_periodic_report_cli_refuses_native_windows_private_path(tmp_path, capsys) -> None:
separator = "\\"
request = _request_with_snapshot_ref(
"C:" + separator + separator.join(["Users", "example-user", "notes.md"])
)
request_path = tmp_path / "request.json"
request_path.write_text(json.dumps(request), encoding="utf-8")

assert (
main(
[
"--format",
"json",
"periodic-report",
"compose-run",
"--request-json",
str(request_path),
]
)
== 1
)
payload = json.loads(capsys.readouterr().out)
assert payload["ok"] is False
assert "private path or credential-like value" in payload["error"]


class _MeteredStdin(io.StringIO):
"""Text stdin that records how many UTF-8 bytes the reader pulled."""

Expand Down
Loading