Flask

Add Nyoxis threat detection to a Flask application using a before_request hook and the requests library.

Prerequisites

Python 3.9 or later
Flask 2.0 or later
A Nyoxis workspace API key — get one here

Install

bash

pip install requests

Integration

Create nyoxis_waf.py:

python

from __future__ import annotations

import logging
import os
from typing import Any

import requests
from flask import Flask, g, jsonify, request

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


def init_nyoxis(app: Flask, *, block_on_high: bool = False, on_error: str = "open") -> None:
    """
    Register Nyoxis WAF hooks on a Flask app.

    Args:
        app:           The Flask application instance.
        block_on_high: If True, abort with 403 when risk == "high".
        on_error:      "open" (default) passes through on failure;
                       "closed" aborts with 503.
    """
    api_key = os.environ.get("NYOXIS_API_KEY")
    if not api_key:
        raise RuntimeError("NYOXIS_API_KEY environment variable is required")

    @app.before_request
    def check_request():
        payload: dict[str, Any] = {
            "method": request.method,
            "path": request.path,
            "query": request.query_string.decode() or None,
            "ip_addr": request.remote_addr,
            "headers": dict(request.headers),
            "body": request.get_data(as_text=True) or None,
        }

        try:
            resp = requests.post(
                f"{NYO_API}/v0/predict",
                params={"api_key": api_key},
                json=payload,
                timeout=3,
            )
            verdict = resp.json()
            # Store on Flask's request-scoped g object
            g.nyoxis = verdict

            if block_on_high:
                risk = (verdict.get("prediction") or {}).get("risk")
                if risk == "high":
                    return jsonify({"error": "Forbidden"}), 403

        except Exception as exc:
            logger.warning("nyoxis: prediction error: %s", exc)
            if on_error == "closed":
                return jsonify({"error": "Service unavailable"}), 503

Register on your app

python

# app.py
from flask import Flask, g
from nyoxis_waf import init_nyoxis

app = Flask(__name__)
init_nyoxis(app, block_on_high=True)


@app.route("/")
def index():
    verdict = getattr(g, "nyoxis", {})
    risk = (verdict.get("prediction") or {}).get("risk", "unknown")
    return {"ok": True, "risk": risk}

Acting on the verdict

The verdict is available as g.nyoxis in every route and after_request hook:

python

from flask import Flask, g, abort


@app.route("/admin")
def admin():
    verdict = getattr(g, "nyoxis", {})
    prediction = verdict.get("prediction") or {}

    if prediction.get("risk") in ("medium", "high"):
        abort(403)

    attacks = prediction.get("attacks", [])
    for attack in attacks:
        if attack["kind"] == "path_traversal":
            app.logger.warning(
                "Path traversal signal on %s from %s",
                request.path,
                request.remote_addr,
            )

    return {"data": "admin content"}

Next steps

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