Refactor configuration and enhance Telegram initData validation

- Improved formatting and readability in config.py and other files by adding line breaks.
- Introduced INIT_DATA_MAX_AGE_SECONDS to enforce replay protection for Telegram initData.
- Updated validate_init_data function to include max_age_seconds parameter for validation.
- Enhanced API to reject old initData based on the new max_age_seconds setting.
- Added tests for auth_date expiry and validation of initData in test_telegram_auth.py.
- Updated README with details on the new INIT_DATA_MAX_AGE_SECONDS configuration.

api/app.py

@@ -1,4 +1,5 @@
 """FastAPI app: /api/duties and static webapp."""
+
 import logging
 import re
 from pathlib import Path
@@ -50,7 +51,15 @@ def _fetch_duties_response(from_date: str, to_date: str) -> list[DutyWithUser]:
 
 
 def _is_private_client(client_host: str | None) -> bool:
-    """True if client is localhost or private LAN (dev / same-machine access)."""
+    """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"):
@@ -84,19 +93,28 @@ def list_duties(
     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()))
+    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:
         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("duties: allowing without initData (MINI_APP_SKIP_AUTH is set)")
+                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)
+    max_age = config.INIT_DATA_MAX_AGE_SECONDS or None
+    username = validate_init_data(init_data, config.BOT_TOKEN, max_age_seconds=max_age)
     if username is None:
-        log.warning("duties: initData validation failed (invalid signature or no username)")
+        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")

api/telegram_auth.py

@@ -1,17 +1,26 @@
 """Validate Telegram Web App initData and extract user username."""
+
 import hashlib
 import hmac
 import json
+import time
 from urllib.parse import unquote
 
 # Telegram algorithm: https://core.telegram.org/bots/webapps#validating-data-received-via-the-mini-app
 # Data-check string must use the same key=value pairs as received (sorted by key); we preserve raw values.
 
 
-def validate_init_data(init_data: str, bot_token: str) -> str | None:
+def validate_init_data(
+    init_data: str,
+    bot_token: str,
+    max_age_seconds: int | None = None,
+) -> str | None:
     """
     Validate initData signature and return the Telegram username (lowercase, no @).
     Returns None if data is invalid, forged, or user has no username.
+
+    If max_age_seconds is set, initData must include auth_date and it must be no older
+    than max_age_seconds (replay protection). Example: 86400 = 24 hours.
     """
     if not init_data or not bot_token:
         return None
@@ -45,6 +54,17 @@ def validate_init_data(init_data: str, bot_token: str) -> str | None:
     ).hexdigest()
     if not hmac.compare_digest(computed.lower(), hash_val.lower()):
         return None
+    # Optional replay protection: reject initData older than max_age_seconds
+    if max_age_seconds is not None and max_age_seconds > 0:
+        auth_date_raw = params.get("auth_date")
+        if not auth_date_raw:
+            return None
+        try:
+            auth_date = int(float(auth_date_raw))
+        except (ValueError, TypeError):
+            return None
+        if time.time() - auth_date > max_age_seconds:
+            return None
     # Parse user JSON (value may be URL-encoded in the raw string)
     user_raw = params.get("user")
     if not user_raw:

api/test_app.py

@@ -1,10 +1,14 @@
 """Tests for FastAPI app /api/duties."""
+
+import time
 from unittest.mock import patch
 
 import pytest
 from fastapi.testclient import TestClient
 
+import config
 from api.app import app
+from api.test_telegram_auth import _make_init_data
 
 
 @pytest.fixture
@@ -101,3 +105,28 @@ def test_duties_200_with_allowed_user(mock_can_access, mock_validate, client):
     assert len(r.json()) == 1
     assert r.json()[0]["full_name"] == "Иван Иванов"
     mock_fetch.assert_called_once_with("2025-01-01", "2025-01-31")
+
+
+def test_duties_e2e_auth_real_validation(client, monkeypatch):
+    """E2E: valid initData + allowlist, no mocks on validate_init_data; full auth path."""
+    test_token = "123:ABC"
+    test_username = "e2euser"
+    monkeypatch.setattr(config, "BOT_TOKEN", test_token)
+    monkeypatch.setattr(config, "ALLOWED_USERNAMES", {test_username})
+    monkeypatch.setattr(config, "ADMIN_USERNAMES", set())
+    monkeypatch.setattr(config, "INIT_DATA_MAX_AGE_SECONDS", 0)
+    init_data = _make_init_data(
+        {"id": 1, "username": test_username},
+        test_token,
+        auth_date=int(time.time()),
+    )
+    with patch("api.app._fetch_duties_response") as mock_fetch:
+        mock_fetch.return_value = []
+        r = client.get(
+            "/api/duties",
+            params={"from": "2025-01-01", "to": "2025-01-31"},
+            headers={"X-Telegram-Init-Data": init_data},
+        )
+    assert r.status_code == 200
+    assert r.json() == []
+    mock_fetch.assert_called_once_with("2025-01-01", "2025-01-31")

api/test_telegram_auth.py

@@ -1,19 +1,25 @@
 """Tests for api.telegram_auth.validate_init_data."""
+
 import hashlib
 import hmac
 import json
-from urllib.parse import quote, urlencode
+from urllib.parse import quote
 
-import pytest
 
 from api.telegram_auth import validate_init_data
 
 
-def _make_init_data(user: dict | None, bot_token: str) -> str:
+def _make_init_data(
+    user: dict | None,
+    bot_token: str,
+    auth_date: int | None = None,
+) -> str:
     """Build initData string with valid HMAC for testing."""
     params = {}
     if user is not None:
         params["user"] = quote(json.dumps(user))
+    if auth_date is not None:
+        params["auth_date"] = str(auth_date)
     pairs = sorted(params.items())
     data_string = "\n".join(f"{k}={v}" for k, v in pairs)
     secret_key = hmac.new(
@@ -82,3 +88,36 @@ def test_empty_bot_token_returns_none():
     user = {"id": 1, "username": "u"}
     init_data = _make_init_data(user, "token")
     assert validate_init_data(init_data, "") is None
+
+
+def test_auth_date_expiry_rejects_old_init_data():
+    """When max_age_seconds is set, initData older than that is rejected."""
+    import time as t
+
+    bot_token = "123:ABC"
+    user = {"id": 1, "username": "testuser"}
+    # auth_date 100 seconds ago
+    old_ts = int(t.time()) - 100
+    init_data = _make_init_data(user, bot_token, auth_date=old_ts)
+    assert validate_init_data(init_data, bot_token, max_age_seconds=60) is None
+    assert validate_init_data(init_data, bot_token, max_age_seconds=200) == "testuser"
+
+
+def test_auth_date_expiry_accepts_fresh_init_data():
+    """Fresh auth_date within max_age_seconds is accepted."""
+    import time as t
+
+    bot_token = "123:ABC"
+    user = {"id": 1, "username": "testuser"}
+    fresh_ts = int(t.time()) - 10
+    init_data = _make_init_data(user, bot_token, auth_date=fresh_ts)
+    assert validate_init_data(init_data, bot_token, max_age_seconds=60) == "testuser"
+
+
+def test_auth_date_expiry_requires_auth_date_when_max_age_set():
+    """When max_age_seconds is set but auth_date is missing, return None."""
+    bot_token = "123:ABC"
+    user = {"id": 1, "username": "testuser"}
+    init_data = _make_init_data(user, bot_token)  # no auth_date
+    assert validate_init_data(init_data, bot_token, max_age_seconds=86400) is None
+    assert validate_init_data(init_data, bot_token) == "testuser"