#!/usr/bin/env python3 # SPDX-License-Identifier: AGPL-3.0-only # Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. # security-audit.yml's npm-scan-packages job depends on this path. """Scans npm lockfile tarballs for IOCs without running them; downloads only from registry.npmjs.org; SHA-512 checked before opening, mismatch aborts; safe_extract guards; stdlib only; exit 1 on HIGH+.""" from __future__ import annotations import argparse import atexit import base64 as _b64 # imported only so the IOC string-scan can detect it import bisect import hashlib import io import itertools import json import os import re import shutil import sys import tarfile import tempfile import urllib.parse import urllib.request from dataclasses import dataclass, field from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[1] # ───────────────────────────────────────────────────────────────────── # Calibrated against the real frontend closure. Native binaries get a larger cap and skip # pattern scanning; text keeps a tight cap (typescript.js at 9.1 MB is the ceiling). # ───────────────────────────────────────────────────────────────────── HARD_MAX_TARBALL_BYTES = 256 * 1024 * 1024 # 256 MiB compressed HARD_MAX_TEXT_FILE_BYTES = 16 * 1024 * 1024 # 16 MiB per text file HARD_MAX_BINARY_FILE_BYTES = 256 * 1024 * 1024 # 256 MiB per .node etc HARD_MAX_TOTAL_BYTES = 512 * 1024 * 1024 # 512 MiB cumulative HARD_MAX_MEMBERS = 50_000 # entries per tarball HARD_HTTP_TIMEOUT_S = 60 # per request # Suffix shortlist; the content-magic check covers extensionless and versioned binaries. _BINARY_SUFFIXES = ( ".node", ".wasm", ".so", ".dll", ".dylib", ".exe", ".a", ".lib", ".o", ".obj", ".bin", ".dat", ".woff", ".woff2", ".ttf", ".otf", ".eot", ".png", ".jpg", ".jpeg", ".gif", ".webp", ".ico", ".mp3", ".mp4", ".webm", ".zip", ".tar", ".gz", ".tgz", ".xz", ".bz2", ) _VERSIONED_LIB = re.compile( r"\.(?:so|dylib)(?:\.\d+)+$", re.IGNORECASE, ) # Sniffed from the first bytes to catch extensionless binaries. _BINARY_MAGICS = ( b"\x7fELF", # ELF (Linux executable / .so) b"MZ", # PE / .exe / .dll (DOS header prefix) b"\xfe\xed\xfa\xce", # Mach-O 32 BE b"\xfe\xed\xfa\xcf", # Mach-O 64 BE b"\xce\xfa\xed\xfe", # Mach-O 32 LE b"\xcf\xfa\xed\xfe", # Mach-O 64 LE b"\xca\xfe\xba\xbe", # Mach-O fat / Java class (also starts with this) b"\x00asm", # WASM b"PK\x03\x04", # ZIP / JAR / nupkg / xpi b"PK\x05\x06", # ZIP (empty) b"\x1f\x8b", # gzip b"BZh", # bzip2 b"\xfd7zXZ", # xz b"7z\xbc\xaf\x27\x1c", # 7zip b"\x89PNG", # PNG b"\xff\xd8\xff", # JPEG b"GIF8", # GIF b"RIFF", # WAV / WEBP / AVI container b"\x00\x00\x01\x00", # ICO b"OggS", # Ogg b"\x1aE\xdf\xa3", # Matroska / WebM ) def _looks_binary(name: str, header: bytes) -> bool: """True if `name` or first bytes suggest a non-text file.""" lower = name.lower() if lower.endswith(_BINARY_SUFFIXES): return True if _VERSIONED_LIB.search(lower): return True for magic in _BINARY_MAGICS: if header.startswith(magic): return True if header and (header.count(b"\x00") / len(header)) > 0.02: return True return False ALLOWED_DOWNLOAD_HOST = "registry.npmjs.org" # ───────────────────────────────────────────────────────────────────── # Severities and Finding shape mirror scripts/scan_packages.py. # ───────────────────────────────────────────────────────────────────── CRITICAL = "CRITICAL" HIGH = "HIGH" MEDIUM = "MEDIUM" INFO = "INFO" _SEVERITY_RANK = {CRITICAL: 0, HIGH: 1, MEDIUM: 2, INFO: 3} @dataclass class Finding: severity: str package: str # name@version filename: str # relative path inside the tarball pattern: str # what matched evidence: str = "" # short surrounding snippet detail: str = "" # human-readable description def __str__(self) -> str: head = f" [{self.severity}] {self.package} :: {self.filename}" body = f" pattern: {self.pattern}" if self.detail: body += f"\n detail: {self.detail}" if self.evidence: ev = self.evidence if len(ev) > 240: ev = ev[:240] + "..." body += f"\n evidence: {ev!r}" return f"{head}\n{body}" @dataclass class PackageEntry: name: str version: str resolved: str integrity: str | None lockfile_key: str @property def display(self) -> str: return f"{self.name}@{self.version}" # ───────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────── # Keep IOC patterns short and factual: speculative patterns dull the signal. KNOWN_IOC_STRINGS: dict[str, tuple[str, str]] = { # Shai-Hulud TanStack wave (2026-05-11, GHSA-g7cv-rxg3-hmpx). "router_init.js": (HIGH, "filename associated with TanStack worm"), "tanstack_runner.js": (HIGH, "filename associated with TanStack worm"), "router_runtime.js": (HIGH, "filename associated with TanStack worm"), "A Mini Shai-Hulud has Appeared": ( CRITICAL, "TanStack worm campaign stdout marker", ), "github:tanstack/router#79ac49eedf774dd4b0cfa308722bc463cfe5885c": ( CRITICAL, "TanStack worm dropper pinned commit", ), "filev2.getsession.org": (CRITICAL, "exfiltration C2 host"), "getsession.org/file/": (CRITICAL, "exfiltration C2 endpoint"), # Mini Shai-Hulud May-12 2026 wave additions. "git-tanstack.com": (CRITICAL, "May-12 dropper host"), "transformers.pyz": (HIGH, "May-12 PyPI dropper artifact"), "/tmp/transformers.pyz": (CRITICAL, "May-12 dropper drop path"), "With Love TeamPCP": (CRITICAL, "May-12 campaign signature"), "We've been online over 2 hours": (CRITICAL, "May-12 campaign signature"), # Aikido (May-12 wave): payload SHA-256 hashes published in IOCs. "ab4fcadaec49c03278063dd269ea5eef82d24f2124a8e15d7b90f2fa8601266c": ( HIGH, "router_init.js payload SHA-256", ), "2ec78d556d696e208927cc503d48e4b5eb56b31abc2870c2ed2e98d6be27fc96": ( HIGH, "tanstack_runner.js payload SHA-256", ), # The new dependency vector: optional dep -> Bun-executed prepare script. "bun run tanstack_runner.js": ( CRITICAL, "TanStack-wave Bun prepare-script dropper invocation", ), "@tanstack/setup": ( CRITICAL, "TanStack-wave optional-dep dropper carrier (no legit pkg of this name)", ), } # Confirmed malicious versions; matched at the lockfile walk, so no tarball is fetched. # Keep in sync with scripts/lockfile_supply_chain_audit.py. BLOCKED_NPM_VERSIONS: dict[str, set[str]] = { # GHSA-g7cv-rxg3-hmpx -- TanStack May-11 2026 (84 versions). "@tanstack/arktype-adapter": {"1.166.12", "1.166.15"}, "@tanstack/eslint-plugin-router": {"1.161.9", "1.161.12"}, "@tanstack/eslint-plugin-start": {"0.0.4", "0.0.7"}, "@tanstack/history": {"1.161.9", "1.161.12"}, "@tanstack/nitro-v2-vite-plugin": {"1.154.12", "1.154.15"}, "@tanstack/react-router": {"1.169.5", "1.169.8"}, "@tanstack/react-router-devtools": {"1.166.16", "1.166.19"}, "@tanstack/react-router-ssr-query": {"1.166.15", "1.166.18"}, "@tanstack/react-start": {"1.167.68", "1.167.71"}, "@tanstack/react-start-client": {"1.166.51", "1.166.54"}, "@tanstack/react-start-rsc": {"0.0.47", "0.0.50"}, "@tanstack/react-start-server": {"1.166.55", "1.166.58"}, "@tanstack/router-cli": {"1.166.46", "1.166.49"}, "@tanstack/router-core": {"1.169.5", "1.169.8"}, "@tanstack/router-devtools": {"1.166.16", "1.166.19"}, "@tanstack/router-devtools-core": {"1.167.6", "1.167.9"}, "@tanstack/router-generator": {"1.166.45", "1.166.48"}, "@tanstack/router-plugin": {"1.167.38", "1.167.41"}, "@tanstack/router-ssr-query-core": {"1.168.3", "1.168.6"}, "@tanstack/router-utils": {"1.161.11", "1.161.14"}, "@tanstack/router-vite-plugin": {"1.166.53", "1.166.56"}, "@tanstack/solid-router": {"1.169.5", "1.169.8"}, "@tanstack/solid-router-devtools": {"1.166.16", "1.166.19"}, "@tanstack/solid-router-ssr-query": {"1.166.15", "1.166.18"}, "@tanstack/solid-start": {"1.167.65", "1.167.68"}, "@tanstack/solid-start-client": {"1.166.50", "1.166.53"}, "@tanstack/solid-start-server": {"1.166.54", "1.166.57"}, "@tanstack/start-client-core": {"1.168.5", "1.168.8"}, "@tanstack/start-fn-stubs": {"1.161.9", "1.161.12"}, "@tanstack/start-plugin-core": {"1.169.23", "1.169.26"}, "@tanstack/start-server-core": {"1.167.33", "1.167.36"}, "@tanstack/start-static-server-functions": {"1.166.44", "1.166.47"}, "@tanstack/start-storage-context": {"1.166.38", "1.166.41"}, "@tanstack/valibot-adapter": {"1.166.12", "1.166.15"}, "@tanstack/virtual-file-routes": {"1.161.10", "1.161.13"}, "@tanstack/vue-router": {"1.169.5", "1.169.8"}, "@tanstack/vue-router-devtools": {"1.166.16", "1.166.19"}, "@tanstack/vue-router-ssr-query": {"1.166.15", "1.166.18"}, "@tanstack/vue-start": {"1.167.61", "1.167.64"}, "@tanstack/vue-start-client": {"1.166.46", "1.166.49"}, "@tanstack/vue-start-server": {"1.166.50", "1.166.53"}, "@tanstack/zod-adapter": {"1.166.12", "1.166.15"}, # Mini Shai-Hulud May-12 wave: OpenSearch JS client. "@opensearch-project/opensearch": {"3.5.3", "3.6.2", "3.7.0", "3.8.0"}, # Mini Shai-Hulud May-12 wave: @squawk/* (22 packages, 5 versions each; https://safedep.io/mass-npm-supply-chain-attack-tanstack-mistral/). "@squawk/airport-data": {"0.7.4", "0.7.5", "0.7.6", "0.7.7", "0.7.8"}, "@squawk/airports": {"0.6.2", "0.6.3", "0.6.4", "0.6.5", "0.6.6"}, "@squawk/airspace": {"0.8.1", "0.8.2", "0.8.3", "0.8.4", "0.8.5"}, "@squawk/airspace-data": {"0.5.3", "0.5.4", "0.5.5", "0.5.6", "0.5.7"}, "@squawk/airway-data": {"0.5.4", "0.5.5", "0.5.6", "0.5.7", "0.5.8"}, "@squawk/airways": {"0.4.2", "0.4.3", "0.4.4", "0.4.5", "0.4.6"}, "@squawk/fix-data": {"0.6.4", "0.6.5", "0.6.6", "0.6.7", "0.6.8"}, "@squawk/fixes": {"0.3.2", "0.3.3", "0.3.4", "0.3.5", "0.3.6"}, "@squawk/flight-math": {"0.5.4", "0.5.5", "0.5.6", "0.5.7", "0.5.8"}, "@squawk/flightplan": {"0.5.2", "0.5.3", "0.5.4", "0.5.5", "0.5.6"}, "@squawk/geo": {"0.4.4", "0.4.5", "0.4.6", "0.4.7", "0.4.8"}, "@squawk/icao-registry": {"0.5.2", "0.5.3", "0.5.4", "0.5.5", "0.5.6"}, "@squawk/icao-registry-data": {"0.8.4", "0.8.5", "0.8.6", "0.8.7", "0.8.8"}, "@squawk/mcp": {"0.9.1", "0.9.2", "0.9.3", "0.9.4", "0.9.5"}, "@squawk/navaid-data": {"0.6.4", "0.6.5", "0.6.6", "0.6.7", "0.6.8"}, "@squawk/navaids": {"0.4.2", "0.4.3", "0.4.4", "0.4.5", "0.4.6"}, "@squawk/notams": {"0.3.6", "0.3.7", "0.3.8", "0.3.9", "0.3.10"}, "@squawk/procedure-data": {"0.7.3", "0.7.4", "0.7.5", "0.7.6", "0.7.7"}, "@squawk/procedures": {"0.5.2", "0.5.3", "0.5.4", "0.5.5", "0.5.6"}, "@squawk/types": {"0.8.1", "0.8.2", "0.8.3", "0.8.4", "0.8.5"}, "@squawk/units": {"0.4.3", "0.4.4", "0.4.5", "0.4.6", "0.4.7"}, "@squawk/weather": {"0.5.6", "0.5.7", "0.5.8", "0.5.9", "0.5.10"}, # Mini Shai-Hulud May-12 wave: @uipath/* (64 packages, single version each; https://www.aikido.dev/blog/mini-shai-hulud-is-back-tanstack-compromised). "@uipath/apollo-react": {"4.24.5"}, "@uipath/apollo-wind": {"2.16.2"}, "@uipath/cli": {"1.0.1"}, "@uipath/rpa-tool": {"0.9.5"}, "@uipath/apollo-core": {"5.9.2"}, "@uipath/filesystem": {"1.0.1"}, "@uipath/solutionpackager-tool-core": {"0.0.34"}, "@uipath/solution-tool": {"1.0.1"}, "@uipath/maestro-tool": {"1.0.1"}, "@uipath/codedapp-tool": {"1.0.1"}, "@uipath/agent-tool": {"1.0.1"}, "@uipath/orchestrator-tool": {"1.0.1"}, "@uipath/integrationservice-tool": {"1.0.2"}, "@uipath/rpa-legacy-tool": {"1.0.1"}, "@uipath/vertical-solutions-tool": {"1.0.1"}, "@uipath/flow-tool": {"1.0.2"}, "@uipath/codedagent-tool": {"1.0.1"}, "@uipath/common": {"1.0.1"}, "@uipath/resource-tool": {"1.0.1"}, "@uipath/auth": {"1.0.1"}, "@uipath/docsai-tool": {"1.0.1"}, "@uipath/case-tool": {"1.0.1"}, "@uipath/api-workflow-tool": {"1.0.1"}, "@uipath/test-manager-tool": {"1.0.2"}, "@uipath/robot": {"1.3.4"}, "@uipath/traces-tool": {"1.0.1"}, "@uipath/agent-sdk": {"1.0.2"}, "@uipath/integrationservice-sdk": {"1.0.2"}, "@uipath/maestro-sdk": {"1.0.1"}, "@uipath/data-fabric-tool": {"1.0.2"}, "@uipath/tasks-tool": {"1.0.1"}, "@uipath/insights-tool": {"1.0.1"}, "@uipath/insights-sdk": {"1.0.1"}, "@uipath/uipath-python-bridge": {"1.0.1"}, "@uipath/ap-chat": {"1.5.7"}, "@uipath/project-packager": {"1.1.16"}, "@uipath/packager-tool-case": {"0.0.9"}, "@uipath/packager-tool-workflowcompiler-browser": {"0.0.34"}, "@uipath/packager-tool-connector": {"0.0.19"}, "@uipath/packager-tool-workflowcompiler": {"0.0.16"}, "@uipath/packager-tool-webapp": {"1.0.6"}, "@uipath/packager-tool-apiworkflow": {"0.0.19"}, "@uipath/packager-tool-functions": {"0.1.1"}, "@uipath/widget.sdk": {"1.2.3"}, "@uipath/resources-tool": {"0.1.11"}, "@uipath/agent.sdk": {"0.0.18"}, "@uipath/codedagents-tool": {"0.1.12"}, "@uipath/aops-policy-tool": {"0.3.1"}, "@uipath/solution-packager": {"0.0.35"}, "@uipath/packager-tool-bpmn": {"0.0.9"}, "@uipath/packager-tool-flow": {"0.0.19"}, "@uipath/telemetry": {"0.0.7"}, "@uipath/tool-workflowcompiler": {"0.0.12"}, "@uipath/vss": {"0.1.6"}, "@uipath/solutionpackager-sdk": {"1.0.11"}, "@uipath/ui-widgets-multi-file-upload": {"1.0.1"}, "@uipath/access-policy-tool": {"0.3.1"}, "@uipath/context-grounding-tool": {"0.1.1"}, "@uipath/gov-tool": {"0.3.1"}, "@uipath/admin-tool": {"0.1.1"}, "@uipath/identity-tool": {"0.1.1"}, "@uipath/llmgw-tool": {"1.0.1"}, "@uipath/resourcecatalog-tool": {"0.1.1"}, "@uipath/functions-tool": {"1.0.1"}, "@uipath/access-policy-sdk": {"0.3.1"}, "@uipath/platform-tool": {"1.0.1"}, # Mini Shai-Hulud May-12 wave: @mistralai/* (npm, not PyPI mistralai). "@mistralai/mistralai": {"2.2.2", "2.2.3", "2.2.4"}, "@mistralai/mistralai-gcp": {"1.7.1", "1.7.2", "1.7.3"}, "@mistralai/mistralai-azure": {"1.7.1", "1.7.2", "1.7.3"}, # Mini Shai-Hulud May-12 wave: @tallyui/*. "@tallyui/components": {"1.0.1", "1.0.2", "1.0.3"}, "@tallyui/connector-medusa": {"1.0.1", "1.0.2", "1.0.3"}, "@tallyui/connector-shopify": {"1.0.1", "1.0.2", "1.0.3"}, "@tallyui/connector-vendure": {"1.0.1", "1.0.2", "1.0.3"}, "@tallyui/connector-woocommerce": {"1.0.1", "1.0.2", "1.0.3"}, "@tallyui/core": {"0.2.1", "0.2.2", "0.2.3"}, "@tallyui/database": {"1.0.1", "1.0.2", "1.0.3"}, "@tallyui/pos": {"0.1.1", "0.1.2", "0.1.3"}, "@tallyui/storage-sqlite": {"0.2.1", "0.2.2", "0.2.3"}, "@tallyui/theme": {"0.2.1", "0.2.2", "0.2.3"}, # Mini Shai-Hulud May-12 wave. "@beproduct/nestjs-auth": { "0.1.2", "0.1.3", "0.1.4", "0.1.5", "0.1.6", "0.1.7", "0.1.8", "0.1.9", "0.1.10", "0.1.11", "0.1.12", "0.1.13", "0.1.14", "0.1.15", "0.1.16", "0.1.17", "0.1.18", "0.1.19", }, # Mini Shai-Hulud May-12 wave: @draftlab/*, @draftauth/*. "@draftauth/client": {"0.2.1", "0.2.2"}, "@draftauth/core": {"0.13.1", "0.13.2"}, "@draftlab/auth": {"0.24.1", "0.24.2"}, "@draftlab/auth-router": {"0.5.1", "0.5.2"}, "@draftlab/db": {"0.16.1"}, # Mini Shai-Hulud May-12 wave. "@taskflow-corp/cli": {"0.1.24", "0.1.25", "0.1.26", "0.1.27", "0.1.28", "0.1.29"}, "@tolka/cli": {"1.0.2", "1.0.3", "1.0.4", "1.0.5", "1.0.6"}, # Mini Shai-Hulud May-12 wave: @ml-toolkit-ts/* + @mesadev/* + @dirigible-ai/sdk + @supersurkhet/*. "@dirigible-ai/sdk": {"0.6.2", "0.6.3"}, "@mesadev/rest": {"0.28.3"}, "@mesadev/saguaro": {"0.4.22"}, "@mesadev/sdk": {"0.28.3"}, "@ml-toolkit-ts/preprocessing": {"1.0.2", "1.0.3"}, "@ml-toolkit-ts/xgboost": {"1.0.3", "1.0.4"}, "@supersurkhet/cli": {"0.0.2", "0.0.3", "0.0.4", "0.0.5", "0.0.6", "0.0.7"}, "@supersurkhet/sdk": {"0.0.2", "0.0.3", "0.0.4", "0.0.5", "0.0.6", "0.0.7"}, # Mini Shai-Hulud May-12 wave: unscoped packages. "safe-action": {"0.8.3", "0.8.4"}, "ts-dna": {"3.0.1", "3.0.2", "3.0.3", "3.0.4"}, "cross-stitch": {"1.1.3", "1.1.4", "1.1.5", "1.1.6"}, "cmux-agent-mcp": {"0.1.3", "0.1.4", "0.1.5", "0.1.6", "0.1.7", "0.1.8"}, "agentwork-cli": {"0.1.4", "0.1.5"}, "git-branch-selector": {"1.3.3", "1.3.4", "1.3.5", "1.3.6", "1.3.7"}, "wot-api": {"0.8.1", "0.8.2", "0.8.3", "0.8.4"}, "git-git-git": {"1.0.8", "1.0.9", "1.0.10", "1.0.11", "1.0.12"}, "nextmove-mcp": {"0.1.3", "0.1.4", "0.1.5", "0.1.6", "0.1.7"}, "ml-toolkit-ts": {"1.0.4", "1.0.5"}, # Mini Shai-Hulud Apr-30 wave (npm side of PyPI lightning 2.6.2/2.6.3); 7.0.3 is safe. "intercom-client": {"7.0.4"}, } # Two tiers: ALWAYS_BAD has no legitimate use; NEEDS_CONTEXT also appears in defensive code # (SSRF blocklists), so it fires only with a fetch verb or inside a URL. CRED_HOST_ALWAYS_BAD: tuple[tuple[str, str], ...] = ( ("registry.npmjs.org/-/npm/v1/tokens", "npm publish-token enumeration endpoint"), ("ACTIONS_ID_TOKEN_REQUEST_URL", "GitHub Actions OIDC token-exchange endpoint env"), ("ACTIONS_ID_TOKEN_REQUEST_TOKEN", "GitHub Actions OIDC token-exchange token env"), ) CRED_HOST_NEEDS_CONTEXT: tuple[tuple[str, str], ...] = ( ("169.254.169.254", "AWS / GCP / Azure instance metadata service (IMDS)"), ("169.254.170.2", "ECS task metadata service"), ("metadata.google.internal", "GCE metadata service"), ("vault.svc.cluster.local", "in-cluster HashiCorp Vault endpoint"), ( "/var/run/secrets/kubernetes.io/serviceaccount", "Kubernetes ServiceAccount token path", ), ) # Flagged only inside lifecycle scripts (auto-run on npm ci). Joined from pieces at import, as # in scan_packages.py, so the file does not look like a credential stealer. CRED_PATH_SUBSTRINGS: tuple[tuple[str, str], ...] = ( ("".join(("/.n", "pmrc")), "npm credentials file"), ("".join(("/.a", "ws/cred", "entials")), "AWS shared credentials file"), ("".join(("/.s", "sh/id_", "rsa")), "SSH private key"), ("".join(("/.s", "sh/id_", "ed25519")), "SSH private key"), ("".join(("/.d", "ocker/con", "fig.json")), "Docker registry credentials"), ("".join(("/.k", "ube/con", "fig")), "Kubernetes kubeconfig"), ) _FETCH_VERBS_PAT = ( r"(?:fetch|axios|XMLHttpRequest|got\b|undici|" r"http\.get|https\.get|http\.request|https\.request|" r"new\s+URL|url\.parse|net\.connect|" r"\.request\s*\(|\.get\s*\(\s*['\"]\s*https?://)" ) _JS_FETCH_EVAL = re.compile( r"""(?xs) (?: Function\s*\(\s*['"`] # new Function("...") | eval\s*\(\s*['"`] | \(\s*0\s*,\s*eval\s*\)\s*\( ) .{0,200} (?:atob\s*\(|Buffer\s*\.from\s*\([^)]+,\s*['"]base64) """, ) _JS_ENV_TOKEN = re.compile( r"""(process\.env\.|os\.environ\[?['"])(?: GITHUB_TOKEN | GH_TOKEN | NPM_TOKEN | NODE_AUTH_TOKEN | AWS_ACCESS_KEY_ID | AWS_SECRET_ACCESS_KEY | AWS_SESSION_TOKEN | GOOGLE_APPLICATION_CREDENTIALS | DOCKER_AUTH_CONFIG | VAULT_TOKEN )['"]?\]?""", re.VERBOSE, ) # Only the fetch+exec chain is blocked; bare curl/wget is legit. _LIFECYCLE_FETCH_EXEC = re.compile( r"""(?xs) (?:curl|wget|fetch|http\.get|axios\.get)\s+ # fetch verb .{0,200} (?:\|\s*(?:sh|bash|node|python|eval)\b # pipe to interpreter | \&\&\s*(?:sh|bash|node|python|eval)\b # &&-chain to interpreter | -o\s+\S+\s*&&\s*(?:sh|bash|node|python) # download then run | --post-file\s+ | \$\(.*\) # command-sub of fetched content ) """, ) # Large base64-ish blob passed to Function() or eval(); tuned on the router_init.js shape. _OBFUSC_BLOB = re.compile( r"""(?xs) (?:Function|eval)\s*\(\s*['"`]? [A-Za-z0-9+/=_-]{2048,} # >=2 KiB of b64-ish """, ) # ───────────────────────────────────────────────────────────────────── def parse_lockfile(path: Path) -> tuple[list[PackageEntry], list[Finding]]: """Return (entries, structural_findings). Structural findings are HIGH-severity refusals that short-circuit the scan (e.g. non-registry resolved URLs), surfaced here so this scanner is standalone-runnable.""" entries: list[PackageEntry] = [] findings: list[Finding] = [] try: lock = json.loads(path.read_text(encoding = "utf-8")) except (OSError, json.JSONDecodeError) as exc: findings.append( Finding( severity = CRITICAL, package = "", filename = str(path), pattern = "lockfile-unreadable", detail = f"could not parse: {exc}", ) ) return entries, findings if lock.get("lockfileVersion") not in (2, 3): findings.append( Finding( severity = HIGH, package = "", filename = str(path), pattern = "unsupported-lockfile-version", detail = ( f"only lockfileVersion 2 or 3 supported; got " f"{lock.get('lockfileVersion')!r}" ), ) ) return entries, findings for key, entry in (lock.get("packages") or {}).items(): if key == "" or entry.get("link"): continue # Nested fold-ins are covered by the parent tarball's integrity. if key.count("/node_modules/") >= 1: continue resolved = entry.get("resolved") if not resolved: continue # Strict registry origin check, so the lockfile cannot pick the fetch URL. parsed = urllib.parse.urlparse(resolved) if parsed.scheme != "https" or parsed.hostname != ALLOWED_DOWNLOAD_HOST: findings.append( Finding( severity = CRITICAL, package = key, filename = str(path), pattern = "non-registry-resolved-url", detail = ( f"resolved={resolved!r}; only " f"https://{ALLOWED_DOWNLOAD_HOST}/ is " "permitted. Refusing to download." ), ) ) continue integrity = entry.get("integrity") if not integrity: findings.append( Finding( severity = HIGH, package = key, filename = str(path), pattern = "missing-integrity-hash", detail = "no `integrity` field; cannot verify download", ) ) continue nm = "node_modules/" name = key[len(nm) :] if key.startswith(nm) else key version = entry.get("version") or "" entries.append( PackageEntry( name = name, version = version, resolved = resolved, integrity = integrity, lockfile_key = key, ) ) return entries, findings # ───────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────── def _decode_integrity(integrity: str) -> tuple[str, bytes] | None: """Parse SRI integrity 'sha512-' -> (algo, digest_bytes).""" if "-" not in integrity: return None algo, b64 = integrity.split("-", 1) algo = algo.strip().lower() if algo not in ("sha256", "sha384", "sha512"): return None try: digest = _b64.b64decode(b64, validate = True) except Exception: return None return algo, digest def download_tarball( entry: PackageEntry, dest: Path, *, timeout: float = HARD_HTTP_TIMEOUT_S, max_bytes: int = HARD_MAX_TARBALL_BYTES, ) -> tuple[Path, str | None]: """Stream-download entry.resolved to dest and verify SRI integrity. Returns (downloaded_path, error_or_none); on error the path may not exist. Network access is restricted to ALLOWED_DOWNLOAD_HOST.""" # Re-assert the hostname as defence in depth against a future refactor. parsed = urllib.parse.urlparse(entry.resolved) if parsed.scheme != "https" or parsed.hostname != ALLOWED_DOWNLOAD_HOST: return dest, (f"refused download from non-allowlisted URL {entry.resolved!r}") decoded = _decode_integrity(entry.integrity or "") if decoded is None: return dest, f"unparseable integrity field {entry.integrity!r}" algo, expected_digest = decoded h = hashlib.new(algo) req = urllib.request.Request( entry.resolved, headers = { "User-Agent": "unsloth-scan-npm-packages/1.0 (+supply-chain audit)", "Accept": "application/octet-stream", }, method = "GET", ) try: with urllib.request.urlopen(req, timeout = timeout) as r: cl = r.headers.get("Content-Length") if cl is not None: try: cl_int = int(cl) if cl_int > max_bytes: return dest, (f"Content-Length {cl_int} > cap {max_bytes}") except ValueError: pass written = 0 with open(dest, "wb") as out: while True: chunk = r.read(64 * 1024) if not chunk: break written += len(chunk) if written > max_bytes: return dest, ( f"download exceeded cap {max_bytes} bytes " f"after {written} bytes" ) h.update(chunk) out.write(chunk) except Exception as exc: return dest, f"download failed: {exc}" actual = h.digest() if actual != expected_digest: return dest, ( f"integrity mismatch: expected {algo}={_b64.b64encode(expected_digest).decode()!r}, " f"got {algo}={_b64.b64encode(actual).decode()!r}" ) return dest, None # ───────────────────────────────────────────────────────────────────── # ───────────────────────────────────────────────────────────────────── def _is_within(root: Path, candidate: Path) -> bool: try: return candidate.resolve().is_relative_to(root.resolve()) except (AttributeError, ValueError): try: candidate.resolve().relative_to(root.resolve()) return True except Exception: return False def safe_extract( tarball_path: Path, extract_root: Path, *, max_total_bytes: int = HARD_MAX_TOTAL_BYTES, max_members: int = HARD_MAX_MEMBERS, ) -> str | None: """Extract tarball_path under extract_root with policed members, returning None on success or a string describing the refusal. Streams via `r|gz` so we can abort mid-extraction without having materialised the rest of the archive.""" extract_root.mkdir(parents = True, exist_ok = True) total = 0 count = 0 try: with tarfile.open(tarball_path, mode = "r|gz") as tf: for member in tf: count += 1 if count > max_members: return f"member count {count} exceeded cap {max_members}" name = member.name if name.startswith("/") or ".." in Path(name).parts: return f"refused unsafe member name {name!r}" if member.issym() or member.islnk(): return f"refused link member {name!r} (sym/lnk)" if member.isdev() or member.isfifo(): return f"refused special member {name!r}" declared = max(member.size, 0) if declared > HARD_MAX_BINARY_FILE_BYTES: return ( f"member {name!r} declared size {declared} > " f"absolute cap {HARD_MAX_BINARY_FILE_BYTES}" ) if total + declared > max_total_bytes: return ( f"cumulative bytes {total + declared} > cap " f"{max_total_bytes} at {name!r}" ) # Refuse anything escaping root; do not trust the npm package/ convention. dest = extract_root / name if not _is_within(extract_root, dest): return f"refused escape: {name!r} resolved outside root" if member.isdir(): dest.mkdir(parents = True, exist_ok = True) continue if not member.isfile(): return f"refused unknown member type for {name!r}" dest.parent.mkdir(parents = True, exist_ok = True) src = tf.extractfile(member) if src is None: continue header = src.read(16) is_binary = _looks_binary(name, header) file_cap = HARD_MAX_BINARY_FILE_BYTES if is_binary else HARD_MAX_TEXT_FILE_BYTES if declared > file_cap: return ( f"member {name!r} declared size {declared} > " f"cap {file_cap} ({'binary' if is_binary else 'text'})" ) remainder_cap = file_cap - len(header) rest = src.read(remainder_cap + 1) data = header + rest if len(data) > file_cap: return ( f"member {name!r} body exceeded declared size cap " f"({'binary' if is_binary else 'text'})" ) total += len(data) with open(dest, "wb") as out: out.write(data) os.chmod(dest, 0o644) except tarfile.TarError as exc: return f"tar parse error: {exc}" except Exception as exc: return f"unexpected extract error: {exc!r}" return None # Backward window for an enclosing opener; over-binding only adds context. _MAX_CONT_LINES = 200 # Forward cap measured from the match, so the tail is always reachable. _MAX_GROUP_LINES = 200 # Blank strings before counting brackets. _RE_JS_STR = re.compile(r"'(?:[^'\\]|\\.)*'|\"(?:[^\"\\]|\\.)*\"|`(?:[^`\\]|\\.)*`") _RE_BRACKETS = re.compile(r"[()\[\]{}]") _OPENERS = frozenset("([{") def _bracket_lr(line: str) -> tuple[int, int]: """Order-aware bracket reduction of one already-string-blanked line: ``(L, R)``, where ``L`` counts closers with no opener earlier on the line and ``R`` counts openers with no closer later. A plain net count collapses order and masks a trailing opener that follows leading closers (``}); const opts = {`` nets -1 and hides the ``{`` that opens the host-config object), so tracking the running minimum keeps that opener visible. Only bracket characters are walked so a long minified line stays cheap.""" depth = 0 low = 0 for ch in _RE_BRACKETS.findall(line): if ch in _OPENERS: depth += 1 else: depth -= 1 if depth < low: low = depth return -low, depth - low def _find_unescaped(line: str, quote: str, start: int) -> int: """Index of the next ``quote`` at or after ``start`` not escaped by a backslash, or -1. Skips ``\\x`` pairs so an escaped quote inside the string is ignored.""" i, n = start, len(line) while i < n: if line[i] == "\\": i += 2 continue if line[i] == quote: return i i += 1 return -1 # `/` starts a regex after these; misdetection only grows the bound span. _JS_REGEX_PRECEDERS = frozenset("([{,;:?=&|!+-*/%^~<>") def _blank_js_strings(lines: list[str]) -> list[str]: """Replace string contents (single, double, multi-line backtick template literals) AND regex literal bodies with spaces across ``lines``, keeping the line count and every bracket OUTSIDE a string/regex intact, so bracket counting never miscounts a ``)`` inside a string, including a template literal spanning several lines or a ``/)/`` regex, which a per-line regex cannot blank. Escapes are honoured.""" out: list[str] = [] in_back = False # inside a multi-line `template` literal prev_sig = "" # last significant non-space char (for regex-vs-division) for line in lines: buf: list[str] = [] i, n = 0, len(line) while i < n: if in_back: end = _find_unescaped(line, "`", i) if end == -1: buf.append(" " * (n - i)) i = n else: buf.append(" " * (end - i + 1)) i = end + 1 in_back = False prev_sig = "`" continue ch = line[i] if ch in " \t": buf.append(ch) i += 1 continue if ch in "'\"`": end = _find_unescaped(line, ch, i + 1) if end == -1: buf.append(" " * (n - i)) i = n if ch == "`": # opens a template literal that runs past this line in_back = True else: buf.append(" " * (end - i + 1)) i = end + 1 prev_sig = "v" # a string is a value: a following `/` is division continue if ch == "/" and (prev_sig == "" or prev_sig in _JS_REGEX_PRECEDERS): # A regex never spans lines, so no close on this line means division. j, in_class, closed = i + 1, False, False while j < n: c = line[j] if c == "\\": j += 2 continue if c == "[": in_class = True elif c == "]": in_class = False elif c == "/" and not in_class: j += 1 closed = True break j += 1 if closed: buf.append(" " * (j - i)) i = j prev_sig = "v" # a regex is a value continue buf.append(ch) i += 1 prev_sig = "/" continue buf.append(ch) i += 1 prev_sig = ch out.append("".join(buf)) return out def _index_text(text: str) -> tuple[list[str], list[str], list[str], list[int]]: """Precompute once per evidence call: raw lines for display, two string-blanked views for bracket counting (single-line via regex, and multi-line aware so a template literal spanning lines is blanked), and newline offsets for O(log n) offset-to-line mapping. Avoids re-splitting and re-counting the whole file on every match, which was O(matches x file size).""" lines = text.split("\n") sl_blanked = [_RE_JS_STR.sub("", ln) for ln in lines] ml_blanked = _blank_js_strings(lines) nl = [p for p, ch in enumerate(text) if ch == "\n"] return lines, sl_blanked, ml_blanked, nl # Past this cap, remaining matches fold into one digest so evidence stays bounded. _MAX_EVIDENCE_MATCHES = 64 def _scan_group(blanked: list[str], idx: int) -> tuple[int, int]: """(start, end) line indices of the bracket group enclosing line ``idx`` in one blanked view: scan back to the still-open opener, then forward to its close.""" # Order-aware reduction keeps a trailing opener visible; net counts would hide changed headers. start = idx depth = 0 for j in range(max(0, idx - _MAX_CONT_LINES), idx): left, right = _bracket_lr(blanked[j]) if left >= depth: depth = 0 # everything opened so far in the window has closed start = idx else: depth -= left if right > 0: if depth == 0: start = j # outermost still-open opener begins here depth += right # Same order-aware reduction forward; the cap is measured from the match. depth = 0 end = start for j in range(start, min(len(blanked), idx + _MAX_GROUP_LINES)): left, right = _bracket_lr(blanked[j]) depth = max(0, depth - left) + right end = j if j >= idx and depth <= 0: break return start, end def _canon_preserve_strings(text: str) -> str: """Whitespace canon that collapses runs OUTSIDE string literals to a single space (so a reindent stays stable) while preserving whitespace INSIDE single/double/backtick literals (so a changed payload body reopens). A plain ``" ".join(text.split())`` erases both, suppressing an intra-literal payload edit along with harmless indentation. Used for the evidence hash and the logical-line digests so the two stay consistent.""" out: list[str] = [] i, n = 0, len(text) quote: str | None = None pending_space = False while i < n: ch = text[i] if quote is not None: out.append(ch) if ch == "\\" and i + 1 < n: out.append(text[i + 1]) i += 2 continue if ch == quote: quote = None i += 1 continue if ch.isspace(): pending_space = True i += 1 continue if pending_space and out: out.append(" ") pending_space = False out.append(ch) if ch in "'\"`": quote = ch i += 1 return "".join(out) def _logical_line_text( lines: list[str], sl_blanked: list[str], ml_blanked: list[str], idx: int ) -> str: """The matched line plus the bracket group it belongs to, so a changed ``path``/``headers``/body on another line binds. Returns the UNION of the groups found in the single-line-blanked view (a payload embedded inside a template still counts) and the multi-line-blanked view (a bracket inside a multi-line template no longer closes the group early); unioning never shrinks the span below either view.""" s1, e1 = _scan_group(sl_blanked, idx) s2, e2 = _scan_group(ml_blanked, idx) start, end = min(s1, s2), max(e1, e2) return " ".join(lines[start : end + 1]) def _format_match( text: str, lines: list[str], sl_blanked: list[str], ml_blanked: list[str], nl: list[int], m: re.Match, max_chars: int, ) -> str: # Append a digest of the full logical line when the snippet does not show it all. idx = bisect.bisect_left(nl, m.start()) # 0-based line index of the match line_start = nl[idx - 1] + 1 if idx > 0 else 0 ke = bisect.bisect_left(nl, m.end()) line_end = nl[ke] if ke < len(nl) else len(text) full_logical = _logical_line_text(lines, sl_blanked, ml_blanked, idx) start = max(line_start, m.start() - 30) end = min(line_end, m.end() + 30) snippet = text[start:end].replace("\n", " ") if len(snippet) > max_chars: snippet = snippet[:max_chars] + "..." if snippet != full_logical: # Normalize like _evidence_hash: reindent does not reopen, string contents do. canon = _canon_preserve_strings(full_logical) digest = hashlib.sha256(canon.encode("utf-8", "replace")).hexdigest() snippet = f"{snippet} sha256:{digest}" return snippet def _stream_overflow_digest( matches, lines: list[str], sl_blanked: list[str], ml_blanked: list[str], nl: list[int] ) -> tuple[int, str]: """A single digest binding the LOGICAL line (the bound bracket-group context, not just the regex match text) of every overflow match in the iterable, plus the count folded. Streams the matches so a huge overflow never materializes a list. Whitespace-normalized to match _evidence_hash so a reindent does not reopen.""" h = hashlib.sha256() count = 0 for m in matches: _fold_overflow_match(h, m, lines, sl_blanked, ml_blanked, nl) count += 1 return count, h.hexdigest() def _fold_overflow_match( h, m: re.Match, lines: list[str], sl_blanked: list[str], ml_blanked: list[str], nl: list[int] ) -> None: """Fold one overflow match's whitespace-normalized logical-line context into the running hash ``h``. Shared by _stream_overflow_digest and the inline overflow fold in _outbound_host_evidence so both produce the identical digest.""" idx = bisect.bisect_left(nl, m.start()) ll = _logical_line_text(lines, sl_blanked, ml_blanked, idx) h.update(b"\x00") h.update(_canon_preserve_strings(ll).encode("utf-8", "replace")) def _evidence( text: str, pat: re.Pattern, max_chars: int = 200, ) -> str: # Stream every match, not a sample; materializing millions of matches would OOM. it = pat.finditer(text) shown_matches = list(itertools.islice(it, _MAX_EVIDENCE_MATCHES)) if not shown_matches: return "" lines, sl_blanked, ml_blanked, nl = _index_text(text) shown = [ _format_match(text, lines, sl_blanked, ml_blanked, nl, m, max_chars) for m in shown_matches ] # Byte-identical to digesting matches[_MAX_EVIDENCE_MATCHES:]. overflow_count, digest = _stream_overflow_digest(it, lines, sl_blanked, ml_blanked, nl) if overflow_count: shown.append(f"(+{overflow_count} more) sha256:{digest}") return " | ".join(shown) def _ioc_evidence(text: str, needle: str) -> str: """Matched-line context (with bracket-group continuation) for a literal IOC needle, so a changed adjacent fetch/exfil body reopens the key instead of riding the bare constant. Falls back to the needle itself if, defensively, nothing matches.""" return _evidence(text, re.compile(re.escape(needle))) or needle LIFECYCLE_HOOKS = ("preinstall", "install", "postinstall", "prepare") # ───────────────────────────────────────────────────────────────────── # Blank JS comments (top false-positive source) but not strings, where droppers hide payloads. # JS sibling of scan_packages.py::_strip_noncode. # ───────────────────────────────────────────────────────────────────── _JS_FAMILY_SUFFIXES = (".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx") _REGEX_PRECEDING_KEYWORDS = frozenset( { "return", "typeof", "instanceof", "in", "of", "new", "delete", "void", "throw", "yield", "await", "do", "else", "case", } ) _IDENT_CHARS = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_$") def _slash_is_regex(prev_tok: str) -> bool: """Disambiguate a lone ``/``: regex literal vs division operator. Biased toward regex when ambiguous, since regex state never blanks, so a wrong guess only costs FP reduction (or a fail-open), never a missed detection.""" if prev_tok == "": return True # start of file -> expression position if prev_tok in _REGEX_PRECEDING_KEYWORDS: return True last = prev_tok[-1] if last.isalnum() or last in "_$)]": return False # previous token ends a value -> division return True # operators, punctuation, `{`, `}` -> regex (safe bias) def _strip_js_noncode(text: str) -> str: """Blank JS/TS comments, preserving byte geometry. Fail-open on confusion.""" if "//" not in text and "/*" not in text: return text n = len(text) out = list(text) nl = ("\n", "\r") def _blank(a: int, b: int) -> None: for k in range(a, b): if out[k] not in nl: out[k] = " " state = "code" prev_tok = "" tmpl_stack: list[str] = [] i = 0 try: while i < n: c = text[i] nxt = text[i + 1] if i + 1 < n else "" if state == "code": if c == "/" and nxt == "/": start = i i += 2 while i < n and text[i] not in nl: i += 1 _blank(start, i) continue if c == "/" and nxt == "*": start = i i += 2 closed = False while i < n: if text[i] == "*" and i + 1 < n and text[i + 1] == "/": i += 2 closed = True break i += 1 if not closed: return text _blank(start, i) continue if c == "'": state = "sq" i += 1 continue if c == '"': state = "dq" i += 1 continue if c == "`": state = "tmpl" i += 1 continue if c == "/": if _slash_is_regex(prev_tok): state = "regex" i += 1 continue prev_tok = "/" i += 1 continue if c.isspace(): i += 1 continue if c in _IDENT_CHARS: j = i while j < n and text[j] in _IDENT_CHARS: j += 1 prev_tok = text[i:j] i = j continue if c == "}" and tmpl_stack: state = tmpl_stack.pop() i += 1 continue prev_tok = c i += 1 continue elif state in ("sq", "dq"): q = "'" if state == "sq" else '"' if c == "\\": i += 2 continue if c == q: state = "code" prev_tok = "_v" i += 1 continue if c in nl: return text i += 1 continue elif state == "tmpl": if c == "\\": i += 2 continue if c == "`": state = "code" prev_tok = "_v" i += 1 continue if c == "$" and nxt == "{": tmpl_stack.append("tmpl") state = "code" prev_tok = "{" i += 2 continue i += 1 continue elif state == "regex": if c == "\\": i += 2 continue if c == "[": state = "regex_cc" i += 1 continue if c == "/": state = "code" prev_tok = "_v" i += 1 continue if c in nl: return text i += 1 continue elif state == "regex_cc": if c == "\\": i += 2 continue if c == "]": state = "regex" i += 1 continue if c in nl: return text i += 1 continue else: return text if state != "code" or tmpl_stack: return text except Exception: return text return "".join(out) def scan_package_json(pkg: PackageEntry, rel: str, text: str) -> list[Finding]: findings: list[Finding] = [] try: meta = json.loads(text) except Exception: return findings if not isinstance(meta, dict): return findings scripts = meta.get("scripts") or {} if not isinstance(scripts, dict): return findings for hook in LIFECYCLE_HOOKS: body = scripts.get(hook) if not isinstance(body, str): continue # One digest pins the whole lifecycle body; real edits reopen it, but a reindent alone does not. body_digest = hashlib.sha256( _canon_preserve_strings(body).encode("utf-8", "replace") ).hexdigest() if _LIFECYCLE_FETCH_EXEC.search(body): findings.append( Finding( severity = CRITICAL, package = pkg.display, filename = rel, pattern = f"lifecycle-fetch-exec ({hook})", evidence = f"{_evidence(body, _LIFECYCLE_FETCH_EXEC)} body-sha256:{body_digest}", detail = ( f"`scripts.{hook}` fetches an external " "resource and pipes/chains it to an " "interpreter; this is the install-time RCE " "vector. Refusing to install." ), ) ) for path_substr, why in CRED_PATH_SUBSTRINGS: if path_substr in body: findings.append( Finding( severity = HIGH, package = pkg.display, filename = rel, pattern = f"cred-path-in-lifecycle ({hook})", evidence = ( f"{_evidence(body, re.compile(re.escape(path_substr)))} " f"body-sha256:{body_digest}" ), detail = ( f"`scripts.{hook}` references {why} " f"({path_substr!r}); install-time access " "to local credential files is the " "exfiltration prep step" ), ) ) if _JS_ENV_TOKEN.search(body): findings.append( Finding( severity = HIGH, package = pkg.display, filename = rel, pattern = f"cred-env-in-lifecycle ({hook})", evidence = f"{_evidence(body, _JS_ENV_TOKEN)} body-sha256:{body_digest}", detail = ( f"`scripts.{hook}` references a credential " "env var (GITHUB_TOKEN / NPM_TOKEN / AWS_* " "/ etc); install-time access to runner " "secrets is the exfiltration prep step" ), ) ) # Optional deps pointing at github: are the TanStack-style injection vector. opt = meta.get("optionalDependencies") or {} if isinstance(opt, dict): for k, v in opt.items(): if isinstance(v, str) and ( v.startswith("github:") or v.startswith("git+") or v.startswith("git://") ): findings.append( Finding( severity = HIGH, package = pkg.display, filename = rel, pattern = "optional-dep-non-registry", evidence = f"{k}={v}", detail = ( "package.json `optionalDependencies` " "points at a non-registry source; this " "is the Shai-Hulud worm injection shape." ), ) ) return findings def _host_in_outbound_context(text: str, host: str) -> bool: """True if `host` appears consistent with an outbound call. A bare array literal (defensive blocklist) is safe; co-occurrence with an HTTP URL scheme or a fetch verb in a short window is not.""" host_re = re.escape(host) url_form = re.compile( rf"(?:https?:)?//{host_re}(?:[:/\"'?#]|$)", ) if url_form.search(text): return True fetch_context = re.compile( rf"(?:{_FETCH_VERBS_PAT})[^\n]{{0,200}}{host_re}" rf"|{host_re}[^\n]{{0,200}}(?:{_FETCH_VERBS_PAT})", re.IGNORECASE, ) if fetch_context.search(text): return True cfg_form = re.compile( rf"(?:host|hostname)\s*:\s*['\"`]{host_re}['\"`]", re.IGNORECASE, ) if cfg_form.search(text): return True return False def _outbound_host_evidence(text: str, host: str) -> str: """Evidence capturing the host WITH its outbound context (URL path, fetch call, host config), so a changed path/headers/body reopens the key instead of riding the bare host literal. Falls back to the host if none matches.""" host_re = re.escape(host) patterns = ( re.compile(rf"(?:https?:)?//{host_re}(?:[:/\"'?#][^\n]*)?", re.IGNORECASE), re.compile( rf"(?:{_FETCH_VERBS_PAT})[^\n]{{0,200}}{host_re}[^\n]{{0,200}}" rf"|{host_re}[^\n]{{0,200}}(?:{_FETCH_VERBS_PAT})[^\n]{{0,200}}", re.IGNORECASE, ), # Capture the whole line so a changed payload on the same host line reopens. re.compile(rf"[^\n]*(?:host|hostname)\s*:\s*['\"`]{host_re}['\"`][^\n]*", re.IGNORECASE), ) # Record every outbound context for the host, capped and folded into a digest past the cap. lines, sl_blanked, ml_blanked, nl = _index_text(text) claimed: list[tuple[int, int]] = [] chosen: list[re.Match] = [] overflow_count = 0 overflow_hash = hashlib.sha256() for pat in patterns: for m in pat.finditer(text): if len(chosen) < _MAX_EVIDENCE_MATCHES: if any(m.start() < e and s < m.end() for s, e in claimed): continue claimed.append((m.start(), m.end())) chosen.append(m) else: _fold_overflow_match(overflow_hash, m, lines, sl_blanked, ml_blanked, nl) overflow_count += 1 if not chosen: return host chosen.sort(key = lambda m: m.start()) shown = [_format_match(text, lines, sl_blanked, ml_blanked, nl, m, 1000) for m in chosen] if overflow_count: shown.append(f"(+{overflow_count} more) sha256:{overflow_hash.hexdigest()}") return " | ".join(shown) def scan_text_blob(pkg: PackageEntry, rel: str, text: str) -> list[Finding]: findings: list[Finding] = [] if rel.lower().endswith(_JS_FAMILY_SUFFIXES): text = _strip_js_noncode(text) # Evidence includes the bracket-group context so a changed adjacent payload reopens. for needle, (sev, why) in KNOWN_IOC_STRINGS.items(): if needle in text: findings.append( Finding( severity = sev, package = pkg.display, filename = rel, pattern = "known-ioc-string", evidence = _ioc_evidence(text, needle), detail = f"{why}: {needle!r}", ) ) for needle, why in CRED_HOST_ALWAYS_BAD: if needle in text: findings.append( Finding( severity = HIGH, package = pkg.display, filename = rel, pattern = "cred-surface-host (always-bad)", evidence = _outbound_host_evidence(text, needle), detail = ( f"references {why} ({needle!r}); no legitimate " "frontend use of this surface" ), ) ) for needle, why in CRED_HOST_NEEDS_CONTEXT: if needle in text and _host_in_outbound_context(text, needle): findings.append( Finding( severity = HIGH, package = pkg.display, filename = rel, pattern = "cred-surface-host (outbound)", evidence = _outbound_host_evidence(text, needle), detail = ( f"references {why} ({needle!r}) in an outbound " "call / URL / host config; a defensive blocklist " "literal would not match this rule" ), ) ) # Credential paths are checked only in lifecycle scripts (scan_package_json). if _JS_FETCH_EVAL.search(text): findings.append( Finding( severity = HIGH, package = pkg.display, filename = rel, pattern = "js-fetch-eval", evidence = _evidence(text, _JS_FETCH_EVAL), detail = ("Function/eval against base64-decoded payload (obfuscated dropper shape)"), ) ) if _JS_ENV_TOKEN.search(text): findings.append( Finding( severity = MEDIUM, package = pkg.display, filename = rel, pattern = "js-env-token", evidence = _evidence(text, _JS_ENV_TOKEN), detail = ("references credential env vars in package source"), ) ) if _OBFUSC_BLOB.search(text): findings.append( Finding( severity = HIGH, package = pkg.display, filename = rel, pattern = "obfuscated-blob", evidence = _evidence(text, _OBFUSC_BLOB), detail = ( "large base64-ish blob fed to Function/eval; " "matches the TanStack worm dropper shape" ), ) ) return findings _TEXT_SUFFIXES = ( ".js", ".mjs", ".cjs", ".ts", ".tsx", ".json", ".html", ".htm", ".sh", ".bash", ".zsh", ".py", ".rb", ".yml", ".yaml", ) def scan_extracted_tree(pkg: PackageEntry, root: Path) -> list[Finding]: findings: list[Finding] = [] for path in sorted(root.rglob("*")): if not path.is_file(): continue rel = path.relative_to(root).as_posix() lower = rel.lower() if not lower.endswith(_TEXT_SUFFIXES): try: if path.stat().st_size > HARD_MAX_TEXT_FILE_BYTES: continue with open(path, "rb") as fh: header = fh.read(16) if _looks_binary(rel, header): continue data = header + path.read_bytes()[len(header) :] except OSError: continue text = data.decode("utf-8", errors = "replace") for needle, (sev, why) in KNOWN_IOC_STRINGS.items(): if needle in text: findings.append( Finding( severity = sev, package = pkg.display, filename = rel, pattern = "known-ioc-string", evidence = _ioc_evidence(text, needle), detail = f"{why}: {needle!r}", ) ) continue try: data = path.read_bytes() except OSError: continue text = data.decode("utf-8", errors = "replace") if rel.endswith("package.json"): findings.extend(scan_package_json(pkg, rel, text)) findings.extend(scan_text_blob(pkg, rel, text)) return findings # ───────────────────────────────────────────────────────────────────── def scan_one(pkg: PackageEntry, workspace: Path) -> tuple[list[Finding], str | None]: """Download, extract and scan a single package, cleaning up its dir. Returns (findings, error); `error` is non-None only on hard failures (download, integrity mismatch, malformed tarball), and on a clean run with findings the caller decides the exit code from severity.""" # Opaque dir: lockfile names are untrusted path components and the dir is rmtree'd. pkg_dir = Path(tempfile.mkdtemp(prefix = "pkg-", dir = workspace)) tarball = pkg_dir / "pkg.tgz" extract = pkg_dir / "x" try: _, err = download_tarball(pkg, tarball) if err: return [], err err = safe_extract(tarball, extract) if err: return [], err return scan_extracted_tree(pkg, extract), None finally: try: shutil.rmtree(pkg_dir, ignore_errors = True) except Exception: pass # ───────────────────────────────────────────────────────────────────── # Matched on (package, relative path, pattern), not evidence text, so version bumps do not reopen. # ───────────────────────────────────────────────────────────────────── _DEFAULT_BASELINE_PATH = str(Path(__file__).resolve().parent / "scan_npm_packages_baseline.json") # v3 adds an evidence hash; a pre-v3 baseline with entries is ignored (fail closed). _BASELINE_SCHEMA_VERSION = 3 def _norm_pkg_name(display: str) -> str: """``@scope/pkg@1.2.3`` / ``pkg@1.2.3`` -> name without the version. The version is the LAST ``@``-separated field; a leading ``@`` (scope) is preserved. Lower-cased, since npm names are case-insensitive. Sentinels like ```` pass through unchanged.""" s = (display or "").strip() at = s.rfind("@") if at > 0: # >0 so a leading @scope is not treated as the version sep s = s[:at] return s.lower() _NPM_TARBALL_ROOT = "package/" def _relpath_in_package(filename: str) -> str: """Path within the published package, stable across version bumps. npm tarballs root every file at ``package/``; strip it so the key is the real source path and a new file with the same basename in a different directory is not silently suppressed.""" f = (filename or "").replace("\\", "/") return f[len(_NPM_TARBALL_ROOT) :] if f.startswith(_NPM_TARBALL_ROOT) else f def _evidence_hash(evidence: str) -> str: """Stable digest of the matched evidence. The npm snippet carries no line markers, so it is already version-stable; whitespace outside string literals is collapsed (reindent-stable) while whitespace inside literals is preserved, so a changed payload body reopens but a formatter reindent does not.""" canon = _canon_preserve_strings(evidence or "") return hashlib.sha256(canon.encode("utf-8", "replace")).hexdigest() def _finding_key(f: Finding) -> tuple[str, str, str, str]: """Allowlist key: normalized package, package-relative path, pattern, and a hash of the matched evidence, so changed flagged code under an already-listed package/path/pattern reopens instead of riding the reviewed entry.""" return ( _norm_pkg_name(f.package), _relpath_in_package(f.filename), f.pattern, _evidence_hash(f.evidence or f.detail), ) def _load_baseline(path: str) -> set[tuple[str, str, str, str]]: """Load an allowlist JSON into a set of match keys. Missing file -> empty.""" try: with open(path, "r", encoding = "utf-8") as fh: data = json.load(fh) except FileNotFoundError: return set() except (OSError, json.JSONDecodeError) as exc: print(f" [WARN] could not read baseline {path}: {exc}", file = sys.stderr) return set() if not isinstance(data, dict): print(f" [WARN] baseline {path} is not a JSON object", file = sys.stderr) return set() entries = data.get("entries", []) if not isinstance(entries, list): print(f" [WARN] baseline {path} entries is not a list", file = sys.stderr) return set() if entries and data.get("version") not in (_BASELINE_SCHEMA_VERSION, 2): print( f" [WARN] baseline schema v{data.get('version')} predates package-relative " f"keys; ignoring {len(entries)} entr(y/ies). Regenerate with --write-baseline.", file = sys.stderr, ) return set() keys: set[tuple[str, str, str, str]] = set() legacy = 0 for e in entries: if not isinstance(e, dict): continue try: evidence_hash = e.get("evidence_hash") or _evidence_hash(e.get("evidence") or "") if not e.get("evidence_hash"): legacy += 1 keys.add( ( _norm_pkg_name(e["package"]), _relpath_in_package(e["file"]), e["pattern"], evidence_hash, ) ) except (KeyError, TypeError): continue if legacy: print( f" [WARN] baseline {path}: {legacy} entries lack evidence_hash and may " f"not suppress until regenerated with --write-baseline (findings reopen " f"rather than risk hiding changed code under a coarse key)", file = sys.stderr, ) return keys def _write_baseline(path: str, findings: list[Finding], threshold_rank: int) -> int: """Persist at-or-above-threshold findings as an allowlist for triage.""" entries = [] seen: set[tuple[str, str, str, str]] = set() for f in sorted(findings, key = lambda f: (_SEVERITY_RANK[f.severity], f.package)): if _SEVERITY_RANK[f.severity] > threshold_rank: continue key = _finding_key(f) if key in seen: continue seen.add(key) evidence = f.evidence or f.detail entries.append( { "package": _norm_pkg_name(f.package), "file": _relpath_in_package(f.filename), "pattern": f.pattern, "severity": f.severity, "evidence": evidence, "evidence_hash": _evidence_hash(evidence), } ) doc = { "_comment": ( "scan_npm_packages.py allowlist. Each entry is a HIGH/CRITICAL " "finding manually judged benign. Matched on (package, " "package-relative path, pattern, evidence hash); a new payload under " "an already-listed package/path/pattern reopens. severity is for " "review only. Regenerate with --write-baseline AFTER reviewing every line." ), "version": _BASELINE_SCHEMA_VERSION, "entries": entries, } with open(path, "w", encoding = "utf-8") as fh: json.dump(doc, fh, indent = 2, sort_keys = False) fh.write("\n") print(f" Wrote {len(entries)} baseline entr(y/ies) to {path}") return len(entries) def _partition_baseline( findings: list[Finding], baseline: set[tuple[str, str, str, str]] ) -> tuple[list[Finding], list[Finding]]: """Split findings into (active, suppressed) by allowlist membership.""" if not baseline: return list(findings), [] active, suppressed = [], [] for f in findings: (suppressed if _finding_key(f) in baseline else active).append(f) return active, suppressed def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description = "Pre-install npm tarball content scanner.", ) parser.add_argument( "--lockfile", default = str(REPO_ROOT / "studio" / "frontend" / "package-lock.json"), help = "Path to package-lock.json (default: studio/frontend).", ) parser.add_argument( "--max-packages", type = int, default = 0, help = ( "Cap on number of packages to scan (0 = no cap). Useful " "for local triage; CI runs with 0." ), ) parser.add_argument( "--fail-on", choices = ("info", "medium", "high", "critical"), default = "high", help = ( "Lowest severity that fails the run (default: high). " "Medium and below print but exit 0." ), ) parser.add_argument( "--baseline", metavar = "FILE", default = None, help = ( "Allowlist JSON of triaged known-good findings to suppress. " "Defaults to scan_npm_packages_baseline.json next to this script " "if present." ), ) parser.add_argument( "--no-baseline", action = "store_true", help = "Ignore the auto-discovered baseline allowlist.", ) parser.add_argument( "--write-baseline", metavar = "FILE", default = None, help = ( "Write the current at/above-threshold findings to FILE as an " "allowlist, then exit 0. Review every entry before committing it." ), ) args = parser.parse_args(argv) lockfile = Path(args.lockfile).resolve() if not lockfile.exists(): print(f"[scan-npm] lockfile not found: {lockfile}", file = sys.stderr) return 2 entries, struct_findings = parse_lockfile(lockfile) if struct_findings: print( f"[scan-npm] {len(struct_findings)} structural finding(s) " "from lockfile pass; subsequent download scan skipped for " "those entries.", flush = True, ) if args.max_packages > 0: entries = entries[: args.max_packages] workspace = Path(tempfile.mkdtemp(prefix = "npm-scan-")).resolve() atexit.register(lambda: shutil.rmtree(workspace, ignore_errors = True)) print( f"[scan-npm] workspace: {workspace}\n" f"[scan-npm] scanning {len(entries)} package(s) from {lockfile}", flush = True, ) all_findings: list[Finding] = list(struct_findings) hard_errors: list[tuple[str, str]] = [] for i, pkg in enumerate(entries, start = 1): print( f"[scan-npm] [{i}/{len(entries)}] {pkg.display}", flush = True, ) blocked = BLOCKED_NPM_VERSIONS.get(pkg.name, set()) if pkg.version in blocked: finding = Finding( severity = CRITICAL, package = pkg.display, filename = "", pattern = "blocked-known-malicious", detail = f"{pkg.name}@{pkg.version} is on the BLOCKED_NPM_VERSIONS list", ) all_findings.append(finding) print(str(finding), flush = True) continue findings, err = scan_one(pkg, workspace) if err: hard_errors.append((pkg.display, err)) print(f"[scan-npm] ERROR {pkg.display}: {err}", flush = True) continue all_findings.extend(findings) for f in findings: print(str(f), flush = True) all_findings.sort(key = lambda f: (_SEVERITY_RANK[f.severity], f.package)) print( f"\n[scan-npm] summary: {len(entries)} package(s), " f"{len(all_findings)} finding(s), " f"{len(hard_errors)} hard error(s)", flush = True, ) if hard_errors: print("\n[scan-npm] HARD ERRORS:", file = sys.stderr) for pkg, err in hard_errors: print(f" {pkg}: {err}", file = sys.stderr) threshold = { "info": INFO, "medium": MEDIUM, "high": HIGH, "critical": CRITICAL, }[args.fail_on] threshold_rank = _SEVERITY_RANK[threshold] # A hard error means the scan was incomplete, so a baseline from it would allow failures. if args.write_baseline: if hard_errors: print( f" [WARN] {len(hard_errors)} hard error(s): baseline may be " "incomplete (some packages did not scan).", file = sys.stderr, ) _write_baseline(args.write_baseline, all_findings, threshold_rank) return 0 if args.no_baseline: baseline_path = None elif args.baseline: baseline_path = args.baseline elif os.path.isfile(_DEFAULT_BASELINE_PATH): baseline_path = _DEFAULT_BASELINE_PATH else: baseline_path = None baseline = _load_baseline(baseline_path) if baseline_path else set() active, suppressed = _partition_baseline(all_findings, baseline) if suppressed: crit_s = sum(1 for f in suppressed if f.severity == CRITICAL) high_s = sum(1 for f in suppressed if f.severity == HIGH) print( f"\n[scan-npm] {len(suppressed)} finding(s) suppressed by baseline " f"{baseline_path} ({crit_s} CRITICAL, {high_s} HIGH).", flush = True, ) blocking = [f for f in active if _SEVERITY_RANK[f.severity] <= threshold_rank] if hard_errors or blocking: if blocking: print( f"\n[scan-npm] FAIL: {len(blocking)} finding(s) " f"at or above {threshold}", file = sys.stderr, ) return 1 print("\n[scan-npm] OK", flush = True) return 0 if __name__ == "__main__": sys.exit(main())