Merge pull request chigwell#233 from omid-io/feat/chat-access-allowlist

feat(security): add per-chat allowlist filter for privacy control (chigwell#212)

Author: chigwell

README.md

@@ -33,6 +33,7 @@ Message sent successfully:
 - [Device Identity](#device-identity)
 - [Proxy Support](#proxy-support)
 - [File Path Security](#file-path-security)
+- [Chat Access Privacy (Allowlist)](#chat-access-privacy-allowlist)
 - [Docker](#docker)
 - [Development](#development)
 - [Security Notes](#security-notes)
@@ -597,6 +598,23 @@ From an MCP client configuration, pass the same roots after `main.py`:
 }
 ```
 
+## Chat Access Privacy (Allowlist)
+
+To restrict AI agents or MCP clients to specific chats only (preventing access to all private or corporate conversations), configure `TELEGRAM_ALLOWED_CHAT_IDS`:
+
+```env
+# Comma-separated list of allowed chat IDs, supergroup IDs, or usernames
+TELEGRAM_ALLOWED_CHAT_IDS=12345678,-100123456789,@allowed_channel
+```
+
+When `TELEGRAM_ALLOWED_CHAT_IDS` is set:
+- **`list_chats` / `get_chats`:** Only chats matching the allowlist are returned; all other conversations remain completely invisible to the agent.
+- **Messaging & Chat Tools (`send_message`, `get_messages`, `get_chat`, `create_poll`, etc.):** Any attempt to interact with a chat outside the allowlist is rejected with a structured error message: `Access to chat '<chat_id>' is restricted by privacy policy (TELEGRAM_ALLOWED_CHAT_IDS)`.
+- **Search & Drafts (`search_global`, `get_drafts`):** Global searches and draft listings omit unallowed conversations.
+- **Incoming Events:** Notifications and debounce feeds only process events from allowed chats.
+
+If `TELEGRAM_ALLOWED_CHAT_IDS` is unset or empty, the server operates in unrestricted mode (default), preserving 100% backward compatibility.
+
 ## Docker
 
 Build the image:

main.py

@@ -16,14 +16,34 @@
 from telegram_mcp.runner import _main, main
 from telegram_mcp.tools import *
 
-# Backward-compatible alias for callers/tests that monkeypatch main.SERVER_ALLOWED_ROOTS.
+# Backward-compatible alias for callers/tests that monkeypatch main.SERVER_ALLOWED_ROOTS / main.ALLOWED_CHAT_IDS.
 SERVER_ALLOWED_ROOTS = _runtime.SERVER_ALLOWED_ROOTS
+ALLOWED_CHAT_IDS = _runtime.ALLOWED_CHAT_IDS
 
 
 def _sync_runtime_roots() -> None:
     _runtime.SERVER_ALLOWED_ROOTS = SERVER_ALLOWED_ROOTS
 
 
+def _sync_runtime_chat_allowlist() -> None:
+    _runtime.ALLOWED_CHAT_IDS = ALLOWED_CHAT_IDS
+
+
+def is_chat_allowlist_enabled() -> bool:
+    _sync_runtime_chat_allowlist()
+    return _runtime.is_chat_allowlist_enabled()
+
+
+def is_chat_allowed(chat_identifier=None, entity=None) -> bool:
+    _sync_runtime_chat_allowlist()
+    return _runtime.is_chat_allowed(chat_identifier, entity)
+
+
+def check_chat_access(chat_identifier=None, entity=None):
+    _sync_runtime_chat_allowlist()
+    return _runtime.check_chat_access(chat_identifier, entity)
+
+
 async def _get_effective_allowed_roots(ctx):
     _sync_runtime_roots()
     return await _runtime._get_effective_allowed_roots(ctx)

telegram_mcp/runtime.py

@@ -13,7 +13,7 @@
 from difflib import SequenceMatcher
 from datetime import datetime, timedelta, timezone
 from enum import Enum
-from typing import List, Dict, Optional, Union, Any, get_args
+from typing import List, Dict, Optional, Union, Any, Iterable, get_args
 from pathlib import Path
 from urllib.parse import unquote, urlparse
 
@@ -879,6 +879,137 @@ async def ensure_connected(cl: TelegramClient = None):
 ROOTS_REQUEST_TIMEOUT_DEFAULT = 10.0
 
 
+# Per-chat access control allowlist configuration (TELEGRAM_ALLOWED_CHAT_IDS)
+ALLOWED_CHAT_IDS: Optional[set[Union[int, str]]] = None
+CHAT_PARAM_NAMES: frozenset[str] = frozenset({"chat_id", "from_chat_id", "to_chat_id", "channel"})
+
+
+def _parse_allowed_chat_ids(
+    raw: Optional[Union[str, Iterable[Union[int, str]]]],
+) -> Optional[set[Union[int, str]]]:
+    """Parse TELEGRAM_ALLOWED_CHAT_IDS into a set of allowed IDs and usernames.
+
+    Supports comma-separated integer IDs (e.g. '12345678,-100123456789') and
+    usernames/handles (e.g. '@mychat,other_channel').
+    Also automatically indexes marked variants for bare integers and vice versa
+    so that both marked IDs (-100...) and bare positive IDs match.
+    """
+    if raw is None:
+        return None
+    if isinstance(raw, str):
+        raw = raw.strip()
+        if not raw:
+            return None
+        tokens = [t.strip() for t in raw.split(",") if t.strip()]
+    elif isinstance(raw, (list, tuple, set)):
+        tokens = [str(t).strip() for t in raw if str(t).strip()]
+    else:
+        return None
+
+    if not tokens:
+        return None
+
+    allowed: set[Union[int, str]] = set()
+    for token in tokens:
+        try:
+            val = int(token)
+            allowed.add(val)
+            # If negative supergroup: -100XXXXXXXXXX
+            if str(val).startswith("-100") and len(str(val)) > 4:
+                try:
+                    channel_id = int(str(val)[4:])
+                    allowed.add(channel_id)
+                except ValueError:
+                    pass
+            # If positive bare ID: add supergroup (-100...) and group (-) variants
+            elif val > 0:
+                allowed.add(-1000000000000 - val)
+                allowed.add(-val)
+            elif val < 0:
+                # Basic group negative ID: -XXXXXX
+                allowed.add(-val)
+        except ValueError:
+            clean_handle = token.lstrip("@").strip().lower()
+            if clean_handle:
+                allowed.add(clean_handle)
+
+    return allowed if allowed else None
+
+
+def _load_allowed_chat_ids() -> Optional[set[Union[int, str]]]:
+    """Load ALLOWED_CHAT_IDS from the TELEGRAM_ALLOWED_CHAT_IDS environment variable."""
+    return _parse_allowed_chat_ids(os.getenv("TELEGRAM_ALLOWED_CHAT_IDS"))
+
+
+# Initial load from environment
+ALLOWED_CHAT_IDS = _load_allowed_chat_ids()
+
+
+def get_effective_allowed_chat_ids() -> Optional[set[Union[int, str]]]:
+    """Return the currently effective set of allowed chat IDs, or None if allowlist is disabled."""
+    env_raw = os.getenv("TELEGRAM_ALLOWED_CHAT_IDS")
+    if env_raw is not None:
+        return _parse_allowed_chat_ids(env_raw)
+    return ALLOWED_CHAT_IDS
+
+
+def is_chat_allowlist_enabled() -> bool:
+    """Return True if chat allowlist filtering is active."""
+    return get_effective_allowed_chat_ids() is not None
+
+
+def is_chat_allowed(chat_identifier: Any, entity: Any = None) -> bool:
+    """Check whether a chat identifier or entity is permitted by the allowlist.
+
+    If allowlist is not enabled, always returns True.
+    """
+    allowed = get_effective_allowed_chat_ids()
+    if allowed is None:
+        return True
+
+    if chat_identifier is not None:
+        if isinstance(chat_identifier, int):
+            if chat_identifier in allowed:
+                return True
+        elif isinstance(chat_identifier, str):
+            try:
+                int_id = int(chat_identifier)
+                if int_id in allowed:
+                    return True
+            except ValueError:
+                clean = chat_identifier.lstrip("@").strip().lower()
+                if clean and clean in allowed:
+                    return True
+
+    if entity is not None:
+        try:
+            marked_id = get_marked_id(entity)
+            if marked_id in allowed:
+                return True
+        except Exception:
+            pass
+
+        bare_id = getattr(entity, "id", None)
+        if isinstance(bare_id, int) and bare_id in allowed:
+            return True
+
+        username = getattr(entity, "username", None)
+        if username and str(username).lower() in allowed:
+            return True
+
+    return False
+
+
+def check_chat_access(chat_identifier: Any, entity: Any = None) -> Optional[str]:
+    """Return an error message if chat access is restricted, or None if allowed."""
+    if not is_chat_allowed(chat_identifier, entity):
+        return (
+            f"Access to chat '{chat_identifier}' is restricted by privacy policy "
+            "(TELEGRAM_ALLOWED_CHAT_IDS)."
+        )
+    return None
+
+
 # Error code prefix mapping for better error tracing
 class ErrorCategory(str, Enum):
     CHAT = "CHAT"
@@ -890,6 +1021,13 @@ class ErrorCategory(str, Enum):
     AUTH = "AUTH"
     ADMIN = "ADMIN"
     FOLDER = "FOLDER"
+    PRIVACY = "PRIVACY"
+
+
+class ChatAccessDeniedError(Exception):
+    """Exception raised when access to a chat is restricted by privacy policy."""
+
+    pass
 
 
 def _is_flood_wait(error: Exception) -> bool:
@@ -1075,6 +1213,34 @@ def validate_single_id(value, p_name):
                         )
                     kwargs[param_name] = validated_value
 
+                # Per-chat privacy allowlist enforcement
+                if is_chat_allowlist_enabled() and param_name in CHAT_PARAM_NAMES:
+                    check_target = kwargs[param_name]
+                    target_items = (
+                        check_target if isinstance(check_target, list) else [check_target]
+                    )
+                    for item in target_items:
+                        if not is_chat_allowed(item):
+                            resolved_allowed = False
+                            if isinstance(item, str):
+                                try:
+                                    cl = get_client(kwargs.get("account"))
+                                    if cl:
+                                        ent = await resolve_entity(item, cl)
+                                        if is_chat_allowed(item, ent):
+                                            resolved_allowed = True
+                                except Exception:
+                                    pass
+                            if not resolved_allowed:
+                                err = check_chat_access(item)
+                                return log_and_format_error(
+                                    func.__name__,
+                                    ChatAccessDeniedError(err),
+                                    prefix=ErrorCategory.PRIVACY,
+                                    user_message=err,
+                                    **{param_name: param_value},
+                                )
+
             return await func(*args, **kwargs)
 
         return wrapper

telegram_mcp/tools/chats.py

@@ -172,6 +172,8 @@ async def get_chats(account: str = None, page: int = 1, page_size: int = 20) ->
             dialogs = await cl.get_dialogs()
         except BotMethodInvalidError:
             return "Listing chats/dialogs is not supported for bot accounts (Telegram API restriction: bots cannot fetch dialog lists)."
+        if is_chat_allowlist_enabled():
+            dialogs = [d for d in dialogs if is_chat_allowed(get_marked_id(d.entity), d.entity)]
         start = (page - 1) * page_size
         end = start + page_size
         if start >= len(dialogs):
@@ -492,6 +494,10 @@ async def list_chats(
         for dialog in dialogs:
             entity = dialog.entity
 
+            # Enforce privacy allowlist
+            if is_chat_allowlist_enabled() and not is_chat_allowed(get_marked_id(entity), entity):
+                continue
+
             # Filter by type if requested
             current_type = get_entity_filter_type(entity)
 
@@ -612,6 +618,16 @@ async def get_chat(chat_id: Union[int, str], account: str = None) -> str:
         cl = get_client(account)
         entity = await resolve_entity(chat_id, cl)
 
+        if is_chat_allowlist_enabled() and not is_chat_allowed(chat_id, entity):
+            err = check_chat_access(chat_id, entity)
+            return log_and_format_error(
+                "get_chat",
+                ChatAccessDeniedError(err),
+                prefix=ErrorCategory.PRIVACY,
+                user_message=err,
+                chat_id=chat_id,
+            )
+
         record = {"id": get_marked_id(entity)}
 
         is_user = isinstance(entity, User)
@@ -710,7 +726,10 @@ async def search_public_chats(query: str, limit: int = 20, account: str = None)
         cl = get_client(account)
         await ensure_connected(cl)
         result = await cl(functions.contacts.SearchRequest(q=query, limit=limit))
-        entities = [format_entity(e) for e in result.chats + result.users]
+        all_entities = result.chats + result.users
+        if is_chat_allowlist_enabled():
+            all_entities = [e for e in all_entities if is_chat_allowed(get_marked_id(e), e)]
+        entities = [format_entity(e) for e in all_entities]
         return json.dumps(entities, indent=2)
     except Exception as e:
         return log_and_format_error("search_public_chats", e, query=query, limit=limit)
@@ -737,6 +756,7 @@ async def resolve_username(username: str, account: str = None) -> str:
     annotations=ToolAnnotations(title="Get Full Chat", openWorldHint=True, readOnlyHint=True)
 )
 @with_account(readonly=True)
+@validate_id("chat_id")
 async def get_full_chat(chat_id: Union[int, str], account: str = None) -> str:
     """
     Get full info of a channel or group including description/about text.
@@ -752,6 +772,16 @@ async def get_full_chat(chat_id: Union[int, str], account: str = None) -> str:
         await ensure_connected(cl)
         entity = await resolve_entity(chat_id, cl)
 
+        if is_chat_allowlist_enabled() and not is_chat_allowed(chat_id, entity):
+            err = check_chat_access(chat_id, entity)
+            return log_and_format_error(
+                "get_full_chat",
+                ChatAccessDeniedError(err),
+                prefix=ErrorCategory.PRIVACY,
+                user_message=err,
+                chat_id=chat_id,
+            )
+
         # Basic ("legacy") groups are not channels: GetFullChannelRequest cannot
         # cast an InputPeerChat and raises TypeError. They are served by
         # messages.GetFullChatRequest instead.

telegram_mcp/tools/events.py

@@ -56,6 +56,8 @@ def _scan_settled(
     """
     soonest_remaining = None
     for cid, rec in list(_pending_msgs.items()):
+        if is_chat_allowlist_enabled() and not is_chat_allowed(cid):
+            continue
         if only is not None and cid != only:
             continue
         quiet = now - rec["last_ts"]
@@ -207,6 +209,8 @@ async def _on_new_incoming(event) -> None:
         if getattr(sender, "bot", False) or getattr(sender, "is_self", False):
             return
         chat_id = event.chat_id
+        if is_chat_allowlist_enabled() and not is_chat_allowed(chat_id, sender):
+            return
         now = time.monotonic()
         msg_id = event.message.id
         rec = _pending_msgs.get(chat_id)
@@ -252,6 +256,7 @@ def register_incoming_handlers() -> None:
         title="Wait For New Message", openWorldHint=True, readOnlyHint=True
     )
 )
+@validate_id("chat_id")
 async def wait_for_new_message(
     timeout: float = 50.0,
     chat_id: Optional[Union[int, str]] = None,
@@ -283,7 +288,10 @@ async def wait_for_new_message(
         deadline = time.monotonic() + timeout
         while True:
             pending = {
-                cid: rec for cid, rec in _pending_msgs.items() if target is None or cid == target
+                cid: rec
+                for cid, rec in _pending_msgs.items()
+                if (target is None or cid == target)
+                and (not is_chat_allowlist_enabled() or is_chat_allowed(cid))
             }
             if pending:
                 chats = [
@@ -323,6 +331,7 @@ async def wait_for_new_message(
         title="Wait For Settled Message", openWorldHint=True, readOnlyHint=True
     )
 )
+@validate_id("chat_id")
 async def wait_for_settled_message(
     settle_ms: int = 6000,
     max_wait_ms: int = 50000,