#!/usr/bin/env python3.12 import logging from fastapi import FastAPI, Response, HTTPException, Security from fastapi.security import APIKeyHeader, APIKeyQuery global api_key_query global api_key_header class Utilities: def __init__(self, app: FastAPI, constants): self.constants = constants self.blocked_response_status = 422 self.blocked_response_content = None self.api_key_query = APIKeyQuery(name=constants.API_KEY_NAME, auto_error=False) self.api_key_header = APIKeyHeader(name=f"x-{constants.API_KEY_NAME}", auto_error=False) def get_blocked_response(self, path: str | None = None): logging.error("Rejected request: Blocked") return Response(content=self.blocked_response_content, status_code=self.blocked_response_status) def get_no_endpoint_found(self, path: str | None = None): logging.error("Rejected request: No such endpoint") raise HTTPException(detail="Unknown endpoint", status_code=404) def check_key(self, path: str, key: str): """ Accepts path as an argument to allow fine tuning access for each API key, not currently in use. """ print(f"Testing with path: {path}, key: {key}") if not key or not key.startswith("Bearer "): return False key = key.split("Bearer ", maxsplit=1)[1].strip() if not key in self.constants.API_KEYS: print("Auth failed.") return False print("Auth succeeded") return True