# middleware/storefront_access.py """ Storefront subscription access guard. Blocks storefront access for stores without an active subscription. Inserted in the middleware chain AFTER LanguageMiddleware, BEFORE ThemeContextMiddleware, so request.state has: platform, store, frontend_type, language. For page requests: renders a standalone "unavailable" HTML page. For API requests: returns JSON 403. Multi-platform aware: checks subscription for the detected platform first, falls back to the store's primary platform subscription. """ import logging from starlette.middleware.base import BaseHTTPMiddleware from starlette.requests import Request from starlette.responses import JSONResponse, Response from app.core.database import get_db from app.modules.enums import FrontendType logger = logging.getLogger(__name__) # Paths that should never be blocked (health, static, etc.) SKIP_PATH_PREFIXES = ("/static/", "/uploads/", "/health", "/docs", "/redoc", "/openapi.json") STATIC_EXTENSIONS = ( ".ico", ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".woff", ".woff2", ".ttf", ".eot", ".webp", ".map", ) # Multilingual messages (en, fr, de, lb) MESSAGES = { "not_found": { "en": { "title": "Storefront Not Found", "message": "This storefront does not exist or has been removed.", }, "fr": { "title": "Boutique introuvable", "message": "Cette boutique n'existe pas ou a \u00e9t\u00e9 supprim\u00e9e.", }, "de": { "title": "Shop nicht gefunden", "message": "Dieser Shop existiert nicht oder wurde entfernt.", }, "lb": { "title": "Buttek net fonnt", "message": "D\u00ebse Buttek exist\u00e9iert net oder gouf ewechgeholl.", }, }, "not_activated": { "en": { "title": "Storefront Not Activated", "message": "This storefront is not yet activated. Please contact the store owner.", }, "fr": { "title": "Boutique non activ\u00e9e", "message": "Cette boutique n'est pas encore activ\u00e9e. Veuillez contacter le propri\u00e9taire.", }, "de": { "title": "Shop nicht aktiviert", "message": "Dieser Shop ist noch nicht aktiviert. Bitte kontaktieren Sie den Inhaber.", }, "lb": { "title": "Buttek net aktiv\u00e9iert", "message": "D\u00ebse Buttek ass nach net aktiv\u00e9iert. Kontakt\u00e9iert w.e.g. den Bes\u00ebtzer.", }, }, } def _is_static_request(path: str) -> bool: """Check if path targets a static resource.""" lower = path.lower() if any(lower.startswith(p) for p in SKIP_PATH_PREFIXES): return True if lower.endswith(STATIC_EXTENSIONS): return True return "favicon.ico" in lower class StorefrontAccessMiddleware(BaseHTTPMiddleware): """ Gate storefront requests behind an active subscription. Execution position (request flow): ... -> LanguageMiddleware -> **StorefrontAccessMiddleware** -> ThemeContextMiddleware -> Router """ async def dispatch(self, request: Request, call_next) -> Response: frontend_type = getattr(request.state, "frontend_type", None) # Only gate storefront requests if frontend_type != FrontendType.STOREFRONT: return await call_next(request) # Skip static files if _is_static_request(request.url.path): return await call_next(request) store = getattr(request.state, "store", None) # Case 1: No store detected at all if not store: return self._render_unavailable(request, "not_found") # Case 2: Store exists — check subscription db = next(get_db()) try: subscription = self._get_subscription(db, store, request) if not subscription or not subscription.is_active: logger.info( f"[STOREFRONT_ACCESS] Blocked store '{store.subdomain}' " f"(merchant_id={store.merchant_id}): no active subscription" ) return self._render_unavailable(request, "not_activated", store) # Store subscription info for downstream use request.state.subscription = subscription request.state.subscription_tier = subscription.tier finally: db.close() return await call_next(request) def _get_subscription(self, db, store, request): """Resolve subscription for the detected platform. No fallback.""" from app.modules.billing.services.subscription_service import ( subscription_service, ) platform = getattr(request.state, "platform", None) if not platform: logger.warning( f"[STOREFRONT_ACCESS] No platform context for store '{store.subdomain}'" ) return None return subscription_service.get_merchant_subscription( db, store.merchant_id, platform.id ) def _render_unavailable( self, request: Request, reason: str, store=None ) -> Response: """Return an appropriate response for blocked requests.""" is_api = request.url.path.startswith("/api/") if is_api: return JSONResponse( status_code=403, content={ "error": "storefront_not_available", "reason": reason, }, ) # Page request — render HTML language = getattr(request.state, "language", "en") if language not in MESSAGES[reason]: language = "en" msgs = MESSAGES[reason][language] theme = getattr(request.state, "theme", None) from app.templates_config import templates context = { "request": request, "reason": reason, "title": msgs["title"], "message": msgs["message"], "language": language, "store": store, "theme": theme, } return templates.TemplateResponse( "storefront/unavailable.html", context, status_code=403, )