Intégrer CaptchaAI à Google Cloud Functions

Vous n'avez pas besoin d'une machine allumée en permanence pour résoudre des CAPTCHA. Sur Google Cloud Functions, chaque appel déclenche une instance à la demande : vous ne payez que le temps d'exécution réel, la mise à l'échelle est automatique et l'intégration avec Secret Manager, Pub/Sub et IAM reste native. Ce guide déploie une fonction de résolution CaptchaAI en mode serverless — d'abord une fonction HTTP synchrone, ensuite un pipeline par lots piloté par Pub/Sub — avec les coûts réels et les pièges d'exploitation à connaître.

La logique est la même quel que soit le type de défi : la fonction envoie la tâche à l'API CaptchaAI, interroge le résultat, puis renvoie le token. En changeant simplement le paramètre method, la même fonction couvre reCAPTCHA v2 et v3 (userrecaptcha), Cloudflare Turnstile (turnstile) ou GeeTest v3 (geetest) — tous pris en charge en version stable.

Serverless ou VM permanente : le calcul de coût

Avant d'écrire une ligne de code, posez la question du modèle de coût. L'intérêt du serverless apparaît sur les charges intermittentes. Une VM allumée en continu coûte le même prix qu'elle traite dix ou dix mille résolutions ; une fonction ne facture que l'exécution réelle. Le point de bascule se situe autour de plusieurs milliers de résolutions par jour, où une instance permanente redevient compétitive.

Ces montants ne couvrent que l'infrastructure GCP. Le coût de résolution CaptchaAI est séparé et dépend de votre plan par threads, pas du nombre de résolutions.

Une fonction HTTP qui résout le CAPTCHA

Le point d'entrée reçoit une requête JSON contenant la method et les params du CAPTCHA, récupère la clé API depuis Secret Manager, puis soumet la tâche et interroge le résultat. La clé n'apparaît jamais dans le code ni dans les variables d'environnement : c'est le point sensible d'un déploiement serverless, où le code source est souvent lisible par plusieurs équipes.

# main.py
import json
import time
import urllib.request
import urllib.parse
import functions_framework


@functions_framework.http
def solve_captcha(request):
    """HTTP Cloud Function for CAPTCHA solving."""
    # Parse request
    request_json = request.get_json(silent=True)
    if not request_json:
        return json.dumps({"error": "JSON body required"}), 400

    method = request_json.get("method", "userrecaptcha")
    params = request_json.get("params", {})

    # Get API key from Secret Manager
    api_key = _get_secret("captchaai-key")

    try:
        token = _solve(api_key, method, params)
        return json.dumps({"token": token})
    except Exception as e:
        return json.dumps({"error": str(e)}), 500


def _get_secret(secret_id):
    """Get secret from GCP Secret Manager."""
    from google.cloud import secretmanager
    client = secretmanager.SecretManagerServiceClient()
    name = f"projects/{_get_project_id()}/secrets/{secret_id}/versions/latest"
    response = client.access_secret_version(request={"name": name})
    return response.payload.data.decode("UTF-8")


def _get_project_id():
    """Get current GCP project ID."""
    import urllib.request
    req = urllib.request.Request(
        "http://metadata.google.internal/computeMetadata/v1/project/project-id",
        headers={"Metadata-Flavor": "Google"},
    )
    with urllib.request.urlopen(req) as resp:
        return resp.read().decode()


def _solve(api_key, method, params, timeout=90):
    """Solve CAPTCHA via CaptchaAI API."""
    # Submit
    submit_data = urllib.parse.urlencode({
        "key": api_key,
        "method": method,
        "json": 1,
        **params,
    }).encode()

    req = urllib.request.Request(
        "https://ocr.captchaai.com/in.php",
        data=submit_data,
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        result = json.loads(resp.read())

    if result.get("status") != 1:
        raise RuntimeError(f"Submit error: {result.get('request')}")

    task_id = result["request"]

    # Poll
    start = time.time()
    while time.time() - start < timeout:
        time.sleep(5)
        poll_url = (
            f"https://ocr.captchaai.com/res.php"
            f"?key={api_key}&action=get&id={task_id}&json=1"
        )
        with urllib.request.urlopen(poll_url, timeout=15) as resp:
            data = json.loads(resp.read())

        if data["request"] != "CAPCHA_NOT_READY":
            if data.get("status") == 1:
                return data["request"]
            raise RuntimeError(f"Solve error: {data['request']}")

    raise TimeoutError("Solve timeout")

Le module urllib de la bibliothèque standard est utilisé volontairement à la place de requests : moins de dépendances signifie un paquet plus léger et un démarrage à froid plus court, deux facteurs qui pèsent directement sur la latence d'une fonction serverless.

Les dépendances Python à déclarer

Gardez le fichier requirements.txt minimal. Seuls le framework des fonctions et le client Secret Manager sont nécessaires ; la résolution passe par urllib, déjà présent dans la bibliothèque standard.

# requirements.txt
functions-framework==3.*
google-cloud-secret-manager==2.*

Déployer la fonction sur Google Cloud Functions

Créez d'abord le secret, puis déployez la fonction en génération 2. La commande ci-dessous cible us-central1 ; pour un public francophone, remplacez la région par europe-west9 (Paris) ou europe-west1 (Belgique) afin de réduire la latence et de garder le traitement à l'intérieur de l'UE — un point utile pour vos obligations RGPD si les données appelantes contiennent des informations personnelles.

# Create secret
echo -n "YOUR_API_KEY" | gcloud secrets create captchaai-key --data-file=-

# Deploy function
gcloud functions deploy solve-captcha \
  --gen2 \
  --runtime=python311 \
  --region=us-central1 \
  --source=. \
  --entry-point=solve_captcha \
  --trigger-http \
  --allow-unauthenticated \
  --timeout=120s \
  --memory=256MB \
  --max-instances=100

# Test
curl -X POST https://us-central1-PROJECT.cloudfunctions.net/solve-captcha \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

Le paramètre --max-instances=100 plafonne le nombre d'instances parallèles. C'est aussi le levier à aligner sur votre plan CaptchaAI : la facturation se fait par thread concurrent, un thread correspondant à un CAPTCHA en cours. BASIC ($15/mois, 5 threads) suffit pour une charge légère ; montez vers STANDARD ($30/mois, 15 threads) ou ADVANCE ($90/mois, 50 threads) à mesure que le pic d'instances simultanées augmente, chaque plan offrant des résolutions illimitées par thread. Une fois l'URL affichée, la commande curl de test valide la chaîne complète, du secret à l'API : un code 200 renvoyant un champ token confirme que tout fonctionne.

Traitement par lots avec Pub/Sub

Pour un volume soutenu, l'appel HTTP synchrone n'est pas idéal : chaque requête bloque une instance pendant toute la résolution. Le modèle recommandé découple la soumission du résultat via Pub/Sub. Une fonction consomme les messages d'un sujet, résout le CAPTCHA, puis publie le token sur un sujet de résultats — la montée en charge suit automatiquement le nombre de messages en attente.

import base64
import json
import functions_framework
from google.cloud import pubsub_v1


@functions_framework.cloud_event
def process_captcha_task(cloud_event):
    """Process CAPTCHA task from Pub/Sub message."""
    data = base64.b64decode(cloud_event.data["message"]["data"])
    task = json.loads(data)

    api_key = _get_secret("captchaai-key")

    try:
        token = _solve(api_key, task["method"], task["params"])
        # Publish result
        publisher = pubsub_v1.PublisherClient()
        topic = f"projects/{_get_project_id()}/topics/captcha-results"
        publisher.publish(topic, json.dumps({
            "task_id": task["id"],
            "status": "success",
            "token": token,
        }).encode())

    except Exception as e:
        print(f"Task {task.get('id')} failed: {e}")

Pub/Sub réessaie tout message pour lequel la fonction lève une erreur. Attention donc aux échecs définitifs (paramètres invalides, page introuvable) : les renvoyer en erreur crée une boucle de nouvelles tentatives inutiles. Consignez l'échec, renvoyez un succès pour les cas non rejouables, et configurez un sujet de type dead-letter pour isoler les messages problématiques.

gcloud functions deploy process-captcha-task \
  --gen2 \
  --runtime=python311 \
  --trigger-topic=captcha-tasks \
  --timeout=120s \
  --memory=256MB

Envoyer les tâches dans la file Pub/Sub

Côté producteur, il suffit de publier un message JSON par URL à traiter. La file absorbe les pics, et les instances de la fonction consommatrice se multiplient pour suivre le rythme, dans la limite de votre allocation de threads.

from google.cloud import pubsub_v1
import json

publisher = pubsub_v1.PublisherClient()
topic = "projects/YOUR_PROJECT/topics/captcha-tasks"

# Submit batch
urls = ["https://site1.com", "https://site2.com", "https://site3.com"]
for i, url in enumerate(urls):
    task = {
        "id": f"task-{i}",
        "method": "userrecaptcha",
        "params": {"googlekey": "SITE_KEY", "pageurl": url},
    }
    publisher.publish(topic, json.dumps(task).encode())
    print(f"Published task-{i}")

Ajoutez un identifiant stable par tâche (id) pour rendre le traitement idempotent : si Pub/Sub livre deux fois le même message, vous pourrez ignorer le doublon côté résultats.

Dépannage

Questions fréquentes

Quel plan CaptchaAI choisir pour une fonction serverless ?

Alignez le plan sur votre pic d'instances simultanées, pas sur le volume total. La facturation étant par thread concurrent, BASIC ($15/mois, 5 threads) convient aux petites charges ; STANDARD ($30/mois, 15 threads) et ADVANCE ($90/mois, 50 threads) accompagnent une valeur --max-instances plus élevée. Toutes les résolutions par thread sont illimitées.

La même fonction peut-elle résoudre Turnstile et GeeTest v3 ?

Oui. Il suffit de changer le paramètre method : userrecaptcha pour reCAPTCHA v2/v3, turnstile pour Cloudflare Turnstile, geetest pour GeeTest v3 — tous pris en charge en version stable. En revanche, hCaptcha et FunCaptcha (Arkose Labs) ne sont pas pris en charge, et GeeTest v4 est annoncé « à venir ».

Faut-il déployer dans une région européenne ?

Si les données appelantes contiennent des informations personnelles, déployer sur europe-west9 (Paris) ou europe-west1 (Belgique) réduit la latence et garde le traitement dans l'UE. Minimisez les données personnelles transmises à la fonction et vérifiez vos obligations RGPD au cas par cas.

Comment limiter les démarrages à froid ?

Deux leviers : réduire la taille du paquet (d'où le choix de urllib) et maintenir une instance chaude. Cloud Scheduler peut envoyer un ping toutes les 5 minutes, ou définissez --min-instances=1 pour garder une instance prête en continu (environ 7 $/mois).

Guides connexes

• Déployer CaptchaAI sur AWS Lambda
• Intégrer CaptchaAI à Azure Functions

Serverless sur GCP — récupérez votre clé CaptchaAI dès aujourd'hui.