#!/usr/bin/env -S uv --quiet run --script # /// script # requires-python = ">=3.12" # dependencies = [] # /// """ Scan JavaScript/TypeScript dependency files for malicious packages. This script scans all repositories for common JS/TS dependency files (package.json, yarn.lock, package-lock.json, etc.) and checks them against a CSV of known malicious packages. """ import json import csv import re from pathlib import Path from typing import Set, Dict, List import argparse import logging logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger(__name__) # Common JavaScript/TypeScript dependency files to scan JS_DEPENDENCY_FILES = { 'package.json', 'package-lock.json', 'yarn.lock', 'npm-shrinkwrap.json', 'pnpm-lock.yaml', 'bun.lockb', 'lerna.json' } class MaliciousPackageScanner: def __init__(self, malicious_csv_path: str, repos_base_path: str): self.malicious_csv_path = Path(malicious_csv_path) self.repos_base_path = Path(repos_base_path) self.malicious_packages: Set[str] = set() self.malicious_scoped_packages: Set[str] = set() # Scoped malicious packages (@author/package) self.malicious_unscoped_packages: Set[str] = set() # Unscoped malicious packages (package) self.findings: List[Dict[str, str]] = [] self.author_matches: List[Dict[str, str]] = [] self.malicious_authors: Set[str] = set() def load_malicious_packages(self): """Load malicious package names from CSV file.""" try: with open(self.malicious_csv_path, 'r', encoding='utf-8') as f: print('opened file') reader = csv.DictReader(f) for row in reader: # CSV has "Package" and "Version" columns package_name = row.get('Package', '').strip() print(package_name) if package_name: # Store in the main set self.malicious_packages.add(package_name) # Separate scoped and unscoped packages if '/' in package_name and package_name.startswith('@'): # This is a scoped package (@author/package) self.malicious_scoped_packages.add(package_name) author_name = package_name.split('/')[0] # Gets @author part self.malicious_authors.add(author_name) else: # This is an unscoped package self.malicious_unscoped_packages.add(package_name) logger.info(f"Loaded {len(self.malicious_packages)} malicious packages from CSV") logger.info(f"Identified {len(self.malicious_authors)} malicious authors") except Exception as e: logger.error(f"Failed to load malicious packages CSV: {e}") raise def find_dependency_files(self, repo_path: Path) -> List[Path]: """Find all JS/TS dependency files in a repository.""" dependency_files = [] # Search recursively but skip common directories to ignore skip_dirs = {'.git', 'node_modules', '.venv', '__pycache__', 'dist', 'build', 'coverage'} def should_skip_dir(path: Path) -> bool: return any(part in skip_dirs for part in path.parts) for file_path in repo_path.rglob('*'): if file_path.is_file() and file_path.name in JS_DEPENDENCY_FILES: if not should_skip_dir(file_path): dependency_files.append(file_path) return dependency_files def extract_packages_from_package_json(self, file_path: Path) -> Set[str]: """Extract package names from package.json files.""" packages = set() try: with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) # Check dependencies, devDependencies, peerDependencies, optionalDependencies for dep_type in ['dependencies', 'devDependencies', 'peerDependencies', 'optionalDependencies']: if dep_type in data and isinstance(data[dep_type], dict): packages.update(data[dep_type].keys()) except Exception as e: logger.debug(f"Error parsing {file_path}: {e}") return packages def extract_packages_from_lock_files(self, file_path: Path) -> Set[str]: """Extract package names from lock files (yarn.lock, package-lock.json, etc.).""" packages = set() try: if file_path.name == 'yarn.lock': # Parse yarn.lock format with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # Yarn.lock format: "package-name@version:", "package-name@npm:version:" yarn_pattern = re.compile(r'^"?([^@\s"]+)@', re.MULTILINE) matches = yarn_pattern.findall(content) packages.update(matches) elif file_path.name == 'package-lock.json': # Parse package-lock.json with open(file_path, 'r', encoding='utf-8') as f: data = json.load(f) # Check both dependencies and packages sections if 'dependencies' in data: packages.update(data['dependencies'].keys()) if 'packages' in data: for pkg_path in data['packages'].keys(): if pkg_path and pkg_path != '': # Skip root package # Extract package name from node_modules path parts = pkg_path.split('/') if 'node_modules' in parts: idx = parts.index('node_modules') if idx + 1 < len(parts): pkg_name = parts[idx + 1] if pkg_name.startswith('@') and idx + 2 < len(parts): pkg_name = f"{pkg_name}/{parts[idx + 2]}" packages.add(pkg_name) elif file_path.name == 'pnpm-lock.yaml': # Basic pnpm-lock.yaml parsing (would need PyYAML for full support) with open(file_path, 'r', encoding='utf-8') as f: content = f.read() # Look for package references in the format "/package-name/version" pnpm_pattern = re.compile(r'/([^/\s:]+)/') matches = pnpm_pattern.findall(content) packages.update(matches) except Exception as e: logger.debug(f"Error parsing lock file {file_path}: {e}") return packages def scan_file(self, file_path: Path, repo_name: str) -> List[str]: """Scan a single dependency file for malicious packages.""" found_malicious = [] if file_path.name == 'package.json' or file_path.name == 'lerna.json': packages = self.extract_packages_from_package_json(file_path) else: packages = self.extract_packages_from_lock_files(file_path) # Track scoped packages by author for author matching scoped_packages_by_author = {} for package in packages: if '/' in package and package.startswith('@'): author = package.split('/')[0] if author not in scoped_packages_by_author: scoped_packages_by_author[author] = [] scoped_packages_by_author[author].append(package) # Check for author matches with malicious packages for author in scoped_packages_by_author: if author in self.malicious_authors: # Find which malicious packages this author has published malicious_by_author = [pkg for pkg in self.malicious_packages if pkg.startswith(author + '/')] for malicious_pkg in malicious_by_author: for scoped_pkg in scoped_packages_by_author[author]: self.author_matches.append({ 'repository': repo_name, 'author_name': author, 'malicious_package': malicious_pkg, 'scoped_package_found': scoped_pkg, 'file': str(file_path.relative_to(self.repos_base_path / repo_name)) }) # Check for malicious packages for package in packages: is_scoped_package = '/' in package and package.startswith('@') package_name_only = package.split('/')[-1] if is_scoped_package else package matched = False # Rule 1: If malicious package is scoped, only match exactly if package in self.malicious_scoped_packages: found_malicious.append(package) self.findings.append({ 'repository': repo_name, 'file': str(file_path.relative_to(self.repos_base_path / repo_name)), 'malicious_package': package }) matched = True # Rule 2: If malicious package is unscoped, match both scoped and unscoped occurrences if not matched: # Check if the base package name (without scope) matches an unscoped malicious package if package_name_only in self.malicious_unscoped_packages: found_malicious.append(package) # Include the full package name (with scope if present) in findings self.findings.append({ 'repository': repo_name, 'file': str(file_path.relative_to(self.repos_base_path / repo_name)), 'malicious_package': package # This shows the actual package found, including scope }) matched = True return found_malicious def scan_repository(self, repo_path: Path) -> int: """Scan a single repository for malicious packages.""" repo_name = repo_path.name dependency_files = self.find_dependency_files(repo_path) if not dependency_files: return 0 total_found = 0 logger.info(f"Scanning {repo_name}: found {len(dependency_files)} dependency files") for file_path in dependency_files: malicious_found = self.scan_file(file_path, repo_name) if malicious_found: logger.warning(f"Found malicious packages in {repo_name}/{file_path.name}: {malicious_found}") total_found += len(malicious_found) return total_found def scan_all_repositories(self) -> None: """Scan all repositories in the base directory.""" if not self.repos_base_path.exists(): logger.error(f"Repository base path does not exist: {self.repos_base_path}") return # Find all subdirectories that look like repositories repo_dirs = [d for d in self.repos_base_path.iterdir() if d.is_dir() and not d.name.startswith('.')] logger.info(f"Found {len(repo_dirs)} potential repositories to scan") total_malicious_found = 0 repos_with_issues = 0 for repo_dir in repo_dirs: found_count = self.scan_repository(repo_dir) if found_count > 0: repos_with_issues += 1 total_malicious_found += found_count logger.info(f"Scan complete: {total_malicious_found} malicious packages found across {repos_with_issues} repositories") if self.author_matches: logger.info(f"Found {len(self.author_matches)} author matches between malicious and dependency packages") def write_results_csv(self, output_path: str) -> None: """Write findings to CSV file.""" output_file = Path(output_path) with open(output_file, 'w', newline='', encoding='utf-8') as f: if self.findings: writer = csv.DictWriter(f, fieldnames=['repository', 'file', 'malicious_package']) writer.writeheader() writer.writerows(self.findings) logger.info(f"Results written to {output_file} ({len(self.findings)} findings)") def write_author_matches_csv(self, output_path: str = 'author_matches.csv') -> None: """Write author matches to CSV file.""" output_file = Path(output_path) with open(output_file, 'w', newline='', encoding='utf-8') as f: if self.author_matches: writer = csv.DictWriter(f, fieldnames=['repository', 'author_name', 'malicious_package', 'scoped_package_found', 'file']) writer.writeheader() writer.writerows(self.author_matches) logger.info(f"Author matches written to {output_file} ({len(self.author_matches)} matches)") def main(): parser = argparse.ArgumentParser(description='Scan repositories for malicious JavaScript/TypeScript packages') parser.add_argument('--malicious-csv', default='malicious-packages.csv', help='Path to malicious packages CSV file') parser.add_argument('--repos-path', default='.', help='Base path containing repositories to scan') parser.add_argument('--output', default='malicious_findings.csv', help='Output CSV file for findings') args = parser.parse_args() scanner = MaliciousPackageScanner(args.malicious_csv, args.repos_path) try: scanner.load_malicious_packages() scanner.scan_all_repositories() scanner.write_results_csv(args.output) scanner.write_author_matches_csv('author_matches.csv') if scanner.findings: logger.warning(f"⚠️ Found {len(scanner.findings)} instances of malicious packages!") logger.warning(f"📄 Details saved to {args.output}") else: logger.info("✅ No malicious packages found in scanned repositories") if scanner.author_matches: logger.warning(f"⚠️ Found {len(scanner.author_matches)} author matches!") logger.warning("📄 Author matches saved to author_matches.csv") except Exception as e: logger.error(f"Scanner failed: {e}") return 1 return 0 if __name__ == '__main__': exit(main())