diff --git a/README.md b/README.md index 3c1b4d70..e49ed011 100644 --- a/README.md +++ b/README.md @@ -291,6 +291,12 @@ GuardDog's behavior can be customized using environment variables: | `GUARDDOG_TOP_PACKAGES_CACHE_LOCATION` | Location of the top packages cache directory | `guarddog/analyzer/metadata/resources` | | `GUARDDOG_YARA_EXT_EXCLUDE` | Comma-separated list of file extensions to exclude from YARA scanning | `ini,md,rst,txt,lock,json,yaml,yml,toml,xml,html,csv,sql,pdf,doc,docx,ppt,pptx,xls,xlsx,odt,changelog,readme,makefile,dockerfile,pkg-info,d.ts` | +#### Metadata Rule Configuration + +| Environment Variable | Description | Default Value | +|---------------------|-------------|---------------| +| `GUARDDOG_NEW_DEPENDENCY_RISK_THRESHOLD` | Minimum risk score for a newly introduced dependency to flag the parent package in the `risky_new_dependency` rule | `5.0` | + #### Archive Extraction Security Limits GuardDog implements multiple security checks when extracting package archives to protect against compression bombs and file descriptor exhaustion attacks: diff --git a/RULES.md b/RULES.md index ecde0a89..b65d97b4 100644 --- a/RULES.md +++ b/RULES.md @@ -86,5 +86,6 @@ Rules are categorized by their `identifies` field which determines how they part | deceptive_author | `threat.metadata.deceptive-author` | This heuristic detects when an author is using a disposable email | medium | initial-access | :white_check_mark: | :white_check_mark: | | | | | | metadata_mismatch | `threat.metadata.manifest-mismatch` | Identify packages with mismatches between registry metadata and the actual package manifest | medium | execution | :white_check_mark: | :white_check_mark: | | | | | | direct_url_dependency | `threat.metadata.direct-url-dep` | Identify packages with direct URL dependencies. Dependencies fetched this way are not immutable and can be used to inject untrusted code or reduce the likelihood of a reproducible install. | medium | initial-access | | :white_check_mark: | | | | | +| risky_new_dependency | `threat.npm.risky-new-dependency` | Identify newly added dependencies that are themselves risky. A dependency introduced in this version but absent from the previous one is scanned as a package; it is flagged when its risk score is high. | high | initial-access | | :white_check_mark: | | | | | diff --git a/guarddog/analyzer/metadata/npm/__init__.py b/guarddog/analyzer/metadata/npm/__init__.py index faf0890d..76a6c688 100644 --- a/guarddog/analyzer/metadata/npm/__init__.py +++ b/guarddog/analyzer/metadata/npm/__init__.py @@ -11,6 +11,9 @@ from guarddog.analyzer.metadata.npm.metadata_mismatch import NPMMetadataMismatchDetector from guarddog.analyzer.metadata.npm.bundled_binary import NPMBundledBinary from guarddog.analyzer.metadata.npm.deceptive_author import NPMDeceptiveAuthor +from guarddog.analyzer.metadata.npm.risky_new_dependency import ( + NPMRiskyNewDependencyDetector, +) NPM_METADATA_RULES = {} @@ -22,6 +25,7 @@ NPMMetadataMismatchDetector, NPMBundledBinary, NPMDeceptiveAuthor, + NPMRiskyNewDependencyDetector, ] for detectorClass in classes: diff --git a/guarddog/analyzer/metadata/npm/risky_new_dependency.py b/guarddog/analyzer/metadata/npm/risky_new_dependency.py new file mode 100644 index 00000000..637e405d --- /dev/null +++ b/guarddog/analyzer/metadata/npm/risky_new_dependency.py @@ -0,0 +1,283 @@ +"""Risky New Dependency Detector + +When a new version of a package adds a dependency that was not present in the +previous published version, that dependency is scanned like any other package. +If it scores at or above the risk threshold, the parent package is flagged so a +maliciously introduced dependency surfaces on the parent scan. +""" + +import json +import logging +import os +import subprocess +import sys +from dataclasses import dataclass, field +from typing import List, Optional + +from guarddog.analyzer.metadata.detector import Detector +from guarddog.analyzer.risk_engine import RiskLabel +from guarddog.utils.config import NEW_DEPENDENCY_RISK_THRESHOLD +from guarddog.utils.npm import highest_matching_version, resolve_npm_alias + +log = logging.getLogger("guarddog") + + +@dataclass +class DependencyRisk: + """Outcome of scanning a single newly added dependency.""" + + version: str + score: float + label: str + matched_rules: List[str] = field(default_factory=list) + + +# Upper bound on a single sub-dependency scan so one slow scan can't hang the parent. +SUBSCAN_TIMEOUT_SECONDS = 300 + +# Keys in the registry `time` object that are not version publish timestamps. +_NON_VERSION_TIME_KEYS = {"created", "modified"} + +# Risk labels phrased to match GuardDog's overall assessment wording. +_LABEL_PHRASE = { + RiskLabel.HIGH_RISK.value: "high risk", + RiskLabel.SUSPICIOUS.value: "suspicious", + RiskLabel.LOW.value: "low risk", +} + + +class NPMRiskyNewDependencyDetector(Detector): + """Detects dependencies newly introduced in a version that are themselves risky. + + The previous published version is determined from the registry `time` map; any + dependency name present in the scanned version but absent from the previous one + is scanned as a standalone package. Each sub-scan runs as a subprocess invocation + of guarddog, so it is sandboxed identically to the parent, and excludes this rule + so the check never recurses beyond one level.""" + + def __init__(self): + super().__init__( + name="risky_new_dependency", + description="Identify newly added dependencies that are themselves risky. " + "A dependency introduced in this version but absent from the previous one " + "is scanned as a package; it is flagged when its risk score is high.", + identifies="threat.npm.risky-new-dependency", + severity="high", + mitre_tactics="initial-access", + specificity="high", + sophistication="low", + ) + + def detect( + self, + package_info, + path: Optional[str] = None, + name: Optional[str] = None, + version: Optional[str] = None, + ) -> tuple[bool, Optional[str]]: + package_name = name or package_info.get("name", "") + versions = package_info.get("versions", {}) + current_version = version or package_info.get("dist-tags", {}).get("latest") + if not current_version or current_version not in versions: + log.debug( + f"[{self.name}] No usable version for '{package_name}' " + f"(resolved '{current_version}'); skipping" + ) + return False, None + + previous_version = self._previous_published_version( + package_info, current_version + ) + if previous_version is None: + log.debug( + f"[{self.name}] '{package_name}@{current_version}' has no previous " + f"published version; skipping" + ) + return False, None + + current_deps = self._installed_dependencies(versions.get(current_version, {})) + previous_deps = self._installed_dependencies(versions.get(previous_version, {})) + new_dependencies = set(current_deps) - set(previous_deps) + if not new_dependencies: + log.debug( + f"[{self.name}] '{package_name}@{current_version}' adds no new " + f"dependencies vs '{previous_version}'; skipping" + ) + return False, None + + log.debug( + f"[{self.name}] '{package_name}@{current_version}' adds " + f"{len(new_dependencies)} new dependency(ies) vs '{previous_version}': " + f"{', '.join(sorted(new_dependencies))}" + ) + + findings = [] + for dep_name in sorted(new_dependencies): + risk = self._scan_dependency(dep_name, current_deps[dep_name]) + if risk is None: + log.debug(f"[{self.name}] Could not score new dependency '{dep_name}'") + continue + flagged = risk.score >= NEW_DEPENDENCY_RISK_THRESHOLD + log.debug( + f"[{self.name}] New dependency '{dep_name}@{risk.version}' scored " + f"{risk.score} ({risk.label}); threshold {NEW_DEPENDENCY_RISK_THRESHOLD} " + f"-> {'FLAGGED' if flagged else 'below threshold'}; " + f"matched rules: {', '.join(risk.matched_rules) or 'none'}" + ) + if flagged: + matched = ( + f" Matched rules: {', '.join(risk.matched_rules)}." + if risk.matched_rules + else "" + ) + phrase = _LABEL_PHRASE.get(risk.label, risk.label or "risky") + findings.append( + f"Newly added dependency {dep_name}@{risk.version} is {phrase} " + f"(risk score {risk.score}/10).{matched} It was introduced in " + f"{package_name}@{current_version} and was not a dependency of " + f"the previous version {previous_version}." + ) + + log.debug( + f"[{self.name}] '{package_name}@{current_version}': " + f"{len(findings)} risky new dependency(ies) found" + ) + return len(findings) != 0, "\n".join(findings) + + def _previous_published_version( + self, package_info, current_version: str + ) -> Optional[str]: + """Return the version published immediately before `current_version`. + + Publish times come from the registry `time` map; ISO 8601 timestamps sort + lexicographically, so the previous version is the one with the greatest + timestamp strictly before the current version's. + """ + published = { + v: t + for v, t in package_info.get("time", {}).items() + if v not in _NON_VERSION_TIME_KEYS and v in package_info.get("versions", {}) + } + current_time = published.get(current_version) + if current_time is None: + return None + + earlier = [(t, v) for v, t in published.items() if t < current_time] + if not earlier: + return None + return max(earlier)[1] + + @staticmethod + def _installed_dependencies(version_info: dict) -> dict: + """Map real package name -> version selector for the dependencies npm + installs by default: `dependencies` and `optionalDependencies` (optional + installs are non-fatal but still run). npm aliases are resolved to the real + package so the diff and sub-scan target the aliased package, not the local + alias name (e.g. "x": "npm:evil@1" -> {"evil": "1"}).""" + resolved: dict = {} + for section in ("dependencies", "optionalDependencies"): + for name, spec in (version_info.get(section) or {}).items(): + real_name, selector = resolve_npm_alias(name, spec) + resolved[real_name] = selector + return resolved + + def _scan_dependency(self, dep_name: str, spec: str) -> Optional[DependencyRisk]: + """Scan a single dependency as a subprocess and return its risk outcome. + + The sub-scan runs with source-code rules only: every metadata rule is + excluded. A maliciously introduced dependency reveals itself through its + code (obfuscation, exec, exfiltration, install hooks), whereas metadata + rules (typosquatting, manifest mismatch, ...) are a weak, noisy signal in + this context. Excluding all metadata rules also excludes this one, so the + check never recurses. Returns None when the scan can't be run or produced + no score. The subprocess takes the same CLI path as a top-level scan, so + it is sandboxed identically to the parent.""" + resolved_version = None + try: + resolved_version = highest_matching_version(dep_name, spec) + except Exception as e: + log.debug(f"Could not resolve version for {dep_name} ({spec}): {e}") + log.debug( + f"[{self.name}] Resolved '{dep_name}' spec '{spec}' to version " + f"'{resolved_version or 'latest'}'" + ) + + command = [ + sys.executable, + "-m", + "guarddog", + "npm", + "scan", + dep_name, + "--output-format", + "json", + ] + for metadata_rule in self._metadata_rule_names(): + command += ["--exclude-rules", metadata_rule] + if resolved_version: + command += ["--version", resolved_version] + + sandbox_choice = os.environ.get("GUARDDOG_SUBSCAN_SANDBOX") + if sandbox_choice == "1": + command.append("--sandbox") + elif sandbox_choice == "0": + command.append("--no-sandbox") + + log.debug(f"[{self.name}] Scanning new dependency: {' '.join(command)}") + try: + completed = subprocess.run( + command, + capture_output=True, + text=True, + timeout=SUBSCAN_TIMEOUT_SECONDS, + ) + except subprocess.TimeoutExpired: + log.debug(f"Timed out scanning new dependency {dep_name}") + return None + + try: + result = json.loads(completed.stdout) + except json.JSONDecodeError: + log.debug( + f"Could not parse scan output for new dependency {dep_name}: " + f"{completed.stderr.strip()}" + ) + return None + + risk_score = result.get("risk_score") or {} + score = risk_score.get("score") + if score is None: + return None + + return DependencyRisk( + version=resolved_version or "latest", + score=float(score), + label=risk_score.get("label", ""), + matched_rules=self._matched_rules(result), + ) + + @staticmethod + def _matched_rules(result: dict) -> List[str]: + """Distinct rule names that flagged the dependency, in first-seen order. + + Prefers the threat rules behind each risk; falls back to any metadata rule + that produced a message when no source-code risks were formed.""" + rules: list[str] = [] + for risk in result.get("risks", []): + rule = risk.get("threat_rule") + if rule and rule not in rules: + rules.append(rule) + if not rules: + for rule, message in (result.get("results") or {}).items(): + if message and rule not in rules: + rules.append(rule) + return rules + + @staticmethod + def _metadata_rule_names() -> List[str]: + """All npm metadata rule names, excluded from the dependency sub-scan so it + runs source-code rules only. Imported lazily to avoid a circular import.""" + from guarddog.analyzer.metadata import get_metadata_detectors + from guarddog.ecosystems import ECOSYSTEM + + return list(get_metadata_detectors(ECOSYSTEM.NPM).keys()) diff --git a/guarddog/cli.py b/guarddog/cli.py index 3284769d..79757225 100644 --- a/guarddog/cli.py +++ b/guarddog/cli.py @@ -219,6 +219,10 @@ def _scan( else: sandbox = True + # Propagate the resolved sandbox decision so any sub-dependency scan spawned by a + # metadata rule (risky_new_dependency) is sandboxed identically to this scan. + os.environ["GUARDDOG_SUBSCAN_SANDBOX"] = "1" if sandbox else "0" + rule_param = _get_rule_param(rules, exclude_rules, ecosystem) scanner = get_package_scanner(ecosystem) if scanner is None: diff --git a/guarddog/reporters/human_readable.py b/guarddog/reporters/human_readable.py index 6c5520dd..b762b478 100644 --- a/guarddog/reporters/human_readable.py +++ b/guarddog/reporters/human_readable.py @@ -231,16 +231,18 @@ def _format_one_risk( ] if desc: block.append(" " + colored(desc, sev_color)) - - inspector_url = ( - HumanReadableReporter._pypi_finding_inspector_url(deep_base, loc_raw) - if deep_base and loc_raw - else None - ) - loc_line = colored(f"{loc_kw} {_sanitize(loc)}", "dark_grey") - if inspector_url: - loc_line = HumanReadableReporter._hyperlink(inspector_url, loc_line) - block.append(" " + loc_line) + # Metadata findings carry no file location; only show the line when there is + # one so it never renders as a bare "in ". + if loc: + inspector_url = ( + HumanReadableReporter._pypi_finding_inspector_url(deep_base, loc_raw) + if deep_base and loc_raw + else None + ) + loc_line = colored(f"{loc_kw} {_sanitize(loc)}", "dark_grey") + if inspector_url: + loc_line = HumanReadableReporter._hyperlink(inspector_url, loc_line) + block.append(" " + loc_line) code = risk.get("threat_code", "") if code: diff --git a/guarddog/scanners/npm_project_scanner.py b/guarddog/scanners/npm_project_scanner.py index d3c30b62..eed43fdb 100644 --- a/guarddog/scanners/npm_project_scanner.py +++ b/guarddog/scanners/npm_project_scanner.py @@ -4,19 +4,17 @@ import re from typing import List -import requests -from semantic_version import NpmSpec, Version # type: ignore - from guarddog.scanners.npm_package_scanner import NPMPackageScanner from guarddog.scanners.scanner import Dependency, DependencyVersion, ProjectScanner from guarddog.utils.config import VERIFY_EXHAUSTIVE_DEPENDENCIES +from guarddog.utils.npm import ( + find_all_versions, + get_matched_versions, + resolve_npm_alias, +) log = logging.getLogger("guarddog") -NPM_ALIAS_PATTERN = re.compile( - r"^npm:(?P@[^/@\s]+/[^@\s]+|[^@\s]+)(?:@(?P.+))?$" -) - class NPMRequirementsScanner(ProjectScanner): """ @@ -54,64 +52,12 @@ def parse_requirements(self, raw_requirements: str) -> List[Dependency]: ) raw_requirement_lines = raw_requirements.splitlines() - def resolve_dependency_spec( - package_name: str, selector: str - ) -> tuple[str, str]: - """ - Normalizes npm alias selectors so scanning targets the real package. - ex: {"alias": "npm:react@19.2.3"} -> ("react", "19.2.3") - """ - match = NPM_ALIAS_PATTERN.match(selector) - if match is None: - return package_name, selector - - resolved_selector = match.group("selector") or "*" - return match.group("package"), resolved_selector - - def get_matched_versions(versions: set[str], semver_range: str) -> set[str]: - """ - Retrieves all versions that match a given semver selector - """ - result = [] - - # Filters to specified versions - try: - spec = NpmSpec(semver_range) - result = [Version(m) for m in versions if spec.match(Version(m))] - except ValueError: - # use it raw - return set([semver_range]) - - # If just the best matched version scan is required we only keep one - if not VERIFY_EXHAUSTIVE_DEPENDENCIES and result: - result = [sorted(result).pop()] - - return set([str(r) for r in result]) - - def find_all_versions(package_name: str) -> set[str]: - """ - This helper function retrieves all versions availables for the package - """ - url = f"https://registry.npmjs.org/{package_name}" - log.debug(f"Retrieving npm package metadata from {url}") - response = requests.get(url) - if response.status_code != 200: - log.debug(f"No version available, status code {response.status_code}") - return set() - - data = response.json() - versions = set(data["versions"].keys()) - log.debug(f"Retrieved versions {', '.join(versions)}") - return versions - merged = {} # type: dict[str, set[str]] merged_original_names = {} # type: dict[str, set[str]] for package, selector in list(dependencies_attr.items()) + list( dev_dependencies_attr.items() ): - resolved_package, resolved_selector = resolve_dependency_spec( - package, selector - ) + resolved_package, resolved_selector = resolve_npm_alias(package, selector) if resolved_package not in merged: merged[resolved_package] = set() merged[resolved_package].add(resolved_selector) @@ -125,7 +71,11 @@ def find_all_versions(package_name: str) -> set[str]: versions = set() # type: set[str] for selector in all_selectors: versions = versions.union( - get_matched_versions(find_all_versions(package), selector) + get_matched_versions( + find_all_versions(package), + selector, + exhaustive=VERIFY_EXHAUSTIVE_DEPENDENCIES, + ) ) if len(versions) == 0: diff --git a/guarddog/utils/config.py b/guarddog/utils/config.py index 1824cf11..8b10d7b7 100644 --- a/guarddog/utils/config.py +++ b/guarddog/utils/config.py @@ -18,6 +18,18 @@ os.environ.get("GUARDDOG_VERIFY_EXHAUSTIVE_DEPENDENCIES", "false").lower() == "true" ) +""" +Risk score (0-10) at or above which a newly added npm dependency is flagged by the +risky_new_dependency rule. The default of 5.0 (GuardDog's `suspicious` band) is +viable because the sub-scan runs source-code rules only: across 3000 popular npm +packages the 5.0-6.9 band was empty of false positives, so 5.0 catches genuinely +suspicious dependencies without the metadata noise that previously required 7.0. +- Default: 5.0 +""" +NEW_DEPENDENCY_RISK_THRESHOLD: float = float( + os.environ.get("GUARDDOG_NEW_DEPENDENCY_RISK_THRESHOLD", 5.0) +) + """ This parameter specifies the location of the top packages cache - Default: guarddog/analyzer/metadata/resources diff --git a/guarddog/utils/npm.py b/guarddog/utils/npm.py new file mode 100644 index 00000000..20b536fb --- /dev/null +++ b/guarddog/utils/npm.py @@ -0,0 +1,80 @@ +"""Shared helpers for resolving npm package versions from the registry.""" + +import logging +import re + +import requests +from semantic_version import NpmSpec, Version # type: ignore + +log = logging.getLogger("guarddog") + +# Matches npm alias specifiers, e.g. "npm:react@19.2.3" or "npm:@scope/pkg@^1". +NPM_ALIAS_PATTERN = re.compile( + r"^npm:(?P@[^/@\s]+/[^@\s]+|[^@\s]+)(?:@(?P.+))?$" +) + + +def resolve_npm_alias(package_name: str, selector: str) -> tuple[str, str]: + """Normalize an npm alias so scanning targets the real package. + + ex: ("alias", "npm:react@19.2.3") -> ("react", "19.2.3"). + Non-alias specifiers are returned unchanged. + """ + match = NPM_ALIAS_PATTERN.match(selector) + if match is None: + return package_name, selector + return match.group("package"), match.group("selector") or "*" + + +def find_all_versions(package_name: str) -> set[str]: + """Retrieve all published versions of a package from the npm registry.""" + url = f"https://registry.npmjs.org/{package_name}" + log.debug(f"Retrieving npm package metadata from {url}") + response = requests.get(url) + if response.status_code != 200: + log.debug(f"No version available, status code {response.status_code}") + return set() + + data = response.json() + versions = set(data["versions"].keys()) + log.debug(f"Retrieved versions {', '.join(versions)}") + return versions + + +def get_matched_versions( + versions: set[str], semver_range: str, exhaustive: bool = False +) -> set[str]: + """Return the versions matching a semver selector. + + When `exhaustive` is False only the single highest matching version is kept. + An unparseable range is returned verbatim so it can still be scanned as-is. + """ + try: + spec = NpmSpec(semver_range) + result = [Version(m) for m in versions if spec.match(Version(m))] + except ValueError: + return {semver_range} + + if not exhaustive and result: + result = [sorted(result).pop()] + + return {str(r) for r in result} + + +def highest_matching_version(package_name: str, semver_range: str) -> str | None: + """Resolve a semver range to the highest published version that matches it. + + Returns None when the package has no published versions, none match, or the + range is not a concrete semver selector (e.g. a URL or git dependency) so the + caller can fall back to scanning the latest version. + """ + matched = get_matched_versions(find_all_versions(package_name), semver_range) + valid = [] + for m in matched: + try: + valid.append(Version(m)) + except ValueError: + continue + if not valid: + return None + return str(sorted(valid).pop()) diff --git a/tests/analyzer/metadata/test_npm_risky_new_dependency.py b/tests/analyzer/metadata/test_npm_risky_new_dependency.py new file mode 100644 index 00000000..0dfd323a --- /dev/null +++ b/tests/analyzer/metadata/test_npm_risky_new_dependency.py @@ -0,0 +1,382 @@ +import json +from types import SimpleNamespace + +import pytest + +from guarddog.analyzer.metadata.npm import risky_new_dependency +from guarddog.analyzer.metadata.npm.risky_new_dependency import ( + DependencyRisk, + NPMRiskyNewDependencyDetector, +) + + +def make_info(versions_deps, times, latest, name="parent"): + """Build a minimal npm registry metadata dict.""" + return { + "name": name, + "dist-tags": {"latest": latest}, + "versions": {v: {"dependencies": deps} for v, deps in versions_deps.items()}, + "time": times, + } + + +# 1.0.0 published before 1.1.0; 1.1.0 adds dependency "b". +ADDED_DEP_INFO = make_info( + versions_deps={ + "1.0.0": {"a": "^1.0.0"}, + "1.1.0": {"a": "^1.0.0", "b": "^2.0.0"}, + }, + times={ + "created": "2020-01-01T00:00:00.000Z", + "1.0.0": "2020-01-01T00:00:00.000Z", + "1.1.0": "2020-06-01T00:00:00.000Z", + "modified": "2020-06-01T00:00:00.000Z", + }, + latest="1.1.0", +) + + +class TestDetectLogic: + detector = NPMRiskyNewDependencyDetector() + + def _patch_scan(self, monkeypatch, mapping): + """Replace the subprocess sub-scan with a deterministic mapping.""" + monkeypatch.setattr( + self.detector, + "_scan_dependency", + lambda dep_name, spec: mapping.get(dep_name), + ) + + def test_risky_new_dependency_flagged(self, monkeypatch): + self._patch_scan( + monkeypatch, + { + "b": DependencyRisk( + "2.0.0", 7.5, "high_risk", ["npm-exfiltrate-sensitive-data"] + ) + }, + ) + matched, message = self.detector.detect(ADDED_DEP_INFO, version="1.1.0") + assert matched is True + assert message is not None + assert "b@2.0.0" in message + # label-based phrasing (mirrors GuardDog's assessment wording), not "malicious" + assert risky_new_dependency._LABEL_PHRASE["high_risk"] in message + assert "malicious" not in message + assert "npm-exfiltrate-sensitive-data" in message # which rule matched + assert "1.0.0" in message # references the previous version + + def test_low_scoring_new_dependency_not_flagged(self, monkeypatch): + self._patch_scan(monkeypatch, {"b": DependencyRisk("2.0.0", 3.0, "low")}) + matched, message = self.detector.detect(ADDED_DEP_INFO, version="1.1.0") + assert matched is False + assert message == "" + + def test_version_bump_of_existing_dependency_not_flagged(self, monkeypatch): + info = make_info( + versions_deps={ + "1.0.0": {"a": "^1.0.0"}, + "1.1.0": {"a": "^2.0.0"}, + }, + times={ + "1.0.0": "2020-01-01T00:00:00.000Z", + "1.1.0": "2020-06-01T00:00:00.000Z", + }, + latest="1.1.0", + ) + + def fail(*args, **kwargs): + raise AssertionError("should not scan when no dependency was added") + + monkeypatch.setattr(self.detector, "_scan_dependency", fail) + matched, _ = self.detector.detect(info, version="1.1.0") + assert matched is False + + def test_earliest_version_has_no_previous(self, monkeypatch): + def fail(*args, **kwargs): + raise AssertionError("should not scan when there is no previous version") + + monkeypatch.setattr(self.detector, "_scan_dependency", fail) + matched, _ = self.detector.detect(ADDED_DEP_INFO, version="1.0.0") + assert matched is False + + def test_latest_resolved_when_version_omitted(self, monkeypatch): + self._patch_scan(monkeypatch, {"b": DependencyRisk("2.0.0", 8.0, "high_risk")}) + matched, message = self.detector.detect(ADDED_DEP_INFO) + assert matched is True + assert message is not None + assert "b@2.0.0" in message + + def test_only_risky_dep_appears_in_message(self, monkeypatch): + info = make_info( + versions_deps={ + "1.0.0": {"a": "^1.0.0"}, + "1.1.0": {"a": "^1.0.0", "good": "^1.0.0", "bad": "^2.0.0"}, + }, + times={ + "1.0.0": "2020-01-01T00:00:00.000Z", + "1.1.0": "2020-06-01T00:00:00.000Z", + }, + latest="1.1.0", + ) + self._patch_scan( + monkeypatch, + { + "good": DependencyRisk("1.0.0", 1.0, "low"), + "bad": DependencyRisk("2.0.0", 9.0, "high_risk"), + }, + ) + matched, message = self.detector.detect(info, version="1.1.0") + assert matched is True + assert message is not None + assert "bad@2.0.0" in message + assert "good" not in message + + +def make_info_full(versions, times, latest, name="parent"): + """Like make_info, but each version maps to a full version_info dict so + optionalDependencies / alias specifiers can be expressed.""" + return { + "name": name, + "dist-tags": {"latest": latest}, + "versions": versions, + "time": times, + } + + +TIMES = { + "1.0.0": "2020-01-01T00:00:00.000Z", + "1.1.0": "2020-06-01T00:00:00.000Z", +} + + +class TestDependencyCollection: + detector = NPMRiskyNewDependencyDetector() + + def test_includes_dependencies_and_optional_dependencies(self): + info = { + "dependencies": {"a": "^1.0.0"}, + "optionalDependencies": {"b": "^2.0.0"}, + "devDependencies": {"c": "^3.0.0"}, # not installed for consumers + } + assert self.detector._installed_dependencies(info) == { + "a": "^1.0.0", + "b": "^2.0.0", + } + + def test_resolves_npm_alias_to_real_package(self): + info = {"dependencies": {"local-name": "npm:evil-pkg@1.2.3"}} + assert self.detector._installed_dependencies(info) == {"evil-pkg": "1.2.3"} + + def test_alias_without_version_uses_wildcard(self): + info = {"dependencies": {"local": "npm:evil-pkg"}} + assert self.detector._installed_dependencies(info) == {"evil-pkg": "*"} + + +class TestDetectAliasAndOptional: + detector = NPMRiskyNewDependencyDetector() + + def _capture(self, monkeypatch, risk_for): + scanned = [] + + def fake(dep_name, spec): + scanned.append((dep_name, spec)) + return risk_for.get(dep_name) + + monkeypatch.setattr(self.detector, "_scan_dependency", fake) + return scanned + + def test_new_optional_dependency_is_scanned_and_flagged(self, monkeypatch): + info = make_info_full( + versions={ + "1.0.0": {"dependencies": {"a": "^1.0.0"}}, + "1.1.0": { + "dependencies": {"a": "^1.0.0"}, + "optionalDependencies": {"evil": "^2.0.0"}, + }, + }, + times=TIMES, + latest="1.1.0", + ) + scanned = self._capture( + monkeypatch, {"evil": DependencyRisk("2.0.0", 9.0, "high_risk")} + ) + matched, message = self.detector.detect(info, version="1.1.0") + assert ("evil", "^2.0.0") in scanned + assert matched is True + assert "evil@2.0.0" in message + + def test_aliased_new_dependency_scans_real_package(self, monkeypatch): + info = make_info_full( + versions={ + "1.0.0": {"dependencies": {"a": "^1.0.0"}}, + "1.1.0": {"dependencies": {"a": "^1.0.0", "x": "npm:evil-pkg@1.2.3"}}, + }, + times=TIMES, + latest="1.1.0", + ) + scanned = self._capture( + monkeypatch, {"evil-pkg": DependencyRisk("1.2.3", 8.0, "high_risk")} + ) + matched, message = self.detector.detect(info, version="1.1.0") + # scans the real package, not the local alias key + assert ("evil-pkg", "1.2.3") in scanned + assert "x" not in [name for name, _ in scanned] + assert matched is True + assert "evil-pkg@1.2.3" in message + + def test_alias_retarget_same_key_is_detected(self, monkeypatch): + # the local key "x" is unchanged but it now aliases a different package + info = make_info_full( + versions={ + "1.0.0": {"dependencies": {"x": "npm:safe-pkg@1.0.0"}}, + "1.1.0": {"dependencies": {"x": "npm:evil-pkg@1.0.0"}}, + }, + times=TIMES, + latest="1.1.0", + ) + scanned = self._capture( + monkeypatch, {"evil-pkg": DependencyRisk("1.0.0", 9.0, "high_risk")} + ) + matched, _ = self.detector.detect(info, version="1.1.0") + assert ("evil-pkg", "1.0.0") in scanned + assert matched is True + + +class TestScanDependencySubprocess: + detector = NPMRiskyNewDependencyDetector() + + def _fake_run(self, captured, stdout): + def run(command, **kwargs): + captured.append(command) + return SimpleNamespace(stdout=stdout, stderr="") + + return run + + def test_subscan_excludes_all_metadata_rules(self, monkeypatch): + from guarddog.analyzer.metadata import get_metadata_detectors + from guarddog.ecosystems import ECOSYSTEM + + monkeypatch.setattr( + risky_new_dependency, "highest_matching_version", lambda n, s: "2.0.0" + ) + captured: list = [] + stdout = json.dumps({"risk_score": {"score": 6.0, "label": "suspicious"}}) + monkeypatch.setattr( + risky_new_dependency.subprocess, "run", self._fake_run(captured, stdout) + ) + monkeypatch.delenv("GUARDDOG_SUBSCAN_SANDBOX", raising=False) + + result = self.detector._scan_dependency("b", "^2.0.0") + assert result == DependencyRisk("2.0.0", 6.0, "suspicious", []) + + command = captured[0] + excluded = { + command[i + 1] for i, tok in enumerate(command) if tok == "--exclude-rules" + } + metadata_rules = set(get_metadata_detectors(ECOSYSTEM.NPM).keys()) + # source-code only: every metadata rule excluded, including this one (no recursion) + assert excluded == metadata_rules + assert "risky_new_dependency" in excluded + assert command[command.index("--version") + 1] == "2.0.0" + assert "--sandbox" not in command and "--no-sandbox" not in command + + @pytest.mark.parametrize( + "env_value,expected_flag", + [("1", "--sandbox"), ("0", "--no-sandbox")], + ) + def test_sandbox_flag_propagated(self, monkeypatch, env_value, expected_flag): + monkeypatch.setattr( + risky_new_dependency, "highest_matching_version", lambda n, s: "2.0.0" + ) + captured: list = [] + stdout = json.dumps({"risk_score": {"score": 6.0, "label": "suspicious"}}) + monkeypatch.setattr( + risky_new_dependency.subprocess, "run", self._fake_run(captured, stdout) + ) + monkeypatch.setenv("GUARDDOG_SUBSCAN_SANDBOX", env_value) + + self.detector._scan_dependency("b", "^2.0.0") + assert expected_flag in captured[0] + + def test_unparseable_output_returns_none(self, monkeypatch): + monkeypatch.setattr( + risky_new_dependency, "highest_matching_version", lambda n, s: None + ) + monkeypatch.setattr( + risky_new_dependency.subprocess, + "run", + lambda command, **kwargs: SimpleNamespace(stdout="not json", stderr="boom"), + ) + assert self.detector._scan_dependency("b", "^2.0.0") is None + + def test_timeout_returns_none(self, monkeypatch): + import subprocess + + monkeypatch.setattr( + risky_new_dependency, "highest_matching_version", lambda n, s: "2.0.0" + ) + + def raise_timeout(command, **kwargs): + raise subprocess.TimeoutExpired(command, 1) + + monkeypatch.setattr(risky_new_dependency.subprocess, "run", raise_timeout) + assert self.detector._scan_dependency("b", "^2.0.0") is None + + def test_missing_risk_score_returns_none(self, monkeypatch): + monkeypatch.setattr( + risky_new_dependency, "highest_matching_version", lambda n, s: "2.0.0" + ) + stdout = json.dumps({"issues": 0, "errors": {"download-package": "404"}}) + monkeypatch.setattr( + risky_new_dependency.subprocess, + "run", + lambda command, **kwargs: SimpleNamespace(stdout=stdout, stderr=""), + ) + assert self.detector._scan_dependency("b", "^2.0.0") is None + + def test_matched_rules_collected_from_risks(self, monkeypatch): + monkeypatch.setattr( + risky_new_dependency, "highest_matching_version", lambda n, s: "2.0.0" + ) + stdout = json.dumps( + { + "risk_score": {"score": 8.0, "label": "high_risk"}, + "risks": [ + {"threat_rule": "npm-exec-base64"}, + {"threat_rule": "npm-exfiltrate-sensitive-data"}, + {"threat_rule": "npm-exec-base64"}, # duplicate, deduped + ], + } + ) + monkeypatch.setattr( + risky_new_dependency.subprocess, + "run", + lambda command, **kwargs: SimpleNamespace(stdout=stdout, stderr=""), + ) + result = self.detector._scan_dependency("b", "^2.0.0") + assert result is not None + assert result.matched_rules == [ + "npm-exec-base64", + "npm-exfiltrate-sensitive-data", + ] + + def test_matched_rules_fall_back_to_metadata_results(self, monkeypatch): + monkeypatch.setattr( + risky_new_dependency, "highest_matching_version", lambda n, s: "2.0.0" + ) + stdout = json.dumps( + { + "risk_score": {"score": 6.0, "label": "suspicious"}, + "risks": [], + "results": {"typosquatting": "looks like express", "shady": None}, + } + ) + monkeypatch.setattr( + risky_new_dependency.subprocess, + "run", + lambda command, **kwargs: SimpleNamespace(stdout=stdout, stderr=""), + ) + result = self.detector._scan_dependency("b", "^2.0.0") + assert result is not None + assert result.matched_rules == ["typosquatting"]