Samir Boulahtit
6735d99df2
Implement secure customer authentication system with dedicated JWT tokens, separate from admin/vendor authentication. Backend Changes: - Add customer JWT token support in deps.py - New get_current_customer_from_cookie_or_header dependency - Validates customer-specific tokens with type checking - Returns Customer object instead of User for shop routes - Extend AuthService with customer token support - Add verify_password() method - Add create_access_token_with_data() for custom token payloads - Update CustomerService authentication - Generate customer-specific JWT tokens with type="customer" - Use vendor-scoped customer lookup - Enhance exception handler - Sanitize validation errors to prevent password leaks in logs - Fix shop login redirect to support multi-access routing - Improve vendor context detection from Referer header - Consistent "path" detection method for cookie path logic Schema Changes: - Rename UserLogin.username to email_or_username for flexibility - Update field validators accordingly API Changes: - Update admin/vendor auth endpoints to use email_or_username - Customer auth already uses email field correctly Route Changes: - Update shop account routes to use Customer dependency - Add /account redirect (without trailing slash) - Change parameter names from current_user to current_customer Frontend Changes: - Update login forms to use email_or_username in API calls - Change button text from "Log in" to "Sign in" for consistency - Improve loading spinner layout with flexbox Security Improvements: - Customer tokens scoped to vendor_id - Token type validation prevents cross-context token usage - Password inputs redacted from validation error logs 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
423 lines
15 KiB
Python
423 lines
15 KiB
Python
# app/exceptions/handler.py
|
|
"""
|
|
Exception handler for FastAPI application.
|
|
|
|
Provides consistent error responses and logging for all custom exceptions.
|
|
This module provides classes and functions for:
|
|
- Unified exception handling for all application exceptions
|
|
- Consistent error response formatting
|
|
- Comprehensive logging with structured data
|
|
- Context-aware HTML error pages
|
|
"""
|
|
|
|
import logging
|
|
from typing import Union
|
|
|
|
from fastapi import Request, HTTPException
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse, RedirectResponse
|
|
|
|
from .base import WizamartException
|
|
from .error_renderer import ErrorPageRenderer
|
|
from middleware.context import RequestContext, get_request_context
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def setup_exception_handlers(app):
|
|
"""Setup exception handlers for the FastAPI app."""
|
|
|
|
@app.exception_handler(WizamartException)
|
|
async def custom_exception_handler(request: Request, exc: WizamartException):
|
|
"""Handle custom exceptions with context-aware rendering."""
|
|
|
|
# Special handling for 401 on HTML page requests (redirect to login)
|
|
if exc.status_code == 401 and _is_html_page_request(request):
|
|
logger.info(
|
|
f"401 on HTML page request - redirecting to login: {request.url.path}",
|
|
extra={
|
|
"path": request.url.path,
|
|
"accept": request.headers.get("accept", ""),
|
|
"method": request.method
|
|
}
|
|
)
|
|
|
|
# Redirect to appropriate login page based on context
|
|
return _redirect_to_login(request)
|
|
|
|
# Log the exception
|
|
logger.error(
|
|
f"Custom exception in {request.method} {request.url}: "
|
|
f"{exc.error_code} - {exc.message}",
|
|
extra={
|
|
"error_code": exc.error_code,
|
|
"status_code": exc.status_code,
|
|
"details": exc.details,
|
|
"url": str(request.url),
|
|
"method": request.method,
|
|
"exception_type": type(exc).__name__,
|
|
}
|
|
)
|
|
|
|
# Check if this is an API request
|
|
if _is_api_request(request):
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=exc.to_dict()
|
|
)
|
|
|
|
# Check if this is an HTML page request
|
|
if _is_html_page_request(request):
|
|
return ErrorPageRenderer.render_error_page(
|
|
request=request,
|
|
status_code=exc.status_code,
|
|
error_code=exc.error_code,
|
|
message=exc.message,
|
|
details=exc.details,
|
|
show_debug=True, # Will be filtered by ErrorPageRenderer based on user role
|
|
)
|
|
|
|
# Default to JSON for unknown request types
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content=exc.to_dict()
|
|
)
|
|
|
|
@app.exception_handler(HTTPException)
|
|
async def http_exception_handler(request: Request, exc: HTTPException):
|
|
"""Handle FastAPI HTTPExceptions with consistent format."""
|
|
|
|
logger.error(
|
|
f"HTTP exception in {request.method} {request.url}: "
|
|
f"{exc.status_code} - {exc.detail}",
|
|
extra={
|
|
"status_code": exc.status_code,
|
|
"detail": exc.detail,
|
|
"url": str(request.url),
|
|
"method": request.method,
|
|
"exception_type": "HTTPException",
|
|
}
|
|
)
|
|
|
|
# Check if this is an API request
|
|
if _is_api_request(request):
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={
|
|
"error_code": f"HTTP_{exc.status_code}",
|
|
"message": exc.detail,
|
|
"status_code": exc.status_code,
|
|
}
|
|
)
|
|
|
|
# Check if this is an HTML page request
|
|
if _is_html_page_request(request):
|
|
return ErrorPageRenderer.render_error_page(
|
|
request=request,
|
|
status_code=exc.status_code,
|
|
error_code=f"HTTP_{exc.status_code}",
|
|
message=exc.detail,
|
|
details={},
|
|
show_debug=True,
|
|
)
|
|
|
|
# Default to JSON
|
|
return JSONResponse(
|
|
status_code=exc.status_code,
|
|
content={
|
|
"error_code": f"HTTP_{exc.status_code}",
|
|
"message": exc.detail,
|
|
"status_code": exc.status_code,
|
|
}
|
|
)
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def validation_exception_handler(request: Request, exc: RequestValidationError):
|
|
"""Handle Pydantic validation errors with consistent format."""
|
|
|
|
# Sanitize errors to remove sensitive data from logs
|
|
sanitized_errors = []
|
|
for error in exc.errors():
|
|
sanitized_error = error.copy()
|
|
# Remove 'input' field which may contain passwords
|
|
if 'input' in sanitized_error:
|
|
sanitized_error['input'] = '<redacted>'
|
|
sanitized_errors.append(sanitized_error)
|
|
|
|
logger.error(
|
|
f"Validation error in {request.method} {request.url}: {len(sanitized_errors)} validation error(s)",
|
|
extra={
|
|
"validation_errors": sanitized_errors,
|
|
"url": str(request.url),
|
|
"method": request.method,
|
|
"exception_type": "RequestValidationError",
|
|
}
|
|
)
|
|
|
|
# Clean up validation errors to ensure JSON serializability
|
|
clean_errors = []
|
|
for error in exc.errors():
|
|
clean_error = {}
|
|
for key, value in error.items():
|
|
if key == 'input' and isinstance(value, bytes):
|
|
# Convert bytes to string representation for JSON serialization
|
|
clean_error[key] = f"<bytes: {len(value)} bytes>"
|
|
elif key == 'ctx' and isinstance(value, dict):
|
|
# Handle the 'ctx' field that contains ValueError objects
|
|
clean_ctx = {}
|
|
for ctx_key, ctx_value in value.items():
|
|
if isinstance(ctx_value, Exception):
|
|
clean_ctx[ctx_key] = str(ctx_value) # Convert exception to string
|
|
else:
|
|
clean_ctx[ctx_key] = ctx_value
|
|
clean_error[key] = clean_ctx
|
|
elif isinstance(value, bytes):
|
|
# Handle any other bytes objects
|
|
clean_error[key] = f"<bytes: {len(value)} bytes>"
|
|
else:
|
|
clean_error[key] = value
|
|
clean_errors.append(clean_error)
|
|
|
|
# Check if this is an API request
|
|
if _is_api_request(request):
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content={
|
|
"error_code": "VALIDATION_ERROR",
|
|
"message": "Request validation failed",
|
|
"status_code": 422,
|
|
"details": {
|
|
"validation_errors": clean_errors
|
|
}
|
|
}
|
|
)
|
|
|
|
# Check if this is an HTML page request
|
|
if _is_html_page_request(request):
|
|
return ErrorPageRenderer.render_error_page(
|
|
request=request,
|
|
status_code=422,
|
|
error_code="VALIDATION_ERROR",
|
|
message="Request validation failed",
|
|
details={"validation_errors": clean_errors},
|
|
show_debug=True,
|
|
)
|
|
|
|
# Default to JSON
|
|
return JSONResponse(
|
|
status_code=422,
|
|
content={
|
|
"error_code": "VALIDATION_ERROR",
|
|
"message": "Request validation failed",
|
|
"status_code": 422,
|
|
"details": {
|
|
"validation_errors": clean_errors
|
|
}
|
|
}
|
|
)
|
|
|
|
@app.exception_handler(Exception)
|
|
async def generic_exception_handler(request: Request, exc: Exception):
|
|
"""Handle unexpected exceptions."""
|
|
|
|
logger.error(
|
|
f"Unexpected exception in {request.method} {request.url}: {str(exc)}",
|
|
exc_info=True,
|
|
extra={
|
|
"url": str(request.url),
|
|
"method": request.method,
|
|
"exception_type": type(exc).__name__,
|
|
}
|
|
)
|
|
|
|
# Check if this is an API request
|
|
if _is_api_request(request):
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={
|
|
"error_code": "INTERNAL_SERVER_ERROR",
|
|
"message": "Internal server error",
|
|
"status_code": 500,
|
|
}
|
|
)
|
|
|
|
# Check if this is an HTML page request
|
|
if _is_html_page_request(request):
|
|
return ErrorPageRenderer.render_error_page(
|
|
request=request,
|
|
status_code=500,
|
|
error_code="INTERNAL_SERVER_ERROR",
|
|
message="Internal server error",
|
|
details={},
|
|
show_debug=True,
|
|
)
|
|
|
|
# Default to JSON
|
|
return JSONResponse(
|
|
status_code=500,
|
|
content={
|
|
"error_code": "INTERNAL_SERVER_ERROR",
|
|
"message": "Internal server error",
|
|
"status_code": 500,
|
|
}
|
|
)
|
|
|
|
@app.exception_handler(404)
|
|
async def not_found_handler(request: Request, exc):
|
|
"""Handle all 404 errors with consistent format."""
|
|
logger.warning(f"404 Not Found: {request.method} {request.url}")
|
|
|
|
# Check if this is an API request
|
|
if _is_api_request(request):
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={
|
|
"error_code": "ENDPOINT_NOT_FOUND",
|
|
"message": f"Endpoint not found: {request.url.path}",
|
|
"status_code": 404,
|
|
"details": {
|
|
"path": request.url.path,
|
|
"method": request.method
|
|
}
|
|
}
|
|
)
|
|
|
|
# Check if this is an HTML page request
|
|
if _is_html_page_request(request):
|
|
return ErrorPageRenderer.render_error_page(
|
|
request=request,
|
|
status_code=404,
|
|
error_code="ENDPOINT_NOT_FOUND",
|
|
message=f"The page you're looking for doesn't exist or has been moved.",
|
|
details={"path": request.url.path},
|
|
show_debug=True,
|
|
)
|
|
|
|
# Default to JSON
|
|
return JSONResponse(
|
|
status_code=404,
|
|
content={
|
|
"error_code": "ENDPOINT_NOT_FOUND",
|
|
"message": f"Endpoint not found: {request.url.path}",
|
|
"status_code": 404,
|
|
"details": {
|
|
"path": request.url.path,
|
|
"method": request.method
|
|
}
|
|
}
|
|
)
|
|
|
|
|
|
def _is_api_request(request: Request) -> bool:
|
|
"""
|
|
Check if the request is for an API endpoint.
|
|
|
|
API requests ALWAYS return JSON.
|
|
"""
|
|
return request.url.path.startswith("/api/")
|
|
|
|
|
|
def _is_html_page_request(request: Request) -> bool:
|
|
"""
|
|
Check if the request is for an HTML page (not an API endpoint).
|
|
|
|
More precise detection:
|
|
- Must NOT have /api/ in path
|
|
- Must be GET request
|
|
- Must explicitly accept text/html
|
|
- Must not already be on login page
|
|
"""
|
|
logger.debug(
|
|
f"Checking if HTML page request: {request.url.path}",
|
|
extra={
|
|
"path": request.url.path,
|
|
"method": request.method,
|
|
"accept": request.headers.get("accept", "")
|
|
}
|
|
)
|
|
|
|
# Don't redirect API calls
|
|
if _is_api_request(request):
|
|
logger.debug("Not HTML page: API endpoint")
|
|
return False
|
|
|
|
# Don't redirect if already on login page
|
|
if request.url.path.endswith("/login"):
|
|
logger.debug("Not HTML page: Already on login page")
|
|
return False
|
|
|
|
# Only redirect GET requests (page loads)
|
|
if request.method != "GET":
|
|
logger.debug(f"Not HTML page: Method is {request.method}, not GET")
|
|
return False
|
|
|
|
# MUST explicitly accept HTML (strict check)
|
|
accept_header = request.headers.get("accept", "")
|
|
if "text/html" not in accept_header:
|
|
logger.debug(f"Not HTML page: Accept header doesn't include text/html: {accept_header}")
|
|
return False
|
|
|
|
logger.debug("IS HTML page request")
|
|
return True
|
|
|
|
|
|
def _redirect_to_login(request: Request) -> RedirectResponse:
|
|
"""
|
|
Redirect to appropriate login page based on request context.
|
|
|
|
Uses context detection to determine admin vs vendor vs shop login.
|
|
Properly handles multi-access routing (domain, subdomain, path-based).
|
|
"""
|
|
context_type = get_request_context(request)
|
|
|
|
if context_type == RequestContext.ADMIN:
|
|
logger.debug("Redirecting to /admin/login")
|
|
return RedirectResponse(url="/admin/login", status_code=302)
|
|
elif context_type == RequestContext.VENDOR_DASHBOARD:
|
|
logger.debug("Redirecting to /vendor/login")
|
|
return RedirectResponse(url="/vendor/login", status_code=302)
|
|
elif context_type == RequestContext.SHOP:
|
|
# For shop context, redirect to shop login (customer login)
|
|
# Calculate base_url for proper routing (supports domain, subdomain, and path-based access)
|
|
vendor = getattr(request.state, 'vendor', None)
|
|
vendor_context = getattr(request.state, 'vendor_context', None)
|
|
access_method = vendor_context.get('detection_method', 'unknown') if vendor_context else 'unknown'
|
|
|
|
base_url = "/"
|
|
if access_method == "path" and vendor:
|
|
full_prefix = vendor_context.get('full_prefix', '/vendor/') if vendor_context else '/vendor/'
|
|
base_url = f"{full_prefix}{vendor.subdomain}/"
|
|
|
|
login_url = f"{base_url}shop/account/login"
|
|
logger.debug(f"Redirecting to {login_url}")
|
|
return RedirectResponse(url=login_url, status_code=302)
|
|
else:
|
|
# Fallback to root for unknown contexts
|
|
logger.debug("Unknown context, redirecting to /")
|
|
return RedirectResponse(url="/", status_code=302)
|
|
|
|
|
|
# Utility functions for common exception scenarios
|
|
def raise_not_found(resource_type: str, identifier: str) -> None:
|
|
"""Convenience function to raise ResourceNotFoundException."""
|
|
from .base import ResourceNotFoundException
|
|
raise ResourceNotFoundException(resource_type, identifier)
|
|
|
|
|
|
def raise_validation_error(message: str, field: str = None, details: dict = None) -> None:
|
|
"""Convenience function to raise ValidationException."""
|
|
from .base import ValidationException
|
|
raise ValidationException(message, field, details)
|
|
|
|
|
|
def raise_auth_error(message: str = "Authentication failed") -> None:
|
|
"""Convenience function to raise AuthenticationException."""
|
|
from .base import AuthenticationException
|
|
raise AuthenticationException(message)
|
|
|
|
|
|
def raise_permission_error(message: str = "Access denied") -> None:
|
|
"""Convenience function to raise AuthorizationException."""
|
|
from .base import AuthorizationException
|
|
raise AuthorizationException(message)
|