Use Telegram numeric user IDs for whitelist authorization

README.en.md

@@ -12,7 +12,7 @@
 ## Overview
 
 Telegram Resender is a focused bot that checks incoming text messages against a whitelist of
-Telegram usernames and forwards approved requests to a configured target chat.
+immutable Telegram numeric user IDs and forwards approved requests to a configured target chat.
 
 The project is intentionally small and practical:
 
@@ -26,7 +26,7 @@ The project is intentionally small and practical:
 - `/start`, `/help`, `/template`, `/avto` commands
 - whitelist-based access control
 - localized Russian and English bot messages
-- separate guidance for unknown users, missing Telegram usernames, and incomplete requests
+- separate guidance for unknown users, missing sender information, and incomplete requests
 - unsupported message guidance for photos, documents, stickers, and voice messages
 - required-field validation and optional confirmation before forwarding
 - admin commands for status checks and whitelist reloads without restart
@@ -79,12 +79,12 @@ All settings are read from environment variables (or `.env`).
 | `TELEGRAM_RESENDER_LOG_FORMAT` | no | `TEXT` or `JSON`, default `TEXT` |
 | `TELEGRAM_RESENDER_POLLING_TIMEOUT` | no | Polling timeout, default `30` |
 
-Whitelist format:
+Whitelist format (one immutable numeric Telegram user ID per line; use `/whoami` to discover IDs):
 
 ```csv
 # whitelist.csv
-alice
-@bob
+123456789
+987654321
 ```
 
 ## Request format

README.md

@@ -23,9 +23,9 @@
 ## Возможности
 
 - `/start`, `/help`, `/template`, `/avto` для пользовательского сценария.
-- Белый список Telegram username из CSV.
+- Белый список числовых Telegram user ID из CSV.
 - Понятный отказ для неизвестного пользователя и отдельный отказ для пользователя
-  без Telegram username.
+  без данных отправителя.
 - Защита от случайной пересылки коротких приветствий вместо заявки.
 - Ответ на неподдерживаемые типы сообщений: фото, документы, стикеры, голосовые.
 - Диагностика конфигурации через `telegram-resender doctor`.
@@ -70,7 +70,7 @@ telegram-resender
 | `TELEGRAM_RESENDER_REQUEST_ACCEPTED_MESSAGE` | нет | переопределение текста успешной заявки |
 | `TELEGRAM_RESENDER_ACCESS_DENIED_MESSAGE` | нет | переопределение текста отказа неизвестному пользователю |
 
-Формат whitelist:
+Формат whitelist (по одному неизменяемому числовому Telegram user ID в строке; ID можно узнать через `/whoami`):
 
 ```csv
 alice

README.ru.md

@@ -20,9 +20,9 @@
 ## Возможности
 
 - `/start`, `/help`, `/template`, `/avto` для пользовательского сценария.
-- Белый список Telegram username из CSV.
+- Белый список числовых Telegram user ID из CSV.
 - Понятный отказ для неизвестного пользователя и отдельный отказ для пользователя
-  без Telegram username.
+  без данных отправителя.
 - Защита от случайной пересылки коротких приветствий вместо заявки.
 - Ответ на неподдерживаемые типы сообщений: фото, документы, стикеры, голосовые.
 - Диагностика конфигурации через `telegram-resender doctor`.
@@ -67,7 +67,7 @@ telegram-resender
 | `TELEGRAM_RESENDER_REQUEST_ACCEPTED_MESSAGE` | нет | переопределение текста успешной заявки |
 | `TELEGRAM_RESENDER_ACCESS_DENIED_MESSAGE` | нет | переопределение текста отказа неизвестному пользователю |
 
-Формат whitelist:
+Формат whitelist (по одному неизменяемому числовому Telegram user ID в строке; ID можно узнать через `/whoami`):
 
 ```csv
 alice

src/telegram_resender/app.py

@@ -56,6 +56,7 @@ def incoming_from_message(message: Message) -> IncomingMessage:
         chat_id=message.chat.id,
         text=message.text or "",
         user=UserProfile(
+            id=user.id if user else None,
             username=user.username if user else None,
             first_name=user.first_name if user else None,
             last_name=user.last_name if user else None,

src/telegram_resender/cli.py

@@ -76,7 +76,7 @@ def run_doctor(
     print(f"Locale: {settings.locale}", file=stdout)
     print(f"Forward chat id: {settings.forward_chat_id}", file=stdout)
     print(f"Whitelist path: {_display_path(settings.whitelist_path)}", file=stdout)
-    print(f"Whitelist users: {len(whitelist.usernames)}", file=stdout)
+    print(f"Whitelist users: {len(whitelist.user_ids)}", file=stdout)
     print(f"Routes path: {_display_optional_path(settings.routes_path)}", file=stdout)
     print(f"Routes: {len(routes)}", file=stdout)
     print(f"Storage path: {_display_path(settings.storage_path)}", file=stdout)

src/telegram_resender/models.py

@@ -11,7 +11,8 @@
 class UserProfile:
     """Telegram user fields needed by the resender."""
 
-    username: str | None
+    id: int | None = None
+    username: str | None = None
     first_name: str | None = None
     last_name: str | None = None
 
@@ -28,9 +29,9 @@ class IncomingMessage:
 
 
 DecisionReason = Literal[
-    "allowed_username",
-    "missing_username",
-    "unknown_username",
+    "allowed_user_id",
+    "missing_user_id",
+    "unknown_user_id",
     "invalid_request",
     "no_route_matched",
 ]

src/telegram_resender/routes.py

@@ -22,7 +22,7 @@ class RouteRule:
     template: str | None
     enabled: bool = True
 
-    def matches(self, *, username: str, text: str) -> bool:
+    def matches(self, *, username: str | None, text: str) -> bool:
         """Return whether this route should receive the request."""
 
         if not self.enabled:

src/telegram_resender/service.py

@@ -40,9 +40,9 @@ def __init__(
 
     @property
     def whitelist_count(self) -> int:
-        """Number of usernames currently allowed by the service."""
+        """Number of Telegram user ids currently allowed by the service."""
 
-        return len(self._whitelist.usernames)
+        return len(self._whitelist.user_ids)
 
     def reload_whitelist(self, path: Path) -> int:
         """Reload whitelist from disk and return the new user count."""
@@ -53,23 +53,22 @@ def reload_whitelist(self, path: Path) -> int:
     def handle_text(self, message: IncomingMessage) -> ForwardingDecision:
         """Decide whether a text message should be forwarded.
 
-        A missing Telegram username is denied even if the chat id is known. The original
-        bot used usernames as the trust boundary, so this keeps the rule explicit and
-        testable instead of silently broadening access.
+        A missing Telegram numeric user id is denied explicitly. Usernames are public,
+        mutable handles, so they must not be used as the trust boundary.
         """
 
-        if message.user.username is None:
+        if message.user.id is None:
             return ForwardingDecision(
                 should_forward=False,
                 response_text=self._missing_username_message.format(chat_id=message.chat_id),
-                reason="missing_username",
+                reason="missing_user_id",
             )
 
-        if not self._whitelist.contains(message.user.username):
+        if not self._whitelist.contains(message.user.id):
             return ForwardingDecision(
                 should_forward=False,
                 response_text=self._access_denied_message,
-                reason="unknown_username",
+                reason="unknown_user_id",
             )
 
         if not message.text.strip() or not self._looks_like_request(message.text):
@@ -108,7 +107,7 @@ def handle_text(self, message: IncomingMessage) -> ForwardingDecision:
         return ForwardingDecision(
             should_forward=True,
             response_text=self._request_accepted_message,
-            reason="allowed_username",
+            reason="allowed_user_id",
             forward_text=forward_text,
             request_id=request_id,
             target_chat_ids=tuple(route.target_chat_id for route in matching_routes),

src/telegram_resender/settings.py

@@ -34,7 +34,7 @@ class Settings(BaseSettings):
     )
     whitelist_path: Path = Field(
         default=Path("whitelist.csv"),
-        description="CSV file with allowed Telegram usernames.",
+        description="CSV file with allowed Telegram numeric user IDs.",
         validation_alias=AliasChoices("TELEGRAM_RESENDER_WHITELIST_PATH", "WHITELIST_PATH"),
     )
     routes_path: Path | None = Field(

src/telegram_resender/whitelist.py

@@ -7,8 +7,25 @@
 from pathlib import Path
 
 
+def parse_user_id(value: int | str | None) -> int | None:
+    """Parse an immutable Telegram numeric user id from configuration or updates."""
+
+    if value is None:
+        return None
+    if isinstance(value, int):
+        return value
+    stripped = value.strip()
+    if not stripped:
+        return None
+    try:
+        return int(stripped)
+    except ValueError as exc:
+        msg = f"Telegram user id must be an integer: {value!r}"
+        raise ValueError(msg) from exc
+
+
 def normalize_username(username: str | None) -> str | None:
-    """Normalize Telegram usernames for case-insensitive whitelist matching."""
+    """Normalize Telegram usernames for display-only route matching."""
 
     if username is None:
         return None
@@ -17,15 +34,15 @@ def normalize_username(username: str | None) -> str | None:
 
 
 class Whitelist:
-    """Immutable set of allowed Telegram usernames."""
+    """Immutable set of allowed Telegram numeric user ids."""
 
-    def __init__(self, usernames: Iterable[str]) -> None:
-        normalized = {username for item in usernames if (username := normalize_username(item))}
-        self._usernames = frozenset(normalized)
+    def __init__(self, user_ids: Iterable[int | str]) -> None:
+        parsed = {user_id for item in user_ids if (user_id := parse_user_id(item)) is not None}
+        self._user_ids = frozenset(parsed)
 
     @classmethod
     def from_file(cls, path: Path) -> Whitelist:
-        """Load usernames from the first column of a CSV file.
+        """Load Telegram numeric user ids from the first column of a CSV file.
 
         The CSV reader is intentionally used even for one-value-per-line files: it keeps
         quoted values, UTF-8 BOMs, and future metadata columns predictable.
@@ -35,30 +52,36 @@ def from_file(cls, path: Path) -> Whitelist:
             msg = f"Whitelist file does not exist: {path}"
             raise FileNotFoundError(msg)
 
-        usernames: list[str] = []
+        user_ids: list[str] = []
         with path.open("r", encoding="utf-8-sig", newline="") as stream:
             for row in csv.reader(stream):
                 if not row:
                     continue
                 candidate = row[0].strip()
                 if candidate and not candidate.startswith("#"):
-                    usernames.append(candidate)
-        return cls(usernames)
+                    user_ids.append(candidate)
+        return cls(user_ids)
+
+    def contains(self, user_id: int | str | None) -> bool:
+        """Return whether a Telegram numeric user id may submit requests."""
 
-    def contains(self, username: str | None) -> bool:
-        """Return whether a Telegram username is allowed to submit requests."""
+        parsed = parse_user_id(user_id)
+        return parsed in self._user_ids if parsed is not None else False
+
+    @property
+    def user_ids(self) -> frozenset[int]:
+        """Allowed Telegram numeric user ids."""
 
-        normalized = normalize_username(username)
-        return normalized in self._usernames if normalized else False
+        return self._user_ids
 
     @property
     def usernames(self) -> frozenset[str]:
-        """Allowed usernames normalized to lowercase and without leading ``@``."""
+        """Backward-compatible empty username set; usernames are not trusted."""
 
-        return self._usernames
+        return frozenset()
 
     @property
     def is_empty(self) -> bool:
         """Whether the whitelist contains no users."""
 
-        return not self._usernames
+        return not self._user_ids

tests/conversation/test_conversation.py

@@ -25,7 +25,7 @@ def test_conversation_commands_and_request_flow() -> None:
     """Mirror an operator conversation without Telegram SDK dependency."""
 
     service = ResenderService(
-        whitelist=Whitelist(["alice"]),
+        whitelist=Whitelist([10]),
         formatter=MessageFormatter(),
         request_accepted_message=REQUEST_ACCEPTED_MESSAGE,
         access_denied_message=RU_MESSAGES.access_denied_unknown,
@@ -55,7 +55,7 @@ def _run(message: IncomingMessage) -> str:
             IncomingMessage(
                 chat_id=100,
                 text=VALID_REQUEST,
-                user=UserProfile(username="alice"),
+                user=UserProfile(id=10, username="alice"),
             )
         )
     )
@@ -64,7 +64,7 @@ def _run(message: IncomingMessage) -> str:
             IncomingMessage(
                 chat_id=100,
                 text=VALID_REQUEST,
-                user=UserProfile(username="stranger"),
+                user=UserProfile(id=30, username="stranger"),
             )
         )
     )
@@ -73,7 +73,7 @@ def _run(message: IncomingMessage) -> str:
             IncomingMessage(
                 chat_id=100,
                 text="",
-                user=UserProfile(username=None),
+                user=UserProfile(id=None, username=None),
             )
         )
     )
@@ -82,7 +82,7 @@ def _run(message: IncomingMessage) -> str:
             IncomingMessage(
                 chat_id=100,
                 text="hi",
-                user=UserProfile(username="alice"),
+                user=UserProfile(id=10, username="alice"),
             )
         )
     )

tests/integration/test_bootstrap.py

@@ -19,7 +19,7 @@ def test_build_service_uses_project_files(tmp_path: Path) -> None:
     """Settings should load and inject dependencies without direct Telegram SDK calls."""
 
     whitelist_file = tmp_path / "whitelist.csv"
-    whitelist_file.write_text("alice\n", encoding="utf-8")
+    whitelist_file.write_text("10\n", encoding="utf-8")
     settings = Settings(
         bot_token="123:abc",
         forward_chat_id=100,
@@ -30,7 +30,7 @@ def test_build_service_uses_project_files(tmp_path: Path) -> None:
         IncomingMessage(
             chat_id=100,
             text=VALID_REQUEST,
-            user=UserProfile(username="alice"),
+            user=UserProfile(id=10, username="alice"),
         )
     )
 

tests/unit/test_app.py

@@ -83,7 +83,7 @@ async def answer(self, text: str) -> None:
 
 def _settings(tmp_path: Path, *, confirm_before_forward: bool = False) -> Settings:
     whitelist_path = tmp_path / "whitelist.csv"
-    whitelist_path.write_text("alice\n", encoding="utf-8")
+    whitelist_path.write_text("10\n", encoding="utf-8")
     return Settings(
         bot_token="123:abc",
         forward_chat_id=200,
@@ -96,7 +96,7 @@ def _settings(tmp_path: Path, *, confirm_before_forward: bool = False) -> Settin
 
 def _settings_with_routes(tmp_path: Path) -> Settings:
     whitelist_path = tmp_path / "whitelist.csv"
-    whitelist_path.write_text("alice\n", encoding="utf-8")
+    whitelist_path.write_text("10\n", encoding="utf-8")
     routes_path = tmp_path / "routes.json"
     routes_path.write_text(
         """
@@ -216,7 +216,7 @@ async def test_admin_can_reload_whitelist(tmp_path: Path) -> None:
     router = create_router(settings, service)
     handlers = _message_handlers(router)
     whitelist_path = settings.whitelist_path
-    whitelist_path.write_text("alice\nmallory\n", encoding="utf-8")
+    whitelist_path.write_text("10\n999\n", encoding="utf-8")
     admin = FakeMessage(user_id=10)
 
     await handlers["reload_whitelist"](admin)

tests/unit/test_cli.py

@@ -57,7 +57,7 @@ def test_doctor_reports_loaded_configuration(
     """Doctor should validate settings and whitelist without polling Telegram."""
 
     whitelist_path = tmp_path / "whitelist.csv"
-    whitelist_path.write_text("alice\nbob\n", encoding="utf-8")
+    whitelist_path.write_text("10\n20\n", encoding="utf-8")
     monkeypatch.chdir(tmp_path)
     monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", "123:abc")
     monkeypatch.setenv("TELEGRAM_RESENDER_FORWARD_CHAT_ID", "100")
@@ -107,7 +107,7 @@ def test_health_reports_ok_for_valid_configuration(
     """Health should be concise for process managers."""
 
     whitelist_path = tmp_path / "whitelist.csv"
-    whitelist_path.write_text("alice\n", encoding="utf-8")
+    whitelist_path.write_text("10\n", encoding="utf-8")
     monkeypatch.setenv("TELEGRAM_RESENDER_BOT_TOKEN", "123:abc")
     monkeypatch.setenv("TELEGRAM_RESENDER_FORWARD_CHAT_ID", "100")
     monkeypatch.setenv("TELEGRAM_RESENDER_WHITELIST_PATH", str(whitelist_path))