CAPTCHA-Lösung für Audit-Protokolle: Verfolgung und Lösung von Compliance-Anfragen

Wenn im nächsten Audit die Frage kommt „Wer hat diese CAPTCHA-Lösung ausgelöst, für welche Seite und zu welchen Kosten?", entscheidet ein strukturiertes Audit-Log darüber, ob Sie in Sekunden antworten oder tagelang Log-Dateien durchsuchen. Die Antwort ist ein separates, fortlaufendes Protokoll, das jede einzelne Lösungsanfrage als eigene JSON-Zeile festhält. Dieser Leitfaden zeigt Schritt für Schritt, wie Sie eine solche Audit-Protokollierung für CaptchaAI-Operationen in Python und Node.js aufbauen – vom Datenmodell über die Speicherung bis zur nachvollziehbaren Auswertung.

Welche Felder ein Audit-Eintrag enthalten sollte

Jede Lösung sollte genug Metadaten hinterlassen, um den Vorgang später zweifelsfrei zu rekonstruieren – aber keine sensiblen Werte. Diese Felder haben sich in der Praxis bewährt:

Die request_id verknüpft dabei jeden Eintrag eindeutig mit einem Vorgang – so lassen sich Übermittlung, Ergebnis und etwaige Fehler auch nachträglich lückenlos einer einzigen Lösung zuordnen.

Niemals protokollieren: API-Schlüssel, CAPTCHA-Tokens (sie sind ohnehin nach 60–300 Sekunden abgelaufen) und personenbezogene Daten von Zielseiten. Gerade target_site-URLs und mitgeloggte IP-Adressen können unter die DSGVO fallen: Protokollieren Sie nur, was Sie wirklich benötigen, und pseudonymisieren Sie im Zweifel.

Audit-Logging in Python umsetzen

Die Kernidee: ein eigener Logger, der ausschließlich Audit-Einträge in eine rotierende JSONL-Datei schreibt – getrennt von den normalen Anwendungs-Logs. So bleibt der Prüfpfad intakt, auch wenn Debug-Ausgaben längst rotiert wurden. Die Funktion solve_with_audit protokolliert jeden Zustand: Übermittlung, Erfolg, Fehler und Timeout.

# audit_solver.py
import os
import uuid
import time
import json
import logging
from datetime import datetime, timezone
import requests

API_KEY = os.environ.get("CAPTCHAAI_KEY", "YOUR_API_KEY")

# Configure audit logger — separate from application logs
audit_logger = logging.getLogger("captcha_audit")
audit_logger.setLevel(logging.INFO)

# File handler with rotation
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler(
    "captcha_audit.jsonl",
    maxBytes=50_000_000,  # 50 MB per file
    backupCount=10,
)
handler.setFormatter(logging.Formatter("%(message)s"))
audit_logger.addHandler(handler)

def log_audit(record):
    """Write a structured audit record."""
    audit_logger.info(json.dumps(record, default=str))

def solve_with_audit(sitekey, pageurl, captcha_type="userrecaptcha",
                      initiator="unknown"):
    """Solve a CAPTCHA with full audit logging."""
    request_id = str(uuid.uuid4())
    start = time.time()

    audit_record = {
        "request_id": request_id,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "captcha_type": captcha_type,
        "target_site": pageurl,
        "initiator": initiator,
        "status": "submitted",
    }

    session = requests.Session()

    try:
        # Submit
        resp = session.get("https://ocr.captchaai.com/in.php", params={
            "key": API_KEY,
            "method": captcha_type,
            "googlekey": sitekey,
            "pageurl": pageurl,
            "json": "1",
        })
        result = resp.json()

        if result.get("status") != 1:
            audit_record.update({
                "status": "submit_failed",
                "error_code": result.get("request"),
                "solve_time_ms": int((time.time() - start) * 1000),
            })
            log_audit(audit_record)
            return None

        task_id = result["request"]
        audit_record["task_id"] = task_id

        # Poll
        time.sleep(15)
        for _ in range(25):
            poll = session.get("https://ocr.captchaai.com/res.php", params={
                "key": API_KEY, "action": "get",
                "id": task_id, "json": "1",
            })
            poll_result = poll.json()

            if poll_result.get("status") == 1:
                solve_time = int((time.time() - start) * 1000)
                audit_record.update({
                    "status": "solved",
                    "solve_time_ms": solve_time,
                    "cost_estimate": 0.003,  # Adjust per your rate
                })
                log_audit(audit_record)
                return poll_result["request"]

            if poll_result.get("request") != "CAPCHA_NOT_READY":
                audit_record.update({
                    "status": "failed",
                    "error_code": poll_result.get("request"),
                    "solve_time_ms": int((time.time() - start) * 1000),
                })
                log_audit(audit_record)
                return None

            time.sleep(5)

        audit_record.update({
            "status": "timeout",
            "solve_time_ms": int((time.time() - start) * 1000),
        })
        log_audit(audit_record)
        return None

    except Exception as e:
        audit_record.update({
            "status": "error",
            "error_code": str(e)[:200],
            "solve_time_ms": int((time.time() - start) * 1000),
        })
        log_audit(audit_record)
        raise

# Usage
token = solve_with_audit(
    sitekey="6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-",
    pageurl="https://www.google.com/recaptcha/api2/demo",
    initiator="price-scraper-v2",
)

So sieht ein einzelner Audit-Eintrag aus

Jede Zeile ist gültiges JSON und damit direkt maschinenlesbar – ideal für jq, Log-Aggregatoren oder einen späteren Datenbank-Import:

{"request_id":"a1b2c3d4-...","timestamp":"2026-04-04T14:30:00+00:00","captcha_type":"userrecaptcha","target_site":"https://www.google.com/recaptcha/api2/demo","initiator":"price-scraper-v2","status":"solved","task_id":"73829451","solve_time_ms":18432,"cost_estimate":0.003}

Dieselbe Logik in Node.js

Läuft Ihre Automatisierung auf Node.js, bildet dieselbe Struktur den Audit-Eintrag ab. fs.appendFileSync genügt für moderate Volumina; bei hoher Last sollten Sie auf gepufferte oder asynchrone Schreibvorgänge umstellen.

// audit_solver.js
const fs = require('fs');
const { v4: uuidv4 } = require('uuid');
const axios = require('axios');

const API_KEY = process.env.CAPTCHAAI_KEY || 'YOUR_API_KEY';
const AUDIT_FILE = 'captcha_audit.jsonl';

function logAudit(record) {
  fs.appendFileSync(AUDIT_FILE, JSON.stringify(record) + '\n');
}

async function solveWithAudit(sitekey, pageurl, initiator = 'unknown') {
  const requestId = uuidv4();
  const start = Date.now();
  const record = {
    request_id: requestId,
    timestamp: new Date().toISOString(),
    captcha_type: 'userrecaptcha',
    target_site: pageurl,
    initiator,
    status: 'submitted',
  };

  try {
    const submit = await axios.get('https://ocr.captchaai.com/in.php', {
      params: {
        key: API_KEY, method: 'userrecaptcha',
        googlekey: sitekey, pageurl, json: '1',
      },
    });

    if (submit.data.status !== 1) {
      record.status = 'submit_failed';
      record.error_code = submit.data.request;
      record.solve_time_ms = Date.now() - start;
      logAudit(record);
      return null;
    }

    record.task_id = submit.data.request;
    await new Promise(r => setTimeout(r, 15000));

    for (let i = 0; i < 25; i++) {
      const poll = await axios.get('https://ocr.captchaai.com/res.php', {
        params: { key: API_KEY, action: 'get', id: submit.data.request, json: '1' },
      });

      if (poll.data.status === 1) {
        record.status = 'solved';
        record.solve_time_ms = Date.now() - start;
        record.cost_estimate = 0.003;
        logAudit(record);
        return poll.data.request;
      }
      if (poll.data.request !== 'CAPCHA_NOT_READY') {
        record.status = 'failed';
        record.error_code = poll.data.request;
        record.solve_time_ms = Date.now() - start;
        logAudit(record);
        return null;
      }
      await new Promise(r => setTimeout(r, 5000));
    }

    record.status = 'timeout';
    record.solve_time_ms = Date.now() - start;
    logAudit(record);
    return null;
  } catch (e) {
    record.status = 'error';
    record.error_code = e.message.slice(0, 200);
    record.solve_time_ms = Date.now() - start;
    logAudit(record);
    throw e;
  }
}

Audit-Logs auswerten

Weil jede Zeile eigenständiges JSON ist, brauchen Sie für Auswertungen keine Datenbank. Eine Tagesübersicht – Anzahl der Anfragen, Statusverteilung, geschätzte Kosten und mediane Lösungszeit – lässt sich in wenigen Zeilen erzeugen:

import json
from collections import Counter
from datetime import date

def daily_summary(log_file, target_date=None):
    """Generate a daily summary from audit logs."""
    target = target_date or date.today().isoformat()
    statuses = Counter()
    total_cost = 0
    solve_times = []

    with open(log_file) as f:
        for line in f:
            record = json.loads(line)
            if record["timestamp"].startswith(target):
                statuses[record["status"]] += 1
                total_cost += record.get("cost_estimate", 0)
                if record.get("solve_time_ms"):
                    solve_times.append(record["solve_time_ms"])

    print(f"Date: {target}")
    print(f"Total requests: {sum(statuses.values())}")
    print(f"Statuses: {dict(statuses)}")
    print(f"Estimated cost: ${total_cost:.2f}")
    if solve_times:
        print(f"Median solve time: {sorted(solve_times)[len(solve_times)//2]}ms")

daily_summary("captcha_audit.jsonl")

Für Ad-hoc-Abfragen auf der Kommandozeile genügt oft jq: So filtern Sie etwa alle fehlgeschlagenen Lösungen eines Tages heraus oder summieren die Kosten pro initiator, ohne ein einziges Skript zu schreiben. Die aufsummierten cost_estimate-Werte eignen sich zudem, um Ihre Audit-Zahlen regelmäßig mit dem CaptchaAI-Dashboard abzugleichen und Abweichungen in der Abrechnung früh zu erkennen.

Aufbewahrung, Speicherung und Skalierung

Wie Sie Audit-Logs ablegen, hängt allein vom Volumen ab. Als Faustregel:

In DACH-Setups landen solche Logs häufig in einer GitLab-CI-Pipeline oder auf einem Hetzner-Worker mit anschließendem Versand an einen zentralen Stack. Entscheidend ist nur, dass die Rotation (RotatingFileHandler, logrotate) greift, bevor eine Datei die Platte füllt.

Typische Probleme und ihre Lösung

Häufige Fragen

Warum eine separate Audit-Datei statt der normalen Anwendungs-Logs?

Weil ein Prüfpfad andere Anforderungen hat als Debug-Ausgaben. Audit-Einträge müssen vollständig, unverändert und langfristig auffindbar sein. Ein eigener Logger mit eigener Rotation stellt sicher, dass sie nicht zwischen Millionen Debug-Zeilen verloren gehen oder vorzeitig überschrieben werden.

Zählen die Audit-Daten unter die DSGVO?

Möglicherweise ja. Felder wie target_site oder eine mitgeloggte IP-Adresse können personenbezogen sein. Prüfen Sie Rechtsgrundlage und Aufbewahrungsfrist, protokollieren Sie nur das Nötige und pseudonymisieren Sie im Zweifel. CaptchaAI trifft dazu keine Zusagen – die Datenhoheit über Ihre Logs liegt bei Ihnen.

Wie werte ich die JSONL-Logs ohne Datenbank aus?

Für die meisten Fälle reichen Bordmittel: jq filtert und aggregiert einzelne Felder direkt auf der Kommandozeile, das Python-Skript oben liefert fertige Tagesstatistiken. Erst ab etwa 10.000 Lösungen/Tag lohnt sich der Weg in einen Aggregator wie ELK oder Datadog.

Wie lange sollte ich Audit-Logs aufbewahren?

90 Tage sind für operative Audit-Logs üblich. Unterliegt Ihr Einsatz regulatorischen Vorgaben (SOC 2, DSGVO, HIPAA), richten Sie die Frist nach der jeweils längsten geforderten Aufbewahrungsdauer.

Skaliert dateibasiertes Logging bei 100.000 Lösungen pro Tag?

Bedingt. Bis in den mittleren fünfstelligen Bereich genügt eine rotierende Datei; darüber werden synchrone Schreibvorgänge zum Flaschenhals. Puffern Sie Einträge dann im Speicher und schreiben Sie asynchron in Batches oder direkt an einen zentralen Log-Endpunkt.

Verwandte Leitfäden

• CaptchaAI Schnellstart
• API-Antwortformate und Fehlercodes
• reCAPTCHA v2 per API lösen