Add global access token setting and validation

docs/src/examples/code/tiler_with_auth.md

@@ -140,12 +140,12 @@ def DatasetPathParams(
     """Create dataset path from args"""
 
     if not api_key_query:
-        raise HTTPException(status_code=403, detail="Missing `access_token`")
+        raise HTTPException(status_code=401, detail="Missing `access_token`")
 
     try:
         AccessToken.from_string(api_key_query)
     except JWTError:
-        raise HTTPException(status_code=403, detail="Invalid `access_token`")
+        raise HTTPException(status_code=401, detail="Invalid `access_token`")
 
     return url
 ```

docs/src/intro.md

@@ -43,6 +43,7 @@ The default application can be customized using environment variables defined in
 - `DISABLE_STAC` (bool): disable `/stac` endpoints.
 - `DISABLE_MOSAIC` (bool): disable `/mosaic` endpoints.
 - `LOWER_CASE_QUERY_PARAMETERS` (bool): transform all query-parameters to lower case (see https://github.com/developmentseed/titiler/pull/321).
+- `GLOBAL_ACCESS_TOKEN` (str | None): a string which is required in the `?access_token=` query param with every request.
 
 ## Customized, minimal app
 
@@ -108,11 +109,11 @@ api_key_query = APIKeyQuery(name="access_token", auto_error=False)
 def token_validation(access_token: str = Security(api_key_query)):
     """stupid token validation."""
     if not access_token:
-        raise HTTPException(status_code=403, detail="Missing `access_token`")
+        raise HTTPException(status_code=401, detail="Missing `access_token`")
 
     # if access_token == `token` then OK
     if not access_token == "token":
-        raise HTTPException(status_code=403, detail="Invalid `access_token`")
+        raise HTTPException(status_code=401, detail="Invalid `access_token`")
 
     return True
 

src/titiler/application/titiler/application/main.py

@@ -3,7 +3,8 @@
 import logging
 
 import jinja2
-from fastapi import FastAPI
+from fastapi import Depends, FastAPI, HTTPException, Security
+from fastapi.security.api_key import APIKeyQuery
 from rio_tiler.io import STACReader
 from starlette.middleware.cors import CORSMiddleware
 from starlette.requests import Request
@@ -47,6 +48,30 @@
 
 api_settings = ApiSettings()
 
+###############################################################################
+# Setup a global API access key, if configured
+api_key_query = APIKeyQuery(name="access_token", auto_error=False)
+
+
+def validate_access_token(access_token: str = Security(api_key_query)):
+    """Validates API key access token, set as the `api_settings.global_access_token` value.
+    Returns True if no access token is required, or if the access token is valid.
+    Raises an HTTPException (401) if the access token is required but invalid/missing."""
+    if api_settings.global_access_token is None:
+        return True
+
+    if not access_token:
+        raise HTTPException(status_code=401, detail="Missing `access_token`")
+
+    # if access_token == `token` then OK
+    if access_token != api_settings.global_access_token:
+        raise HTTPException(status_code=401, detail="Invalid `access_token`")
+
+    return True
+
+
+###############################################################################
+
 app = FastAPI(
     title=api_settings.name,
     openapi_url="/api",
@@ -63,6 +88,7 @@
     """,
     version=titiler_version,
     root_path=api_settings.root_path,
+    dependencies=[Depends(validate_access_token)],
 )
 
 ###############################################################################
@@ -77,7 +103,11 @@
         ],
     )
 
-    app.include_router(cog.router, prefix="/cog", tags=["Cloud Optimized GeoTIFF"])
+    app.include_router(
+        cog.router,
+        prefix="/cog",
+        tags=["Cloud Optimized GeoTIFF"],
+    )
 
 
 ###############################################################################
@@ -92,24 +122,36 @@
     )
 
     app.include_router(
-        stac.router, prefix="/stac", tags=["SpatioTemporal Asset Catalog"]
+        stac.router,
+        prefix="/stac",
+        tags=["SpatioTemporal Asset Catalog"],
     )
 
 ###############################################################################
 # Mosaic endpoints
 if not api_settings.disable_mosaic:
     mosaic = MosaicTilerFactory(router_prefix="/mosaicjson")
-    app.include_router(mosaic.router, prefix="/mosaicjson", tags=["MosaicJSON"])
+    app.include_router(
+        mosaic.router,
+        prefix="/mosaicjson",
+        tags=["MosaicJSON"],
+    )
 
 ###############################################################################
 # TileMatrixSets endpoints
 tms = TMSFactory()
-app.include_router(tms.router, tags=["Tiling Schemes"])
+app.include_router(
+    tms.router,
+    tags=["Tiling Schemes"],
+)
 
 ###############################################################################
 # Algorithms endpoints
 algorithms = AlgorithmFactory()
-app.include_router(algorithms.router, tags=["Algorithms"])
+app.include_router(
+    algorithms.router,
+    tags=["Algorithms"],
+)
 
 add_exception_handlers(app, DEFAULT_STATUS_CODES)
 add_exception_handlers(app, MOSAIC_STATUS_CODES)

src/titiler/application/titiler/application/settings.py

@@ -1,5 +1,7 @@
 """Titiler API settings."""
 
+from typing import Union
+
 from pydantic import field_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
@@ -20,6 +22,9 @@ class ApiSettings(BaseSettings):
 
     lower_case_query_parameters: bool = False
 
+    # an API key required to access any endpoint, passed via the ?access_token= query parameter
+    global_access_token: Union[str, None] = None
+
     model_config = SettingsConfigDict(env_prefix="TITILER_API_", env_file=".env")
 
     @field_validator("cors_origins")