Samir Boulahtit
f47c680cb8
- Extract store/platform context from Referer header for storefront API requests
(StoreContextMiddleware and PlatformContextMiddleware) so login POST works in
dev mode where API paths lack /platforms/{code}/ prefix
- Set customer token cookie path to "/" for cross-route compatibility
- Fix double storefront in URLs: replace {{ base_url }}storefront/ with {{ base_url }}
across all 24 storefront templates
- Fix auth error redirect to include platform prefix and use store_code
- Update seed script to output correct storefront login URLs
- Add 20 new unit tests covering all fixes; fix 9 pre-existing test failures
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
582 lines
23 KiB
Python
582 lines
23 KiB
Python
# middleware/platform_context.py
|
|
"""
|
|
Platform Context Middleware
|
|
|
|
Detects platform from host/domain/path and injects into request.state.
|
|
This middleware runs BEFORE StoreContextMiddleware to establish platform context.
|
|
|
|
Handles two routing modes:
|
|
1. Production: Domain-based (omsflow.lu, rewardflow.lu → Platform detection)
|
|
2. Development: Path-based (localhost:9999/platforms/oms/*, localhost:9999/platforms/loyalty/*)
|
|
|
|
URL Structure:
|
|
- Main marketing site: localhost:9999/ (no platform prefix) → uses 'main' platform
|
|
- Platform sites: localhost:9999/platforms/{code}/ → uses specific platform
|
|
|
|
Also provides platform_clean_path for downstream middleware to use.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import Request
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.database import get_db
|
|
from app.core.frontend_detector import FrontendDetector
|
|
from app.modules.tenancy.models import Platform
|
|
|
|
# Note: We use pure ASGI middleware (not BaseHTTPMiddleware) to enable path rewriting
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Platform code for the main marketing site (localhost without /platforms/ prefix)
|
|
MAIN_PLATFORM_CODE = "main"
|
|
|
|
# Hosts treated as local development (including Starlette TestClient's "testserver")
|
|
_LOCAL_HOSTS = {"localhost", "127.0.0.1", "testserver"}
|
|
|
|
|
|
class PlatformContextManager:
|
|
"""Manages platform context detection for multi-platform routing."""
|
|
|
|
@staticmethod
|
|
def detect_platform_context(request: Request) -> dict | None:
|
|
"""
|
|
Detect platform context from request.
|
|
|
|
Priority order:
|
|
1. Domain-based (production): omsflow.lu → platform code "oms"
|
|
2. Path-based (development): localhost:9999/platforms/oms/* → platform code "oms"
|
|
3. Default: localhost without /platforms/ prefix → 'main' platform (marketing site)
|
|
|
|
URL Structure:
|
|
- / → Main marketing site ('main' platform)
|
|
- /about → Main marketing site about page
|
|
- /platforms/oms/ → OMS platform homepage
|
|
- /platforms/oms/pricing → OMS platform pricing
|
|
- /platforms/loyalty/ → Loyalty platform homepage
|
|
|
|
Returns dict with platform info or None if not detected.
|
|
"""
|
|
host = request.headers.get("host", "")
|
|
path = request.url.path
|
|
|
|
# Remove port from host if present (e.g., localhost:9999 -> localhost)
|
|
host_without_port = host.split(":")[0] if ":" in host else host
|
|
|
|
# Skip platform detection for admin routes - admin is global
|
|
if FrontendDetector.is_admin(host, path):
|
|
return None
|
|
|
|
# Method 1: Domain-based detection (production)
|
|
# Check if the host matches a known platform domain
|
|
# This will be resolved in get_platform_from_context by DB lookup
|
|
if host_without_port and host_without_port not in _LOCAL_HOSTS:
|
|
# Could be a platform domain or a store subdomain/custom domain
|
|
# Check if it's a known platform domain pattern
|
|
# For now, assume non-localhost hosts that aren't subdomains are platform domains
|
|
if "." in host_without_port:
|
|
# This could be:
|
|
# - Platform domain: omsflow.lu, rewardflow.lu
|
|
# - Store subdomain: store.omsflow.lu
|
|
# - Custom domain: shop.mymerchant.com
|
|
# We detect platform domain vs subdomain by checking if it's a root domain
|
|
parts = host_without_port.split(".")
|
|
if len(parts) == 2: # e.g., omsflow.lu (root domain)
|
|
return {
|
|
"domain": host_without_port,
|
|
"detection_method": "domain",
|
|
"host": host,
|
|
"original_path": path,
|
|
}
|
|
|
|
# Method 2: Path-based detection (development) - ONLY for /platforms/ prefix
|
|
# Check for path prefix like /platforms/oms/, /platforms/loyalty/
|
|
if path.startswith("/platforms/"):
|
|
# Extract: /platforms/oms/pricing → code="oms", clean_path="/pricing"
|
|
path_after_platforms = path[11:] # Remove "/platforms/"
|
|
parts = path_after_platforms.split("/", 1)
|
|
platform_code = parts[0].lower()
|
|
|
|
if platform_code:
|
|
clean_path = "/" + parts[1] if len(parts) > 1 and parts[1] else "/"
|
|
return {
|
|
"path_prefix": platform_code,
|
|
"detection_method": "path",
|
|
"host": host,
|
|
"original_path": path,
|
|
"clean_path": clean_path,
|
|
}
|
|
|
|
# Method 3: Default platform for localhost without /platforms/ prefix
|
|
# This serves the main marketing site
|
|
# Store routes require explicit platform via /platforms/{code}/store/...
|
|
if host_without_port in _LOCAL_HOSTS:
|
|
if path.startswith(("/store/", "/stores/")):
|
|
return None # No platform — handlers will show appropriate error
|
|
return {
|
|
"path_prefix": MAIN_PLATFORM_CODE,
|
|
"detection_method": "default",
|
|
"host": host,
|
|
"original_path": path,
|
|
"clean_path": path, # No path rewrite for main site
|
|
}
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_platform_from_context(db: Session, context: dict) -> Platform | None:
|
|
"""
|
|
Get platform from database using context information.
|
|
|
|
Supports:
|
|
1. Domain-based lookup (Platform.domain)
|
|
2. Path-prefix lookup (Platform.path_prefix)
|
|
3. Default lookup (Platform.code)
|
|
"""
|
|
if not context:
|
|
return None
|
|
|
|
platform = None
|
|
|
|
# Method 1: Domain-based lookup (also handles subdomain detection)
|
|
if context.get("detection_method") in ("domain", "subdomain"):
|
|
domain = context.get("domain")
|
|
if domain:
|
|
# Try Platform.domain first
|
|
platform = (
|
|
db.query(Platform)
|
|
.filter(Platform.domain == domain)
|
|
.filter(Platform.is_active.is_(True))
|
|
.first()
|
|
)
|
|
|
|
if platform:
|
|
logger.debug(
|
|
f"[PLATFORM] Platform found via domain: {domain} → {platform.name}"
|
|
)
|
|
return platform
|
|
|
|
# Fallback: Check StoreDomain for custom store domains
|
|
from app.modules.tenancy.models import StoreDomain
|
|
store_domain = (
|
|
db.query(StoreDomain)
|
|
.filter(StoreDomain.domain == domain, StoreDomain.is_active.is_(True))
|
|
.first()
|
|
)
|
|
if store_domain and store_domain.platform_id:
|
|
platform = (
|
|
db.query(Platform)
|
|
.filter(Platform.id == store_domain.platform_id, Platform.is_active.is_(True))
|
|
.first()
|
|
)
|
|
if platform:
|
|
# Mark as store domain so __call__ knows to rewrite path
|
|
context["is_store_domain"] = True
|
|
logger.debug(
|
|
f"[PLATFORM] Platform found via store domain: {domain} → {platform.name}"
|
|
)
|
|
return platform
|
|
|
|
# Fallback: Check MerchantDomain for merchant-level domains
|
|
from app.modules.tenancy.models.merchant_domain import MerchantDomain
|
|
merchant_domain = (
|
|
db.query(MerchantDomain)
|
|
.filter(
|
|
MerchantDomain.domain == domain,
|
|
MerchantDomain.is_active.is_(True),
|
|
MerchantDomain.is_verified.is_(True),
|
|
)
|
|
.first()
|
|
)
|
|
if merchant_domain and merchant_domain.platform_id:
|
|
platform = (
|
|
db.query(Platform)
|
|
.filter(
|
|
Platform.id == merchant_domain.platform_id,
|
|
Platform.is_active.is_(True),
|
|
)
|
|
.first()
|
|
)
|
|
if platform:
|
|
# Mark as store domain so __call__ knows to rewrite path
|
|
context["is_store_domain"] = True
|
|
logger.debug(
|
|
f"[PLATFORM] Platform found via merchant domain: {domain} → {platform.name}"
|
|
)
|
|
return platform
|
|
|
|
logger.debug(f"[PLATFORM] No platform found for domain: {domain}")
|
|
|
|
# Method 2: Path-prefix lookup
|
|
if context.get("detection_method") == "path":
|
|
path_prefix = context.get("path_prefix")
|
|
if path_prefix:
|
|
# First try path_prefix, then try code
|
|
platform = (
|
|
db.query(Platform)
|
|
.filter(
|
|
(Platform.path_prefix == path_prefix) | (Platform.code == path_prefix)
|
|
)
|
|
.filter(Platform.is_active.is_(True))
|
|
.first()
|
|
)
|
|
|
|
if platform:
|
|
logger.debug(
|
|
f"[PLATFORM] Platform found via path prefix: {path_prefix} → {platform.name}"
|
|
)
|
|
return platform
|
|
logger.debug(f"[PLATFORM] No platform found for path prefix: {path_prefix}")
|
|
|
|
# Method 3: Default lookup
|
|
if context.get("detection_method") == "default":
|
|
platform = (
|
|
db.query(Platform)
|
|
.filter(Platform.code == MAIN_PLATFORM_CODE)
|
|
.filter(Platform.is_active.is_(True))
|
|
.first()
|
|
)
|
|
|
|
if platform:
|
|
logger.debug(
|
|
f"[PLATFORM] Default platform found: {platform.name}"
|
|
)
|
|
return platform
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def extract_clean_path(request: Request, platform_context: dict | None) -> str:
|
|
"""
|
|
Extract clean path without platform prefix for routing.
|
|
|
|
Downstream middleware (like StoreContextMiddleware) should use this
|
|
clean path for their detection logic.
|
|
"""
|
|
if not platform_context:
|
|
return request.url.path
|
|
|
|
# For path-based detection, use the pre-computed clean path
|
|
if platform_context.get("detection_method") == "path":
|
|
return platform_context.get("clean_path", request.url.path)
|
|
|
|
# For domain-based or default, path remains unchanged
|
|
return request.url.path
|
|
|
|
@staticmethod
|
|
def is_static_file_request(request: Request) -> bool:
|
|
"""Check if request is for static files."""
|
|
path = request.url.path.lower()
|
|
|
|
static_extensions = (
|
|
".ico", ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg",
|
|
".woff", ".woff2", ".ttf", ".eot", ".webp", ".map", ".json",
|
|
".xml", ".txt", ".pdf", ".webmanifest",
|
|
)
|
|
|
|
static_paths = ("/static/", "/media/", "/assets/", "/.well-known/")
|
|
|
|
if path.endswith(static_extensions):
|
|
return True
|
|
|
|
if any(path.startswith(static_path) for static_path in static_paths):
|
|
return True
|
|
|
|
return "favicon.ico" in path
|
|
|
|
|
|
class PlatformContextMiddleware:
|
|
"""
|
|
ASGI Middleware to inject platform context and rewrite URL paths.
|
|
|
|
This middleware:
|
|
1. Detects platform from domain (production) or path prefix (development)
|
|
2. Rewrites the URL path to remove platform prefix for routing
|
|
3. Stores platform info in request state for handlers
|
|
|
|
Runs BEFORE StoreContextMiddleware to establish platform context.
|
|
|
|
Sets in scope['state']:
|
|
platform: Platform object
|
|
platform_context: Detection metadata
|
|
platform_clean_path: Path without platform prefix
|
|
platform_original_path: Original path before rewrite
|
|
"""
|
|
|
|
def __init__(self, app):
|
|
self.app = app
|
|
|
|
async def __call__(self, scope, receive, send):
|
|
"""ASGI interface - allows path rewriting before routing."""
|
|
if scope["type"] != "http":
|
|
await self.app(scope, receive, send)
|
|
return
|
|
|
|
# Initialize state dict if not present
|
|
if "state" not in scope:
|
|
scope["state"] = {}
|
|
|
|
path = scope["path"]
|
|
host = ""
|
|
referer = ""
|
|
for header_name, header_value in scope.get("headers", []):
|
|
if header_name == b"host":
|
|
host = header_value.decode("utf-8")
|
|
elif header_name == b"referer":
|
|
referer = header_value.decode("utf-8")
|
|
|
|
# Skip for static files
|
|
if self._is_static_file(path):
|
|
scope["state"]["platform"] = None
|
|
scope["state"]["platform_context"] = None
|
|
scope["state"]["platform_clean_path"] = path
|
|
scope["state"]["platform_original_path"] = path
|
|
await self.app(scope, receive, send)
|
|
return
|
|
|
|
# Skip for system endpoints
|
|
if path in ["/health", "/docs", "/redoc", "/openapi.json"]:
|
|
scope["state"]["platform"] = None
|
|
scope["state"]["platform_context"] = None
|
|
scope["state"]["platform_clean_path"] = path
|
|
scope["state"]["platform_original_path"] = path
|
|
await self.app(scope, receive, send)
|
|
return
|
|
|
|
# Skip for admin requests
|
|
if FrontendDetector.is_admin(host, path):
|
|
scope["state"]["platform"] = None
|
|
scope["state"]["platform_context"] = None
|
|
scope["state"]["platform_clean_path"] = path
|
|
scope["state"]["platform_original_path"] = path
|
|
await self.app(scope, receive, send)
|
|
return
|
|
|
|
# Detect platform context
|
|
platform_context = self._detect_platform_context(path, host)
|
|
|
|
# For storefront API requests on localhost, the path doesn't contain
|
|
# /platforms/{code}/, so extract platform from the Referer header instead.
|
|
# e.g., Referer: http://localhost:8000/platforms/loyalty/storefront/FASHIONHUB/...
|
|
host_without_port = host.split(":")[0] if ":" in host else host
|
|
if (
|
|
host_without_port in _LOCAL_HOSTS
|
|
and path.startswith("/api/v1/storefront/")
|
|
and referer
|
|
and platform_context
|
|
and platform_context.get("detection_method") == "default"
|
|
):
|
|
referer_platform = self._extract_platform_from_referer(referer)
|
|
if referer_platform:
|
|
# Keep the original API path — don't rewrite to the Referer's path
|
|
referer_platform["clean_path"] = path
|
|
platform_context = referer_platform
|
|
|
|
if platform_context:
|
|
db_gen = get_db()
|
|
db = next(db_gen)
|
|
try:
|
|
platform = PlatformContextManager.get_platform_from_context(
|
|
db, platform_context
|
|
)
|
|
|
|
if platform:
|
|
clean_path = platform_context.get("clean_path", path)
|
|
|
|
# Store in scope state
|
|
scope["state"]["platform"] = platform
|
|
scope["state"]["platform_context"] = platform_context
|
|
scope["state"]["platform_clean_path"] = clean_path
|
|
scope["state"]["platform_original_path"] = path
|
|
|
|
# REWRITE THE PATH for routing
|
|
# This is the key: FastAPI will route based on this rewritten path
|
|
detection_method = platform_context.get("detection_method")
|
|
|
|
if detection_method == "path":
|
|
# Dev mode: strip /platforms/{code}/ prefix
|
|
scope["path"] = clean_path
|
|
if "raw_path" in scope:
|
|
scope["raw_path"] = clean_path.encode("utf-8")
|
|
|
|
# Prod mode: subdomain or custom store domain
|
|
# Prepend /storefront/ so routes registered at /storefront/ prefix match
|
|
is_storefront_domain = (
|
|
detection_method == "subdomain"
|
|
or platform_context.get("is_store_domain", False)
|
|
)
|
|
if is_storefront_domain:
|
|
_RESERVED = (
|
|
"/store/", "/admin/", "/api/", "/static/",
|
|
"/storefront/", "/health", "/docs", "/redoc",
|
|
"/media/", "/assets/",
|
|
)
|
|
if not any(clean_path.startswith(p) for p in _RESERVED):
|
|
new_path = "/storefront" + clean_path
|
|
scope["path"] = new_path
|
|
scope["state"]["platform_clean_path"] = new_path
|
|
if "raw_path" in scope:
|
|
scope["raw_path"] = new_path.encode("utf-8")
|
|
|
|
logger.debug(
|
|
f"[PLATFORM] Detected: {platform.code}, "
|
|
f"original={path}, routed={scope['path']}"
|
|
)
|
|
else:
|
|
# Platform code not found in database
|
|
scope["state"]["platform"] = None
|
|
scope["state"]["platform_context"] = None
|
|
scope["state"]["platform_clean_path"] = path
|
|
scope["state"]["platform_original_path"] = path
|
|
finally:
|
|
db.close()
|
|
else:
|
|
scope["state"]["platform"] = None
|
|
scope["state"]["platform_context"] = None
|
|
scope["state"]["platform_clean_path"] = path
|
|
scope["state"]["platform_original_path"] = path
|
|
|
|
await self.app(scope, receive, send)
|
|
|
|
def _detect_platform_context(self, path: str, host: str) -> dict | None:
|
|
"""
|
|
Detect platform from path or host.
|
|
|
|
URL Structure:
|
|
- / → Main marketing site ('main' platform)
|
|
- /about → Main marketing site about page
|
|
- /platforms/oms/ → OMS platform homepage
|
|
- /platforms/oms/pricing → OMS platform pricing
|
|
- /platforms/loyalty/ → Loyalty platform homepage
|
|
"""
|
|
host_without_port = host.split(":")[0] if ":" in host else host
|
|
|
|
# Method 1: Domain-based (production)
|
|
if host_without_port and host_without_port not in _LOCAL_HOSTS:
|
|
if "." in host_without_port:
|
|
parts = host_without_port.split(".")
|
|
if len(parts) == 2: # Root domain like omsflow.lu
|
|
return {
|
|
"domain": host_without_port,
|
|
"detection_method": "domain",
|
|
"host": host,
|
|
"original_path": path,
|
|
"clean_path": path,
|
|
}
|
|
if len(parts) >= 3: # Subdomain like wizatech.omsflow.lu
|
|
root_domain = ".".join(parts[-2:])
|
|
return {
|
|
"domain": root_domain,
|
|
"subdomain": parts[0],
|
|
"detection_method": "subdomain",
|
|
"host": host,
|
|
"original_path": path,
|
|
"clean_path": path,
|
|
}
|
|
|
|
# Method 2: Path-based (development) - ONLY for /platforms/ prefix
|
|
if path.startswith("/platforms/"):
|
|
# Extract: /platforms/oms/pricing → code="oms", clean_path="/pricing"
|
|
path_after_platforms = path[11:] # Remove "/platforms/"
|
|
parts = path_after_platforms.split("/", 1)
|
|
platform_code = parts[0].lower()
|
|
|
|
if platform_code:
|
|
clean_path = "/" + parts[1] if len(parts) > 1 and parts[1] else "/"
|
|
return {
|
|
"path_prefix": platform_code,
|
|
"detection_method": "path",
|
|
"host": host,
|
|
"original_path": path,
|
|
"clean_path": clean_path,
|
|
}
|
|
|
|
# Method 3: Default for localhost - serves main marketing site
|
|
# Store/storefront routes require explicit platform via /platforms/{code}/...
|
|
if host_without_port in _LOCAL_HOSTS:
|
|
if path.startswith(("/store/", "/stores/", "/storefront/")):
|
|
return None # No platform — require /platforms/{code}/ prefix
|
|
return {
|
|
"path_prefix": MAIN_PLATFORM_CODE,
|
|
"detection_method": "default",
|
|
"host": host,
|
|
"original_path": path,
|
|
"clean_path": path, # No path rewrite for main site
|
|
}
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def _extract_platform_from_referer(referer: str) -> dict | None:
|
|
"""
|
|
Extract platform context from Referer header.
|
|
|
|
Used for storefront API requests on localhost where the API path
|
|
doesn't contain /platforms/{code}/ but the Referer does.
|
|
|
|
e.g., Referer: http://localhost:8000/platforms/loyalty/storefront/FASHIONHUB/...
|
|
→ platform_code = "loyalty"
|
|
"""
|
|
try:
|
|
from urllib.parse import urlparse
|
|
|
|
parsed = urlparse(referer)
|
|
referer_path = parsed.path or ""
|
|
|
|
if referer_path.startswith("/platforms/"):
|
|
path_after = referer_path[11:] # Remove "/platforms/"
|
|
parts = path_after.split("/", 1)
|
|
platform_code = parts[0].lower()
|
|
if platform_code:
|
|
logger.debug(
|
|
f"[PLATFORM] Extracted platform from Referer: {platform_code}"
|
|
)
|
|
return {
|
|
"path_prefix": platform_code,
|
|
"detection_method": "path",
|
|
"host": parsed.hostname or "",
|
|
"original_path": referer_path,
|
|
"clean_path": "/" + parts[1] if len(parts) > 1 and parts[1] else "/",
|
|
}
|
|
except Exception as e:
|
|
logger.warning(f"[PLATFORM] Failed to extract platform from Referer: {e}")
|
|
|
|
return None
|
|
|
|
def _is_static_file(self, path: str) -> bool:
|
|
"""Check if path is for static files."""
|
|
path_lower = path.lower()
|
|
static_extensions = (
|
|
".ico", ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg",
|
|
".woff", ".woff2", ".ttf", ".eot", ".webp", ".map", ".json",
|
|
".xml", ".txt", ".pdf", ".webmanifest",
|
|
)
|
|
static_paths = ("/static/", "/media/", "/assets/", "/.well-known/")
|
|
|
|
if path_lower.endswith(static_extensions):
|
|
return True
|
|
if any(path_lower.startswith(p) for p in static_paths):
|
|
return True
|
|
return "favicon.ico" in path_lower
|
|
|
|
def get_current_platform(request: Request) -> Platform | None:
|
|
"""Helper function to get current platform from request state."""
|
|
return getattr(request.state, "platform", None)
|
|
|
|
|
|
def require_platform_context():
|
|
"""Dependency to require platform context in endpoints."""
|
|
|
|
def dependency(request: Request):
|
|
platform = get_current_platform(request)
|
|
if not platform:
|
|
from fastapi import HTTPException
|
|
raise HTTPException(
|
|
status_code=404,
|
|
detail="Platform not found"
|
|
)
|
|
return platform
|
|
|
|
return dependency
|