scan_litellm.py

scan_litellm.py

#!/usr/bin/env python3
"""
LiteLLM Supply Chain Attack Scanner (March 2026)

Scans directories and the system for LiteLLM installations affected by the
TeamPCP backdoor (versions 1.82.7 and 1.82.8).

IOCs sourced from: ARMO, Sonatype, Wiz, Snyk, LiteLLM official advisory.

Usage:
    python scan_litellm.py /path/to/folder1 /path/to/folder2 ...
    python scan_litellm.py --system                  # scan all Python environments on the machine
    python scan_litellm.py /path/to/folder --deep    # also check dependency files
    python scan_litellm.py --system --deep            # full system scan
"""

import argparse
import os
import re
import subprocess
import sys
from dataclasses import dataclass, field
from pathlib import Path

# --- Known IOCs from TeamPCP / March 2026 attack ---

COMPROMISED_VERSIONS = {"1.82.7", "1.82.8"}

# Malicious files dropped by the payload
MALICIOUS_FILES_SITE_PACKAGES = [
    "litellm_init.pth",       # v1.82.8 - auto-executes on any Python startup
]

MALICIOUS_FILES_SYSTEM = [
    "/tmp/pglog",              # staging file used by the payload
    "/tmp/.pg_state",          # persistence state file
]

MALICIOUS_DROPPED_FILES = [
    "sysmon.py",               # persistence mechanism written to disk
    "tpcp.tar.gz",             # exfiltration archive
]

# C2 domains
C2_DOMAINS = [
    "models.litellm.cloud",    # fake LiteLLM domain for exfil
    "checkmarx.zone",          # secondary C2
]

# Suspicious strings in source code
SUSPICIOUS_STRINGS = [
    "tpcp",
    "models.litellm.cloud",
    "checkmarx.zone",
    "tpcp.tar.gz",
    "sysmon.py",
    "pglog",
    ".pg_state",
    "tpcp-docs",
]

RED = "\033[91m"
GREEN = "\033[92m"
YELLOW = "\033[93m"
CYAN = "\033[96m"
BOLD = "\033[1m"
RESET = "\033[0m"


@dataclass
class Finding:
    path: str
    version: str
    source: str  # "installed", "dependency_file", "system"
    is_compromised: bool


@dataclass
class IOCResult:
    severity: str   # "CRITICAL", "SUSPICIOUS", "INFO"
    message: str
    path: str = ""


def get_installed_version(litellm_path: Path) -> str | None:
    """Extract version from installed litellm package metadata."""
    site_packages = litellm_path.parent
    for item in site_packages.iterdir():
        if item.is_dir() and item.name.startswith("litellm-") and item.name.endswith(".dist-info"):
            metadata = item / "METADATA"
            if metadata.exists():
                for line in metadata.read_text().splitlines():
                    if line.startswith("Version:"):
                        return line.split(":", 1)[1].strip()
    version_file = litellm_path / "version.py"
    if version_file.exists():
        content = version_file.read_text()
        match = re.search(r'version\s*=\s*["\']([^"\']+)["\']', content)
        if match:
            return match.group(1)
    return None


def check_iocs(litellm_path: Path) -> list[IOCResult]:
    """Run all IOC checks against an installed litellm package."""
    results = []
    site_packages = litellm_path.parent

    # 1. Malicious .pth file (v1.82.8 auto-exec payload)
    for fname in MALICIOUS_FILES_SITE_PACKAGES:
        target = site_packages / fname
        if target.exists():
            results.append(IOCResult(
                severity="CRITICAL",
                message=f"Malicious file found: {fname} (auto-executes on Python startup)",
                path=str(target),
            ))

    # 2. Malicious proxy_server.py payload (v1.82.7 + v1.82.8)
    proxy_server = litellm_path / "proxy" / "proxy_server.py"
    if proxy_server.exists():
        try:
            content = proxy_server.read_text()
            # Large base64 blobs indicate injected payload
            b64_blobs = re.findall(r'[A-Za-z0-9+/]{200,}={0,2}', content)
            if b64_blobs:
                results.append(IOCResult(
                    severity="CRITICAL",
                    message=f"Large base64 payload in proxy_server.py ({len(b64_blobs)} blob(s))",
                    path=str(proxy_server),
                ))
            # C2 domain references
            for domain in C2_DOMAINS:
                if domain in content:
                    results.append(IOCResult(
                        severity="CRITICAL",
                        message=f"C2 domain '{domain}' found in proxy_server.py",
                        path=str(proxy_server),
                    ))
        except Exception:
            pass

    # 3. Suspicious strings across all .py files
    for root, _, files in os.walk(litellm_path):
        for fname in files:
            if not fname.endswith(".py"):
                continue
            fpath = Path(root) / fname
            try:
                content = fpath.read_text(errors="ignore")
                for sig in SUSPICIOUS_STRINGS:
                    if sig in content.lower():
                        results.append(IOCResult(
                            severity="SUSPICIOUS",
                            message=f"String '{sig}' found in {fpath.name}",
                            path=str(fpath),
                        ))
                        break  # one hit per file is enough
            except Exception:
                pass

    # 4. Dropped persistence/exfil files near site-packages
    for fname in MALICIOUS_DROPPED_FILES:
        for match in site_packages.rglob(fname):
            results.append(IOCResult(
                severity="CRITICAL",
                message=f"Dropped malicious file: {fname}",
                path=str(match),
            ))

    return results


def check_system_iocs() -> list[IOCResult]:
    """Check system-wide IOCs outside of any specific installation."""
    results = []

    # Temp files used by the payload
    for fpath in MALICIOUS_FILES_SYSTEM:
        if Path(fpath).exists():
            results.append(IOCResult(
                severity="CRITICAL",
                message=f"Malicious temp file found: {fpath}",
                path=fpath,
            ))

    # Check for sysmon.py persistence service
    try:
        ps_output = subprocess.run(
            ["ps", "aux"], capture_output=True, text=True, timeout=5
        ).stdout
        if "sysmon.py" in ps_output:
            results.append(IOCResult(
                severity="CRITICAL",
                message="Malicious 'sysmon.py' process running (TeamPCP persistence)",
            ))
    except Exception:
        pass

    # Check for active C2 connections
    try:
        netstat_output = subprocess.run(
            ["lsof", "-i", "-n", "-P"], capture_output=True, text=True, timeout=5
        ).stdout
        for domain in C2_DOMAINS:
            if domain in netstat_output:
                results.append(IOCResult(
                    severity="CRITICAL",
                    message=f"Active network connection to C2 domain: {domain}",
                ))
    except Exception:
        pass

    return results


def find_all_python_site_packages() -> list[Path]:
    """Find all Python site-packages directories on the system."""
    paths = set()

    # From sys.path of all discoverable pythons
    for python_cmd in ["python3", "python"]:
        try:
            result = subprocess.run(
                [python_cmd, "-c", "import site; print('\\n'.join(site.getsitepackages()))"],
                capture_output=True, text=True, timeout=5,
            )
            for line in result.stdout.strip().splitlines():
                p = Path(line.strip())
                if p.is_dir():
                    paths.add(p)
        except Exception:
            pass

    # Common venv/conda locations
    home = Path.home()
    search_roots = [
        home / ".local" / "lib",
        home / ".conda",
        home / "miniconda3",
        home / "anaconda3",
        Path("/usr/lib"),
        Path("/usr/local/lib"),
    ]
    for root in search_roots:
        if root.is_dir():
            for sp in root.rglob("site-packages"):
                if sp.is_dir():
                    paths.add(sp)

    return sorted(paths)


def scan_installed_packages(folder: Path) -> list[Finding]:
    """Find litellm in site-packages directories under a folder."""
    findings = []
    for litellm_dir in folder.rglob("site-packages/litellm"):
        if not litellm_dir.is_dir():
            continue
        version = get_installed_version(litellm_dir)
        is_compromised = version in COMPROMISED_VERSIONS if version else False
        findings.append(Finding(
            path=str(litellm_dir),
            version=version or "unknown",
            source="installed",
            is_compromised=is_compromised,
        ))
    return findings


def scan_system_packages() -> list[Finding]:
    """Find litellm in all system Python environments."""
    findings = []
    seen = set()
    for sp in find_all_python_site_packages():
        litellm_dir = sp / "litellm"
        if litellm_dir.is_dir() and str(litellm_dir) not in seen:
            seen.add(str(litellm_dir))
            version = get_installed_version(litellm_dir)
            is_compromised = version in COMPROMISED_VERSIONS if version else False
            findings.append(Finding(
                path=str(litellm_dir),
                version=version or "unknown",
                source="system",
                is_compromised=is_compromised,
            ))
    return findings


def scan_dependency_files(folder: Path) -> list[Finding]:
    """Scan requirements/pyproject/etc for litellm references."""
    findings = []
    dep_patterns = [
        "requirements*.txt", "Pipfile", "Pipfile.lock",
        "pyproject.toml", "setup.py", "setup.cfg", "poetry.lock",
    ]

    for pattern in dep_patterns:
        for dep_file in folder.rglob(pattern):
            if "site-packages" in str(dep_file):
                continue
            try:
                content = dep_file.read_text(errors="ignore")
            except Exception:
                continue

            for line in content.splitlines():
                if "litellm" not in line.lower():
                    continue
                version_match = re.search(
                    r'litellm\s*[=~><]=?\s*([0-9][0-9a-zA-Z.\-*]+)', line, re.IGNORECASE
                )
                version = version_match.group(1) if version_match else "unpinned"
                is_compromised = version in COMPROMISED_VERSIONS
                findings.append(Finding(
                    path=str(dep_file),
                    version=version,
                    source="dependency_file",
                    is_compromised=is_compromised,
                ))
                break
    return findings


def print_finding(finding: Finding) -> None:
    if finding.is_compromised:
        icon = f"{RED}{BOLD}[COMPROMISED]{RESET}"
    elif finding.version in ("unknown", "unpinned"):
        icon = f"{YELLOW}[WARNING]{RESET}"
    else:
        icon = f"{GREEN}[SAFE]{RESET}"

    labels = {"installed": "Installed", "dependency_file": "Dependency", "system": "System"}
    print(f"  {icon} {labels[finding.source]} v{finding.version}")
    print(f"         {finding.path}")


def print_ioc(ioc: IOCResult) -> None:
    colors = {"CRITICAL": RED, "SUSPICIOUS": YELLOW, "INFO": CYAN}
    color = colors.get(ioc.severity, RESET)
    print(f"         {color}[{ioc.severity}]{RESET} {ioc.message}")
    if ioc.path:
        print(f"                  -> {ioc.path}")


def main():
    parser = argparse.ArgumentParser(
        description="Scan for LiteLLM installations affected by the March 2026 TeamPCP supply chain attack.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
IOCs checked:
  - Compromised versions: 1.82.7, 1.82.8
  - Malicious files: litellm_init.pth, sysmon.py, tpcp.tar.gz
  - System artifacts: /tmp/pglog, /tmp/.pg_state
  - C2 domains: models.litellm[.]cloud, checkmarx[.]zone
  - Suspicious processes: sysmon.py persistence daemon
  - Injected base64 payloads in proxy_server.py

References:
  - https://docs.litellm.ai/blog/security-update-march-2026
  - https://www.armosec.io/blog/litellm-supply-chain-attack-backdoor-analysis/
  - https://www.sonatype.com/blog/compromised-litellm-pypi-package-delivers-multi-stage-credential-stealer
        """,
    )
    parser.add_argument("folders", nargs="*", help="Folders to scan")
    parser.add_argument("--system", action="store_true", help="Scan all Python environments on this machine")
    parser.add_argument("--deep", action="store_true", help="Also scan dependency files (requirements.txt, pyproject.toml, etc.)")
    args = parser.parse_args()

    if not args.folders and not args.system:
        parser.error("Provide at least one folder or use --system")

    print(f"\n{BOLD}LiteLLM Supply Chain Attack Scanner{RESET}")
    print(f"Compromised versions: {', '.join(sorted(COMPROMISED_VERSIONS))}")
    print(f"C2 domains: {', '.join(C2_DOMAINS)}")
    print("=" * 60)

    total_findings = 0
    compromised_count = 0
    ioc_count = 0

    # System-wide scan
    if args.system:
        print(f"\n{BOLD}[System-wide scan]{RESET}")

        # System IOCs (temp files, processes, network)
        sys_iocs = check_system_iocs()
        if sys_iocs:
            print(f"\n  {BOLD}System-level IOCs:{RESET}")
            for ioc in sys_iocs:
                print_ioc(ioc)
                ioc_count += 1
        else:
            print(f"  {GREEN}No system-level IOCs detected.{RESET}")

        # System Python packages
        sys_findings = scan_system_packages()
        if sys_findings:
            print(f"\n  {BOLD}System Python environments:{RESET}")
            for finding in sys_findings:
                print_finding(finding)
                total_findings += 1
                if finding.is_compromised:
                    compromised_count += 1
                iocs = check_iocs(Path(finding.path))
                for ioc in iocs:
                    print_ioc(ioc)
                    ioc_count += 1
        else:
            print(f"  {GREEN}No LiteLLM in system Python environments.{RESET}")

    # Folder scans
    for folder_str in (args.folders or []):
        folder = Path(folder_str).expanduser().resolve()
        if not folder.is_dir():
            print(f"\n{YELLOW}[SKIP]{RESET} Not a directory: {folder}")
            continue

        print(f"\n{BOLD}Scanning: {folder}{RESET}")

        findings = scan_installed_packages(folder)
        if args.deep:
            findings += scan_dependency_files(folder)

        if not findings:
            print(f"  {GREEN}No LiteLLM found.{RESET}")
            continue

        for finding in findings:
            print_finding(finding)
            total_findings += 1
            if finding.is_compromised:
                compromised_count += 1
            if finding.source in ("installed", "system"):
                iocs = check_iocs(Path(finding.path))
                for ioc in iocs:
                    print_ioc(ioc)
                    ioc_count += 1

    # Summary
    print("\n" + "=" * 60)
    print(f"{BOLD}Summary{RESET}")
    print(f"  Installations found : {total_findings}")
    print(f"  IOC hits            : {ioc_count}")
    print(f"  Compromised         : {compromised_count}")

    if compromised_count > 0 or ioc_count > 0:
        print(f"\n{RED}{BOLD}!! ACTION REQUIRED !!{RESET}")
        print(f"{RED}Remediation steps:{RESET}")
        print("  1. pip uninstall litellm (in every affected venv)")
        print("  2. Remove malicious artifacts:")
        print("     rm -f /tmp/pglog /tmp/.pg_state")
        print("     find / -name 'litellm_init.pth' -delete")
        print("     find / -name 'sysmon.py' -name 'tpcp.tar.gz' -delete")
        print("  3. Kill suspicious processes: pkill -f sysmon.py")
        print("  4. Rotate ALL credentials (AWS, GCP, Azure, SSH, K8s, DB)")
        print("  5. Audit cloud account access logs for unauthorized activity")
        print("  6. Reinstall safe version: pip install litellm==1.82.6")
        print("  7. Consider rebuilding affected systems from clean backups")
    elif total_findings > 0:
        print(f"\n{GREEN}No compromised versions or IOCs detected.{RESET}")
    else:
        print(f"\n{GREEN}No LiteLLM installations found. You're clean.{RESET}")

    sys.exit(1 if (compromised_count > 0 or ioc_count > 0) else 0)


if __name__ == "__main__":
    main()