diff --git a/CHANGELOG.md b/CHANGELOG.md index c6a8c31e..907ab6b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,7 @@ ### Features/Bug Fixes * Inspect hidden and nested ZIP-compatible artifacts under cumulative safety bounds. * Report HIGH SC9 findings for executables concealed in documents or hidden/disguised artifacts. +* Report HIGH SC10 findings when package-manager configuration changes a dependency source trust boundary. --- ### 2.9.6 (Tuesday, August 18, 2026) ### Features/Bug Fixes diff --git a/README.md b/README.md index 5d77b399..3c88c44c 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ SkillSpector is part of the [NVIDIA Verified Skills pipeline](https://docs.nvidi ## Features - **Multi-format input**: Scan Git repos, URLs, zip files, directories, or single files -- **70 vulnerability patterns** across 17 categories: prompt injection, data exfiltration, privilege escalation, supply chain, excessive agency, output handling, system prompt leakage, memory poisoning, tool misuse, rogue agent, anti-refusal, trigger abuse, dangerous code (AST), taint tracking, YARA signatures, MCP least privilege, and MCP tool poisoning +- **71 vulnerability patterns** across 17 categories: prompt injection, data exfiltration, privilege escalation, supply chain, excessive agency, output handling, system prompt leakage, memory poisoning, tool misuse, rogue agent, anti-refusal, trigger abuse, dangerous code (AST), taint tracking, YARA signatures, MCP least privilege, and MCP tool poisoning - **Two-stage analysis**: Fast static analysis + optional LLM semantic evaluation - **Live vulnerability lookups**: SC4 queries [OSV.dev](https://osv.dev) for real-time CVE data with automatic offline fallback - **Multiple output formats**: Terminal, JSON, Markdown, and SARIF reports @@ -353,7 +353,7 @@ claude mcp add skillspector -- skillspector mcp ## Vulnerability Patterns -SkillSpector detects **70 vulnerability patterns** across 17 categories: +SkillSpector detects **71 vulnerability patterns** across 17 categories: ### Prompt Injection (6 patterns) @@ -391,7 +391,7 @@ SkillSpector detects **70 vulnerability patterns** across 17 categories: | PE2 | Sudo/Root Execution | MEDIUM | Invoking elevated system privileges | | PE3 | Credential Access | HIGH | Reading SSH keys, tokens, passwords | -### Supply Chain (9+ patterns) +### Supply Chain (10+ patterns) | ID | Pattern | Severity | Description | |----|---------|----------|-------------| @@ -403,6 +403,7 @@ SkillSpector detects **70 vulnerability patterns** across 17 categories: | SC6 | Typosquatting | HIGH | Package names similar to popular packages | | SC8 | Shipped Python Bytecode | HIGH | `__pycache__` / `.pyc` present (discovery skips; malicious bytecode bypass) | | SC9 | Concealed Executable Artifact | HIGH | Executable nested in a document container or hidden/disguised artifact | +| SC10 | Dependency Source Redirection | HIGH | Package-manager source added, replaced, or unresolved | ### Excessive Agency (4 patterns) diff --git a/docs/DEPENDENCY_SOURCE_REDIRECTION.md b/docs/DEPENDENCY_SOURCE_REDIRECTION.md new file mode 100644 index 00000000..5734b644 --- /dev/null +++ b/docs/DEPENDENCY_SOURCE_REDIRECTION.md @@ -0,0 +1,41 @@ +# Dependency Source Redirection + +SkillSpector reports deterministic HIGH SC10 findings when skill content adds or replaces a +package-manager source, or when the destination cannot be resolved from simple local assignments. +This makes the dependency trust-boundary change explicit without making a reputation judgment +about the destination. + +## Supported ecosystems and surfaces + +| Ecosystem | Direct configuration | Commands and environment | Generated configuration | +|---|---|---|---| +| npm | `.npmrc` registry and scoped registry | `npm config set`, `NPM_CONFIG_REGISTRY` | `.npmrc` heredoc | +| Yarn | `.yarnrc`, `.yarnrc.yml` | `yarn config set` | Yarn config heredoc | +| pip | `pip.conf`, `pip.ini` | index flags, `pip config set`, `PIP_INDEX_URL`, `PIP_EXTRA_INDEX_URL` | pip config heredoc | +| Poetry | `pyproject.toml` sources | `poetry source add`, repository config | `pyproject.toml` heredoc | +| Maven | `settings.xml`, `pom.xml` repositories and mirrors | Maven CLI repository override | Maven XML heredoc | +| Cargo | `.cargo/config`, `.cargo/config.toml` sources and registries | Cargo registry-index environment variables | Cargo config heredoc | + +Commands in executable scripts and shell-language Markdown fences are actionable scan surfaces. +Explanatory prose, comments, and non-shell fences do not create SC10 findings. + +## Evidence + +Each finding records the ecosystem, add/replace operation, configuration surface, scope, +destination, and whether that destination was resolved. Simple literal variables defined in the +same file are resolved without evaluating shell code. Dynamic destinations are reported as +`unresolved` rather than ignored. + +Credentials and sensitive query values embedded in URLs are redacted from findings and every +report format. The analyzer never logs credentials, executes configuration, or contacts the +destination. + +## Trust model + +Canonical public defaults are built into the analyzer solely to avoid reporting an unchanged +default as a redirection. Every other resolved destination is reported uniformly: SkillSpector +does not maintain an organization allowlist, infer whether a host is public or private, perform +DNS resolution, or make network/reputation calls. + +SC10 remains HIGH through optional LLM meta-analysis. An explicit, user-selected baseline retains +its existing ability to suppress reviewed findings. diff --git a/src/skillspector/dependency_sources.py b/src/skillspector/dependency_sources.py new file mode 100644 index 00000000..dd4e8caa --- /dev/null +++ b/src/skillspector/dependency_sources.py @@ -0,0 +1,2119 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Deterministic dependency-source redirection analysis. + +The analyzer models package-manager configuration locally. It does not contact +registries, infer ownership/reputation, or trust explanatory prose. +""" + +from __future__ import annotations + +import configparser +import re +import shlex +import tomllib +import urllib.parse +import xml.etree.ElementTree as ET +from dataclasses import dataclass +from pathlib import PurePosixPath + +from skillspector.models import Finding + +_URL_RE = re.compile( + r"(?:https?|ssh|git\+https?|git\+ssh|sparse\+https)://[^\s'\"<>]+", + re.IGNORECASE, +) +_VARIABLE_RE = re.compile( + r"(?[A-Za-z_][A-Za-z0-9_]*)\}|(?P[A-Za-z_][A-Za-z0-9_]*))" +) +_ASSIGNMENT_WORD_RE = re.compile( + r"(?P[A-Za-z_][A-Za-z0-9_]*)=" + r"(?P'[^']*'|\"(?:\\.|[^\"\\])*\"|[^\s]*)" +) +_FUNCTION_DECLARATION_RE = re.compile( + r"^\s*(?:function\s+(?P[A-Za-z_][A-Za-z0-9_]*)(?:\s*\(\s*\))?" + r"|(?P[A-Za-z_][A-Za-z0-9_]*)\s*\(\s*\))(?P.*)$" +) +_SENSITIVE_QUERY_KEY = re.compile(r"(?:auth|credential|key|pass|secret|signature|token)", re.I) +_SHELL_SUFFIXES = frozenset({".sh", ".bash", ".zsh"}) +_SHELL_SHEBANG_RE = re.compile(r"^#![^\n]*(?:^|/|\s)(?:ba|z|da|k)?sh(?:\s|$)", re.I) + +Assignments = dict[str, list[tuple[int, str | None]]] + +_CANONICAL_DESTINATIONS: dict[str, frozenset[str]] = { + "npm": frozenset({"https://registry.npmjs.org/"}), + "yarn": frozenset( + { + "https://registry.npmjs.org/", + "https://registry.yarnpkg.com/", + } + ), + "pip": frozenset({"https://pypi.org/simple/"}), + "poetry": frozenset({"https://pypi.org/simple/"}), + "maven": frozenset( + { + "https://repo.maven.apache.org/maven2/", + "https://repo1.maven.org/maven2/", + } + ), + "cargo": frozenset( + { + "sparse+https://index.crates.io/", + "https://github.com/rust-lang/crates.io-index/", + } + ), +} + + +@dataclass(frozen=True) +class SourceChange: + """One dependency-source trust-boundary change.""" + + ecosystem: str + operation: str + surface: str + scope: str | None + destination: str + file: str + line: int + matched_text: str + + +@dataclass(frozen=True) +class _HeredocRegion: + target: str + body: str + declaration_line: int + start_line: int + end_line: int + expand_variables: bool + complete: bool + + +@dataclass(frozen=True) +class _ShellHeredocSpec: + """One statically bounded heredoc declaration on a shell command line.""" + + delimiter: str + strip_tabs: bool + expand_variables: bool + input_fd: int + segment: int + command_depth: int + + +@dataclass(frozen=True) +class _HeredocBody: + """The completed body associated with one ordered heredoc declaration.""" + + spec: _ShellHeredocSpec + body: str + start_line: int + end_line: int + + +@dataclass(frozen=True) +class _ShellWord: + """One statically tokenized shell word with its raw assignment shape.""" + + raw: str + value: str + assignment: tuple[str, str] | None + + +_HEREDOC_WORD_BOUNDARIES = frozenset(";|&<>()") + + +def _strip_shell_comment(value: str) -> str: + """Remove an unquoted shell comment without interpreting the command.""" + quote: str | None = None + for index, character in enumerate(value): + if character in {'"', "'"}: + quote = None if quote == character else character if quote is None else quote + elif character == "#" and quote is None and (index == 0 or value[index - 1].isspace()): + return value[:index].rstrip() + return value.strip() + + +def _brace_delta(value: str) -> int: + """Count shell grouping braces while ignoring quotes and parameter expansion.""" + quote: str | None = None + escaped = False + parameter_depth = 0 + delta = 0 + index = 0 + while index < len(value): + character = value[index] + if escaped: + escaped = False + index += 1 + continue + if character == "\\" and quote != "'": + escaped = True + index += 1 + continue + if character in {'"', "'"}: + quote = None if quote == character else character if quote is None else quote + index += 1 + continue + if quote is not None: + index += 1 + continue + if character == "#" and (index == 0 or value[index - 1].isspace()): + break + if character == "$" and value[index : index + 2] == "${": + parameter_depth += 1 + index += 2 + continue + if character == "}" and parameter_depth: + parameter_depth -= 1 + elif character == "{": + delta += 1 + elif character == "}": + delta -= 1 + index += 1 + return delta + + +def _command_segment_body(segment: str, *, allow_case_arm: bool = False) -> str: + """Remove bounded shell-control wrappers around one simple command.""" + candidate = segment.strip().removesuffix("}").strip() + candidate = candidate.lstrip("{").lstrip() + keyword = re.match(r"^(?:then|do|else)\b\s*(?P.*)$", candidate) + if keyword: + candidate = keyword.group("rest") + if allow_case_arm: + case_arm = re.match(r"^[A-Za-z0-9_.*?|/-]+\)\s*(?P.+)$", candidate) + if case_arm: + candidate = case_arm.group("rest") + return candidate.strip() + + +def _leading_assignments(segment: str) -> tuple[list[tuple[str, str]], str]: + """Return leading assignment words and the remaining simple command.""" + candidate = segment.strip() + position = 0 + export = re.match(r"export\b\s*", candidate) + if export and _ASSIGNMENT_WORD_RE.match(candidate, export.end()): + position = export.end() + + assignments: list[tuple[str, str]] = [] + while match := _ASSIGNMENT_WORD_RE.match(candidate, position): + assignments.append((match.group("name"), match.group("value"))) + position = match.end() + if position >= len(candidate): + break + if not candidate[position].isspace(): + return [], candidate + position += len(candidate[position:]) - len(candidate[position:].lstrip()) + remainder = candidate[position:].strip() + if ( + export + and assignments + and all(re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) for name in remainder.split()) + ): + remainder = "" + return assignments, remainder + + +def _assignment_from_word(word: str) -> tuple[str, str] | None: + """Return a static ``NAME=value`` operand.""" + name, separator, value = word.partition("=") + if not separator or re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", name) is None: + return None + return name, value + + +def _raw_shell_words(value: str) -> list[str] | None: + """Split shell words while retaining quoting and command substitutions.""" + words: list[str] = [] + current: list[str] = [] + quote: str | None = None + escaped = False + substitution_depth = 0 + index = 0 + while index < len(value): + character = value[index] + if escaped: + current.append(character) + escaped = False + index += 1 + continue + if character == "\\" and quote != "'": + current.append(character) + escaped = True + index += 1 + continue + if character in {'"', "'", "`"}: + quote = None if quote == character else character if quote is None else quote + current.append(character) + index += 1 + continue + if quote is None and value[index : index + 2] == "$(": + current.extend(("$", "(")) + substitution_depth += 1 + index += 2 + continue + if quote is None and substitution_depth and character == "(": + substitution_depth += 1 + elif quote is None and substitution_depth and character == ")": + substitution_depth -= 1 + if character.isspace() and quote is None and substitution_depth == 0: + if current: + words.append("".join(current)) + current = [] + index += 1 + continue + current.append(character) + index += 1 + if escaped or quote is not None or substitution_depth: + return None + if current: + words.append("".join(current)) + return words + + +def _shell_words(value: str) -> list[_ShellWord] | None: + """Tokenize one bounded segment without losing shell word boundaries.""" + raw_words = _raw_shell_words(value) + if raw_words is None: + return None + words: list[_ShellWord] = [] + for raw in raw_words: + assignment = _assignment_from_word(raw) + try: + normalized = shlex.split(raw, comments=False, posix=True) + except ValueError: + return None + token = normalized[0] if len(normalized) == 1 else raw + if re.search(r"(?:\\[$`]|'[^']*[$`][^']*')", raw): + # Quote removal must not turn a literal dollar/backtick into an + # expandable value when the destination is resolved later. + token = raw + words.append(_ShellWord(raw=raw, value=token, assignment=assignment)) + return words + + +_STATIC_REDIRECTION_TARGET = r"(?:'[^'\n]+'|\"[^\"$`\n]+\"|[-A-Za-z0-9_./:+@%=,]+)" +_STATIC_SUBSHELL_REDIRECTIONS = re.compile( + rf"(?:\d*>&(?:\d+|-)|\d*>>?\s*{_STATIC_REDIRECTION_TARGET})" + rf"(?:\s*(?:\d*>&(?:\d+|-)|\d*>>?\s*{_STATIC_REDIRECTION_TARGET}))*\s*" +) + + +def _strip_outer_subshell(value: str) -> str: + """Strip bounded outer ``(...)`` wrappers around static command lists.""" + candidate = value.strip() + for _ in range(8): + # ``((...))`` is arithmetic syntax in the supported shells, not two + # nested subshells. Requiring whitespace distinguishes ``( (...) )``. + if not candidate.startswith("(") or candidate.startswith("(("): + return candidate + + quote: str | None = None + escaped = False + depth = 0 + closing_index: int | None = None + for index, character in enumerate(candidate): + if escaped: + escaped = False + continue + if character == "\\" and quote != "'": + escaped = True + continue + if character in {'"', "'", "`"}: + quote = None if quote == character else character if quote is None else quote + continue + if quote is not None: + continue + if character == "(": + depth += 1 + elif character == ")": + depth -= 1 + if depth < 0: + return candidate + if depth == 0: + closing_index = index + break + if closing_index is None: + return candidate + + tail = candidate[closing_index + 1 :].strip() + if tail and _STATIC_SUBSHELL_REDIRECTIONS.fullmatch(tail) is None: + return candidate + + inner = candidate[1:closing_index].strip() + if inner.endswith(";") and not inner.endswith(r"\;"): + without_terminator = inner[:-1].rstrip() + if without_terminator.endswith(";"): + return candidate + inner = without_terminator + if not inner: + return candidate + candidate = inner + return candidate + + +def _prepared_shell_segment(segment: str) -> str: + """Remove bounded control, prompt, and subshell wrappers from a segment.""" + candidate = _command_segment_body(segment, allow_case_arm=True) + candidate = re.sub(r"^(?:[$>]\s+)", "", candidate) + return _strip_outer_subshell(candidate) + + +def _persistent_environment_assignments(segment: str) -> list[tuple[str, str]]: + """Return assignments from an assignment-only or ``export`` command.""" + words = _shell_words(_prepared_shell_segment(segment)) + if not words: + return [] + + if words[0].value == "export" and words[0].assignment is None: + index = 1 + if index < len(words) and words[index].value == "--": + index += 1 + elif index < len(words) and words[index].value.startswith("-"): + return [] + assignments: list[tuple[str, str]] = [] + for word in words[index:]: + assignment = word.assignment or _assignment_from_word(word.value) + if assignment is not None: + assignments.append(assignment) + elif re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", word.value) is None: + return [] + return assignments + + if any(word.assignment is None for word in words): + return [] + return [word.assignment for word in words if word.assignment is not None] + + +def _consume_assignment_words( + words: list[_ShellWord], index: int, *, utility_operands: bool = False +) -> tuple[int, list[tuple[str, str]]]: + assignments: list[tuple[str, str]] = [] + while index < len(words): + assignment = words[index].assignment + if utility_operands and assignment is None: + assignment = _assignment_from_word(words[index].value) + if assignment is None: + break + assignments.append(assignment) + index += 1 + return index, assignments + + +def _consume_env_wrapper( + words: list[_ShellWord], index: int +) -> tuple[int, list[tuple[str, str]], bool, set[str]] | None: + """Consume a bounded subset of static ``env`` options and assignments.""" + clear_environment = False + unset_names: set[str] = set() + while index < len(words) and words[index].value.startswith("-"): + option = words[index].value + if option == "--": + index += 1 + break + if option in {"-i", "--ignore-environment"}: + clear_environment = True + index += 1 + continue + if option in {"-u", "--unset"}: + if index + 1 >= len(words): + return None + unset_names.add(words[index + 1].value) + index += 2 + continue + if option in {"-C", "--chdir", "--argv0"}: + if index + 1 >= len(words): + return None + index += 2 + continue + if option.startswith("-u") and len(option) > 2: + unset_names.add(option[2:]) + index += 1 + continue + if option.startswith("--unset="): + unset_names.add(option.split("=", 1)[1]) + index += 1 + continue + if ( + (option.startswith("-C") and len(option) > 2) + or option.startswith("--chdir=") + or option.startswith("--argv0=") + ): + index += 1 + continue + return None + index, assignments = _consume_assignment_words(words, index, utility_operands=True) + return (index, assignments, clear_environment, unset_names) if index < len(words) else None + + +def _consume_sudo_wrapper(words: list[_ShellWord], index: int) -> int | None: + """Consume static sudo execution options, rejecting informational modes.""" + no_argument = { + "-E", + "-H", + "-S", + "-b", + "-n", + "--background", + "--non-interactive", + "--preserve-env", + "--set-home", + "--stdin", + } + with_argument = { + "-C", + "-D", + "-R", + "-T", + "-g", + "-h", + "-p", + "-r", + "-t", + "-u", + "--chdir", + "--chroot", + "--close-from", + "--group", + "--host", + "--prompt", + "--role", + "--type", + "--user", + } + while index < len(words) and words[index].value.startswith("-"): + option = words[index].value + if option == "--": + return index + 1 + if option in no_argument or option.startswith("--preserve-env="): + index += 1 + continue + if option in with_argument: + if index + 1 >= len(words): + return None + index += 2 + continue + if re.match(r"^-[CDRTghprtu].+", option) or re.match( + r"^--(?:chdir|chroot|close-from|group|host|prompt|role|type|user)=.+", option + ): + index += 1 + continue + return None + return index if index < len(words) else None + + +def _consume_command_wrapper(words: list[_ShellWord], index: int) -> int | None: + """Consume execution-preserving options for the shell ``command`` builtin.""" + while index < len(words) and words[index].value.startswith("-"): + option = words[index].value + if option == "--": + index += 1 + break + if option == "-p": + index += 1 + continue + return None + return index if index < len(words) else None + + +def _normalize_executable_command( + segment: str, +) -> tuple[list[str], list[tuple[str, str]]] | None: + """Normalize common static execution wrappers around one simple command.""" + words = _shell_words(_prepared_shell_segment(segment)) + if words is None: + return None + index, assignments = _consume_assignment_words(words, 0) + + for _ in range(8): + if index >= len(words): + return [], assignments + wrapper = words[index].value + if wrapper == "env": + consumed = _consume_env_wrapper(words, index + 1) + if consumed is None: + return None + index, wrapper_assignments, clear_environment, unset_names = consumed + if clear_environment: + assignments.clear() + if unset_names: + assignments = [ + assignment for assignment in assignments if assignment[0] not in unset_names + ] + assignments.extend(wrapper_assignments) + elif wrapper == "sudo": + consumed_index = _consume_sudo_wrapper(words, index + 1) + if consumed_index is None: + return None + index = consumed_index + index, wrapper_assignments = _consume_assignment_words( + words, index, utility_operands=True + ) + assignments.extend(wrapper_assignments) + elif wrapper == "command": + consumed_index = _consume_command_wrapper(words, index + 1) + if consumed_index is None: + return None + index = consumed_index + else: + break + else: + return None + + if ( + index >= len(words) + or re.fullmatch(r"(?:npm|yarn|pip3?|python3?|poetry|mvn|cargo)", words[index].value, re.I) + is None + ): + return None + return [word.value for word in words[index:]], assignments + + +def _environment_source_details(name: str) -> tuple[str, str, str | None] | None: + normalized = name.upper() + if normalized == "NPM_CONFIG_REGISTRY": + return "npm", "replace", None + if normalized == "PIP_INDEX_URL": + return "pip", "replace", None + if normalized == "PIP_EXTRA_INDEX_URL": + return "pip", "add", None + cargo = re.fullmatch(r"CARGO_REGISTRIES_([A-Z0-9_]+)_INDEX", normalized) + if cargo: + return "cargo", "add", cargo.group(1).lower() + return None + + +def _command_environment_ecosystem(command: list[str]) -> str | None: + if command and command[0].lower() == "npm": + return "npm" + if command and command[0].lower() in {"pip", "pip3"}: + return "pip" + if ( + len(command) >= 3 + and command[0].lower() in {"python", "python3"} + and command[1] == "-m" + and command[2].lower() in {"pip", "pip3"} + ): + return "pip" + if command and command[0].lower() == "cargo": + return "cargo" + return None + + +def _command_destination(word: str) -> str | None: + """Keep one argv destination without inventing boundaries inside strings.""" + if any(character.isspace() for character in word) and "$(" not in word and "`" not in word: + return None + return word + + +def _command_source_specs( + command: list[str], +) -> list[tuple[str, str, str, str | None, str]]: + """Return dependency-source changes from one normalized argv vector.""" + if not command: + return [] + lowered = [word.lower() for word in command] + specs: list[tuple[str, str, str, str | None, str]] = [] + + if len(command) >= 5 and lowered[:3] == ["npm", "config", "set"]: + key = command[3] + scope: str | None = None + if key.lower() == "registry": + pass + elif re.fullmatch(r"@[\w.-]+:registry", key, re.I): + scope = key.rsplit(":", 1)[0] + else: + return specs + destination = _command_destination(command[4]) + if destination is not None: + specs.append(("npm", "replace", "npm config set", scope, destination)) + return specs + + if ( + len(command) >= 5 + and lowered[:3] == ["yarn", "config", "set"] + and lowered[3] in {"registry", "npmregistryserver"} + ): + destination = _command_destination(command[4]) + if destination is not None: + specs.append(("yarn", "replace", "yarn config set", None, destination)) + return specs + + pip_args: list[str] | None = None + if lowered[0] in {"pip", "pip3"}: + pip_args = command[1:] + elif ( + len(command) >= 3 + and lowered[0] in {"python", "python3"} + and lowered[1] == "-m" + and lowered[2] in {"pip", "pip3"} + ): + pip_args = command[3:] + if pip_args is not None: + lowered_args = [word.lower() for word in pip_args] + if len(pip_args) >= 4 and lowered_args[:2] == ["config", "set"]: + key = lowered_args[2].removeprefix("global.") + destination = _command_destination(pip_args[3]) + if destination is not None and key in {"index-url", "extra-index-url"}: + specs.append( + ( + "pip", + "add" if key == "extra-index-url" else "replace", + "pip config set", + None, + destination, + ) + ) + return specs + for index, argument in enumerate(pip_args): + lowered_argument = argument.lower() + option: str | None = None + option_destination: str | None = None + for candidate in ("--extra-index-url", "--index-url", "-i"): + if lowered_argument == candidate and index + 1 < len(pip_args): + option = candidate + option_destination = pip_args[index + 1] + break + if lowered_argument.startswith(candidate + "="): + option = candidate + option_destination = argument.split("=", 1)[1] + break + if option is None or option_destination is None: + continue + destination = _command_destination(option_destination) + if destination is not None: + extra = option == "--extra-index-url" + specs.append( + ( + "pip", + "add" if extra else "replace", + "pip --extra-index-url" if extra else "pip --index-url", + None, + destination, + ) + ) + return specs + + if len(command) >= 5 and lowered[:3] == ["poetry", "source", "add"]: + index = 3 + while index < len(command) and command[index].startswith("-"): + index += 1 + if index + 1 < len(command): + destination = _command_destination(command[index + 1]) + if destination is not None and re.fullmatch(r"[\w.-]+", command[index]): + specs.append( + ( + "poetry", + "add", + "poetry source add", + command[index], + destination, + ) + ) + return specs + + if ( + len(command) >= 4 + and lowered[:2] == ["poetry", "config"] + and lowered[2].startswith("repositories.") + ): + scope = command[2].split(".", 1)[1] + destination = _command_destination(command[3]) + if destination is not None and re.fullmatch(r"[\w.-]+", scope): + specs.append(("poetry", "add", "poetry config repositories", scope, destination)) + return specs + + if lowered[0] == "mvn": + prefix = "-Dmaven.repo.remote=" + for argument in command[1:]: + if argument.lower().startswith(prefix.lower()): + destination = _command_destination(argument[len(prefix) :]) + if destination is not None: + specs.append(("maven", "replace", "Maven CLI repository", None, destination)) + return specs + + return specs + + +def _normalize_heredoc_word(word: str) -> tuple[str, bool] | None: + """Apply bounded shell quote removal to one static heredoc word.""" + if not word or word.startswith("#"): + return None + delimiter: list[str] = [] + quoted = False + index = 0 + while index < len(word): + character = word[index] + if character == "$" and index + 1 < len(word) and word[index + 1] == "(": + # The surrounding scanner deliberately fails open for command and + # arithmetic substitutions instead of accepting a partial prefix. + return None + if character == "$" and index + 1 < len(word) and word[index + 1] == "'": + # Support the common static subset of Bash ANSI-C quoting. Escape + # decoding is intentionally rejected rather than approximated. + end = word.find("'", index + 2) + if end < 0 or "\\" in word[index + 2 : end]: + return None + delimiter.append(word[index + 2 : end]) + quoted = True + index = end + 1 + continue + if character == "$" and index + 1 < len(word) and word[index + 1] == '"': + # Treat Bash locale quoting like ordinary quoting. A translated + # delimiter that no longer matches the literal terminator leaves the + # shell input incomplete, so masking the literal complete form is the + # conservative inert-data result. + quoted = True + index += 1 + continue + if character == "'": + end = word.find("'", index + 1) + if end < 0: + return None + delimiter.append(word[index + 1 : end]) + quoted = True + index = end + 1 + continue + if character == '"': + quoted = True + index += 1 + while index < len(word) and word[index] != '"': + if word[index] == "\\": + if index + 1 >= len(word): + return None + escaped = word[index + 1] + if escaped in {"$", "`", '"', "\\"}: + delimiter.append(escaped) + else: + delimiter.extend(("\\", escaped)) + index += 2 + else: + delimiter.append(word[index]) + index += 1 + if index >= len(word): + return None + index += 1 + continue + if character == "\\": + if index + 1 >= len(word): + return None + delimiter.append(word[index + 1]) + quoted = True + index += 2 + continue + if character == "`" or character.isspace() or character in _HEREDOC_WORD_BOUNDARIES: + return None + delimiter.append(character) + index += 1 + normalized = "".join(delimiter) + return (normalized, quoted) if normalized else None + + +def _function_context(content: str, data_lines: set[int]) -> tuple[set[int], dict[str, set[str]]]: + """Locate function definitions and variables they may assign, without executing them.""" + lines = content.splitlines() + function_lines: set[int] = set() + assigned_by_function: dict[str, set[str]] = {} + index = 0 + while index < len(lines): + line_number = index + 1 + if line_number in data_lines: + index += 1 + continue + declaration = _FUNCTION_DECLARATION_RE.match(_strip_shell_comment(lines[index])) + if not declaration: + index += 1 + continue + name = declaration.group("bash") or declaration.group("posix") or "" + rest = declaration.group("rest") + opening_index = index if rest.lstrip().startswith("{") else None + if opening_index is None: + candidate = index + 1 + while candidate < len(lines) and not _strip_shell_comment(lines[candidate]).strip(): + candidate += 1 + if candidate >= len(lines) or not _strip_shell_comment( + lines[candidate] + ).lstrip().startswith("{"): + index += 1 + continue + opening_index = candidate + + function_lines.add(line_number) + depth = 0 + cursor = opening_index + assigned_names: set[str] = set() + while cursor < len(lines): + function_lines.add(cursor + 1) + fragment = rest if cursor == index else lines[cursor] + depth += _brace_delta(fragment) + for _, segment in _shell_parts(fragment): + assignment_words, remainder = _leading_assignments( + _command_segment_body(segment, allow_case_arm=True) + ) + if not remainder: + assigned_names.update(name for name, _ in assignment_words) + cursor += 1 + if depth <= 0: + break + assigned_by_function.setdefault(name, set()).update(assigned_names) + index = max(index + 1, cursor) + return function_lines, assigned_by_function + + +def _literal_assignments(content: str) -> Assignments: + """Collect definite top-level assignments without evaluating shell syntax. + + Heredoc data and function bodies are inert at their physical location, so + their assignment-shaped text is ignored. Assignments in conditional or + iterative control flow are recorded as ambiguous so they cannot silently + make an earlier possible destination appear canonical. + """ + assignments: Assignments = {} + heredoc_data_lines = _heredoc_data_lines(content) + function_lines, assigned_by_function = _function_context(content, heredoc_data_lines) + control_depth = 0 + for line_number, line in enumerate(content.splitlines(), 1): + if line_number in heredoc_data_lines or line_number in function_lines: + continue + for separator, segment in _shell_parts(line): + stripped = _command_segment_body(segment, allow_case_arm=bool(control_depth)) + if re.match(r"^(?:fi|done|esac)\b", stripped): + control_depth = max(0, control_depth - 1) + continue + control = re.match( + r"^(?Pif|elif|case|for|while|until|select)\b\s*(?P.*)$", + stripped, + ) + if control: + keyword = control.group("keyword") + if keyword != "elif": + control_depth += 1 + if keyword == "case": + case_arm = re.match(r"^[^)]*\)\s*(?P.+)$", control.group("rest")) + if not case_arm: + continue + stripped = case_arm.group("rest") + elif keyword in {"if", "elif", "while", "until"}: + stripped = control.group("rest") + else: + continue + + assignment_words, call_candidate = _leading_assignments(stripped) + if assignment_words and not call_candidate: + for name, raw_value in assignment_words: + value = _strip_shell_comment(raw_value).strip() + if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: + value = value[1:-1] + resolved_value: str | None = value + if ( + control_depth + or separator in {"&&", "||", "|", "|&"} + or not value + or "$" in value + or "`" in value + ): + resolved_value = None + assignments.setdefault(name, []).append((line_number, resolved_value)) + continue + + call = re.match(r"^(?:command\s+)?(?P[A-Za-z_][A-Za-z0-9_]*)\b", call_candidate) + if call and call.group("name") in assigned_by_function: + for name in assigned_by_function[call.group("name")]: + assignments.setdefault(name, []).append((line_number, None)) + return assignments + + +def _resolve_value(value: str, assignments: Assignments, use_line: int) -> tuple[str, bool]: + """Resolve simple variables from the latest literal assignment before use.""" + resolved = _strip_shell_comment(value).strip().strip(";,)") + single_quoted = len(resolved) >= 2 and resolved[0] == resolved[-1] == "'" + if single_quoted: + literal = resolved[1:-1] + dynamic = bool("$" in literal or "`" in literal) + return ( + "unresolved" if dynamic or not literal else literal, + not dynamic and bool(literal), + ) + if len(resolved) >= 2 and resolved[0] == resolved[-1] == '"': + resolved = resolved[1:-1] + + def replacement(match: re.Match[str]) -> str: + name = match.group("braced") or match.group("plain") or "" + prior = [assigned for line, assigned in assignments.get(name, []) if line < use_line] + return prior[-1] if prior and prior[-1] is not None else match.group(0) + + resolved = _VARIABLE_RE.sub(replacement, resolved).strip().strip("\"'") + dynamic = bool("$" in resolved or "`" in resolved) + return ("unresolved" if dynamic or not resolved else resolved, not dynamic and bool(resolved)) + + +def _normalize_destination(destination: str) -> str: + """Normalize a URL for comparison with built-in canonical endpoints.""" + if destination == "unresolved": + return destination + try: + parsed = urllib.parse.urlsplit(destination) + except ValueError: + return destination.rstrip("/") + "/" + if not parsed.scheme or not parsed.hostname: + return destination.rstrip("/") + "/" + scheme = parsed.scheme.lower() + hostname = parsed.hostname.lower().rstrip(".") + try: + port = parsed.port + except ValueError: + return destination.rstrip("/") + "/" + if port and not ( + (scheme in {"https", "sparse+https", "git+https"} and port == 443) + or (scheme == "http" and port == 80) + ): + hostname = f"{hostname}:{port}" + path = re.sub(r"/+", "/", parsed.path or "/") + if not path.endswith("/"): + path += "/" + return urllib.parse.urlunsplit((scheme, hostname, path, "", "")) + + +def redact_url(destination: str) -> str: + """Remove URL credentials and sensitive query values from report evidence.""" + if destination == "unresolved": + return destination + try: + parsed = urllib.parse.urlsplit(destination) + except ValueError: + return "" + if not parsed.scheme or not parsed.hostname: + if "@" in destination or _SENSITIVE_QUERY_KEY.search(destination.partition("?")[2]): + return "" + return destination + hostname = parsed.hostname + try: + port = parsed.port + except ValueError: + return "" + if port: + hostname = f"{hostname}:{port}" + if parsed.username is not None or parsed.password is not None: + hostname = f"***@{hostname}" + query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True) + safe_query = [ + (key, "***" if _SENSITIVE_QUERY_KEY.search(key) else value) for key, value in query + ] + return urllib.parse.urlunsplit( + (parsed.scheme, hostname, parsed.path, urllib.parse.urlencode(safe_query), "") + ) + + +def redact_text(text: str) -> str: + """Redact every URL-like token in source evidence.""" + + def replacement(match: re.Match[str]) -> str: + raw = match.group(0) + suffix = "" + while raw and raw[-1] in ".,;)]}": + suffix = raw[-1] + suffix + raw = raw[:-1] + return redact_url(raw) + suffix + + return _URL_RE.sub(replacement, text) + + +def _is_canonical(ecosystem: str, destination: str) -> bool: + normalized = _normalize_destination(destination) + return normalized in _CANONICAL_DESTINATIONS[ecosystem] + + +def _line_for(content: str, needle: str, default: int = 1) -> int: + for index, line in enumerate(content.splitlines(), 1): + if needle and needle in line: + return index + return default + + +def _add_change( + changes: list[SourceChange], + *, + ecosystem: str, + operation: str, + surface: str, + scope: str | None, + raw_destination: str, + file: str, + line: int, + matched_text: str, + assignments: Assignments, +) -> None: + destination, resolved = _resolve_value(raw_destination, assignments, line) + if resolved and _is_canonical(ecosystem, destination): + return + changes.append( + SourceChange( + ecosystem=ecosystem, + operation=operation, + surface=surface, + scope=scope, + destination=destination, + file=file, + line=line, + matched_text=matched_text, + ) + ) + + +def _add_environment_assignment_changes( + changes: list[SourceChange], + assignment_words: list[tuple[str, str]], + *, + file: str, + line: int, + matched_text: str, + assignments: Assignments, + required_ecosystem: str | None = None, +) -> None: + """Add one finding for every supported dependency-source assignment.""" + effective_reversed: list[tuple[str, str]] = [] + seen_names: set[str] = set() + for name, raw_destination in reversed(assignment_words): + if name in seen_names: + continue + seen_names.add(name) + effective_reversed.append((name, raw_destination)) + + for name, raw_destination in reversed(effective_reversed): + details = _environment_source_details(name) + if details is None: + continue + ecosystem, operation, scope = details + if required_ecosystem is not None and ecosystem != required_ecosystem: + continue + _add_change( + changes, + ecosystem=ecosystem, + operation=operation, + surface="environment variable", + scope=scope, + raw_destination=raw_destination, + file=file, + line=line, + matched_text=matched_text, + assignments=assignments, + ) + + +def _parse_npmrc( + content: str, file: str, start_line: int, assignments: Assignments +) -> list[SourceChange]: + changes: list[SourceChange] = [] + for offset, line in enumerate(content.splitlines()): + stripped = line.strip() + if not stripped or stripped.startswith(("#", ";")): + continue + match = re.match(r"(?P(?:@[\w.-]+:)?registry)\s*=\s*(?P.+)$", stripped, re.I) + if not match: + continue + scope = match.group("key").split(":", 1)[0] if match.group("key").startswith("@") else None + _add_change( + changes, + ecosystem="npm", + operation="replace", + surface=".npmrc", + scope=scope, + raw_destination=match.group("value"), + file=file, + line=start_line + offset, + matched_text=line, + assignments=assignments, + ) + return changes + + +def _parse_yarnrc( + content: str, file: str, start_line: int, assignments: Assignments +) -> list[SourceChange]: + changes: list[SourceChange] = [] + current_scope: str | None = None + scope_indent = -1 + for offset, line in enumerate(content.splitlines()): + stripped = line.strip() + if not stripped or stripped.startswith(("#", ";")): + continue + indent = len(line) - len(line.lstrip()) + scope_match = re.match(r"(?P[\w.-]+):\s*$", stripped) + if scope_match and "npmScopes" not in stripped and indent > 0: + current_scope = scope_match.group("scope") + scope_indent = indent + continue + if current_scope and indent <= scope_indent: + current_scope = None + match = re.match( + r"(?Pregistry|npmRegistryServer)\s*(?::|\s)\s*(?P.+)$", + stripped, + re.I, + ) + if not match: + continue + _add_change( + changes, + ecosystem="yarn", + operation="replace", + surface=".yarnrc.yml" if file.lower().endswith((".yml", ".yaml")) else ".yarnrc", + scope=current_scope, + raw_destination=match.group("value"), + file=file, + line=start_line + offset, + matched_text=line, + assignments=assignments, + ) + return changes + + +def _parse_pip_config( + content: str, file: str, start_line: int, assignments: Assignments +) -> list[SourceChange]: + changes: list[SourceChange] = [] + section: str | None = None + for offset, line in enumerate(content.splitlines()): + stripped = line.strip() + if not stripped or stripped.startswith(("#", ";")): + continue + if stripped.startswith("[") and stripped.endswith("]"): + section = stripped[1:-1] + continue + match = re.match(r"(?Pindex-url|extra-index-url)\s*=\s*(?P.+)$", stripped, re.I) + if not match: + continue + key = match.group("key").lower() + _add_change( + changes, + ecosystem="pip", + operation="add" if key == "extra-index-url" else "replace", + surface="pip config", + scope=section, + raw_destination=match.group("value"), + file=file, + line=start_line + offset, + matched_text=line, + assignments=assignments, + ) + return changes + + +def _parse_poetry(content: str, file: str, assignments: Assignments) -> list[SourceChange]: + changes: list[SourceChange] = [] + try: + parsed = tomllib.loads(content) + except tomllib.TOMLDecodeError: + return changes + poetry = parsed.get("tool", {}).get("poetry", {}) + if not isinstance(poetry, dict): + return changes + sources = poetry.get("source", []) + if isinstance(sources, dict): + sources = [sources] + if not isinstance(sources, list): + return changes + for source in sources: + if not isinstance(source, dict) or not isinstance(source.get("url"), str): + continue + destination = str(source["url"]) + _add_change( + changes, + ecosystem="poetry", + operation="add", + surface="pyproject.toml source", + scope=str(source.get("name")) if source.get("name") is not None else None, + raw_destination=destination, + file=file, + line=_line_for(content, destination), + matched_text=next( + (line for line in content.splitlines() if destination in line), destination + ), + assignments=assignments, + ) + return changes + + +def _parse_maven(content: str, file: str, assignments: Assignments) -> list[SourceChange]: + changes: list[SourceChange] = [] + try: + root = ET.fromstring(content) + except ET.ParseError: + return changes + + def local_name(tag: str) -> str: + return tag.rsplit("}", 1)[-1] + + for element in root.iter(): + if local_name(element.tag) not in {"mirror", "repository", "pluginRepository"}: + continue + values = {local_name(child.tag): (child.text or "").strip() for child in element} + destination = values.get("url") + if not destination: + continue + is_mirror = local_name(element.tag) == "mirror" + _add_change( + changes, + ecosystem="maven", + operation="replace" if is_mirror else "add", + surface="settings.xml mirror" if is_mirror else "Maven repository", + scope=values.get("mirrorOf") or values.get("id"), + raw_destination=destination, + file=file, + line=_line_for(content, destination), + matched_text=next( + (line for line in content.splitlines() if destination in line), destination + ), + assignments=assignments, + ) + return changes + + +def _parse_cargo(content: str, file: str, assignments: Assignments) -> list[SourceChange]: + changes: list[SourceChange] = [] + try: + parsed = tomllib.loads(content) + except tomllib.TOMLDecodeError: + return changes + sources = parsed.get("source", {}) + if isinstance(sources, dict): + for name, source in sources.items(): + if not isinstance(source, dict): + continue + replacement = source.get("replace-with") + if isinstance(replacement, str): + target = sources.get(replacement, {}) + destination = target.get("registry") if isinstance(target, dict) else None + raw_destination = str(destination) if destination else "unresolved" + _add_change( + changes, + ecosystem="cargo", + operation="replace", + surface="Cargo source.replace-with", + scope=str(name), + raw_destination=raw_destination, + file=file, + line=_line_for(content, "replace-with"), + matched_text=next( + (line for line in content.splitlines() if "replace-with" in line), + "replace-with", + ), + assignments=assignments, + ) + elif isinstance(source.get("registry"), str): + destination = str(source["registry"]) + _add_change( + changes, + ecosystem="cargo", + operation="add" if name != "crates-io" else "replace", + surface="Cargo source registry", + scope=str(name), + raw_destination=destination, + file=file, + line=_line_for(content, destination), + matched_text=next( + (line for line in content.splitlines() if destination in line), destination + ), + assignments=assignments, + ) + registries = parsed.get("registries", {}) + if isinstance(registries, dict): + for name, registry in registries.items(): + if not isinstance(registry, dict) or not isinstance(registry.get("index"), str): + continue + destination = str(registry["index"]) + _add_change( + changes, + ecosystem="cargo", + operation="add", + surface="Cargo registry index", + scope=str(name), + raw_destination=destination, + file=file, + line=_line_for(content, destination), + matched_text=next( + (line for line in content.splitlines() if destination in line), destination + ), + assignments=assignments, + ) + return changes + + +def _redirection_word(line: str, start: int) -> tuple[str, int, bool]: + """Read one redirection word without accepting a static prefix.""" + index = start + while index < len(line) and line[index].isspace(): + index += 1 + word_start = index + quote: str | None = None + escaped = False + substitution_depth = 0 + dynamic = False + while index < len(line): + character = line[index] + if escaped: + escaped = False + index += 1 + continue + if character == "\\" and quote != "'": + escaped = True + index += 1 + continue + if quote is not None: + if character == quote: + quote = None + index += 1 + continue + if substitution_depth: + if character in {'"', "'"}: + quote = character + elif line[index : index + 2] == "$(": + substitution_depth += 1 + index += 2 + continue + elif character == "(": + substitution_depth += 1 + elif character == ")": + substitution_depth -= 1 + index += 1 + continue + if line[index : index + 2] == "$(": + dynamic = True + substitution_depth = 1 + index += 2 + continue + if character == "`": + dynamic = True + quote = "`" + index += 1 + continue + if character in {'"', "'"}: + quote = character + index += 1 + continue + if character.isspace() or character in _HEREDOC_WORD_BOUNDARIES: + break + index += 1 + malformed = escaped or quote is not None or substitution_depth > 0 + return line[word_start:index], index, dynamic or malformed + + +def _arithmetic_end(line: str, start: int) -> int: + """Skip one balanced ``((...))`` or ``$((...))`` arithmetic expression.""" + opener_length = 3 if line[start : start + 3] == "$((" else 2 + depth = 2 + index = start + opener_length + quote: str | None = None + escaped = False + while index < len(line) and depth: + character = line[index] + if escaped: + escaped = False + elif character == "\\" and quote != "'": + escaped = True + elif character in {'"', "'"}: + quote = None if quote == character else character if quote is None else quote + elif quote is None and character == "(": + depth += 1 + elif quote is None and character == ")": + depth -= 1 + index += 1 + return index + + +def _redirection_start(line: str, operator_index: int) -> int: + """Return the start of an adjacent shell IO number, if one exists.""" + start = operator_index + while start > 0 and line[start - 1] in "0123456789": + start -= 1 + if start > 0 and not (line[start - 1].isspace() or line[start - 1] in ";|&()"): + return operator_index + return start + + +def _redirection_fd(line: str, operator_index: int, default: int) -> int: + """Return an adjacent shell IO number, or the operator's default fd.""" + start = _redirection_start(line, operator_index) + if start == operator_index: + return default + normalized = line[start:operator_index].lstrip("0") or "0" + return int(normalized) if len(normalized) <= 6 else -1 + + +def _static_redirection_target(raw: str, dynamic: bool) -> str | None: + """Normalize one static output-redirection target without expanding it.""" + if dynamic or not raw: + return None + try: + words = shlex.split(raw, comments=False, posix=True) + except ValueError: + return None + return words[0] if len(words) == 1 else None + + +def _scan_shell_redirections( + line: str, +) -> tuple[ + list[_ShellHeredocSpec], + dict[tuple[int, int], str | None], + dict[tuple[int, int], int | None], + str, + bool, +]: + """Scan heredoc and stdout-file redirects in one linear lexical pass.""" + specs: list[_ShellHeredocSpec] = [] + stdout_targets: dict[tuple[int, int], str | None] = {} + stdin_heredocs: dict[tuple[int, int], int | None] = {} + command_characters = list(line) + valid_heredocs = True + quote: str | None = None + escaped = False + segment = 0 + command_depth = 0 + return_quote: str | None = None + return_segment = 0 + index = 0 + + def mask_command_redirection(start: int, end: int) -> None: + if command_depth == 0 and segment == 0: + command_characters[start:end] = [" "] * (end - start) + + while index < len(line): + character = line[index] + if escaped: + escaped = False + index += 1 + continue + if character == "\\" and quote != "'": + escaped = True + index += 1 + continue + if quote == '"' and line[index : index + 2] == "$(": + # A command substitution inside double quotes has its own shell + # grammar; heredoc bodies there remain data, not executable lines. + if command_depth == 0: + return_quote = quote + return_segment = segment + segment = 0 + quote = None + command_depth += 1 + index += 2 + continue + if character in {'"', "'", "`"}: + quote = None if quote == character else character if quote is None else quote + index += 1 + continue + if quote is not None: + index += 1 + continue + if character == "#" and (index == 0 or line[index - 1].isspace()): + break + if line[index : index + 3] == "$((" or line[index : index + 2] == "((": + index = _arithmetic_end(line, index) + continue + if line[index : index + 2] == "$(": + if command_depth == 0: + return_quote = None + return_segment = segment + segment = 0 + command_depth += 1 + index += 2 + continue + if command_depth and character == "(": + command_depth += 1 + index += 1 + continue + if command_depth and character == ")": + command_depth -= 1 + index += 1 + if command_depth == 0: + quote = return_quote + segment = return_segment + return_quote = None + continue + if character == "&" and line[index : index + 2] == "&>": + stdout_targets[(command_depth, segment)] = None + cursor = index + (3 if line[index : index + 3] == "&>>" else 2) + _, end, _ = _redirection_word(line, cursor) + mask_command_redirection(index, end) + index = end + continue + if character in ";|&": + pair = line[index : index + 2] + segment += 1 + index += 2 if pair in {"&&", "||", "|&"} else 1 + continue + if line[index : index + 2] == "<<" and ( + (index == 0 or line[index - 1] != "<") and line[index : index + 3] != "<<<" + ): + cursor = index + 2 + strip_tabs = cursor < len(line) and line[cursor] == "-" + if strip_tabs: + cursor += 1 + raw_word, end, dynamic = _redirection_word(line, cursor) + partial_before_parenthesis = end < len(line) and line[end] == "(" + normalized = ( + None if dynamic or partial_before_parenthesis else _normalize_heredoc_word(raw_word) + ) + if normalized is None: + valid_heredocs = False + else: + delimiter, quoted = normalized + input_fd = _redirection_fd(line, index, 0) + spec_index = len(specs) + specs.append( + _ShellHeredocSpec( + delimiter=delimiter, + strip_tabs=strip_tabs, + expand_variables=not quoted, + input_fd=input_fd, + segment=segment, + command_depth=command_depth, + ) + ) + if input_fd == 0: + stdin_heredocs[(command_depth, segment)] = spec_index + mask_command_redirection(_redirection_start(line, index), end) + index = max(end, cursor) + continue + if character == ">" and (index == 0 or line[index - 1] not in "<>"): + cursor = index + (2 if line[index : index + 2] == ">>" else 1) + fd = _redirection_fd(line, index, 1) + supported_file_redirect = True + if cursor < len(line) and line[cursor] in "&|": + supported_file_redirect = False + cursor += 1 + raw_target, end, dynamic = _redirection_word(line, cursor) + if fd == 1: + stdout_targets[(command_depth, segment)] = ( + _static_redirection_target(raw_target, dynamic) + if supported_file_redirect + else None + ) + mask_command_redirection(_redirection_start(line, index), end) + index = max(end, cursor) + continue + if character == "<" and (index == 0 or line[index - 1] not in "<>"): + if line[index : index + 2] == "<<": + # Here-strings were excluded from the heredoc branch above. + cursor = index + 3 + else: + cursor = index + (2 if line[index : index + 2] in {"<&", "<>"} else 1) + fd = _redirection_fd(line, index, 0) + raw_target, end, _ = _redirection_word(line, cursor) + if fd == 0: + stdin_heredocs[(command_depth, segment)] = None + elif fd == 1: + stdout_targets[(command_depth, segment)] = None + mask_command_redirection(_redirection_start(line, index), end) + index = max(end, cursor if raw_target else cursor) + continue + index += 1 + if not valid_heredocs: + specs = [] + stdin_heredocs = {} + return specs, stdout_targets, stdin_heredocs, "".join(command_characters), valid_heredocs + + +def _ordered_heredoc_bodies( + lines: list[str], header_index: int, specs: list[_ShellHeredocSpec] +) -> tuple[list[_HeredocBody], int, bool]: + """Bind sequential heredoc bodies to their declarations in shell order.""" + bodies: list[_HeredocBody] = [] + body_index = header_index + 1 + for spec in specs: + end = body_index + while end < len(lines): + terminator = lines[end].lstrip("\t") if spec.strip_tabs else lines[end] + if terminator == spec.delimiter: + break + end += 1 + if end >= len(lines): + return bodies, len(lines), False + body_lines = lines[body_index:end] + if spec.strip_tabs: + body_lines = [line.lstrip("\t") for line in body_lines] + bodies.append( + _HeredocBody( + spec=spec, + body="\n".join(body_lines), + start_line=body_index + 1, + end_line=end + 1, + ) + ) + body_index = end + 1 + return bodies, body_index, True + + +def _cat_reads_stdin(command_text: str) -> bool: + """Return whether a bounded simple ``cat`` consumes its stdin.""" + parts = _shell_parts(command_text) + if not parts: + return False + words = _shell_words(parts[0][1]) + if not words or words[0].value != "cat": + return False + + operands: list[_ShellWord] = [] + parse_options = True + informational = {"--help", "--version"} + long_options = { + "--number-nonblank", + "--number", + "--show-all", + "--show-ends", + "--show-nonprinting", + "--show-tabs", + "--squeeze-blank", + } + for word in words[1:]: + value = word.value + if parse_options and value == "--": + parse_options = False + continue + if parse_options and value in informational: + return False + if parse_options and value in long_options: + continue + if parse_options and value.startswith("-") and value != "-": + if re.fullmatch(r"-[AbEenstTuv]+", value) is None: + return False + continue + operands.append(word) + + if not operands: + return True + if any(word.value == "-" for word in operands): + return True + # A dynamic operand can still resolve to the conventional stdin marker. + return any("$" in word.raw or "`" in word.raw for word in operands) + + +def _generated_cat_heredoc( + line: str, + command_text: str, + specs: list[_ShellHeredocSpec], + stdout_targets: dict[tuple[int, int], str | None], + stdin_heredocs: dict[tuple[int, int], int | None], +) -> tuple[str, int] | None: + """Return the target and effective stdin heredoc for a simple ``cat``.""" + if re.match(r"^\s*cat\b", line) is None or not _cat_reads_stdin(command_text): + return None + target = stdout_targets.get((0, 0)) + if target is None: + return None + spec_index = stdin_heredocs.get((0, 0)) + if spec_index is None or spec_index >= len(specs): + return None + return target, spec_index + + +def _heredocs(content: str) -> list[_HeredocRegion]: + """Return generated-config heredocs using ordered, linear redirection scans.""" + lines = content.splitlines() + regions: list[_HeredocRegion] = [] + index = 0 + while index < len(lines): + specs, stdout_targets, stdin_heredocs, command_text, valid = _scan_shell_redirections( + lines[index] + ) + if not valid or not specs: + index += 1 + continue + bodies, next_index, complete = _ordered_heredoc_bodies(lines, index, specs) + if not complete: + # Avoid repeated suffix scans; executable command parsing remains + # fail-open because the unmatched body is not added to data lines. + break + generated = _generated_cat_heredoc( + lines[index], command_text, specs, stdout_targets, stdin_heredocs + ) + if generated is not None: + target, spec_index = generated + selected = bodies[spec_index] + regions.append( + _HeredocRegion( + target=target, + body=selected.body, + declaration_line=index + 1, + start_line=selected.start_line, + end_line=selected.end_line, + expand_variables=selected.spec.expand_variables, + complete=True, + ) + ) + index = next_index + return regions + + +def _shell_heredoc_specs(line: str) -> list[tuple[str, bool]]: + """Return ordered static heredoc delimiters declared by one shell line.""" + specs, _, _, _, valid = _scan_shell_redirections(line) + if not valid: + return [] + return [(spec.delimiter, spec.strip_tabs) for spec in specs] + + +def _heredoc_data_lines(content: str) -> set[int]: + """Return all complete shell heredoc body and terminator lines in one pass.""" + lines = content.splitlines() + data_lines: set[int] = set() + index = 0 + while index < len(lines): + specs, _, _, _, valid = _scan_shell_redirections(lines[index]) + if not valid or not specs: + index += 1 + continue + bodies, next_index, complete = _ordered_heredoc_bodies(lines, index, specs) + for body in bodies: + data_lines.update(range(body.start_line, body.end_line + 1)) + if not complete: + # Fail open for the unmatched body while retaining already completed + # bodies from earlier declarations on the same command line. + return data_lines + index = next_index + return data_lines + + +def _parse_generated_configs( + content: str, file: str, assignments: Assignments +) -> list[SourceChange]: + changes: list[SourceChange] = [] + heredoc_data_lines = _heredoc_data_lines(content) + for region in _heredocs(content): + if not region.complete or region.declaration_line in heredoc_data_lines: + continue + lower = region.target.lower() + region_assignments = assignments if region.expand_variables else {} + if lower.endswith(".npmrc"): + changes.extend(_parse_npmrc(region.body, file, region.start_line, region_assignments)) + elif lower.endswith(".yarnrc") or lower.endswith((".yarnrc.yml", ".yarnrc.yaml")): + changes.extend(_parse_yarnrc(region.body, file, region.start_line, region_assignments)) + elif lower.endswith(("pip.conf", "pip.ini")): + changes.extend( + _parse_pip_config(region.body, file, region.start_line, region_assignments) + ) + elif lower.endswith(("settings.xml", "pom.xml")): + generated = _parse_maven(region.body, file, region_assignments) + changes.extend( + SourceChange( + ecosystem=change.ecosystem, + operation=change.operation, + surface=f"generated {change.surface}", + scope=change.scope, + destination=change.destination, + file=change.file, + line=region.start_line + change.line - 1, + matched_text=change.matched_text, + ) + for change in generated + ) + elif lower.endswith("pyproject.toml"): + generated = _parse_poetry(region.body, file, region_assignments) + changes.extend( + SourceChange( + ecosystem=change.ecosystem, + operation=change.operation, + surface=f"generated {change.surface}", + scope=change.scope, + destination=change.destination, + file=change.file, + line=region.start_line + change.line - 1, + matched_text=change.matched_text, + ) + for change in generated + ) + elif ".cargo/" in lower and lower.endswith(("/config", "/config.toml")): + generated = _parse_cargo(region.body, file, region_assignments) + changes.extend( + SourceChange( + ecosystem=change.ecosystem, + operation=change.operation, + surface=f"generated {change.surface}", + scope=change.scope, + destination=change.destination, + file=change.file, + line=region.start_line + change.line - 1, + matched_text=change.matched_text, + ) + for change in generated + ) + return changes + + +def _shell_parts(line: str) -> list[tuple[str | None, str]]: + """Split shell command lists while retaining the preceding control operator.""" + parts: list[tuple[str | None, str]] = [] + current: list[str] = [] + separator: str | None = None + quote: str | None = None + escaped = False + substitution_depth = 0 + grouping_depth = 0 + index = 0 + while index < len(line): + character = line[index] + if escaped: + current.append(character) + escaped = False + index += 1 + continue + if character == "\\" and quote != "'": + current.append(character) + escaped = True + index += 1 + continue + if character in {'"', "'", "`"}: + quote = None if quote == character else character if quote is None else quote + current.append(character) + index += 1 + continue + if quote is None and line[index : index + 2] == "$(": + current.extend(("$", "(")) + substitution_depth += 1 + index += 2 + continue + if quote is None and substitution_depth and character == "(": + substitution_depth += 1 + elif quote is None and substitution_depth and character == ")": + substitution_depth -= 1 + elif quote is None and character == "(": + grouping_depth += 1 + elif quote is None and character == ")" and grouping_depth: + grouping_depth -= 1 + if ( + quote is None + and substitution_depth == 0 + and character == "#" + and (not current or current[-1].isspace()) + ): + break + pair = line[index : index + 2] + delimiter = pair if pair in {"&&", "||", "|&"} else character + if ( + quote is None + and substitution_depth == 0 + and grouping_depth == 0 + and (character in {";", "|"} or pair in {"&&", "||", "|&"}) + ): + segment = "".join(current).strip() + if segment: + parts.append((separator, segment)) + current = [] + separator = delimiter + index += 2 if pair in {"&&", "||", "|&"} else 1 + continue + current.append(character) + index += 1 + segment = "".join(current).strip() + if segment: + parts.append((separator, segment)) + return parts + + +def _shell_segments(line: str) -> list[str]: + """Split executable shell command lists without evaluating shell syntax.""" + segments: list[str] = [] + for _, segment in _shell_parts(line): + unwrapped = _strip_outer_subshell(segment) + if unwrapped != segment: + segments.extend(_shell_segments(unwrapped)) + else: + segments.append(segment) + return segments + + +def _parse_commands(content: str, file: str, assignments: Assignments) -> list[SourceChange]: + changes: list[SourceChange] = [] + heredoc_data_lines = _heredoc_data_lines(content) + for line_number, line in enumerate(content.splitlines(), 1): + if line_number in heredoc_data_lines: + continue + for segment in _shell_segments(line): + _add_environment_assignment_changes( + changes, + _persistent_environment_assignments(segment), + file=file, + line=line_number, + matched_text=line, + assignments=assignments, + ) + + normalized = _normalize_executable_command(segment) + if normalized is None: + continue + command_candidate, command_assignments = normalized + command_ecosystem = _command_environment_ecosystem(command_candidate) + if command_ecosystem is not None: + _add_environment_assignment_changes( + changes, + command_assignments, + file=file, + line=line_number, + matched_text=line, + assignments=assignments, + required_ecosystem=command_ecosystem, + ) + for ecosystem, operation, surface, scope, destination in _command_source_specs( + command_candidate + ): + _add_change( + changes, + ecosystem=ecosystem, + operation=operation, + surface=surface, + scope=scope, + raw_destination=destination, + file=file, + line=line_number, + matched_text=line, + assignments=assignments, + ) + return changes + + +def _markdown_shell_content(content: str) -> str: + """Keep actionable shell fences while blanking prose and preserving lines.""" + output: list[str] = [] + in_shell = False + for line in content.splitlines(): + fence = re.match(r"^\s*```\s*([\w+-]*)", line) + if fence: + language = fence.group(1).lower() + if in_shell: + in_shell = False + else: + in_shell = language in {"bash", "sh", "shell", "zsh", "console"} + output.append("") + else: + output.append(line if in_shell else "") + return "\n".join(output) + + +def _changes_for_file(content: str, file: str, *, executable: bool = False) -> list[SourceChange]: + normalized = file.replace("\\", "/") + lower = normalized.lower() + name = PurePosixPath(normalized).name.lower() + assignments = _literal_assignments(content) + changes: list[SourceChange] = [] + if name == ".npmrc": + changes.extend(_parse_npmrc(content, file, 1, assignments)) + elif name == ".yarnrc": + changes.extend(_parse_yarnrc(content, file, 1, assignments)) + elif name in {".yarnrc.yml", ".yarnrc.yaml"}: + changes.extend(_parse_yarnrc(content, file, 1, assignments)) + elif name in {"pip.conf", "pip.ini"}: + # ConfigParser validates basic INI structure without executing interpolation. + parser = configparser.ConfigParser(interpolation=None) + try: + parser.read_string(content) + except configparser.Error: + pass + changes.extend(_parse_pip_config(content, file, 1, assignments)) + elif name == "pyproject.toml": + changes.extend(_parse_poetry(content, file, assignments)) + elif name in {"settings.xml", "pom.xml"}: + changes.extend(_parse_maven(content, file, assignments)) + elif name in {"config", "config.toml"} and "/.cargo/" in f"/{lower}": + changes.extend(_parse_cargo(content, file, assignments)) + + suffix = PurePosixPath(normalized).suffix.lower() + is_script = suffix in _SHELL_SUFFIXES or ( + not suffix and executable and bool(_SHELL_SHEBANG_RE.search(content[:256])) + ) + actionable = _markdown_shell_content(content) if name in {"skill.md", "readme.md"} else content + if is_script or actionable != content: + command_assignments = _literal_assignments(actionable) or assignments + changes.extend(_parse_generated_configs(actionable, file, command_assignments)) + changes.extend(_parse_commands(actionable, file, command_assignments)) + return changes + + +def _finding(change: SourceChange, *, local_only: bool) -> Finding: + destination = redact_url(change.destination) + matched_text = redact_text(change.matched_text) + resolved = change.destination != "unresolved" + scope = change.scope or "global" + tags = ["supply-chain", "dependency-source"] + evidence: dict[str, object] = { + "ecosystem": change.ecosystem, + "operation": change.operation, + "surface": change.surface, + "scope": scope, + "destination": destination, + "destination_status": "resolved" if resolved else "unresolved", + } + if local_only: + tags.append("local-only") + evidence["local_only"] = True + return Finding( + rule_id="SC10", + message=( + f"{change.ecosystem} dependency source {change.operation} changes the " + f"trust boundary to {destination}." + ), + severity="HIGH", + confidence=1.0, + file=change.file, + start_line=change.line, + category="Supply Chain", + pattern="Dependency Source Redirection", + finding=matched_text[:200], + explanation=( + "Dependency resolution is redirected away from a canonical default, adds another " + "source, or uses a destination that cannot be resolved statically." + ), + remediation=( + "Review the destination and configuration scope as a dependency trust-boundary " + "change, and keep the intended source explicit and reviewable." + ), + tags=tags, + context=matched_text, + matched_text=matched_text[:200], + evidence=evidence, + ) + + +def analyze_dependency_sources( + components: list[str], + file_cache: dict[str, str], + component_metadata: list[dict[str, object]] | None = None, +) -> list[Finding]: + """Return deterministic HIGH findings for dependency-source trust changes.""" + local_only_paths = { + str(metadata.get("path", "")) + for metadata in component_metadata or [] + if metadata.get("local_only") is True + } + executable_paths = { + str(metadata.get("path", "")) + for metadata in component_metadata or [] + if metadata.get("executable") is True + } + changes: list[SourceChange] = [] + for file in components: + content = file_cache.get(file) + if content is None or "\x00" in content[:8192]: + continue + changes.extend(_changes_for_file(content, file, executable=file in executable_paths)) + + findings: list[Finding] = [] + seen: set[tuple[object, ...]] = set() + for change in changes: + key = ( + change.ecosystem, + change.operation, + change.surface, + change.scope, + change.destination, + change.file, + change.line, + ) + if key in seen: + continue + seen.add(key) + findings.append(_finding(change, local_only=change.file in local_only_paths)) + return findings diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index 2b62ae6d..06312169 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/src/skillspector/llm_analyzer_base.py @@ -40,6 +40,7 @@ from langchain_openai import ChatOpenAI from pydantic import BaseModel, Field, ValidationError, field_validator +from skillspector.inference_usage import InferenceUsageRecord from skillspector.inspection_ledger import ( AnalyzerStatusEvent, InspectionLedgerEvent, @@ -505,7 +506,7 @@ def _model_for_call(self) -> tuple[object, object | None]: return llm, structured @property - def inference_usage(self) -> list[dict[str, object]]: + def inference_usage(self) -> list[InferenceUsageRecord]: """Provider-reported usage captured for this analyzer instance.""" return list(self._usage_collector.snapshot()) diff --git a/src/skillspector/nodes/analyzers/pattern_defaults.py b/src/skillspector/nodes/analyzers/pattern_defaults.py index edbe2f7b..2db27506 100644 --- a/src/skillspector/nodes/analyzers/pattern_defaults.py +++ b/src/skillspector/nodes/analyzers/pattern_defaults.py @@ -95,6 +95,7 @@ class PatternCategory(StrEnum): "SC7": "Code pulls a container image with signature or registry verification disabled (--disable-content-trust, DOCKER_CONTENT_TRUST=0, --insecure-registry). This accepts tampered or unverified images and is a container supply-chain risk.", "SC8": "Skill ships Python bytecode (__pycache__/ or .pyc/.pyo). Discovery skips these paths, so malicious bytecode can score SAFE while decoy sources look clean.", "SC9": "Executable content is concealed inside a document container or hidden/disguised artifact, where extension-based review can miss it.", + "SC10": "Package-manager configuration redirects dependency resolution away from a canonical default, adds another source, or uses an unresolved destination.", # Trigger Abuse "TR1": "Skill uses overly broad trigger patterns that match common words or phrases, causing it to activate in unintended contexts and potentially shadow other skills.", "TR2": "Skill trigger shadows a common built-in command or another skill's trigger, potentially intercepting requests meant for trusted functionality.", @@ -195,6 +196,7 @@ class PatternCategory(StrEnum): "SC7": PatternCategory.SUPPLY_CHAIN.value, "SC8": PatternCategory.SUPPLY_CHAIN.value, "SC9": PatternCategory.SUPPLY_CHAIN.value, + "SC10": PatternCategory.SUPPLY_CHAIN.value, "TR1": PatternCategory.TRIGGER_ABUSE.value, "TR2": PatternCategory.TRIGGER_ABUSE.value, "TR3": PatternCategory.TRIGGER_ABUSE.value, @@ -282,6 +284,7 @@ class PatternCategory(StrEnum): "SC7": "Untrusted Container Image", "SC8": "Shipped Python Bytecode", "SC9": "Concealed Executable Artifact", + "SC10": "Dependency Source Redirection", "TR1": "Overly Broad Trigger", "TR2": "Shadow Command Trigger", "TR3": "Keyword Baiting Trigger", @@ -378,6 +381,7 @@ class PatternCategory(StrEnum): "SC7": "Keep image signature verification (Docker Content Trust / cosign) and registry TLS enabled. Pull only signed images from trusted registries; never disable content-trust or use insecure registries in skill code.", "SC8": "Do not ship __pycache__/ or .pyc/.pyo in skills. Delete bytecode before packaging; if presence is intentional for a lab fixture, quarantine it outside the skill install path.", "SC9": "Keep executable files explicit and directly reviewable. Review the artifact provenance and why executable content is packaged inside a document, hidden file, or disguised container.", + "SC10": "Review the destination and configuration scope as a dependency trust-boundary change, and keep the intended package source explicit and reviewable.", # Trigger Abuse "TR1": "Use specific, narrow trigger patterns that match only the skill's intended use case. Avoid single-word or common-phrase triggers.", "TR2": "Choose triggers that do not conflict with built-in commands or other skills. Prefix with a unique namespace if necessary.", diff --git a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py index 8de22759..c77c5ccc 100644 --- a/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py +++ b/src/skillspector/nodes/analyzers/static_patterns_supply_chain.py @@ -13,7 +13,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Static patterns: supply chain (SC1–SC9) and trigger analysis (TR1–TR3). +"""Static patterns: supply chain (SC1–SC10) and trigger analysis (TR1–TR3). SC1–SC3: regex-based pattern matching (original implementation). SC4: Known vulnerable dependencies — live OSV.dev lookup with static fallback. @@ -22,6 +22,7 @@ SC7: Untrusted container image — flags image signature / registry-verification bypass. SC8: Shipped Python bytecode — flags __pycache__/ and *.pyc/*.pyo that discovery skips. SC9: Concealed executable artifact — flags executables nested in document or hidden artifacts. +SC10: Dependency source redirection — flags noncanonical package registries and indexes. TR1–TR3: Trigger analysis — flags overly broad, shadowing, or baiting triggers. Node and analyze() in one module. @@ -40,6 +41,7 @@ from packaging.requirements import InvalidRequirement, Requirement from packaging.version import InvalidVersion, Version +from skillspector.dependency_sources import analyze_dependency_sources from skillspector.inspection_ledger import LedgerOutcome, analyzer_status_for_events, ledger_event from skillspector.logging_config import get_logger from skillspector.models import AnalyzerFinding, Finding, Location, Severity @@ -1377,7 +1379,7 @@ def _analyze_concealed_executables( def node(state: SkillspectorState) -> AnalyzerNodeResponse: - """Run supply_chain patterns (SC1–SC9) and trigger analysis (TR1–TR3).""" + """Run supply_chain patterns (SC1–SC10) and trigger analysis (TR1–TR3).""" # SC1–SC3 via static_runner response = static_runner.run_static_patterns_with_ledger(state, [sys.modules[__name__]]) findings = response["findings"] @@ -1476,6 +1478,20 @@ def record_extra_findings( f"{ANALYZER_ID}_concealed_executable", ) + # SC10: deterministic dependency registry/source trust-boundary changes. + dependency_source_findings = analyze_dependency_sources( + components, + file_cache, + component_metadata, + ) + findings.extend(dependency_source_findings) + for finding_path in sorted({finding.file for finding in dependency_source_findings}): + record_extra_findings( + finding_path, + [finding for finding in dependency_source_findings if finding.file == finding_path], + f"{ANALYZER_ID}_dependency_source", + ) + logger.info("%s: %d findings", ANALYZER_ID, len(findings)) response["analyzer_status_events"] = [ analyzer_status_for_events(ANALYZER_ID, response["inspection_ledger"]) diff --git a/src/skillspector/nodes/meta_analyzer.py b/src/skillspector/nodes/meta_analyzer.py index f82e1779..e1cc5503 100644 --- a/src/skillspector/nodes/meta_analyzer.py +++ b/src/skillspector/nodes/meta_analyzer.py @@ -28,6 +28,7 @@ from pydantic import BaseModel, Field, field_validator from skillspector.constants import _SKILLSPECTOR_DEFAULT_MODEL +from skillspector.dependency_sources import redact_text from skillspector.inspection_ledger import ( AnalyzerStatusEvent, InspectionLedgerEvent, @@ -221,10 +222,11 @@ def _format_findings_for_prompt(findings: list[Finding]) -> str: for i, f in enumerate(findings, 1): end = f"–{f.end_line}" if f.end_line and f.end_line != f.start_line else "" loc = f"{f.file}:{f.start_line}{end}" - matched = f.matched_text or f.message - ctx = f.context or "" + message = redact_text(f.message) + matched = redact_text(f.matched_text or f.message) + ctx = redact_text(f.context or "") lines.append( - f"{i}. [{f.rule_id}] {f.message} ({f.severity})\n" + f"{i}. [{f.rule_id}] {message} ({f.severity})\n" f" Location: {loc}\n" f" Matched: {matched}\n" f" Context:\n " + "\n ".join(ctx.splitlines()) @@ -235,6 +237,7 @@ def _format_findings_for_prompt(findings: list[Finding]) -> str: _NO_LLM_CONFIDENCE_THRESHOLD = 0.4 _HIGH_SEVERITY_PASS_THROUGH = frozenset({"CRITICAL", "HIGH"}) _CODE_EXAMPLE_DOWNWEIGHT = 0.5 +_AUTHORITATIVE_DETERMINISTIC_RULES = frozenset({"SC9", "SC10"}) def _fallback_filtered(findings: list[Finding]) -> list[Finding]: @@ -253,6 +256,9 @@ def _fallback_filtered(findings: list[Finding]) -> list[Finding]: result: list[Finding] = [] for f in findings: + if f.rule_id in _AUTHORITATIVE_DETERMINISTIC_RULES: + result.append(f) + continue severity_upper = (f.severity or "LOW").upper() confidence = f.confidence if f.context and is_code_example(f.context): @@ -299,7 +305,9 @@ def _passthrough_with_defaults(findings: list[Finding]) -> list[Finding]: should fail-closed — showing more findings is safer than silently dropping. """ return [ - Finding( + f + if f.rule_id in _AUTHORITATIVE_DETERMINISTIC_RULES + else Finding( rule_id=f.rule_id, message=f.message, finding_id=f.finding_id, @@ -347,7 +355,7 @@ def _estimate_extra_overhead(self, findings: list[Finding]) -> int: return estimate_tokens(_format_findings_for_prompt(findings)) def build_prompt(self, batch: Batch, **kwargs: object) -> str: - metadata_text = kwargs.get("metadata_text", "No metadata available") + metadata_text = redact_text(str(kwargs.get("metadata_text", "No metadata available"))) findings_text = _format_findings_for_prompt(batch.findings) return self.base_prompt.format( metadata=metadata_text, @@ -356,6 +364,19 @@ def build_prompt(self, batch: Batch, **kwargs: object) -> str: static_findings=findings_text, ) + def get_batches( + self, + file_paths: list[str], + file_cache: dict[str, str], + findings: list[Finding] | None = None, + ) -> list[Batch]: + """Redact credential-bearing SC10 source text before provider batching.""" + batches = super().get_batches(file_paths, file_cache, findings) + for batch in batches: + if any(finding.rule_id == "SC10" for finding in batch.findings): + batch.content = redact_text(batch.content) + return batches + def parse_response( # type: ignore[override] # Base class permits custom parsed values. self, response: MetaAnalyzerResult, @@ -438,6 +459,9 @@ def apply_filter( result: list[Finding] = [] for f in findings: + if f.rule_id in _AUTHORITATIVE_DETERMINISTIC_RULES: + result.append(f) + continue exact_key = (f.file, f.rule_id, f.start_line, f.end_line) start_only_key = (f.file, f.rule_id, f.start_line, None) coarse_key = (f.file, f.rule_id) diff --git a/src/skillspector/nodes/report.py b/src/skillspector/nodes/report.py index 921fe534..a01eaa76 100644 --- a/src/skillspector/nodes/report.py +++ b/src/skillspector/nodes/report.py @@ -35,6 +35,7 @@ from rich.table import Table from skillspector import __version__ as skillspector_version +from skillspector.dependency_sources import redact_text from skillspector.inference_usage import sanitize_inference_usage from skillspector.inspection_ledger import AnalysisCompleteness from skillspector.llm_utils import is_llm_available @@ -104,19 +105,24 @@ def _clean_text(value: str | None) -> str | None: def _sanitize_finding(finding: Finding) -> Finding: """Return a copy of *finding* with control/ANSI bytes stripped from text fields.""" + + def clean(value: str | None) -> str | None: + cleaned = _clean_text(value) + return redact_text(cleaned) if isinstance(cleaned, str) else cleaned + evidence = { - _clean_text(str(key)) or "": _clean_text(value) if isinstance(value, str) else value + clean(str(key)) or "": clean(value) if isinstance(value, str) else value for key, value in finding.evidence.items() } return replace( finding, - message=_clean_text(finding.message) or "", - explanation=_clean_text(finding.explanation), - remediation=_clean_text(finding.remediation), - finding=_clean_text(finding.finding), - context=_clean_text(finding.context), - matched_text=_clean_text(finding.matched_text), - code_snippet=_clean_text(finding.code_snippet), + message=clean(finding.message) or "", + explanation=clean(finding.explanation), + remediation=clean(finding.remediation), + finding=clean(finding.finding), + context=clean(finding.context), + matched_text=clean(finding.matched_text), + code_snippet=clean(finding.code_snippet), evidence=evidence, ) diff --git a/tests/integration/test_graph.py b/tests/integration/test_graph.py index 8ec668e3..c6e1d57f 100644 --- a/tests/integration/test_graph.py +++ b/tests/integration/test_graph.py @@ -42,6 +42,117 @@ def test_graph_invoke_with_output_format_json(tmp_path: Path) -> None: assert "components" in data +@pytest.mark.parametrize("output_format", ["terminal", "json", "markdown", "sarif"]) +@pytest.mark.parametrize( + ("script", "expected_destination"), + [ + ( + "MARKER=1 NPM_CONFIG_REGISTRY=https://packages.example.invalid\n", + "https://packages.example.invalid", + ), + ( + "export MARKER=1 PIP_INDEX_URL=https://packages.example.invalid/simple\n", + "https://packages.example.invalid/simple", + ), + ( + "export MARKER=1 PIP_EXTRA_INDEX_URL=https://packages.example.invalid/simple\n", + "https://packages.example.invalid/simple", + ), + ( + "export MARKER=1 CARGO_REGISTRIES_PRIVATE_INDEX=" + "sparse+https://packages.example.invalid/index\n", + "sparse+https://packages.example.invalid/index", + ), + ( + "env MARKER=1 npm config set registry https://packages.example.invalid\n", + "https://packages.example.invalid", + ), + ( + "sudo -E npm config set registry https://packages.example.invalid\n", + "https://packages.example.invalid", + ), + ( + "command -- npm config set registry https://packages.example.invalid\n", + "https://packages.example.invalid", + ), + ( + "( npm config set registry https://packages.example.invalid )\n", + "https://packages.example.invalid", + ), + ( + """cat > .npmrc < None: + """SC10 survives the complete static graph and every public report format.""" + (tmp_path / "SKILL.md").write_text( + "---\nname: dependency-source-test\n---\n# Dependency Source Test\n", + encoding="utf-8", + ) + (tmp_path / "setup.sh").write_text(script, encoding="utf-8") + + result = graph.invoke( + { + "skill_path": str(tmp_path), + "output_format": output_format, + "use_llm": False, + } + ) + + finding = next(item for item in result["findings"] if item.rule_id == "SC10") + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == expected_destination + rendered = ( + json.dumps(result["sarif_report"]) if output_format == "sarif" else result["report_body"] + ) + assert "SC10" in rendered + assert "packages.example.invalid" in rendered + + +@pytest.mark.parametrize("output_format", ["terminal", "json", "markdown", "sarif"]) +@pytest.mark.parametrize( + "script", + [ + "NPM_CONFIG_REGISTRY=https://registry.npmjs.org/ MARKER=1\n", + '"npm config set registry https://packages.example.invalid"\n', + """cat < None: + """Reviewed negative forms remain clear in every public report format.""" + (tmp_path / "SKILL.md").write_text( + "---\nname: dependency-source-negative-test\n---\n# Dependency Source Negative Test\n", + encoding="utf-8", + ) + (tmp_path / "setup.sh").write_text(script, encoding="utf-8") + + result = graph.invoke( + { + "skill_path": str(tmp_path), + "output_format": output_format, + "use_llm": False, + } + ) + + assert all(item.rule_id != "SC10" for item in result["findings"]) + rendered = ( + json.dumps(result["sarif_report"]) if output_format == "sarif" else result["report_body"] + ) + assert "SC10" not in rendered + + def test_graph_excludes_valid_oms_signature_from_static_findings(tmp_path: Path) -> None: """A real OMS signature remains inventoried without producing scan findings.""" fixture = Path(__file__).parents[1] / "fixtures" / "oms" / "mcore-split-pr.skill.oms.sig" diff --git a/tests/nodes/analyzers/test_dependency_sources.py b/tests/nodes/analyzers/test_dependency_sources.py new file mode 100644 index 00000000..eb52465d --- /dev/null +++ b/tests/nodes/analyzers/test_dependency_sources.py @@ -0,0 +1,1871 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Deterministic regression tests for dependency-source redirection.""" + +from __future__ import annotations + +import json + +import pytest + +from skillspector.dependency_sources import analyze_dependency_sources +from skillspector.llm_analyzer_base import Batch +from skillspector.models import Finding +from skillspector.nodes.meta_analyzer import ( + PER_FILE_ANALYSIS_PROMPT, + LLMMetaAnalyzer, + _fallback_filtered, + _passthrough_with_defaults, +) +from skillspector.nodes.report import report +from skillspector.state import SkillspectorState + + +def _analyze( + files: dict[str, str], metadata: list[dict[str, object]] | None = None +) -> list[Finding]: + return analyze_dependency_sources(sorted(files), files, metadata or []) + + +def test_generated_npm_and_yarn_configs_resolve_simple_local_indirection() -> None: + script = """#!/bin/sh +SOURCE_URL="https://packages.example.invalid" +cat > "$PROJECT/.npmrc" << EOF +registry=${SOURCE_URL} +EOF +cat > "$PROJECT/.yarnrc" << EOF +registry "${SOURCE_URL}" +EOF +""" + + findings = _analyze({"scripts/setup.sh": script}) + + assert [(finding.evidence["ecosystem"], finding.start_line) for finding in findings] == [ + ("npm", 4), + ("yarn", 7), + ] + assert all(finding.rule_id == "SC10" for finding in findings) + assert all(finding.severity == "HIGH" for finding in findings) + assert all(finding.evidence["operation"] == "replace" for finding in findings) + assert all( + finding.evidence["destination"] == "https://packages.example.invalid" + for finding in findings + ) + + +def test_supported_direct_configuration_surfaces_cover_all_ecosystems() -> None: + files = { + ".npmrc": "@team:registry=https://npm.example.invalid\n", + ".yarnrc.yml": ( + "npmScopes:\n team:\n npmRegistryServer: https://yarn.example.invalid\n" + ), + "pip.conf": ( + "[global]\n" + "index-url = https://python.example.invalid/simple\n" + "extra-index-url = https://extra.example.invalid/simple\n" + ), + "pyproject.toml": ( + "[[tool.poetry.source]]\n" + 'name = "mirror"\n' + 'url = "https://poetry.example.invalid/simple"\n' + ), + "settings.xml": ( + "all*" + "https://maven.example.invalid/repository" + "" + ), + ".cargo/config.toml": ( + '[source.crates-io]\nreplace-with = "mirror"\n' + '[source.mirror]\nregistry = "sparse+https://cargo.example.invalid/index"\n' + ), + } + + findings = _analyze(files) + + assert {finding.evidence["ecosystem"] for finding in findings} == { + "npm", + "yarn", + "pip", + "poetry", + "maven", + "cargo", + } + npm = next(finding for finding in findings if finding.evidence["ecosystem"] == "npm") + assert npm.evidence["scope"] == "@team" + yarn = next(finding for finding in findings if finding.evidence["ecosystem"] == "yarn") + assert yarn.evidence["scope"] == "team" + pip_operations = { + finding.evidence["operation"] + for finding in findings + if finding.evidence["ecosystem"] == "pip" + } + assert pip_operations == {"replace", "add"} + cargo = [finding for finding in findings if finding.evidence["ecosystem"] == "cargo"] + assert any(finding.evidence["operation"] == "replace" for finding in cargo) + + +def test_supported_command_and_environment_surfaces() -> None: + script = """#!/bin/sh +npm config set registry https://npm.example.invalid +yarn config set npmRegistryServer https://yarn.example.invalid +pip install --index-url https://pip.example.invalid/simple example +pip config set global.extra-index-url https://extra.example.invalid/simple +poetry source add private https://poetry.example.invalid/simple +mvn -Dmaven.repo.remote=https://maven.example.invalid/repo verify +export CARGO_REGISTRIES_PRIVATE_INDEX=sparse+https://cargo.example.invalid/index +""" + + findings = _analyze({"setup.sh": script}) + + assert {finding.evidence["ecosystem"] for finding in findings} == { + "npm", + "yarn", + "pip", + "poetry", + "maven", + "cargo", + } + assert all(finding.evidence["destination_status"] == "resolved" for finding in findings) + + +@pytest.mark.parametrize( + ("name", "destination", "ecosystem", "operation", "scope"), + [ + ( + "NPM_CONFIG_REGISTRY", + "https://npm.example.invalid", + "npm", + "replace", + "global", + ), + ( + "PIP_INDEX_URL", + "https://pip.example.invalid/simple", + "pip", + "replace", + "global", + ), + ( + "PIP_EXTRA_INDEX_URL", + "https://extra.example.invalid/simple", + "pip", + "add", + "global", + ), + ( + "CARGO_REGISTRIES_PRIVATE_INDEX", + "sparse+https://cargo.example.invalid/index", + "cargo", + "add", + "private", + ), + ], +) +@pytest.mark.parametrize( + "template", + [ + "MARKER=1 {name}={destination}", + "export MARKER=1 {name}={destination}", + "{name}={destination} MARKER=1", + "export {name}={destination} MARKER", + ], +) +def test_dependency_environment_variable_can_be_any_assignment_word( + name: str, + destination: str, + ecosystem: str, + operation: str, + scope: str, + template: str, +) -> None: + script = template.format(name=name, destination=destination) + "\n" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + finding = findings[0] + assert finding.rule_id == "SC10" + assert finding.severity == "HIGH" + assert finding.evidence["ecosystem"] == ecosystem + assert finding.evidence["operation"] == operation + assert finding.evidence["surface"] == "environment variable" + assert finding.evidence["scope"] == scope + assert finding.evidence["destination"] == destination + assert finding.evidence["destination_status"] == "resolved" + + +@pytest.mark.parametrize( + "script", + [ + "NPM_CONFIG_REGISTRY=https://registry.npmjs.org/ MARKER=1\n", + "export MARKER=1 NPM_CONFIG_REGISTRY=https://registry.npmjs.org/\n", + "PIP_INDEX_URL=https://pypi.org/simple/ MARKER=1\n", + "export MARKER=1 PIP_INDEX_URL=https://pypi.org/simple/\n", + ], +) +def test_canonical_environment_assignment_with_other_words_is_not_high(script: str) -> None: + assert _analyze({"setup.sh": script}) == [] + + +def test_multiple_dependency_environment_assignments_are_independent() -> None: + script = """export MARKER=1 NPM_CONFIG_REGISTRY=https://npm.example.invalid PIP_INDEX_URL=https://pip.example.invalid/simple +""" + + findings = _analyze({"setup.sh": script}) + + assert [finding.evidence["ecosystem"] for finding in findings] == ["npm", "pip"] + assert [finding.evidence["destination"] for finding in findings] == [ + "https://npm.example.invalid", + "https://pip.example.invalid/simple", + ] + + +def test_nonfirst_environment_assignment_resolves_prior_literal_value() -> None: + script = """SOURCE=https://packages.example.invalid +MARKER=1 NPM_CONFIG_REGISTRY=$SOURCE +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.evidence["destination"] == "https://packages.example.invalid" + assert finding.evidence["destination_status"] == "resolved" + + +def test_nonfirst_dynamic_environment_assignment_remains_unresolved() -> None: + finding = _analyze({"setup.sh": "MARKER=1 NPM_CONFIG_REGISTRY=$RUNTIME_SOURCE\n"})[0] + + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + assert finding.severity == "HIGH" + + +@pytest.mark.parametrize( + "script", + [ + "export NPM_CONFIG_REGISTRY=https://packages.example.invalid " + "NPM_CONFIG_REGISTRY=https://registry.npmjs.org/\n", + "env NPM_CONFIG_REGISTRY=https://packages.example.invalid " + "NPM_CONFIG_REGISTRY=https://registry.npmjs.org/ npm install\n", + ], +) +def test_last_duplicate_environment_assignment_takes_precedence(script: str) -> None: + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize( + "script", + [ + "export NPM_CONFIG_REGISTRY=https://registry.npmjs.org/ " + "NPM_CONFIG_REGISTRY=https://packages.example.invalid\n", + "env NPM_CONFIG_REGISTRY=https://registry.npmjs.org/ " + "NPM_CONFIG_REGISTRY=https://packages.example.invalid npm install\n", + ], +) +def test_last_noncanonical_duplicate_environment_assignment_is_high(script: str) -> None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize( + "script", + [ + "export MARKER NPM_CONFIG_REGISTRY=https://packages.example.invalid\n", + "export -- MARKER=1 NPM_CONFIG_REGISTRY=https://packages.example.invalid\n", + 'export MARKER=1 "NPM_CONFIG_REGISTRY=https://packages.example.invalid"\n', + ], +) +def test_export_assignment_operand_forms_are_detected(script: str) -> None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize( + "script", + [ + "MARKER=1 NPM_CONFIG_REGISTRY=$(printf %s https://packages.example.invalid)\n", + "export MARKER=1 NPM_CONFIG_REGISTRY=$(printf %s https://packages.example.invalid)\n", + "env NPM_CONFIG_REGISTRY=$(printf %s https://packages.example.invalid) npm install\n", + "SOURCE=https://registry.npmjs.org/\n" + r"MARKER=1 NPM_CONFIG_REGISTRY=\$SOURCE" + "\n", + ], +) +def test_complex_or_escaped_environment_values_remain_unresolved(script: str) -> None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["destination"] == "unresolved" + assert findings[0].evidence["destination_status"] == "unresolved" + + +@pytest.mark.parametrize( + "line", + [ + "NPM_CONFIG_REGISTRY='$SOURCE'", + "env NPM_CONFIG_REGISTRY='$SOURCE' npm install", + "npm config set registry '$SOURCE'", + ], +) +def test_single_quoted_dependency_source_variable_is_literal_and_unresolved(line: str) -> None: + script = f"SOURCE=https://registry.npmjs.org/\n{line}\n" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["destination"] == "unresolved" + assert findings[0].evidence["destination_status"] == "unresolved" + + +@pytest.mark.parametrize( + "line", + [ + 'NPM_CONFIG_REGISTRY="$SOURCE"', + 'env NPM_CONFIG_REGISTRY="$SOURCE" npm install', + 'npm config set registry "$SOURCE"', + ], +) +def test_double_quoted_dependency_source_variable_expands_statically(line: str) -> None: + script = f"SOURCE=https://registry.npmjs.org/\n{line}\n" + + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize( + "substitution", + [ + "$(printf %s https://packages.example.invalid | tr a-z A-Z)", + "$(printf %s https://packages.example.invalid; printf /simple)", + ], +) +def test_assignment_command_substitution_keeps_internal_control_operators( + substitution: str, +) -> None: + script = f"MARKER=1 NPM_CONFIG_REGISTRY={substitution}\n" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["destination"] == "unresolved" + assert findings[0].evidence["destination_status"] == "unresolved" + + +@pytest.mark.parametrize( + "script", + [ + "npm config set registry `printf %s https://packages.example.invalid | tr a-z A-Z`\n", + "MARKER=1 NPM_CONFIG_REGISTRY=" + "`printf %s https://packages.example.invalid; printf /simple`\n", + "env NPM_CONFIG_REGISTRY=" + "`printf %s https://packages.example.invalid | tr a-z A-Z` npm install\n", + ], +) +def test_backtick_substitution_keeps_internal_control_operators(script: str) -> None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["destination"] == "unresolved" + assert findings[0].evidence["destination_status"] == "unresolved" + + +def test_generated_configs_support_pip_poetry_maven_and_cargo() -> None: + script = """#!/bin/sh +cat > "$ROOT/pip.conf" << EOF +[global] +index-url = https://pip.example.invalid/simple +EOF +cat > "$ROOT/pyproject.toml" << EOF +[[tool.poetry.source]] +name = "private" +url = "https://poetry.example.invalid/simple" +EOF +cat > "$ROOT/settings.xml" << EOF +*https://maven.example.invalid/repo +EOF +cat > "$ROOT/.cargo/config.toml" << EOF +[registries.private] +index = "sparse+https://cargo.example.invalid/index" +EOF +""" + + findings = _analyze({"generate.sh": script}) + + assert {finding.evidence["ecosystem"] for finding in findings} == { + "pip", + "poetry", + "maven", + "cargo", + } + assert all( + str(finding.evidence["surface"]).startswith("generated") + or finding.evidence["ecosystem"] == "pip" + for finding in findings + ) + + +def test_canonical_defaults_do_not_produce_sc10() -> None: + files = { + ".npmrc": "registry=https://registry.npmjs.org/\n", + ".yarnrc": 'registry "https://registry.npmjs.org"\n', + "pip.conf": "[global]\nindex-url=https://pypi.org/simple/\n", + "pyproject.toml": ( + '[[tool.poetry.source]]\nname = "pypi"\nurl = "https://pypi.org/simple"\n' + ), + "settings.xml": ( + "" + "https://repo.maven.apache.org/maven2/" + "" + ), + ".cargo/config.toml": ( + '[source.crates-io]\nreplace-with = "canonical"\n' + '[source.canonical]\nregistry = "sparse+https://index.crates.io/"\n' + ), + } + + assert _analyze(files) == [] + + +@pytest.mark.parametrize("filename", [".yarnrc", ".yarnrc.yml"]) +def test_yarn_documented_public_default_does_not_produce_sc10(filename: str) -> None: + content = ( + 'registry "https://registry.yarnpkg.com"\n' + if filename == ".yarnrc" + else "npmRegistryServer: https://registry.yarnpkg.com\n" + ) + + assert _analyze({filename: content}) == [] + + +def test_variable_resolution_uses_assignment_visible_at_command_line() -> None: + script = """SRC=https://packages.example.invalid +npm config set registry "$SRC" +SRC=https://registry.npmjs.org/ +""" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].start_line == 2 + assert findings[0].evidence["destination"] == "https://packages.example.invalid" + + +def test_assignment_text_in_unrelated_heredoc_cannot_suppress_sc10() -> None: + script = """#!/bin/sh +SRC=https://packages.example.invalid +cat <<'EOF' > instructions.txt +SRC=https://registry.npmjs.org/ +EOF +npm config set registry "$SRC" +""" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].start_line == 6 + assert findings[0].evidence["destination"] == "https://packages.example.invalid" + + +def test_assignment_in_uncalled_function_cannot_suppress_sc10() -> None: + script = """#!/bin/sh +SRC=https://packages.example.invalid +configure_later() { + SRC=https://registry.npmjs.org/ +} +npm config set registry "$SRC" +""" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].start_line == 6 + assert findings[0].evidence["destination"] == "https://packages.example.invalid" + + +def test_assignment_in_split_line_function_declaration_cannot_suppress_sc10() -> None: + script = """#!/bin/sh +SRC=https://packages.example.invalid +configure_later() +{ + SRC=https://registry.npmjs.org/ +} +npm config set registry "$SRC" +""" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].start_line == 7 + assert findings[0].evidence["destination"] == "https://packages.example.invalid" + + +def test_called_function_assignment_keeps_possible_redirect_high() -> None: + script = """#!/bin/sh +SRC=https://registry.npmjs.org/ +use_private() { + SRC=https://packages.example.invalid +} +use_private +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 7 + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + + +def test_conditionally_called_function_keeps_possible_redirect_high() -> None: + script = """SRC=https://registry.npmjs.org/ +use_private() { SRC=https://packages.example.invalid; } +if test -f use-private; then use_private; fi +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 4 + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "unresolved" + + +@pytest.mark.parametrize( + "invocation", + [ + "if use_private; then :; fi", + "MARKER=1 use_private", + "{ use_private; }", + ], +) +def test_function_invocation_shapes_keep_possible_redirect_high(invocation: str) -> None: + script = f"""SRC=https://registry.npmjs.org/ +use_private() {{ SRC=https://packages.example.invalid; }} +{invocation} +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 4 + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + + +@pytest.mark.parametrize( + ("script", "ecosystem", "surface"), + [ + ( + "MARKER=1 npm config set registry https://packages.example.invalid\n", + "npm", + "npm config set", + ), + ( + "if :; then yarn config set registry https://packages.example.invalid; fi\n", + "yarn", + "yarn config set", + ), + ( + "{ pip install --index-url https://packages.example.invalid demo; }\n", + "pip", + "pip --index-url", + ), + ( + "while false; do pip config set global.index-url " + "https://packages.example.invalid; done\n", + "pip", + "pip config set", + ), + ( + "MARKER=1 pip install --extra-index-url https://packages.example.invalid demo\n", + "pip", + "pip --extra-index-url", + ), + ( + "{ pip config set global.extra-index-url https://packages.example.invalid; }\n", + "pip", + "pip config set", + ), + ( + "MARKER=1 poetry source add private https://packages.example.invalid\n", + "poetry", + "poetry source add", + ), + ( + "{ poetry config repositories.private https://packages.example.invalid; }\n", + "poetry", + "poetry config repositories", + ), + ( + "if :; then mvn -Dmaven.repo.remote=https://packages.example.invalid verify; fi\n", + "maven", + "Maven CLI repository", + ), + ], +) +def test_package_manager_commands_remain_detectable_in_shell_wrappers( + script: str, ecosystem: str, surface: str +) -> None: + finding = _analyze({"setup.sh": script})[0] + + assert finding.rule_id == "SC10" + assert finding.severity == "HIGH" + assert finding.evidence["ecosystem"] == ecosystem + assert finding.evidence["surface"] == surface + assert finding.evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize( + ("command", "ecosystem", "surface"), + [ + ( + "npm config set registry https://packages.example.invalid", + "npm", + "npm config set", + ), + ( + "yarn config set npmRegistryServer https://packages.example.invalid", + "yarn", + "yarn config set", + ), + ( + "python3 -m pip install --index-url https://packages.example.invalid demo", + "pip", + "pip --index-url", + ), + ( + "pip install --extra-index-url https://packages.example.invalid demo", + "pip", + "pip --extra-index-url", + ), + ( + "pip config set global.index-url https://packages.example.invalid", + "pip", + "pip config set", + ), + ( + "pip config set global.extra-index-url https://packages.example.invalid", + "pip", + "pip config set", + ), + ( + "poetry source add private https://packages.example.invalid", + "poetry", + "poetry source add", + ), + ( + "poetry config repositories.private https://packages.example.invalid", + "poetry", + "poetry config repositories", + ), + ( + "mvn -Dmaven.repo.remote=https://packages.example.invalid verify", + "maven", + "Maven CLI repository", + ), + ], +) +@pytest.mark.parametrize( + "wrapper", + [ + "env MARKER=1 {command}", + "sudo -E {command}", + "command -- {command}", + "( {command} )", + ], +) +def test_common_static_execution_wrappers_cover_every_command_family( + command: str, ecosystem: str, surface: str, wrapper: str +) -> None: + findings = _analyze({"setup.sh": wrapper.format(command=command) + "\n"}) + + assert len(findings) == 1 + finding = findings[0] + assert finding.rule_id == "SC10" + assert finding.severity == "HIGH" + assert finding.evidence["ecosystem"] == ecosystem + assert finding.evidence["surface"] == surface + assert finding.evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize( + "wrapper", + [ + "( {command}; )", + "( {command} ) >review.log", + "( {command} ) 2>/dev/null", + "( {command} ) >review.log 2>&1", + "( ( {command} ) )", + "( {command} && true )", + "( true && {command} )", + ], +) +def test_bounded_subshell_variants_preserve_actionable_command(wrapper: str) -> None: + command = "npm config set registry https://packages.example.invalid" + + findings = _analyze({"setup.sh": wrapper.format(command=command) + "\n"}) + + assert len(findings) == 1 + assert findings[0].evidence["surface"] == "npm config set" + assert findings[0].evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize( + "script", + [ + "env -i -u HOME MARKER=1 command -- npm config set registry " + "https://packages.example.invalid\n", + "sudo -u root -E -- npm config set registry https://packages.example.invalid\n", + "( sudo -E env -i MARKER=1 command -- npm config set registry " + "https://packages.example.invalid )\n", + ], +) +def test_nested_and_option_bearing_execution_wrappers_are_bounded(script: str) -> None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize( + ("script", "ecosystem"), + [ + ( + "env MARKER=1 NPM_CONFIG_REGISTRY=https://npm.example.invalid npm install\n", + "npm", + ), + ( + "env MARKER=1 PIP_INDEX_URL=https://pip.example.invalid/simple pip install demo\n", + "pip", + ), + ( + "env MARKER=1 CARGO_REGISTRIES_PRIVATE_INDEX=" + "sparse+https://cargo.example.invalid/index cargo build\n", + "cargo", + ), + ], +) +def test_env_wrapped_dependency_environment_assignments_are_detected( + script: str, ecosystem: str +) -> None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["ecosystem"] == ecosystem + assert findings[0].evidence["surface"] == "environment variable" + + +def test_python_module_pip3_uses_pip_environment_source() -> None: + script = ( + "env PIP_INDEX_URL=https://packages.example.invalid/simple python -m pip3 install demo\n" + ) + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["ecosystem"] == "pip" + assert findings[0].evidence["surface"] == "environment variable" + assert findings[0].evidence["destination"] == "https://packages.example.invalid/simple" + + +@pytest.mark.parametrize( + "script", + [ + "PIP_INDEX_URL=https://packages.example.invalid/simple " + "pip_index_url=https://pypi.org/simple/\n", + "env PIP_INDEX_URL=https://packages.example.invalid/simple " + "pip_index_url=https://pypi.org/simple/ pip install demo\n", + ], +) +def test_last_write_wins_only_for_exact_environment_variable_name(script: str) -> None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["ecosystem"] == "pip" + assert findings[0].evidence["destination"] == "https://packages.example.invalid/simple" + + +@pytest.mark.parametrize( + "script", + [ + "PIP_INDEX_URL=https://packages.example.invalid/simple env -i pip install demo\n", + "PIP_INDEX_URL=https://packages.example.invalid/simple " + "env --ignore-environment pip install demo\n", + "PIP_INDEX_URL=https://packages.example.invalid/simple " + "env -u PIP_INDEX_URL pip install demo\n", + "PIP_INDEX_URL=https://packages.example.invalid/simple " + "env --unset=PIP_INDEX_URL pip install demo\n", + "env PIP_INDEX_URL=https://packages.example.invalid/simple env -i pip install demo\n", + "env PIP_INDEX_URL=https://packages.example.invalid/simple " + "env --unset PIP_INDEX_URL pip install demo\n", + ], +) +def test_env_clear_and_unset_remove_accumulated_assignments(script: str) -> None: + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize( + "script", + [ + "PIP_INDEX_URL=https://pypi.org/simple env -i " + "PIP_INDEX_URL=https://packages.example.invalid/simple pip install demo\n", + "env PIP_INDEX_URL=https://pypi.org/simple env -u PIP_INDEX_URL " + "PIP_INDEX_URL=https://packages.example.invalid/simple pip install demo\n", + ], +) +def test_env_assignments_after_clear_or_unset_remain_effective(script: str) -> None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].evidence["destination"] == "https://packages.example.invalid/simple" + + +@pytest.mark.parametrize( + "script", + [ + "command -v npm config set registry https://packages.example.invalid\n", + "command -V npm config set registry https://packages.example.invalid\n", + "sudo -V npm config set registry https://packages.example.invalid\n", + "sudo -l npm config set registry https://packages.example.invalid\n", + "env -S 'npm config set registry https://packages.example.invalid'\n", + "( npm config set registry https://packages.example.invalid\n", + '"npm config set registry https://packages.example.invalid"\n', + "'npm config set registry https://packages.example.invalid'\n", + r"npm\ config\ set\ registry\ https://packages.example.invalid" + "\n", + '( "npm config set registry https://packages.example.invalid" )\n', + 'env "npm config set registry https://packages.example.invalid"\n', + "npm 'config set registry https://packages.example.invalid'\n", + "pip 'install --index-url https://packages.example.invalid demo'\n", + "poetry 'source add private https://packages.example.invalid'\n", + "mvn '-Dmaven.repo.remote=https://packages.example.invalid verify'\n", + '"NPM_CONFIG_REGISTRY=https://packages.example.invalid MARKER=1"\n', + "command NPM_CONFIG_REGISTRY=https://packages.example.invalid npm install\n", + "command -- NPM_CONFIG_REGISTRY=https://packages.example.invalid npm install\n", + ], +) +def test_nonexecuting_or_malformed_wrappers_are_not_actionable(script: str) -> None: + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize( + "script", + [ + "( npm config set registry https://packages.example.invalid ) arbitrary-tail\n", + "( npm config set registry https://packages.example.invalid ) >\n", + "( npm config set registry https://packages.example.invalid\n", + "( ( npm config set registry https://packages.example.invalid )\n", + "(( npm config set registry https://packages.example.invalid ))\n", + "( 'npm config set registry https://packages.example.invalid'; )\n", + "( npm 'config set registry https://packages.example.invalid'; )\n", + "( npm config set registry https://packages.example.invalid ) >review.log arbitrary-tail\n", + "echo `printf 'npm config set registry https://packages.example.invalid' | cat`\n", + ], +) +def test_malformed_or_inert_grouping_is_not_actionable(script: str) -> None: + assert _analyze({"setup.sh": script}) == [] + + +def test_wrapped_command_text_in_unrelated_heredoc_is_not_actionable() -> None: + script = """cat <<'EOF' +env MARKER=1 npm config set registry https://packages.example.invalid +sudo -E pip config set global.index-url https://packages.example.invalid +EOF +""" + + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize("output_format", ["terminal", "json", "markdown", "sarif"]) +def test_assignment_prefixed_command_is_preserved_in_all_reports(output_format: str) -> None: + finding = _analyze( + {"setup.sh": ("MARKER=1 npm config set registry https://packages.example.invalid\n")} + )[0] + state: SkillspectorState = { + "filtered_findings": [finding], + "component_metadata": [], + "has_executable_scripts": True, + "manifest": {}, + "output_format": output_format, + } + + result = report(state) + rendered = json.dumps(result.get("sarif_report", result.get("report_body", ""))) + + assert "SC10" in rendered + assert "packages.example.invalid" in rendered + + +@pytest.mark.parametrize("output_format", ["terminal", "json", "markdown", "sarif"]) +@pytest.mark.parametrize( + "script", + [ + "MARKER=1 NPM_CONFIG_REGISTRY=https://packages.example.invalid\n", + "env MARKER=1 npm config set registry https://packages.example.invalid\n", + "sudo -E npm config set registry https://packages.example.invalid\n", + "command -- npm config set registry https://packages.example.invalid\n", + "( npm config set registry https://packages.example.invalid )\n", + "cat > .npmrc < .npmrc < None: + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + state: SkillspectorState = { + "filtered_findings": findings, + "component_metadata": [], + "has_executable_scripts": True, + "manifest": {}, + "output_format": output_format, + } + result = report(state) + rendered = json.dumps(result.get("sarif_report", result.get("report_body", ""))) + + assert "SC10" in rendered + assert "packages.example.invalid" in rendered + + +@pytest.mark.parametrize("output_format", ["terminal", "json", "markdown", "sarif"]) +def test_nonfirst_environment_assignment_credentials_are_redacted_in_all_reports( + output_format: str, +) -> None: + username = "second-assignment-user-sentinel" + password = "second-assignment-password-sentinel" + token = "second-assignment-token-sentinel" + destination = f"https://{username}:{password}@packages.example.invalid/simple?token={token}" + findings = _analyze({"setup.sh": f"export MARKER=1 PIP_INDEX_URL={destination}\n"}) + + assert len(findings) == 1 + state: SkillspectorState = { + "filtered_findings": findings, + "component_metadata": [], + "has_executable_scripts": True, + "manifest": {}, + "output_format": output_format, + } + result = report(state) + rendered = json.dumps(result.get("sarif_report", result.get("report_body", ""))) + + for secret in (username, password, token): + assert secret not in json.dumps(findings[0].to_dict()) + assert secret not in rendered + assert "packages.example.invalid" in rendered + + +def test_assignment_in_case_arm_keeps_possible_redirect_high() -> None: + script = """SRC=https://registry.npmjs.org/ +case "$MODE" in + private) SRC=https://packages.example.invalid ;; +esac +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 5 + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + + +@pytest.mark.parametrize( + "case_body", + [ + "SRC=https://packages.example.invalid", + "use_private", + ], +) +def test_one_line_case_arm_keeps_possible_redirect_high(case_body: str) -> None: + function = ( + "use_private() { SRC=https://packages.example.invalid; }\n" + if case_body == "use_private" + else "" + ) + script = f"""MODE=private +SRC=https://registry.npmjs.org/ +{function}case "$MODE" in private) {case_body} ;; esac +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + + +def test_definite_assignment_after_one_line_case_clears_ambiguity() -> None: + script = """MODE=private +SRC=https://packages.example.invalid +case "$MODE" in private) SRC=https://other.example.invalid ;; esac +SRC=https://registry.npmjs.org/ +npm config set registry "$SRC" +""" + + assert _analyze({"setup.sh": script}) == [] + + +def test_assignment_shaped_command_cannot_override_real_assignment() -> None: + script = """SRC=https://packages.example.invalid +SRC = https://registry.npmjs.org/ || true +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "https://packages.example.invalid" + assert finding.evidence["destination_status"] == "resolved" + + +def test_export_assignment_remains_effective_with_trailing_variable_name() -> None: + script = """SRC=https://registry.npmjs.org/ +export SRC=https://packages.example.invalid MARKER +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "https://packages.example.invalid" + assert finding.evidence["destination_status"] == "resolved" + + +def test_multiple_assignment_words_update_each_variable() -> None: + script = """SRC=https://registry.npmjs.org/ +MARKER=1 SRC=https://packages.example.invalid +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 3 + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "https://packages.example.invalid" + assert finding.evidence["destination_status"] == "resolved" + + +def test_conditional_assignment_keeps_possible_noncanonical_redirect_high() -> None: + script = """#!/bin/sh +SRC=https://packages.example.invalid +if test -f use-default; then + SRC=https://registry.npmjs.org/ +fi +npm config set registry "$SRC" +""" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].start_line == 6 + assert findings[0].severity == "HIGH" + assert findings[0].evidence["destination"] == "unresolved" + assert findings[0].evidence["destination_status"] == "unresolved" + + +def test_inline_conditional_assignment_keeps_possible_redirect_high() -> None: + script = """SRC=https://registry.npmjs.org/ +if test -f use-private; then SRC=https://packages.example.invalid; fi +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 3 + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "unresolved" + + +@pytest.mark.parametrize("operator", ["&&", "||"]) +def test_short_circuit_assignment_keeps_possible_redirect_high(operator: str) -> None: + script = f"""SRC=https://registry.npmjs.org/ +test -f use-private {operator} SRC=https://packages.example.invalid +npm config set registry "$SRC" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 3 + assert finding.severity == "HIGH" + assert finding.evidence["destination"] == "unresolved" + + +def test_definite_assignment_after_inline_conditional_clears_ambiguity() -> None: + script = """SRC=https://packages.example.invalid +if test -f use-private; then SRC=https://other.example.invalid; fi +SRC=https://registry.npmjs.org/ +npm config set registry "$SRC" +""" + + assert _analyze({"setup.sh": script}) == [] + + +def test_single_prior_literal_assignment_resolves_statically() -> None: + script = """SRC=https://packages.example.invalid +npm config set registry "${SRC}" +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.evidence["destination"] == "https://packages.example.invalid" + assert finding.evidence["destination_status"] == "resolved" + + +@pytest.mark.parametrize( + "expression", + [ + "${SRC:-https://packages.example.invalid}", + "$(printf https://packages.example.invalid)", + "`printf https://packages.example.invalid`", + "$UNASSIGNED_SOURCE", + ], +) +def test_dynamic_or_unsupported_shell_expansions_remain_unresolved(expression: str) -> None: + finding = _analyze({"setup.sh": f"npm config set registry {expression}\n"})[0] + + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + assert finding.severity == "HIGH" + + +def test_unresolved_destination_is_high_trust_boundary_change() -> None: + script = """#!/bin/sh +cat > .npmrc << EOF +registry=${SOURCE_FROM_RUNTIME} +EOF +""" + + findings = _analyze({"setup.sh": script}) + + assert len(findings) == 1 + assert findings[0].severity == "HIGH" + assert findings[0].evidence["destination"] == "unresolved" + assert findings[0].evidence["destination_status"] == "unresolved" + + +def test_prose_comments_and_unrelated_registry_words_do_not_change_result() -> None: + docs = """# Package Registry Notes +The word registry appears here with https://packages.example.invalid. +```text +npm config set registry https://packages.example.invalid +``` +""" + script = """#!/bin/sh +# This audited internal registry is completely safe. +# npm config set registry https://comment.example.invalid +echo registry +""" + + assert _analyze({"README.md": docs, "setup.sh": script}) == [] + + +def test_actionable_shell_fence_is_analyzed_without_trusting_surrounding_prose() -> None: + markdown = """# Setup +This source is approved and audited. +```bash +npm config set registry https://packages.example.invalid +``` +""" + + findings = _analyze({"SKILL.md": markdown}) + + assert len(findings) == 1 + assert findings[0].evidence["ecosystem"] == "npm" + + +@pytest.mark.parametrize("output_format", ["terminal", "json", "markdown", "sarif"]) +def test_url_credentials_are_redacted_from_findings_and_all_reports(output_format: str) -> None: + username = "registry-user-sentinel" + password = "registry-password-sentinel" + query_token = "registry-token-sentinel" + content = ( + f"registry=https://{username}:{password}@packages.example.invalid/" + f"?token={query_token}&channel=stable\n" + ) + finding = _analyze({".npmrc": content})[0] + + serialized_finding = json.dumps(finding.to_dict()) + for secret in (username, password, query_token): + assert secret not in serialized_finding + assert "***@packages.example.invalid" in serialized_finding + + state: SkillspectorState = { + "filtered_findings": [finding], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "output_format": output_format, + } + rendered = report(state)["report_body"] + for secret in (username, password, query_token): + assert secret not in rendered + + +@pytest.mark.parametrize( + "destination", + [ + "ssh://registry-user-sentinel:registry-password-sentinel@packages.example.invalid/index?token=registry-token-sentinel", + "git+https://registry-user-sentinel:registry-password-sentinel@packages.example.invalid/index?token=registry-token-sentinel", + "sparse+https://registry-user-sentinel:registry-password-sentinel@packages.example.invalid/index?token=registry-token-sentinel", + ], +) +def test_cargo_url_credentials_are_redacted_for_supported_schemes(destination: str) -> None: + content = f'[registries.private]\nindex = "{destination}"\n' + + finding = _analyze({".cargo/config.toml": content})[0] + serialized = json.dumps(finding.to_dict()) + + for secret in ( + "registry-user-sentinel", + "registry-password-sentinel", + "registry-token-sentinel", + ): + assert secret not in serialized + assert "packages.example.invalid" in serialized + + +def test_sc10_credentials_are_redacted_before_provider_prompt_construction() -> None: + username = "provider-user-sentinel" + password = "provider-password-sentinel" + token = "provider-token-sentinel" + content = f"registry=ssh://{username}:{password}@packages.example.invalid/index?token={token}\n" + finding = _analyze({".npmrc": content})[0] + analyzer = LLMMetaAnalyzer.__new__(LLMMetaAnalyzer) + analyzer.base_prompt = PER_FILE_ANALYSIS_PROMPT + analyzer._input_budget = 100_000 + + batch = analyzer.get_batches([".npmrc"], {".npmrc": content}, [finding])[0] + prompt = analyzer.build_prompt(batch, metadata_text="No metadata available") + + for secret in (username, password, token): + assert secret not in batch.content + assert secret not in prompt + assert "packages.example.invalid" in prompt + + +def test_hidden_source_finding_is_marked_local_only() -> None: + findings = _analyze( + {".npmrc": "registry=https://packages.example.invalid\n"}, + [{"path": ".npmrc", "local_only": True}], + ) + + assert findings[0].evidence["local_only"] is True + assert "local-only" in findings[0].tags + + +def test_sc10_survives_optional_llm_filtering_when_unconfirmed() -> None: + content = "registry=https://packages.example.invalid\n" + finding = _analyze({".npmrc": content})[0] + batch = Batch(file_path=".npmrc", content=content, findings=[finding]) + analyzer = LLMMetaAnalyzer.__new__(LLMMetaAnalyzer) + + kept = analyzer.apply_filter([finding], [(batch, [])]) + + assert len(kept) == 1 + assert kept[0].rule_id == "SC10" + assert kept[0].severity == "HIGH" + assert kept[0] is finding + assert kept[0].tags == ["supply-chain", "dependency-source"] + + +def test_sc10_provider_confirmation_cannot_replace_deterministic_fields() -> None: + finding = _analyze({".npmrc": "registry=https://packages.example.invalid\n"})[0] + original = finding.to_dict() + batch = Batch(file_path=".npmrc", content="redacted", findings=[finding]) + provider_item = { + "pattern_id": "SC10", + "is_vulnerability": True, + "confidence": 0.6, + "start_line": finding.start_line, + "explanation": "provider alternate explanation", + "remediation": "provider alternate remediation", + "_file": ".npmrc", + } + analyzer = LLMMetaAnalyzer.__new__(LLMMetaAnalyzer) + + kept = analyzer.apply_filter([finding], [(batch, [provider_item])]) + + assert kept == [finding] + assert kept[0].to_dict() == original + assert kept[0].confidence == 1.0 + assert kept[0].message == finding.message + + +def test_sc10_static_only_and_provider_failure_paths_preserve_canonical_record() -> None: + finding = _analyze({".npmrc": "registry=https://packages.example.invalid\n"})[0] + + assert _fallback_filtered([finding]) == [finding] + assert _passthrough_with_defaults([finding]) == [finding] + + +def test_common_heredoc_redirection_order_is_detected_at_config_line() -> None: + script = """cat < .npmrc +registry=https://packages.example.invalid +EOF +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 2 + assert finding.evidence["surface"] == ".npmrc" + + +@pytest.mark.parametrize("delimiter", ["'END-OF'", "END-OF"]) +def test_hyphenated_heredoc_delimiter_is_detected(delimiter: str) -> None: + script = f"""cat > "$HOME/.npmrc" <<{delimiter} +registry=https://packages.example.invalid +END-OF +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 2 + assert finding.evidence["surface"] == ".npmrc" + assert finding.evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize("delimiter", ["END'-'OF", 'END"-"OF', r"END\-OF"]) +def test_word_quoted_heredoc_delimiter_generates_config(delimiter: str) -> None: + script = f"""cat > "$HOME/.npmrc" <<{delimiter} +registry=https://packages.example.invalid +END-OF +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 2 + assert finding.severity == "HIGH" + assert finding.evidence["surface"] == ".npmrc" + assert finding.evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize("delimiter", ["END'-'OF", 'END"-"OF', r"END\-OF"]) +def test_word_quoted_unrelated_heredoc_data_is_not_actionable(delimiter: str) -> None: + script = f"""cat <<{delimiter} > instructions.txt +npm config set registry https://packages.example.invalid +END-OF +""" + + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize( + ("target", "body", "ecosystem"), + [ + (".npmrc", "registry=https://packages.example.invalid", "npm"), + (".yarnrc", 'registry "https://packages.example.invalid"', "yarn"), + ( + "pip.conf", + "[global]\nindex-url=https://packages.example.invalid/simple", + "pip", + ), + ( + "pyproject.toml", + '[[tool.poetry.source]]\nname="private"\nurl="https://packages.example.invalid/simple"', + "poetry", + ), + ( + "settings.xml", + "*" + "https://packages.example.invalid/repository" + "", + "maven", + ), + ( + ".cargo/config.toml", + '[registries.private]\nindex="sparse+https://packages.example.invalid/index"', + "cargo", + ), + ], +) +def test_literal_dollar_heredoc_generates_every_supported_config( + target: str, body: str, ecosystem: str +) -> None: + script = f"cat > {target} <= 2 + assert findings[0].severity == "HIGH" + assert findings[0].evidence["ecosystem"] == ecosystem + assert "packages.example.invalid" in str(findings[0].evidence["destination"]) + + +@pytest.mark.parametrize( + "header", + [ + "tee instructions.txt < instructions.txt", + "cat <> instructions.txt", + "cat 3<&3", + ], +) +def test_literal_dollar_heredoc_data_is_not_actionable(header: str) -> None: + script = f"""{header} +npm config set registry https://packages.example.invalid +MARKER=1 PIP_INDEX_URL=https://packages.example.invalid/simple +END$OF +""" + + assert _analyze({"setup.sh": script}) == [] + + +def test_command_after_complete_literal_dollar_heredoc_is_actionable() -> None: + script = """cat < None: + script = """SRC=https://packages.example.invalid +cat < None: + script = """tee instructions.txt < .npmrc +registry=https://packages.example.invalid +EOF +END$OF +""" + + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize( + ("delimiter", "status"), + [ + ("END$OF", "resolved"), + ("'END$OF'", "unresolved"), + ('"END$OF"', "unresolved"), + (r"END\$OF", "unresolved"), + ("END'$'OF", "unresolved"), + ], +) +def test_literal_dollar_delimiter_preserves_expansion_semantics( + delimiter: str, status: str +) -> None: + script = f"""SOURCE=https://packages.example.invalid +cat > .npmrc <<{delimiter} +registry=$SOURCE +END$OF +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.evidence["destination_status"] == status + assert finding.evidence["destination"] == ( + "https://packages.example.invalid" if status == "resolved" else "unresolved" + ) + + +def test_tab_stripping_literal_dollar_heredoc_is_supported() -> None: + script = "cat > .npmrc <<-END$OF\n\tregistry=https://packages.example.invalid\n\tEND$OF\n" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 2 + assert finding.evidence["destination"] == "https://packages.example.invalid" + + +@pytest.mark.parametrize("delimiter", ["END${OF}", "END$?", "END#OF", "END!OF"]) +def test_static_punctuation_heredoc_words_are_literal_and_inert(delimiter: str) -> None: + script = f"""cat <<{delimiter} +npm config set registry https://packages.example.invalid +{delimiter} +""" + + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize("delimiter", ["END${OF}", "END$?", "END#OF", "END!OF"]) +def test_static_punctuation_heredoc_words_generate_config(delimiter: str) -> None: + script = f"""cat > .npmrc <<{delimiter} +registry=https://packages.example.invalid +{delimiter} +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 2 + assert finding.evidence["destination"] == "https://packages.example.invalid" + + +def test_bare_braced_dollar_heredoc_keeps_body_expansion_enabled() -> None: + script = """SOURCE=https://packages.example.invalid +cat > .npmrc < None: + script = """SOURCE=https://packages.example.invalid +cat > .npmrc <<$'END$OF' +registry=$SOURCE +END$OF +""" + + finding = _analyze({"setup.bash": script})[0] + + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + + +def test_locale_quoted_heredoc_word_is_inert_and_disables_expansion() -> None: + script = """SOURCE=https://packages.example.invalid +cat > .npmrc <<$"END$OF" +registry=$SOURCE +END$OF +""" + + finding = _analyze({"setup.bash": script})[0] + + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + + +def test_heredoc_inside_quoted_command_substitution_is_inert() -> None: + script = """value="$(cat < None: + script = """cat > .npmrc < None: + script = """cat > .npmrc < None: + script = """cat > .npmrc <", True), + (">>", True), + ("1>", True), + ("1>>", True), + ("2>", False), + ("3>>", False), + ], +) +def test_generated_config_requires_stdout_file_redirect(redirect: str, expected: bool) -> None: + script = f"""cat {redirect} .npmrc < None: + inert = """cat > .npmrc > instructions.txt < instructions.txt >> .npmrc < None: + script = f"""cat {operands} > .npmrc < None: + overridden = """cat > .npmrc < .npmrc < existing.txt < None: + script = f"""{arithmetic} +npm config set registry https://packages.example.invalid +2 +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 2 + assert finding.evidence["destination"] == "https://packages.example.invalid" + + +def test_generated_heredoc_header_scan_is_bounded_on_one_long_line() -> None: + script = "cat " + "x>" * 20_000 + " no-redirection-target\n" + + assert _analyze({"setup.sh": script}) == [] + + +def test_hyphenated_generic_heredoc_does_not_hide_later_command() -> None: + script = """cat < instructions.txt +not executable +END-OF +npm config set registry https://packages.example.invalid +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 4 + assert finding.evidence["surface"] == "npm config set" + + +def test_unterminated_literal_dollar_heredoc_does_not_hide_later_command() -> None: + script = """cat < instructions.txt +not executable +npm config set registry https://packages.example.invalid +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 3 + assert finding.evidence["surface"] == "npm config set" + + +def test_mismatched_literal_dollar_terminator_does_not_hide_later_command() -> None: + script = """cat < instructions.txt +not executable +ENDOF +npm config set registry https://packages.example.invalid +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 4 + assert finding.evidence["surface"] == "npm config set" + + +def test_dynamic_heredoc_word_is_not_partially_accepted() -> None: + script = """cat < instructions.txt +npm config set registry https://packages.example.invalid +END$ +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 2 + assert finding.evidence["surface"] == "npm config set" + + +def test_unmatched_word_quote_does_not_partially_consume_later_command() -> None: + script = """cat < instructions.txt +not executable +npm config set registry https://packages.example.invalid +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 3 + assert finding.evidence["surface"] == "npm config set" + + +def test_command_text_in_unrelated_heredoc_is_not_actionable() -> None: + script = """#!/bin/sh +cat <<'EOF' > instructions.txt +npm config set registry https://packages.example.invalid +EOF +""" + + assert _analyze({"setup.sh": script}) == [] + + +@pytest.mark.parametrize( + "header", + [ + "tee instructions.txt <<'EOF'", + "cat <<'EOF'", + "cat <<'EOF' >> instructions.txt", + "cat 3<<'EOF' 1>&3", + ], +) +def test_command_text_in_generic_heredoc_is_not_actionable(header: str) -> None: + script = f"{header}\nnpm config set registry https://packages.example.invalid\nEOF\n" + + assert _analyze({"setup.sh": script}) == [] + + +def test_generated_config_text_nested_in_unrelated_heredoc_is_not_actionable() -> None: + script = """tee instructions.txt <<'OUTER' +cat < .npmrc +registry=https://packages.example.invalid +EOF +OUTER +""" + + assert _analyze({"setup.sh": script}) == [] + + +def test_dependency_source_command_in_pipeline_stage_is_actionable() -> None: + script = "printf y | npm config set registry https://packages.example.invalid\n" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 1 + assert finding.severity == "HIGH" + assert finding.evidence["ecosystem"] == "npm" + assert finding.evidence["destination"] == "https://packages.example.invalid" + + +def test_quoted_heredoc_delimiter_does_not_expand_variables() -> None: + script = """SOURCE=https://packages.example.invalid +cat <<'EOF' > .npmrc +registry=${SOURCE} +EOF +""" + + finding = _analyze({"setup.sh": script})[0] + + assert finding.start_line == 3 + assert finding.evidence["destination"] == "unresolved" + assert finding.evidence["destination_status"] == "unresolved" + + +@pytest.mark.parametrize("delimiter", ["EOF", "END$OF"]) +def test_repeated_unmatched_heredocs_are_bounded_and_do_not_produce_sc10( + delimiter: str, +) -> None: + script = "\n".join(f"cat <<{delimiter} > .npmrc" for _ in range(2_000)) + + assert _analyze({"setup.sh": script}) == [] + + +def test_echoed_and_source_language_command_text_is_not_actionable() -> None: + destination = "https://packages.example.invalid" + files = { + "setup.sh": f"echo npm config set registry {destination}\n", + "example.py": f'command = "npm config set registry {destination}"\n', + "example.js": f'const command = "npm config set registry {destination}";\n', + } + + assert _analyze(files) == [] + + +def test_pip_short_index_option_is_detected() -> None: + finding = _analyze( + {"setup.sh": "pip install -i https://packages.example.invalid/simple package-name\n"} + )[0] + + assert finding.evidence["ecosystem"] == "pip" + assert finding.evidence["operation"] == "replace" + + +def test_extensionless_executable_shell_script_is_actionable() -> None: + content = "#!/bin/sh\nnpm config set registry https://packages.example.invalid\n" + metadata = [{"path": "bootstrap", "executable": True}] + + finding = _analyze({"bootstrap": content}, metadata)[0] + + assert finding.start_line == 2 + assert finding.evidence["ecosystem"] == "npm" diff --git a/tests/nodes/test_report_sanitizer.py b/tests/nodes/test_report_sanitizer.py index 0f2b5ba1..a6d66dbd 100644 --- a/tests/nodes/test_report_sanitizer.py +++ b/tests/nodes/test_report_sanitizer.py @@ -17,6 +17,8 @@ from __future__ import annotations +import json + import pytest from skillspector.models import Finding @@ -74,3 +76,42 @@ def test_report_emits_clean_utf8_for_all_formats(fmt: str) -> None: assert "\x1b" not in body, f"ESC leaked into {fmt}" # The readable content survives the sanitization. assert "leak" in body and "here" in body + + +@pytest.mark.parametrize("fmt", ["markdown", "json", "sarif", "terminal"]) +@pytest.mark.parametrize("scheme", ["https", "ssh", "git+https", "sparse+https"]) +def test_report_redacts_url_credentials_from_every_finding_field(fmt: str, scheme: str) -> None: + username = "output-user-sentinel" + password = "output-password-sentinel" + token = "output-token-sentinel" + url = f"{scheme}://{username}:{password}@packages.example.invalid/?token={token}" + finding = Finding( + rule_id="E2", + message=f"credential-bearing destination {url}", + severity="HIGH", + confidence=0.9, + file="setup.sh", + start_line=1, + finding=url, + explanation=url, + remediation=url, + context=url, + matched_text=url, + code_snippet=url, + evidence={"destination": url}, + ) + state: SkillspectorState = { + "filtered_findings": [finding], + "component_metadata": [], + "has_executable_scripts": False, + "manifest": {}, + "skill_path": None, + "output_format": fmt, + } + + result = report(state) + rendered = result["report_body"] + serialized_findings = json.dumps([item.to_dict() for item in result["filtered_findings"]]) + for secret in (username, password, token): + assert secret not in rendered + assert secret not in serialized_findings