AWS Lambda + CaptchaAI: Giải mã CAPTCHA không cần máy chủ

AWS Lambda xử lý việc giải CAPTCHA mà không cần quản lý máy chủ. Chỉ trả tiền khi giải quyết, tự động mở rộng quy mô và tích hợp với API Gateway, SQS hoặc Step Functions.

Trình xử lý 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")

Khóa API bảo mật với Trình quản lý bí mật

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"]

Lưu trữ bí mật:

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

Mẫu SAM (Cơ sở hạ tầng dưới dạng mã)

# 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"

Triển khai

# 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"
    }
  }'

Xử lý hàng loạt được kích hoạt SQS

Xử lý các tác vụ CAPTCHA từ hàng đợi SQS:

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}

Cân nhắc về Lambda

Khắc phục sự cố

Câu hỏi thường gặp

Lambda có hiệu quả về mặt chi phí khi giải CAPTCHA không?

Vâng. Với mức giá ~$0,0001 cho mỗi lệnh gọi (256 MB, 60 giây), Lambda tính thêm chi phí không đáng kể ngoài phí API CaptchaAI. Bạn tránh được chi phí máy chủ trong thời gian nhàn rỗi.

Còn thời gian chờ 15 phút của Lambda thì sao?

Hầu hết CAPTCHA giải quyết trong 10-60 giây. Đặt thời gian chờ Lambda của bạn thành 120 giây. Đối với các loại phức tạp như reCAPTCHA Enterprise, hãy sử dụng 180 giây.

Tôi có thể sử dụng các lớp Lambda cho thư viện yêu cầu không?

Có, nhưng urllib.request (tích hợp sẵn) hoạt động tốt với API HTTP đơn giản của CaptchaAI. Điều này tránh hoàn toàn việc quản lý lớp.

Hướng dẫn liên quan

• Chức năng đám mây của Google + CaptchaAI
• Chức năng Azure + CaptchaAI

Trở nên không có máy chủ —lấy khóa CaptchaAI của bạnHôm nay.