flares/scripts/scan_dependencies.py

#!/usr/bin/env python3
"""
Dependency vulnerability scanner for FLARES project.

Runs security audits on Python dependencies to detect known vulnerabilities.
Supports multiple scanners (pip-audit, safety) with fallback.

Usage:
    python scripts/scan_dependencies.py [--detailed] [--fix]
    
Environment Variables:
    SCAN_DEPENDENCIES_FAIL_FAST: Set to "true" to exit on first vulnerability
    
Exit Codes:
    0: No vulnerabilities found
    1: Vulnerabilities detected
    2: Scanner execution failed

Security Notes:
- Does not output sensitive dependency versions publicly
- Can be integrated into CI/CD pipeline
- Results can be exported in JSON format for tooling integration
"""

import os
import sys
import json
import subprocess
from pathlib import Path
from typing import List, Dict, Optional, Tuple
from datetime import datetime


class DependencyScanner:
    """
    Wrapper around multiple dependency vulnerability scanners.
    
    Falls back through scanner list if primary scanner is unavailable:
        1. pip-audit (preferred)
        2. safety (alternative)
        3. Manual requirements.txt review (last resort)
    """
    
    SCANNERS = ["pip-audit", "safety", "pipdeptree"]
    
    def __init__(self, requirements_path: str = "requirements.txt"):
        self.requirements_path = Path(requirements_path)
        self.results: Dict[str, any] = {
            "timestamp": datetime.now().isoformat(),
            "scanner_used": None,
            "vulnerabilities": [],
            "packages_scanned": 0,
            "exit_code": 2  # Default to error
        }
        
    def check_requirements_file_exists(self) -> bool:
        """Verify requirements file exists."""
        if not self.requirements_path.exists():
            print(f"❌ Requirements file not found: {self.requirements_path}")
            return False
        
        print(f"✅ Found requirements file: {self.requirements_path}")
        with open(self.requirements_path) as f:
            lines = [l.strip() for l in f if l.strip() and not l.startswith("#")]
            print(f"   Contains {len(lines)} package specifications")
        
        return True
    
    def run_pip_audit(self, detailed: bool = False) -> Tuple[bool, List[str]]:
        """
        Run pip-audit scanner (preferred method).
        
        Args:
            detailed: Show detailed vulnerability information
            
        Returns:
            Tuple of (success, list of vulnerability messages)
            
        Exit Codes:
            0: No vulnerabilities
            Non-zero: Vulnerabilities found or execution failed
        """
        print("\n🔍 Running pip-audit...")
        
        cmd = ["pip", "audit", "--requirement", str(self.requirements_path)]
        
        if not detailed:
            cmd.append("--quiet")
        
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=300  # 5 minute timeout
            )
            
            if result.returncode == 0:
                print("✅ No vulnerabilities found by pip-audit")
                self.results["scanner_used"] = "pip-audit"
                self.results["exit_code"] = 0
                return True, []
            
            # Parse vulnerability output
            messages = []
            if result.stdout:
                messages = result.stdout.strip().split("\n")
            if result.stderr:
                messages.extend(result.stderr.strip().split("\n"))
                
            print(f"⚠️  pip-audit found issues ({len(messages)} findings):")
            for msg in messages[:10]:  # Show first 10
                if msg.strip():
                    print(f"   {msg}")
            
            self.results["scanner_used"] = "pip-audit"
            self.results["vulnerabilities"] = messages
            self.results["exit_code"] = 1
            
            return False, messages
            
        except FileNotFoundError:
            print("⚠️  pip-audit not installed. Install with: pip install pip-audit")
            return None, []  # Return None to trigger fallback
        except subprocess.TimeoutExpired:
            print("❌ pip-audit timed out after 5 minutes")
            return False, ["Scanner timeout"]
    
    def run_safety_check(self, detailed: bool = False) -> Tuple[bool, List[str]]:
        """
        Run safety scanner (fallback method).
        
        Args:
            detailed: Show detailed vulnerability information
            
        Returns:
            Tuple of (success, list of vulnerability messages)
        """
        print("\n🔍 Running safety check...")
        
        cmd = ["safety", "check", "-r", str(self.requirements_path), "--json"]
        
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=300
            )
            
            if result.returncode == 0 and result.stdout:
                data = json.loads(result.stdout)
                
                if "vulnerabilities" in data and data["vulnerabilities"]:
                    print(f"⚠️  safety found {len(data['vulnerabilities'])} vulnerabilities:")
                    
                    messages = []
                    for vuln in data["vulnerabilities"][:10]:
                        msg = f"[{vuln.get('id', 'UNKNOWN')}] {vuln.get('package', '')} {vuln.get('version', '')}"
                        messages.append(msg)
                        print(f"   {msg}")
                    
                    self.results["scanner_used"] = "safety"
                    self.results["vulnerabilities"] = messages
                    self.results["exit_code"] = 1
                    
                    return False, messages
                
                print("✅ No vulnerabilities found by safety")
                self.results["scanner_used"] = "safety"
                self.results["exit_code"] = 0
                return True, []
            
            # Safety returns non-zero when vulnerabilities are found
            if result.returncode == 54:  # Standard safety exit code for vulnerabilities
                print("⚠️  safety found vulnerabilities (parse output manually)")
                self.results["scanner_used"] = "safety"
                self.results["exit_code"] = 1
                return False, [result.stdout]
            
            return False, [f"safety check failed with code {result.returncode}"]
            
        except FileNotFoundError:
            print("⚠️  safety not installed. Install with: pip install safety")
            return None, []
        except json.JSONDecodeError:
            print("❌ Invalid JSON output from safety")
            return False, ["Invalid scanner output"]
    
    def check_dependencies(self, detailed: bool = False) -> int:
        """
        Run vulnerability scan with fallback through available scanners.
        
        Args:
            detailed: Show detailed vulnerability information
            
        Returns:
            Exit code (0=clean, 1=vulnerabilities, 2=error)
        """
        # Step 1: Verify requirements file exists
        if not self.check_requirements_file_exists():
            return 2
        
        # Step 2: Try scanners in order of preference
        results = []
        
        # Try pip-audit first (preferred)
        success, messages = self.run_pip_audit(detailed)
        if success is not None:  # Not a fallback trigger
            return self.results["exit_code"]
        
        print("\n→ Falling back to safety check...")
        
        # Try safety as fallback
        success, messages = self.run_safety_check(detailed)
        if success is not None:
            return self.results["exit_code"]
        
        # Both scanners unavailable
        print("\n⚠️  No vulnerability scanners available.")
        print("   Install pip-audit: pip install pip-audit")
        print("   OR install safety: pip install safety")
        print("\n   Alternative: Manual review of dependencies recommended")
        
        self.results["scanner_used"] = "none"
        self.results["exit_code"] = 2
        return 2
    
    def export_results(self, output_path: Optional[str] = None) -> str:
        """
        Export scan results to JSON file.
        
        Args:
            output_path: Output file path (default: .security/scan_results.json)
            
        Returns:
            Path to exported file
        """
        if output_path is None:
            output_dir = Path(".security")
            output_dir.mkdir(exist_ok=True)
            output_path = str(output_dir / "scan_results.json")
        
        with open(output_path, "w") as f:
            json.dump(self.results, f, indent=2)
        
        print(f"\n📄 Results exported to: {output_path}")
        return output_path


def main() -> int:
    """Main entry point for dependency scanner."""
    import argparse
    
    parser = argparse.ArgumentParser(
        description="Scan Python dependencies for vulnerabilities",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
    python scripts/scan_dependencies.py              # Basic scan
    python scripts/scan_dependencies.py --detailed   # Verbose output
    python scripts/scan_dependencies.py --export     # Save results to JSON
    python scripts/scan_dependencies.py -r custom.txt  # Custom requirements file

Exit codes:
    0 = No vulnerabilities found
    1 = Vulnerabilities detected (review output)
    2 = Scanner execution failed (install pip-audit or safety)

CI/CD Integration:
    Add to GitHub Actions workflow:
    
    - name: Scan dependencies
      run: python scripts/scan_dependencies.py
      
    The script exits with code 1 if vulnerabilities are found, failing the build.
        """
    )
    
    parser.add_argument(
        "-r", "--requirements",
        default="requirements.txt",
        help="Path to requirements file (default: requirements.txt)"
    )
    parser.add_argument(
        "-d", "--detailed",
        action="store_true",
        help="Show detailed vulnerability information"
    )
    parser.add_argument(
        "-e", "--export",
        action="store_true",
        help="Export results to JSON file"
    )
    
    args = parser.parse_args()
    
    print("=" * 60)
    print("FLARES Dependency Vulnerability Scanner")
    print(f"Target: {args.requirements}")
    print("=" * 60)
    
    # Run scanner
    scanner = DependencyScanner(args.requirements)
    exit_code = scanner.check_dependencies(args.detailed)
    
    # Export results if requested
    if args.export:
        scanner.export_results()
    
    # Print summary
    print("\n" + "=" * 60)
    if exit_code == 0:
        print("✅ SCAN COMPLETE: No vulnerabilities detected")
    elif exit_code == 1:
        print("⚠️  SCAN COMPLETE: Vulnerabilities found (review output)")
        print(f"   Scanner used: {scanner.results.get('scanner_used', 'unknown')}")
        print(f"   Findings: {len(scanner.results.get('vulnerabilities', []))}")
    else:
        print("❌ SCAN FAILED: No scanners available or execution error")
        print("   Install pip-audit or safety to enable scanning")
    
    print("=" * 60)
    
    return exit_code


if __name__ == "__main__":
    sys.exit(main())