Ssrf in Fastapi with Basic Auth

Server-Side Request Forgery (SSRF) in a FastAPI service that uses HTTP Basic Authentication can occur when user-supplied input is used to construct outbound HTTP requests without adequate validation or network segregation. In this combination, the API endpoint may accept a URL or host parameter from a client, then forward that request while presenting Basic Auth credentials (username and password) obtained from the request or from a service account. Because FastAPI does not inherently prevent an attacker from directing outbound requests to internal resources, an SSRF vector can allow scanning of internal services such as metadata endpoints (e.g., http://169.254.169.254 on cloud providers), internal management interfaces, or other applications reachable from the server network.

When Basic Auth is involved, the risk is compounded if the credentials are extracted from an untrusted source (such as a header or form field) and reused. An attacker may supply a malicious URL that causes the server to authenticate to an internal service using those credentials, potentially bypassing access controls that would otherwise block unauthenticated requests. Even if the endpoint is designed to call a fixed service, SSRF can occur if the underlying HTTP client follows redirects to internal destinations or if hostname resolution is manipulated (e.g., via DNS rebinding or malicious redirect targets).

During a black-box scan, middleBrick tests SSRF by probing endpoints that accept URLs or host parameters, attempting to reach internal IP ranges and cloud metadata services. If the endpoint uses Basic Auth and forwards requests with those credentials, findings may include unauthenticated reachability of internal systems and exposure of sensitive metadata. The scan also checks whether input validation restricts internal network destinations and whether the runtime environment limits outbound connectivity to prevent data exfiltration.

To mitigate SSRF in FastAPI while using Basic Auth, validate and restrict outbound destinations, avoid propagating extracted credentials, and enforce network-level controls. Below are concrete code examples that demonstrate secure patterns.

Example 1: Whitelisted destinations with strict input validation

Allow only known, safe targets and reject any user-controlled hostname or scheme. Use a pre-defined mapping instead of passing raw user input to the HTTP client.

from fastapi import FastAPI, Depends, HTTPException, Header
import httpx
from pydantic import AnyUrl, url_validator

app = FastAPI()

def validate_safe_url(raw_url: str) -> str:
    # Only allow HTTPS to known domains
    allowed_hosts = {"api.example.com", "data.example.com"}
    parsed = url_validator(raw_url)
    if parsed.scheme != "https":
        raise HTTPException(status_code=400, detail="Only HTTPS URLs are allowed")
    if parsed.hostname not in allowed_hosts:
        raise HTTPException(status_code=400, detail="Destination not permitted")
    return parsed.geturl()

@app.get("/fetch-external")
async def fetch_external(url: str, x_api_key: str = Header(None)):
    target = validate_safe_url(url)
    # Do NOT reuse user-supplied credentials; use a service identity with limited scope
    async with httpx.AsyncClient() as client:
        resp = await client.get(target, headers={"Authorization": f"Bearer {x_api_key}"})
    return {"status": resp.status_code, "body": resp.text[:200]}

Example 2: Avoid propagating Basic Auth extracted from user input

Do not forward credentials taken directly from request headers or parameters. Instead, use a server-side identity with minimal permissions and inject authentication securely.

from fastapi import FastAPI, Depends, HTTPException
import httpx
from pydantic import AnyUrl

app = FastAPI()

# Server-side credentials stored securely (e.g., from environment/secrets)
SERVER_USER = "service_user"
SERVER_PASS = "service_password"

def safe_url(value: str) -> str:
    # Basic validation to block internal IPs and metadata addresses
    if value.startswith("http://"):
        raise HTTPException(status_code=400, detail="Use HTTPS only")
    # Reject private IPs and localhost
    # In production, use a library to properly parse and validate host/IP
    if ".internal" in value or value.startswith("http://169.254."):
        raise HTTPException(status_code=400, detail="Internal destinations not allowed")
    return value

@app.post("/proxy")
async def proxy_request(target: AnyUrl):
    safe_target = safe_url(str(target))
    async with httpx.AsyncClient() as client:
        # Use server-managed auth, not user-supplied credentials
        resp = await client.get(safe_target, auth=(SERVER_USER, SERVER_PASS))
    return {"status": resp.status_code}

General recommendations

• Do not accept raw URLs from clients; use a curated set of endpoints or a validated allowlist.
• Restrict outbound network access at the host level (firewall rules, container network policies) to prevent connections to internal services and metadata endpoints.
• Avoid reusing Basic Auth credentials extracted from user input; prefer short-lived tokens or service identities with least privilege.
• Disable redirect following or validate each redirect location to prevent SSRF via open redirects.