Webhook Abuse in Fastapi with Jwt Tokens
Webhook Abuse in Fastapi with Jwt Tokens — how this specific combination creates or exposes the vulnerability
Webhook abuse in Fastapi applications that rely on JWT tokens for authentication can occur when webhook endpoints do not adequately validate the origin and integrity of incoming requests. A webhook URL exposed in client-side code or predictable endpoint paths can be targeted by attackers who send crafted HTTP requests. Even when requests include JWT tokens, misuse is possible if token validation is incomplete or if tokens are accepted from untrusted sources.
In Fastapi, if a route intended for external webhooks also accepts bearer tokens via the Authorization header and does not verify issuer, audience, or token binding, an attacker can replay captured tokens or inject tokens obtained through other means. For example, consider a route that processes events from a third-party service:
from fastapi import FastAPI, Depends, Header, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
app = FastAPI()
security = HTTPBearer()
def verify_token(credentials: HTTPAuthorizationCredentials = Depends(security)):
# Incomplete validation: only checks presence, not signature or claims
if not credentials.credentials:
raise HTTPException(status_code=403, detail="Missing token")
return credentials.credentials
@app.post("/webhook")
async def webhook_handler(token: str = Depends(verify_token)):
# Process webhook event
return {"status": "received"}
In this pattern, an attacker who obtains a valid JWT token—perhaps through logs, client-side exposure, or a separate vulnerability—can send it to the webhook endpoint. If the endpoint does not validate scopes, issuer, or intended audience, the request is accepted as legitimate. Additionally, if the webhook URL is exposed in public repositories or client-side JavaScript, automated tools can discover it and flood it with requests using valid tokens, leading to denial of service or unauthorized event processing.
Another vector arises when JWT tokens are used without HTTPS termination at the webhook receiver. Tokens intercepted in transit can be reused to abuse otherwise protected endpoints. Even with HTTPS, if token revocation or expiration checks are not enforced, compromised tokens remain usable until natural expiry.
The interaction between webhook triggers and JWT tokens also surfaces risks around event replay. An attacker can capture a signed request and replay it to the same endpoint, especially when idempotency is not enforced. Without nonce or timestamp validation within the token claims or webhook payload, the system cannot distinguish legitimate replays from malicious ones.
Jwt Tokens-Specific Remediation in Fastapi — concrete code fixes
To secure webhook endpoints in Fastapi that use JWT tokens, implement strict token validation and explicit trust boundaries. Always verify the token signature, issuer (iss), audience (aud), and expiration (exp) using a trusted library such as PyJWT or python-jose. Avoid accepting tokens from arbitrary sources, and ensure webhook routes either rely on a shared secret or a public key that is explicitly managed.
Below is a concrete example using PyJWT with RSA public key validation:
from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
import jwt
from jwt import PyJWKClient
from datetime import datetime, timezone
app = FastAPI()
security = HTTPBearer()
JWKS_URL = "https://example.com/.well-known/jwks.json"
def get_jwks_client():
return PyJWKClient(JWKS_URL)
def decode_token(token: str):
jwks_client = get_jwks_client()
signing_key = jwks_client.get_signing_key_from_jwt(token)
decoded = jwt.decode(
token,
signing_key.key,
algorithms=["RS256"],
options={"require": ["iss", "aud", "exp", "nbf"]},
issuer="https://auth.example.com/",
audience="https://api.example.com/webhook",
)
return decoded
@app.post("/webhook")
async def webhook_handler(credentials: HTTPAuthorizationCredentials = Depends(security)):
if not credentials.credentials.startswith("Bearer "):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authorization scheme")
token = credentials.credentials.split(" ")[1]
try:
payload = decode_token(token)
# Optional: verify custom claims such as event source or scope
if payload.get("event_scope") != "webhook":
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Insufficient scope")
except jwt.ExpiredSignatureError:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired")
except jwt.InvalidTokenError:
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Invalid token")
# Process webhook event safely
return {"status": "accepted"}
Additional remediation practices include enforcing HTTPS for all webhook communications, implementing idempotency keys for event processing, and validating the origin of requests using a shared secret or mutual TLS where feasible. Avoid placing webhook URLs in publicly accessible locations and rotate signing keys regularly. If you use the middleBrick CLI to scan this Fastapi service, it can surface insecure token handling patterns and map findings to relevant compliance controls.