feat(entitlements): enforce plans on every write, limit, redirect and edge path

Gated writes answer 403 plan_required and count limits answer 403
limit_reached from the resolved map, never from config. The redirect path
shapes owned links by the owner's entitlement version through the URL cache,
custom domains stop serving once their owner lost them, and the analytics
window clamps instead of rejecting. Edge promotion picks the tier threshold
from the owner's plan and flags tracked entries; over-limit items pause newest first.

Author: Zingzy

.env.example

@@ -143,7 +143,6 @@ GITHUB_REPO="" # org/repo format
 # Access is feature-flag gated per user on top of this master switch.
 # WEBHOOKS_ENABLED="false"
 # WEBHOOKS_RUNTIME="auto"                        # auto | worker | embedded | off
-# WEBHOOKS_MAX_ENDPOINTS="5"                     # per user
 # WEBHOOKS_DELIVERY_TIMEOUT_SECONDS="15"
 # WEBHOOKS_MAX_CONSECUTIVE_FAILURES="10"         # exhausted deliveries → auto-disable
 # WEBHOOKS_DELIVERY_LOG_TTL_DAYS="30"            # events + delivery log retention

config.py

@@ -161,7 +161,7 @@ class CustomDomainSettings(BaseSettings):
     even when False so the rollout has a clean code path to flip.
 
     All env vars must be prefixed ``CUSTOM_DOMAINS_`` so generic names
-    like ``ENABLED`` or ``MAX_PER_USER`` set elsewhere in the deploy
+    like ``ENABLED`` or ``MOCK_DCV`` set elsewhere in the deploy
     environment don't accidentally configure this feature.
     """
 
@@ -186,7 +186,6 @@ class CustomDomainSettings(BaseSettings):
     # All counts must be >= 1 — a zero quota silently bricks the feature
     # (every create raises QuotaExceeded with no log signal that the cause
     # is config, not abuse). Validators below fail container startup instead.
-    max_per_user: int = Field(default=1, ge=1)
     # Generous because CF's own DCV cadence can take 5-15 min per probe;
     # legitimate users may poll many times during initial activation.
     verify_attempts_per_hour: int = Field(default=60, ge=1)
@@ -457,7 +456,6 @@ class WebhookSettings(BaseSettings):
     enabled: bool = False
     runtime: Literal["auto", "worker", "embedded", "off"] = "auto"
 
-    max_endpoints: int = Field(default=5, ge=1)
     delivery_timeout_seconds: float = Field(default=15.0, gt=0)
     max_payload_bytes: int = Field(default=20_480, ge=1024)
     max_consecutive_failures: int = Field(default=10, ge=1)
@@ -755,8 +753,6 @@ def blocked_self_domains(self) -> tuple[str, ...]:
     hcaptcha_sitekey: str = ""
 
     # Service limits (overridable by self-hosters via env vars)
-    max_active_api_keys: int = 20
-    max_date_range_days: int = 90
     http_client_timeout: float = 5.0
     # Account deletion (GDPR Art. 17): days between the deletion request
     # and the erasure sweep purging the account (0 = purge on the next
@@ -798,8 +794,6 @@ def blocked_self_domains(self) -> tuple[str, ...]:
     # ── Field validators for safety-critical config ────────────────────
 
     @field_validator(
-        "max_active_api_keys",
-        "max_date_range_days",
         "max_emoji_alias_length",
         "emoji_generated_alias_length",
         "geo_rules_max_countries",

dependencies/auth.py

@@ -121,6 +121,16 @@ async def get_current_user(
                 )
                 return None
 
+            if key.paused_by_limit:
+                log.warning(
+                    "api_key_auth_failed",
+                    reason="paused_by_limit",
+                    key_prefix=key.token_prefix,
+                    key_id=str(key.id),
+                    user_id=str(key.user_id),
+                )
+                return None
+
             now = datetime.now(timezone.utc)
             exp = as_aware_utc(key.expires_at)
             if exp is not None and exp <= now:

dependencies/entitlements.py

@@ -29,6 +29,7 @@ async def get_entitlements(
         user.user_id if user else None,
         plan_hint=user.plan_claim if user else None,
     )
+    request.state.entitlements = resolved
     if user is not None and not resolved.degraded:
         request.state.entitlements_version = resolved.version
     return resolved

dependencies/wiring.py

@@ -86,6 +86,7 @@
 from services.domain_intel_service import DomainIntelService
 from services.edge_cache.og_writethrough import OgEdgeWritethrough
 from services.entitlements import EntitlementService
+from services.entitlements.over_limit import OverLimitService
 from services.events.sinks import (
     InlineDomainEventSink,
     NullDomainEventSink,
@@ -291,21 +292,34 @@ def build_entitlement_service(
     cache, events, subscriptions, overrides = store or build_entitlement_store(
         db, redis_client
     )
+    domain_repo = CustomDomainRepository(db["custom_domains"])
+    endpoint_repo = WebhookEndpointRepository(db["webhook-endpoints"])
+    key_repo = ApiKeyRepository(db["api-keys"])
+    reconciler = OverLimitService(
+        endpoints=endpoint_repo,
+        domains=domain_repo,
+        keys=key_repo,
+        tenant_resolver=CachedMongoTenantResolver(
+            repo=domain_repo,
+            redis_client=redis_client,
+            system_default_domain=settings.system_default_domain,
+        ),
+        webhook_owner_cache=OwnerSubscriptionCache(
+            redis_client, ttl_seconds=settings.webhooks.matcher_cache_ttl_seconds
+        ),
+    )
     return EntitlementService(
         subscriptions,
         overrides,
         events,
         cache,
         selfhost=settings.billing.selfhost,
         usage={
-            "custom_domains_max": CustomDomainRepository(
-                db["custom_domains"]
-            ).count_by_owner,
-            "webhook_endpoints_max": WebhookEndpointRepository(
-                db["webhook-endpoints"]
-            ).count_by_user,
-            "api_keys_max": ApiKeyRepository(db["api-keys"]).count_by_user,
+            "custom_domains_max": domain_repo.count_by_owner,
+            "webhook_endpoints_max": endpoint_repo.count_by_user,
+            "api_keys_max": key_repo.count_by_user,
         },
+        over_limit=reconciler,
     )
 
 
@@ -601,7 +615,6 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
         webhook_executor,
         webhook_owner_cache,
         master_secret=settings.secret_key,
-        max_endpoints=wh_settings.max_endpoints,
     )
 
     # ── Safety pipeline ──────────────────────────────────────────────
@@ -764,6 +777,7 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
         events=app.state.domain_event_sink,
         user_repo=user_repo,
         tag_service=app.state.tag_service,
+        entitlements=app.state.entitlement_service,
     )
     app.state.bulk_url_service = BulkUrlService(
         url_repo,
@@ -778,7 +792,6 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
     app.state.stats_service = StatsService(
         click_repo,
         url_repo,
-        max_date_range_days=settings.max_date_range_days,
         tag_service=app.state.tag_service,
     )
     # One resolver serves BOTH public read-only surfaces (preview + stats)
@@ -790,7 +803,15 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
         emoji_repo,
         system_default_domain=settings.system_default_domain,
     )
-    app.state.public_preview_service = PublicPreviewService(public_link_resolver)
+    app.state.feature_flag_service = FeatureFlagService(
+        feature_flag_repo, feature_flag_cache
+    )
+
+    app.state.public_preview_service = PublicPreviewService(
+        public_link_resolver,
+        app.state.entitlement_service,
+        app.state.feature_flag_service,
+    )
     app.state.url_expand_service = UrlExpandService(
         blocked_url_repo,
         MetaFetchCache(redis_client, prefix="url_expand"),
@@ -808,7 +829,8 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
     app.state.public_stats_service = PublicStatsService(
         public_link_resolver,
         app.state.stats_service,
-        max_date_range_days=settings.max_date_range_days,
+        app.state.entitlement_service,
+        app.state.feature_flag_service,
     )
     # Report intake shares the resolver (existence checks answer from the
     # same generation the redirect serves) and the ops notifier + captcha
@@ -832,7 +854,6 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
     )
     app.state.api_key_service = ApiKeyService(
         api_key_repo,
-        max_active_keys=settings.max_active_api_keys,
     )
     app.state.page_layout_service = PageLayoutService(page_layout_repo)
     token_factory = TokenFactory(
@@ -940,10 +961,6 @@ def wire_services(app: FastAPI, settings: AppSettings, redis_client) -> None:
 
     app.state.app_grant_repo = app_grant_repo
 
-    app.state.feature_flag_service = FeatureFlagService(
-        feature_flag_repo, feature_flag_cache
-    )
-
     # Whenever clicks are tracked inline, link.clicked webhooks must fan
     # out at emit time — the worker's stream group only sees clicks that
     # ride the stream. Keying on the CLICK sink (not the domain sink)

errors.py

@@ -131,6 +131,41 @@ class InvalidTransitionError(ConflictError):
     error_code = "invalid_transition"
 
 
+class PlanRequiredError(ForbiddenError):
+    """The account's plan does not include the feature being written."""
+
+    error_code = "plan_required"
+
+    def __init__(self, feature: str) -> None:
+        super().__init__(f"{feature} is not included in your plan")
+        self.feature = feature
+
+    def to_dict(self) -> ErrorBody:
+        d = super().to_dict()
+        d["feature"] = self.feature
+        return d
+
+
+class LimitReachedError(ForbiddenError):
+    """A per-plan count limit is full; ``max`` and ``current`` let the UI
+    render the counter."""
+
+    error_code = "limit_reached"
+
+    def __init__(self, limit: str, *, max_: int, current: int) -> None:
+        super().__init__(f"{limit} limit reached ({current}/{max_})")
+        self.limit = limit
+        self.max = max_
+        self.current = current
+
+    def to_dict(self) -> ErrorBody:
+        d = super().to_dict()
+        d["limit"] = self.limit
+        d["max"] = self.max
+        d["current"] = self.current
+        return d
+
+
 class BlockedUrlError(AppError):
     status_code = 451
     error_code = "blocked"

infrastructure/cache/url_cache.py

@@ -61,6 +61,9 @@ class UrlCacheData(BaseModel):
     # Entries cached before this field existed deserialize to None.
     start_time: int | None = None
     pre_start_url: str | None = None
+    # Owner entitlement version this entry was shaped under; None for
+    # unowned links and for entries written before the owner was consulted.
+    owner_ent_version: int | None = None
 
     @classmethod
     def from_v2_doc(cls, doc: UrlV2Doc) -> UrlCacheData:

middleware/rate_limiter.py

@@ -10,12 +10,16 @@
 import hashlib
 import math
 import os
+import re
+from collections.abc import Callable
 
 from fastapi import Request
 from slowapi import Limiter
 from starlette.datastructures import MutableHeaders
 
 from infrastructure.logging import get_logger
+from services.entitlements.resolver import Resolved
+from services.features.catalog import UNLIMITED
 from shared.ip_utils import get_client_ip
 
 log = get_logger(__name__)
@@ -91,8 +95,8 @@ class Limits:
     # so the request budget must never make bulk scarcer than looping the
     # per-item routes — that would push clients back to the fan-out these
     # endpoints exist to kill. Per-minute is kept high for bursts (a mass
-    # takedown chunks at 100 ids/request); the daily cap is a lid, not a
-    # ration. Blast radius per request is bounded by the 100-id cap and
+    # takedown chunks at the plan's bulk_batch_max); the daily cap is a lid, not a
+    # ration. Blast radius per request is bounded by that batch cap and
     # ownership scoping, and each batch is ~4 local calls plus at most
     # one CF call, so these are cheap requests. Delete stays the tighter
     # pair because it is irreversible.
@@ -190,12 +194,26 @@ class Limits:
 # ── Key resolution ───────────────────────────────────────────────────────────
 
 
+# Plan multiplier suffix on the key, so slowapi counts each tier in its own
+# bucket and the limit function can scale the string it applies.
+_MULTIPLIER_SEP = "|x"
+# What "unlimited" means for a rate: still bounded, just far above any client.
+_UNLIMITED_MULTIPLIER = 1000
+
+
 def rate_limit_key(request: Request) -> str:
-    """Three-tier rate limit key: API key hash → JWT hash → client IP.
+    """Three-tier rate limit key: API key hash → JWT hash → client IP, plus
+    the caller's plan rate multiplier when the route resolved entitlements.
 
     Lightweight header inspection only — no DB queries, no JWT verification.
     Provides consistent per-session bucketing for rate limiting purposes.
     """
+    key = _identity_key(request)
+    multiplier = plan_multiplier(request)
+    return f"{key}{_MULTIPLIER_SEP}{multiplier}" if multiplier > 1 else key
+
+
+def _identity_key(request: Request) -> str:
     auth_header = request.headers.get("Authorization", "")
 
     if auth_header.lower().startswith("bearer "):
@@ -214,6 +232,49 @@ def rate_limit_key(request: Request) -> str:
     return get_client_ip(request)
 
 
+def plan_multiplier(request: Request) -> int:
+    """The ``api_rate_multiplier`` of the entitlements the ``Entitled``
+    dependency stored on this request; 1 when the route did not resolve them."""
+    resolved = getattr(request.state, "entitlements", None)
+    if not isinstance(resolved, Resolved):
+        return 1
+    value = resolved.limit("api_rate_multiplier")
+    if value == UNLIMITED:
+        return _UNLIMITED_MULTIPLIER
+    return value if value > 1 else 1
+
+
+def multiplier_of(key: str) -> int:
+    _, sep, suffix = key.rpartition(_MULTIPLIER_SEP)
+    return int(suffix) if sep and suffix.isdigit() else 1
+
+
+def scale_limit(limit: str, factor: int) -> str:
+    """``"60 per minute; 5000 per day"`` times ``factor`` on every count."""
+    if factor <= 1:
+        return limit
+    return "; ".join(
+        re.sub(r"^\d+", lambda m: str(int(m.group(0)) * factor), part.strip())
+        for part in limit.split(";")
+    )
+
+
+def plan_scaled(limit: str) -> Callable[[str], str]:
+    """Limit callable for ``@limiter.limit``: the base string scaled by the
+    plan multiplier carried on the key. The route must declare ``Entitled``
+    so the multiplier is known before the limiter runs.
+
+    Scaled: the per-request API budgets an integration spends on links,
+    stats and exports. Flat: auth, keys, domains, webhooks and bulk, which
+    are admin actions (bulk already scales through ``bulk_batch_max``).
+    """
+
+    def _limit(key: str) -> str:
+        return scale_limit(limit, multiplier_of(key))
+
+    return _limit
+
+
 # ── Limiter singleton ────────────────────────────────────────────────────────
 
 _redis_uri = os.environ.get("REDIS_URI")
@@ -306,7 +367,7 @@ async def endpoint(request: Request, ...): ...
 
     def _limit(key: str) -> str:
         if key.startswith("apikey:") or key.startswith("jwt:"):
-            return authenticated
+            return scale_limit(authenticated, multiplier_of(key))
         return anonymous
 
     return _limit, rate_limit_key