Add whole-app password protection

src/resume_agent/webapp.py

@@ -1,15 +1,18 @@
 from __future__ import annotations
 
 import asyncio
+import base64
+import binascii
 import logging
 import os
+import secrets
 from collections.abc import Awaitable, Callable
 from pathlib import Path
 from typing import Annotated, Any, Literal
 from urllib.parse import urlparse
 from uuid import uuid4
 
-from fastapi import FastAPI, File, Form, HTTPException, UploadFile
+from fastapi import FastAPI, File, Form, HTTPException, Request, UploadFile
 from fastapi.concurrency import run_in_threadpool
 from fastapi.responses import FileResponse, PlainTextResponse
 from fastapi.staticfiles import StaticFiles
@@ -174,6 +177,32 @@ def _save_app_settings(settings_path: Path, settings: AppSettings) -> None:
     settings_path.write_text(settings.model_dump_json(indent=2), encoding="utf-8")
 
 
+def _valid_basic_credentials(
+    authorization: str | None,
+    expected_username: str,
+    expected_password: str,
+) -> bool:
+    if not authorization:
+        return False
+    scheme, separator, token = authorization.partition(" ")
+    if not separator or scheme.lower() != "basic":
+        return False
+    try:
+        decoded = base64.b64decode(token, validate=True).decode("utf-8")
+    except (binascii.Error, UnicodeDecodeError):
+        return False
+    username, separator, password = decoded.partition(":")
+    if not separator:
+        return False
+    username_matches = secrets.compare_digest(
+        username.encode("utf-8"), expected_username.encode("utf-8")
+    )
+    password_matches = secrets.compare_digest(
+        password.encode("utf-8"), expected_password.encode("utf-8")
+    )
+    return username_matches and password_matches
+
+
 def create_app(
     *,
     profile_path: Path = DEFAULT_PROFILE,
@@ -181,8 +210,53 @@ def create_app(
     llm_factory: Callable[[], OpenAILLM] = OpenAILLM,
     cv_dist: Path = DEFAULT_CV_DIST,
     settings_path: Path | None = None,
+    auth_username: str | None = None,
+    auth_password: str | None = None,
 ) -> FastAPI:
     web_app = FastAPI(title="Resume Agent", version="0.1.0")
+    resolved_auth_username = (
+        auth_username if auth_username is not None else os.getenv("RESUME_AGENT_USERNAME", "")
+    )
+    resolved_auth_password = (
+        auth_password if auth_password is not None else os.getenv("RESUME_AGENT_PASSWORD", "")
+    )
+    auth_configured = bool(
+        resolved_auth_username
+        and resolved_auth_password
+        and ":" not in resolved_auth_username
+        and not resolved_auth_username.lower().startswith("replace-with-")
+        and not resolved_auth_password.lower().startswith("replace-with-")
+    )
+
+    @web_app.middleware("http")
+    async def require_authentication(request: Request, call_next: Callable[..., Awaitable[Any]]):
+        if request.url.path == "/healthz":
+            return await call_next(request)
+        if not auth_configured:
+            return PlainTextResponse(
+                "Authentication is not configured. Set RESUME_AGENT_USERNAME and "
+                "RESUME_AGENT_PASSWORD.",
+                status_code=503,
+                headers={"Cache-Control": "no-store"},
+            )
+        if not _valid_basic_credentials(
+            request.headers.get("Authorization"),
+            resolved_auth_username,
+            resolved_auth_password,
+        ):
+            return PlainTextResponse(
+                "Authentication required.",
+                status_code=401,
+                headers={
+                    "Cache-Control": "no-store",
+                    "WWW-Authenticate": 'Basic realm="Resume Agent", charset="UTF-8"',
+                },
+            )
+        response = await call_next(request)
+        if request.url.path.startswith("/api/"):
+            response.headers["Cache-Control"] = "no-store"
+        return response
+
     web_app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
     editor_available = cv_dist.is_dir()
     if editor_available:
@@ -191,6 +265,12 @@ def create_app(
     task_records: dict[str, TaskStatusResponse] = {}
     active_tasks: set[asyncio.Task[None]] = set()
 
+    @web_app.get("/healthz", include_in_schema=False)
+    async def healthz() -> PlainTextResponse:
+        if not auth_configured:
+            return PlainTextResponse("authentication is not configured", status_code=503)
+        return PlainTextResponse("ok")
+
     async def run_task(task_id: str, operation: Awaitable[BaseModel]) -> None:
         TASK_LOGGER.info("task=%s event=start", task_id)
         try: