Gérer les CAPTCHA dans une application Flask avec CaptchaAI

Deux besoins distincts reviennent dès qu'un CAPTCHA croise une application Flask :

Résolution côté serveur : votre script d'automatisation rencontre un CAPTCHA et doit obtenir un token valide via l'API CaptchaAI.
Vérification côté formulaire : un visiteur remplit l'un de vos formulaires protégé par Cloudflare Turnstile, et vous validez sa réponse avant de traiter la soumission.

CaptchaAI couvre le premier cas via une simple API HTTP ; Cloudflare fournit l'endpoint de vérification pour le second. Ce guide construit les deux, étape par étape : une classe de service réutilisable, des endpoints Flask synchrones, une résolution en arrière-plan par threads, puis une organisation en Blueprint pour les projets plus conséquents.

Mise en place du projet Flask

Deux paquets suffisent pour démarrer : Flask pour le serveur et requests pour les appels à l'API CaptchaAI.

pip install flask requests

Arborescence du projet

Isolez la logique de résolution dans un module services/ : elle reste ainsi testable et réutilisable d'un endpoint à l'autre.

myapp/
├── app.py
├── config.py
├── services/
│   └── captcha_solver.py
└── templates/
    └── form.html

Trois emplacements structurent le projet :

services/captcha_solver.py — la classe qui parle à l'API CaptchaAI, indépendante de Flask.
app.py — le point d'entrée qui expose les routes et charge la configuration.
templates/form.html — le gabarit du formulaire protégé par Turnstile.

Service CaptchaAI

Toute la communication avec l'API suit le même cycle, quel que soit le type de CAPTCHA :

Envoyer la tâche sur in.php et récupérer un identifiant.
Interroger res.php à intervalle régulier jusqu'à ce que le statut passe à 1.
Retourner le token, ou lever une erreur si la résolution échoue ou expire.

La classe CaptchaSolver encapsule ce cycle une fois pour toutes, avec une méthode dédiée par type (reCAPTCHA v2, Turnstile, image) et une gestion centralisée dans _submit_and_poll.

# services/captcha_solver.py
import time
import requests


class CaptchaSolver:
    """CaptchaAI solver service for Flask applications."""

    API_BASE = "https://ocr.captchaai.com"

    def __init__(self, api_key):
        self.api_key = api_key

    def solve_recaptcha_v2(self, sitekey, page_url):
        """Solve reCAPTCHA v2."""
        return self._submit_and_poll({
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": page_url,
        })

    def solve_turnstile(self, sitekey, page_url):
        """Solve Cloudflare Turnstile."""
        return self._submit_and_poll({
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": page_url,
        })

    def solve_image(self, image_base64):
        """Solve image CAPTCHA."""
        return self._submit_and_poll({
            "method": "base64",
            "body": image_base64,
        })

    def get_balance(self):
        """Check API balance."""
        resp = requests.get(f"{self.API_BASE}/res.php", params={
            "key": self.api_key,
            "action": "getbalance",
            "json": 1,
        }, timeout=30)
        return float(resp.json().get("request", 0))

    def _submit_and_poll(self, params, timeout=120):
        """Submit and poll for result."""
        submit_data = {"key": self.api_key, "json": 1, **params}

        resp = requests.post(f"{self.API_BASE}/in.php", data=submit_data, timeout=30)
        resp.raise_for_status()
        data = resp.json()

        if data.get("status") != 1:
            raise CaptchaSolveError(f"Submit failed: {data.get('request')}")

        task_id = data["request"]

        start = time.time()
        while time.time() - start < timeout:
            time.sleep(5)
            result = requests.get(f"{self.API_BASE}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1,
            }, timeout=30).json()

            if result.get("status") == 1:
                return result["request"]
            if result.get("request") == "ERROR_CAPTCHA_UNSOLVABLE":
                raise CaptchaSolveError("CAPTCHA unsolvable")

        raise CaptchaSolveError("Solve timed out")


class CaptchaSolveError(Exception):
    pass

Le paramètre timeout=120 borne l'attente : la résolution est levée en erreur plutôt que de bloquer indéfiniment un worker.

Exposer la résolution via des endpoints Flask

Instanciez le service une seule fois au démarrage de l'application, puis réutilisez-le dans chaque route. Chaque endpoint valide ses entrées, renvoie 400 si le sitekey ou l'URL manque, et 500 si la résolution échoue.

# app.py
from flask import Flask, request, jsonify
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

app = Flask(__name__)
app.config["CAPTCHAAI_API_KEY"] = "YOUR_API_KEY"

solver = CaptchaSolver(app.config["CAPTCHAAI_API_KEY"])


@app.route("/solve/recaptcha", methods=["POST"])
def solve_recaptcha():
    """Solve reCAPTCHA v2 via API."""
    data = request.get_json()
    sitekey = data.get("sitekey")
    page_url = data.get("url")

    if not sitekey or not page_url:
        return jsonify({"error": "sitekey and url required"}), 400

    try:
        token = solver.solve_recaptcha_v2(sitekey, page_url)
        return jsonify({"token": token})
    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/solve/turnstile", methods=["POST"])
def solve_turnstile():
    """Solve Cloudflare Turnstile via API."""
    data = request.get_json()
    sitekey = data.get("sitekey")
    page_url = data.get("url")

    if not sitekey or not page_url:
        return jsonify({"error": "sitekey and url required"}), 400

    try:
        token = solver.solve_turnstile(sitekey, page_url)
        return jsonify({"token": token})
    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@app.route("/balance", methods=["GET"])
def check_balance():
    """Check CaptchaAI balance."""
    balance = solver.get_balance()
    return jsonify({"balance": balance})


if __name__ == "__main__":
    app.run(debug=True, port=5000)

Tester les endpoints

Trois appels couvrent la surface exposée :

/solve/recaptcha — renvoie un token reCAPTCHA v2 pour un sitekey et une URL.
/solve/turnstile — le même contrat, pour Cloudflare Turnstile.
/balance — vérifie le solde ; c'est le réflexe le plus simple pour confirmer que la clé API est valide avant de lancer un lot.

# Solve reCAPTCHA
curl -X POST http://localhost:5000/solve/recaptcha \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "6Le-wvkSAAAA...", "url": "https://example.com/login"}'

# Solve Turnstile
curl -X POST http://localhost:5000/solve/turnstile \
  -H "Content-Type: application/json" \
  -d '{"sitekey": "0x4AAAAAAAC3DHQ...", "url": "https://example.com/signup"}'

# Check balance
curl http://localhost:5000/balance

Protéger un formulaire Flask avec Cloudflare Turnstile

L'autre moitié du sujet : vous ne résolvez plus un CAPTCHA, vous vérifiez celui qu'un visiteur a produit. Le mécanisme Turnstile tient en deux temps :

le widget ajoute un champ cf-turnstile-response au formulaire ;
côté serveur, vous validez ce token auprès de Cloudflare avant de traiter la soumission.

# app.py
from flask import Flask, request, render_template, redirect, url_for, flash
import requests as http_requests

app = Flask(__name__)
app.secret_key = "your-secret-key"
app.config["TURNSTILE_SITE_KEY"] = "0x4AAAAAAAC3DHQhMMQ_Rxrg"
app.config["TURNSTILE_SECRET_KEY"] = "0x4AAAAAAAC3DHQhYYY_secret"


def verify_turnstile(token, remote_ip=None):
    """Verify Turnstile token with Cloudflare."""
    data = {
        "secret": app.config["TURNSTILE_SECRET_KEY"],
        "response": token,
    }
    if remote_ip:
        data["remoteip"] = remote_ip

    resp = http_requests.post(
        "https://challenges.cloudflare.com/turnstile/v0/siteverify",
        data=data,
        timeout=10,
    )
    return resp.json().get("success", False)


@app.route("/contact", methods=["GET", "POST"])
def contact():
    if request.method == "POST":
        turnstile_token = request.form.get("cf-turnstile-response")

        if not turnstile_token:
            flash("CAPTCHA required")
            return redirect(url_for("contact"))

        if not verify_turnstile(turnstile_token, request.remote_addr):
            flash("CAPTCHA verification failed")
            return redirect(url_for("contact"))

        # Process the form
        name = request.form.get("name")
        email = request.form.get("email")
        # ... save or email the data
        flash("Message sent successfully")
        return redirect(url_for("contact"))

    return render_template("form.html",
                           turnstile_sitekey=app.config["TURNSTILE_SITE_KEY"])

<!-- templates/form.html -->
<!DOCTYPE html>
<html>
<body>
    <form method="post">
        <input name="name" placeholder="Name" required>
        <input name="email" type="email" placeholder="Email" required>
        <textarea name="message" placeholder="Message" required></textarea>
        <div class="cf-turnstile" data-sitekey="{{ turnstile_sitekey }}"></div>
        <button type="submit">Send</button>
    </form>
    <script src="https://challenges.cloudflare.com/turnstile/v0/api.js" async defer></script>
</body>
</html>

Côté RGPD, restez sobre sur les données : le token Turnstile et l'adresse IP servent uniquement à la vérification et n'ont pas à être conservés une fois le formulaire validé. Ne journalisez que ce qui est nécessaire au diagnostic.

Résolution en arrière-plan avec des threads

Flask traite les requêtes de façon synchrone : une résolution qui dure une minute bloque le worker qui la porte. La parade tient en deux gestes :

déportez le travail de résolution dans un thread dédié ;
rendez immédiatement un identifiant que le client interrogera ensuite.

import uuid
import threading
from flask import Flask, request, jsonify
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

app = Flask(__name__)
solver = CaptchaSolver("YOUR_API_KEY")

# In-memory task storage (use Redis in production)
tasks = {}


def solve_in_background(task_id, captcha_type, sitekey, page_url):
    """Background CAPTCHA solver."""
    try:
        if captcha_type == "recaptcha_v2":
            token = solver.solve_recaptcha_v2(sitekey, page_url)
        elif captcha_type == "turnstile":
            token = solver.solve_turnstile(sitekey, page_url)
        else:
            raise ValueError(f"Unknown type: {captcha_type}")

        tasks[task_id] = {"status": "solved", "token": token}

    except CaptchaSolveError as e:
        tasks[task_id] = {"status": "failed", "error": str(e)}


@app.route("/solve/async", methods=["POST"])
def solve_async():
    """Submit CAPTCHA for background solving."""
    data = request.get_json()
    captcha_type = data.get("type", "recaptcha_v2")
    sitekey = data.get("sitekey")
    page_url = data.get("url")

    if not sitekey or not page_url:
        return jsonify({"error": "sitekey and url required"}), 400

    task_id = str(uuid.uuid4())
    tasks[task_id] = {"status": "pending"}

    thread = threading.Thread(
        target=solve_in_background,
        args=(task_id, captcha_type, sitekey, page_url),
    )
    thread.start()

    return jsonify({"task_id": task_id}), 202


@app.route("/solve/status/<task_id>")
def solve_status(task_id):
    """Check solving status."""
    task = tasks.get(task_id)
    if not task:
        return jsonify({"error": "Task not found"}), 404
    return jsonify(task)

Le client rappelle /solve/status/<task_id> jusqu'à obtenir solved ou failed. Ce stockage en mémoire suffit pour une démo, pas pour la production : sur un hébergeur comme OVHcloud ou Scaleway, plusieurs workers WSGI ne partagent pas le même dictionnaire tasks. Passez à Redis dès que vous dépassez un seul processus.

# Submit async solve
curl -X POST http://localhost:5000/solve/async \
  -H "Content-Type: application/json" \
  -d '{"type": "turnstile", "sitekey": "0x4AAA...", "url": "https://example.com"}'
# Returns: {"task_id": "abc-123-..."}

# Check status
curl http://localhost:5000/solve/status/abc-123-...
# Returns: {"status": "pending"}  or  {"status": "solved", "token": "..."}

Organiser les routes avec un Flask Blueprint

Sur un projet qui grossit, regroupez les routes de résolution dans un Blueprint dédié plutôt que de les empiler dans app.py. Le service est reconstruit à la demande à partir de current_app.config, ce qui évite toute clé API codée en dur au niveau du module.

# blueprints/captcha.py
from flask import Blueprint, request, jsonify, current_app
from services.captcha_solver import CaptchaSolver, CaptchaSolveError

captcha_bp = Blueprint("captcha", __name__, url_prefix="/api/captcha")


def get_solver():
    return CaptchaSolver(current_app.config["CAPTCHAAI_API_KEY"])


@captcha_bp.route("/solve", methods=["POST"])
def solve():
    data = request.get_json()
    captcha_type = data.get("type")
    sitekey = data.get("sitekey")
    url = data.get("url")

    solver = get_solver()

    try:
        if captcha_type == "recaptcha_v2":
            token = solver.solve_recaptcha_v2(sitekey, url)
        elif captcha_type == "turnstile":
            token = solver.solve_turnstile(sitekey, url)
        elif captcha_type == "image":
            image_b64 = data.get("image")
            token = solver.solve_image(image_b64)
        else:
            return jsonify({"error": f"Unknown type: {captcha_type}"}), 400

        return jsonify({"token": token})

    except CaptchaSolveError as e:
        return jsonify({"error": str(e)}), 500


@captcha_bp.route("/balance")
def balance():
    solver = get_solver()
    return jsonify({"balance": solver.get_balance()})

# app.py
from flask import Flask
from blueprints.captcha import captcha_bp

app = Flask(__name__)
app.config["CAPTCHAAI_API_KEY"] = "YOUR_API_KEY"
app.register_blueprint(captcha_bp)

Cette organisation apporte trois avantages concrets :

Préfixe d'URL unique : toutes les routes sont servies sous /api/captcha, faciles à versionner.
Aucune clé en dur : le service se reconstruit depuis current_app.config à chaque requête.
Isolation : vous pouvez appliquer une authentification ou un rate limiting au seul Blueprint, sans toucher au reste de l'application.

Questions fréquentes

Flask ou Django pour intégrer CaptchaAI ?

Flask convient mieux aux API légères et aux microservices ; Django prend l'avantage sur les applications web complètes avec back-office. Le motif d'intégration CaptchaAI reste identique dans les deux cas : une classe de service qui gère le cycle envoi/interrogation.

Comment protéger ma clé API CaptchaAI dans une application Flask ?

Ne l'écrivez jamais en clair dans le code. Chargez-la depuis une variable d'environnement ou un gestionnaire de secrets, injectez-la dans app.config, et récupérez-la via current_app.config comme dans l'exemple du Blueprint.

Ai-je besoin de Redis pour les tâches en arrière-plan ?

Pas pour prototyper : le dictionnaire en mémoire suffit tant que vous restez sur un seul processus. Dès que vous passez à plusieurs workers Gunicorn, chacun a sa propre mémoire, et un stockage partagé comme Redis devient nécessaire pour suivre l'état des tâches.

Combien coûte la résolution de CAPTCHA avec CaptchaAI ?

La facturation est basée sur les threads, pas sur le nombre de résolutions : chaque plan inclut des résolutions illimitées sur ses threads concurrents. L'offre d'entrée BASIC ($15/mois, 5 threads) suffit à un petit service Flask ; montez en gamme quand votre débit l'exige.

Comment gérer les délais d'expiration des requêtes ?

Alignez le timeout de votre serveur WSGI sur la durée de résolution (Gunicorn : --timeout 180). Une résolution prend généralement de 15 à 120 secondes, d'où l'intérêt du motif asynchrone pour ne pas immobiliser un worker pendant tout ce temps.

Dépannage

Symptôme	Cause probable	Correctif
La requête reste bloquée plus de 2 minutes	Une résolution synchrone bloque le worker Flask	Basculez sur le motif threads ou async
`ConnectionError`	API CaptchaAI injoignable	Vérifiez le réseau et les règles de pare-feu
Token renvoyé vide	Réponse JSON mal interprétée	Contrôlez le format de la réponse et le champ `request`
La vérification Turnstile échoue	`TURNSTILE_SECRET_KEY` incorrecte	Revérifiez la clé secrète côté serveur
La mémoire grimpe sur les tâches en arrière-plan	Le dictionnaire `tasks` n'est jamais purgé	Ajoutez un TTL et un nettoyage périodique