51 lines
1.9 KiB
Python
51 lines
1.9 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi import Request
|
|
from fastapi.responses import JSONResponse
|
|
from sqlmodel import Session
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
from bootstrap.auth_access import RouteAccessMode, extract_bot_id, resolve_route_access_mode
|
|
from core.database import engine
|
|
from services.platform_auth_service import (
|
|
resolve_bot_request_auth,
|
|
resolve_panel_request_auth,
|
|
)
|
|
|
|
|
|
def _unauthorized(detail: str) -> JSONResponse:
|
|
return JSONResponse(status_code=401, content={"detail": detail})
|
|
|
|
|
|
class PasswordProtectionMiddleware(BaseHTTPMiddleware):
|
|
async def dispatch(self, request: Request, call_next):
|
|
if request.method.upper() == "OPTIONS":
|
|
return await call_next(request)
|
|
|
|
path = request.url.path
|
|
access_mode = resolve_route_access_mode(path, request.method)
|
|
if access_mode == RouteAccessMode.PUBLIC:
|
|
return await call_next(request)
|
|
|
|
bot_id = extract_bot_id(path)
|
|
with Session(engine) as session:
|
|
panel_principal = resolve_panel_request_auth(session, request)
|
|
if panel_principal.authenticated:
|
|
request.state.auth_principal = panel_principal
|
|
return await call_next(request)
|
|
|
|
if access_mode == RouteAccessMode.PANEL_ONLY:
|
|
return _unauthorized("Panel authentication required")
|
|
|
|
if not bot_id:
|
|
return _unauthorized("Bot authentication required")
|
|
|
|
bot_principal = resolve_bot_request_auth(session, request, bot_id)
|
|
if bot_principal.authenticated:
|
|
request.state.auth_principal = bot_principal
|
|
return await call_next(request)
|
|
|
|
if access_mode == RouteAccessMode.PUBLIC_BOT_OR_PANEL:
|
|
return _unauthorized("Bot or panel authentication required to access this resource")
|
|
return _unauthorized("Bot or panel authentication required")
|