"""FastAPI app: /api/duties and static webapp.""" import logging from pathlib import Path from typing import Annotated, Generator import config from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles from sqlalchemy.orm import Session 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 from utils.dates import validate_date_range log = logging.getLogger(__name__) def _validate_duty_dates(from_date: str, to_date: str) -> None: """Raise HTTPException 400 if dates are invalid or from_date > to_date.""" try: validate_date_range(from_date, to_date) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) from e def get_validated_dates( from_date: str = Query(..., description="ISO date YYYY-MM-DD", alias="from"), to_date: str = Query(..., description="ISO date YYYY-MM-DD", alias="to"), ) -> tuple[str, str]: """FastAPI dependency: validate from_date/to_date and return (from_date, to_date). Raises 400 if invalid.""" _validate_duty_dates(from_date, to_date) return (from_date, to_date) def get_db_session() -> Generator[Session, None, None]: """FastAPI dependency: yield a DB session from session_scope.""" with session_scope(config.DATABASE_URL) as session: yield session def require_miniapp_username( request: Request, x_telegram_init_data: Annotated[ str | None, Header(alias="X-Telegram-Init-Data") ] = None, ) -> str: """FastAPI dependency: return authenticated username or raise 403.""" return get_authenticated_username(request, x_telegram_init_data) def _fetch_duties_response( session: Session, from_date: str, to_date: str ) -> list[DutyWithUser]: """Fetch duties in range and return list of DutyWithUser.""" 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, x_telegram_init_data: str | None = Header(None, alias="X-Telegram-Init-Data"), dates: tuple[str, str] = Depends(get_validated_dates), _username: str = Depends(require_miniapp_username), session: Session = Depends(get_db_session), ) -> list[DutyWithUser]: from_date_val, to_date_val = dates log.info( "GET /api/duties from %s, has initData: %s", request.client.host if request.client else "?", bool((x_telegram_init_data or "").strip()), ) return _fetch_duties_response(session, from_date_val, to_date_val) @app.get("/api/calendar-events", response_model=list[CalendarEvent]) def list_calendar_events( dates: tuple[str, str] = Depends(get_validated_dates), _username: str = Depends(require_miniapp_username), ) -> list[CalendarEvent]: from_date_val, to_date_val = dates url = config.EXTERNAL_CALENDAR_ICS_URL if not url: return [] events = get_calendar_events(url, from_date=from_date_val, to_date=to_date_val) 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")