Authentication Model

class AuthMiddleware(BaseHTTPMiddleware):
    """Authenticate every request via Bearer token or session cookie."""

    # Paths that do not require authentication
    _PUBLIC_PATHS = frozenset({"/login", "/login/", "/health"})

    async def dispatch(
        self, request: Request, call_next: RequestResponseEndpoint
    ) -> Response:
        if request.url.path in self._PUBLIC_PATHS:
            return await call_next(request)

        settings = get_settings()
        user = None
        is_admin = False
        username = ""

        # 1. Check Authorization header (API clients)
        auth_header = request.headers.get("authorization", "")
        if auth_header.startswith("Bearer "):
            token = auth_header[7:]
            if token == settings.admin_key:
                is_admin = True
                username = "admin"
            else:
                user = await get_user_by_key(token)

        # 2. Check session cookie (browser) -- opaque session token
        if not user and not is_admin:
            session_token = request.cookies.get("docsfy_session")
            if session_token:
                session = await get_session(session_token)
                if session:
                    is_admin = bool(session["is_admin"])
                    username = str(session["username"])
                    # Fix 8: For DB users (not ADMIN_KEY admin), verify user still exists
                    if username != "admin":
                        user = await get_user_by_username(username)
                        if not user:
                            # User was deleted since session was created
                            if request.url.path.startswith("/api/"):
                                return JSONResponse(
                                    status_code=401, content={"detail": "Unauthorized"}
                                )
                            return RedirectResponse(url="/login", status_code=302)

        if not user and not is_admin:
            # Not authenticated
            if request.url.path.startswith("/api/"):
                return JSONResponse(status_code=401, content={"detail": "Unauthorized"})
            return RedirectResponse(url="/login", status_code=302)

def hash_api_key(key: str, hmac_secret: str = "") -> str:
    """Hash an API key with HMAC-SHA256 for storage.

    Uses ADMIN_KEY as the HMAC secret so that even if the source is read,
    keys cannot be cracked without the environment secret.
    """
    # NOTE: ADMIN_KEY is used as the HMAC secret. Rotating ADMIN_KEY will
    # invalidate all existing api_key_hash values, requiring all users to
    # regenerate their API keys.
    secret = hmac_secret or os.getenv("ADMIN_KEY", "")
    if not secret:
        msg = "ADMIN_KEY environment variable is required for key hashing"
        raise RuntimeError(msg)
    return hmac.new(secret.encode(), key.encode(), hashlib.sha256).hexdigest()


async def get_user_by_key(api_key: str) -> dict[str, str | int | None] | None:
    """Look up a user by their raw API key."""
    key_hash = hash_api_key(api_key)
    async with aiosqlite.connect(DB_PATH) as db:
        db.row_factory = aiosqlite.Row
        cursor = await db.execute(
            "SELECT * FROM users WHERE api_key_hash = ?", (key_hash,)
        )
        row = await cursor.fetchone()
        return dict(row) if row else None

@app.post("/login", response_model=None)
async def login(request: Request) -> RedirectResponse | HTMLResponse:
    """Authenticate with username + API key and set a session cookie."""
    form = await request.form()
    username = str(form.get("username", ""))
    api_key = str(form.get("api_key", ""))
    settings = get_settings()

    is_admin = False
    authenticated = False

    # Check admin -- username must be "admin" and key must match
    if username == "admin" and api_key == settings.admin_key:
        is_admin = True
        authenticated = True
    else:
        # Check user key -- verify username matches the key's owner
        user = await get_user_by_key(api_key)
        if user and user["username"] == username:
            authenticated = True
            is_admin = user.get("role") == "admin"

    if authenticated:
        session_token = await create_session(username, is_admin=is_admin)
        response = RedirectResponse(url="/", status_code=302)
        response.set_cookie(
            "docsfy_session",
            session_token,
            httponly=True,
            samesite="strict",
            secure=settings.secure_cookies,
            max_age=SESSION_TTL_SECONDS,
        )
        return response

<!-- Field name="api_key" matches the POST handler in main.py (form.get("api_key")).
     Label says "Password" for UX, but the backend field name is api_key. -->
<label for="api_key">Password</label>
<input type="password" id="api_key" name="api_key" placeholder="Enter your password" required>

SESSION_TTL_SECONDS = 28800  # 8 hours
SESSION_TTL_HOURS = SESSION_TTL_SECONDS // 3600

def _hash_session_token(token: str) -> str:
    """Hash a session token for storage."""
    return hashlib.sha256(token.encode()).hexdigest()

async def create_session(
    username: str, is_admin: bool = False, ttl_hours: int = SESSION_TTL_HOURS
) -> str:
    """Create an opaque session token."""
    token = secrets.token_urlsafe(32)
    token_hash = _hash_session_token(token)
    expires_at = datetime.now(timezone.utc) + timedelta(hours=ttl_hours)
    expires_str = expires_at.strftime("%Y-%m-%d %H:%M:%S")
    async with aiosqlite.connect(DB_PATH) as db:
        await db.execute(
            "INSERT INTO sessions (token, username, is_admin, expires_at) VALUES (?, ?, ?, ?)",
            (token_hash, username, 1 if is_admin else 0, expires_str),
        )
        await db.commit()
    return token

@app.get("/logout")
async def logout(request: Request) -> RedirectResponse:
    """Clear the session cookie, delete session from DB, and redirect to login."""
    session_token = request.cookies.get("docsfy_session")
    if session_token:
        await delete_session(session_token)
    settings = get_settings()
    response = RedirectResponse(url="/login", status_code=302)
    response.delete_cookie(
        "docsfy_session",
        httponly=True,
        samesite="strict",
        secure=settings.secure_cookies,
    )
    return response

services:
  docsfy:
    env_file: .env
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

async def test_login_redirect_when_unauthenticated(
    unauthed_client: AsyncClient,
) -> None:
    """Browser requests to protected pages should redirect to /login."""
    response = await unauthed_client.get("/", follow_redirects=False)
    assert response.status_code == 302
    assert response.headers["location"] == "/login"


async def test_api_returns_401_when_unauthenticated(
    unauthed_client: AsyncClient,
) -> None:
    """API requests without auth should return 401."""
    response = await unauthed_client.get("/api/status")
    assert response.status_code == 401
    assert response.json()["detail"] == "Unauthorized"

def _require_write_access(request: Request) -> None:
    """Raise 403 if user is a viewer (read-only)."""
    if request.state.role not in ("admin", "user"):
        raise HTTPException(
            status_code=403,
            detail="Write access required.",
        )

def _require_admin(request: Request) -> None:
    """Raise 403 if the user is not an admin."""
    if not request.state.is_admin:
        raise HTTPException(status_code=403, detail="Admin access required")

# Don't call _require_write_access -- viewers should be able to change their password
if request.state.is_admin and not request.state.user:
    raise HTTPException(
        status_code=400,
        detail="ADMIN_KEY users cannot rotate keys. Change the ADMIN_KEY env var instead.",
    )

async def _check_ownership(
    request: Request, project_name: str, project: dict[str, Any]
) -> None:
    """Raise 404 if the requesting user does not own the project (unless admin)."""
    if request.state.is_admin:
        return
    project_owner = str(project.get("owner", ""))
    if project_owner == request.state.username:
        return
    # Check if user has been granted access (scoped by project_owner)
    access = await get_project_access(project_name, project_owner=project_owner)
    if request.state.username in access:
        return
    raise HTTPException(status_code=404, detail="Not found")

# GET /api/projects/{name} - returns 404 to avoid leaking existence
response = await ac.get("/api/projects/secret-proj")
assert response.status_code == 404

# REQUIRED - Admin key for user management (minimum 16 characters)
ADMIN_KEY=your-secure-admin-key-here-min-16-chars

# Set to false for local HTTP development
# SECURE_COOKIES=false

class Settings(BaseSettings):
    admin_key: str = ""  # Required — validated at startup
    ai_provider: str = "claude"
    ai_model: str = "claude-opus-4-6[1m]"
    ai_cli_timeout: int = Field(default=60, gt=0)
    log_level: str = "INFO"
    data_dir: str = "/data"
    secure_cookies: bool = True  # Set to False for local HTTP dev

if not settings.admin_key:
    logger.error("ADMIN_KEY environment variable is required")
    raise SystemExit(1)

if len(settings.admin_key) < 16:
    logger.error("ADMIN_KEY must be at least 16 characters long")
    raise SystemExit(1)

[env.unittests]
deps = ["uv"]
commands = [["uv", "run", "--extra", "dev", "pytest", "-n", "auto", "tests"]]