Python
Check signups in Python
Check each signup in your Python view before you create the user. The file needs the requests package.
Install
A function for Django, Flask, FastAPI or any Python backend. The file needs the requests package.
- Save the file as signupbouncer.py in your project.
- Create an API key for the app on its Integration tab in your dashboard, then set SIGNUPBOUNCER_API_KEY in your server's environment to that key (sb_live_...).
- In your signup view, call signup_refusal(email, ip) before you create the user. When it returns a message, refuse the signup and show that message.
- To show the "Why was this refused?" link too, call signup_refusal_with_link(email, ip) instead. It returns None when the signup may go ahead, which includes a failed check. When you should refuse the signup, it returns a dict with "message" and "why_url". "why_url" is None unless you turn the link on in the Rules tab. For example: refusal = signup_refusal_with_link(email, ip), then if refusal: show refusal["message"], and refusal["why_url"] as a link when it is set.
signupbouncer.py
# SignupBouncer signup check. Docs: https://signupbouncer.com/docs/code/python
# signupbouncer.py: checks one signup with SignupBouncer. Needs the requests package and SIGNUPBOUNCER_API_KEY in
# the environment. Leave SIGNUPBOUNCER_BASE_URL unset. It points the helper at another copy of the API, such as one on
# your machine.
#
# The call has a 1.5-second limit in total, and any failure lets the signup through: a timeout, a network error, a
# non-2xx answer or an unreadable body. Only a complete, consistent enforce-mode block answer refuses the signup: then
# it returns the message to show, the one you set on your dashboard's Rules tab. With our default message, we word it
# to fit the reason, such as "Did you mean ann@gmail.com?" for a typo domain. Otherwise it returns None.
# signup_refusal_with_link returns None in the same cases, or {"message": ..., "why_url": ...}: the same message and,
# when you turn on the "Why was this refused?" link on the Rules tab, the link to show after it (None otherwise).
import ipaddress
import json
import math
import os
import re
from concurrent.futures import ThreadPoolExecutor
from threading import BoundedSemaphore
from time import monotonic
from urllib.parse import urlsplit
import requests
MAX_ANSWER_BYTES = 65536 # a real answer is far smaller
DEFAULT_MESSAGE = "Please sign up with a permanent email address."
# requests' own timeout limits each network read, not the whole call, so the call runs in a worker thread and your
# signup view stops waiting after 1.5 seconds whatever the API does. At most eight calls, with no waiting backlog.
_pool = ThreadPoolExecutor(max_workers=8)
_slots = BoundedSemaphore(8)
def _is_block(v):
# A partial or contradictory answer must never refuse a real signup.
number = lambda x: type(x) in (int, float) and math.isfinite(x)
string = lambda x: isinstance(x, str)
nullable = lambda x: x is None or string(x)
if not isinstance(v, dict) or any(v.get(k) != val for k, val in
(("action", "block"), ("verdict", "block"), ("mode", "enforce"))):
return False
if v.get("quota_exceeded") is not False or "error" in v:
return False
if not number(v.get("score")) or not string(v.get("check_id")) or not number(v.get("latency_ms")):
return False
for field, points in (("reasons", True), ("shadow", False)):
rows = v.get(field)
if not isinstance(rows, list) or not all(isinstance(r, dict) and
all(string(r.get(k)) for k in ("code", "detail", "tier")) and
(not points or number(r.get("points"))) for r in rows):
return False
n = v.get("normalized")
if not isinstance(n, dict) or "provider" not in n or not nullable(n["provider"]) or not number(n.get("repeat_signups_7d")):
return False
if "network" not in v:
return False
net = v["network"]
return net is None or (isinstance(net, dict) and
all(k in net for k in ("asn", "asn_org", "kinds", "label", "prefix")) and
(net["asn"] is None or number(net["asn"])) and nullable(net["asn_org"]) and
isinstance(net["kinds"], list) and all(string(k) for k in net["kinds"]) and
nullable(net["label"]) and string(net["prefix"]))
def _message(v):
# The message sent with a block answer (your Rules tab's, or our default for the reason); ours if it is missing.
m = v.get("block_message")
return m if isinstance(m, str) and m.strip() and len(m) <= 500 else DEFAULT_MESSAGE
def _why_url(v):
# The "Why was this refused?" link, sent only when you turn it on in your dashboard's Rules tab; None otherwise.
# Only a well-formed http(s) URL with a host counts, returned unchanged, so a bad value never becomes a link.
u = v.get("why_url")
if not isinstance(u, str) or len(u) > 2048 or "\\" in u or any(c.isspace() for c in u):
return None
try:
parts = urlsplit(u)
parts.port # raises ValueError for a port that is not a number in range
host = parts.hostname
if parts.scheme.lower() not in ("http", "https") or not u.lower().startswith(parts.scheme.lower() + "://"):
return None
if not host:
return None
if "[" in parts.netloc:
ipaddress.IPv6Address(host.split("%", 1)[0]) # raises ValueError for anything else in brackets
elif not re.fullmatch(r"[\w.~!$&'()*+,;=-]+", host):
return None
except ValueError:
return None
return u
def signup_refusal(email=None, ip=None, auth_provider=None, email_verified=None):
# Returns the message to show when the signup should be refused, or None to let it through.
refusal = signup_refusal_with_link(email, ip, auth_provider, email_verified)
return refusal["message"] if refusal else None
def signup_refusal_with_link(email=None, ip=None, auth_provider=None, email_verified=None):
# Like signup_refusal, but returns {"message": ..., "why_url": ...} or None. why_url is the "Why was this
# refused?" link to show after the message, or None when you have not turned the link on in the Rules tab.
# No real address is longer than 320 characters. Refusing them here keeps the request small enough to be checked.
if email and len(email) > 320:
return {"message": DEFAULT_MESSAGE, "why_url": None}
# auth_provider: the provider id, for OAuth/SSO signups only. email_verified: your auth platform's own flag that
# the provider verified this address (an OIDC email_verified claim), never what the user typed. Only True counts.
body = {k: v for k, v in {"email": email, "ip": ip, "auth_provider": auth_provider}.items() if v}
if email_verified is True:
body["email_verified"] = True
# No unbounded queue of checks after their signup already continued.
deadline = monotonic() + 1.5
if not _slots.acquire(blocking=False):
return None
try:
future = _pool.submit(_check, body, deadline)
except Exception:
_slots.release()
return None
future.add_done_callback(lambda _: _slots.release())
try:
return future.result(timeout=max(0, deadline - monotonic()))
except Exception:
future.cancel()
return None
def _check(body, deadline):
remaining = deadline - monotonic()
if remaining <= 0:
return None
with requests.post(
(os.environ.get("SIGNUPBOUNCER_BASE_URL") or "https://signupbouncer.com").rstrip("/") + "/v1/check",
headers={"Authorization": f"Bearer {os.environ.get('SIGNUPBOUNCER_API_KEY', '')}"},
json=body,
timeout=remaining,
allow_redirects=False, # a redirect is a failed check: your key and the address go nowhere else
stream=True,
) as r: # closes the connection on every path
if r.status_code != 200:
return None
data = bytearray()
# One decoded byte at a time also checks the deadline on a continuously trickling response. A blocked
# network read has its own timeout; its slot stays occupied until it exits, even after the caller returned.
for chunk in r.iter_content(chunk_size=1):
if monotonic() >= deadline:
return None
data.extend(chunk)
if len(data) > MAX_ANSWER_BYTES:
return None
if monotonic() >= deadline:
return None
answer = json.loads(data)
return {"message": _message(answer), "why_url": _why_url(answer)} if _is_block(answer) else None
if __name__ == "__main__":
import sys
refusal = signup_refusal(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else None)
print("allow" if refusal is None else "refuse")
- You wait at most 1.5 seconds for the check. At most eight Python checks run at once, with no queue. When all eight slots are busy, the signup goes through unchecked. The helper cancels a check that runs out of time before it starts, and stops reading the answer at the deadline. A network read that has already started can finish later, and holds its slot until then.
- Send the IP only from a header your host overwrites with the client's address, because the client can set a header that your host passes through unchanged. If you're unsure, leave ip out and we'll still check the email.
What the code does when a check fails
In every snippet, the call has a 1.5-second limit, and the code lets the signup through on any failure: a timeout, a network error, a non-2xx answer or an unreadable or incomplete body. It refuses a signup only after a complete answer with "action" and "verdict" both "block", "mode" set to "enforce", and no quota or error flag. The exception is an address that can't be real, which the code refuses without a call: one over 320 characters, or, in the Better Auth plugin, one that isn't an address at all. New apps start in shadow mode, where "action" is always "allow". So the code refuses no other signup until you switch the app to enforce in your dashboard. Each snippet checks only the signups that pass through it. So if your app also creates users somewhere else, such as an OAuth callback or an admin tool, call the check there too.