Google Cloud Functions + CaptchaAI-Integration

Eine Always-On-VM nur für gelegentliche CAPTCHA-Lösungen laufen zu lassen, bezahlt vor allem Leerlauf. Google Cloud Functions dreht das um: Ihre Funktion startet pro Anfrage, ruft die CaptchaAI-API auf und skaliert automatisch mit der Last – abgerechnet wird nur die tatsächliche Ausführungszeit. Diese Anleitung zeigt eine HTTP-Funktion, Batch-Verarbeitung über Pub/Sub und den sicheren Umgang mit dem API-Schlüssel.

Wann sich der serverlose Ansatz lohnt

Serverlos passt zu ungleichmäßiger Last: mal keine, mal hunderte Lösungen pro Minute. Für GCP-Compute zahlen Sie nur bei tatsächlicher Ausführung; der Preis für die CaptchaAI-Lösung läuft davon getrennt. CaptchaAI rechnet pro gleichzeitigem Thread ab, nicht pro Lösung – Pläne starten bei BASIC (15 $/Monat, 5 Threads) mit unbegrenzten Lösungen pro Thread. Der einzige Haken sind Kaltstarts: Die erste Anfrage nach einer Leerlaufphase dauert ein paar hundert Millisekunden länger. Bei konstant hoher Dauerlast kann eine feste VM günstiger sein – der serverlose Vorteil liegt im schwankenden Bedarf, nicht im Rund-um-die-Uhr-Betrieb.

HTTP-Funktion für die CAPTCHA-Lösung

Die Funktion nimmt einen JSON-Body entgegen, übermittelt die Aufgabe an in.php, fragt den Status über res.php ab und gibt das Token zurück. Den API-Schlüssel holt sie zur Laufzeit aus dem GCP Secret Manager – so landet er weder im Quellcode noch in Umgebungsvariablen.

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

Abhängigkeiten

Zwei Pakete genügen – das Functions Framework und der Secret-Manager-Client:

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

Funktion bereitstellen und testen

Legen Sie den Schlüssel als Secret an und deployen Sie die Funktion als Gen2. Wichtig: Das --timeout muss über der maximalen Lösungszeit liegen, sonst bricht GCP die Funktion ab, bevor das Token bereit ist. Der abschließende curl-Aufruf prüft den Endpunkt.

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

Welche CAPTCHA-Typen die Funktion abdeckt

Der method-Parameter im JSON-Body steuert, welchen Typ CaptchaAI löst – der restliche Code bleibt unverändert. Für reCAPTCHA v2 und v3 (auch Enterprise) verwenden Sie userrecaptcha, für Cloudflare Turnstile turnstile, für die Cloudflare Challenge cloudflare_challenge, für GeeTest v3 geetest und für Bild- oder Rasterbild-CAPTCHAs post. CaptchaFox, Friendly Captcha und Lemin sind als Beta verfügbar. hCaptcha, FunCaptcha und GeeTest v4 werden derzeit nicht unterstützt – GeeTest v4 ist als „bald verfügbar“ angekündigt. So bündeln Sie mehrere unterstützte CAPTCHA-Typen hinter einer einzigen serverlosen Funktion.

Batch-Verarbeitung über Pub/Sub

Für hohes Volumen entkoppeln Sie Einreichung und Verarbeitung: Ein Producer legt Aufgaben in ein Pub/Sub-Thema, eine Event-Funktion verarbeitet sie parallel und schreibt die Ergebnisse in ein zweites Thema. So verarbeiten Sie CAPTCHA-Aufgaben aus einem Pub/Sub-Thema:

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}")

Für Pub/Sub bereitstellen:

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

Aufgaben an Pub/Sub senden

Der Producer bleibt schlank – er veröffentlicht die Aufgaben und überlässt Skalierung und Wiederholungen der Plattform:

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}")

Kosten: serverlos vs. Always-On-VM

Bei geringem und mittlerem Volumen ist der serverlose Ansatz deutlich günstiger, weil keine Leerlaufkosten anfallen. Ab etwa 10.000 Lösungen pro Tag gleichen sich beide Modelle an. Als Always-On-Alternative im DACH-Raum bietet sich zum Beispiel ein Hetzner-VPS an.

Hinweis: Die Werte sind grobe Schätzungen für die reinen GCP-Compute-Kosten und enthalten nicht den CaptchaAI-Plan. Eigene Messungen mit realem Traffic sind maßgeblich.

Fehlerbehebung

Endpunkt absichern und Datenflüsse prüfen

Deployen Sie öffentliche Funktionen nur bewusst mit --allow-unauthenticated. Für den internen Gebrauch setzen Sie --no-allow-unauthenticated und legen bei Bedarf ein API Gateway mit API-Schlüsseln davor. Wer die gelösten CAPTCHAs im Rahmen von Web-Scraping einsetzt, sollte zusätzlich die eigenen Datenflüsse prüfen: IP-Adressen und personenbezogene Daten unterliegen der DSGVO, unabhängig davon, in welcher GCP-Region die Funktion läuft.

Häufige Fragen

Wie lange darf das Timeout einer Cloud Function sein?

Bei Gen2 bis zu 60 Minuten. Für die CAPTCHA-Lösung reichen meist 120 Sekunden – der Wert muss aber über der internen Poll-Zeitüberschreitung liegen, sonst bricht die Funktion vorzeitig ab.

Wie viele CAPTCHAs kann ich gleichzeitig lösen?

Die Parallelität ergibt sich aus zwei Grenzen: --max-instances in GCP und Ihrer Thread-Zahl bei CaptchaAI. Ein BASIC-Plan (5 Threads) verarbeitet fünf gleichzeitige Lösungen, unabhängig davon, wie viele Funktionsinstanzen laufen.

Wo speichere ich den API-Schlüssel am besten?

Im GCP Secret Manager, abgerufen zur Laufzeit. So steht der Schlüssel weder im Repository noch in Klartext-Umgebungsvariablen und lässt sich ohne erneutes Deployment rotieren.

Lohnt es sich, die Funktion warm zu halten?

Nur bei latenzkritischen Abläufen. Ein Ping über Cloud Scheduler alle 5 Minuten oder --min-instances=1 hält eine Instanz bereit – Letzteres kostet rund 7 $/Monat und hebt den Leerlaufvorteil teilweise auf.

Verwandte Leitfäden

• CAPTCHAs mit AWS Lambda lösen
• Serverlose Integration mit Azure Functions

Serverlos auf GCP – Holen Sie sich Ihren CaptchaAI-Schlüssel und lösen Sie Ihr erstes CAPTCHA in wenigen Minuten.