FastAPI Fraud Detection: One Dependency, Failing Open

The call itself is one line. Everything that makes a fraud check safe to deploy is around it: which routes it runs on, what it does when the vendor is having a bad day, and, specific to FastAPI, which side of the event loop it waits on.

This is written as a dependency rather than middleware, because dependencies are per-route, typed, and overridable in tests. Middleware sees every request including your static files and your health check, and then you spend the rest of the afternoon writing path prefixes to keep it off the fast path. It follows the three steps the Maskbreak console lists (add the script to the page with your form, send the check from your server, submit your form once) and starts in watch mode, where every check is logged and nothing is refused.

Step 1: Add the script to the page with your form

The check needs evidence from the visitor's browser. Load the script on the page with your form. For a server-rendered form, put class="monocle-enriched" on the <form> itself and the script adds three hidden fields: monocle (network token), sentinel_fp (browser check id) and sentinel_tz (time zone).

<script async src="https://maskbreak.com/assets/sentinel.js"></script>

<form method="post" action="/signup" class="monocle-enriched">
  <input type="email" name="email" required>
  <button type="submit">Create account</button>
</form>

FastAPI usually sits behind a JavaScript frontend that posts JSON. Then ask the script for the values at submit time and put them in the body. Check that window.Sentinel exists, and never hold the form back when it does not:

const evidence = window.Sentinel
  ? await window.Sentinel.collect().catch(() => ({}))
  : {};                     // script blocked: submit anyway
await fetch('/signup', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ email, ...evidence }),   // token, fingerprintEventId, tz
});

If your pages send a Content-Security-Policy, add https://maskbreak.com, https://mcl.maskbreak.com, https://*.mcl.maskbreak.com, https://mesu.maskbreak.com and 'wasm-unsafe-eval' to script-src, the mcl hosts (https and wss) and mesu to connect-src, and https://mesu.maskbreak.com blob: to worker-src.

Step 2: Send the check from your server

Keep your live key in a MASKBREAK_API_KEY environment variable; nothing goes in your settings module, and the key never reaches the browser.

# .env (never committed)
MASKBREAK_API_KEY=sk_live_...
# MASKBREAK_MODE=enforce   (later; unset means watch)

The client is an httpx.AsyncClient (pip install httpx; fastapi[standard] already includes it), so waiting on the network never blocks the event loop. The function maps the form's names to the API's and calls even when a field is missing: a check without a token or a browser check id is how the console tells you which half is missing. It never raises.

# app/maskbreak.py
import logging
import os

import httpx

log = logging.getLogger("maskbreak")
MODE = os.environ.get("MASKBREAK_MODE", "watch")  # "watch" or "enforce"

# One client for the process, which runs one event loop: its pooled
# connections belong to that loop. Short timeout: this sits in front of a
# user-facing action, so a slow answer is worth less than no answer.
client = httpx.AsyncClient(timeout=3.0)


async def check_visitor(data: dict, email: str | None = None,
                        account_id: str | None = None) -> dict | None:
    """Returns the verdict, or None when there is none. Never raises."""
    payload = {
        "token": data.get("monocle") or data.get("token") or "",
        "fingerprintEventId": data.get("sentinel_fp") or data.get("fingerprintEventId"),
        "tz": data.get("sentinel_tz") or data.get("tz"),
        "email": email,
        "accountId": account_id,
    }
    result = None
    try:
        resp = await client.post(
            "https://maskbreak.com/v1/evaluate",
            json={k: v for k, v in payload.items() if v is not None},
            headers={
                "Authorization": "Bearer " + os.environ.get("MASKBREAK_API_KEY", ""),
                "Content-Type": "application/json",
            },
        )
        if resp.is_success:
            result = resp.json()
        else:
            log.warning("maskbreak HTTP %s", resp.status_code)  # 401: the key
    except Exception as exc:   # timeout, network, bad JSON, anything: no verdict, never a 500
        log.warning("maskbreak unavailable: %s", type(exc).__name__)
    # One line per check. Never log the key, the token or form data.
    log.info("maskbreak mode=%s decision=%s reasons=%s device_evidence=%s degraded=%s",
             MODE,
             result.get("decision") if result else None,
             result.get("reasons") if result else None,
             result.get("device_evidence") if result else None,
             bool(result.get("degraded")) if result else None)
    return result

The dependency

One factory, parameterised by how strict the route should be. It reads the evidence from a JSON body or a form (form parsing needs python-multipart), records the verdict, and in watch mode never refuses.

# app/deps.py
from fastapi import HTTPException, Request

from app.maskbreak import MODE, check_visitor


async def _evidence(request: Request) -> dict:
    try:
        if request.headers.get("content-type", "").startswith("application/json"):
            data = await request.json()
        else:
            data = dict(await request.form())
    except Exception:
        return {}
    return data if isinstance(data, dict) else {}


def clean_allow(v: dict | None) -> bool:
    """A complete, production allow."""
    return bool(v) and v.get("decision") == "allow" and v.get("device_evidence") == "ok" \
        and not v.get("degraded") and not v.get("test")


def screen(strict: bool = False):
    """Enforce mode: strict=False refuses only a block; strict=True continues
    only on a clean, complete allow. Watch mode never refuses anything."""

    async def dependency(request: Request) -> dict | None:
        data = await _evidence(request)
        email = data.get("email")
        result = await check_visitor(
            data,
            email=email if isinstance(email, str) else None,
            account_id=getattr(request.state, "user_id", None),
        )
        if MODE == "enforce":
            if result and result.get("decision") == "block":
                raise HTTPException(status_code=403, detail="Request declined")
            if strict and not clean_allow(result):
                raise HTTPException(status_code=409, detail="Additional verification required")
        return result

    return dependency


screen_default = screen()
screen_strict = screen(strict=True)

FastAPI has already read the body for your route by the time the dependency runs, and Starlette keeps it on the request, so reading it again here costs nothing and leaves your body model intact.

The FastAPI-specific part

The dependency is async def and the client is async. That pairing is deliberate, and it is the one detail in this post that is genuinely about FastAPI rather than about fraud.

FastAPI runs async def dependencies on the event loop. Put a blocking client inside one (requests, urllib, a synchronous SDK) and you have moved a blocking network read onto the loop: one slow answer now stalls every concurrent request in the process, and the symptom under load is a latency cliff that looks nothing like a fraud problem. If you would rather use a blocking client, declare the dependency with plain def, and FastAPI runs it in a threadpool instead. Either shape is fine. Blocking the loop is not.

Different actions, different strictness

Attach the dependency where the decision matters, not globally. A single threshold across every route is always wrong somewhere: too tight on a newsletter form, too loose on a withdrawal.

# app/routes.py
from fastapi import APIRouter, Depends

from app.deps import screen_default, screen_strict
from app.maskbreak import MODE

router = APIRouter()

# Signup: in enforce mode, refuse only a block. VPN users are customers.
@router.post("/signup")
async def signup(payload: SignupIn, verdict=Depends(screen_default)):
    ...

# Money leaving: in enforce mode, only a clean, complete allow continues.
@router.post("/withdraw", dependencies=[Depends(screen_strict)])
async def withdraw(payload: WithdrawIn):
    ...

# Checkout: review costs a step, not the sale.
@router.post("/checkout")
async def checkout(payload: CheckoutIn, verdict=Depends(screen_default)):
    if MODE == "enforce" and verdict and verdict.get("decision") == "review":
        return await step_up(payload)        # OTP, 3DS, manual queue
    return await charge(payload)

A JSON body model that forbids extra fields would reject token, fingerprintEventId and tz: either add them to the model as optional strings or keep extra fields allowed.

The rule that matters here: review must not route to the same place as block. If it does, you have built a hard block with extra steps, and every VPN user in your funnel is a refused customer you will never hear about. There is more on that split in route on the decision, sort on the score.

Step 3: Submit your form once

Run the app on your computer or deploy it, open the page in an ordinary browser, wait a moment, and submit the form. The first check then appears in the Maskbreak console (if half of it did not arrive, the console says which half), and your log shows one maskbreak line (configure the logger at INFO; uvicorn's defaults do not show it). degraded=True means no network token arrived; device_evidence=not_sent means no browser check id arrived; device_evidence=ok with degraded=False is a complete check. When it looks right after a few days, set MASKBREAK_MODE=enforce and redeploy.

Pass an account id and get multi-accounting for free

The account_id argument is doing more than logging. When it is present, the same device arriving under a second account is linkable, which is the signal that catches trial abuse, referral farming and bonus abuse, none of which look wrong on any single request.

Two constraints worth knowing before you wire it up. Linking is per-customer and hash-only: your accounts are never linked against another customer's, and the raw device identifier is not what gets stored. And it only works if you pass a stable id from your own server: a session id changes on every login and links nothing.

Testing it in CI

The public sandbox key returns fixed decisions for a set of test tokens, with no account and no live pipeline behind it: test_clean answers allow, test_vpn review, test_proxy and test_tor block. Every sandbox answer carries test: true, which is why a strict route never lets one through.

# tests/conftest.py
import os
os.environ.setdefault("MASKBREAK_API_KEY", "sk_test_sandbox")
os.environ.setdefault("MASKBREAK_MODE", "enforce")

import pytest
from fastapi.testclient import TestClient

from app.main import app


@pytest.fixture(scope="session")
def client():
    # One TestClient held open runs every request on one event loop, as in
    # production. Without the with-block each request gets a new loop, and the
    # module's AsyncClient would reuse a connection from a closed one.
    with TestClient(app) as c:
        yield c

# tests/test_withdraw.py
def test_withdraw_refuses_a_proxy(client):
    r = client.post("/withdraw", json={"amount": 100, "token": "test_proxy"})
    assert r.status_code == 403


def test_withdraw_holds_a_test_result(client):
    r = client.post("/withdraw", json={"amount": 100, "token": "test_clean"})
    assert r.status_code == 409

For the failure path, override the dependency instead of mocking the network. That is the reason to use a dependency in the first place:

app.dependency_overrides[screen_default] = lambda: None   # vendor down: signup proceeds

Notice what that test asserts: when the fraud check is unavailable, the signup still goes through. If that makes you uncomfortable on a payout, the strict route already holds it; the fix elsewhere is a lower rate limit and a review queue, not a check that fails closed everywhere.

The short version

• Script on the page; collect() for a JSON frontend, the class on the <form> for a server-rendered one.
• Forward token, fingerprintEventId and tz, and call every time, even when one is missing.
• Use a dependency, not middleware: per-route, typed, overridable in tests.
• An async client in an async def dependency, or a blocking client in a plain def one. Never a blocking call on the loop.
• Watch first; in enforce mode, strict on money, block-only on signup, nothing on static routes.
• Pass a stable account_id if multi-accounting is part of your problem.