36 lines
1.0 KiB
Python
36 lines
1.0 KiB
Python
from fastapi import Request, Response
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
# Shared mutable state — updated by cloud_sync.py
|
|
license_state: dict = {
|
|
"licensed": True,
|
|
"locked": False,
|
|
"expires_at": None,
|
|
"last_sync": None,
|
|
"sync_failed": False,
|
|
}
|
|
|
|
EXEMPT_PATHS = {"/api/system/health"}
|
|
|
|
|
|
class LicenseCheckMiddleware(BaseHTTPMiddleware):
|
|
async def dispatch(self, request: Request, call_next):
|
|
if request.url.path in EXEMPT_PATHS:
|
|
return await call_next(request)
|
|
|
|
if license_state.get("locked"):
|
|
return Response(
|
|
content='{"detail": "System is locked by cloud administrator"}',
|
|
status_code=423,
|
|
media_type="application/json",
|
|
)
|
|
|
|
if not license_state.get("licensed", True):
|
|
return Response(
|
|
content='{"detail": "License expired or invalid"}',
|
|
status_code=402,
|
|
media_type="application/json",
|
|
)
|
|
|
|
return await call_next(request)
|