LiteLLM / TeamPCP Exposure Checker

#!/usr/bin/env python3
"""
PlayCISO LiteLLM / TeamPCP Exposure Checker
=============================================

Checks this host for indicators associated with the March 2026 LiteLLM supply
chain attack (TeamPCP), which backdoored litellm versions 1.82.7 and 1.82.8 on
PyPI. Background: https://playciso.com/blog/litellm-supply-chain-attack-teampcp-what-to-do

This script is:
  - Read-only. It never modifies, deletes, or "cleans" anything.
  - 100% local. It makes zero network calls: no telemetry, no phone-home, no
    live DNS lookups. Read the source below before you run it — that's the
    whole point of a security-checking script being a single readable file.
  - Best-effort. It cannot prove your secrets were or were not exfiltrated.
    A clean result here does not replace rotating credentials if you know you
    ran the bad versions in March 2026 — see the blog post above for that.

Usage:
    python3 litellm_check.py [--dir PATH] [--json]

    --dir PATH   Directory to scan for lockfiles/requirements (default: cwd)
    --json       Emit machine-readable JSON instead of the text report

Exit codes:
    0  clean — no indicators found
    1  one or more indicators found — treat as a lead, not a verdict
    2  script error
"""

import argparse
import json
import os
import re
import shutil
import subprocess
import sys

BAD_VERSIONS = {"1.82.7", "1.82.8"}
C2_DOMAINS = ["models.litellm.cloud", "checkmarx.zone"]
BACKDOOR_SERVICE_NAMES = ["sysmon.service"]
DROPPED_FILE_NAMES = ["litellm_init.pth"]
LOCKFILE_NAMES = [
    "requirements.txt", "requirements.lock", "poetry.lock", "Pipfile.lock",
    "uv.lock", "pyproject.toml", "constraints.txt",
]
SKIP_DIRS = {".git", "node_modules", "__pycache__", ".venv", "venv", "dist", "build"}

findings = []
json_mode = False


def flag(severity, check, detail):
    findings.append({"severity": severity, "check": check, "detail": detail})


def info(msg):
    if not json_mode:
        print(f"  {msg}")


def check_installed_version():
    version = None
    try:
        from importlib import metadata as importlib_metadata
        version = importlib_metadata.version("litellm")
    except Exception:
        version = None

    if version is None:
        for pip_cmd in (["pip3"], ["pip"], [sys.executable, "-m", "pip"]):
            try:
                out = subprocess.run(
                    pip_cmd + ["show", "litellm"],
                    capture_output=True, text=True, timeout=15,
                )
                m = re.search(r"^Version:\s*(\S+)", out.stdout, re.MULTILINE)
                if m:
                    version = m.group(1)
                    break
            except Exception:
                continue

    if version is None:
        info("litellm not found via importlib.metadata or pip show (may not be installed here).")
        return

    if version in BAD_VERSIONS:
        flag("critical", "installed-version", f"litellm {version} is installed — this is a known-backdoored release.")
    else:
        info(f"Installed litellm version: {version} (not in the known-bad set).")


def check_lockfiles(root):
    hits = []
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        for fname in filenames:
            if fname not in LOCKFILE_NAMES:
                continue
            fpath = os.path.join(dirpath, fname)
            try:
                with open(fpath, "r", errors="ignore") as f:
                    content = f.read()
            except Exception:
                continue
            normalized = content.replace('"', "").replace("'", "")
            for bad in BAD_VERSIONS:
                if re.search(r"litellm\s*[=<>!~]*\s*" + re.escape(bad) + r"\b", normalized, re.IGNORECASE):
                    hits.append((fpath, bad))
    for fpath, bad in hits:
        flag("high", "lockfile-pin", f"{fpath} pins litellm=={bad}.")
    if not hits:
        info(f"No lockfile/requirements pin to a known-bad version found under {root}.")


def check_dropped_files():
    search_dirs = set()
    try:
        import site
        if hasattr(site, "getsitepackages"):
            search_dirs.update(site.getsitepackages())
        search_dirs.add(site.getusersitepackages())
    except Exception:
        pass
    search_dirs = [d for d in search_dirs if d and os.path.isdir(d)]

    hits = []
    for d in search_dirs:
        for name in DROPPED_FILE_NAMES:
            fpath = os.path.join(d, name)
            if os.path.exists(fpath):
                hits.append(fpath)
    for fpath in hits:
        flag("critical", "dropped-file", f"{fpath} exists — this is a known IOC file.")
    if not hits and search_dirs:
        info("No known dropped IOC files found in: " + ", ".join(search_dirs))


def check_systemd():
    if not sys.platform.startswith("linux"):
        return
    if shutil.which("systemctl") is None:
        return
    try:
        out = subprocess.run(
            ["systemctl", "list-unit-files", "--all", "--no-legend"],
            capture_output=True, text=True, timeout=15,
        )
        units = out.stdout
    except Exception:
        units = ""
    hit = False
    for name in BACKDOOR_SERVICE_NAMES:
        if name in units:
            flag("critical", "systemd-persistence", f"Unit '{name}' is registered on this host.")
            hit = True
    if not hit and units:
        info("No known backdoor systemd unit found.")


def check_kubernetes():
    if shutil.which("kubectl") is None:
        return
    try:
        out = subprocess.run(
            ["kubectl", "get", "pods", "-n", "kube-system", "-o", "name"],
            capture_output=True, text=True, timeout=15,
        )
    except Exception:
        return
    if out.returncode != 0:
        return
    hits = [line for line in out.stdout.splitlines() if "node-setup-" in line]
    for line in hits:
        flag("critical", "k8s-lateral-movement", f"Suspicious pod in kube-system: {line.strip()}")
    if not hits:
        info("No 'node-setup-*' pods found in kube-system.")


def check_hosts_file():
    hosts_path = "/etc/hosts" if os.name != "nt" else r"C:\Windows\System32\drivers\etc\hosts"
    if not os.path.exists(hosts_path):
        return
    try:
        with open(hosts_path, "r", errors="ignore") as f:
            content = f.read()
    except Exception:
        return
    hit = False
    for domain in C2_DOMAINS:
        if domain in content:
            flag("critical", "hosts-file", f"{hosts_path} has a static entry for {domain}.")
            hit = True
    if not hit:
        info(f"No C2 domain entries in {hosts_path}.")
    info(
        "This script does not resolve " + " or ".join(C2_DOMAINS) + " live — "
        "grep your own DNS/proxy/egress logs for those domains going back to "
        "late March 2026 for a real answer."
    )


def main():
    global json_mode
    parser = argparse.ArgumentParser(description="Check this host for LiteLLM/TeamPCP supply chain IOCs.")
    parser.add_argument("--dir", default=os.getcwd(), help="Directory to scan for lockfiles (default: cwd)")
    parser.add_argument("--json", action="store_true", help="Emit JSON instead of a text report")
    args = parser.parse_args()
    json_mode = args.json

    if not json_mode:
        print("PlayCISO LiteLLM/TeamPCP exposure check")
        print("=" * 44)
        print("Scanning... (100% local, no network calls)")
        print("")

    check_installed_version()
    check_lockfiles(args.dir)
    check_dropped_files()
    check_systemd()
    check_kubernetes()
    check_hosts_file()

    if json_mode:
        print(json.dumps({"findings": findings, "clean": len(findings) == 0}, indent=2))
    else:
        print("")
        if not findings:
            print("RESULT: No known indicators found on this host.")
            print("This does not prove you were never exposed — it means none of the")
            print("indicators we know about are visible from here. If you know you ran")
            print("1.82.7 or 1.82.8 in March 2026, rotate credentials regardless.")
        else:
            print(f"RESULT: {len(findings)} indicator(s) found — treat as a lead, not a verdict.")
            print("")
            for item in findings:
                print(f"  [{item['severity'].upper()}] {item['check']}")
                print(f"      {item['detail']}")
            print("")
            print("Next steps: https://playciso.com/blog/litellm-supply-chain-attack-teampcp-what-to-do")

    sys.exit(1 if findings else 0)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(2)