LiteLLM March 2026 Compromise

detect-litellm-compromise.sh

#!/usr/bin/env python3
"""
detect_litellm_compromise.py

Detects compromised LiteLLM versions (1.82.7, 1.82.8) from the March 2026
TeamPCP supply chain attack, and identifies MCP servers/projects in your
environment that depend on LiteLLM directly or transitively.

Script has no dependencies, should run standalone if you have python3.

Reference: https://futuresearch.ai/blog/no-prompt-injection-required/

Usage:
    python3 detect_litellm_compromise.py [--json] [--fix] [--deep]

    --json   Output results as JSON (for piping to other tools)
    --fix    Attempt to uninstall compromised versions (interactive)
    --deep   Also scan uvx/pipx caches, node_modules, Docker images
"""

import argparse
import glob
import importlib.metadata
import json
import os
import pathlib
import re
import shutil
import site
import subprocess
import sys
from dataclasses import dataclass, field, asdict
from enum import Enum
from typing import Optional


# -- Constants ----------------------------------------------------------------

COMPROMISED_VERSIONS = {"1.82.7", "1.82.8"}
MALICIOUS_PTH_FILE = "litellm_init.pth"
EXFIL_DOMAIN = "models.litellm.cloud"  # C2 domain used by the malware

# Known MCP servers and projects that depend on litellm (direct dependency).
# Format: (package_or_repo_name, pip_package_name_if_known)
KNOWN_LITELLM_DEPENDENTS = {
    # --- BerriAI official ---
    "litellm":                   "litellm",
    "litellm-agent-mcp":         "litellm-agent-mcp",
    "litellm-linear-mcp":        None,  # Cloudflare Worker, not pip
    # --- Community MCP servers ---
    "litellm-mcp":               "litellm-mcp",
    "LiteLLM-MCP-Server":        None,
    "mcp-litellm-proxy":         None,
    "mcp-server-litellm":        "mcp-server-litellm",
    "mcp_code_review":           "mcp-code-review",
    "Pixelle-MCP":               None,
    "blessed-mcp-server":        None,
    "infra-mcp":                 None,
    "better-code-review-graph":  None,
    # --- MCP-adjacent tools ---
    "tinyclient":                "tinyclient",
    "mcp2py":                    "mcp2py",
    "cognee":                    "cognee",
    "txtai":                     "txtai",
    # --- Frameworks with litellm as transitive dep ---
    "dspy":                      "dspy",
    "dspy-ai":                   "dspy-ai",
    "mlflow":                    "mlflow",
    "crewai":                    "crewai",
    "autogen":                   "autogen",
    "mem0ai":                    "mem0ai",
    "embedchain":                "embedchain",
    "langfuse":                  "langfuse",
    "promptflow":                "promptflow",
    "ragflow":                   None,
}


class Severity(str, Enum):
    CRITICAL = "CRITICAL"
    HIGH = "HIGH"
    MEDIUM = "MEDIUM"
    INFO = "INFO"
    OK = "OK"


@dataclass
class Finding:
    severity: Severity
    category: str
    message: str
    detail: str = ""
    remediation: str = ""


@dataclass
class ScanResult:
    findings: list[Finding] = field(default_factory=list)
    litellm_installed: bool = False
    litellm_version: Optional[str] = None
    litellm_compromised: bool = False
    pth_file_found: bool = False
    pth_file_paths: list[str] = field(default_factory=list)
    dependent_packages: list[str] = field(default_factory=list)
    c2_indicators: list[str] = field(default_factory=list)

    @property
    def max_severity(self) -> Severity:
        order = [Severity.CRITICAL, Severity.HIGH, Severity.MEDIUM, Severity.INFO, Severity.OK]
        for s in order:
            if any(f.severity == s for f in self.findings):
                return s
        return Severity.OK


# -- Detection Functions ------------------------------------------------------

def check_installed_litellm(result: ScanResult):
    """Check if litellm is installed and what version."""
    try:
        dist = importlib.metadata.distribution("litellm")
        version = dist.metadata["Version"]
        result.litellm_installed = True
        result.litellm_version = version

        if version in COMPROMISED_VERSIONS:
            result.litellm_compromised = True
            result.findings.append(Finding(
                severity=Severity.CRITICAL,
                category="compromised_package",
                message=f"Compromised litellm {version} is installed in this Python environment",
                detail=f"Location: {dist._path}" if hasattr(dist, '_path') else "",
                remediation="Immediately run: pip uninstall litellm && pip install 'litellm>=1.82.9'"
            ))
        else:
            result.findings.append(Finding(
                severity=Severity.OK,
                category="package_version",
                message=f"litellm {version} installed (not a compromised version)",
            ))
    except importlib.metadata.PackageNotFoundError:
        result.findings.append(Finding(
            severity=Severity.INFO,
            category="package_version",
            message="litellm is not installed in the current Python environment",
        ))


def check_pth_files(result: ScanResult):
    """Scan site-packages for the malicious .pth file."""
    search_dirs = set()
    search_dirs.update(site.getsitepackages())
    search_dirs.add(site.getusersitepackages())
    # Also check common venv/conda paths
    for env_var in ("VIRTUAL_ENV", "CONDA_PREFIX"):
        prefix = os.environ.get(env_var)
        if prefix:
            for pattern in ("lib/python*/site-packages", "lib/site-packages"):
                search_dirs.update(glob.glob(os.path.join(prefix, pattern)))

    for d in search_dirs:
        pth_path = os.path.join(d, MALICIOUS_PTH_FILE)
        if os.path.exists(pth_path):
            result.pth_file_found = True
            result.pth_file_paths.append(pth_path)
            # Read first 200 bytes to check for known malware signatures
            try:
                content = pathlib.Path(pth_path).read_text(errors="replace")[:500]
                has_import = "import " in content
            except Exception:
                has_import = False

            result.findings.append(Finding(
                severity=Severity.CRITICAL,
                category="malicious_pth",
                message=f"Malicious .pth file found: {pth_path}",
                detail=f"Contains import statement: {has_import}. "
                       "This file executes on EVERY Python process startup.",
                remediation=f"Delete immediately: rm '{pth_path}' — then rotate ALL credentials on this machine."
            ))

    if not result.pth_file_found:
        result.findings.append(Finding(
            severity=Severity.OK,
            category="malicious_pth",
            message=f"No {MALICIOUS_PTH_FILE} found in any site-packages directory",
        ))


def check_dependent_packages(result: ScanResult):
    """Check for installed packages known to depend on litellm."""
    installed = {d.metadata["Name"].lower(): d.metadata["Version"]
                 for d in importlib.metadata.distributions()}

    for name, pip_name in KNOWN_LITELLM_DEPENDENTS.items():
        check_name = (pip_name or name).lower()
        if check_name in installed and check_name != "litellm":
            version = installed[check_name]
            result.dependent_packages.append(f"{check_name}=={version}")
            sev = Severity.HIGH if result.litellm_compromised else Severity.MEDIUM
            result.findings.append(Finding(
                severity=sev,
                category="litellm_dependent",
                message=f"{check_name}=={version} is installed (depends on litellm)",
                detail="This package may have pulled in litellm as a transitive dependency.",
                remediation=f"Audit: pip show {check_name} | grep -i requires"
            ))

    if not result.dependent_packages:
        result.findings.append(Finding(
            severity=Severity.OK,
            category="litellm_dependent",
            message="No known litellm-dependent MCP servers or frameworks installed",
        ))


def check_pip_cache(result: ScanResult):
    """Check pip's download cache for compromised wheels."""
    cache_dir = subprocess.run(
        [sys.executable, "-m", "pip", "cache", "dir"],
        capture_output=True, text=True
    )
    if cache_dir.returncode != 0:
        return

    cache_path = cache_dir.stdout.strip()
    if not os.path.isdir(cache_path):
        return

    for root, _, files in os.walk(cache_path):
        for f in files:
            if "litellm" in f.lower() and any(v in f for v in COMPROMISED_VERSIONS):
                full = os.path.join(root, f)
                result.findings.append(Finding(
                    severity=Severity.HIGH,
                    category="cached_compromise",
                    message=f"Compromised litellm wheel in pip cache: {full}",
                    remediation=f"Remove: rm '{full}' or run: pip cache remove litellm"
                ))


def check_uvx_pipx_caches(result: ScanResult):
    """Check uvx/pipx tool caches for litellm (deep scan)."""
    home = pathlib.Path.home()
    cache_dirs = [
        home / ".local" / "share" / "uv",
        home / ".cache" / "uv",
        home / ".local" / "share" / "pipx",
        home / ".local" / "pipx",
    ]

    for cache_root in cache_dirs:
        if not cache_root.exists():
            continue
        try:
            matches = list(cache_root.rglob("litellm*"))
            for match in matches:
                name = match.name
                if any(v in name for v in COMPROMISED_VERSIONS):
                    result.findings.append(Finding(
                        severity=Severity.CRITICAL,
                        category="tool_cache_compromise",
                        message=f"Compromised litellm in tool cache: {match}",
                        detail=f"uvx/pipx may auto-install this on MCP server startup.",
                        remediation=f"Remove the cache entry and re-install: rm -rf '{match}'"
                    ))
                elif match.is_dir() or match.suffix in (".whl", ".tar.gz"):
                    result.findings.append(Finding(
                        severity=Severity.INFO,
                        category="tool_cache_litellm",
                        message=f"litellm found in tool cache: {match}",
                        detail="Verify version is >= 1.82.9"
                    ))
        except PermissionError:
            pass


def check_c2_indicators(result: ScanResult):
    """Check for indicators of C2 communication."""
    # 1. Check /etc/hosts for the exfil domain
    try:
        hosts = pathlib.Path("/etc/hosts").read_text()
        if EXFIL_DOMAIN in hosts:
            result.c2_indicators.append(f"/etc/hosts contains {EXFIL_DOMAIN}")
    except Exception:
        pass

    # 2. Check DNS cache (systemd-resolve) if available
    if shutil.which("resolvectl"):
        try:
            dns = subprocess.run(
                ["resolvectl", "query", EXFIL_DOMAIN],
                capture_output=True, text=True, timeout=5
            )
            if dns.returncode == 0:
                result.c2_indicators.append(f"DNS resolution succeeded for {EXFIL_DOMAIN}")
        except Exception:
            pass

    # 3. Check shell history for the domain
    history_files = [
        pathlib.Path.home() / ".bash_history",
        pathlib.Path.home() / ".zsh_history",
        pathlib.Path.home() / ".local" / "share" / "nushell" / "history.txt",
        pathlib.Path.home() / ".local" / "share" / "fish" / "fish_history",
    ]
    for hf in history_files:
        try:
            if hf.exists() and EXFIL_DOMAIN in hf.read_text(errors="replace"):
                result.c2_indicators.append(f"C2 domain found in {hf}")
        except Exception:
            pass

    # 4. Check for the systemd persistence backdoor
    systemd_user = pathlib.Path.home() / ".config" / "systemd" / "user"
    if systemd_user.exists():
        for unit in systemd_user.iterdir():
            try:
                content = unit.read_text(errors="replace")
                if EXFIL_DOMAIN in content or "litellm_init" in content:
                    result.c2_indicators.append(f"Suspicious systemd unit: {unit}")
            except Exception:
                pass

    # 5. Check for credential staging directories the malware used
    staging_dirs = [
        pathlib.Path("/tmp/.litellm_cache"),
        pathlib.Path("/tmp/.llm_telemetry"),
        pathlib.Path("/var/tmp/.litellm_cache"),
    ]
    for sd in staging_dirs:
        if sd.exists():
            result.c2_indicators.append(f"Suspicious staging directory: {sd}")

    if result.c2_indicators:
        for indicator in result.c2_indicators:
            result.findings.append(Finding(
                severity=Severity.CRITICAL,
                category="c2_indicator",
                message=f"C2/exfiltration indicator: {indicator}",
                remediation="This machine may be actively compromised. "
                            "Isolate immediately, rotate ALL credentials, "
                            "and conduct a full forensic investigation."
            ))
    else:
        result.findings.append(Finding(
            severity=Severity.OK,
            category="c2_indicator",
            message="No C2/exfiltration indicators found",
        ))


def check_credential_exposure(result: ScanResult):
    """Check if high-value credential files exist (the malware's targets)."""
    home = pathlib.Path.home()
    targets = {
        "SSH private keys":       home / ".ssh" / "id_rsa",
        "SSH ed25519 keys":       home / ".ssh" / "id_ed25519",
        "AWS credentials":        home / ".aws" / "credentials",
        "GCP service account":    home / ".config" / "gcloud" / "application_default_credentials.json",
        "Azure CLI tokens":       home / ".azure" / "accessTokens.json",
        "Kubeconfig":             home / ".kube" / "config",
        "Docker config (creds)":  home / ".docker" / "config.json",
    }

    exposed = []
    for label, path in targets.items():
        if path.exists():
            exposed.append(label)

    if result.litellm_compromised and exposed:
        result.findings.append(Finding(
            severity=Severity.CRITICAL,
            category="credential_exposure",
            message=f"Compromised litellm + {len(exposed)} credential files present on disk",
            detail=f"At risk: {', '.join(exposed)}",
            remediation="ASSUME THESE ARE EXFILTRATED. Rotate immediately: "
                        "SSH keys, cloud provider credentials, Kubernetes tokens, Docker registry creds."
        ))
    elif exposed:
        result.findings.append(Finding(
            severity=Severity.INFO,
            category="credential_exposure",
            message=f"{len(exposed)} credential files present (not at risk unless compromised version was installed)",
            detail=f"Files: {', '.join(exposed)}",
        ))


def check_docker_images(result: ScanResult):
    """Check local Docker images for litellm (deep scan)."""
    if not shutil.which("docker"):
        return

    try:
        images = subprocess.run(
            ["docker", "images", "--format", "{{.Repository}}:{{.Tag}}"],
            capture_output=True, text=True, timeout=10
        )
        if images.returncode != 0:
            return

        for line in images.stdout.strip().splitlines():
            if "litellm" in line.lower():
                sev = Severity.HIGH if any(v in line for v in COMPROMISED_VERSIONS) else Severity.MEDIUM
                result.findings.append(Finding(
                    severity=sev,
                    category="docker_image",
                    message=f"Docker image contains litellm: {line}",
                    detail="LiteLLM Docker images have 57+ known vulnerabilities.",
                    remediation=f"Inspect: docker run --rm --entrypoint pip {line} show litellm"
                ))
    except Exception:
        pass


def check_mcp_configs(result: ScanResult):
    """Check local MCP config files for servers that might use litellm."""
    home = pathlib.Path.home()
    config_locations = [
        home / ".cursor" / "mcp.json",
        home / ".config" / "cursor" / "mcp.json",
        home / ".config" / "claude" / "mcp.json",
        home / ".claude" / "mcp_settings.json",
        home / "Library" / "Application Support" / "Cursor" / "mcp.json",  # macOS
        home / "Library" / "Application Support" / "Claude" / "mcp.json",  # macOS
    ]

    # Also check .mcp.json in common project roots
    cwd = pathlib.Path.cwd()
    for p in [cwd, cwd.parent]:
        config_locations.append(p / ".mcp.json")
        config_locations.append(p / "mcp.json")

    for config_path in config_locations:
        if not config_path.exists():
            continue
        try:
            content = config_path.read_text()
            data = json.loads(content)
            # Look for servers that reference litellm in command, args, or env
            servers = data.get("mcpServers", data.get("servers", {}))
            for name, cfg in servers.items():
                cfg_str = json.dumps(cfg).lower()
                if "litellm" in cfg_str:
                    cmd = cfg.get("command", "")
                    result.findings.append(Finding(
                        severity=Severity.HIGH,
                        category="mcp_config",
                        message=f"MCP config references litellm: server '{name}' in {config_path}",
                        detail=f"Command: {cmd}",
                        remediation="Pin the litellm version in this server's dependencies "
                                    "and verify it's >= 1.82.9"
                    ))
        except (json.JSONDecodeError, KeyError):
            pass


# -- Output -------------------------------------------------------------------

SEVERITY_COLORS = {
    Severity.CRITICAL: "\033[1;91m",  # bold bright red
    Severity.HIGH:     "\033[91m",    # bright red
    Severity.MEDIUM:   "\033[93m",    # yellow
    Severity.INFO:     "\033[36m",    # cyan
    Severity.OK:       "\033[92m",    # green
}
RESET = "\033[0m"
BOLD = "\033[1m"


def print_findings(result: ScanResult, use_json: bool):
    if use_json:
        output = {
            "max_severity": result.max_severity.value,
            "litellm_installed": result.litellm_installed,
            "litellm_version": result.litellm_version,
            "litellm_compromised": result.litellm_compromised,
            "pth_file_found": result.pth_file_found,
            "pth_file_paths": result.pth_file_paths,
            "dependent_packages": result.dependent_packages,
            "c2_indicators": result.c2_indicators,
            "findings": [asdict(f) for f in result.findings],
        }
        print(json.dumps(output, indent=2, default=str))
        return

    print(f"\n{BOLD}{'=' * 72}")
    print(f"  LiteLLM Supply Chain Compromise — Detection Report")
    print(f"{'=' * 72}{RESET}\n")

    max_sev = result.max_severity
    color = SEVERITY_COLORS[max_sev]
    print(f"  Overall: {color}{BOLD}{max_sev.value}{RESET}\n")

    if result.litellm_installed:
        v_color = SEVERITY_COLORS[Severity.CRITICAL] if result.litellm_compromised else SEVERITY_COLORS[Severity.OK]
        print(f"  litellm version: {v_color}{result.litellm_version}{RESET}")
    else:
        print(f"  litellm: {SEVERITY_COLORS[Severity.INFO]}not installed{RESET}")

    if result.dependent_packages:
        print(f"  Dependents:      {', '.join(result.dependent_packages)}")
    print()

    # Group findings by severity
    for severity in [Severity.CRITICAL, Severity.HIGH, Severity.MEDIUM, Severity.INFO, Severity.OK]:
        group = [f for f in result.findings if f.severity == severity]
        if not group:
            continue

        color = SEVERITY_COLORS[severity]
        print(f"  {color}{BOLD}[{severity.value}]{RESET}")
        for f in group:
            print(f"  {color}  • {f.message}{RESET}")
            if f.detail:
                print(f"      {f.detail}")
            if f.remediation and severity in (Severity.CRITICAL, Severity.HIGH):
                print(f"      → {BOLD}{f.remediation}{RESET}")
        print()

    print(f"{BOLD}{'=' * 72}{RESET}")
    if result.litellm_compromised:
        print(f"  {SEVERITY_COLORS[Severity.CRITICAL]}{BOLD}")
        print(f"  ⚠  COMPROMISED VERSION DETECTED")
        print(f"     1. Disconnect this machine from the network")
        print(f"     2. Rotate ALL credentials (SSH, cloud, k8s, API keys)")
        print(f"     3. pip uninstall litellm && pip install 'litellm>=1.82.9'")
        print(f"     4. rm any {MALICIOUS_PTH_FILE} files listed above")
        print(f"     5. Conduct full forensic review{RESET}")
    elif max_sev in (Severity.HIGH, Severity.MEDIUM):
        print(f"  {SEVERITY_COLORS[Severity.MEDIUM]}  Potential exposure detected — review findings above.{RESET}")
    else:
        print(f"  {SEVERITY_COLORS[Severity.OK]}  No compromise indicators found.{RESET}")
    print()


def attempt_fix(result: ScanResult):
    """Interactive remediation for compromised installations."""
    if not result.litellm_compromised and not result.pth_file_found:
        print("Nothing to fix — no compromised version or malicious files detected.")
        return

    print(f"\n{BOLD}Remediation steps:{RESET}\n")

    if result.pth_file_found:
        for p in result.pth_file_paths:
            resp = input(f"  Delete malicious file {p}? [y/N] ").strip().lower()
            if resp == "y":
                try:
                    os.remove(p)
                    print(f"    ✓ Deleted {p}")
                except Exception as e:
                    print(f"    ✗ Failed: {e}")

    if result.litellm_compromised:
        resp = input(f"\n  Uninstall litellm {result.litellm_version} and install safe version? [y/N] ").strip().lower()
        if resp == "y":
            subprocess.run([sys.executable, "-m", "pip", "uninstall", "-y", "litellm"])
            subprocess.run([sys.executable, "-m", "pip", "install", "litellm>=1.82.9"])
            print("    ✓ Reinstalled litellm")

    print(f"\n  {SEVERITY_COLORS[Severity.CRITICAL]}{BOLD}IMPORTANT: You must still rotate all credentials.{RESET}")
    print(f"  The malware may have already exfiltrated SSH keys, cloud creds, and k8s configs.\n")


# -- Main --------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description="Detect compromised LiteLLM (TeamPCP supply chain attack, March 2026)"
    )
    parser.add_argument("--json", action="store_true", help="Output as JSON")
    parser.add_argument("--fix", action="store_true", help="Interactive remediation")
    parser.add_argument("--deep", action="store_true", help="Scan uvx/pipx caches, Docker images")
    args = parser.parse_args()

    result = ScanResult()

    # Core checks (always run)
    check_installed_litellm(result)
    check_pth_files(result)
    check_dependent_packages(result)
    check_pip_cache(result)
    check_c2_indicators(result)
    check_credential_exposure(result)
    check_mcp_configs(result)

    # Deep checks (optional, slower)
    if args.deep:
        check_uvx_pipx_caches(result)
        check_docker_images(result)

    print_findings(result, use_json=args.json)

    if args.fix:
        attempt_fix(result)

    # Exit code: 2=critical, 1=high/medium, 0=ok/info
    if result.max_severity == Severity.CRITICAL:
        sys.exit(2)
    elif result.max_severity in (Severity.HIGH, Severity.MEDIUM):
        sys.exit(1)
    sys.exit(0)


if __name__ == "__main__":
    main()