"""Shared privacy-bounded licensing API abuse controls. The database is the enforcement authority so limits remain effective across multiple API workers and replicas. TLS ingress still owns the earlier network flood/WAF boundary; these controls add authenticated subject and signed-device semantics that a generic reverse proxy cannot safely infer from JSON bodies. """ from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone import hashlib import hmac from ipaddress import ip_address, ip_network import json import math import threading from types import MappingProxyType from typing import Callable, Mapping, MutableMapping, Optional, Sequence from sqlalchemy import case, delete from sqlalchemy.dialects.postgresql import insert as postgresql_insert from sqlalchemy.dialects.sqlite import insert as sqlite_insert from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session, sessionmaker from licensing_shared.serials import SERIAL_LOOKUP_CHARACTERS, SERIAL_PRODUCT_PREFIX from licensing_shared.constants import validate_identifier from .constants import ( FINGERPRINT_PEPPER_MINIMUM_BYTES, RATE_LIMIT_CLEANUP_REQUEST_INTERVAL, RATE_LIMIT_MAXIMUM_PROXY_CHAIN_LENGTH, RATE_LIMIT_MAXIMUM_REQUESTS_PER_WINDOW, RATE_LIMIT_MAXIMUM_SUBJECT_CHARACTERS, RATE_LIMIT_MAXIMUM_TRUSTED_PROXY_NETWORKS, RATE_LIMIT_MAXIMUM_WINDOW_SECONDS, RATE_LIMIT_SERIAL_INPUT_CHARACTERS, ) from .errors import ServerErrorCode, ServerLicensingError from .middleware import BOUNDED_REQUEST_BODY_STATE_KEY from .models import RateLimitBucket RATE_LIMIT_CONTEXT_STATE_KEY = "licensing_rate_limit_context" RATE_LIMIT_DIGEST_DOMAIN = b"APOLON-RATE-LIMIT-V1\x00" RATE_LIMIT_KEY_KINDS = frozenset(("device", "principal", "serial", "source")) RATE_LIMIT_EXEMPT_ACTION = "exempt" RATE_LIMIT_SOURCE_HEADER = b"x-forwarded-for" RATE_LIMIT_JSON_CONTENT_TYPE = b"application/json" RATE_LIMIT_MAXIMUM_HEADER_BYTES = 8192 @dataclass(frozen=True) class RateLimitQuota: name: str key_kind: str maximum_requests: int window_seconds: int def __post_init__(self) -> None: for field_name in ("name", "key_kind"): value = getattr(self, field_name) if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty string") object.__setattr__(self, field_name, value.strip()) object.__setattr__(self, "name", validate_identifier(self.name, "quota name")) if self.key_kind not in RATE_LIMIT_KEY_KINDS: raise ValueError("key_kind is not supported") for field_name in ("maximum_requests", "window_seconds"): value = getattr(self, field_name) if isinstance(value, bool) or not isinstance(value, int): raise TypeError(f"{field_name} must be an integer") if value < 1: raise ValueError(f"{field_name} must be at least one") if self.maximum_requests > RATE_LIMIT_MAXIMUM_REQUESTS_PER_WINDOW: raise ValueError("maximum_requests is too large") if self.window_seconds > RATE_LIMIT_MAXIMUM_WINDOW_SECONDS: raise ValueError("window_seconds is too large") @dataclass(frozen=True) class RequestRateLimitContext: action: str source_address: str device_key_thumbprint: Optional[str] = None serial_lookup_prefix: Optional[str] = None def __post_init__(self) -> None: if not isinstance(self.action, str) or not self.action.strip(): raise ValueError("action must be a non-empty string") object.__setattr__( self, "action", validate_identifier(self.action.strip(), "rate-limit action"), ) if not isinstance(self.source_address, str): raise TypeError("source_address must be a string") try: normalized_source = str(ip_address(self.source_address.strip())) except ValueError as exc: raise ValueError("source_address must be an IP address") from exc object.__setattr__(self, "source_address", normalized_source) for field_name in ("device_key_thumbprint", "serial_lookup_prefix"): value = getattr(self, field_name) if value is None: continue if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty string or None") normalized = value.strip() if len(normalized) > RATE_LIMIT_MAXIMUM_SUBJECT_CHARACTERS: raise ValueError(f"{field_name} is too long") object.__setattr__(self, field_name, normalized) def _quota( name: str, key_kind: str, maximum_requests: int, window_seconds: int, ) -> RateLimitQuota: return RateLimitQuota(name, key_kind, maximum_requests, window_seconds) DEFAULT_REQUEST_QUOTAS: Mapping[str, tuple[RateLimitQuota, ...]] = MappingProxyType( { "public.read": (_quota("public.source.minute", "source", 240, 60),), "serial.redeem": ( _quota("serial.source.minute", "source", 20, 60), _quota("serial.lookup.five_minutes", "serial", 8, 300), _quota("serial.device.minute", "device", 20, 60), ), "trial.start": ( _quota("trial.source.hour", "source", 30, 3600), _quota("trial.device.day", "device", 12, 86400), ), "trial.status": (_quota("trial.status.source.minute", "source", 60, 60),), "activation.refresh": ( _quota("activation.refresh.source.minute", "source", 120, 60), _quota("activation.refresh.device.hour", "device", 120, 3600), ), "activation.mutate": ( _quota("activation.mutate.source.minute", "source", 30, 60), _quota("activation.mutate.device.hour", "device", 30, 3600), ), "account.read": (_quota("account.read.source.minute", "source", 240, 60),), "account.mutate": ( _quota("account.mutate.source.minute", "source", 120, 60), _quota("account.mutate.device.hour", "device", 120, 3600), ), "admin.read": (_quota("admin.read.source.minute", "source", 120, 60),), "admin.mutate": (_quota("admin.mutate.source.minute", "source", 60, 60),), "stripe.webhook": (_quota("stripe.webhook.source.minute", "source", 600, 60),), "api.other": (_quota("api.other.source.minute", "source", 180, 60),), "offline.portal": (_quota("offline.portal.source.minute", "source", 60, 60),), } ) DEFAULT_PRINCIPAL_QUOTAS: Mapping[str, tuple[RateLimitQuota, ...]] = MappingProxyType( { "serial.redeem": (_quota("serial.principal.minute", "principal", 30, 60),), "trial.start": (_quota("trial.principal.hour", "principal", 20, 3600),), "account.read": (_quota("account.read.principal.minute", "principal", 240, 60),), "account.mutate": ( _quota("account.mutate.principal.minute", "principal", 120, 60), ), "admin.read": (_quota("admin.read.principal.minute", "principal", 120, 60),), "admin.mutate": (_quota("admin.mutate.principal.minute", "principal", 60, 60),), } ) def _normalize_quota_mapping( value: Mapping[str, Sequence[RateLimitQuota]], field_name: str, ) -> Mapping[str, tuple[RateLimitQuota, ...]]: if not isinstance(value, Mapping): raise TypeError(f"{field_name} must be a mapping") normalized: dict[str, tuple[RateLimitQuota, ...]] = {} names: set[str] = set() for action, quotas in value.items(): if not isinstance(action, str) or not action.strip(): raise ValueError(f"{field_name} action names must be non-empty strings") if isinstance(quotas, (str, bytes)) or not isinstance(quotas, Sequence): raise TypeError(f"{field_name} quota collections must be sequences") normalized_quotas = tuple(quotas) if any(not isinstance(quota, RateLimitQuota) for quota in normalized_quotas): raise TypeError(f"{field_name} values must contain RateLimitQuota objects") for quota in normalized_quotas: if quota.name in names: raise ValueError("rate-limit quota names must be globally unique") names.add(quota.name) normalized_action = validate_identifier(action.strip(), "rate-limit action") normalized[normalized_action] = normalized_quotas return MappingProxyType(normalized) def classify_rate_limit_action(method: str, path: str) -> str: if not isinstance(method, str) or not method.strip(): raise ValueError("method must be a non-empty string") if not isinstance(path, str) or not path.startswith("/"): raise ValueError("path must be an absolute request path") normalized_method = method.strip().upper() if path in ("/v1/health", "/v1/readiness"): return RATE_LIMIT_EXEMPT_ACTION if path == "/offline-activation": return "offline.portal" if path in ("/v1/catalog/public", "/v1/auth/native/config"): return "public.read" if path in ( "/v1/serials/redeem", "/v1/offline/requests", "/v1/me/licenses/claim-serial", ): return "serial.redeem" if path in ("/v1/trials/start", "/v1/offline/trial-requests"): return "trial.start" if path == "/v1/offline/trial-requests/status": return "trial.status" if path == "/v1/activations/refresh": return "activation.refresh" if path in ("/v1/activations/deactivate", "/v1/activations/release-file"): return "activation.mutate" if path == "/v1/webhooks/stripe": return "stripe.webhook" if path.startswith("/v1/admin/"): return "admin.read" if normalized_method == "GET" else "admin.mutate" if path.startswith("/v1/me/"): return "account.read" if normalized_method == "GET" else "account.mutate" if path.startswith("/v1/"): return "api.other" return RATE_LIMIT_EXEMPT_ACTION def _header_values(scope: Mapping[str, object], name: bytes) -> tuple[bytes, ...]: raw_headers = scope.get("headers", ()) if not isinstance(raw_headers, (list, tuple)): raise TypeError("ASGI headers must be a list or tuple") values: list[bytes] = [] for item in raw_headers: if not isinstance(item, (list, tuple)) or len(item) != 2: raise TypeError("ASGI header entries must be name/value pairs") header_name, header_value = item if not isinstance(header_name, bytes) or not isinstance(header_value, bytes): raise TypeError("ASGI header names and values must be bytes") if len(header_name) + len(header_value) > RATE_LIMIT_MAXIMUM_HEADER_BYTES: continue if header_name.lower() == name: values.append(header_value) return tuple(values) def _source_address( scope: Mapping[str, object], trusted_proxy_networks: Sequence[str], ) -> str: client = scope.get("client") if not isinstance(client, (list, tuple)) or len(client) != 2: raise ValueError("ASGI client address is missing") raw_peer = client[0] if not isinstance(raw_peer, str): raise TypeError("ASGI client host must be a string") peer = ip_address(raw_peer.strip()) networks = tuple(ip_network(value, strict=False) for value in trusted_proxy_networks) if not any(peer in network for network in networks): return str(peer) values = _header_values(scope, RATE_LIMIT_SOURCE_HEADER) if len(values) != 1: return str(peer) try: chain_text = values[0].decode("ascii") except UnicodeError: return str(peer) chain_values = tuple(part.strip() for part in chain_text.split(",") if part.strip()) if not chain_values: return str(peer) candidates = chain_values[-RATE_LIMIT_MAXIMUM_PROXY_CHAIN_LENGTH:] for raw_candidate in reversed(candidates): try: candidate = ip_address(raw_candidate) except ValueError: continue if any(candidate in network for network in networks): continue return str(candidate) try: return str(ip_address(candidates[0])) except ValueError: return str(peer) def _json_body(scope: Mapping[str, object]) -> Mapping[str, object]: state = scope.get("state", {}) if not isinstance(state, Mapping): raise TypeError("ASGI state must be a mapping") body = state.get(BOUNDED_REQUEST_BODY_STATE_KEY, b"") if not isinstance(body, bytes): raise TypeError("bounded ASGI request body must be bytes") if not body: return {} content_types = _header_values(scope, b"content-type") if not content_types or not content_types[0].lower().startswith( RATE_LIMIT_JSON_CONTENT_TYPE ): return {} try: parsed = json.loads(body) except (UnicodeError, json.JSONDecodeError, RecursionError): return {} return parsed if isinstance(parsed, dict) else {} def _nested_text(document: Mapping[str, object], *paths: tuple[str, ...]) -> Optional[str]: for path in paths: current: object = document for component in path: if not isinstance(current, Mapping): current = None break current = current.get(component) if isinstance(current, str) and current.strip(): normalized = current.strip() if len(normalized) <= RATE_LIMIT_MAXIMUM_SUBJECT_CHARACTERS: return normalized return None def _serial_lookup_prefix(document: Mapping[str, object]) -> Optional[str]: serial = document.get("serial") if not isinstance(serial, str) or not serial.strip(): return None if len(serial) > RATE_LIMIT_SERIAL_INPUT_CHARACTERS: return None try: compact = "".join( character for character in serial.upper() if not character.isspace() and character != "-" ) except UnicodeError: return None prefix_characters = len(SERIAL_PRODUCT_PREFIX) + SERIAL_LOOKUP_CHARACTERS return compact[:prefix_characters] or None def extract_rate_limit_context( scope: Mapping[str, object], trusted_proxy_networks: Sequence[str], ) -> RequestRateLimitContext: if not isinstance(scope, Mapping): raise TypeError("scope must be a mapping") if isinstance(trusted_proxy_networks, (str, bytes)) or not isinstance( trusted_proxy_networks, Sequence, ): raise TypeError("trusted_proxy_networks must be a sequence") if len(trusted_proxy_networks) > RATE_LIMIT_MAXIMUM_TRUSTED_PROXY_NETWORKS: raise ValueError("trusted_proxy_networks contains too many networks") method = scope.get("method") path = scope.get("path") if not isinstance(method, str) or not isinstance(path, str): raise ValueError("HTTP scope must provide method and path") body = _json_body(scope) device_key_thumbprint = _nested_text( body, ("device", "deviceKeyThumbprint"), ("deviceProof", "deviceKeyThumbprint"), ("offlineRequest", "payload", "deviceKeyThumbprint"), ) return RequestRateLimitContext( action=classify_rate_limit_action(method, path), source_address=_source_address(scope, trusted_proxy_networks), device_key_thumbprint=device_key_thumbprint, serial_lookup_prefix=_serial_lookup_prefix(body), ) class RequestRateLimitContextMiddleware: def __init__( self, app: Callable, trusted_proxy_networks: Sequence[str], ) -> None: if not callable(app): raise TypeError("app must be callable") if isinstance(trusted_proxy_networks, (str, bytes)) or not isinstance( trusted_proxy_networks, Sequence, ): raise TypeError("trusted_proxy_networks must be a sequence") if len(trusted_proxy_networks) > RATE_LIMIT_MAXIMUM_TRUSTED_PROXY_NETWORKS: raise ValueError("trusted_proxy_networks contains too many networks") normalized_networks = tuple( str(ip_network(value, strict=False)) for value in trusted_proxy_networks ) self._app = app self._trusted_proxy_networks = normalized_networks async def __call__( self, scope: Mapping[str, object], receive: Callable, send: Callable, ) -> None: if not isinstance(scope, Mapping): raise TypeError("scope must be a mapping") if not callable(receive): raise TypeError("receive must be callable") if not callable(send): raise TypeError("send must be callable") if scope.get("type") != "http": await self._app(scope, receive, send) return context = extract_rate_limit_context(scope, self._trusted_proxy_networks) child_scope = dict(scope) raw_state = scope.get("state", {}) if not isinstance(raw_state, Mapping): raise TypeError("ASGI state must be a mapping") state: MutableMapping[str, object] = dict(raw_state) state.pop(BOUNDED_REQUEST_BODY_STATE_KEY, None) state[RATE_LIMIT_CONTEXT_STATE_KEY] = context child_scope["state"] = state await self._app(child_scope, receive, send) class DatabaseRateLimiter: def __init__( self, session_factory: sessionmaker[Session], digest_pepper: bytes, request_quotas: Mapping[str, Sequence[RateLimitQuota]] = DEFAULT_REQUEST_QUOTAS, principal_quotas: Mapping[str, Sequence[RateLimitQuota]] = DEFAULT_PRINCIPAL_QUOTAS, now_provider: Callable[[], datetime] = lambda: datetime.now(timezone.utc), ) -> None: if not isinstance(session_factory, sessionmaker): raise TypeError("session_factory must be a SQLAlchemy sessionmaker") if not isinstance(digest_pepper, bytes): raise TypeError("digest_pepper must be bytes") if len(digest_pepper) < FINGERPRINT_PEPPER_MINIMUM_BYTES: raise ValueError("digest_pepper is too short") if not callable(now_provider): raise TypeError("now_provider must be callable") self._session_factory = session_factory self._digest_pepper = digest_pepper self._request_quotas = _normalize_quota_mapping( request_quotas, "request_quotas", ) self._principal_quotas = _normalize_quota_mapping( principal_quotas, "principal_quotas", ) self._now_provider = now_provider self._cleanup_lock = threading.Lock() self._requests_until_cleanup = 0 def enforce_request(self, context: RequestRateLimitContext) -> None: if not isinstance(context, RequestRateLimitContext): raise TypeError("context must be a RequestRateLimitContext") values = { "source": context.source_address, "serial": context.serial_lookup_prefix, } self._consume(self._request_quotas.get(context.action, ()), values) def enforce_verified_device( self, context: RequestRateLimitContext, device_key_thumbprint: str, ) -> None: if not isinstance(context, RequestRateLimitContext): raise TypeError("context must be a RequestRateLimitContext") if ( not isinstance(device_key_thumbprint, str) or not device_key_thumbprint.strip() ): raise ValueError("device_key_thumbprint must be a non-empty string") normalized_thumbprint = device_key_thumbprint.strip() if len(normalized_thumbprint) > RATE_LIMIT_MAXIMUM_SUBJECT_CHARACTERS: raise ValueError("device_key_thumbprint is too long") device_quotas = tuple( quota for quota in self._request_quotas.get(context.action, ()) if quota.key_kind == "device" ) self._consume(device_quotas, {"device": normalized_thumbprint}) def enforce_principal( self, context: RequestRateLimitContext, issuer: str, subject: str, ) -> None: if not isinstance(context, RequestRateLimitContext): raise TypeError("context must be a RequestRateLimitContext") for field_name, value in (("issuer", issuer), ("subject", subject)): if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty string") if len(value.strip()) > RATE_LIMIT_MAXIMUM_SUBJECT_CHARACTERS: raise ValueError(f"{field_name} is too long") principal = f"{issuer.strip()}\x00{subject.strip()}" self._consume( self._principal_quotas.get(context.action, ()), {"principal": principal}, ) def _consume( self, quotas: Sequence[RateLimitQuota], values: Mapping[str, Optional[str]], ) -> None: applicable = tuple( (quota, values.get(quota.key_kind)) for quota in quotas if values.get(quota.key_kind) is not None ) if not applicable: return now = self._normalized_now() breaches: list[tuple[RateLimitQuota, datetime]] = [] try: with self._session_factory() as session, session.begin(): if self._should_cleanup(): session.execute( delete(RateLimitBucket).where(RateLimitBucket.expires_at <= now) ) for quota, raw_value in applicable: assert raw_value is not None count, expires_at = self._increment( session, quota, raw_value, now, ) if count > quota.maximum_requests: breaches.append((quota, expires_at)) except SQLAlchemyError as exc: raise ServerLicensingError( ServerErrorCode.RATE_LIMIT_UNAVAILABLE, "licensing request protection is temporarily unavailable", status_code=503, retryable=True, ) from exc if breaches: retry_after = max( 1, max( math.ceil((expires_at - now).total_seconds()) for _quota_value, expires_at in breaches ), ) raise ServerLicensingError( ServerErrorCode.RATE_LIMITED, "too many licensing requests; please wait and try again", status_code=429, retryable=True, retry_after_seconds=retry_after, ) def _increment( self, session: Session, quota: RateLimitQuota, raw_value: str, now: datetime, ) -> tuple[int, datetime]: if not isinstance(session, Session): raise TypeError("session must be a Session") if not isinstance(quota, RateLimitQuota): raise TypeError("quota must be a RateLimitQuota") if not isinstance(raw_value, str) or not raw_value: raise ValueError("raw_value must be a non-empty string") if not isinstance(now, datetime): raise TypeError("now must be a datetime") binding = session.get_bind() if binding.dialect.name == "postgresql": insert = postgresql_insert elif binding.dialect.name == "sqlite": insert = sqlite_insert else: raise ServerLicensingError( ServerErrorCode.RATE_LIMIT_UNAVAILABLE, "licensing request protection requires PostgreSQL or SQLite", status_code=503, retryable=True, ) expires_at = now + timedelta(seconds=quota.window_seconds) bucket_id = self._bucket_id(quota, raw_value) reset_window = RateLimitBucket.expires_at <= now statement = ( insert(RateLimitBucket) .values( id=bucket_id, policy=quota.name, request_count=1, window_started_at=now, expires_at=expires_at, ) .on_conflict_do_update( index_elements=[RateLimitBucket.id], set_={ "policy": quota.name, "request_count": case( (reset_window, 1), else_=RateLimitBucket.request_count + 1, ), "window_started_at": case( (reset_window, now), else_=RateLimitBucket.window_started_at, ), "expires_at": case( (reset_window, expires_at), else_=RateLimitBucket.expires_at, ), "updated_at": now, }, ) .returning(RateLimitBucket.request_count, RateLimitBucket.expires_at) ) row = session.execute(statement).one() returned_expiry = row.expires_at if returned_expiry.tzinfo is None: returned_expiry = returned_expiry.replace(tzinfo=timezone.utc) return int(row.request_count), returned_expiry.astimezone(timezone.utc) def _bucket_id(self, quota: RateLimitQuota, raw_value: str) -> str: material = "\x00".join((quota.name, quota.key_kind, raw_value)).encode("utf-8") return hmac.new( self._digest_pepper, RATE_LIMIT_DIGEST_DOMAIN + material, hashlib.sha256, ).hexdigest() def _normalized_now(self) -> datetime: value = self._now_provider() if not isinstance(value, datetime): raise TypeError("now_provider must return a datetime") if value.tzinfo is None or value.utcoffset() is None: raise ValueError("now_provider must return a timezone-aware datetime") return value.astimezone(timezone.utc) def _should_cleanup(self) -> bool: with self._cleanup_lock: if self._requests_until_cleanup <= 0: self._requests_until_cleanup = RATE_LIMIT_CLEANUP_REQUEST_INTERVAL return True self._requests_until_cleanup -= 1 return False def request_rate_limit_context(request_state: object) -> RequestRateLimitContext: context = getattr(request_state, RATE_LIMIT_CONTEXT_STATE_KEY, None) if not isinstance(context, RequestRateLimitContext): raise ServerLicensingError( ServerErrorCode.RATE_LIMIT_UNAVAILABLE, "licensing request protection context is unavailable", status_code=503, retryable=True, ) return context __all__ = [ "DEFAULT_PRINCIPAL_QUOTAS", "DEFAULT_REQUEST_QUOTAS", "DatabaseRateLimiter", "RATE_LIMIT_CONTEXT_STATE_KEY", "RATE_LIMIT_EXEMPT_ACTION", "RateLimitQuota", "RequestRateLimitContext", "RequestRateLimitContextMiddleware", "classify_rate_limit_action", "extract_rate_limit_context", "request_rate_limit_context", ]