Other Package Ecosystems & Developer Protection¶

#!/usr/bin/env python3
"""
Multi-ecosystem package installation monitor
"""

import os
import subprocess
import json
import logging
from datetime import datetime
from pathlib import Path

class PackageMonitor:
    def __init__(self):
        self.log_file = Path.home() / ".package_monitor.log"
        self.setup_logging()

    def setup_logging(self):
        logging.basicConfig(
            filename=self.log_file,
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )

    def monitor_npm_install(self, command_args):
        """Monitor npm install commands"""
        packages = self.extract_npm_packages(command_args)
        for package in packages:
            self.analyze_package_risk('npm', package)

    def monitor_pip_install(self, command_args):
        """Monitor pip install commands"""
        packages = self.extract_pip_packages(command_args)
        for package in packages:
            self.analyze_package_risk('pip', package)

    def analyze_package_risk(self, ecosystem, package_name):
        """Analyze package for risk indicators"""
        risk_score = 0
        risk_factors = []

        # Check package age
        age_days = self.get_package_age(ecosystem, package_name)
        if age_days < 30:
            risk_score += 3
            risk_factors.append("Very new package")
        elif age_days < 90:
            risk_score += 1
            risk_factors.append("Relatively new package")

        # Check download stats
        downloads = self.get_download_stats(ecosystem, package_name)
        if downloads < 1000:
            risk_score += 3
            risk_factors.append("Low download count")
        elif downloads < 10000:
            risk_score += 1
            risk_factors.append("Moderate download count")

        # Check for typosquatting
        if self.check_typosquatting(package_name):
            risk_score += 5
            risk_factors.append("Potential typosquatting")

        # Log risk assessment
        logging.info(f"Package: {ecosystem}/{package_name}, Risk Score: {risk_score}, Factors: {risk_factors}")

        # Alert on high risk
        if risk_score >= 5:
            self.send_alert(ecosystem, package_name, risk_score, risk_factors)

    def send_alert(self, ecosystem, package_name, risk_score, factors):
        """Send alert for high-risk packages"""
        alert_msg = f"""
        🚨 HIGH RISK PACKAGE DETECTED 🚨
        Ecosystem: {ecosystem}
        Package: {package_name}
        Risk Score: {risk_score}/10
        Risk Factors: {', '.join(factors)}
        Time: {datetime.now()}

        Please review this package before proceeding with installation.
        """

        # Log to console
        print(alert_msg)

        # Could also send to Slack, email, etc.
        self.log_to_security_system(alert_msg)

# Usage as command wrapper
monitor = PackageMonitor()

# Wrap package managers
def secure_npm(*args):
    monitor.monitor_npm_install(args)
    return subprocess.run(['npm'] + list(args))

def secure_pip(*args):
    monitor.monitor_pip_install(args)
    return subprocess.run(['pip'] + list(args))

#!/usr/bin/env python3
"""
Automated package compromise response system
"""

import json
import subprocess
import requests
from datetime import datetime
import smtplib
from email.mime.text import MIMEText

class PackageIncidentResponse:
    def __init__(self):
        self.compromised_packages = set()
        self.affected_projects = []

    def detect_compromise(self, ecosystem, package_name, malicious_versions):
        """Detect if compromised package is in use"""
        print(f"🚨 Checking for compromise: {ecosystem}/{package_name}")

        # Add to compromised list
        self.compromised_packages.add(f"{ecosystem}/{package_name}")

        # Scan all projects
        self.scan_projects(ecosystem, package_name, malicious_versions)

        if self.affected_projects:
            self.execute_response()

    def scan_projects(self, ecosystem, package_name, malicious_versions):
        """Scan all projects for compromised package"""
        projects_dir = "/path/to/projects"

        for project in os.listdir(projects_dir):
            project_path = os.path.join(projects_dir, project)

            if ecosystem == "npm":
                self.scan_npm_project(project_path, package_name, malicious_versions)
            elif ecosystem == "pip":
                self.scan_python_project(project_path, package_name, malicious_versions)
            # Add other ecosystems...

    def execute_response(self):
        """Execute incident response procedures"""
        # 1. Notify security team
        self.notify_security_team()

        # 2. Block access to compromised packages
        self.block_package_access()

        # 3. Generate remediation plan
        self.generate_remediation_plan()

        # 4. Execute automated fixes where safe
        self.execute_safe_fixes()

    def notify_security_team(self):
        """Send notifications to security team"""
        subject = f"Package Compromise Detected - {len(self.affected_projects)} projects affected"
        body = f"""
        Compromised packages detected:
        {list(self.compromised_packages)}

        Affected projects:
        {self.affected_projects}

        Immediate action required.
        """

        # Send email, Slack notification, etc.
        self.send_notification(subject, body)

    def block_package_access(self):
        """Block access to compromised packages"""
        # Update firewall rules, proxy configurations, etc.
        for package in self.compromised_packages:
            print(f"Blocking access to {package}")
            # Implementation depends on infrastructure

    def generate_remediation_plan(self):
        """Generate step-by-step remediation plan"""
        plan = {
            "affected_projects": self.affected_projects,
            "recommended_actions": [
                "Update to safe version",
                "Review system for indicators of compromise",
                "Rotate any exposed credentials",
                "Monitor for unusual activity"
            ],
            "timeline": "Execute within 4 hours"
        }

        with open("incident_response_plan.json", "w") as f:
            json.dump(plan, f, indent=2)

# Usage example
response_system = PackageIncidentResponse()
response_system.detect_compromise("npm", "eslint-config-prettier", ["9.1.0"])