39 lines
1017 B
Python
39 lines
1017 B
Python
from fastapi import Request
|
|
from fastapi.responses import JSONResponse
|
|
from app.loader import load_app_config
|
|
|
|
CONFIG = load_app_config()
|
|
|
|
AUTH = CONFIG.get("auth", {})
|
|
AUTH_ENABLED = AUTH.get("enabled", False)
|
|
API_TOKEN = AUTH.get("token")
|
|
PUBLIC_PATHS = set(AUTH.get("public_paths", []))
|
|
|
|
|
|
async def auth_middleware(request: Request, call_next):
|
|
if not AUTH_ENABLED:
|
|
return await call_next(request)
|
|
|
|
path = request.url.path
|
|
|
|
if path in PUBLIC_PATHS:
|
|
return await call_next(request)
|
|
|
|
auth_header = request.headers.get("Authorization")
|
|
|
|
if not auth_header or not auth_header.startswith("Bearer "):
|
|
return JSONResponse(
|
|
status_code=401,
|
|
content={"message": "Missing or invalid Authorization header"},
|
|
)
|
|
|
|
token = auth_header.replace("Bearer ", "").strip()
|
|
|
|
if token != API_TOKEN:
|
|
return JSONResponse(
|
|
status_code=403,
|
|
content={"message": "Invalid token"},
|
|
)
|
|
|
|
return await call_next(request)
|