diff --git a/docker/clawscan-runtime/Dockerfile b/docker/clawscan-runtime/Dockerfile index 671b1c7..70f3805 100644 --- a/docker/clawscan-runtime/Dockerfile +++ b/docker/clawscan-runtime/Dockerfile @@ -1,4 +1,10 @@ +ARG SHFMT_VERSION=v3.14.1 +FROM golang:1.27.3 AS bash-parser +ARG SHFMT_VERSION +RUN CGO_ENABLED=0 GOBIN=/out go install "mvdan.cc/sh/v3/cmd/shfmt@${SHFMT_VERSION}" + FROM python:3.12-slim +ARG SHFMT_VERSION ARG SKILLSPECTOR_REF=95a52d89fa16f608f0b18ccee76b0918ea4b7eef ARG AIG_SKILL_SCAN_REF=a5c5a72c2e5b88fc6e5b27e0c86110943c4d350b @@ -23,6 +29,8 @@ RUN apt-get update \ COPY patches/ /tmp/clawscan-patches/ COPY tests/ /tmp/clawscan-adapter-tests/ +COPY --from=bash-parser /out/shfmt /usr/local/bin/shfmt +RUN test "$(shfmt --version)" = "$SHFMT_VERSION" # Exact upstream pins fail closed if a transport patch stops applying. RUN git init /tmp/skillspector \ @@ -41,7 +49,7 @@ RUN git init /tmp/skillspector \ RUN python -m pip install --no-cache-dir --upgrade pip uv \ && python -m pip install --no-cache-dir /tmp/skillspector pytest \ "cisco-ai-skill-scanner==$CISCO_AI_SKILL_SCANNER_VERSION" \ - && python -m pytest -q /tmp/clawscan-adapter-tests/test_skillspector_responses.py /tmp/clawscan-adapter-tests/test_skillspector_applicability.py /tmp/clawscan-adapter-tests/test_skillspector_accounting.py /tmp/clawscan-adapter-tests/test_skillspector_frontmatter_json.py /tmp/clawscan-adapter-tests/test_skillspector_osv_evidence.py /tmp/clawscan-adapter-tests/test_skillspector_pdf.py /tmp/clawscan-adapter-tests/test_skillspector_markdown_containers.py /tmp/clawscan-adapter-tests/test_skillspector_semantic_packets.py /tmp/clawscan-adapter-tests/test_skillspector_meta_packets.py /tmp/clawscan-adapter-tests/test_skillspector_security_view_cache.py /tmp/clawscan-adapter-tests/test_skillspector_ledger_capacity.py /tmp/clawscan-adapter-tests/test_skillspector_table_markers.py /tmp/clawscan-adapter-tests/test_skillspector_fence_boundaries.py /tmp/clawscan-adapter-tests/test_skillspector_csv_markers.py \ + && python -m pytest -q /tmp/clawscan-adapter-tests/test_skillspector_responses.py /tmp/clawscan-adapter-tests/test_skillspector_applicability.py /tmp/clawscan-adapter-tests/test_skillspector_accounting.py /tmp/clawscan-adapter-tests/test_skillspector_frontmatter_json.py /tmp/clawscan-adapter-tests/test_skillspector_osv_evidence.py /tmp/clawscan-adapter-tests/test_skillspector_pdf.py /tmp/clawscan-adapter-tests/test_skillspector_markdown_containers.py /tmp/clawscan-adapter-tests/test_skillspector_semantic_packets.py /tmp/clawscan-adapter-tests/test_skillspector_meta_packets.py /tmp/clawscan-adapter-tests/test_skillspector_security_view_cache.py /tmp/clawscan-adapter-tests/test_skillspector_ledger_capacity.py /tmp/clawscan-adapter-tests/test_skillspector_table_markers.py /tmp/clawscan-adapter-tests/test_skillspector_fence_boundaries.py /tmp/clawscan-adapter-tests/test_skillspector_csv_markers.py /tmp/clawscan-adapter-tests/test_skillspector_shell_source.py \ && uv tool install /tmp/aig/skill-scan --with pytest \ && /root/.local/share/uv/tools/aig-skill-scan/bin/python -m pytest -q /tmp/clawscan-adapter-tests/test_aig_responses.py \ && uv tool install "snyk-agent-scan==$SNYK_AGENT_SCAN_VERSION" \ diff --git a/docker/clawscan-runtime/patches/skillspector-responses.patch b/docker/clawscan-runtime/patches/skillspector-responses.patch index ad02153..e04a0b3 100644 --- a/docker/clawscan-runtime/patches/skillspector-responses.patch +++ b/docker/clawscan-runtime/patches/skillspector-responses.patch @@ -1917,7 +1917,7 @@ index 74c304a..b8bc498 100644 + assert len(requests) == 2 + assert len(osv._cache) == 1 diff --git a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py -index c323c84..49e4a5f 100644 +index c323c84..de65a73 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py +++ b/src/skillspector/nodes/analyzers/static_patterns_tool_misuse.py @@ -41,6 +41,7 @@ from skillspector.artifacts import ( @@ -1928,7 +1928,58 @@ index c323c84..49e4a5f 100644 from skillspector.logging_config import get_logger from skillspector.models import AnalyzerFinding, Finding, Location, Severity from skillspector.python_ast import ParsedPythonFile, parse_python_source -@@ -4847,6 +4848,8 @@ def _markdown_shell_text( +@@ -48,6 +49,7 @@ from skillspector.python_tokens import PythonLiteralSpans + from skillspector.python_tokens import python_literal_spans as _python_literal_spans + from skillspector.python_tokens import python_string_body as _python_string_body + from skillspector.security_reconstruction import validated_json_string_spans ++from skillspector.shell_source import ShellSourceOwnership + from skillspector.state import AnalyzerNodeResponse, SkillspectorState +- ++ + from . import static_runner +@@ -2026,6 +2028,7 @@ def _has_shell_command_word_exhaustion( + structural_quote_closers: set[int] | None = None, + structural_quote_openers: set[int] | None = None, + python_source: _PythonSourceOwnership | None = None, ++ shell_source: ShellSourceOwnership | None = None, + perl_eval_blocks: bool = False, + _command_string_depth: int = 0, + ) -> bool: +@@ -2049,6 +2052,8 @@ def _has_shell_command_word_exhaustion( + for candidate in _SHELL_COMMAND_WORD_START_RE.finditer(content): + check_runtime() + start = candidate.start() ++ if shell_source is not None and shell_source.is_noncommand_start(start): ++ continue + if start in owned_word_positions: + continue + if structural_quote_closers is not None and start in structural_quote_closers: +@@ -2199,6 +2204,7 @@ def _has_shell_command_word_exhaustion( + check_runtime, + perl_eval_blocks=perl_eval_blocks, + python_source=python_source, ++ shell_source=shell_source, + ): + if command_string is None or _command_string_depth >= 8: + return True +@@ -2247,9 +2253,15 @@ def _shell_command_strings( + *, + perl_eval_blocks: bool = False, + python_source: _PythonSourceOwnership | None = None, ++ shell_source: ShellSourceOwnership | None = None, + ) -> Iterator[str | None]: + """Yield bounded strings reparsed by ``eval`` or a shell ``-c`` wrapper.""" +- for clause_start in _shell_clause_starts(content, check_runtime): ++ clause_starts = ( ++ sorted(set(shell_source.command_starts).union(_shell_clause_starts(content, check_runtime))) ++ if shell_source is not None and shell_source.command_starts is not None ++ else _shell_clause_starts(content, check_runtime) ++ ) ++ for clause_start in clause_starts: + check_runtime() + recognized, command_string = _command_string_from_clause( + content, +@@ -4847,6 +4859,8 @@ def _markdown_shell_text( table_columns: int | None = None table_minimum_indent = 0 table_delimiter_line = -1 @@ -1937,7 +1988,7 @@ index c323c84..49e4a5f 100644 offset = 0 lines = content.splitlines(keepends=True) for line_index, line in enumerate(lines): -@@ -4921,6 +4924,13 @@ def _markdown_shell_text( +@@ -4921,6 +4935,13 @@ def _markdown_shell_text( if (html_end and html_end in leading.lower()) or (not html_end and not leading): html_end = None elif fence is not None: @@ -1951,7 +2002,7 @@ index c323c84..49e4a5f 100644 fence_body = stripped.lstrip(" \t") closing_fence = MARKDOWN_FENCE_CLOSE.fullmatch(fence_body) if ( -@@ -4997,6 +5007,7 @@ def _markdown_shell_text( +@@ -4997,6 +5018,7 @@ def _markdown_shell_text( table_minimum_indent = minimum_indent table_delimiter_line = line_index + 1 if table_columns is not None: @@ -1959,7 +2010,7 @@ index c323c84..49e4a5f 100644 mask_inline_delimiters() if line_index != table_delimiter_line: cells = _markdown_table_cells(stripped, prefix, check_runtime) -@@ -5069,6 +5080,13 @@ def _markdown_shell_text( +@@ -5069,6 +5091,13 @@ def _markdown_shell_text( paragraph_list_indent = 0 offset += len(line) mask_inline_delimiters() @@ -1973,7 +2024,19 @@ index c323c84..49e4a5f 100644 projected = "".join(output) if prose_apostrophes: # A prose-looking line can still be inside an earlier shell word, such -@@ -5152,6 +5170,7 @@ def _has_bounded_parse_exhaustion( +@@ -5143,6 +5172,11 @@ def _has_bounded_parse_exhaustion( + structural_quote_openers = None + json_strings: list[tuple[int, int]] = [] + python_source = None ++ shell_source = None ++ if file_type == "shell" and complete_context: ++ shell_source = ShellSourceOwnership(content, check_runtime) ++ if shell_source.has_opaque_assignment: ++ return True + if file_type == "perl" and complete_context: + content = _perl_literal_print_shell_text(content, check_runtime) + if file_type == "python" and complete_context: +@@ -5152,6 +5186,7 @@ def _has_bounded_parse_exhaustion( if file_type == "markdown": if complete_context: json_strings = validated_json_string_spans(content, check_runtime) @@ -1981,6 +2044,14 @@ index c323c84..49e4a5f 100644 structural_quote_closers = {end - 1 for _, end in json_strings} structural_quote_openers = {start for start, _ in json_strings} content = _markdown_shell_text( +@@ -5168,6 +5203,7 @@ def _has_bounded_parse_exhaustion( + structural_quote_closers=structural_quote_closers, + structural_quote_openers=structural_quote_openers, + python_source=python_source, ++ shell_source=shell_source, + perl_eval_blocks=file_type == "perl", + ): + return True diff --git a/src/skillspector/frontmatter_json.py b/src/skillspector/frontmatter_json.py new file mode 100644 --- /dev/null @@ -3330,3 +3401,286 @@ index 0000000..df398a0 + self._closers = _closing_quotes(self._content, self._check_runtime) + self._check_runtime() + return offset in self._closers +diff --git a/src/skillspector/shell_source.py b/src/skillspector/shell_source.py +new file mode 100644 +index 0000000..8631930 +--- /dev/null ++++ b/src/skillspector/shell_source.py +@@ -0,0 +1,277 @@ ++# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. ++# SPDX-License-Identifier: Apache-2.0 ++"""Optional, bounded source ownership from the canonical Bash syntax parser. ++ ++This grants no safety verdict. It distinguishes actual comments, closing quotes, ++string-test operands and command starts from unrelated lexical candidates. Raw ++finding extraction is unchanged, and decoded command strings get no ownership. ++""" ++ ++import json ++import os ++import selectors ++import shutil ++import signal ++import subprocess ++import tempfile ++import time ++from array import array ++from bisect import bisect_right ++from collections.abc import Callable ++ ++MAX_SOURCE_BYTES = 65_536 ++MAX_AST_BYTES = 8 * 1024 * 1024 ++MAX_AST_NODES = 100_000 ++MAX_AST_DEPTH = 128 ++PARSER_SECONDS = 0.5 ++ ++ ++def _intervals(ranges: list[tuple[int, int]]) -> tuple[list[int], list[int]]: ++ merged: list[tuple[int, int]] = [] ++ for start, end in sorted(ranges): ++ if merged and start <= merged[-1][1]: ++ merged[-1] = (merged[-1][0], max(end, merged[-1][1])) ++ else: ++ merged.append((start, end)) ++ return [start for start, _ in merged], [end for _, end in merged] ++ ++ ++def _contains(intervals: tuple[list[int], list[int]], offset: int) -> bool: ++ starts, ends = intervals ++ index = bisect_right(starts, offset) - 1 ++ return index >= 0 and offset < ends[index] ++ ++ ++def _parse(source: bytes, check_runtime: Callable[[], None]) -> dict | None: ++ executable = shutil.which("shfmt") ++ if executable is None or not os.path.isabs(executable) or os.name != "posix": ++ return None ++ # A regular unlinked input descriptor avoids a pipe writer blocking behind ++ # the parser's output. No source, environment credential or diagnostic is ++ # placed in process arguments or retained in a named file. ++ with tempfile.TemporaryFile() as stdin: ++ stdin.write(source) ++ stdin.seek(0) ++ with subprocess.Popen( ++ [executable, "-ln", "bash", "--to-json"], ++ stdin=stdin, ++ stdout=subprocess.PIPE, ++ stderr=subprocess.DEVNULL, ++ env={}, ++ start_new_session=True, ++ ) as process: ++ output = bytearray() ++ deadline = time.monotonic() + PARSER_SECONDS ++ try: ++ assert process.stdout is not None ++ with selectors.DefaultSelector() as selector: ++ selector.register(process.stdout, selectors.EVENT_READ) ++ while selector.get_map(): ++ check_runtime() ++ remaining = deadline - time.monotonic() ++ if remaining <= 0: ++ return None ++ for key, _ in selector.select(min(remaining, 0.01)): ++ chunk = os.read(key.fd, 65_536) ++ if not chunk: ++ selector.unregister(key.fileobj) ++ else: ++ output.extend(chunk) ++ if len(output) > MAX_AST_BYTES: ++ return None ++ if process.wait(timeout=max(0.001, deadline - time.monotonic())) != 0: ++ return None ++ check_runtime() ++ result = json.loads(output) ++ return result if isinstance(result, dict) and result.get("Type") == "File" else None ++ finally: ++ if process.poll() is None: ++ try: ++ os.killpg(process.pid, signal.SIGKILL) ++ except ProcessLookupError: ++ pass ++ process.wait() ++ ++ ++class ShellSourceOwnership: ++ """All-or-nothing source coordinates; unavailable proofs retain the lexer.""" ++ ++ def __init__(self, content: str, check_runtime: Callable[[], None]): ++ self.command_starts: tuple[int, ...] | None = None ++ self.has_opaque_assignment = False ++ self._ownership_allowed = False ++ self._closers: set[int] = set() ++ self._comments = ([], []) ++ self._tests = ([], []) ++ self._bodies = ([], []) ++ check_runtime() ++ runtime_failed = False ++ caller_check = check_runtime ++ ++ def checked_runtime(): ++ nonlocal runtime_failed ++ try: ++ caller_check() ++ except Exception: ++ runtime_failed = True ++ raise ++ ++ check_runtime = checked_runtime ++ try: ++ source = content.encode("utf-8", errors="strict") ++ if len(source) > MAX_SOURCE_BYTES: ++ return ++ tree = _parse(source, check_runtime) ++ if tree is None: ++ return ++ char_offsets = array("i", [-1]) * (len(source) + 1) ++ cursor = 0 ++ for index, character in enumerate(content): ++ if index % 4096 == 0: ++ check_runtime() ++ char_offsets[cursor] = index ++ cursor += len(character.encode("utf-8")) ++ char_offsets[cursor] = len(content) ++ ++ def offset(position: dict) -> int: ++ raw = position["Offset"] ++ if type(raw) is not int or not 0 <= raw <= len(source) or char_offsets[raw] < 0: ++ raise ValueError("Invalid source coordinate") ++ return char_offsets[raw] ++ ++ def literal_word(word: dict) -> str | None: ++ parts = word.get("Parts", []) ++ if len(parts) != 1 or parts[0].get("Type") != "Lit": ++ return None ++ start, end = offset(word["Pos"]), offset(word["End"]) ++ value = content[start:end] ++ return value if value == parts[0].get("Value") and "\\" not in value else None ++ ++ def can_change_parsing(arguments: list[dict]) -> bool: ++ index = 0 ++ for _ in range(32): ++ if index >= len(arguments): ++ return False ++ name = literal_word(arguments[index]) ++ if name is None: ++ return True ++ if name not in {"command", "builtin"}: ++ return name in {"eval", "alias", "source", ".", "shopt"} ++ index += 1 ++ if ( ++ name == "command" ++ and index < len(arguments) ++ and literal_word(arguments[index]) in {"-v", "-V"} ++ ): ++ return False ++ while index < len(arguments) and literal_word(arguments[index]) in ( ++ {"--", "-p"} if name == "command" else {"--"} ++ ): ++ index += 1 ++ return True ++ ++ closers: set[int] = set() ++ comments, tests, bodies, assignments, commands = [], [], [], [], set() ++ pending = [(tree, 0)] ++ count = 0 ++ ownership_allowed = True ++ opaque_assignment = False ++ while pending: ++ check_runtime() ++ node, depth = pending.pop() ++ count += 1 ++ if count > MAX_AST_NODES or depth > MAX_AST_DEPTH: ++ return ++ if isinstance(node, list): ++ pending.extend((item, depth + 1) for item in node) ++ continue ++ if not isinstance(node, dict): ++ continue ++ kind = node.get("Type") ++ if ( ++ isinstance(node.get("Name"), dict) or node.get("Naked") is True ++ ) and isinstance(node.get("Value"), dict): ++ value = node["Value"] ++ if "Parts" in value: ++ assignments.append((offset(value["Pos"]), offset(value["End"]))) ++ if kind in {"DblQuoted", "SglQuoted"}: ++ closing = offset(node["Right"]) ++ quote = '"' if kind == "DblQuoted" else "'" ++ if closing >= len(content) or content[closing] != quote: ++ return ++ closers.add(closing) ++ if "Hash" in node and "Text" in node: ++ start, end = offset(node["Pos"]), offset(node["End"]) ++ if not start < end or content[start] != "#": ++ return ++ comments.append((start, end)) ++ if kind == "CallExpr": ++ # Dynamic names and eval/alias/source can change how later ++ # source is read. Retain only additive command candidates. ++ if can_change_parsing(node.get("Args", [])): ++ ownership_allowed = False ++ commands.add(offset(node["Pos"])) ++ if (kind == "BinaryTest" and node.get("Op") in {"=", "==", "!=", "=~"}) or ( ++ kind == "UnaryTest" and node.get("Op") in {"-n", "-z"} ++ ): ++ tests.append((offset(node["Pos"]), offset(node["End"]))) ++ if kind == "CmdSubst": ++ left = offset(node["Left"]) ++ opening = "`" if node.get("Backquotes") else "$(" ++ if content[left : left + len(opening)] != opening: ++ return ++ start = left + len(opening) ++ bodies.append((start, offset(node["Right"]))) ++ if kind == "ParamExp" and node.get("Exp", {}).get("Op") in {"=", ":="}: ++ # Parameter assignment may invoke arithmetic evaluation ++ # through target attributes. No local value/attribute flow ++ # proof is available; report incomplete independently of ++ # whitespace and the lexer's command-candidate positions. ++ opaque_assignment = True ++ if kind in {"ArithmExp", "ProcSubst"} or ( ++ kind == "ParamExp" ++ and ( ++ any(key in node for key in ("Index", "Slice", "Excl")) ++ or node.get("Exp", {}).get("Op") in {"=", ":="} ++ ) ++ ): ++ # Arithmetic, indirect and assignment parameter forms can ++ # interpret variable contents as source. Assignment to an ++ # integer/nameref target can evaluate the generated value ++ # even when this expansion sits in a string comparison. ++ bodies.append((offset(node["Pos"]), offset(node["End"]))) ++ pending.extend((value, depth + 1) for value in node.values()) ++ assigned_values = _intervals(assignments) ++ executing_bodies = _intervals(bodies) ++ # Do not refine construction of assigned values: later consumers ++ # may interpret them as source. Nested executable substitutions ++ # retain their own independent lexical boundaries. ++ self._closers = { ++ closing ++ for closing in closers ++ if not _contains(assigned_values, closing) or _contains(executing_bodies, closing) ++ } ++ self._comments = _intervals(comments) ++ self._tests = _intervals(tests) ++ self._bodies = executing_bodies ++ self.command_starts = tuple(sorted(commands)) ++ self.has_opaque_assignment = opaque_assignment ++ self._ownership_allowed = ownership_allowed ++ except ( ++ OSError, ++ ValueError, ++ TypeError, ++ KeyError, ++ RecursionError, ++ subprocess.SubprocessError, ++ ): ++ if runtime_failed: ++ raise ++ return ++ ++ def is_noncommand_start(self, offset: int) -> bool: ++ return self._ownership_allowed and ( ++ offset in self._closers ++ or _contains(self._comments, offset) ++ or (_contains(self._tests, offset) and not _contains(self._bodies, offset)) ++ ) diff --git a/docker/clawscan-runtime/tests/test_skillspector_shell_source.py b/docker/clawscan-runtime/tests/test_skillspector_shell_source.py new file mode 100644 index 0000000..a640ed9 --- /dev/null +++ b/docker/clawscan-runtime/tests/test_skillspector_shell_source.py @@ -0,0 +1,161 @@ +"""Canonical shell ownership must retain actual executable uncertainty.""" + +import os +import shlex +import shutil +import subprocess +import time + +import pytest +from skillspector import shell_source +from skillspector.inspection_ledger import LedgerOutcome +from skillspector.nodes.analyzers import static_patterns_tool_misuse as tm +from skillspector.nodes.analyzers import static_runner + + +@pytest.fixture +def canonical_parser(): + executable = shutil.which("shfmt") + assert executable is not None, "Pinned Bash parser is required in the runtime" + assert subprocess.check_output([executable, "--version"], text=True).strip() == "v3.14.1" + + +def ledger(content): + return static_runner.run_static_patterns_with_ledger( + {"components": ["scripts/example.sh"], "file_cache": {"scripts/example.sh": content}}, + [tm], + ) + + +@pytest.mark.parametrize( + "content", + [ + '[[ "$(printf \'%s\' "$VALUE")" == expected ]]\n', + '# 👋 the command can use "$(cat)" as data\n' + "# padding\n" * 1000 + "true\n", + "COUNT=$(find \"$DIR\" -name '*.md' | wc -l | tr -d ' ')\n" + + "# padding\n" * 1000 + + "true\n", + '# Unicode → offsets\r\n[[ "$(printf \'%s\' "$VALUE")" != expected ]]\r\n', + ], +) +def test_complete_nonexecuting_source_roles(canonical_parser, content): + assert shell_source.ShellSourceOwnership(content, lambda: None).command_starts is not None + result = ledger(content) + assert result["inspection_ledger"][0]["outcome"] is LedgerOutcome.COMPLETED + + +@pytest.mark.parametrize( + "content", + [ + '"$(printf \'%s\' "$CMD")" -rf /\n', + 'exec "$CMD" -rf /\n', + 'eval "$(printf \'%s\' "$CMD")"\n', + 'bash -c "$CMD"\n', + '[[ "$( $CMD -rf / )" == expected ]]\n', + '[[ "$(eval "$CMD")" == expected ]]\n', + 'if [[ "$(printf \'%s\' "$VALUE")" == expected ]]; then "$CMD" -rf /; fi\n', + "eval \"$(python3 - <<'PY'\nprint(input())\nPY\n)\"\n", + 'bash -c \'"$(printf "%s" "$CMD")" -rf /\'\n', + 'CMD=$(printf \'%s\' "$VALUE"); eval "$CMD"\n', + 'alias check=eval\ncheck "$(printf \'%s\' "$CMD")"\n', + '[[ "$(printf \'%s\' "$VALUE")" == x ]]; rm -rf /\n', + ], +) +def test_executable_constructions_remain_partial_or_high(canonical_parser, content): + result = ledger(content) + completed = result["inspection_ledger"][0]["outcome"] is LedgerOutcome.COMPLETED + high = any( + finding.rule_id == "TM1" and str(finding.severity).upper() == "HIGH" + for finding in result["findings"] + ) + assert not completed or high + + +@pytest.mark.parametrize("operator", [":=", "="]) +@pytest.mark.parametrize("spacing", ["", " ", "\t"]) +def test_comparison_parameter_assignments_retain_secondary_evaluation(canonical_parser, operator, spacing): + # Integer/nameref attributes can cause an assignment's value to be read as + # arithmetic source even though its outer context is a string comparison. + content = ( + 'PAYLOAD=\'a[$(eval "$CMD")0]\'; unset number; declare -i number; ' + '[[ "${number' + operator + spacing + '$(printf \'%s\' "$PAYLOAD")}" == 0 ]]\n' + ) + ownership = shell_source.ShellSourceOwnership(content, lambda: None) + opener = content.index("$(printf") + assert not ownership.is_noncommand_start(opener) + result = ledger(content) + assert result["inspection_ledger"][0]["outcome"] is not LedgerOutcome.COMPLETED + + +@pytest.mark.parametrize("declaration", ["declare -i", "typeset -i", "local -i"]) +def test_quoted_declaration_values_do_not_gain_quote_ownership(canonical_parser, declaration): + content = ( + declaration + ' "number=$(printf \'%s\' "$PAYLOAD")"\n' + + "# padding\n" * 1000 + "true\n" + ) + ownership = shell_source.ShellSourceOwnership(content, lambda: None) + outer_closing_quote = content.index(')"') + 1 + assert not ownership.is_noncommand_start(outer_closing_quote) + assert ledger(content)["inspection_ledger"][0]["outcome"] is not LedgerOutcome.COMPLETED + + +def test_unavailable_parser_retains_existing_lexical_limit(monkeypatch): + monkeypatch.setattr(shell_source.shutil, "which", lambda _: None) + content = '[[ "$(printf \'%s\' "$VALUE")" == expected ]]\n' + assert shell_source.ShellSourceOwnership(content, lambda: None).command_starts is None + assert tm.has_bounded_parse_exhaustion(content, lambda: None, file_type="shell") + + +@pytest.mark.parametrize( + "content", ['echo "unterminated', 'echo "👋"\n' * 20_000], + ids=["unterminated", "over-source-budget"], +) +def test_malformed_or_oversized_source_has_no_ownership(canonical_parser, content): + assert shell_source.ShellSourceOwnership(content, lambda: None).command_starts is None + + +def test_outer_runtime_budget_propagates(): + class BudgetExceededError(RuntimeError): + pass + + def exhausted(): + raise BudgetExceededError + + with pytest.raises(BudgetExceededError): + shell_source.ShellSourceOwnership("true\n", exhausted) + + +def test_parser_output_budget_is_bounded(canonical_parser, monkeypatch): + monkeypatch.setattr(shell_source, "MAX_AST_BYTES", 64) + assert shell_source.ShellSourceOwnership("echo harmless\n", lambda: None).command_starts is None + + +@pytest.mark.parametrize("error", [TimeoutError, ValueError, RuntimeError]) +def test_outer_budget_failure_inside_parser_is_not_availability_fallback(canonical_parser, error): + calls = 0 + + def exhausted(): + nonlocal calls + calls += 1 + if calls == 2: + raise error("outer deadline") + + with pytest.raises(error, match="outer deadline"): + shell_source.ShellSourceOwnership("true\n", exhausted) + + +def test_parser_timeout_reaps_the_actual_helper(monkeypatch, tmp_path): + pid_file = tmp_path / "helper.pid" + executable = tmp_path / "parser" + executable.write_text( + "#!/bin/sh\nprintf '%s\\n' $$ > " + shlex.quote(str(pid_file)) + "\nexec /bin/sleep 60\n" + ) + executable.chmod(0o700) + monkeypatch.setattr(shell_source.shutil, "which", lambda _: str(executable)) + monkeypatch.setattr(shell_source, "PARSER_SECONDS", 0.1) + started = time.monotonic() + assert shell_source.ShellSourceOwnership("true\n", lambda: None).command_starts is None + assert time.monotonic() - started < 2 + pid = int(pid_file.read_text()) + with pytest.raises(ProcessLookupError): + os.kill(pid, 0)