SGLangs multimodal generation runtime is vulnerable to unauthenticated remote code execution when the…

#!/usr/bin/env python3
# Verify likely exposure to CVE-2026-7304 on a Linux host
# Exit codes: 0=PATCHED, 1=VULNERABLE, 2=UNKNOWN

import os
import re
import sys
import glob
from pathlib import Path

try:
    from importlib import metadata as importlib_metadata
except Exception:
    import importlib_metadata  # type: ignore

LOWER_AFFECTED = "0.4.1.post7"


def parse_version(v):
    if not v:
        return None
    m = re.match(r"^(\d+)\.(\d+)\.(\d+)(?:\.post(\d+))?", v)
    if not m:
        return None
    major, minor, patch, post = m.groups()
    return (int(major), int(minor), int(patch), int(post or 0))


def version_gte(a, b):
    pa = parse_version(a)
    pb = parse_version(b)
    if pa is None or pb is None:
        return None
    return pa >= pb


def get_installed_version():
    candidates = ["sglang", "SGLang"]
    for name in candidates:
        try:
            return importlib_metadata.version(name)
        except Exception:
            pass
    return None


def get_package_root():
    try:
        dist = importlib_metadata.distribution("sglang")
        for f in dist.files or []:
            p = Path(dist.locate_file(f))
            if p.name == "__init__.py" and "sglang" in str(p.parent):
                return p.parent
    except Exception:
        pass
    for base in sys.path:
        p = Path(base) / "sglang"
        if p.exists() and p.is_dir():
            return p
    return None


def source_has_sink(pkg_root):
    if not pkg_root:
        return False, None
    target = pkg_root / "srt" / "sampling" / "custom_logit_processor.py"
    if not target.exists():
        return False, str(target)
    try:
        txt = target.read_text(encoding="utf-8", errors="ignore")
    except Exception:
        return False, str(target)
    patterns = [
        'dill.loads(bytes.fromhex(data["callable"]))',
        "dill.loads(bytes.fromhex(data['callable']))",
        "dill.loads(",
    ]
    return any(p in txt for p in patterns), str(target)


def find_running_flagged_processes():
    hits = []
    for proc in glob.glob("/proc/[0-9]*/cmdline"):
        pid = proc.split("/")[2]
        try:
            raw = Path(proc).read_bytes()
            if not raw:
                continue
            cmd = raw.replace(b"\x00", b" ").decode("utf-8", errors="ignore").strip()
        except Exception:
            continue
        if "sglang" in cmd and "--enable-custom-logit-processor" in cmd:
            hits.append({"pid": pid, "cmd": cmd})
    return hits


def find_unitfile_flag():
    roots = ["/etc/systemd/system", "/lib/systemd/system", "/usr/lib/systemd/system"]
    hits = []
    for root in roots:
        p = Path(root)
        if not p.exists():
            continue
        for unit in p.rglob("*.service"):
            try:
                txt = unit.read_text(encoding="utf-8", errors="ignore")
            except Exception:
                continue
            if "sglang" in txt and "--enable-custom-logit-processor" in txt:
                hits.append(str(unit))
    return hits


def main():
    version = get_installed_version()
    pkg_root = get_package_root()
    sink_present, sink_file = source_has_sink(pkg_root)
    flagged_procs = find_running_flagged_processes()
    flagged_units = find_unitfile_flag()

    print(f"installed_version={version or 'UNKNOWN'}")
    print(f"package_root={str(pkg_root) if pkg_root else 'UNKNOWN'}")
    print(f"source_file={sink_file or 'UNKNOWN'}")
    print(f"deserialization_sink_present={'yes' if sink_present else 'no'}")
    print(f"running_processes_with_flag={len(flagged_procs)}")
    for hit in flagged_procs[:10]:
        print(f"process_hit pid={hit['pid']} cmd={hit['cmd']}")
    print(f"unit_files_with_flag={len(flagged_units)}")
    for unit in flagged_units[:10]:
        print(f"unit_hit path={unit}")

    affected_by_version = None
    if version:
        affected_by_version = version_gte(version, LOWER_AFFECTED)
        print(f"affected_by_version={'yes' if affected_by_version else 'no' if affected_by_version is False else 'unknown'}")
    else:
        print("affected_by_version=unknown")

    flag_enabled = bool(flagged_procs or flagged_units)

    if affected_by_version is False:
        print("PATCHED")
        sys.exit(0)

    if (affected_by_version is True or sink_present) and flag_enabled:
        print("VULNERABLE")
        sys.exit(1)

    if affected_by_version is True or sink_present:
        print("UNKNOWN")
        sys.exit(2)

    print("UNKNOWN")
    sys.exit(2)


if __name__ == "__main__":
    main()