Arnike/duty-teller
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 14 Wiki Activity

Implement phone number normalization and access control for Telegram users

Browse Source 
- Added functionality to normalize phone numbers for comparison, ensuring only digits are stored and checked.
- Updated configuration to include optional phone number allowlists for users and admins in the environment settings.
- Enhanced authentication logic to allow access based on normalized phone numbers, in addition to usernames.
- Introduced new helper functions for parsing and validating phone numbers, improving code organization and maintainability.
- Added unit tests to validate phone normalization and access control based on phone numbers.
This commit is contained in:
Nikolay Tatarinov
2026-02-18 16:11:44 +03:00
parent d0d22c150a
commit 59ba2a9ca4
10 changed files with 344 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
duty_teller/api/dependencies.py
View File

@@ -4,12 +4,12 @@ import logging
import re
from typing import Annotated, Generator
from fastapi import Header, HTTPException, Query, Request
from fastapi import Depends, Header, HTTPException, Query, Request
from sqlalchemy.orm import Session
import duty_teller.config as config
from duty_teller.api.telegram_auth import validate_init_data_with_reason
from duty_teller.db.repository import get_duties
from duty_teller.db.repository import get_duties, get_user_by_telegram_id
from duty_teller.db.schemas import DutyWithUser
from duty_teller.db.session import session_scope
from duty_teller.i18n import t
@@ -74,8 +74,9 @@ def require_miniapp_username(
x_telegram_init_data: Annotated[
str | None, Header(alias="X-Telegram-Init-Data")
] = None,
session: Session = Depends(get_db_session),
) -> str:
return get_authenticated_username(request, x_telegram_init_data)
return get_authenticated_username(request, x_telegram_init_data, session)
def _is_private_client(client_host: str | None) -> bool:
@@ -95,31 +96,43 @@ def _is_private_client(client_host: str | None) -> bool:
def get_authenticated_username(
request: Request, x_telegram_init_data: str | None
request: Request,
x_telegram_init_data: str | None,
session: Session,
) -> str:
"""Return identifier for miniapp auth (username or full_name or id:...); empty if skip-auth."""
if config.MINI_APP_SKIP_AUTH:
log.warning("allowing without any auth check (MINI_APP_SKIP_AUTH is set)")
return ""
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("allowing without initData (MINI_APP_SKIP_AUTH is set)")
if _is_private_client(client_host):
return ""
log.warning("no X-Telegram-Init-Data header (client=%s)", client_host)
lang = _lang_from_accept_language(request.headers.get("Accept-Language"))
raise HTTPException(status_code=403, detail=t(lang, "api.open_from_telegram"))
max_age = config.INIT_DATA_MAX_AGE_SECONDS or None
username, auth_reason, lang = validate_init_data_with_reason(
telegram_user_id, username, auth_reason, lang = validate_init_data_with_reason(
init_data, config.BOT_TOKEN, max_age_seconds=max_age
)
if username is None:
if auth_reason != "ok":
log.warning("initData validation failed: %s", auth_reason)
raise HTTPException(
status_code=403, detail=_auth_error_detail(auth_reason, lang)
)
if not config.can_access_miniapp(username):
log.warning("username not in allowlist: %s", username)
raise HTTPException(status_code=403, detail=t(lang, "api.access_denied"))
return username
if username and config.can_access_miniapp(username):
return username
if telegram_user_id is not None:
user = get_user_by_telegram_id(session, telegram_user_id)
if user and user.phone and config.can_access_miniapp_by_phone(user.phone):
return username or (user.full_name or "") or f"id:{telegram_user_id}"
log.warning(
"username/phone not in allowlist (username=%s, telegram_id=%s)",
username,
telegram_user_id,
)
raise HTTPException(status_code=403, detail=t(lang, "api.access_denied"))
def fetch_duties_response(

43
duty_teller/api/telegram_auth.py
View File

@@ -16,7 +16,7 @@ def validate_init_data(
max_age_seconds: int | None = None,
) -> str | None:
"""Validate initData and return username; see validate_init_data_with_reason for failure reason."""
username, _, _ = validate_init_data_with_reason(
_, username, _, _ = validate_init_data_with_reason(
init_data, bot_token, max_age_seconds
)
return username
@@ -34,14 +34,16 @@ def validate_init_data_with_reason(
init_data: str,
bot_token: str,
max_age_seconds: int | None = None,
) -> tuple[str | None, str, str]:
) -> tuple[int | None, str | None, str, str]:
"""
Validate initData signature and return (username, reason, lang) or (None, reason, lang).
reason is one of: "ok", "empty", "no_hash", "hash_mismatch", "auth_date_expired", "no_user", "user_invalid", "no_username".
Validate initData signature and return (telegram_user_id, username, reason, lang).
reason is one of: "ok", "empty", "no_hash", "hash_mismatch", "auth_date_expired",
"no_user", "user_invalid", "no_user_id", "no_username" (legacy; no_user_id used when user.id missing).
lang is from user.language_code normalized to 'ru' or 'en'; 'en' when no user.
On success: (user.id, user.get('username') or None, "ok", lang) — username may be None.
"""
if not init_data or not bot_token:
return (None, "empty", "en")
return (None, None, "empty", "en")
init_data = init_data.strip()
params = {}
for part in init_data.split("&"):
@@ -53,7 +55,7 @@ def validate_init_data_with_reason(
params[key] = value
hash_val = params.pop("hash", None)
if not hash_val:
return (None, "no_hash", "en")
return (None, None, "no_hash", "en")
data_pairs = sorted(params.items())
# Data-check string: key=value with URL-decoded values (per Telegram example)
data_string = "\n".join(f"{k}={unquote(v)}" for k, v in data_pairs)
@@ -69,28 +71,37 @@ def validate_init_data_with_reason(
digestmod=hashlib.sha256,
).hexdigest()
if not hmac.compare_digest(computed.lower(), hash_val.lower()):
return (None, "hash_mismatch", "en")
return (None, None, "hash_mismatch", "en")
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, "auth_date_expired", "en")
return (None, None, "auth_date_expired", "en")
try:
auth_date = int(float(auth_date_raw))
except (ValueError, TypeError):
return (None, "auth_date_expired", "en")
return (None, None, "auth_date_expired", "en")
if time.time() - auth_date > max_age_seconds:
return (None, "auth_date_expired", "en")
return (None, None, "auth_date_expired", "en")
user_raw = params.get("user")
if not user_raw:
return (None, "no_user", "en")
return (None, None, "no_user", "en")
try:
user = json.loads(unquote(user_raw))
except (json.JSONDecodeError, TypeError):
return (None, "user_invalid", "en")
return (None, None, "user_invalid", "en")
if not isinstance(user, dict):
return (None, "user_invalid", "en")
return (None, None, "user_invalid", "en")
lang = _normalize_lang(user.get("language_code"))
raw_id = user.get("id")
if raw_id is None:
return (None, None, "no_user_id", lang)
try:
telegram_user_id = int(raw_id)
except (TypeError, ValueError):
return (None, None, "no_user_id", lang)
username = user.get("username")
if not username or not isinstance(username, str):
return (None, "no_username", lang)
return (username.strip().lstrip("@").lower(), "ok", lang)
if username and isinstance(username, str):
username = username.strip().lstrip("@").lower()
else:
username = None
return (telegram_user_id, username, "ok", lang)

44
duty_teller/config.py
View File

@@ -1,5 +1,6 @@
"""Load configuration from environment. BOT_TOKEN is not validated on import; check in main/entry point."""
import re
import os
from dataclasses import dataclass
from pathlib import Path
@@ -11,6 +12,16 @@ load_dotenv()
# Project root (parent of duty_teller package). Used for webapp path, etc.
PROJECT_ROOT = Path(__file__).resolve().parent.parent
# Only digits for phone comparison (same when saving and when checking allowlist).
_PHONE_DIGITS_RE = re.compile(r"\D")
def normalize_phone(phone: str | None) -> str:
"""Return phone as digits only (spaces, +, parentheses, dashes removed). Empty string if None/empty."""
if not phone or not isinstance(phone, str):
return ""
return _PHONE_DIGITS_RE.sub("", phone.strip())
def _normalize_default_language(value: str) -> str:
"""Normalize DEFAULT_LANGUAGE from env to 'ru' or 'en'."""
@@ -20,6 +31,16 @@ def _normalize_default_language(value: str) -> str:
return "ru" if v.startswith("ru") else "en"
def _parse_phone_list(raw: str) -> set[str]:
"""Parse comma-separated phones into set of normalized (digits-only) strings."""
result = set()
for s in (raw or "").strip().split(","):
normalized = normalize_phone(s)
if normalized:
result.add(normalized)
return result
@dataclass(frozen=True)
class Settings:
"""Optional injectable settings built from env. Tests can override or build from env."""
@@ -30,6 +51,8 @@ class Settings:
http_port: int
allowed_usernames: set[str]
admin_usernames: set[str]
allowed_phones: set[str]
admin_phones: set[str]
mini_app_skip_auth: bool
init_data_max_age_seconds: int
cors_origins: list[str]
@@ -49,6 +72,8 @@ class Settings:
admin = {
s.strip().lstrip("@").lower() for s in raw_admin.split(",") if s.strip()
}
allowed_phones = _parse_phone_list(os.getenv("ALLOWED_PHONES", ""))
admin_phones = _parse_phone_list(os.getenv("ADMIN_PHONES", ""))
raw_cors = os.getenv("CORS_ORIGINS", "").strip()
cors = (
[_o.strip() for _o in raw_cors.split(",") if _o.strip()]
@@ -62,6 +87,8 @@ class Settings:
http_port=int(os.getenv("HTTP_PORT", "8080")),
allowed_usernames=allowed,
admin_usernames=admin,
allowed_phones=allowed_phones,
admin_phones=admin_phones,
mini_app_skip_auth=os.getenv("MINI_APP_SKIP_AUTH", "").strip()
in ("1", "true", "yes"),
init_data_max_age_seconds=int(os.getenv("INIT_DATA_MAX_AGE_SECONDS", "0")),
@@ -93,6 +120,9 @@ ADMIN_USERNAMES = {
s.strip().lstrip("@").lower() for s in _raw_admin.split(",") if s.strip()
}
ALLOWED_PHONES = _parse_phone_list(os.getenv("ALLOWED_PHONES", ""))
ADMIN_PHONES = _parse_phone_list(os.getenv("ADMIN_PHONES", ""))
MINI_APP_SKIP_AUTH = os.getenv("MINI_APP_SKIP_AUTH", "").strip() in ("1", "true", "yes")
INIT_DATA_MAX_AGE_SECONDS = int(os.getenv("INIT_DATA_MAX_AGE_SECONDS", "0"))
@@ -123,6 +153,20 @@ def can_access_miniapp(username: str) -> bool:
return u in ALLOWED_USERNAMES or u in ADMIN_USERNAMES
def can_access_miniapp_by_phone(phone: str | None) -> bool:
"""True if normalized phone is in ALLOWED_PHONES or ADMIN_PHONES."""
normalized = normalize_phone(phone)
if not normalized:
return False
return normalized in ALLOWED_PHONES or normalized in ADMIN_PHONES
def is_admin_by_phone(phone: str | None) -> bool:
"""True if normalized phone is in ADMIN_PHONES."""
normalized = normalize_phone(phone)
return bool(normalized and normalized in ADMIN_PHONES)
def require_bot_token() -> None:
"""Raise SystemExit with a clear message if BOT_TOKEN is not set. Call from entry point."""
if not BOT_TOKEN:

7
duty_teller/db/repository.py
View File

@@ -7,6 +7,11 @@ from sqlalchemy.orm import Session
from duty_teller.db.models import User, Duty, GroupDutyPin
def get_user_by_telegram_id(session: Session, telegram_user_id: int) -> User | None:
"""Find user by telegram_user_id. Returns None if not found (no creation)."""
return session.query(User).filter(User.telegram_user_id == telegram_user_id).first()
def get_or_create_user(
session: Session,
telegram_user_id: int,
@@ -15,7 +20,7 @@ def get_or_create_user(
first_name: str | None = None,
last_name: str | None = None,
) -> User:
user = session.query(User).filter(User.telegram_user_id == telegram_user_id).first()
user = get_user_by_telegram_id(session, telegram_user_id)
if user:
user.full_name = full_name
user.username = username