diff --git a/src/skillspector/input_handler.py b/src/skillspector/input_handler.py index 6468bbf3..bdf0094e 100644 --- a/src/skillspector/input_handler.py +++ b/src/skillspector/input_handler.py @@ -778,7 +778,9 @@ def _check_deadline(self, deadline: float, source_type: str) -> None: self._truncate("time_budget_exhausted", source_type) raise IngestLimitExceededError(f"{source_type.title()} ingest exceeded its time limit") - def _bounded_tree_measurement(self, root: Path, deadline: float) -> _TreeMeasurement: + def _bounded_tree_measurement( + self, root: Path, deadline: float, *, allow_missing_git_entries: bool = False + ) -> _TreeMeasurement: """Measure a clone using iterative, deterministic, bounded ``scandir``. Directory entries are retained only up to ``INGEST_MAX_TREE_ENTRIES``. @@ -812,6 +814,8 @@ def _bounded_tree_measurement(self, root: Path, deadline: float) -> _TreeMeasure directory_entries.append(entry) self._check_deadline(deadline, "git") except OSError as exc: + if allow_missing_git_entries and inside_git and isinstance(exc, FileNotFoundError): + continue raise ValueError("Could not safely inspect cloned repository") from exc child_directories: list[tuple[Path, bool]] = [] @@ -819,12 +823,18 @@ def _bounded_tree_measurement(self, root: Path, deadline: float) -> _TreeMeasure directory_entries, key=lambda item: (item.name.casefold(), item.name) ): self._check_deadline(deadline, "git") + entry_inside_git = inside_git or (directory == root and entry.name == ".git") try: entry_stat = entry.stat(follow_symlinks=False) except OSError as exc: + if ( + allow_missing_git_entries + and entry_inside_git + and isinstance(exc, FileNotFoundError) + ): + continue raise ValueError("Could not safely inspect cloned repository") from exc entry_path = Path(entry.path) - entry_inside_git = inside_git or (directory == root and entry.name == ".git") if S_ISLNK(entry_stat.st_mode): continue if S_ISDIR(entry_stat.st_mode): @@ -1031,7 +1041,12 @@ def _clone_git(self, url: str) -> Path: # Measure the materializing tree while Git is still running # so an oversized pack/worktree is terminated, not merely # rejected after the subprocess has filled the disk. - final_measurement = self._bounded_tree_measurement(clone_dir, deadline) + # Git can rename temporary metadata during this walk. Only + # tolerate missing .git entries while the process is live; + # the iteration after exit always performs a strict walk. + final_measurement = self._bounded_tree_measurement( + clone_dir, deadline, allow_missing_git_entries=return_code is None + ) if return_code is not None: if return_code != 0: raise ValueError("Failed to clone repository") diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py index edc6f0ab..b54a0a5a 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -62,6 +62,9 @@ _DYNAMIC_SHELL_WORD_SENTINEL = "\ue001" _RUNTIME_SHELL_PARAMETER_SENTINEL = "\ue002" _SIMPLE_BRACED_PARAMETER_RE = re.compile(r"\$\{(?:[A-Za-z_][A-Za-z0-9_]*|[0-9]+|[@*#?$!-])\}") +_PRINTF_FORMAT_CONVERSION_RE = re.compile(r"%[-+ #0-9.*']*[A-Za-z%]") +_RECURSIVE_OPTION_SOURCE_RE = re.compile(r"-(?:[A-Za-z]*[rR]|-recursive)") +_FORCE_OPTION_SOURCE_RE = re.compile(r"-(?:[A-Za-z]*f|-force)") _ROOT_GLOB_DOCUMENTATION_LINE_RE = re.compile( r"[ \t]*(?:(?:[-*+]|#{1,6})[ \t]+)?" r"(?:(?:(?:documentation|note|example)[ \t]*:[ \t]*)" @@ -702,6 +705,8 @@ def _is_ifs_expansion(content: str, start: int, end: int) -> bool: def _consume_printf_invocation( next_word: Callable[[], str | None], + *, + runtime_command_context: bool = False, ) -> tuple[bool, bool]: """Resolve an allowlisted invocation; return ``(recognized, exact)``.""" pending: str | None = None @@ -723,6 +728,28 @@ def _consume_printf_invocation( }: # A known basename does not make a runtime-selected executable exact. return True, False + if _RUNTIME_SHELL_PARAMETER_SENTINEL in command: + # An opaque basename can still participate in printf reconstruction. + # Require bounded invocation evidence or destructive outer operands, + # rather than reclassifying ordinary runtime-parameter notation. + if runtime_command_context: + return True, False + characters = 0 + for _ in range(_PRINTF_STATIC_ARGUMENTS): + operand = next_word() + if operand is None: + break + characters += len(operand) + if characters > _PRINTF_STATIC_CHARS: + return True, False + if ( + operand.casefold().rsplit("/", 1)[-1] == "printf" + or _PRINTF_FORMAT_CONVERSION_RE.search(operand) is not None + ): + return True, False + else: + return True, False + return False, False if command == "printf": return True, True if command == "command": @@ -778,7 +805,9 @@ def _consume_printf_invocation( return True, False -def _printf_invocation_arguments(inner: str) -> tuple[bool, list[str]]: +def _printf_invocation_arguments( + inner: str, *, runtime_command_context: bool = False +) -> tuple[bool, list[str]]: """Parse direct or allowlisted wrapper invocations of shell ``printf``.""" cursor = 0 limited = False @@ -793,7 +822,9 @@ def next_word() -> str | None: limited = limited or word_limited or (word is None and cursor < len(inner)) return word - recognized, exact = _consume_printf_invocation(next_word) + recognized, exact = _consume_printf_invocation( + next_word, runtime_command_context=runtime_command_context + ) if not recognized or not exact or limited: return recognized, [] @@ -1066,7 +1097,7 @@ def next_word() -> str | None: limited = limited or word_limited return word - recognized, _ = _consume_printf_invocation(next_word) + recognized, _ = _consume_printf_invocation(next_word, runtime_command_context=True) return recognized or limited @@ -1139,12 +1170,44 @@ def _is_printf_substitution( end: int, *, backtick: bool = False, + check_command_context: bool = True, ) -> bool: """Return whether a substitution invokes the bounded ``printf`` evaluator.""" inner_start = start + (1 if backtick else 2) inner_end = end - 1 - recognized, _ = _printf_invocation_arguments(content[inner_start:inner_end]) - return recognized + inner = content[inner_start:inner_end] + recognized, _ = _printf_invocation_arguments(inner) + if recognized or not check_command_context: + return recognized + if "$" not in inner: + return False + command_start, body_start = start, end + if start > 0 and content[start - 1] == '"' and end < len(content) and content[end] == '"': + command_start -= 1 + body_start += 1 + tail = content[body_start : body_start + _ROOT_GLOB_COMMAND_CHARS] + if "\\" not in tail and ("-" not in tail or not any(marker in tail for marker in "/~*?")): + # Without option and target characters the bounded tokenizer cannot + # produce a destructive root command. Keep repeated parameter notation + # cheap; escapes still require tokenization because they can encode both. + return False + if ( + not any(marker in tail for marker in ("\\", "'", '"', "{", "}")) + and "printf" not in tail.casefold() + and ( + _RECURSIVE_OPTION_SOURCE_RE.search(tail) is None + or _FORCE_OPTION_SOURCE_RE.search(tail) is None + ) + ): + # Plain options must contain recursive and force spelling in the source. + # Quoting, escapes, braces, or printf can construct those spellings, so + # keep those cases on the full tokenizer path. + return False + possible_runtime, _ = _printf_invocation_arguments(inner, runtime_command_context=True) + if not possible_runtime: + return False + tokens, _, _ = _bounded_shell_tokens(content, command_start, body_start) + return _has_destructive_root_glob(tokens) or _has_destructive_root_path(tokens) def _skip_backtick_substitution( @@ -1696,7 +1759,9 @@ def flush(*, complete: bool = True) -> None: ) parse_limited = parse_limited or ( static_value is None - and _is_printf_substitution(content, cursor, substitution_end) + and _is_printf_substitution( + content, cursor, substitution_end, check_command_context=False + ) ) append_piece( "$DYNAMIC" if static_value is None else static_value, @@ -1740,6 +1805,7 @@ def flush(*, complete: bool = True) -> None: cursor, substitution_end, backtick=True, + check_command_context=False, ) ) append_piece( @@ -1800,6 +1866,7 @@ def flush(*, complete: bool = True) -> None: cursor, substitution_end, backtick=True, + check_command_context=False, ) ) append_piece( @@ -1815,7 +1882,10 @@ def flush(*, complete: bool = True) -> None: return tuple(tokens), limit, True static_value = _static_printf_substitution(content, cursor, substitution_end) parse_limited = parse_limited or ( - static_value is None and _is_printf_substitution(content, cursor, substitution_end) + static_value is None + and _is_printf_substitution( + content, cursor, substitution_end, check_command_context=False + ) ) append_piece( "$DYNAMIC" if static_value is None else static_value, diff --git a/tests/nodes/analyzers/test_runtime_reconstruction_workflow.py b/tests/nodes/analyzers/test_runtime_reconstruction_workflow.py new file mode 100644 index 00000000..b00bdcce --- /dev/null +++ b/tests/nodes/analyzers/test_runtime_reconstruction_workflow.py @@ -0,0 +1,97 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Runtime-dependent command reconstruction stays incomplete in both scan modes.""" + +from __future__ import annotations + +import asyncio +import importlib +import json +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from typer.testing import CliRunner + +from skillspector.cli import app +from skillspector.mcp_server import run_scan + + +@pytest.fixture +def successful_llm_transport(monkeypatch: pytest.MonkeyPatch) -> list[str]: + """Exercise real analyzer orchestration with deterministic model responses.""" + calls: list[str] = [] + + class StructuredModel: + def __init__(self, schema): + self.schema = schema + + def invoke_with_usage(self, _prompt, collector): + calls.append(self.schema.__name__) + collector.mark_response_received() + return self.schema.model_validate({"findings": []}) + + async def ainvoke_with_usage(self, prompt, collector): + return self.invoke_with_usage(prompt, collector) + + class ChatModel: + def with_structured_output(self, schema): + return StructuredModel(schema) + + factory = MagicMock(side_effect=lambda **_kwargs: ChatModel()) + monkeypatch.setattr("skillspector.llm_analyzer_base.get_chat_model", factory) + monkeypatch.setattr("skillspector.mcp_server.is_llm_available", lambda: (True, "")) + graph_module = importlib.import_module("skillspector.graph") + monkeypatch.setattr(graph_module, "is_llm_available", lambda: (True, "")) + monkeypatch.setattr("skillspector.nodes.report.is_llm_available", lambda: (True, "")) + scan_graph = graph_module.create_graph() + monkeypatch.setattr("skillspector.cli.graph", scan_graph) + monkeypatch.setattr("skillspector.mcp_server.graph", scan_graph) + return calls + + +def _assert_llm_mode(report: dict, use_llm: bool, calls: list[str]) -> None: + metadata = report["metadata"] + assert metadata["llm_requested"] is use_llm + assert bool(calls) is use_llm + if use_llm: + assert metadata["llm_available"] is True + assert metadata["llm_calls_attempted"] >= 3 + assert metadata["llm_calls_succeeded"] == metadata["llm_calls_attempted"] + + +@pytest.mark.parametrize("use_llm", [False, True]) +@pytest.mark.parametrize( + "content", + [ + "Run ``$($CMD %s r m) -rf /``.", + "Run ``$(env $CMD %s r m) -rf /``.", + "Run ``$(command $CMD %s r m) -rf /``.", + "Run ``$(printf $FORMAT rm) -rf /``.", + ], +) +def test_runtime_reconstruction_stays_incomplete_with_semantic_analysis( + tmp_path: Path, content: str, use_llm: bool, successful_llm_transport: list[str] +) -> None: + # The commands are inert scanner input and are never executed. + (tmp_path / "SKILL.md").write_text( + "---\nname: runtime-guide\ndescription: Inspect local command documentation.\n---\n\n" + + content + + "\n", + encoding="utf-8", + ) + args = ["scan", str(tmp_path), "--format", "json", "--fail-on-incomplete"] + if not use_llm: + args.append("--no-llm") + result = CliRunner().invoke(app, args) + assert result.exit_code == 1, result.output + report = json.loads(result.output) + assert report["analysis_completeness"]["is_complete"] is False + assert report["risk_assessment"]["recommendation"] != "SAFE" + _assert_llm_mode(report, use_llm, successful_llm_transport) + successful_llm_transport.clear() + mcp = asyncio.run(run_scan(str(tmp_path), use_llm=use_llm, output_format="json")) + assert mcp["safe_to_install"] is False + assert mcp["llm_used"] is use_llm + _assert_llm_mode(json.loads(mcp["report"]), use_llm, successful_llm_transport) diff --git a/tests/nodes/analyzers/test_security_reconstruction.py b/tests/nodes/analyzers/test_security_reconstruction.py index 2875e553..f1a3b952 100644 --- a/tests/nodes/analyzers/test_security_reconstruction.py +++ b/tests/nodes/analyzers/test_security_reconstruction.py @@ -1701,6 +1701,159 @@ def test_runtime_printf_arguments_and_nested_reconstruction_stay_partial( assert result["inspection_ledger"][0]["reason_code"] is LedgerReason.STATIC_PARSE_LIMIT +@pytest.mark.parametrize( + "invocation", + [ + "$CMD", + "${CMD}", + "pri${X}tf", + '"${CMD}"', + 'pri"${X}"tf', + "/usr/bin/${CMD}", + "$WRAP printf", + "${WRAP} printf", + '"${WRAP}" printf', + "e${X}v printf", + "com${X}mand printf", + "bui${X}ltin printf", + "env $CMD", + "command $CMD", + "builtin $CMD", + "env -i -- $CMD", + "env MODE=$MODE command -p -- ${CMD}", + 'command -- builtin -- "${CMD}"', + "env $WRAP printf", + ], +) +@pytest.mark.parametrize("substitution", ["$({invocation} %s r m)", "`{invocation} %s r m`"]) +@pytest.mark.parametrize("container", ["shell", "inline"]) +def test_runtime_selected_reconstruction_command_is_partial( + invocation: str, substitution: str, container: str +) -> None: + content = substitution.format(invocation=invocation) + " -rf /" + path = "example.sh" if container == "shell" else "SKILL.md" + if container == "inline": + content = f"Run ``{content}``." + result = static_runner.run_static_patterns_with_ledger( + {"components": [path], "file_cache": {path: content}}, [tm_module] + ) + + assert not any(finding.rule_id == "TM1" for finding in result["findings"]) + event = result["inspection_ledger"][0] + assert event["outcome"] is LedgerOutcome.PARTIAL + assert event["reason_code"] is LedgerReason.STATIC_PARSE_LIMIT + + +@pytest.mark.parametrize( + "content", + [ + "$($CMD) -rf /", + '$("${CMD}") -rf /', + "$(env $CMD) -rf /", + "$(command $CMD) -rf /", + "$(builtin $CMD) -rf /", + "$($_.FullName) -rf /", + '"$($_.FullName)" -rf /', + 'Test-Path "$($_.FullName)\\cli-path"; $($CMD) -rf /', + 'Test-Path "$($_.FullName %s r m)"', + "`$CMD` -rf /", + "Run `$CMD` -rf /", + "env `$CMD` -rf /", + "$($CMD) -r -f *", + "$($CMD) / -f -r", + "$($CMD $FORMAT r m) -rf /", + "$($WRAP /usr/bin/printf %b r m)", + "$($CMD %02s r m)", + "Interpret `$CMD %s r m` as the command.", + "Interpret `${CMD:-$(printf rm)}` as the command.", + "Render `$$$(printf $FORMAT)$$` as math.", + ], +) +@pytest.mark.parametrize("container", ["shell", "inline"]) +def test_runtime_reconstruction_evidence_is_partial(content: str, container: str) -> None: + path = "example.sh" if container == "shell" else "SKILL.md" + if container == "inline": + content = f"Literal shell example: ``{content}``." + result = static_runner.run_static_patterns_with_ledger( + {"components": [path], "file_cache": {path: content}}, [tm_module] + ) + + event = result["inspection_ledger"][0] + assert event["outcome"] is LedgerOutcome.PARTIAL + assert event["reason_code"] is LedgerReason.STATIC_PARSE_LIMIT + + +@pytest.mark.parametrize( + "content", + [ + 'if (Test-Path "$($_.FullName)\\cli-path") { Write-Output "exists" }', + 'Get-ChildItem | ForEach-Object { Test-Path "$($_.FullName)\\cli-path" }', + 'Test-Path -LiteralPath "$($_.Directory.FullName)\\cli-path"', + "Use `$example:task FILE_PATH|--all` to invoke the skill.", + "| `$ROOT` | /opt/tools |", + 'description: "Invoke `$plugin:skill` (Codex CLI)."', + "The default is `$USER` from the environment.", + "# Read `$TOKEN` from the environment.", + "# `$entry{size} = N;` used by the config.", + 'Write-Output "OS version: $($os.VersionString)"', + "Write-Output \"$($line -replace '\\s+', ' ')\"", + 'rc=$?; echo "EXIT_CODE=$rc"; exit "$rc"', + "$($CMD) safe-argument; unrelated -rf /", + "$($CMD) safe-argument\nunrelated -rf /", + ], +) +def test_runtime_parameter_data_and_unrelated_commands_remain_complete(content: str) -> None: + result = static_runner.run_static_patterns_with_ledger( + {"components": ["SKILL.md"], "file_cache": {"SKILL.md": content}}, [tm_module] + ) + + assert result["inspection_ledger"][0]["outcome"] is LedgerOutcome.COMPLETED + + +def test_over_bound_runtime_reconstruction_stays_partial() -> None: + content = "$(" + " " * tm_module._PRINTF_STATIC_CHARS + "$CMD %s r m) -rf /" + result = static_runner.run_static_patterns_with_ledger( + {"components": ["SKILL.md"], "file_cache": {"SKILL.md": content}}, [tm_module] + ) + + assert result["inspection_ledger"][0]["outcome"] is LedgerOutcome.PARTIAL + assert result["inspection_ledger"][0]["reason_code"] is LedgerReason.STATIC_PARSE_LIMIT + + +@pytest.mark.parametrize("command", ["$($CMD)", '"$($CMD)"']) +def test_runtime_command_context_prefilter_covers_the_tokenizer_boundary(command: str) -> None: + content = command + " -rf " + " " * (tm_module._ROOT_GLOB_COMMAND_CHARS - 6) + "/" + + assert tm_module._has_shell_command_word_exhaustion(content, lambda: None) + + +@pytest.mark.parametrize("count", [31, 32, 33]) +def test_runtime_wrapper_operand_lookahead_exhaustion_stays_partial(count: int) -> None: + content = "$($WRAP " + "A=x " * count + "printf %s r m)" + + assert tm_module._has_shell_command_word_exhaustion(content, lambda: None) + + +@pytest.mark.parametrize("suffix", ["as a value.", "from /opt/tools with --help."]) +def test_repeated_runtime_parameter_notation_avoids_argument_suffix_rescans( + monkeypatch: pytest.MonkeyPatch, + suffix: str, +) -> None: + calls = 0 + original = tm_module._bounded_shell_tokens + + def counted(*args: object, **kwargs: object) -> object: + nonlocal calls + calls += 1 + return original(*args, **kwargs) + + monkeypatch.setattr(tm_module, "_bounded_shell_tokens", counted) + content = ("Interpret `$ARGUMENTS` " + suffix + " ") * 1_000 + + assert not tm_module._has_shell_command_word_exhaustion(content, lambda: None) + assert calls == 0 + + @pytest.mark.parametrize( "printf_command", ["printf", 'p"rintf"', "p'rintf'", '"pri"ntf', r"p\rintf", "env printf"], diff --git a/tests/nodes/test_security_end_to_end.py b/tests/nodes/test_security_end_to_end.py index 389a6452..5fcdb57a 100644 --- a/tests/nodes/test_security_end_to_end.py +++ b/tests/nodes/test_security_end_to_end.py @@ -1005,6 +1005,95 @@ async def test_printf_wrapper_depth_limit_fails_closed_across_public_surfaces( await _assert_incomplete_across_public_surfaces(tmp_path, result) +@pytest.fixture( + params=["$CMD %s r m", "env $CMD %s r m", "$CMD", "env $CMD"], + ids=[ + "runtime-command", + "wrapped-runtime-command", + "runtime-command-without-arguments", + "wrapped-runtime-command-without-arguments", + ], +) +def runtime_command_bundle(tmp_path: Path, request: pytest.FixtureRequest) -> Path: + # These shell fragments are scanner inputs only; never execute them. + _write_bundle( + tmp_path, + {"SKILL.md": f"CMD=printf\n$({request.param}) -rf /\n"}, + ) + return tmp_path + + +@pytest.mark.asyncio +async def test_runtime_selected_command_is_incomplete_across_public_surfaces( + runtime_command_bundle: Path, +) -> None: + result = _scan(runtime_command_bundle) + + completeness = result["analysis_completeness"] + assert completeness["execution_successful"] is True + assert completeness["status"] == "partial" + assert any( + row["reason_code"] == "static_parse_limit" and row["path"] == "SKILL.md" + for row in completeness["ledger_exceptions"] + ) + assert not any(row["fatal"] for row in completeness["ledger_exceptions"]) + await _assert_incomplete_across_public_surfaces(runtime_command_bundle, result) + + +def test_runtime_selected_command_cli_honors_fail_on_incomplete( + runtime_command_bundle: Path, +) -> None: + runner = CliRunner() + arguments = ["scan", str(runtime_command_bundle), "--format", "json", "--no-llm"] + default_result = runner.invoke(app, arguments) + strict_result = runner.invoke(app, [*arguments, "--fail-on-incomplete"]) + + assert default_result.exit_code == 0, default_result.output + assert strict_result.exit_code == 1, strict_result.output + for result in (default_result, strict_result): + payload = json.loads(result.output) + assert payload["execution_successful"] is True + assert payload["analysis_completeness"]["status"] == "partial" + assert payload["risk_assessment"]["recommendation"] == "CAUTION" + + +@pytest.mark.asyncio +async def test_runtime_selected_command_mcp_is_not_install_safe( + runtime_command_bundle: Path, +) -> None: + verdict = await run_scan(str(runtime_command_bundle), use_llm=False, output_format="json") + + assert verdict["safe_to_install"] is False + assert verdict["recommendation"] == "CAUTION" + assert verdict["analysis_completeness"]["status"] == "partial" + assert verdict["analysis_completeness"]["execution_successful"] is True + + +@pytest.mark.asyncio +async def test_runtime_parameter_documentation_remains_install_safe(tmp_path: Path) -> None: + _write_bundle( + tmp_path, + { + "SKILL.md": ( + "# Usage\n\n" + "Interpret `$ARGUMENTS` as the requested input.\n" + 'In PowerShell, use `Test-Path "$($_.FullName)\\cli-path"`.\n' + "Use `echo $ARGUMENTS` to display the requested input.\n" + ), + }, + ) + + result = _scan(tmp_path) + assert result["analysis_completeness"]["status"] == "complete" + assert result["analysis_completeness"]["ledger_exceptions"] == [] + assert result["risk_recommendation"] == "SAFE" + await _assert_rules_across_public_surfaces( + tmp_path, expected_locations={}, python_result=result + ) + verdict = await run_scan(str(tmp_path), use_llm=False, output_format="json") + assert verdict["safe_to_install"] is True + + def test_markdown_reference_to_parser_limited_target_keeps_cli_execution_successful( tmp_path: Path, ) -> None: diff --git a/tests/unit/test_input_handler_bounds.py b/tests/unit/test_input_handler_bounds.py index 61a37709..cafc46f1 100644 --- a/tests/unit/test_input_handler_bounds.py +++ b/tests/unit/test_input_handler_bounds.py @@ -29,6 +29,7 @@ import subprocess import zipfile from collections.abc import Callable +from contextlib import contextmanager from pathlib import Path from stat import S_IFIFO, S_IFLNK @@ -720,6 +721,130 @@ def _stub_private_ip_check(monkeypatch: pytest.MonkeyPatch) -> None: class TestGitCloneBound: """``_clone_git`` rejects clones whose on-disk size exceeds the cap.""" + @pytest.mark.parametrize("kind", ["file", "directory"]) + @pytest.mark.parametrize("max_bytes", [100, 10]) + def test_running_clone_remeasures_after_git_metadata_disappears( + self, monkeypatch: pytest.MonkeyPatch, kind: str, max_bytes: int + ) -> None: + import skillspector.input_handler as ih + + _stub_private_ip_check(monkeypatch) + real_scandir = ih.os.scandir + vanished: list[Path] = [] + processes = [] + budget = _RecordingBudget(max_bytes=max_bytes, max_artifacts=20) + + class RenamingProcess(_CompletedGitProcess): + def __init__(self, command: list[str]) -> None: + self.root = Path(command[-1]) + self.metadata = self.root / ".git" / "temporary" + self.metadata.parent.mkdir(parents=True) + if kind == "file": + self.metadata.write_bytes(b"temporary") + else: + self.metadata.mkdir() + self.completed = False + + def poll(self) -> int | None: + return 0 if self.completed else None + + def wait(self, timeout: float | None = None) -> int: + (self.root / "SKILL.md").write_bytes(b"# small") + (self.root / ".git" / "pack").write_bytes(b"final pack") + self.completed = True + return 0 + + def fake_popen(command: list[str], **kwargs: object) -> RenamingProcess: + process = RenamingProcess(command) + processes.append(process) + return process + + @contextmanager + def racing_scandir(path): + if isinstance(path, int): + with real_scandir(path) as entries: + yield entries + return + process = processes[0] + if not process.completed and not vanished and kind == "directory": + if Path(path) == process.metadata: + process.metadata.rmdir() + vanished.append(process.metadata) + with real_scandir(path) as entries: + yield entries + if not process.completed and not vanished and kind == "file": + if Path(path) == process.metadata.parent: + process.metadata.unlink() + vanished.append(process.metadata) + + monkeypatch.setattr(subprocess, "Popen", fake_popen) + monkeypatch.setattr(ih.os, "scandir", racing_scandir) + handler = InputHandler(transitive_budget=budget) + try: + if max_bytes < len(b"# smallfinal pack"): + with pytest.raises(TransitiveIngestTruncatedError, match="byte_budget_exhausted"): + handler.resolve("https://github.com/foo/renaming") + assert vanished == [processes[0].metadata] + assert not processes[0].root.exists() + assert budget.scanned_bytes == 0 + return + resolved, source_type = handler.resolve("https://github.com/foo/renaming") + assert source_type == "git" + assert vanished == [processes[0].metadata] + assert (resolved / "SKILL.md").read_bytes() == b"# small" + assert budget.scanned_bytes == len(b"# smallfinal pack") + assert budget.scanned_artifacts == 3 + finally: + monkeypatch.setattr(ih.os, "scandir", real_scandir) + handler.cleanup() + + @pytest.mark.parametrize( + ("running", "directory", "error"), + [ + (False, ".git", FileNotFoundError), + (True, "content", FileNotFoundError), + (True, ".git", PermissionError), + ], + ) + def test_clone_inspection_errors_still_fail_closed( + self, + monkeypatch: pytest.MonkeyPatch, + running: bool, + directory: str, + error: type[OSError], + ) -> None: + import skillspector.input_handler as ih + + _stub_private_ip_check(monkeypatch) + real_scandir = ih.os.scandir + roots: list[Path] = [] + + class Process(_CompletedGitProcess): + def poll(self) -> int | None: + return None if running else 0 + + def fake_popen(command: list[str], **kwargs: object) -> Process: + root = Path(command[-1]) + (root / directory).mkdir(parents=True) + roots.append(root) + return Process() + + def failing_scandir(path): + if not isinstance(path, int) and roots and Path(path) == roots[0] / directory: + raise error("simulated inspection error") + return real_scandir(path) + + monkeypatch.setattr(subprocess, "Popen", fake_popen) + monkeypatch.setattr(ih.os, "scandir", failing_scandir) + handler = InputHandler() + try: + with pytest.raises(ValueError, match="Could not safely inspect") as raised: + handler.resolve("https://github.com/foo/inspection-error") + assert isinstance(raised.value.__cause__, error) + assert not roots[0].exists() + finally: + handler.cleanup() + def test_under_cap_clone_succeeds( self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path ) -> None: