duty-teller/api/app.py

"""FastAPI app: /api/duties and static webapp."""

import logging
import re
from pathlib import Path

import config
from fastapi import FastAPI, Header, HTTPException, Query, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles

from db.session import session_scope
from db.repository import get_duties
from db.schemas import DutyWithUser, CalendarEvent
from api.telegram_auth import validate_init_data_with_reason
from api.calendar_ics import get_calendar_events

log = logging.getLogger(__name__)

# ISO date YYYY-MM-DD
_DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")


def _validate_duty_dates(from_date: str, to_date: str) -> None:
    """Raise HTTPException 400 if dates are invalid or from_date > to_date."""
    if not _DATE_RE.match(from_date) or not _DATE_RE.match(to_date):
        raise HTTPException(
            status_code=400,
            detail="Параметры from и to должны быть в формате YYYY-MM-DD",
        )
    if from_date > to_date:
        raise HTTPException(
            status_code=400,
            detail="Дата from не должна быть позже to",
        )


def _fetch_duties_response(from_date: str, to_date: str) -> list[DutyWithUser]:
    """Fetch duties in range and return list of DutyWithUser. Uses config.DATABASE_URL."""
    with session_scope(config.DATABASE_URL) as session:
        rows = get_duties(session, from_date=from_date, to_date=to_date)
        return [
            DutyWithUser(
                id=duty.id,
                user_id=duty.user_id,
                start_at=duty.start_at,
                end_at=duty.end_at,
                full_name=full_name,
                event_type=(
                    duty.event_type
                    if duty.event_type in ("duty", "unavailable", "vacation")
                    else "duty"
                ),
            )
            for duty, full_name in rows
        ]


def _auth_error_detail(auth_reason: str) -> str:
    """Return user-facing detail message for 403 when initData validation fails."""
    if auth_reason == "hash_mismatch":
        return (
            "Неверная подпись. Убедитесь, что BOT_TOKEN на сервере совпадает с токеном бота, "
            "из которого открыт календарь (тот же бот, что в меню)."
        )
    return "Неверные данные авторизации"


def _is_private_client(client_host: str | None) -> bool:
    """True if client is localhost or private LAN (dev / same-machine access).

    Note: Behind a reverse proxy (e.g. nginx, Caddy), request.client.host is often
    the proxy address (e.g. 127.0.0.1). Then "private client" would be true for all
    requests when initData is missing. For production, either rely on the Mini App
    always sending initData, or configure the proxy to forward the real client IP
    (e.g. X-Forwarded-For) and use that for this check. Do not rely on the private-IP
    bypass when deployed behind a proxy without one of these measures.
    """
    if not client_host:
        return False
    if client_host in ("127.0.0.1", "::1"):
        return True
    parts = client_host.split(".")
    if len(parts) == 4:  # IPv4
        try:
            a, b, c, d = (int(x) for x in parts)
            if (a == 10) or (a == 172 and 16 <= b <= 31) or (a == 192 and b == 168):
                return True
        except (ValueError, IndexError):
            pass
    return False


def get_authenticated_username(
    request: Request,
    x_telegram_init_data: str | None,
) -> str:
    """Validate Mini App auth. Returns username (or "" when bypass allowed); raises HTTPException 403 otherwise."""
    init_data = (x_telegram_init_data or "").strip()
    if not init_data:
        client_host = request.client.host if request.client else None
        if _is_private_client(client_host) or config.MINI_APP_SKIP_AUTH:
            if config.MINI_APP_SKIP_AUTH:
                log.warning("allowing without initData (MINI_APP_SKIP_AUTH is set)")
            return ""
        log.warning("no X-Telegram-Init-Data header (client=%s)", client_host)
        raise HTTPException(status_code=403, detail="Откройте календарь из Telegram")
    max_age = config.INIT_DATA_MAX_AGE_SECONDS or None
    username, auth_reason = validate_init_data_with_reason(
        init_data, config.BOT_TOKEN, max_age_seconds=max_age
    )
    if username is None:
        log.warning("initData validation failed: %s", auth_reason)
        raise HTTPException(status_code=403, detail=_auth_error_detail(auth_reason))
    if not config.can_access_miniapp(username):
        log.warning("username not in allowlist: %s", username)
        raise HTTPException(status_code=403, detail="Доступ запрещён")
    return username


app = FastAPI(title="Duty Teller API")
app.add_middleware(
    CORSMiddleware,
    allow_origins=config.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


@app.get("/api/duties", response_model=list[DutyWithUser])
def list_duties(
    request: Request,
    from_date: str = Query(..., description="ISO date YYYY-MM-DD", alias="from"),
    to_date: str = Query(..., description="ISO date YYYY-MM-DD", alias="to"),
    x_telegram_init_data: str | None = Header(None, alias="X-Telegram-Init-Data"),
) -> list[DutyWithUser]:
    _validate_duty_dates(from_date, to_date)
    log.info(
        "GET /api/duties from %s, has initData: %s",
        request.client.host if request.client else "?",
        bool((x_telegram_init_data or "").strip()),
    )
    get_authenticated_username(request, x_telegram_init_data)
    return _fetch_duties_response(from_date, to_date)


@app.get("/api/calendar-events", response_model=list[CalendarEvent])
def list_calendar_events(
    request: Request,
    from_date: str = Query(..., description="ISO date YYYY-MM-DD", alias="from"),
    to_date: str = Query(..., description="ISO date YYYY-MM-DD", alias="to"),
    x_telegram_init_data: str | None = Header(None, alias="X-Telegram-Init-Data"),
) -> list[CalendarEvent]:
    _validate_duty_dates(from_date, to_date)
    get_authenticated_username(request, x_telegram_init_data)
    url = config.EXTERNAL_CALENDAR_ICS_URL
    if not url:
        return []
    events = get_calendar_events(url, from_date=from_date, to_date=to_date)
    return [CalendarEvent(date=e["date"], summary=e["summary"]) for e in events]


webapp_path = Path(__file__).resolve().parent.parent / "webapp"
if webapp_path.is_dir():
    app.mount("/app", StaticFiles(directory=str(webapp_path), html=True), name="webapp")