Enhance API and configuration for Telegram miniapp
- Added support for CORS origins and a new environment variable for miniapp access control. - Implemented date validation for API requests to ensure correct date formats. - Updated FastAPI app to allow access without Telegram initData for local development. - Enhanced error handling and logging for better debugging. - Added tests for API functionality and Telegram initData validation. - Updated README with new environment variable details and testing instructions. - Modified Docker and Git ignore files to include additional directories and files.
This commit is contained in:
101
api/app.py
101
api/app.py
@@ -1,4 +1,6 @@
|
||||
"""FastAPI app: /api/duties and static webapp."""
|
||||
import logging
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
import config
|
||||
@@ -6,15 +8,68 @@ from fastapi import FastAPI, Header, HTTPException, Query, Request
|
||||
from fastapi.middleware.cors import CORSMiddleware
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from db.session import get_session
|
||||
from db.session import session_scope
|
||||
from db.repository import get_duties
|
||||
from db.schemas import DutyWithUser
|
||||
from api.telegram_auth import validate_init_data
|
||||
|
||||
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,
|
||||
)
|
||||
for duty, full_name in rows
|
||||
]
|
||||
|
||||
|
||||
def _is_private_client(client_host: str | None) -> bool:
|
||||
"""True if client is localhost or private LAN (dev / same-machine access)."""
|
||||
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
|
||||
|
||||
|
||||
app = FastAPI(title="Duty Teller API")
|
||||
app.add_middleware(
|
||||
CORSMiddleware,
|
||||
allow_origins=["*"],
|
||||
allow_origins=config.CORS_ORIGINS,
|
||||
allow_credentials=True,
|
||||
allow_methods=["*"],
|
||||
allow_headers=["*"],
|
||||
@@ -28,47 +83,25 @@ def list_duties(
|
||||
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()))
|
||||
init_data = (x_telegram_init_data or "").strip()
|
||||
if not init_data:
|
||||
# Allow access from localhost without Telegram initData (local dev only)
|
||||
client_host = request.client.host if request.client else None
|
||||
if client_host in ("127.0.0.1", "::1"):
|
||||
session = get_session(config.DATABASE_URL)
|
||||
try:
|
||||
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,
|
||||
)
|
||||
for duty, full_name in rows
|
||||
]
|
||||
finally:
|
||||
session.close()
|
||||
if _is_private_client(client_host) or config.MINI_APP_SKIP_AUTH:
|
||||
if config.MINI_APP_SKIP_AUTH:
|
||||
log.warning("duties: allowing without initData (MINI_APP_SKIP_AUTH is set)")
|
||||
return _fetch_duties_response(from_date, to_date)
|
||||
log.warning("duties: no X-Telegram-Init-Data header (client=%s)", client_host)
|
||||
raise HTTPException(status_code=403, detail="Откройте календарь из Telegram")
|
||||
username = validate_init_data(init_data, config.BOT_TOKEN)
|
||||
if username is None:
|
||||
log.warning("duties: initData validation failed (invalid signature or no username)")
|
||||
raise HTTPException(status_code=403, detail="Неверные данные авторизации")
|
||||
if not config.can_access_miniapp(username):
|
||||
log.warning("duties: username not in allowlist")
|
||||
raise HTTPException(status_code=403, detail="Доступ запрещён")
|
||||
session = get_session(config.DATABASE_URL)
|
||||
try:
|
||||
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,
|
||||
)
|
||||
for duty, full_name in rows
|
||||
]
|
||||
finally:
|
||||
session.close()
|
||||
return _fetch_duties_response(from_date, to_date)
|
||||
|
||||
|
||||
webapp_path = Path(__file__).resolve().parent.parent / "webapp"
|
||||
|
||||
Reference in New Issue
Block a user