How it arises in FastAPI
HttpUrl is a syntax type. Pydantic documents HttpUrl as a type that accepts any http or https URL, with a maximum length of 2083, where a top-level domain is not required and a host is not required. http://localhost:8000/ and http://10.0.0.5/ are valid values. The type removes file:// and gopher://, which is useful, and leaves the destination question open.
Webhook registration. POST /webhooks with {"callback_url": "..."} is close to a standard feature of API products. The service later sends events to that URL from inside its own network. Often a "send test event" endpoint fires immediately and returns the status code, which gives the caller a port scanner.
Service meshes. FastAPI services call each other over plain HTTP with no authentication because the network is trusted. That is precisely what makes an SSRF valuable: the forged request arrives from a trusted peer.
Async clients everywhere. httpx.AsyncClient is the usual choice. HTTPX does not follow redirects by default, which is a safer default than some other clients. Many codebases create one shared client with follow_redirects=True so that normal integrations work, and the webhook sender inherits it.
URL-typed query parameters for import and preview endpoints: source: HttpUrl.
Vulnerable and fixed
import httpx
from fastapi import FastAPI
from pydantic import BaseModel, HttpUrl
app = FastAPI()
client = httpx.AsyncClient(follow_redirects=True, timeout=5)
class Hook(BaseModel):
callback_url: HttpUrl
@app.post("/webhooks/test")
async def test_hook(hook: Hook):
resp = await client.post(str(hook.callback_url), json={"event": "ping"})
return {"status": resp.status_code}
Put the destination rule in a dependency, so every endpoint that accepts a callback shares it. It resolves the host without blocking the event loop, rejects non-public addresses, and the sender uses a client that does not follow redirects:
import asyncio
import ipaddress
import httpx
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel, HttpUrl
app = FastAPI()
hook_client = httpx.AsyncClient(follow_redirects=False, timeout=5)
class Hook(BaseModel):
callback_url: HttpUrl
async def public_hook(hook: Hook) -> Hook:
url = hook.callback_url
loop = asyncio.get_running_loop()
infos = await loop.getaddrinfo(url.host, url.port or 443)
if not infos or any(not ipaddress.ip_address(i[4][0]).is_global for i in infos):
raise HTTPException(status_code=422, detail="callback host is not allowed")
return hook
@app.post("/webhooks/test")
async def test_hook(hook: Hook = Depends(public_hook)):
resp = await hook_client.post(str(hook.callback_url), json={"event": "ping"})
return {"delivered": resp.status_code < 400}
The response no longer echoes the upstream status. Between the check and the send the name is resolved a second time, so pair this with an egress rule or a forward proxy that refuses internal ranges.
What FastAPI, Pydantic and HTTPX already do for you
HttpUrl parses the URL once and gives you host, port and scheme as attributes, so you do not write a URL parser. Dependencies give the destination rule a single home and make it visible in every signature that uses it. HTTPX keeps redirects off unless asked, and applies timeouts by default. Pydantic's IPvAnyAddress type covers the case where callers should supply an address directly.
What es-python adds
For a webhook product the practical question is "where did our servers connect because a customer told them to". es-python answers from the running service. The base events socket getaddrinfo and socket connect fire whenever a destination was steered by untrusted data, and when the address being dialled is loopback, link-local or private, an ssrf event with the ssrf impact tag fires beside them.
The decision is taken on the address, after HttpUrl validation, after name resolution in Python, and after any redirect the client followed. So the shared follow_redirects=True client in the vulnerable version is covered: the redirect produces a second connection, to the internal address, and that is what gets reported.
Registered callbacks are stored and used later by a delivery worker. Switch on the db_rows source together with socket so the URL read back from the database is followed into the send.
Stated limits: alternate encodings of an internal address are not recognised by the address test, a name is identified as internal only when the program resolves it through Python, and DNS rebinding is not covered. The base socket connect event remains as the record of every customer-chosen destination. Under the Block backend network attacks template, or any Raise rule on ssrf, the connect raises a RuntimeError before it is made, which the endpoint returns as an error like any other.
Review checklist
- Find every model field typed
HttpUrl,AnyUrlorstrthat is later fetched. - Move destination checks into a shared dependency.
- Use a dedicated client with
follow_redirects=Falsefor user-directed calls. - Do not return upstream status codes or bodies from test endpoints.
- Authenticate service-to-service calls so a forged request is not automatically trusted.
References
- Reference: Pydantic documentation: Network Types, HttpUrl (checked 2026-10-12)
- Reference: HTTPX documentation: QuickStart, Redirection and History (checked 2026-10-12)
- Reference: FastAPI documentation: Dependencies (checked 2026-10-12)