jcv-dev
551b42eab5
Close the gaps between the documented contract (API-STANDARD, TENANCY, ENTITLEMENTS) and the implementation, and remove committed build artifacts. Security: - verify + process inbound webhooks (HMAC/handler verify, size limit, timestamp tolerance, idempotency via InboundEvent.external_id) - real 2FA login flow (pre-auth challenge; tokens only after verify/recovery) - wire HybridPermission into security/audit views; add API-key rotate and users/<id>/permissions|roles endpoints - tenant-scoped throttling on by default; webhook replay protection - verify MercadoPago webhook signatures - login brute-force guard, configurable password policy, real ALTCHA PoW Correctness: - apply verified billing webhooks idempotently (subscription/entitlement/ invoice/PaymentTransaction); scheduled payment lifecycle jobs - capture audit update diffs automatically; add audit retention purge - working notification retries, per-channel rate limits, log retention - pluggable virus scanner, upload-size limit, pipeline toggle - feature rollout %/environment targeting; settings-driven registrations - workflow guards (instance cap, route depth, self-assignment, clone on re-entry) - wire every previously-dead INFRASYNTH_* setting; drop truly dead ones Delivery: - README + CHANGELOG; CI format check + coverage gate - keep test media out of the tree; untrack .coverage, __pycache__, egg-info, docs/ and invoice artifacts
101 lines
3.9 KiB
Python
101 lines
3.9 KiB
Python
"""``EntitlementService`` — in-process commercial enforcement (``ENTITLEMENTS.md`` §4)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from django.core.cache import cache
|
|
|
|
from infrasynth.shared.settings_utils import get_setting
|
|
|
|
from .models import Entitlement
|
|
|
|
__all__ = ["EntitlementService"]
|
|
|
|
_ENTITLED_STATUSES = {"trialing", "active", "past_due", "grace"}
|
|
_TENANT_OK_STATUSES = {"active", "trialing"}
|
|
|
|
|
|
class EntitlementService:
|
|
"""The only gate for commercial access. Feature flags never replace it."""
|
|
|
|
def __init__(self) -> None:
|
|
self.ttl = int(get_setting("INFRASYNTH_BILLING", "ENTITLEMENT_CACHE_TTL_SECONDS", 60))
|
|
|
|
def _cache_key(self, tenant: Any, app_slug: str) -> str:
|
|
return f"tenant:{tenant.pk}:entitlements:{app_slug}"
|
|
|
|
def get(self, tenant: Any, app_slug: str) -> Entitlement | None:
|
|
if tenant is None:
|
|
return None
|
|
key = self._cache_key(tenant, app_slug)
|
|
cached = cache.get(key)
|
|
if cached is not None:
|
|
return cached if isinstance(cached, Entitlement) else None
|
|
entitlement = (
|
|
Entitlement.all_objects.filter(tenant=tenant, app__slug=app_slug).select_related("plan", "app").first()
|
|
)
|
|
cache.set(key, entitlement, self.ttl)
|
|
return entitlement
|
|
|
|
def is_entitled(self, tenant: Any, app_slug: str, *, feature: str | None = None) -> bool:
|
|
if tenant is None:
|
|
return False
|
|
# Tenant status gates before entitlement state (ENTITLEMENTS.md §4).
|
|
if getattr(tenant, "status", None) not in _TENANT_OK_STATUSES:
|
|
return False
|
|
entitlement = self.get(tenant, app_slug)
|
|
if entitlement is None:
|
|
return False
|
|
if entitlement.status not in _ENTITLED_STATUSES:
|
|
return False
|
|
if feature is None:
|
|
return True
|
|
return bool(self._features(entitlement).get(feature, False))
|
|
|
|
@staticmethod
|
|
def _features(entitlement: Entitlement) -> dict:
|
|
"""``plan.features`` merged under any entitlement-level override."""
|
|
features: dict = {}
|
|
if entitlement.plan:
|
|
features.update(entitlement.plan.features or {})
|
|
override = (entitlement.metadata or {}).get("features")
|
|
if isinstance(override, dict):
|
|
features.update(override)
|
|
return features
|
|
|
|
def check_limit(self, tenant: Any, app_slug: str, limit: str, current: int) -> bool:
|
|
"""True when ``current`` is within the plan's ``limit`` (None = unlimited)."""
|
|
entitlement = self.get(tenant, app_slug)
|
|
if entitlement is None or entitlement.plan is None:
|
|
return False
|
|
max_value = (entitlement.plan.limits or {}).get(limit)
|
|
if max_value is None:
|
|
return True
|
|
return current < int(max_value)
|
|
|
|
def require_limit(self, tenant: Any, app_slug: str, limit: str, current: int) -> None:
|
|
"""Raises :class:`EntitlementError` when the limit would be exceeded."""
|
|
if self.check_limit(tenant, app_slug, limit, current):
|
|
return
|
|
from infrasynth.shared.exceptions import ENTITLEMENT_LIMIT_REACHED, EntitlementError
|
|
|
|
entitlement = self.get(tenant, app_slug)
|
|
max_value = (entitlement.plan.limits or {}).get(limit) if entitlement and entitlement.plan else None
|
|
raise EntitlementError(
|
|
code=ENTITLEMENT_LIMIT_REACHED,
|
|
app=app_slug,
|
|
limit=limit,
|
|
current=current,
|
|
max=max_value,
|
|
)
|
|
|
|
def invalidate(self, tenant: Any, app_slug: str | None = None) -> None:
|
|
if tenant is None:
|
|
return
|
|
if app_slug is not None:
|
|
cache.delete(self._cache_key(tenant, app_slug))
|
|
return
|
|
slugs = Entitlement.all_objects.filter(tenant=tenant).values_list("app__slug", flat=True)
|
|
for slug in set(slugs):
|
|
cache.delete(self._cache_key(tenant, slug))
|