Vous avez des résolutions CAPTCHA à traiter par à-coups, sans vouloir payer un serveur qui tourne en continu ? AWS Lambda répond précisément à ce besoin : vous ne payez que le temps d'exécution réel, le scaling est automatique et l'intégration avec API Gateway, SQS ou Step Functions est immédiate.
La résolution CAPTCHA serverless sur Lambda ne se résume pourtant pas à un appel d'API depuis une fonction. Trois points font la différence entre un prototype et une brique fiable : la gestion de la clé API, le calibrage des timeouts et le choix du déclencheur adapté à votre profil de charge. Ce guide les déroule dans l'ordre, du handler Python jusqu'au traitement par lots via SQS.
Le handler Lambda de résolution CAPTCHA
Le cœur de l'intégration tient dans une seule fonction. Le handler lit la clé API depuis une variable d'environnement, parse la requête entrante, soumet la tâche à l'API CaptchaAI via in.php, puis interroge régulièrement res.php jusqu'à obtenir le token. Le code ci-dessous couvre la soumission et le polling, avec un timeout applicatif par défaut de 90 secondes.
# 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")
Deux détails d'exploitation méritent attention. Le polling interroge le résultat toutes les 5 secondes : c'est un bon compromis entre réactivité et nombre d'appels. Et le timeout applicatif (90 s) doit toujours rester inférieur au timeout configuré sur la fonction Lambda elle-même, sinon AWS coupe l'exécution avant que votre code ne renvoie une erreur exploitable.
Sécuriser la clé API avec AWS Secrets Manager
Ne codez jamais la clé API en dur, et évitez de la laisser en clair dans une variable d'environnement visible depuis la console. AWS Secrets Manager centralise le secret, le chiffre au repos et journalise les accès — un réflexe d'autant plus utile si vous devez documenter vos traitements au titre du RGPD.
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"}'
La fonction peut alors récupérer le secret au démarrage. Mieux encore, laissez CloudFormation le résoudre au déploiement (voir le template SAM ci-dessous) : la clé arrive dans la variable d'environnement sans appel réseau supplémentaire à chaque invocation.
Décrire l'infrastructure avec un template SAM
Plutôt que de cliquer dans la console, décrivez toute la pile en Infrastructure as Code. Le template SAM suivant crée la fonction, injecte la clé API depuis Secrets Manager, expose un endpoint POST /solve via API Gateway et attache la politique IAM minimale nécessaire pour lire le secret.
# 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"
Notez le Timeout: 120 défini au niveau global : il laisse à la fonction le temps de terminer un cycle de polling complet, y compris sur les types les plus lents à résoudre.
Déployer et tester la fonction
Deux commandes suffisent pour construire et déployer la pile. L'option --guided de sam deploy vous pose les questions de configuration la première fois, puis mémorise vos réponses dans samconfig.toml pour les déploiements suivants.
# 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"
}
}'
L'URL renvoyée en sortie (SolveApiUrl) est votre endpoint de production. Pour un public francophone, déployez de préférence dans une région proche : eu-west-3 (Paris) réduit la latence réseau entre API Gateway et vos clients en France, en Belgique ou en Suisse, sans rien changer au code.
Traitement par lots déclenché par SQS
Pour un gros volume, ne lancez pas la résolution en synchrone depuis API Gateway. Placez une file SQS en entrée : elle découple la soumission des tâches de leur exécution, absorbe les pics et rejoue automatiquement les messages en échec. Chaque message porte une tâche, et le handler les traite par lot.
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}
Ce découplage change aussi le modèle de coût : vous lissez la charge sur la concurrence disponible au lieu de risquer un pic d'invocations simultanées, et vous gardez une file de secours si l'API met plus longtemps que prévu à répondre.
Réglages d'exploitation sur Lambda
Quelques valeurs de référence pour dimensionner la fonction sans sur-provisionner :
| Facteur | Valeur |
|---|---|
| Timeout maximum | 15 minutes, mais 120 s suffisent souvent |
| Mémoire | 256 Mo suffisent dans la plupart des cas |
| Concurrence | 1 000 exécutions par défaut selon le compte |
| Cold start | Faible impact face au temps de résolution |
| Coût | Très bas par invocation, hors coût de l'API |
| Dépendances | urllib évite d'ajouter des layers inutiles |
Côté facturation CaptchaAI, gardez en tête que la tarification est basée sur les threads, pas sur le nombre de résolutions : l'offre BASIC ($15/mois, 5 threads) couvre déjà une charge serverless modérée, et vous montez en gamme uniquement quand vous avez besoin de plus de résolutions en parallèle.
Dépannage
| Problème | Cause probable | Correctif |
|---|---|---|
| La fonction expire | Timeout Lambda trop court | Passez à 120 s ou plus |
| Accès refusé au secret | Politique IAM absente | Ajoutez la permission Secrets Manager |
| Latence au premier appel | Cold start visible | Activez la provisioned concurrency si nécessaire |
| Import requests indisponible | Dépendance non embarquée | Utilisez urllib.request ou ajoutez un layer |
FAQ
API Gateway ou SQS : quel mode de déclenchement retenir ?
API Gateway convient aux résolutions unitaires où l'appelant attend le token en réponse directe. SQS s'impose dès que le volume grimpe ou que vous voulez lisser les pics : la file absorbe la charge et rejoue les échecs sans bloquer l'émetteur.
Quelle région AWS privilégier pour un public francophone ?
Déployez dans eu-west-3 (Paris). Vous réduisez la latence réseau pour les utilisateurs en France, en Belgique et en Suisse, et vous restez au plus près de vos autres services européens si votre pipeline y est déjà hébergé.
Le cold start ralentit-il la résolution ?
Marginalement. Le démarrage à froid d'une fonction Python se compte en centaines de millisecondes, négligeable face aux secondes que prend la résolution elle-même. Si vos invocations sont très espacées, activez la provisioned concurrency.
Faut-il embarquer la bibliothèque requests dans un layer ?
Non, ce n'est pas nécessaire. Le module urllib.request, inclus dans la bibliothèque standard, suffit pour l'API HTTP de CaptchaAI et simplifie nettement le package de déploiement.
Guides connexes
Prêt à passer au serverless ? Récupérez votre clé CaptchaAI et branchez-la sur votre pipeline AWS Lambda.
