Return a 401/403 response when admin endpoints should be blocked. Admin endpoints (connections, providers, spend write, routing-config write, selector write) can mutate API keys and routing behavior. Two policies: * ``UNCOMMON_ROUTE_ADMIN_TOKEN`` is set: require ``Authorization: Bear
(request: Request)
| 164 | |
| 165 | |
| 166 | def _admin_auth_failure(request: Request) -> JSONResponse | None: |
| 167 | """Return a 401/403 response when admin endpoints should be blocked. |
| 168 | |
| 169 | Admin endpoints (connections, providers, spend write, routing-config write, selector write) |
| 170 | can mutate API keys and routing behavior. Two policies: |
| 171 | |
| 172 | * ``UNCOMMON_ROUTE_ADMIN_TOKEN`` is set: require ``Authorization: Bearer <token>``. |
| 173 | * Not set: only accept requests from a local client. |
| 174 | """ |
| 175 | token = os.environ.get(_ADMIN_ENV_VAR, "").strip() |
| 176 | if token: |
| 177 | hdr = request.headers.get("authorization", "") |
| 178 | if hdr.startswith("Bearer ") and hdr[7:].strip() == token: |
| 179 | return None |
| 180 | return JSONResponse({"error": f"admin token required (set {_ADMIN_ENV_VAR})"}, status_code=401) |
| 181 | client_host = getattr(request.client, "host", "") if request.client else "" |
| 182 | if client_host in _LOCAL_CLIENT_HOSTS: |
| 183 | return None |
| 184 | return JSONResponse( |
| 185 | { |
| 186 | "error": ( |
| 187 | "admin endpoints require a local client; " |
| 188 | f"set {_ADMIN_ENV_VAR} to allow remote access" |
| 189 | ) |
| 190 | }, |
| 191 | status_code=403, |
| 192 | ) |
| 193 | |
| 194 | |
| 195 | class _SpendReservation: |
no test coverage detected