add required bearer auth token

tailscalesd/main.py

@@ -3,13 +3,15 @@ import logging
 import os
 import sys
 from contextlib import asynccontextmanager
+from functools import lru_cache
 from ipaddress import ip_address
-from typing import Dict, List
+from typing import Annotated, Dict, List
 
 import httpx
 import json_logging  # type: ignore
 import uvicorn
-from fastapi import FastAPI
+from fastapi import Depends, FastAPI, HTTPException, status
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
 from prometheus_client import Counter
 from prometheus_fastapi_instrumentator import Instrumentator
 from pydantic import Field, SecretStr
@@ -48,14 +50,15 @@ class Settings(BaseSettings):
     port: int = 9242
     interval: int = 60
     tailnet: str = Field()
+    bearer_token: str = Field()
     api_key: SecretStr = Field()
+    test_mode: bool = False
 
 
-settings = Settings()  # type: ignore[call-arg]
 CACHE_SD = []
 
 
-async def tailscale_devices() -> List[Dict]:
+async def tailscale_devices(settings: Settings) -> List[Dict]:
     async with httpx.AsyncClient() as client:
         try:
             # https://github.com/tailscale/tailscale/blob/main/api.md#tailnet-devices-get
@@ -170,11 +173,13 @@ def plain_devices_sd(tailnet, devices) -> List[Dict]:
     return sd
 
 
-async def poll_sd():
+async def poll_sd(settings: Settings):
     global CACHE_SD
+    if settings.test_mode:
+        return
     while True:
         try:
-            devices = await tailscale_devices()
+            devices = await tailscale_devices(settings)
             device_targets = plain_devices_sd(settings.tailnet, devices)
             matrix_targets = await matrix_sd(settings.tailnet, devices)
             CACHE_SD = matrix_targets + device_targets
@@ -187,26 +192,52 @@ async def poll_sd():
             )
 
 
+@lru_cache
+def get_settings():
+    return Settings()  # type: ignore[call-arg]
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     instrumentator.expose(app)
-    asyncio.create_task(poll_sd())
+    settings = get_settings()
+    asyncio.create_task(poll_sd(settings))
     yield
 
 
 app = FastAPI(lifespan=lifespan)
+security = HTTPBearer()
 instrumentator = Instrumentator().instrument(app)
 
 json_logging.init_fastapi(enable_json=True)
 json_logging.init_request_instrument(app)
 
 
+async def is_authorized(
+    settings: Annotated[Settings, Depends(get_settings)],
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+):
+    if (
+        credentials.scheme != "Bearer"
+        or credentials.credentials != settings.bearer_token
+    ):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="Invalid authentication credentials",
+        )
+    return True
+
+
 @app.get("/")
-async def sd():
-    return CACHE_SD
+async def sd(
+    is_authed: bool = Depends(is_authorized),
+):
+    if is_authed:
+        return CACHE_SD
 
 
 def main():
+    settings = get_settings()
     uvicorn.run(app, host=settings.host, port=settings.port)