70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
"""Безопасность админки: проверка Telegram-логина, сессия и CSRF."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import hmac
|
||
import secrets
|
||
import time
|
||
from collections.abc import Mapping
|
||
|
||
from fastapi import HTTPException, Request
|
||
|
||
#: Сколько живёт подпись Telegram-логина (секунды).
|
||
TELEGRAM_AUTH_MAX_AGE = 24 * 60 * 60
|
||
|
||
#: Допустимое расхождение часов с Telegram (секунды).
|
||
FUTURE_SKEW = 300
|
||
|
||
|
||
def telegram_check_string(data: Mapping[str, object]) -> str:
|
||
"""Строка для проверки подписи: отсортированные ``key=value`` без ``hash``."""
|
||
pairs = sorted((key, str(value)) for key, value in data.items() if key != "hash")
|
||
return "\n".join(f"{key}={value}" for key, value in pairs)
|
||
|
||
|
||
def verify_telegram_auth(
|
||
data: Mapping[str, object],
|
||
bot_token: str,
|
||
*,
|
||
max_age: int = TELEGRAM_AUTH_MAX_AGE,
|
||
now: float | None = None,
|
||
) -> bool:
|
||
"""Проверить подпись Telegram Login Widget и свежесть ``auth_date``.
|
||
|
||
Секрет — ``SHA256(bot_token)``, подпись — HMAC-SHA256 от check-string.
|
||
"""
|
||
received = str(data.get("hash") or "")
|
||
if not received or not bot_token:
|
||
return False
|
||
|
||
expected = hmac.new(
|
||
hashlib.sha256(bot_token.encode()).digest(),
|
||
telegram_check_string(data).encode(),
|
||
hashlib.sha256,
|
||
).hexdigest()
|
||
if not hmac.compare_digest(expected, received):
|
||
return False
|
||
|
||
try:
|
||
auth_date = int(str(data.get("auth_date") or ""))
|
||
except (TypeError, ValueError):
|
||
return False
|
||
|
||
moment = time.time() if now is None else now
|
||
return -FUTURE_SKEW <= moment - auth_date <= max_age
|
||
|
||
|
||
def check_csrf(request: Request, token: str) -> None:
|
||
"""Проверить CSRF-токен формы; при несовпадении — 400."""
|
||
expected = request.session.get("csrf")
|
||
if not expected or not token or not secrets.compare_digest(str(expected), str(token)):
|
||
raise HTTPException(status_code=400, detail="CSRF-токен неверен или устарел")
|
||
|
||
|
||
def safe_redirect_target(value: str | None) -> str:
|
||
"""Разрешить редирект только на внутренний путь (защита от open redirect)."""
|
||
if value and value.startswith("/") and not value.startswith("//"):
|
||
return value
|
||
return "/"
|