feat: storefront subscription access guard + module-driven nav + URL rename
Add StorefrontAccessMiddleware that blocks storefront access for stores without an active subscription, returning a multilingual unavailable page (en/fr/de/lb) for page requests and JSON 403 for API requests. Multi-platform aware: resolves subscription for detected platform with fallback to primary. Also includes yesterday's session work: - Module-driven storefront navigation via FrontendType.STOREFRONT menu declarations - shop/ → storefront/ URL rename across 30+ templates - Subscription context (tier_code) passed to storefront templates Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
189
middleware/storefront_access.py
Normal file
189
middleware/storefront_access.py
Normal file
@@ -0,0 +1,189 @@
|
||||
# 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, handling multi-platform stores correctly."""
|
||||
from app.modules.billing.services.subscription_service import (
|
||||
subscription_service,
|
||||
)
|
||||
|
||||
platform = getattr(request.state, "platform", None)
|
||||
|
||||
# If we have a detected platform, check subscription for THAT platform
|
||||
if platform:
|
||||
sub = subscription_service.get_merchant_subscription(
|
||||
db, store.merchant_id, platform.id
|
||||
)
|
||||
if sub:
|
||||
return sub
|
||||
|
||||
# Fallback: use store's primary platform (via StorePlatform)
|
||||
return subscription_service.get_subscription_for_store(db, store.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,
|
||||
)
|
||||
Reference in New Issue
Block a user