wip add a bearer token on the current push endpoint

src/summary/summary/config.py

@@ -1,5 +1,9 @@
-"""Module for managing application configuration and settings."""
+"""Application configuration and settings."""
 
+from functools import lru_cache
+from typing import Annotated
+
+from fastapi import Depends
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -8,6 +12,7 @@ class Settings(BaseSettings):
 
     app_name: str = "Awesome API"
     model_config = SettingsConfigDict(env_file=".env")
+    app_api_token: str
 
     # Celery settings
     celery_broker_url: str = "redis://redis/0"
@@ -28,3 +33,12 @@ class Settings(BaseSettings):
     webhook_backoff_factor: float = 0.1
     webhook_api_token: str
     webhook_url: str
+
+
+@lru_cache
+def get_settings():
+    """Load and cache application settings."""
+    return Settings()
+
+
+SettingsDeps = Annotated[Settings, Depends(get_settings)]

src/summary/summary/main.py

@@ -1,26 +1,17 @@
-"""Application entry point."""
-
-from functools import lru_cache
-from typing import Annotated
+"""Application endpoint."""
 
 from fastapi import Depends, FastAPI
 from pydantic import BaseModel
 
 from .celery_worker import send_push_notification
-from .config import Settings
-
+from .config import SettingsDeps
+from .security import verify_token
 
 app = FastAPI()
 
 
-@lru_cache
-def get_settings():
-    """Load and cache application settings."""
-    return Settings()
-
-
 @app.get("/")
-async def root(settings: Annotated[Settings, Depends(get_settings)]):
+async def root(settings: SettingsDeps):
     """Root endpoint that returns app name."""
     return {"message": f"Hello World, using {settings.app_name}"}
 
@@ -46,7 +37,7 @@ class NotificationRequest(BaseModel):
 
 
 @app.post("/push")
-async def notify(request: NotificationRequest):
+async def notify(request: NotificationRequest, token: str = Depends(verify_token)):
     """Push a notification."""
     send_push_notification.delay(request.filename, request.email, request.sub)
     return {"message": "Notification sent"}

src/summary/summary/security.py

@@ -0,0 +1,19 @@
+"""Application security."""
+
+from fastapi import HTTPException, Security
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from .config import SettingsDeps
+
+security = HTTPBearer()
+
+
+def verify_token(
+    settings: SettingsDeps,
+    credentials: HTTPAuthorizationCredentials = Security(security),  # noqa: B008
+):
+    """Verify the bearer token from the Authorization header."""
+    token = credentials.credentials
+    if token != settings.app_api_token:
+        raise HTTPException(status_code=401, detail="Invalid token")
+    return token