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 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, in the three steps the Maskbreak console lists: add the script to the page with your form, send the check from your server, and submit your form once. It 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 server needs evidence only the browser can produce. In your templates, load the script and put class="monocle-enriched" on the <form> element itself (never on an input):

<!-- 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 three hidden inputs to that form, monocle (network token), sentinel_fp (browser check id) and sentinel_tz (time zone), so they arrive in request.form like any other field. For a JSON endpoint, call await window.Sentinel.collect() in the browser (check that window.Sentinel exists, and submit anyway when it does not) and send its token, fingerprintEventId and tz in the body. If you send a Content-Security-Policy (Flask-Talisman does), 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

The key is a server-side credential: put your live key in a MASKBREAK_API_KEY environment variable, never in the repository, and never in a template.

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

The client is requests (pip install requests if the app does not have it yet). One client for the process, created at import time, is right for the threaded and gevent workers people actually run. The function maps the form's names to the API's and calls even when a field is missing, because a check without a token or a browser check id is how the console tells you which half is missing. It never raises.

# app/fraud.py
import logging
import os

import requests

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

# Created at import, per worker (see the fork note below).
session = requests.Session()


def evaluate(data, account_id=None):
    """Returns the verdict (a dict), or None when there is none.
    None never means 'safe'. Never raises."""
    email = data.get("email")
    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 if isinstance(email, str) else None,
        "accountId": account_id,
    }
    result = None
    try:
        # Three seconds: this sits in front of a user-facing action.
        resp = session.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",
            },
            timeout=3,
        )
        if resp.ok:
            result = resp.json()
        elif resp.status_code < 500:
            # A 4xx is your bug (a bad key, a malformed body). Log it loudly.
            log.error("maskbreak rejected the request: HTTP %s", resp.status_code)
        else:
            log.warning("maskbreak unavailable: HTTP %s", resp.status_code)   # their 5xx
    except (requests.RequestException, ValueError) as e:
        # Timeout, connection reset. Weather. Count it.
        log.warning("maskbreak unavailable: %s", type(e).__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

Python's root logger drops INFO unless you configure it; logging.basicConfig(level=logging.INFO) in your app factory (or your gunicorn logging config) makes the line visible.

Now the fork. If you create the connection pool in the gunicorn parent, which is what happens with --preload, 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 session per worker instead:

# gunicorn.conf.py
def post_fork(server, worker):
    import requests
    import app.fraud
    app.fraud.session = requests.Session()

The same rule applies 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 abort, g, request, session as flask_session


def clean_allow(v):
    """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=False):
    """Enforce mode: strict=False refuses a block only; strict=True continues
    only on a clean, complete allow. Watch mode never refuses anything."""
    def outer(view):
        @wraps(view)
        def inner(*args, **kwargs):
            data = request.get_json(silent=True)
            if not isinstance(data, dict):
                data = request.form
            g.verdict = evaluate(data, account_id=flask_session.get("user_id"))

            if MODE == "enforce":
                if g.verdict and g.verdict.get("decision") == "block":
                    abort(403)
                if strict and not clean_allow(g.verdict):
                    abort(409)
            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 flask import g

from app.fraud import MODE, screen

# Signup: in enforce mode, refuse only a block. VPN users are customers.
@bp.post("/register")
@screen()
def register():
    ...

# Money leaving: in enforce mode, only a clean, complete allow continues.
@bp.post("/withdraw")
@screen(strict=True)
def withdraw():
    ...

# Checkout: the review band costs a step, not the sale.
@bp.post("/checkout")
@screen()
def checkout():
    v = g.get("verdict")
    if MODE == "enforce" and v and v.get("decision") == "review":
        return step_up()        # OTP, 3DS, manual queue
    return charge()

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. Put @login_required above @screen() too, so an unauthenticated request is rejected before you spend a check on it.

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

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. 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.

The failure modes, decided in advance

No token. The collector did not run: a content blocker, a content-security-policy mistake, JavaScript disabled, or a client that never loaded your page. It is common on real traffic and it is not evidence of anything. The check still goes out (Maskbreak answers it, marked degraded), so the console can show you how often it happens.

Upstream unavailable. Timeout, reset, or a 5xx. evaluate returns None and the view continues. Count it and alert on the rate rather than on single events.

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

from prometheus_client import Counter

eval_failures = Counter("fraud_eval_failures_total",
                        "Fraud API calls that produced no verdict",
                        ["kind"])
# ...in evaluate(), next to the log lines:
#     eval_failures.labels(kind="upstream").inc()

If you are on async views or ASGI

Flask 2 supports async def views, but they run on a worker thread 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 run Flask under an ASGI server or have moved to Quart, push the blocking call off the loop explicitly:

import anyio

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

Test the paths you hope never run

The public sandbox key returns fixed answers for the test tokens, so the block path is an ordinary test. Run it with MASKBREAK_MODE=enforce:

def test_blocked_signup_is_refused(client, monkeypatch):
    monkeypatch.setenv("MASKBREAK_API_KEY", "sk_test_sandbox")
    monkeypatch.setattr("app.fraud.MODE", "enforce")
    r = client.post("/register", data={
        "monocle": "test_proxy",           # the sandbox answers block
        "email": "a@example.com",
    })
    assert r.status_code == 403


def test_signup_survives_an_outage(client, monkeypatch):
    monkeypatch.setattr("app.fraud.MODE", "enforce")
    monkeypatch.setattr("app.fraud.evaluate", lambda *a, **k: None)
    r = client.post("/register", data={"email": "a@example.com"})
    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

• Script on the page, class on the <form>, three hidden fields forwarded as token, fingerprintEventId and tz.
• Call every time, even without a token, and log one line. Watch mode never refuses.
• @bp.post stays outermost. Above @screen() it registers the unscreened function and the check silently never runs.
• Do not create the session before gunicorn forks. With --preload, rebuild it in post_fork.
• Set a timeout on every call. There is no useful default in front of a user.
• In enforce mode, review is a step-up, not a soft block. A plain VPN is a customer.