bincio-activity/bincio/serve/deps.py

"""Shared state and FastAPI dependency functions for bincio.serve.

All module-level globals live here so routers can import them without
creating circular dependencies through server.py.
The CLI sets these before uvicorn starts.
"""
from __future__ import annotations

import json
import os
import re
import threading
import time
from pathlib import Path
from typing import Optional

from fastapi import Cookie, HTTPException, Request, Response

from bincio.serve.db import (
    User,
    authenticate,
    create_session,
    delete_session,
    get_session,
    get_user,
    open_db,
)
from bincio.edit.ops import VALID_ACTIVITY_ID as _VALID_ACTIVITY_ID
from bincio.shared.images import ALLOWED_IMAGE_TYPES as _ALLOWED_IMAGE_TYPES  # noqa: F401
from bincio.shared.images import MAX_IMAGE_BYTES as _MAX_IMAGE_BYTES  # noqa: F401
from bincio.shared.images import unique_image_name as _unique_image_name  # noqa: F401

# ── Module-level state (set by CLI before uvicorn starts) ─────────────────────

data_dir: Path | None = None
site_dir: Path | None = None
webroot: Path | None = None
strava_client_id: str = ""
strava_client_secret: str = ""
public_url: str = ""
dem_url: str = "https://api.open-elevation.com"
sync_secret: str = ""
_db = None
_strava_sync_running = False
_strava_sync_lock = threading.Lock()

# ── Constants ─────────────────────────────────────────────────────────────────

_VALID_HANDLE = re.compile(r'^[a-z0-9][a-z0-9_-]{0,29}$')
_SESSION_COOKIE = "bincio_session"
_COOKIE_MAX_AGE = 30 * 86400  # 30 days
_SESSION_DOMAIN = os.environ.get("SESSION_DOMAIN") or None

_STRAVA_CREDS_FILE = "strava_credentials.json"

_login_attempts: dict[str, list[float]] = {}
_register_attempts: dict[str, list[float]] = {}
_RATE_WINDOW = 900   # 15 minutes
_LOGIN_RATE_LIMIT = 10
_REGISTER_RATE_LIMIT = 5

# ── Core helpers ──────────────────────────────────────────────────────────────

def _get_data_dir() -> Path:
    if data_dir is None:
        raise HTTPException(500, "Server not configured")
    return data_dir


def _get_db():
    global _db
    if _db is None:
        _db = open_db(_get_data_dir())
    return _db


def _strava_creds(handle: str) -> tuple[str, str]:
    """Return (client_id, client_secret) for a user.

    Per-user credentials take precedence over the instance-level globals.
    Returns ("", "") when neither is configured.
    """
    creds_path = _get_data_dir() / handle / _STRAVA_CREDS_FILE
    if creds_path.exists():
        try:
            d = json.loads(creds_path.read_text(encoding="utf-8"))
            cid = str(d.get("client_id", "")).strip()
            csec = str(d.get("client_secret", "")).strip()
            if cid and csec:
                return cid, csec
        except (OSError, json.JSONDecodeError, KeyError, ValueError):
            pass
    return strava_client_id, strava_client_secret


def _check_id(activity_id: str) -> str:
    if not _VALID_ACTIVITY_ID.match(activity_id):
        raise HTTPException(400, "Invalid activity ID")
    return activity_id

# ── Rate limiting ─────────────────────────────────────────────────────────────

def _check_rate_limit(
    ip: str,
    store: dict[str, list[float]],
    limit: int,
    msg: str = "Too many attempts. Try again later.",
) -> None:
    now = time.time()
    attempts = [t for t in store.get(ip, []) if now - t < _RATE_WINDOW]
    store[ip] = attempts
    if len(attempts) >= limit:
        raise HTTPException(429, msg)
    attempts.append(now)
    store[ip] = attempts

# ── Auth dependency functions ─────────────────────────────────────────────────

def _current_user(bincio_session: Optional[str] = Cookie(default=None)) -> Optional[User]:
    if not bincio_session:
        return None
    return get_session(_get_db(), bincio_session)


def _require_user(bincio_session: Optional[str] = Cookie(default=None)) -> User:
    user = _current_user(bincio_session)
    if not user:
        raise HTTPException(401, "Not authenticated")
    return user


def _require_admin(bincio_session: Optional[str] = Cookie(default=None)) -> User:
    user = _require_user(bincio_session)
    if not user.is_admin:
        raise HTTPException(403, "Admin required")
    return user


def _require_auth(
    request: Request,
    bincio_session: Optional[str] = Cookie(default=None),
) -> User:
    """Accept session cookie (web) OR Authorization: Bearer token (mobile)."""
    token = bincio_session
    if not token:
        auth = request.headers.get("Authorization", "")
        if auth.startswith("Bearer "):
            token = auth[7:]
    if not token:
        raise HTTPException(401, "Not authenticated")
    user = get_session(_get_db(), token)
    if not user:
        raise HTTPException(401, "Invalid or expired session")
    return user


def _set_session_cookie(response: Response, token: str) -> None:
    kwargs: dict = dict(
        key=_SESSION_COOKIE,
        value=token,
        max_age=_COOKIE_MAX_AGE,
        httponly=True,
        samesite="lax",
        secure=False,
    )
    if _SESSION_DOMAIN:
        kwargs["domain"] = _SESSION_DOMAIN
    response.set_cookie(**kwargs)