FastAPI

Add Nyoxis threat detection to a FastAPI application using an ASGI middleware and httpx for async HTTP calls.

Prerequisites

Python 3.9 or later
A FastAPI application
A Nyoxis workspace API key — get one here

Install

bash

pip install httpx

Middleware

Create middleware/nyoxis.py:

python

from __future__ import annotations

import logging
from typing import Any

import httpx
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

NYO_API = "https://api.nyoxis.com"
logger = logging.getLogger("nyoxis")


class NyoxisWAF(BaseHTTPMiddleware):
    """
    ASGI middleware that calls /v0/predict for every incoming request.

    Args:
        app:           The ASGI application to wrap.
        api_key:       Your Nyoxis workspace API token.
        block_on_high: If True, return 403 when risk == "high".
        timeout:       Request timeout in seconds (default 3).
        on_error:      "open" (default) passes the request through on failure;
                       "closed" returns 503.
    """

    def __init__(
        self,
        app,
        *,
        api_key: str,
        block_on_high: bool = False,
        timeout: float = 3.0,
        on_error: str = "open",
    ) -> None:
        super().__init__(app)
        if not api_key:
            raise ValueError("nyoxis: api_key is required")
        self._api_key = api_key
        self._block_on_high = block_on_high
        self._timeout = timeout
        self._on_error = on_error

    async def dispatch(self, request: Request, call_next) -> Response:
        payload: dict[str, Any] = {
            "method": request.method,
            "path": request.url.path,
            "query": request.url.query or None,
            "ip_addr": request.client.host if request.client else None,
            "headers": dict(request.headers),
        }

        try:
            async with httpx.AsyncClient(timeout=self._timeout) as client:
                response = await client.post(
                    f"{NYO_API}/v0/predict",
                    params={"api_key": self._api_key},
                    json=payload,
                )
                verdict = response.json()

            # Store verdict on request state for downstream access
            request.state.nyoxis = verdict

            if self._block_on_high:
                risk = (verdict.get("prediction") or {}).get("risk")
                if risk == "high":
                    return JSONResponse({"detail": "Forbidden"}, status_code=403)

        except Exception as exc:
            logger.warning("nyoxis: prediction error: %s", exc)
            if self._on_error == "closed":
                return JSONResponse({"detail": "Service unavailable"}, status_code=503)

        return await call_next(request)

Register the middleware

python

from fastapi import FastAPI
from middleware.nyoxis import NyoxisWAF
import os

app = FastAPI()

app.add_middleware(
    NyoxisWAF,
    api_key=os.environ["NYOXIS_API_KEY"],
    block_on_high=True,
)


@app.get("/")
async def root(request: Request):
    verdict = getattr(request.state, "nyoxis", None)
    risk = (verdict or {}).get("prediction", {}).get("risk", "unknown")
    return {"ok": True, "risk": risk}

Acting on the verdict

Access request.state.nyoxis in any route or dependency:

python

from fastapi import Request, HTTPException


@app.get("/sensitive")
async def sensitive_endpoint(request: Request):
    verdict = getattr(request.state, "nyoxis", {})
    prediction = verdict.get("prediction") or {}

    if prediction.get("risk") in ("medium", "high"):
        raise HTTPException(status_code=403, detail="Forbidden")

    attacks = prediction.get("attacks", [])
    sqli = next((a for a in attacks if a["kind"] == "sql_injection"), None)
    if sqli and sqli["confidence"] > 0.8:
        # Log for alerting — could be a false positive, don't hard-block here
        print(f"[security] SQL injection signal on {request.url.path}")

    return {"data": "sensitive content"}

Next steps

API Reference — complete field descriptions and status codes.
Overview — how the classifier and redaction pipeline work.