diff --git a/CHANGELOG.md b/CHANGELOG.md index 339c287a..973ce79e 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.5 (Friday, August 14, 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..e1b5d4f2 --- /dev/null +++ b/src/skillspector/dependency_sources.py @@ -0,0 +1,1168 @@ +# 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 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"\$(?:\{(?P[A-Za-z_][A-Za-z0-9_]*)\}|(?P[A-Za-z_][A-Za-z0-9_]*))" +) +_ASSIGNMENT_RE = re.compile( + r"^\s*(?:export\s+)?(?P[A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?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 + start_line: int + end_line: int + expand_variables: bool + complete: bool + + +_HEREDOC_TARGET = r'(?P"[^"]+"|\'[^\']+\'|[^\s;]+)' +_HEREDOC_DELIMITER = r"(?P['\"]?)(?P[A-Za-z_][A-Za-z0-9_]*)(?P=quote)" +_SHELL_HEREDOC_OPERATOR = re.compile( + r"<<(?P-?)(?!<)\s*(?:" + r"(?P['\"])(?P[^'\"]+)(?P=quote)|" + r"\\?(?P[A-Za-z_][A-Za-z0-9_]*)" + r")" +) +_HEREDOC_HEADERS = ( + re.compile( + rf"^\s*cat\b.*?(?])>(?!>)\s*{_HEREDOC_TARGET}\s*" + rf"<<(?P-?)\s*{_HEREDOC_DELIMITER}" + ), + re.compile( + rf"^\s*cat\b.*?<<(?P-?)\s*{_HEREDOC_DELIMITER}\s*" + rf"(?])>(?!>)\s*{_HEREDOC_TARGET}" + ), +) + + +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 _assignment_match(segment: str) -> re.Match[str] | None: + """Return an assignment occupying one bounded shell command segment.""" + candidate = segment.rsplit("{", 1)[-1].strip().removesuffix("}").strip() + keyword = re.match(r"^(?:then|do|else)\b\s*(?P.*)$", candidate) + if keyword: + candidate = keyword.group("rest") + return _ASSIGNMENT_RE.match(candidate) + + +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 = _assignment_match(segment) + if assignment: + assigned_names.add(assignment.group("name")) + 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 = segment.strip() + if re.match(r"^(?:fi|done|esac)\b", stripped): + control_depth = max(0, control_depth - 1) + continue + if re.match(r"^(?:if|case|for|while|until|select)\b", stripped): + control_depth += 1 + continue + + match = _assignment_match(stripped) + if match: + value = _strip_shell_comment(match.group("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(match.group("name"), []).append( + (line_number, resolved_value) + ) + continue + + call_candidate = re.sub(r"^(?:then|do|else)\b\s*", "", stripped) + 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(";,)") + if len(resolved) >= 2 and resolved[0] == resolved[-1] and resolved[0] in {'"', "'"}: + 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 _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 _heredocs(content: str) -> list[_HeredocRegion]: + """Return bounded, linearly parsed generated-configuration heredocs.""" + lines = content.splitlines() + regions: list[_HeredocRegion] = [] + index = 0 + while index < len(lines): + match = next( + ( + candidate + for pattern in _HEREDOC_HEADERS + if (candidate := pattern.search(lines[index])) is not None + ), + None, + ) + if not match: + index += 1 + continue + delimiter = match.group("delimiter") + strip_tabs = match.group("strip_tabs") == "-" + end = index + 1 + while end < len(lines): + terminator = lines[end].lstrip("\t") if strip_tabs else lines[end] + if terminator == delimiter: + break + end += 1 + complete = end < len(lines) + body_lines = lines[index + 1 : end] + if strip_tabs: + body_lines = [line.lstrip("\t") for line in body_lines] + regions.append( + _HeredocRegion( + target=match.group("target").strip("'\""), + body="\n".join(body_lines), + start_line=index + 2, + end_line=end + 1 if complete else len(lines), + expand_variables=not bool(match.group("quote")), + complete=complete, + ) + ) + if not complete: + # An unmatched heredoc consumes the remaining shell input. Stopping + # here both reflects that ambiguity and prevents repeated O(n) scans. + break + index = end + 1 + return regions + + +def _shell_heredoc_specs(line: str) -> list[tuple[str, bool]]: + """Return unquoted heredoc delimiters declared by one shell command line.""" + specs: list[tuple[str, bool]] = [] + quote: str | None = None + escaped = False + index = 0 + while index < len(line): + character = line[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 None and character == "#" and (index == 0 or line[index - 1].isspace()): + break + if ( + quote is None + and line[index : index + 2] == "<<" + and (index == 0 or line[index - 1] != "<") + ): + match = _SHELL_HEREDOC_OPERATOR.match(line, index) + if match: + delimiter = match.group("quoted") or match.group("bare") or "" + specs.append((delimiter, match.group("strip_tabs") == "-")) + index = match.end() + continue + index += 1 + return specs + + +def _heredoc_data_lines(content: str) -> set[int]: + """Return all shell heredoc body and terminator lines in one bounded pass.""" + lines = content.splitlines() + data_lines: set[int] = set() + index = 0 + while index < len(lines): + specs = _shell_heredoc_specs(lines[index]) + if not specs: + index += 1 + continue + body_index = index + 1 + for delimiter, strip_tabs in specs: + end = body_index + while end < len(lines): + terminator = lines[end].lstrip("\t") if strip_tabs else lines[end] + if terminator == delimiter: + break + end += 1 + data_lines.update(range(body_index + 1, min(end + 2, len(lines) + 1))) + if end >= len(lines): + return data_lines + body_index = end + 1 + index = body_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.start_line - 1 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 + 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 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 (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.""" + return [segment for _, segment in _shell_parts(line)] + + +def _parse_commands(content: str, file: str, assignments: Assignments) -> list[SourceChange]: + changes: list[SourceChange] = [] + heredoc_data_lines = _heredoc_data_lines(content) + command_prefix = r"^\s*(?:[$>]\s+)?(?:(?:command|sudo)\s+)?" + patterns: tuple[tuple[str, str, str, str, re.Pattern[str]], ...] = ( + ( + "npm", + "replace", + "npm config set", + "scope", + re.compile( + command_prefix + + r"npm\s+config\s+set\s+(?P@[\w.-]+:)?registry\s+(?P\S+)", + re.I, + ), + ), + ( + "yarn", + "replace", + "yarn config set", + "scope", + re.compile( + command_prefix + + r"yarn\s+config\s+set\s+(?:registry|npmRegistryServer)\s+(?P\S+)", + re.I, + ), + ), + ( + "pip", + "replace", + "pip --index-url", + "none", + re.compile( + command_prefix + + r"(?:python(?:3)?\s+-m\s+)?pip(?:3)?\b[^\n]*?" + + r"(?:--index-url|-i)(?:=|\s+)(?P\S+)", + re.I, + ), + ), + ( + "pip", + "add", + "pip --extra-index-url", + "none", + re.compile( + command_prefix + + r"(?:python(?:3)?\s+-m\s+)?pip(?:3)?\b[^\n]*?" + + r"--extra-index-url(?:=|\s+)(?P\S+)", + re.I, + ), + ), + ( + "pip", + "replace", + "pip config set", + "none", + re.compile( + command_prefix + + r"pip(?:3)?\s+config\s+set\s+(?:global\.)?index-url\s+(?P\S+)", + re.I, + ), + ), + ( + "pip", + "add", + "pip config set", + "none", + re.compile( + command_prefix + + r"pip(?:3)?\s+config\s+set\s+(?:global\.)?extra-index-url\s+(?P\S+)", + re.I, + ), + ), + ( + "poetry", + "add", + "poetry source add", + "poetry", + re.compile( + command_prefix + + r"poetry\s+source\s+add(?:\s+--\S+)*\s+" + + r"(?P[\w.-]+)\s+(?P\S+)", + re.I, + ), + ), + ( + "poetry", + "add", + "poetry config repositories", + "poetry", + re.compile( + command_prefix + + r"poetry\s+config\s+repositories\." + + r"(?P[\w.-]+)\s+(?P\S+)", + re.I, + ), + ), + ( + "maven", + "replace", + "Maven CLI repository", + "none", + re.compile( + command_prefix + r"mvn\b[^\n]*?-Dmaven\.repo\.remote=(?P\S+)", + re.I, + ), + ), + ) + for line_number, line in enumerate(content.splitlines(), 1): + if line_number in heredoc_data_lines: + continue + for segment in _shell_segments(line): + for ecosystem, operation, surface, scope_mode, pattern in patterns: + match = pattern.search(segment) + if not match: + continue + scope = match.groupdict().get("scope") if scope_mode != "none" else None + if scope: + scope = scope.rstrip(":") + _add_change( + changes, + ecosystem=ecosystem, + operation=operation, + surface=surface, + scope=scope, + raw_destination=match.group("dest"), + file=file, + line=line_number, + matched_text=line, + assignments=assignments, + ) + + env_match = re.match( + r"\s*(?:export\s+)?(?PNPM_CONFIG_REGISTRY|PIP_INDEX_URL|PIP_EXTRA_INDEX_URL|CARGO_REGISTRIES_[A-Za-z0-9_]+_INDEX)\s*=\s*(?P.+)$", + segment, + re.I, + ) + if env_match: + name = env_match.group("name").upper() + if name == "NPM_CONFIG_REGISTRY": + ecosystem, operation, scope = "npm", "replace", None + elif name == "PIP_INDEX_URL": + ecosystem, operation, scope = "pip", "replace", None + elif name == "PIP_EXTRA_INDEX_URL": + ecosystem, operation, scope = "pip", "add", None + else: + ecosystem, operation = "cargo", "add" + scope = name.removeprefix("CARGO_REGISTRIES_").removesuffix("_INDEX").lower() + _add_change( + changes, + ecosystem=ecosystem, + operation=operation, + surface="environment variable", + scope=scope, + raw_destination=env_match.group("dest"), + 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 0fbf8d23..f669f5d6 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/src/skillspector/llm_analyzer_base.py @@ -39,6 +39,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, @@ -481,7 +482,7 @@ def __init__(self, base_prompt: str, model: str, *, node: str = "llm_analyzer"): ) @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 8e0d6664..962785b1 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 7e3738eb..aafdc995 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. @@ -39,6 +40,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 @@ -1336,7 +1338,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"] @@ -1435,6 +1437,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 d0bc3e1b..aa377e03 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/nodes/analyzers/test_dependency_sources.py b/tests/nodes/analyzers/test_dependency_sources.py new file mode 100644 index 00000000..a4afbffc --- /dev/null +++ b/tests/nodes/analyzers/test_dependency_sources.py @@ -0,0 +1,653 @@ +# 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) + + +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" + + +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" + + +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" + + +def test_repeated_unmatched_heredocs_are_bounded_and_do_not_produce_sc10() -> None: + script = "\n".join("cat < .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