API Keys Explained: How They Work and Why They Fail

// io.thecodeforge — System Design tutorial
// Tracing a single API call from your app to a third-party service
// Scenario: Your e-commerce checkout calls Stripe to charge a card

// ─────────────────────────────────────────────────────────────
// STEP 1 — Your checkout service builds an HTTP request
// ─────────────────────────────────────────────────────────────

POST https://api.stripe.com/v1/charges

Headers:
  Authorization: Bearer sk_live_4eC39HqLyjWDarjtT1zdp7dc   // <-- the API key
  Content-Type: application/x-www-form-urlencoded

Body:
  amount=2000          // $20.00 in cents
  currency=usd
  source=tok_visa      // tokenised card from Stripe.js

// ─────────────────────────────────────────────────────────────
// STEP 2 — Stripe's server receives the request
// ─────────────────────────────────────────────────────────────

Stripe API Gateway:
  1. Extract key from Authorization header
     key = "sk_live_4eC39HqLyjWDarjtT1zdp7dc"

  2. Look up key in Stripe's internal key store
     SELECT account_id, permissions, rate_limit, is_active
     FROM api_keys
     WHERE key_hash = SHA256("sk_live_4eC39HqLyjWDarjtT1zdp7dc")
     // NOTE: Stripe stores a HASH of your key, not the key itself
     // This means even Stripe can't recover your key if their DB leaks

  3. Key found → account_id = "acct_1A2B3C4D5E6F"
     is_active = true
     permissions = ["charges:write", "refunds:write"]
     rate_limit = 100 requests/second

  4. Check rate limit — current usage: 23/100 req/sec → OK

  5. Process the charge against account acct_1A2B3C4D5E6F

// ─────────────────────────────────────────────────────────────
// STEP 3 — Stripe responds
// ─────────────────────────────────────────────────────────────

HTTP 200 OK
{
  "id": "ch_3MqLiJKZ2eZvKYlo2T9UW2GX",
  "object": "charge",
  "amount": 2000,
  "status": "succeeded"
}

// ─────────────────────────────────────────────────────────────
// WHAT HAPPENS WITH A BAD KEY
// ─────────────────────────────────────────────────────────────

Stripe API Gateway (bad key scenario):
  1. Extract key: "sk_live_INVALIDKEYHERE"
  2. Hash and look up → no matching row in api_keys table
  3. Return immediately — no account check, no charge processing

HTTP 401 Unauthorized
{
  "error": {
    "code": "api_key_invalid",
    "message": "No such API key: 'sk_live_INVA...HERE'"
  }
}

# io.thecodeforge — System Design tutorial
# Scenario: Payment service loading a Stripe key safely at startup
# Demonstrating: env vars (dev), secrets manager (prod), and the logging trap

import os
import boto3
import json
import logging
import requests
from functools import lru_cache

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────
# PATTERN 1 — Environment variable (acceptable for local dev)
# ─────────────────────────────────────────────────────────────

def load_stripe_key_from_env() -> str:
    key = os.environ.get("STRIPE_SECRET_KEY")
    if not key:
        # Fail loud at startup — better than a cryptic 401 at checkout time
        raise EnvironmentError(
            "STRIPE_SECRET_KEY is not set. "
            "Check your .env file or deployment environment variables."
        )
    if key.startswith("sk_live") and os.environ.get("APP_ENV") == "development":
        # Catch the classic mistake: live key used in local dev
        raise EnvironmentError(
            "Live Stripe key detected in development environment. "
            "Use sk_test_ keys for local development."
        )
    return key

# ─────────────────────────────────────────────────────────────
# PATTERN 2 — AWS Secrets Manager (required for production)
# ─────────────────────────────────────────────────────────────

@lru_cache(maxsize=1)  # Cache the secret — don't call Secrets Manager on every request
def load_stripe_key_from_secrets_manager(secret_name: str, region: str) -> str:
    client = boto3.client("secretsmanager", region_name=region)
    try:
        response = client.get_secret_value(SecretId=secret_name)
    except client.exceptions.ResourceNotFoundException:
        raise RuntimeError(f"Secret '{secret_name}' not found in Secrets Manager.")
    except client.exceptions.AccessDeniedException:
        # This usually means your IAM role doesn't have secretsmanager:GetSecretValue
        raise RuntimeError(
            f"IAM permission denied reading '{secret_name}'. "
            "Check your task role policy for secretsmanager:GetSecretValue."
        )
    secret = json.loads(response["SecretString"])
    return secret["stripe_secret_key"]

# ─────────────────────────────────────────────────────────────
# THE LOGGING TRAP — this is how keys end up in Datadog
# ─────────────────────────────────────────────────────────────

def charge_card_unsafe(stripe_key: str, amount_cents: int, card_token: str):
    headers = {"Authorization": f"Bearer {stripe_key}"}
    response = requests.post(
        "https://api.stripe.com/v1/charges",
        headers=headers,
        data={"amount": amount_cents, "currency": "usd", "source": card_token}
    )
    if response.status_code != 200:
        # DANGER: logging response.request exposes the Authorization header
        # If Sentry or Datadog captures this log, your key is now in their system
        logger.error(f"Stripe charge failed. Request: {response.request.headers}")
    return response.json()


def charge_card_safe(stripe_key: str, amount_cents: int, card_token: str):
    headers = {"Authorization": f"Bearer {stripe_key}"}
    response = requests.post(
        "https://api.stripe.com/v1/charges",
        headers=headers,
        data={"amount": amount_cents, "currency": "usd", "source": card_token}
    )
    if response.status_code != 200:
        # Log only what you need to debug — never log headers containing credentials
        logger.error(
            "Stripe charge failed",
            extra={
                "status_code": response.status_code,
                "stripe_error_code": response.json().get("error", {}).get("code"),
                "amount_cents": amount_cents
                # Deliberately omitting: headers, full request object, card_token
            }
        )
    return response.json()


# ─────────────────────────────────────────────────────────────
# STARTUP — how the service wires this together
# ─────────────────────────────────────────────────────────────

if __name__ == "__main__":
    env = os.environ.get("APP_ENV", "development")

    if env == "production":
        stripe_key = load_stripe_key_from_secrets_manager(
            secret_name="prod/payment-service/stripe",
            region="us-east-1"
        )
        print("Loaded Stripe key from Secrets Manager")
    else:
        stripe_key = load_stripe_key_from_env()
        print("Loaded Stripe key from environment variable")

    # Sanity check — log key PREFIX only so you can confirm which key is active
    # Never log the full key, even in debug mode
    print(f"Active Stripe key prefix: {stripe_key[:12]}...")

# io.thecodeforge — System Design tutorial
# Scenario: Internal API gateway managing keys for microservices
# Demonstrates: scoped keys, rotation tracking, and handling 429s correctly

import time
import hashlib
import secrets
import logging
from datetime import datetime, timedelta
from dataclasses import dataclass, field
from typing import Optional
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────
# DATA MODEL — what a managed API key looks like internally
# ─────────────────────────────────────────────────────────────

@dataclass
class ScopedAPIKey:
    service_name: str           # which internal service owns this key
    provider: str               # e.g. "stripe", "sendgrid", "googlemaps"
    permissions: list[str]      # e.g. ["charges:write"] — not ["*"]
    created_at: datetime = field(default_factory=datetime.utcnow)
    rotate_by: datetime = field(default_factory=lambda: datetime.utcnow() + timedelta(days=90))
    _raw_key: str = field(default="", repr=False)  # never printed in logs or repr

    @property
    def key_prefix(self) -> str:
        # Safe to log — enough to identify which key is active without exposing it
        return self._raw_key[:12] + "..."

    @property
    def days_until_rotation(self) -> int:
        return (self.rotate_by - datetime.utcnow()).days

    @property
    def needs_rotation(self) -> bool:
        return self.days_until_rotation <= 7  # warn a week out


# ─────────────────────────────────────────────────────────────
# RETRY LOGIC — handling 429s without hammering the upstream
# ─────────────────────────────────────────────────────────────

def build_resilient_http_session(total_retries: int = 3) -> requests.Session:
    session = requests.Session()

    # Retry on 429 (rate limit) and 503 (upstream temporarily unavailable)
    # backoff_factor=2 means: wait 2s, then 4s, then 8s between retries
    retry_strategy = Retry(
        total=total_retries,
        status_forcelist=[429, 503],
        backoff_factor=2,
        respect_retry_after_header=True  # honour Stripe/SendGrid's Retry-After header
    )
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    return session


# ─────────────────────────────────────────────────────────────
# SCOPED REQUEST BUILDER — enforces least-privilege per service
# ─────────────────────────────────────────────────────────────

class ScopedStripeClient:
    """
    Each internal service gets its own ScopedStripeClient with its own key.
    The checkout service gets charges:write.
    The reporting service gets charges:read only.
    A compromised reporting service cannot create charges.
    """

    def __init__(self, api_key: ScopedAPIKey):
        self._key = api_key
        self._session = build_resilient_http_session()
        self._check_rotation_status()

    def _check_rotation_status(self):
        if self._key.needs_rotation:
            # Warn loudly at startup — gives ops team time to rotate before expiry
            logger.warning(
                "API key rotation due soon",
                extra={
                    "service": self._key.service_name,
                    "provider": self._key.provider,
                    "key_prefix": self._key.key_prefix,
                    "days_remaining": self._key.days_until_rotation
                }
            )

    def get_charge(self, charge_id: str) -> dict:
        # Reporting service uses this — read-only, no ability to create/modify
        if "charges:read" not in self._key.permissions:
            raise PermissionError(
                f"Key for '{self._key.service_name}' lacks charges:read permission. "
                f"Granted permissions: {self._key.permissions}"
            )
        response = self._session.get(
            f"https://api.stripe.com/v1/charges/{charge_id}",
            headers={"Authorization": f"Bearer {self._key._raw_key}"}
        )
        response.raise_for_status()
        return response.json()

    def create_charge(self, amount_cents: int, card_token: str) -> dict:
        # Checkout service uses this — requires explicit write permission
        if "charges:write" not in self._key.permissions:
            raise PermissionError(
                f"Key for '{self._key.service_name}' lacks charges:write permission. "
                f"This is likely a scoping error — do not expand permissions. "
                f"Create a dedicated key with charges:write for the checkout service."
            )
        response = self._session.post(
            "https://api.stripe.com/v1/charges",
            headers={"Authorization": f"Bearer {self._key._raw_key}"},
            data={"amount": amount_cents, "currency": "usd", "source": card_token}
        )
        response.raise_for_status()
        return response.json()


# ─────────────────────────────────────────────────────────────
# EXAMPLE USAGE — wiring up two services with different scopes
# ─────────────────────────────────────────────────────────────

if __name__ == "__main__":
    # Checkout service key — write access
    checkout_api_key = ScopedAPIKey(
        service_name="checkout-service",
        provider="stripe",
        permissions=["charges:write", "refunds:write"],
        rotate_by=datetime.utcnow() + timedelta(days=5)  # triggers rotation warning
    )
    checkout_api_key._raw_key = "sk_live_checkout_key_here"

    # Reporting service key — read access only
    reporting_api_key = ScopedAPIKey(
        service_name="reporting-service",
        provider="stripe",
        permissions=["charges:read"],
        rotate_by=datetime.utcnow() + timedelta(days=60)
    )
    reporting_api_key._raw_key = "sk_live_reporting_key_here"

    checkout_client = ScopedStripeClient(checkout_api_key)
    reporting_client = ScopedStripeClient(reporting_api_key)

    # This works:
    print("Checkout client permissions:", checkout_api_key.permissions)

    # This raises PermissionError — the reporting client cannot create charges
    try:
        reporting_client.create_charge(2000, "tok_visa")
    except PermissionError as e:
        print(f"Caught expected permission error: {e}")

# io.thecodeforge — System Design tutorial
# Scenario: A lightweight health-check system that validates API keys are alive
# and alerts on anomalous usage patterns before the bill arrives

import logging
import time
from datetime import datetime, timedelta
from collections import deque
from typing import Callable
import requests

logger = logging.getLogger(__name__)

# ─────────────────────────────────────────────────────────────
# PATTERN: Sliding window usage tracker
# Detects anomalous request spikes that could indicate a leaked key
# being used by someone else against your quota
# ─────────────────────────────────────────────────────────────

class APIKeyUsageMonitor:
    def __init__(
        self,
        service_name: str,
        rate_limit_per_minute: int,
        spike_alert_threshold: float = 0.75  # alert at 75% of rate limit
    ):
        self.service_name = service_name
        self.rate_limit_per_minute = rate_limit_per_minute
        self.spike_alert_threshold = spike_alert_threshold
        # Deque of timestamps — we keep only the last 60 seconds of requests
        self._request_timestamps: deque = deque()

    def record_request(self):
        now = time.monotonic()
        self._request_timestamps.append(now)
        self._evict_old_timestamps(now)
        self._check_for_spike()

    def _evict_old_timestamps(self, now: float):
        # Remove timestamps older than 60 seconds
        cutoff = now - 60.0
        while self._request_timestamps and self._request_timestamps[0] < cutoff:
            self._request_timestamps.popleft()

    def _check_for_spike(self):
        current_rate = len(self._request_timestamps)
        alert_threshold = int(self.rate_limit_per_minute * self.spike_alert_threshold)
        if current_rate >= alert_threshold:
            logger.warning(
                "API key usage spike detected — possible key leak or runaway client",
                extra={
                    "service": self.service_name,
                    "requests_last_60s": current_rate,
                    "rate_limit": self.rate_limit_per_minute,
                    "threshold": alert_threshold,
                    "pct_of_limit": round(current_rate / self.rate_limit_per_minute * 100, 1)
                }
            )

    def current_usage(self) -> dict:
        now = time.monotonic()
        self._evict_old_timestamps(now)
        return {
            "service": self.service_name,
            "requests_last_60s": len(self._request_timestamps),
            "rate_limit_per_minute": self.rate_limit_per_minute,
            "headroom_remaining": self.rate_limit_per_minute - len(self._request_timestamps)
        }


# ─────────────────────────────────────────────────────────────
# PATTERN: Active key health check
# Run this from your /healthz endpoint — does NOT use lru_cache
# so it always validates the current key, even after rotation
# ─────────────────────────────────────────────────────────────

def validate_stripe_key_is_active(stripe_key: str) -> dict:
    """
    Stripe's /v1/account endpoint requires a valid key and returns
    account metadata. It's the canonical 'is this key alive?' check.
    Costs one API call. Cache the RESULT for 60 seconds max, never the key.
    """
    try:
        response = requests.get(
            "https://api.stripe.com/v1/account",
            headers={"Authorization": f"Bearer {stripe_key}"},
            timeout=5  # never let a health check block indefinitely
        )
        if response.status_code == 200:
            account_data = response.json()
            return {
                "status": "healthy",
                "account_id": account_data.get("id"),
                "charges_enabled": account_data.get("charges_enabled"),
                "checked_at": datetime.utcnow().isoformat()
            }
        elif response.status_code == 401:
            # The key is dead — either revoked, rotated, or never valid
            return {
                "status": "invalid_key",
                "error": response.json().get("error", {}).get("message"),
                "action_required": "Rotate key immediately and update secrets manager"
            }
        else:
            return {
                "status": "unexpected_response",
                "http_status": response.status_code
            }
    except requests.Timeout:
        return {"status": "timeout", "note": "Stripe API did not respond within 5s"}
    except requests.ConnectionError:
        return {"status": "network_error", "note": "Cannot reach api.stripe.com"}


# ─────────────────────────────────────────────────────────────
# DEMO — simulating usage tracking and health check
# ─────────────────────────────────────────────────────────────

if __name__ == "__main__":
    monitor = APIKeyUsageMonitor(
        service_name="checkout-service",
        rate_limit_per_minute=100,
        spike_alert_threshold=0.75
    )

    # Simulate normal traffic (30 requests)
    for _ in range(30):
        monitor.record_request()
    print("After 30 requests:", monitor.current_usage())

    # Simulate spike (76 more requests — crosses 75% threshold)
    for _ in range(46):
        monitor.record_request()
    print("After 76 requests:", monitor.current_usage())

    # Health check output (mocked — would hit real Stripe in prod)
    print("\nKey health check result:")
    print({
        "status": "healthy",
        "account_id": "acct_1A2B3C4D5E6F",
        "charges_enabled": True,
        "checked_at": datetime.utcnow().isoformat()
    })