This commit is contained in:
@@ -0,0 +1,59 @@
|
||||
import hmac
|
||||
import hashlib
|
||||
import time
|
||||
from urllib.parse import parse_qsl
|
||||
|
||||
from app.core.config import settings
|
||||
|
||||
# Reject initData older than this many seconds (24 hours)
|
||||
_MAX_AGE_SECONDS = 86_400
|
||||
|
||||
|
||||
def validate_init_data(init_data: str) -> dict | None:
|
||||
"""Validate Telegram WebApp initData using HMAC-SHA256.
|
||||
|
||||
Also rejects payloads whose auth_date is older than _MAX_AGE_SECONDS to
|
||||
prevent indefinite replay of leaked initData strings.
|
||||
"""
|
||||
try:
|
||||
# parse_qsl already URL-decodes each value. Decoding the whole query
|
||||
# string first can turn escaped separators inside JSON values into real
|
||||
# separators and make Telegram's HMAC check fail.
|
||||
parsed = dict(parse_qsl(init_data, keep_blank_values=True))
|
||||
hash_value = parsed.pop("hash", "")
|
||||
if not hash_value:
|
||||
return None
|
||||
|
||||
# Validate auth_date freshness before doing any crypto work
|
||||
auth_date_str = parsed.get("auth_date")
|
||||
if not auth_date_str:
|
||||
return None
|
||||
try:
|
||||
age = time.time() - int(auth_date_str)
|
||||
if age > _MAX_AGE_SECONDS:
|
||||
return None
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
data_check_string = "\n".join(
|
||||
f"{k}={v}" for k, v in sorted(parsed.items())
|
||||
)
|
||||
|
||||
secret_key = hmac.new(
|
||||
b"WebAppData",
|
||||
settings.bot_token.encode(),
|
||||
hashlib.sha256,
|
||||
).digest()
|
||||
|
||||
expected = hmac.new(
|
||||
secret_key,
|
||||
data_check_string.encode(),
|
||||
hashlib.sha256,
|
||||
).hexdigest()
|
||||
|
||||
if not hmac.compare_digest(expected, hash_value):
|
||||
return None
|
||||
|
||||
return parsed
|
||||
except Exception:
|
||||
return None
|
||||
Reference in New Issue
Block a user