AWS Lambda et CaptchaAI pour une résolution CAPTCHA serverless

AWS Lambda permet de traiter la résolution CAPTCHA sans maintenir de serveur permanent. C'est un bon choix si vous avez des besoins intermittents, des pics de charge ou un pipeline déjà construit autour d'API Gateway, SQS ou Step Functions. Vous payez l'exécution, vous laissez AWS gérer le scaling, et vous gardez une architecture simple à déployer.

L'enjeu n'est pas seulement d'appeler CaptchaAI depuis une fonction. Il faut aussi gérer proprement la clé API, calibrer les timeouts et choisir un mode de déclenchement adapté à votre charge.

Handler Lambda

# lambda_function.py
import json
import os
import time
import urllib.request
import urllib.parse


def lambda_handler(event, context):
    """AWS Lambda handler for CaptchaAI solving."""
    api_key = os.environ["CAPTCHAAI_KEY"]

    # Parse input
    body = json.loads(event.get("body", "{}")) if isinstance(event.get("body"), str) else event

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

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


def solve_captcha(api_key, method, params, timeout=90):
    """Solve CAPTCHA using CaptchaAI API."""
    # Submit task
    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 for result
    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")

Sécuriser la clé API avec Secrets Manager

import json
import boto3


def get_api_key():
    """Retrieve CaptchaAI key from AWS Secrets Manager."""
    client = boto3.client("secretsmanager")
    response = client.get_secret_value(SecretId="captchaai/api-key")
    secret = json.loads(response["SecretString"])
    return secret["api_key"]

Création du secret :

aws secretsmanager create-secret \
  --name captchaai/api-key \
  --secret-string '{"api_key":"YOUR_API_KEY"}'

Template SAM pour l'infrastructure

# template.yaml
AWSTemplateFormatVersion: "2010-09-09"
Transform: AWS::Serverless-2016-10-31

Globals:
  Function:
    Timeout: 120
    MemorySize: 256
    Runtime: python3.11

Resources:
  CaptchaSolverFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: lambda_function.lambda_handler
      Environment:
        Variables:
          CAPTCHAAI_KEY: !Sub "{{resolve:secretsmanager:captchaai/api-key:SecretString:api_key}}"
      Events:
        SolveApi:
          Type: Api
          Properties:
            Path: /solve
            Method: post
      Policies:

        - AWSSecretsManagerGetSecretValuePolicy:
            SecretArn: !Sub "arn:aws:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:captchaai/api-key-*"

Outputs:
  SolveApiUrl:
    Value: !Sub "https://${ServerlessRestApi}.execute-api.${AWS::Region}.amazonaws.com/Prod/solve"

Déploiement

# Build and deploy
sam build
sam deploy --guided

# Test
curl -X POST https://YOUR_API_ID.execute-api.us-east-1.amazonaws.com/Prod/solve \
  -H "Content-Type: application/json" \
  -d '{
    "method": "userrecaptcha",
    "params": {
      "googlekey": "SITE_KEY",
      "pageurl": "https://example.com"
    }
  }'

Traitement batch déclenché par SQS

Si vous avez beaucoup de résolutions à exécuter, une file SQS en entrée permet de découpler la soumission des tâches de leur exécution.

import json
import os
import time
import urllib.request
import urllib.parse


def sqs_handler(event, context):
    """Process CAPTCHA tasks from SQS queue."""
    api_key = os.environ["CAPTCHAAI_KEY"]
    results = []

    for record in event["Records"]:
        task = json.loads(record["body"])
        try:
            token = solve_captcha(
                api_key,
                task["method"],
                task["params"],
            )
            results.append({
                "task_id": task.get("id"),
                "status": "success",
                "token": token[:50],
            })
        except Exception as e:
            results.append({
                "task_id": task.get("id"),
                "status": "error",
                "error": str(e),
            })

    return {"results": results}

Points d'exploitation Lambda

Depannage

FAQ

Lambda est-il pertinent économiquement ?

Oui, surtout si la charge est variable. Vous évitez un serveur dédié pour un composant qui n'a pas besoin de tourner en permanence.

Quel timeout faut-il configurer ?

Dans la plupart des cas, 120 secondes suffisent. Si vous traitez des types plus lourds ou si votre pipeline ajoute de la latence réseau, prévoyez une marge supplémentaire.

Faut-il embarquer requests dans Lambda ?

Pas obligatoirement. Pour cet usage, urllib.request suffit souvent et simplifie nettement le package de déploiement.

Guides connexes

• Fonctions Google Cloud + CaptchaAI
• Fonctions Azure + CaptchaAI

Passez à une résolution serverless : récupérez votre clé CaptchaAI et branchez-la sur votre pipeline AWS.