Flask Fraud Detection: One Decorator, Three Failure Modes

Flask makes this integration short enough that the interesting part is not the code. A decorator, a module-level client, and the fraud check runs before your view does. Three things around it decide whether it survives real traffic, and all three are Flask and WSGI concerns rather than fraud concerns.

Those three: a requests.Session that gets created in the wrong place relative to the gunicorn fork, a request with no timeout, and a view that returns 500 because the vendor had a bad minute. This walks through all of it with the official Python SDK, and notes the raw-HTTP equivalents for anyone who would rather not add a dependency.

Install and configure

pip install sentinelsup

# .env — never committed
SENTINEL_KEY=sk_live_your_key_here

The key is a server-side credential. It never reaches the browser, and the collector script does not need one.

The browser half

The server needs a token only the browser can produce. In your base template:

<!-- templates/base.html -->
<script async src="https://maskbreak.com/assets/sentinel.js">
</script>

<!-- templates/register.html -->
<form method="post" action="/register" class="monocle-enriched">
  {{ form.csrf_token }}
  <input type="email" name="email" required>
  <input type="password" name="password" required>
  <button type="submit">Create account</button>
</form>

The script adds a hidden monocle input to any form with that class, so the token arrives in request.form like any other field. For a JSON endpoint, send it in the body under the same name.

The client, and where to create it

One client for the process, created at import time, is right for the threaded and gevent workers people actually run:

# app/fraud.py
import logging
from sentinel import Sentinel, SentinelError

log = logging.getLogger(__name__)

# Short timeout on purpose: this sits in front of a user-facing
# action, so a slow answer is worth less than no answer.
client = Sentinel(timeout=1.5)   # reads SENTINEL_KEY


def evaluate(token, account_id=None):
    """Returns a verdict, or None when we have no opinion.
    None never means 'safe'."""
    if not token:
        return None
    try:
        return client.evaluate(token=token, account_id=account_id)
    except SentinelError as e:
        # Timeout, connection reset, their 5xx. Weather. Count it.
        log.warning("maskbreak unavailable: %s", e)
        return None

Now the fork. If you create the underlying connection pool in the gunicorn parent — which is what happens with --preload, or with an app.py that builds the client inside an if __name__ guard the master also executes — every worker inherits the same open sockets and they interfere with one another. The symptoms are the confusing kind: occasional truncated responses, or SSL errors that never reproduce locally because your dev server never forks.

Module-level creation in a module imported by the app factory is safe under the default gunicorn behaviour, because each worker imports it after forking. If you use --preload, rebuild the client per worker instead:

# gunicorn.conf.py
def post_fork(server, worker):
    import app.fraud
    app.fraud.client = app.fraud.Sentinel(timeout=1.5)

The same rule applies to a raw requests.Session, and to any database pool you keep next to it. It is the most common production-only bug in this integration and it has nothing to do with fraud.

The decorator

# app/fraud.py (continued)
from functools import wraps
from flask import g, request, abort, session

def screen(block_at="block", required=False):
    """block_at='block'  refuse hard blocks only
       block_at='review' refuse anything not clean
       required=True     400 if no token arrived"""
    def outer(view):
        @wraps(view)
        def inner(*args, **kwargs):
            body = request.get_json(silent=True) or {}
            token = request.form.get("monocle") or body.get("monocle")

            if not token:
                # Ad blockers and CSP mistakes stop the collector for
                # real users. Degraded, not hostile — unless this is
                # money.
                if required:
                    abort(400, "missing security token")
                g.verdict = None
                return view(*args, **kwargs)

            uid = session.get("user_id")
            g.verdict = evaluate(token, account_id=uid)

            if g.verdict is not None:
                refused = (g.verdict.decision != "allow"
                           if block_at == "review"
                           else g.verdict.decision == "block")
                if refused:
                    abort(403)

            return view(*args, **kwargs)
        return inner
    return outer

@wraps is not decoration for its own sake here. Without it every screened view reports the same __name__, and Flask raises on the second one you register because it maps endpoints by function name.

Applying it

# app/routes.py
from app.fraud import screen

# Signup: refuse only a hard block — proxy, Tor, automation,
# emulator, tampering. VPN users are customers.
@bp.post("/register")
@screen()
def register():
    ...

# Money leaving: refuse anything that is not clean, and insist on
# a token.
@bp.post("/withdraw")
@screen(block_at="review", required=True)
def withdraw():
    ...

Decorator order matters and gets this wrong quietly. @bp.post must be outermost: it registers whatever function is beneath it, so putting @screen() above it registers the undecorated view and your screening never runs. The route still works, which is what makes it hard to notice. The same applies to @login_required — put it above @screen() so an unauthenticated request is rejected before you spend an API call on it.

Use the review band

The verdict is on g, so views can act on the middle band rather than treating it as a weaker block:

@bp.post("/checkout")
@screen()
def checkout():
    v = g.get("verdict")
    if v and v.decision == "review":
        return step_up()        # OTP, 3DS, manual queue
    return charge()

A plain VPN returns review deliberately. Refusing it costs you a large population of privacy-conscious customers to catch a small number of abusers; the arithmetic rarely works. See risk score thresholds for where each band belongs.

If you are on async views or ASGI

Flask 2 supports async def views, but they run on a worker thread pool rather than on an event loop, so a blocking call inside one is not the disaster it would be in a native ASGI framework. If you are running Flask under an ASGI server through asgiref, or you have moved to Quart, push the blocking call off the loop explicitly:

import anyio

verdict = await anyio.to_thread.run_sync(
    lambda: evaluate(token, account_id))

Blocking the event loop for a network round trip is the standard way an async deployment gets slower than the sync one it replaced.

The three failure modes, decided in advance

No token. The collector did not run: an ad blocker, a content-security-policy mistake, JavaScript disabled, or a client that never loaded your page. This is common on real traffic and it is not evidence of anything. Let it through on signup, count it, and require the token only where money leaves.

Upstream unavailable. Timeout, reset, or a 5xx. Fail open on signup and login. Count it as a metric and alert on the rate rather than on single events, because single failures are constant on the internet and mean nothing.

Rejected request. A 4xx is your bug — a bad key, a malformed body, an expired token — and it will not fix itself. Log it at error, distinctly from the outage case, or it hides inside the noise for a month.

Wire the counter in the same place you swallow the exception:

from prometheus_client import Counter

eval_failures = Counter("fraud_eval_failures_total",
                        "Fraud API calls that produced no verdict",
                        ["kind"])
# ...
except SentinelError as e:
    eval_failures.labels(kind="upstream").inc()
    log.warning("maskbreak unavailable: %s", e)
    return None

A fail-open policy with no counter behind it is indistinguishable from a broken integration, and that is exactly how integrations stay broken for months.

Test the paths you hope never run

def test_blocked_signup_is_refused(client):
    r = client.post("/register", data={
        "monocle": "test_token_block",
        "email": "[email protected]",
    })
    assert r.status_code == 403


def test_signup_survives_an_outage(client, monkeypatch):
    monkeypatch.setattr("app.fraud.evaluate",
                        lambda *a, **k: None)
    r = client.post("/register", data={
        "monocle": "anything",
        "email": "[email protected]",
    })
    assert r.status_code in (200, 302)

The second test is the one that earns its place. Fail-open is a policy, and without a test asserting it, a later refactor turns it into fail-closed and nobody notices until the vendor has an incident.

The short version

• A decorator, a module-level client, and the check runs before the view.
• @bp.post stays outermost. Above @screen() it registers the unscreened function and the check silently never runs.
• Do not create the client before gunicorn forks. With --preload, rebuild it in post_fork.
• Set a timeout of one to two seconds. There is no useful default in front of a user.
• Three failure modes, three decisions: no token is common and benign, an outage fails open and gets counted, a 4xx is your bug and gets logged loudly.
• review is a step-up, not a soft block. A plain VPN is a customer.
• Test the block path and the outage path, or the policy will drift.

Go deeper

• The FastAPI version, with the async traps
• Django: middleware instead of a decorator
• Where to put your thresholds, band by band
• API reference: every field in the verdict

Fifteen lines in front of your signup

Maskbreak returns a decision, a score and the reasons behind it in a single call. Free tier: 1,000 requests per hour, no credit card.

Try Maskbreak free →