diff --git a/src/routers/keys.py b/src/routers/keys.py index c40b0ae7..69e91c10 100644 --- a/src/routers/keys.py +++ b/src/routers/keys.py @@ -1,9 +1,9 @@ import datetime import logging -import os from fastapi import APIRouter, Depends, Query +from src.config import settings from src.exceptions import DisabledException, ValidationException from src.security import ( JWTParams, @@ -13,8 +13,6 @@ from src.security import ( logger = logging.getLogger(__name__) -USE_AUTH = os.getenv("USE_AUTH", "False").lower() == "true" - router = APIRouter( prefix="/keys", tags=["keys"], @@ -35,7 +33,7 @@ async def create_key( expires_at: datetime.datetime | None = None, ): """Create a new Key""" - if not USE_AUTH: + if not settings.AUTH.USE_AUTH: raise DisabledException() # Validate that at least one parameter is provided for proper scoping diff --git a/src/security.py b/src/security.py index 0f182d49..66063d32 100644 --- a/src/security.py +++ b/src/security.py @@ -15,15 +15,6 @@ from .exceptions import AuthenticationException logger = logging.getLogger(__name__) -USE_AUTH = settings.AUTH.USE_AUTH -AUTH_JWT_SECRET = settings.AUTH.JWT_SECRET - -if USE_AUTH and not AUTH_JWT_SECRET: - print( - "\n ERROR: No JWT secret provided. Set the AUTH_JWT_SECRET environment variable.\n" - ) - exit(1) - security = HTTPBearer( auto_error=False, ) @@ -81,9 +72,9 @@ def create_admin_jwt() -> str: def create_jwt(params: JWTParams) -> str: """Create a JWT token from the given parameters.""" payload = {k: v for k, v in params.__dict__.items() if v is not None} - if not AUTH_JWT_SECRET: + if not settings.AUTH.JWT_SECRET: raise ValueError("AUTH_JWT_SECRET is not set, cannot create JWT.") - return jwt.encode(payload, AUTH_JWT_SECRET.encode("utf-8"), algorithm="HS256") + return jwt.encode(payload, settings.AUTH.JWT_SECRET.encode("utf-8"), algorithm="HS256") async def verify_jwt(token: str) -> JWTParams: @@ -91,10 +82,10 @@ async def verify_jwt(token: str) -> JWTParams: params = JWTParams() try: - if not AUTH_JWT_SECRET: + if not settings.AUTH.JWT_SECRET: raise ValueError("AUTH_JWT_SECRET is not set, cannot verify JWT.") decoded = jwt.decode( - token, AUTH_JWT_SECRET.encode("utf-8"), algorithms=["HS256"] + token, settings.AUTH.JWT_SECRET.encode("utf-8"), algorithms=["HS256"] ) if "t" in decoded: params.t = decoded["t"] @@ -180,7 +171,7 @@ async def auth( collection_id: Optional[str] = None, ) -> JWTParams: """Authenticate the given JWT and return the decoded parameters.""" - if not USE_AUTH: + if not settings.AUTH.USE_AUTH: return JWTParams(t="", ad=True) if not credentials or not credentials.credentials: logger.warning("No access token provided") diff --git a/tests/conftest.py b/tests/conftest.py index 78a2b8b8..1488cac7 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -54,9 +54,8 @@ CONNECTION_URI = make_url(DB_URI) TEST_DB_URL = CONNECTION_URI.set(database="test_db") DEFAULT_DB_URL = str(CONNECTION_URI.set(database="postgres")) -# Test API authorization -USE_AUTH = settings.AUTH.USE_AUTH -AUTH_JWT_SECRET = settings.AUTH.JWT_SECRET or "test-secret" +# Test API authorization - no longer needed as module-level constants +# We'll use settings.AUTH directly where needed def create_test_database(db_url): @@ -157,7 +156,7 @@ async def client(db_session): app.dependency_overrides[get_db] = override_get_db with TestClient(app) as c: - if USE_AUTH: + if settings.AUTH.USE_AUTH: # give the test client the admin JWT c.headers["Authorization"] = f"Bearer {create_admin_jwt()}" yield c @@ -181,11 +180,7 @@ def auth_client(client, request, monkeypatch): Always ensures USE_AUTH is set to True. """ # Ensure USE_AUTH is always True for this fixture - import src.routers.keys as keys_module - import src.security as security - - monkeypatch.setattr(keys_module, "USE_AUTH", "true") - monkeypatch.setattr(security, "USE_AUTH", "true") + monkeypatch.setattr(settings.AUTH, "USE_AUTH", True) # Clear any existing Authorization header client.headers.pop("Authorization", None)