GET /api/system/identity (unauthenticated, license-exempt) returns
{app: 'xenia-pos', site_id, venue_name, version, api_version: 1} so a phone
can confirm which venue a server is when pairing and when rediscovering the
server after an IP change. No secrets: SITE_KEY never leaves the server.
/api/system/status now also returns site_id and lan_ip (HOST_IP only - in
Docker any auto-detected address is the unreachable bridge IP).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
from fastapi import Request, Response
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
# Shared mutable state — updated by cloud_sync.py
|
|
# Fields:
|
|
# licensed bool — False only after 72h offline OR after expiry grace passes
|
|
# locked bool — True once lock_pending is enforced at workday close
|
|
# lock_pending bool — Cloud requested lock; waiting for workday to close
|
|
# expires_at str|None — ISO timestamp from cloud
|
|
# days_until_expiry int|None — negative when expired
|
|
# grace_expires_at str|None — ISO timestamp of expiry + 5 days
|
|
# last_sync str|None — ISO timestamp of last successful heartbeat
|
|
# sync_failed bool
|
|
# latest_version str|None
|
|
license_state: dict = {
|
|
"licensed": True,
|
|
"locked": False,
|
|
"lock_pending": False,
|
|
"expires_at": None,
|
|
"days_until_expiry": None,
|
|
"grace_expires_at": None,
|
|
"last_sync": None,
|
|
"sync_failed": False,
|
|
"latest_version": None,
|
|
}
|
|
|
|
# Paths that bypass all license checks (health probe, venue identity for pairing)
|
|
EXEMPT_PATHS = {"/api/system/health", "/api/system/identity"}
|
|
|
|
# Paths that are always allowed so the frontend can read license status
|
|
# and managers can still log in / close the workday when restricted
|
|
STATUS_ALLOWED_PATHS = {
|
|
"/api/system/status",
|
|
"/api/system/sync-license",
|
|
"/api/auth/login",
|
|
"/api/auth/me",
|
|
"/api/business-day/current",
|
|
"/api/business-day/close",
|
|
}
|
|
|
|
|
|
class LicenseCheckMiddleware(BaseHTTPMiddleware):
|
|
async def dispatch(self, request: Request, call_next):
|
|
path = request.url.path
|
|
|
|
if path in EXEMPT_PATHS or path in STATUS_ALLOWED_PATHS:
|
|
return await call_next(request)
|
|
|
|
# Hard block: licensed=False means either 72h offline grace expired
|
|
# OR expiry grace period (5 days) has passed. In both cases the
|
|
# business_day router already prevented opening a new workday, so
|
|
# existing operations can still complete — we only block new ones.
|
|
# The business_day /open endpoint has its own detailed error message.
|
|
if not license_state.get("licensed", True):
|
|
return Response(
|
|
content='{"detail":"license_expired","code":"LICENSE_EXPIRED"}',
|
|
status_code=402,
|
|
media_type="application/json",
|
|
)
|
|
|
|
# Hard block: locked=True (lock_pending was enforced at workday close)
|
|
if license_state.get("locked"):
|
|
return Response(
|
|
content='{"detail":"system_locked","code":"SYSTEM_LOCKED"}',
|
|
status_code=423,
|
|
media_type="application/json",
|
|
)
|
|
|
|
return await call_next(request)
|