CAPTCHA Solving in Multi-Tenant SaaS Architectures

Building a SaaS platform that includes CAPTCHA solving requires per-tenant isolation, usage tracking, and cost allocation. Here's how to architect it.

Architecture Overview

┌──────────────────────────────────────┐
│            SaaS Platform             │
│  ┌──────────┐  ┌──────────────────┐  │
│  │ Tenant A │  │  CAPTCHA Service  │  │
│  │ Tenant B │──│  - Rate limiter   │──│── CaptchaAI API
│  │ Tenant C │  │  - Usage tracker  │  │
│  └──────────┘  │  - Cost allocator │  │
│                └──────────────────┘  │
└──────────────────────────────────────┘

Tenant-Isolated Solver

import threading
import time
import requests
from collections import defaultdict


class TenantCaptchaService:
    """CAPTCHA solving service with per-tenant isolation."""

    def __init__(self, api_key):
        self.api_key = api_key
        self.usage = defaultdict(lambda: {"solves": 0, "cost": 0.0, "failures": 0})
        self.limits = {}
        self._lock = threading.Lock()

    def configure_tenant(self, tenant_id, daily_limit=1000, rate_per_minute=10):
        """Set limits for a tenant."""
        self.limits[tenant_id] = {
            "daily_limit": daily_limit,
            "rate_per_minute": rate_per_minute,
            "minute_counter": 0,
            "minute_start": time.time(),
        }

    def solve(self, tenant_id, captcha_type, params):
        """Solve CAPTCHA for a specific tenant."""
        self._check_limits(tenant_id)

        # Solve via CaptchaAI
        data = {"key": self.api_key, "json": 1, **params}
        resp = requests.post(
            "https://ocr.captchaai.com/in.php", data=data, timeout=30,
        )
        result = resp.json()

        if result.get("status") != 1:
            with self._lock:
                self.usage[tenant_id]["failures"] += 1
            raise RuntimeError(result.get("request"))

        task_id = result["request"]

        # Poll
        time.sleep(10)
        for _ in range(24):
            resp = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)
            data = resp.json()

            if data.get("status") == 1:
                self._record_usage(tenant_id, captcha_type)
                return data["request"]
            if data["request"] != "CAPCHA_NOT_READY":
                with self._lock:
                    self.usage[tenant_id]["failures"] += 1
                raise RuntimeError(data["request"])
            time.sleep(5)

        raise TimeoutError("Solve timeout")

    def _check_limits(self, tenant_id):
        """Enforce tenant rate and daily limits."""
        with self._lock:
            limits = self.limits.get(tenant_id)
            if not limits:
                return  # No limits configured

            # Daily limit
            if self.usage[tenant_id]["solves"] >= limits["daily_limit"]:
                raise TenantLimitExceeded(
                    f"Tenant {tenant_id} daily limit reached "
                    f"({limits['daily_limit']} solves)"
                )

            # Rate limit
            now = time.time()
            if now - limits["minute_start"] > 60:
                limits["minute_counter"] = 0
                limits["minute_start"] = now

            if limits["minute_counter"] >= limits["rate_per_minute"]:
                raise TenantRateLimited(
                    f"Tenant {tenant_id} rate limited "
                    f"({limits['rate_per_minute']}/min)"
                )
            limits["minute_counter"] += 1

    def _record_usage(self, tenant_id, captcha_type):
        """Record successful solve for billing."""
        cost_map = {
            "recaptcha_v2": 0.003,
            "recaptcha_v3": 0.004,
            "turnstile": 0.002,
            "image": 0.001,
        }
        cost = cost_map.get(captcha_type, 0.003)

        with self._lock:
            self.usage[tenant_id]["solves"] += 1
            self.usage[tenant_id]["cost"] += cost

    def get_tenant_usage(self, tenant_id):
        """Get usage stats for a tenant."""
        with self._lock:
            return dict(self.usage[tenant_id])

    def get_all_usage(self):
        """Get usage for all tenants."""
        with self._lock:
            return {tid: dict(data) for tid, data in self.usage.items()}


class TenantLimitExceeded(Exception):
    pass

class TenantRateLimited(Exception):
    pass

Usage Tracking for Billing

import csv
import os
from datetime import datetime, timezone


class TenantBillingTracker:
    """Track per-tenant usage for billing purposes."""

    def __init__(self, log_dir="billing_logs"):
        os.makedirs(log_dir, exist_ok=True)
        self.log_dir = log_dir

    def record_solve(self, tenant_id, captcha_type, cost, task_id):
        """Record a billable solve."""
        date = datetime.now(timezone.utc).strftime("%Y-%m")
        filepath = os.path.join(self.log_dir, f"{tenant_id}_{date}.csv")

        write_header = not os.path.exists(filepath)
        with open(filepath, "a", newline="") as f:
            writer = csv.writer(f)
            if write_header:
                writer.writerow(["timestamp", "task_id", "type", "cost_usd"])
            writer.writerow([
                datetime.now(timezone.utc).isoformat(),
                task_id, captcha_type, f"{cost:.6f}",
            ])

    def get_monthly_total(self, tenant_id, year_month=None):
        """Get monthly billing total for a tenant."""
        if not year_month:
            year_month = datetime.now(timezone.utc).strftime("%Y-%m")

        filepath = os.path.join(self.log_dir, f"{tenant_id}_{year_month}.csv")
        if not os.path.exists(filepath):
            return {"solves": 0, "total_cost": 0.0}

        total = 0.0
        count = 0
        with open(filepath, "r") as f:
            reader = csv.DictReader(f)
            for row in reader:
                total += float(row["cost_usd"])
                count += 1

        return {"solves": count, "total_cost": round(total, 4)}

Per-Tenant API Key Pattern

For larger platforms, give each tenant their own CaptchaAI key:

class MultiKeyService:
    """Use per-tenant API keys for full isolation."""

    def __init__(self, default_key=None):
        self.tenant_keys = {}
        self.default_key = default_key

    def register_tenant(self, tenant_id, api_key):
        """Register a tenant's API key."""
        self.tenant_keys[tenant_id] = api_key

    def solve(self, tenant_id, params):
        """Solve using tenant's own key."""
        key = self.tenant_keys.get(tenant_id, self.default_key)
        if not key:
            raise ValueError(f"No API key for tenant {tenant_id}")

        data = {"key": key, "json": 1, **params}
        resp = requests.post(
            "https://ocr.captchaai.com/in.php", data=data, timeout=30,
        )
        return resp.json()

    def get_tenant_balance(self, tenant_id):
        """Check a tenant's balance."""
        key = self.tenant_keys.get(tenant_id)
        if not key:
            return None

        resp = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": key, "action": "getbalance", "json": 1,
        }, timeout=10)
        data = resp.json()
        if data.get("status") == 1:
            return float(data["request"])
        return None

FAQ

Should each tenant have their own CaptchaAI key?

For small platforms, a shared key with usage tracking works. For enterprise platforms, per-tenant keys provide full billing isolation and let tenants manage their own balance.