add api-key header and swagger authentication

src/adapters/incoming/api_routes.py

@@ -18,7 +18,11 @@ from pathlib import Path
 from typing import Iterator, List, Optional
 
 from fastapi import APIRouter, Depends, FastAPI, File, Form, HTTPException, UploadFile, status
+from fastapi.openapi.docs import get_swagger_ui_html, get_redoc_html
 from fastapi.responses import JSONResponse
+from fastapi.security import HTTPBasicCredentials
+
+from .auth import check_docs_credentials, validate_api_key
 
 from ...core.config import get_settings
 from ...core.domain.exceptions import (
@@ -41,11 +45,6 @@ from .api_schemas import (
 
 logger = logging.getLogger(__name__)
 
-
-# =============================================================================
-# Application Setup
-# =============================================================================
-
 # Load settings
 settings = get_settings()
 
@@ -53,12 +52,19 @@ app = FastAPI(
     title="Text Processor API",
     description="Text extraction and chunking system using Hexagonal Architecture",
     version="1.0.0",
-    docs_url="/docs",
+    # docs_url=None,
-    redoc_url="/redoc",
+    # redoc_url=None,
 )
 
-router = APIRouter(prefix="/api/v1", tags=["Text Processing"])
+router = APIRouter(
+    prefix="/api/v1",
+    tags=["Text Processing"],
+    dependencies=[Depends(validate_api_key)]
+)
 
+public_router = APIRouter(
+    tags=["System"],
+)
 
 # =============================================================================
 # Global Exception Handler
@@ -339,7 +345,7 @@ async def process_file(
     )
 
 
-@router.get(
+@public_router.get(
     "/health",
     response_model=HealthCheckResponse,
     status_code=status.HTTP_200_OK,
@@ -356,21 +362,29 @@ async def health_check() -> HealthCheckResponse:
     )
 
 
+# =============================================================================
+# Protected Documentation Routes
+# =============================================================================
+
+@app.get("/docs", include_in_schema=False)
+def api_docs(_: HTTPBasicCredentials = Depends(check_docs_credentials)):
+    return get_swagger_ui_html(
+        openapi_url="/openapi.json",
+        title="Protected Text-Processor API Docs"
+    )
+
+
+@app.get("/redoc", include_in_schema=False)
+def api_docs(_: HTTPBasicCredentials = Depends(check_docs_credentials)):
+    return get_redoc_html(
+        openapi_url="/openapi.json",
+        title="Protected Text-Processor API Docs"
+    )
+
 # =============================================================================
 # Application Setup
 # =============================================================================
 
-# Include router in app
+# Include routers in app
 app.include_router(router)
-
+app.include_router(public_router)
-
-@app.get("/")
-async def root():
-    """Root endpoint with API information."""
-    return {
-        "name": "Text Processor API",
-        "version": "1.0.0",
-        "description": "Text extraction and chunking system using Hexagonal Architecture",
-        "docs_url": "/docs",
-        "api_prefix": "/api/v1",
-    }

src/adapters/incoming/auth.py

@@ -0,0 +1,34 @@
+import secrets
+from fastapi import Depends, HTTPException, Security, status
+from fastapi.security import APIKeyHeader, HTTPBasic, HTTPBasicCredentials
+from ...core.config import get_settings
+
+settings = get_settings()
+# This allows Swagger UI to detect the "Authorize" button
+api_key_header = APIKeyHeader(name=settings.API_KEY_NAME, auto_error=False)
+http_basic = HTTPBasic()
+
+async def validate_api_key(api_key: str = Security(api_key_header)):
+    """
+    Validates the X-API-Key header.
+    Using secrets.compare_digest protects against timing attacks.
+    """
+    if not api_key or not secrets.compare_digest(api_key, settings.API_KEY):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Could not validate credentials. Invalid or missing API Key.",
+        )
+    return api_key
+
+
+security = HTTPBasic()
+
+def check_docs_credentials(credentials: HTTPBasicCredentials = Depends(security)):
+    is_correct_user = secrets.compare_digest(credentials.username, settings.DOCS_USERNAME)
+    is_correct_password = secrets.compare_digest(credentials.password, settings.DOCS_PASSWORD)
+
+    if not (is_correct_user and is_correct_password):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            headers={"WWW-Authenticate": "Basic"},
+        )

src/core/config.py

@@ -14,6 +14,13 @@ class Settings(BaseSettings):
     S3_ENDPOINT_URL: Optional[str] = "https://cdn.d.aiengines.ir"
     S3_PRESIGNED_URL_EXPIRATION: int = 3600
     S3_UPLOAD_PATH_PREFIX: str = "extractions"
+
+    API_KEY: str = "some-secret-api-key"
+    API_KEY_NAME: str = "API-Key"
+
+    DOCS_USERNAME: str = "admin"
+    DOCS_PASSWORD: str = "admin"
+
     LOG_LEVEL: str = "INFO"
 
     model_config = SettingsConfigDict(