Flask bot detection and email validation
A decorator you apply to the handful of routes worth protecting, rather than a global before_request that bills you for favicons.
1. Install
pip install requests flask
2. Add the file
conure.py
import functools
import logging
import os
import requests
from flask import jsonify, request
log = logging.getLogger(__name__)
CONURE = "https://uat.conureapi.com"
API_KEY = os.environ.get("CONURE_API_KEY", "")
def client_ip():
forwarded = request.headers.get("X-Forwarded-For", "")
if forwarded:
return forwarded.split(",")[0].strip()
return request.remote_addr or ""
def block_bots(view):
@functools.wraps(view)
def wrapper(*args, **kwargs):
ip = client_ip()
if not (ip and API_KEY):
return view(*args, **kwargs)
try:
response = requests.post(
CONURE + "/v1/bot-check",
headers={"Authorization": "Bearer " + API_KEY},
json={"ip": ip, "user_agent": request.headers.get("User-Agent", "")},
timeout=3,
)
response.raise_for_status()
verdict = response.json()
except (requests.RequestException, ValueError) as error:
log.warning("conure unreachable: %s", error)
return view(*args, **kwargs) # fail open
if verdict.get("is_bot") is True:
return jsonify(
error="Automated traffic is not permitted.",
reasons=verdict.get("reasons", []),
), 403
return view(*args, **kwargs)
return wrapper3. Wire it up
from flask import Flask
from conure import block_bots
app = Flask(__name__)
@app.post("/signup")
@block_bots
def signup():
...4. Verify
Confirm the API answers before you debug your Flask
wiring. The sample address sits inside a published AWS range, so a correct setup
returns is_bot: true.
curl -X POST https://uat.conureapi.com/v1/bot-check \
-H "Authorization: Bearer YOUR_API_KEY" \
-H "Content-Type: application/json" \
-d '{"ip":"52.1.2.3","user_agent":"curl/8.4.0"}'
import requests
HEADERS = {"Authorization": "Bearer YOUR_API_KEY"}
response = requests.post(
"https://uat.conureapi.com/v1/bot-check",
headers=HEADERS,
json={"ip": "52.1.2.3", "user_agent": "curl/8.4.0"},
timeout=5,
)
response.raise_for_status()
print(response.json())
const response = await fetch("https://uat.conureapi.com/v1/bot-check", {
method: "POST",
headers: {
Authorization: "Bearer YOUR_API_KEY",
"Content-Type": "application/json",
},
body: JSON.stringify({"ip":"52.1.2.3","user_agent":"curl/8.4.0"}),
signal: AbortSignal.timeout(5000),
});
if (!response.ok) throw new Error("conure: " + response.status);
console.log(await response.json());
$curl = curl_init("https://uat.conureapi.com/v1/bot-check");
curl_setopt_array($curl, [
CURLOPT_RETURNTRANSFER => true,
CURLOPT_TIMEOUT => 5,
CURLOPT_POST => true,
CURLOPT_POSTFIELDS => '{"ip":"52.1.2.3","user_agent":"curl/8.4.0"}',
CURLOPT_HTTPHEADER => [
"Authorization: Bearer YOUR_API_KEY",
"Content-Type: application/json",
],
]);
$body = curl_exec($curl);
$status = curl_getinfo($curl, CURLINFO_RESPONSE_CODE);
curl_close($curl);
$verdict = $status === 200 ? json_decode($body, true) : null;
var_dump($verdict);
require "net/http"
require "json"
uri = URI("https://uat.conureapi.com/v1/bot-check")
request = Net::HTTP::Post.new(uri)
request["Authorization"] = "Bearer YOUR_API_KEY"
request["Content-Type"] = "application/json"
request.body = { ip: "52.1.2.3", user_agent: "curl/8.4.0" }.to_json
response = Net::HTTP.start(uri.hostname, uri.port,
use_ssl: uri.scheme == "https",
open_timeout: 5, read_timeout: 5) do |http|
http.request(request)
end
puts JSON.parse(response.body)
Notes
- requests is synchronous. Under gunicorn, run enough workers that a three-second upstream cannot exhaust the pool.
- Behind nginx or Cloudflare, use ProxyFix if you drop the X-Forwarded-For parsing above.