Merge pull request chigwell#260 from vsem-azamat/feat/server-roots-only

feat(roots): opt-in server-only roots; pin symlink escape tests

Author: chigwell

.env.example

@@ -80,6 +80,9 @@ TELEGRAM_SESSION_NAME=telegram_session
 # if server CLI roots are configured. Set this to 1/true to fall back to the
 # server CLI roots in those cases. Default is deny-all.
 # TELEGRAM_ALLOW_SERVER_ROOTS_FALLBACK=1
+# Use the server roots as-is and never send roots/list (for clients that never
+# answer it). No effect without server roots. Default is off.
+# TELEGRAM_SERVER_ROOTS_ONLY=1
 
 # --- HTTP/SSE transport security (optional) ---
 # Only relevant when MCP_TRANSPORT=http or sse. If the server is reachable via

README.md

@@ -642,6 +642,9 @@ Security behavior:
   `TELEGRAM_ALLOW_SERVER_ROOTS_FALLBACK=1` to fall back to the server CLI roots
   in that case (opt-in; the default stays deny-all). The same opt-in also applies
   when `list_roots` fails unexpectedly and no client paths could be recovered.
+- `roots/list` times out after `TELEGRAM_ROOTS_TIMEOUT_SECONDS` (default 10).
+  With server roots configured, `TELEGRAM_SERVER_ROOTS_ONLY=1` uses them directly
+  and skips `roots/list`; client Roots are then ignored.
 - Paths are resolved through real paths and must stay inside an allowed root.
 - Traversal, wildcard-like, shell-like, and null-byte path patterns are rejected.
 - Relative paths resolve under the first allowed root.
@@ -844,6 +847,7 @@ Telegram messages, display names, chat titles, and button labels are untrusted c
 - **Database is locked:** make sure no other process is using the same file session.
 - **`AuthKeyDuplicatedError` / "Another telegram-mcp process is already connected with this session":** two processes tried to connect the same Telegram session at once (e.g. an MCP client restarted the connector before the old process exited), which Telegram rejects and can invalidate the session for both. The server now takes an exclusive lock per session before connecting; a second concurrent launch waits briefly (default 20s, override with `TELEGRAM_LOCK_GRACE_SECONDS`) for the first to release it and otherwise exits without ever calling `connect()`, instead of racing into a duplicate connection. Retry once only one instance is running — the refusal names the PID holding the lock. If several instances on this host are meant to share one session (all reaching Telegram from the same IP), set `TELEGRAM_SESSION_LOCK=shared`; see [Sharing one session from one host](#sharing-one-session-from-one-host).
 - **File tools are disabled:** pass allowed roots or configure MCP Roots in your client.
+- **File tools wait for `roots/list`:** configure server roots and set `TELEGRAM_SERVER_ROOTS_ONLY=1`.
 - **Path rejected:** ensure the path is inside an allowed root and does not use traversal or wildcard patterns.
 - **Auth errors after password changes:** regenerate your session string.
 - **Bot-only tool rejected:** regular user accounts cannot manage bot command settings.

telegram_mcp/runtime.py

@@ -926,6 +926,7 @@ async def ensure_connected(cl: TelegramClient = None):
 ROOTS_STATUS_UNSUPPORTED_FALLBACK = "unsupported_fallback"
 ROOTS_STATUS_CLIENT_DENY_ALL = "client_deny_all"
 ROOTS_STATUS_SERVER_FALLBACK = "server_fallback"
+ROOTS_STATUS_SERVER_ONLY = "server_only"
 ROOTS_STATUS_ERROR = "error"
 ROOTS_STATUS_TIMEOUT = "timeout"
 # Some clients accept the server-initiated roots/list request but never answer
@@ -2156,6 +2157,12 @@ def _server_roots_fallback_enabled(value: Optional[str] = None) -> bool:
     return _parse_bool_env(raw_value, False)
 
 
+def _server_roots_only_enabled(value: Optional[str] = None) -> bool:
+    """TELEGRAM_SERVER_ROOTS_ONLY: use server roots and skip roots/list (default off)."""
+    raw_value = os.getenv("TELEGRAM_SERVER_ROOTS_ONLY") if value is None else value
+    return _parse_bool_env(raw_value, False)
+
+
 def _roots_request_timeout(value: Optional[str] = None) -> Optional[float]:
     """Seconds to wait for the client's ``roots/list`` reply.
 
@@ -2180,6 +2187,8 @@ async def _get_effective_allowed_roots_with_status(
         if fallback_roots:
             return fallback_roots, ROOTS_STATUS_READY
         return [], ROOTS_STATUS_NOT_CONFIGURED
+    if fallback_roots and _server_roots_only_enabled():
+        return fallback_roots, ROOTS_STATUS_SERVER_ONLY
 
     try:
         timeout = _roots_request_timeout()
@@ -2269,6 +2278,7 @@ async def _ensure_allowed_roots(
                 (
                     f"{tool_name} is disabled because the MCP client never answered the "
                     "roots/list request. Configure server CLI roots and set "
+                    "TELEGRAM_SERVER_ROOTS_ONLY=1 (skip roots/list) or "
                     "TELEGRAM_ALLOW_SERVER_ROOTS_FALLBACK=1, or raise "
                     "TELEGRAM_ROOTS_TIMEOUT_SECONDS."
                 ),

tests/test_file_path_security.py

@@ -100,6 +100,87 @@ async def test_readable_path_rejects_outside_root(tmp_path, monkeypatch):
     assert error == "Path is outside allowed roots."
 
 
+def _symlink(link, target, *, directory=False):
+    try:
+        link.symlink_to(target, target_is_directory=directory)
+    except (OSError, NotImplementedError) as exc:
+        pytest.skip(f"symlinks are not available here: {exc}")
+
+
+def _symlinked_layout(tmp_path):
+    """An allowed root with symlinks pointing out of it, plus an outside dir."""
+    root = (tmp_path / "root").resolve()
+    outside = (tmp_path / "outside").resolve()
+    root.mkdir()
+    outside.mkdir()
+    (outside / "secret.txt").write_text("secret")
+    _symlink(root / "secret_link.txt", outside / "secret.txt")
+    _symlink(root / "outside_dir", outside, directory=True)
+    _symlink(root / "dangling.bin", outside / "created.bin")
+    return root, outside
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("raw_path", ["secret_link.txt", "outside_dir/secret.txt"])
+async def test_readable_path_rejects_symlink_escape(tmp_path, monkeypatch, raw_path):
+    root, _outside = _symlinked_layout(tmp_path)
+    monkeypatch.setattr(main, "SERVER_ALLOWED_ROOTS", [root])
+
+    for path in (raw_path, str(root / raw_path)):
+        resolved, error = await main._resolve_readable_file_path(
+            raw_path=path, ctx=None, tool_name="send_file"
+        )
+        assert resolved is None
+        assert error == "Path is outside allowed roots."
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "raw_path",
+    ["dangling.bin", "outside_dir/new.bin", "outside_dir/nested/new.bin"],
+)
+async def test_writable_path_rejects_symlink_escape(tmp_path, monkeypatch, raw_path):
+    root, outside = _symlinked_layout(tmp_path)
+    monkeypatch.setattr(main, "SERVER_ALLOWED_ROOTS", [root])
+
+    resolved, error = await main._resolve_writable_file_path(
+        raw_path=raw_path,
+        default_filename="ignored.bin",
+        ctx=None,
+        tool_name="download_media",
+    )
+
+    assert resolved is None
+    assert error == "Path is outside allowed roots."
+    # Rejection happens before any directory is created on the target side.
+    assert sorted(p.name for p in outside.iterdir()) == ["secret.txt"]
+
+
+@pytest.mark.asyncio
+async def test_symlinks_that_stay_inside_root_resolve_to_their_target(tmp_path, monkeypatch):
+    root = (tmp_path / "root").resolve()
+    (root / "real").mkdir(parents=True)
+    target = root / "real" / "file.txt"
+    target.write_text("ok")
+    _symlink(root / "linked", root / "real", directory=True)
+    monkeypatch.setattr(main, "SERVER_ALLOWED_ROOTS", [root])
+
+    readable, error = await main._resolve_readable_file_path(
+        raw_path="linked/file.txt", ctx=None, tool_name="send_file"
+    )
+    assert error is None
+    assert readable == target
+
+    writable, error = await main._resolve_writable_file_path(
+        raw_path="linked/new.bin",
+        default_filename="ignored.bin",
+        ctx=None,
+        tool_name="download_media",
+    )
+    assert error is None
+    assert writable == root / "real" / "new.bin"
+
+
 @pytest.mark.asyncio
 async def test_client_roots_replace_server_allowlist(tmp_path, monkeypatch):
     server_root = (tmp_path / "server_root").resolve()

tests/test_runtime.py

@@ -1205,6 +1205,99 @@ async def test_list_roots_timeout_denies_without_opt_in(tmp_path, monkeypatch):
     )
     assert error is not None
     assert "roots/list" in error
+    assert "TELEGRAM_SERVER_ROOTS_ONLY" in error
+
+
+class _RootsSessionThatMustNotBeCalled:
+    """Client that would hang on roots/list; the server must not ask it."""
+
+    def __init__(self):
+        self.calls = 0
+
+    async def list_roots(self):
+        self.calls += 1
+        await asyncio.sleep(3600)
+
+
+def test_server_roots_only_enabled_parsing(monkeypatch):
+    monkeypatch.delenv("TELEGRAM_SERVER_ROOTS_ONLY", raising=False)
+    assert runtime._server_roots_only_enabled() is False
+    assert runtime._server_roots_only_enabled("1") is True
+    assert runtime._server_roots_only_enabled("true") is True
+    assert runtime._server_roots_only_enabled("off") is False
+    monkeypatch.setenv("TELEGRAM_SERVER_ROOTS_ONLY", "yes")
+    assert runtime._server_roots_only_enabled() is True
+
+
+@pytest.mark.asyncio
+async def test_server_roots_only_skips_client_roots_request(tmp_path, monkeypatch):
+    root = tmp_path / "root"
+    root.mkdir()
+    monkeypatch.setattr(runtime, "SERVER_ALLOWED_ROOTS", [root.resolve()])
+    monkeypatch.setenv("TELEGRAM_SERVER_ROOTS_ONLY", "1")
+    monkeypatch.delenv("TELEGRAM_ALLOW_SERVER_ROOTS_FALLBACK", raising=False)
+    monkeypatch.delenv("TELEGRAM_ROOTS_TIMEOUT_SECONDS", raising=False)
+    session = _RootsSessionThatMustNotBeCalled()
+    ctx = SimpleNamespace(session=session)
+
+    roots, status = await asyncio.wait_for(
+        runtime._get_effective_allowed_roots_with_status(ctx), timeout=1
+    )
+    assert status == runtime.ROOTS_STATUS_SERVER_ONLY
+    assert roots == [root.resolve()]
+
+    resolved, error = await asyncio.wait_for(
+        runtime._ensure_allowed_roots(ctx, "download_media"), timeout=1
+    )
+    assert error is None
+    assert resolved == [root.resolve()]
+    assert session.calls == 0
+
+
+@pytest.mark.asyncio
+async def test_server_roots_only_ignores_client_roots_when_server_roots_set(tmp_path, monkeypatch):
+    server_root = tmp_path / "server"
+    client_root = tmp_path / "client"
+    server_root.mkdir()
+    client_root.mkdir()
+    monkeypatch.setattr(runtime, "SERVER_ALLOWED_ROOTS", [server_root.resolve()])
+    monkeypatch.setenv("TELEGRAM_SERVER_ROOTS_ONLY", "1")
+
+    roots, status = await runtime._get_effective_allowed_roots_with_status(
+        _ctx_with_roots([SimpleNamespace(uri=client_root.as_uri())])
+    )
+    assert status == runtime.ROOTS_STATUS_SERVER_ONLY
+    assert roots == [server_root.resolve()]
+
+
+@pytest.mark.asyncio
+async def test_server_roots_only_keeps_client_roots_without_server_roots(tmp_path, monkeypatch):
+    client_root = tmp_path / "client"
+    client_root.mkdir()
+    monkeypatch.setattr(runtime, "SERVER_ALLOWED_ROOTS", [])
+    monkeypatch.setenv("TELEGRAM_SERVER_ROOTS_ONLY", "1")
+
+    roots, status = await runtime._get_effective_allowed_roots_with_status(
+        _ctx_with_roots([SimpleNamespace(uri=client_root.as_uri())])
+    )
+    assert status == runtime.ROOTS_STATUS_READY
+    assert roots == [client_root.resolve()]
+
+
+@pytest.mark.asyncio
+async def test_client_roots_still_replace_server_roots_by_default(tmp_path, monkeypatch):
+    server_root = tmp_path / "server"
+    client_root = tmp_path / "client"
+    server_root.mkdir()
+    client_root.mkdir()
+    monkeypatch.setattr(runtime, "SERVER_ALLOWED_ROOTS", [server_root.resolve()])
+    monkeypatch.delenv("TELEGRAM_SERVER_ROOTS_ONLY", raising=False)
+
+    roots, status = await runtime._get_effective_allowed_roots_with_status(
+        _ctx_with_roots([SimpleNamespace(uri=client_root.as_uri())])
+    )
+    assert status == runtime.ROOTS_STATUS_READY
+    assert roots == [client_root.resolve()]
 
 
 def test_get_file_extension_overrides_rejects_duplicate_tool_name():