fix: platform admin authentication and UserContext completeness

Issues fixed:
- Platform selection returned LoginResponse requiring user timestamps,
  but UserContext doesn't have created_at/updated_at. Created dedicated
  PlatformSelectResponse that returns only token and platform info.

- UserContext was missing platform context fields (token_platform_id,
  token_platform_code). JWT token included them but they weren't
  extracted into UserContext, causing fallback warnings.

- admin_menu_config.py accessed admin_platforms (SQLAlchemy relationship)
  on UserContext (Pydantic schema). Changed to use accessible_platform_ids.

- Static file mount order in main.py caused 404 for locale files.
  More specific paths (/static/modules/X/locales) must be mounted
  before less specific paths (/static/modules/X).

Changes:
- models/schema/auth.py: Add PlatformSelectResponse, token_platform_id,
  token_platform_code, can_access_platform(), get_accessible_platform_ids()
- admin_auth.py: Use PlatformSelectResponse for select-platform endpoint
- admin_platform_service.py: Accept User | UserContext in validation
- admin_menu_config.py: Use accessible_platform_ids instead of admin_platforms
- main.py: Mount locales before static for correct path priority

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-02-02 22:31:35 +01:00
parent 9a0dd84035
commit b935592430
6 changed files with 344 additions and 20 deletions

View File

@@ -431,14 +431,14 @@ async def get_rendered_admin_menu(
else: else:
# Platform admin: use platform-level config # Platform admin: use platform-level config
# Get the selected platform from the JWT token # Get the selected platform from the JWT token
platform_id = getattr(current_user, "token_platform_id", None) platform_id = current_user.token_platform_id
# Fallback to first platform if no platform in token (shouldn't happen) # Fallback to first accessible platform if no platform in token (shouldn't happen)
if platform_id is None and current_user.admin_platforms: if platform_id is None and current_user.accessible_platform_ids:
platform_id = current_user.admin_platforms[0].id platform_id = current_user.accessible_platform_ids[0]
logger.warning( logger.warning(
f"[MENU_CONFIG] No platform_id in token for {current_user.email}, " f"[MENU_CONFIG] No platform_id in token for {current_user.email}, "
f"falling back to first platform: {platform_id}" f"falling back to first accessible platform: {platform_id}"
) )
menu = menu_service.get_menu_for_rendering( menu = menu_service.get_menu_for_rendering(

View File

@@ -23,7 +23,7 @@ from app.modules.core.services.auth_service import auth_service
from middleware.auth import AuthManager from middleware.auth import AuthManager
from app.modules.tenancy.models import Platform # noqa: API-007 - Admin needs to query platforms from app.modules.tenancy.models import Platform # noqa: API-007 - Admin needs to query platforms
from models.schema.auth import UserContext from models.schema.auth import UserContext
from models.schema.auth import LoginResponse, LogoutResponse, UserLogin, UserResponse from models.schema.auth import LoginResponse, LogoutResponse, PlatformSelectResponse, UserLogin, UserResponse
admin_auth_router = APIRouter(prefix="/auth") admin_auth_router = APIRouter(prefix="/auth")
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -160,7 +160,7 @@ def get_accessible_platforms(
} }
@admin_auth_router.post("/select-platform") @admin_auth_router.post("/select-platform", response_model=PlatformSelectResponse)
def select_platform( def select_platform(
platform_id: int, platform_id: int,
response: Response, response: Response,
@@ -177,7 +177,7 @@ def select_platform(
platform_id: Platform ID to select platform_id: Platform ID to select
Returns: Returns:
LoginResponse with new token containing platform context PlatformSelectResponse with new token and platform info
""" """
if current_user.is_super_admin: if current_user.is_super_admin:
raise InvalidCredentialsException( raise InvalidCredentialsException(
@@ -213,9 +213,10 @@ def select_platform(
logger.info(f"Admin {current_user.username} selected platform {platform.code}") logger.info(f"Admin {current_user.username} selected platform {platform.code}")
return LoginResponse( return PlatformSelectResponse(
access_token=token_data["access_token"], access_token=token_data["access_token"],
token_type=token_data["token_type"], token_type=token_data["token_type"],
expires_in=token_data["expires_in"], expires_in=token_data["expires_in"],
user=current_user, platform_id=platform.id,
platform_code=platform.code,
) )

View File

@@ -23,6 +23,7 @@ from app.modules.tenancy.exceptions import (
from app.modules.tenancy.models import AdminPlatform from app.modules.tenancy.models import AdminPlatform
from app.modules.tenancy.models import Platform from app.modules.tenancy.models import Platform
from app.modules.tenancy.models import User from app.modules.tenancy.models import User
from models.schema.auth import UserContext
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@@ -223,14 +224,14 @@ class AdminPlatformService:
def validate_admin_platform_access( def validate_admin_platform_access(
self, self,
user: User, user: User | UserContext,
platform_id: int, platform_id: int,
) -> None: ) -> None:
""" """
Validate that an admin has access to a platform. Validate that an admin has access to a platform.
Args: Args:
user: User object user: User model or UserContext schema (both have can_access_platform)
platform_id: Platform ID to check platform_id: Platform ID to check
Raises: Raises:

View File

@@ -0,0 +1,280 @@
# Middleware Frontend Detection - Problem Statement
**Date**: 2026-02-02
**Status**: Pending
**Priority**: Medium
---
## Context
During the fix for admin API authentication (where `/api/v1/admin/*` routes returned 401), we identified architectural issues in how the middleware detects frontend context (admin/vendor/storefront/platform).
The immediate authentication issue was fixed by making `FrontendType` mandatory in `require_module_access()`. However, the middleware still has design issues that should be addressed.
---
## Current State
### Middleware Files Involved
- `middleware/platform_context.py` - Detects platform from host/domain/path
- `middleware/vendor_context.py` - Detects vendor from subdomain/domain/path
### Current Detection Logic
```python
# In both PlatformContextManager and VendorContextManager
def is_admin_request(request: Request) -> bool:
host = request.headers.get("host", "")
path = request.url.path
# Production: domain-based
if host.startswith("admin."):
return True
# Development: path-based
return path.startswith("/admin")
```
### Routing Modes
**Development (localhost, path-based)**:
```
localhost:9999/admin/* → Admin pages
localhost:9999/api/v1/admin/* → Admin API
localhost:9999/vendor/* → Vendor pages
localhost:9999/api/v1/vendor/* → Vendor API
localhost:9999/* → Platform/storefront
```
**Production (domain/subdomain-based)**:
```
admin.platform.com/* → Admin (all paths)
vendor.platform.com/* → Vendor portal
shop.mystore.com/* → Vendor custom domain
api.platform.com/v1/* → Shared API domain (?)
platform.com/* → Marketing/platform pages
```
---
## Problems Identified
### 1. Incomplete Path Detection for Development Mode
The middleware only checks `/admin` but not `/api/v1/admin/*`:
```python
return path.startswith("/admin") # Misses /api/v1/admin/*
```
**Impact**: In development, API routes like `/api/v1/admin/messages/unread-count` are not recognized as admin requests, causing incorrect context detection.
**Note**: This doesn't break authentication anymore (fixed via `require_module_access`), but may affect context detection (vendor/platform context might be incorrectly applied to admin API routes).
### 2. Code Duplication
Same `is_admin_request` logic exists in 3 places:
- `PlatformContextManager.is_admin_request()` (static method)
- `PlatformContextMiddleware._is_admin_request()` (instance method)
- `VendorContextManager.is_admin_request()` (static method)
### 3. Hardcoded Paths
Path patterns are hardcoded in multiple locations:
- Middleware: `/admin`, `/vendor`
- Routes discovery: `/api/v1/admin`, `/api/v1/vendor` (in `app/modules/routes.py`)
- API main: `/v1/admin`, `/v1/vendor` (in `app/api/main.py`)
### 4. No Single Source of Truth
There's no centralized configuration that defines:
- What domains/subdomains map to which frontend
- What path patterns map to which frontend
- Whether we're in dev mode (path-based) or prod mode (domain-based)
### 5. Incomplete Frontend Coverage
Only `is_admin_request()` exists. No equivalent methods for:
- `is_vendor_request()`
- `is_storefront_request()`
- `is_platform_request()`
### 6. Not Using FrontendType Enum
Middleware returns `bool` instead of using the `FrontendType` enum that exists in `app/modules/enums.py`.
---
## Production Deployment Scenarios to Consider
### Scenario A: Subdomain per Frontend
```
admin.platform.com → Admin
vendor.platform.com → Vendor portal
*.platform.com → Vendor shops (wildcard subdomain)
platform.com → Marketing site
```
### Scenario B: Shared Domain with Path Routing
```
platform.com/admin/* → Admin
platform.com/vendor/* → Vendor
platform.com/api/v1/admin/* → Admin API
platform.com/* → Marketing/storefront
```
### Scenario C: Separate API Domain
```
admin.platform.com/* → Admin pages
api.platform.com/v1/admin/* → Admin API (different domain!)
```
**Issue**: Host is `api.platform.com`, path is `/v1/admin/*` (no `/api` prefix)
### Scenario D: Multi-Platform (current architecture)
```
oms.platform.com/* → OMS platform
loyalty.platform.com/* → Loyalty platform
admin.platform.com/* → Global admin for all platforms
```
---
## Proposed Solution Options
### Option A: Centralized FrontendDetector
Create a single utility class that handles all frontend detection:
```python
# app/core/frontend_detector.py
from app.core.config import settings
from app.modules.enums import FrontendType
class FrontendDetector:
"""Centralized frontend detection for both dev and prod modes."""
# Configurable patterns
ADMIN_SUBDOMAINS = ["admin"]
VENDOR_SUBDOMAINS = ["vendor", "portal"]
@classmethod
def detect(cls, host: str, path: str) -> FrontendType | None:
"""
Detect frontend type from request host and path.
Priority:
1. Domain/subdomain check (production)
2. Path prefix check (development)
"""
host_without_port = host.split(":")[0] if ":" in host else host
# Production: subdomain-based
subdomain = cls._get_subdomain(host_without_port)
if subdomain:
if subdomain in cls.ADMIN_SUBDOMAINS:
return FrontendType.ADMIN
if subdomain in cls.VENDOR_SUBDOMAINS:
return FrontendType.VENDOR
# Development: path-based
return cls._detect_from_path(path)
@classmethod
def _detect_from_path(cls, path: str) -> FrontendType | None:
# Check both page routes and API routes
admin_patterns = ["/admin", f"{settings.API_PREFIX}/admin"]
vendor_patterns = ["/vendor", f"{settings.API_PREFIX}/vendor"]
storefront_patterns = [f"{settings.API_PREFIX}/storefront"]
platform_patterns = [f"{settings.API_PREFIX}/platform"]
for pattern in admin_patterns:
if path.startswith(pattern):
return FrontendType.ADMIN
# ... etc
```
### Option B: Configuration-Driven Detection
Define all patterns in settings:
```python
# app/core/config.py
class Settings:
API_PREFIX: str = "/api/v1"
FRONTEND_DETECTION = {
FrontendType.ADMIN: {
"subdomains": ["admin"],
"path_prefixes": ["/admin"], # API prefix added automatically
},
FrontendType.VENDOR: {
"subdomains": ["vendor", "portal"],
"path_prefixes": ["/vendor"],
},
# ...
}
```
### Option C: Route-Level State Setting
Have the router set `request.state.frontend_type` when the route matches, eliminating detection entirely:
```python
# In route discovery or middleware
@app.middleware("http")
async def set_frontend_type(request: Request, call_next):
# Determine from matched route's tags or prefix
if request.scope.get("route"):
route = request.scope["route"]
if "admin" in route.tags:
request.state.frontend_type = FrontendType.ADMIN
return await call_next(request)
```
---
## Questions to Answer Before Implementation
1. **What's the production deployment model?**
- Subdomains per frontend?
- Separate API domain?
- Path-based on shared domain?
2. **Should detection be configurable?**
- Environment-specific patterns?
- Runtime configuration vs build-time?
3. **What does the middleware actually need?**
- `PlatformContextMiddleware`: Needs to know "is this NOT a platform-specific request?"
- `VendorContextMiddleware`: Needs to know "should I detect vendor context?"
- Maybe they just need `is_global_admin_request()` not full frontend detection
4. **API domain considerations?**
- Will API be on separate domain in production?
- If so, what's the path structure (`/v1/admin` vs `/api/v1/admin`)?
---
## Files to Modify (when implementing)
- `app/core/config.py` - Add centralized path/domain configuration
- `app/core/frontend_detector.py` - New centralized detection utility
- `middleware/platform_context.py` - Use centralized detector
- `middleware/vendor_context.py` - Use centralized detector
- `app/modules/routes.py` - Use centralized path configuration
---
## Related Commits
- `9a0dd84` - fix: make FrontendType mandatory in require_module_access
- `01e7602` - fix: add missing db argument to get_admin_context calls
---
## Next Steps
1. Answer the deployment model questions above
2. Choose solution option (A, B, or C)
3. Implement centralized detection
4. Update middleware to use centralized detection
5. Add tests for both dev and prod modes

18
main.py
View File

@@ -183,6 +183,7 @@ logger.info("=" * 80)
# Module static files must be mounted before the main /static mount # Module static files must be mounted before the main /static mount
# Mount module static files FIRST (self-contained modules) # Mount module static files FIRST (self-contained modules)
# NOTE: More specific paths must be mounted BEFORE less specific paths
MODULES_DIR = BASE_DIR / "app" / "modules" MODULES_DIR = BASE_DIR / "app" / "modules"
if MODULES_DIR.exists(): if MODULES_DIR.exists():
for module_dir in sorted(MODULES_DIR.iterdir()): for module_dir in sorted(MODULES_DIR.iterdir()):
@@ -191,20 +192,21 @@ if MODULES_DIR.exists():
module_name = module_dir.name module_name = module_dir.name
# Mount module static files # Mount module locale files FIRST (more specific path)
module_static = module_dir / "static" # Must come before static mount to avoid being intercepted
if module_static.exists():
mount_path = f"/static/modules/{module_name}"
app.mount(mount_path, StaticFiles(directory=str(module_static)), name=f"{module_name}_static")
logger.info(f"Mounted module static files: {mount_path} -> {module_static}")
# Mount module locale files (for JS i18n)
module_locales = module_dir / "locales" module_locales = module_dir / "locales"
if module_locales.exists(): if module_locales.exists():
mount_path = f"/static/modules/{module_name}/locales" mount_path = f"/static/modules/{module_name}/locales"
app.mount(mount_path, StaticFiles(directory=str(module_locales)), name=f"{module_name}_locales") app.mount(mount_path, StaticFiles(directory=str(module_locales)), name=f"{module_name}_locales")
logger.info(f"Mounted module locales: {mount_path} -> {module_locales}") logger.info(f"Mounted module locales: {mount_path} -> {module_locales}")
# Mount module static files (less specific path)
module_static = module_dir / "static"
if module_static.exists():
mount_path = f"/static/modules/{module_name}"
app.mount(mount_path, StaticFiles(directory=str(module_static)), name=f"{module_name}_static")
logger.info(f"Mounted module static files: {mount_path} -> {module_static}")
# Mount main static directory AFTER module statics # Mount main static directory AFTER module statics
if STATIC_DIR.exists(): if STATIC_DIR.exists():
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")

View File

@@ -39,6 +39,16 @@ class LoginResponse(BaseModel):
user: UserResponse user: UserResponse
class PlatformSelectResponse(BaseModel):
"""Response for platform selection (no user data - client already has it)."""
access_token: str
token_type: str = "bearer"
expires_in: int
platform_id: int
platform_code: str
class UserDetailResponse(UserResponse): class UserDetailResponse(UserResponse):
"""Extended user response with additional details.""" """Extended user response with additional details."""
@@ -198,6 +208,10 @@ class UserContext(BaseModel):
is_super_admin: bool = False is_super_admin: bool = False
accessible_platform_ids: list[int] | None = None # None = all platforms (super admin) accessible_platform_ids: list[int] | None = None # None = all platforms (super admin)
# Admin platform context (from JWT token after platform selection)
token_platform_id: int | None = None
token_platform_code: str | None = None
# Vendor-specific fields (from JWT token) # Vendor-specific fields (from JWT token)
token_vendor_id: int | None = None token_vendor_id: int | None = None
token_vendor_code: str | None = None token_vendor_code: str | None = None
@@ -227,6 +241,28 @@ class UserContext(BaseModel):
"""Check if user is a vendor.""" """Check if user is a vendor."""
return self.role == "vendor" return self.role == "vendor"
def can_access_platform(self, platform_id: int) -> bool:
"""
Check if user can access a specific platform.
Super admins (accessible_platform_ids=None) can access all platforms.
Platform admins can only access their assigned platforms.
"""
if self.is_super_admin:
return True
if self.accessible_platform_ids is None:
return True # Super admin fallback
return platform_id in self.accessible_platform_ids
def get_accessible_platform_ids(self) -> list[int] | None:
"""
Get list of platform IDs this user can access.
Returns None for super admins (all platforms accessible).
Returns list of platform IDs for platform admins.
"""
return self.accessible_platform_ids
@classmethod @classmethod
def from_user(cls, user, include_vendor_context: bool = True) -> "UserContext": def from_user(cls, user, include_vendor_context: bool = True) -> "UserContext":
""" """
@@ -262,6 +298,10 @@ class UserContext(BaseModel):
ap.platform_id for ap in admin_platforms if ap.is_active ap.platform_id for ap in admin_platforms if ap.is_active
] ]
# Add platform context from JWT token (for platform admins after selection)
data["token_platform_id"] = getattr(user, "token_platform_id", None)
data["token_platform_code"] = getattr(user, "token_platform_code", None)
# Add vendor context from JWT token if present # Add vendor context from JWT token if present
if include_vendor_context: if include_vendor_context:
data["token_vendor_id"] = getattr(user, "token_vendor_id", None) data["token_vendor_id"] = getattr(user, "token_vendor_id", None)