"""Server-side feature gating — the enforcement the skill requires be real, not a client-side hide. A route that a plan doesn't include is genuinely refused here. Usage in a route: from backend.billing.gating import require_feature @router.post("/xr/model3d/builder") async def build(..., _=Depends(require_feature(Feature.AR_MODEL_GEN))): ... """ from __future__ import annotations from fastapi import Depends, Header, HTTPException, Request from backend.billing.plans import Feature, OWNER_USER_ID from backend.billing.subscription import effective_plan from backend.billing import store # OWNER_USER_ID lives in plans.py (shared by gating + subscription). # # SECURITY: this used to be `return (x_user_id or OWNER_USER_ID)` — i.e. a request that # sent NO identity header resolved to the owner, inheriting the hidden omega_owner plan: # every feature including FLEET_ADMIN, unmetered, and exempt from rate limiting. Combined # with the billing router being mounted without verify_token, that meant an entirely # unauthenticated caller was treated as the operator. Omitting the credential granted # strictly more access than presenting one. # # Identity now comes from the authenticated bearer token, which is the only thing that # actually proves who the caller is. The header may narrow identity (so the owner can # QA a lower plan without losing owner status — an owner-plan feature the skill asks # for), but it can never elevate: an unauthenticated caller cannot claim to be the # owner, and cannot reach owner identity by omitting the header either. ANONYMOUS_USER_ID = "anonymous" def _bearer_token(request: Request | None) -> str: if request is None: return "" raw = request.headers.get("authorization") or "" scheme, _, value = raw.partition(" ") return value.strip() if scheme.lower() == "bearer" else "" def is_owner_request(request: Request | None) -> bool: """True only when the caller presented the real operator token.""" token = _bearer_token(request) if not token: return False try: from backend.dependencies.auth import get_or_create_master_token import secrets as _secrets return _secrets.compare_digest(token, get_or_create_master_token()) except Exception: # Fail closed: if the master token can't be resolved, nobody is the owner. return False def resolve_user_id(request: Request, x_user_id: str | None = Header(default=None)) -> str: supplied = (x_user_id or "").strip() if is_owner_request(request): # Authenticated operator: may impersonate any account for QA, or stay owner. return supplied or OWNER_USER_ID # Not the operator. A supplied identity is honoured (subscribers are not yet issued # their own tokens), but claiming the owner identity is refused outright, and the # absence of a header resolves to an unprivileged anonymous account — never owner. if supplied and supplied != OWNER_USER_ID: return supplied return ANONYMOUS_USER_ID def require_owner(request: Request) -> str: """Dependency for genuinely owner-only routes (audit trail, kill-switch, reconcile). Requires the real operator token — a missing header is a 403, not a free pass.""" if not is_owner_request(request): raise HTTPException(status_code=403, detail="forbidden") return OWNER_USER_ID def rate_limit(bucket_prefix: str, limit: int, window_seconds: float = 60.0): """Per-user AND per-IP sliding-window throttle for AI-heavy / billing routes, so the metering cap can't be bypassed by hammering faster than cap accounting settles. 429 with Retry-After when either bucket is over. Owner is never limited. """ def _dep(request: Request, user_id: str = Depends(resolve_user_id)) -> str: if user_id == OWNER_USER_ID: return user_id ip = (request.client.host if request and request.client else "unknown") for bucket in (f"{bucket_prefix}:user:{user_id}", f"{bucket_prefix}:ip:{ip}"): r = store.rate_check(bucket, limit, window_seconds) if not r["allowed"]: raise HTTPException(status_code=429, headers={"Retry-After": str(int(window_seconds))}, detail={"error": "rate_limited", "bucket": bucket_prefix, "limit": limit, "window_seconds": window_seconds}) return user_id return _dep def require_flag(flag_name: str): """Kill-switch: a risky surface can be disabled operator-side without a rollback. 503 when the flag is off. Defaults to enabled if never set.""" def _dep() -> None: if not store.flag_enabled(flag_name, default=True): raise HTTPException(status_code=503, detail={ "error": "feature_disabled", "flag": flag_name, "message": "Temporarily disabled by the operator."}) return _dep def require_feature(feature: Feature): def _dep(user_id: str = Depends(resolve_user_id)) -> str: plan = effective_plan(user_id) if not plan.grants(feature): raise HTTPException( status_code=402, # Payment Required — the honest gated status detail={"error": "feature_not_in_plan", "feature": feature.value, "plan": plan.id, "upgrade_required": True}) return user_id return _dep def gate_and_meter(feature: Feature): """Gate on the feature AND consume one metered unit (reserve-then-do). Use on the company-provisioned model-generation routes so a plan's cap is enforced server-side, per-user. 402 if the plan lacks the feature; 429 if the cap is reached (body carries the plan's configurable at-cap policy so the UI can show the right prompt). Imported lazily to avoid a circular import with metering.""" def _dep(user_id: str = Depends(resolve_user_id)) -> str: from backend.billing.metering import check_and_consume result = check_and_consume(user_id, feature) if result["allowed"]: return user_id reason = result["reason"] if reason == "feature_not_in_plan": raise HTTPException(status_code=402, detail={ "error": "feature_not_in_plan", "feature": feature.value, "upgrade_required": True, "usage": result["status"]}) raise HTTPException(status_code=429, detail={ "error": "usage_cap_reached", "feature": feature.value, "cap_policy": result["status"]["cap_policy"], "usage": result["status"]}) return _dep