Files
duty-teller/api/app.py
Nikolay Tatarinov 7cdf1edc34 Enhance API access control and update Docker configuration
- Added port mapping to docker-compose for local development.
- Modified the API to allow access from localhost without Telegram initData for local development.
- Updated the web application to check for localhost before denying access based on initData.
2026-02-17 14:15:06 +03:00

77 lines
2.8 KiB
Python

"""FastAPI app: /api/duties and static webapp."""
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 get_session
from db.repository import get_duties
from db.schemas import DutyWithUser
from api.telegram_auth import validate_init_data
app = FastAPI(title="Duty Teller API")
app.add_middleware(
CORSMiddleware,
allow_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]:
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()
raise HTTPException(status_code=403, detail="Откройте календарь из Telegram")
username = validate_init_data(init_data, config.BOT_TOKEN)
if username is None:
raise HTTPException(status_code=403, detail="Неверные данные авторизации")
if not config.can_access_miniapp(username):
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()
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")