From fc6240b72e5da90fed732ed26a2ebe1e20770406 Mon Sep 17 00:00:00 2001 From: Jason Morcos Date: Sun, 2 Aug 2026 15:33:45 -0700 Subject: [PATCH] test(safety): allow public GitHub attachments --- tests/test_public_api_contract.py | 99 ++++++++++++++++++++----------- tests/test_share_safety.py | 62 +++++++++++++++++++ tests/test_worker_cleanup.py | 8 ++- tools/check_share_safety.py | 91 ++++++++++++++++++++++++++-- 4 files changed, 217 insertions(+), 43 deletions(-) diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index 0a33e78..8467f6a 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -9,23 +9,37 @@ from smartthings_local.ocf.state_cache import StateCache from smartthings_local.protocol.dtls_session import DtlsCoapSession -def _parameter_names(callable_object) -> list[str]: - return list(inspect.signature(callable_object).parameters) +def _assert_compatible_signature(callable_object, expected: list[str]) -> None: + """Require the existing call surface while allowing safe extensions.""" + parameters = list(inspect.signature(callable_object).parameters.values()) + assert [parameter.name for parameter in parameters[: len(expected)]] == expected + for parameter in parameters[len(expected) :]: + assert ( + parameter.kind + in ( + inspect.Parameter.VAR_POSITIONAL, + inspect.Parameter.VAR_KEYWORD, + ) + or parameter.default is not inspect.Parameter.empty + ) def test_dtls_session_constructor_keeps_file_memory_and_local_port_inputs(): - assert _parameter_names(DtlsCoapSession) == [ - "host", - "port", - "cert_path", - "key_path", - "cert_pem", - "key_pem", - "on_notification", - "mtu", - "rate_limit_rps", - "local_port", - ] + _assert_compatible_signature( + DtlsCoapSession, + [ + "host", + "port", + "cert_path", + "key_path", + "cert_pem", + "key_pem", + "on_notification", + "mtu", + "rate_limit_rps", + "local_port", + ], + ) def test_dtls_session_keeps_current_consumer_methods(): @@ -42,23 +56,32 @@ def test_dtls_session_keeps_current_consumer_methods(): "subscribe", } assert expected <= set(dir(DtlsCoapSession)) - assert _parameter_names(DtlsCoapSession.get) == [ - "self", - "path_segs", - "query", - "timeout", - ] - assert _parameter_names(DtlsCoapSession.post) == [ - "self", - "path_segs", - "body_cbor", - "timeout", - ] - assert _parameter_names(DtlsCoapSession.subscribe) == ["self", "path_segs"] + _assert_compatible_signature( + DtlsCoapSession.get, + [ + "self", + "path_segs", + "query", + "timeout", + ], + ) + _assert_compatible_signature( + DtlsCoapSession.post, + [ + "self", + "path_segs", + "body_cbor", + "timeout", + ], + ) + _assert_compatible_signature( + DtlsCoapSession.subscribe, + ["self", "path_segs"], + ) def test_state_cache_keeps_current_consumer_surface(): - assert _parameter_names(StateCache) == ["descriptor"] + _assert_compatible_signature(StateCache, ["descriptor"]) expected = { "apply_optimistic", "apply_rep", @@ -73,10 +96,16 @@ def test_state_cache_keeps_current_consumer_surface(): def test_observe_refresh_task_keeps_current_consumer_surface(): - assert _parameter_names(ObserveRefreshTask) == [ - "session", - "paths", - "interval_s", - "logger", - ] - assert _parameter_names(ObserveRefreshTask.run_forever) == ["self", "stop"] + _assert_compatible_signature( + ObserveRefreshTask, + [ + "session", + "paths", + "interval_s", + "logger", + ], + ) + _assert_compatible_signature( + ObserveRefreshTask.run_forever, + ["self", "stop"], + ) diff --git a/tests/test_share_safety.py b/tests/test_share_safety.py index 0b971fc..d948b3d 100644 --- a/tests/test_share_safety.py +++ b/tests/test_share_safety.py @@ -13,6 +13,22 @@ def test_documentation_addresses_and_synthetic_uuid_are_safe(): assert check_share_safety.scan_text("fixture.txt", text) == [] +def test_dotted_object_identifiers_are_not_ipv4_addresses(): + text = "extendedKeyUsage = 1.3.6.1.4.1.51414.0.1.2" + + assert check_share_safety.scan_text("fixture.txt", text) == [] + + +def test_public_github_attachment_uuid_is_safe_but_bare_uuid_is_not(): + value = "cc1dca15-f272-4625-" + "a13c-2dc82283ff95" + public_url = f"https://github.com/user-attachments/assets/{value}" + + assert check_share_safety.scan_text("README.md", public_url) == [] + assert check_share_safety.scan_text("fixture.txt", value) == [ + check_share_safety.Finding("fixture.txt", 1, "UUID") + ] + + def test_findings_never_echo_matched_content(): cases = { "PEM_PRIVATE_KEY": "-----BEGIN " + "PRIVATE KEY-----", @@ -94,3 +110,49 @@ def test_changed_paths_include_staged_unstaged_and_untracked_files( "staged.txt", "untracked.txt", ] + + +def test_committed_scan_ignores_unchanged_findings_but_checks_added_lines( + tmp_path, monkeypatch +): + def git(*args): + return subprocess.run( + [ + "git", + "-c", + "commit.gpgsign=false", + "-c", + "user.name=Test", + "-c", + "user.email=" + "test" + chr(64) + "example.invalid", + *args, + ], + cwd=tmp_path, + capture_output=True, + check=True, + text=True, + ) + + candidate = tmp_path / "candidate.txt" + private_one = "10." + "24.8.9" + private_two = "10." + "24.8.10" + git("init", "--quiet") + candidate.write_text(f"existing {private_one}\n") + git("add", "candidate.txt") + git("commit", "--quiet", "-m", "baseline") + base = git("rev-parse", "HEAD").stdout.strip() + + candidate.write_text(f"existing {private_one}\nsafe addition\n") + git("add", "candidate.txt") + git("commit", "--quiet", "-m", "safe change") + monkeypatch.chdir(tmp_path) + assert check_share_safety.check_changed(base) == [] + + candidate.write_text( + f"existing {private_one}\nsafe addition\nintroduced {private_two}\n" + ) + git("add", "candidate.txt") + git("commit", "--quiet", "-m", "unsafe change") + assert check_share_safety.check_changed(base) == [ + check_share_safety.Finding("candidate.txt", 3, "NON_DOCUMENTATION_IPV4") + ] diff --git a/tests/test_worker_cleanup.py b/tests/test_worker_cleanup.py index 995fd59..11694b6 100644 --- a/tests/test_worker_cleanup.py +++ b/tests/test_worker_cleanup.py @@ -9,6 +9,8 @@ from smartthings_local.ocf.observe_refresh import ObserveRefreshTask from smartthings_local.ocf.poll_scheduler import PollScheduler, PollTier from smartthings_local.ocf.state_cache import StateCache +_THREAD_DEADLINE_S = 2.0 + class _Session: def ping(self): @@ -45,9 +47,11 @@ def _assert_worker_stops(target, name: str): worker = threading.Thread(target=run, name=name, daemon=True) worker.start() - assert stop.waiting.wait(0.5), f"{name} did not enter an interruptible wait" + assert stop.waiting.wait(_THREAD_DEADLINE_S), ( + f"{name} did not enter an interruptible wait" + ) stop.set() - worker.join(0.5) + worker.join(_THREAD_DEADLINE_S) assert not worker.is_alive(), f"{name} did not stop" assert errors == [], f"{name} raised {errors[0]}" diff --git a/tools/check_share_safety.py b/tools/check_share_safety.py index b35907d..24ca068 100755 --- a/tools/check_share_safety.py +++ b/tools/check_share_safety.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Check changed public files for common private-data and secret shapes. +"""Check introduced public content for common private-data and secret shapes. Findings contain only path, line, and rule ID. Matched content is never printed because it may itself be sensitive. @@ -92,11 +92,14 @@ PATTERNS = ( re.compile(r"(?i)\b(?:qr[_-]?payload|setup[_-]?payload)\b\s*(?::|=)"), ), ) -UUID_RE = re.compile( - r"(?i)(?{UUID_PATTERN})", + re.IGNORECASE, ) IPV4_RE = re.compile( - r"(? bool: def scan_text(path: str, text: str) -> list[Finding]: findings: set[Finding] = set() for line_number, line in enumerate(text.splitlines(), start=1): + public_attachment_uuids = { + match.group("uuid").lower() + for match in PUBLIC_GITHUB_ATTACHMENT_RE.finditer(line) + } for rule_id, pattern in PATTERNS: if pattern.search(line): findings.add(Finding(path, line_number, rule_id)) for match in UUID_RE.finditer(line): - if match.group(0).lower() not in SAFE_UUIDS: + value = match.group(0).lower() + if value not in SAFE_UUIDS and value not in public_attachment_uuids: findings.add(Finding(path, line_number, "UUID")) for match in IPV4_RE.finditer(line): if not _safe_ipv4(match.group(0)): @@ -200,6 +208,73 @@ def _changed_paths(base: str) -> list[str]: ) +def _local_changed_paths() -> set[str]: + commands = ( + ["git", "diff", "--cached", "--name-only", "--diff-filter=ACMR", "-z", "--"], + ["git", "diff", "--name-only", "--diff-filter=ACMR", "-z", "--"], + ["git", "ls-files", "--others", "--exclude-standard", "-z"], + ) + outputs = ( + subprocess.run(command, capture_output=True, check=True).stdout + for command in commands + ) + return { + value.decode("utf-8") + for output in outputs + for value in output.split(b"\0") + if value + } + + +HUNK_RE = re.compile(r"^@@ -\d+(?:,\d+)? \+(?P\d+)(?:,(?P\d+))? @@") + + +def _introduced_lines(base: str, path: str) -> set[int]: + result = subprocess.run( + [ + "git", + "diff", + "--no-color", + "--no-ext-diff", + "--unified=0", + "--diff-filter=ACMR", + f"{base}..HEAD", + "--", + path, + ], + capture_output=True, + check=True, + text=True, + ) + lines: set[int] = set() + for value in result.stdout.splitlines(): + match = HUNK_RE.match(value) + if match is None: + continue + start = int(match.group("start")) + count = int(match.group("count") or 1) + lines.update(range(start, start + count)) + return lines + + +def check_changed(base: str) -> list[Finding]: + """Scan introduced committed lines and all local-only file content.""" + local_paths = _local_changed_paths() + findings: list[Finding] = [] + for path in _changed_paths(base): + path_findings = scan_file(Path(path), path) + if path in local_paths: + findings.extend(path_findings) + continue + introduced = _introduced_lines(base, path) + findings.extend( + finding + for finding in path_findings + if finding.line == 0 or finding.line in introduced + ) + return sorted(set(findings)) + + def check_paths(paths: Iterable[str]) -> list[Finding]: findings: list[Finding] = [] for value in paths: @@ -216,7 +291,11 @@ def main() -> int: paths = _changed_paths(args.changed_since) if args.changed_since else args.paths if not paths: raise ValueError("no paths selected") - findings = check_paths(paths) + findings = ( + check_changed(args.changed_since) + if args.changed_since + else check_paths(paths) + ) except (OSError, UnicodeDecodeError, subprocess.SubprocessError, ValueError): print("share-safety scan failed closed:SCAN_ERROR") return 2