diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 233ccb2f8..b27f2d9c4 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -7,6 +7,7 @@ import logging import math import mimetypes import re +import stat import time from collections import OrderedDict from collections.abc import Awaitable, Callable, Mapping @@ -52,6 +53,7 @@ from deerflow.skills.slash import parse_slash_skill_reference from deerflow.skills.storage import get_or_new_skill_storage from deerflow.skills.storage.skill_storage import SkillStorage from deerflow.trace_context import ensure_trace_context +from deerflow.uploads.manager import apply_upload_sandbox_permits from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY logger = logging.getLogger(__name__) @@ -1082,6 +1084,19 @@ def _prepare_artifact_delivery( return response_text, attachments +def _make_inbound_file_sandbox_readable(file_path: Path) -> None: + """Make a channel-downloaded upload readable by the sandbox process. + + The gateway writes inbound files as root with 0o600; in AIO/Docker sandbox + mode the sandbox runs as a non-root user on the bind-mounted path and + cannot read the file without group/other read bits. Delegates to the shared + apply_upload_sandbox_permits helper so the permission change stays bound to + the validated upload inode (O_NOFOLLOW + fchmod) and cannot be redirected + through a symlink swapped in after validation. + """ + apply_upload_sandbox_permits(file_path, stat.S_IRGRP | stat.S_IROTH) + + async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: str | None = None) -> list[dict[str, Any]]: if not msg.files: return [] @@ -1160,6 +1175,9 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: dest = uploads_dir / safe_name try: dest = await asyncio.to_thread(write_upload_file_no_symlink, uploads_dir, safe_name, data) + # Root-written 0o600 files are unreadable to the non-root + # sandbox; grant group/other read like the HTTP upload path. + await asyncio.to_thread(_make_inbound_file_sandbox_readable, dest) except UnsafeUploadPathError: logger.warning("[Manager] skipping inbound file with unsafe destination: %s", safe_name) continue diff --git a/backend/app/channels/telegram.py b/backend/app/channels/telegram.py index 0d323ecfa..eced42059 100644 --- a/backend/app/channels/telegram.py +++ b/backend/app/channels/telegram.py @@ -93,6 +93,14 @@ class TelegramChannel(Channel): self._thread: threading.Thread | None = None self._tg_loop: asyncio.AbstractEventLoop | None = None self._main_loop: asyncio.AbstractEventLoop | None = None + # Dedicated Bot for inbound file downloads. The Application's Bot is + # built on the manager/main loop (see start()), so its httpx + # connection pool is bound to that loop; opening a fresh download + # connection from the Telegram loop then trips "Event bound to a + # different event loop". A Bot initialized on the Telegram loop keeps + # its pool on the loop that performs the download. + self._download_bot: Any = None + self._download_bot_lock = asyncio.Lock() # Tasks submitted from the main dispatcher loop back to PTB's loop. # Only the Telegram loop mutates this set. self._tg_bridge_tasks: set[asyncio.Task[Any]] = set() @@ -175,7 +183,7 @@ class TelegramChannel(Channel): try: if telegram_loop and telegram_loop.is_running(): - drain_future = asyncio.run_coroutine_threadsafe(self._cancel_telegram_bridge_tasks(), telegram_loop) + drain_future = asyncio.run_coroutine_threadsafe(self._shutdown_telegram_resources(), telegram_loop) try: remaining = max(0.0, deadline - shutdown_loop.time()) await asyncio.wait_for(asyncio.wrap_future(drain_future), timeout=remaining) @@ -184,9 +192,9 @@ class TelegramChannel(Channel): raise except TimeoutError: drain_future.cancel() - logger.warning("[Telegram] timed out cancelling inbound file downloads during shutdown") + logger.warning("[Telegram] timed out shutting down Telegram loop resources during shutdown") except Exception as exc: - logger.warning("[Telegram] failed to cancel inbound file downloads during shutdown: %s", type(exc).__name__) + logger.warning("[Telegram] failed to shut down Telegram loop resources during shutdown: %s", type(exc).__name__) finally: if telegram_loop and telegram_loop.is_running(): try: @@ -462,7 +470,6 @@ class TelegramChannel(Channel): if not msg.files: return msg - bot = self._application.bot if self._application is not None else None materialized: list[dict[str, Any]] = [] unavailable: list[str] = [] @@ -478,6 +485,24 @@ class TelegramChannel(Channel): unavailable.append(f"{filename} (exceeds the 20 MB download limit)") continue + # Resolve the loop-bound download Bot. It is created on the + # Telegram loop so its httpx pool is bound to the loop that + # performs the download (the Application's Bot is main-loop-bound). + # First-time resolution performs a network getMe, so it runs inside + # its own guard: an init failure must mark only this attachment + # unavailable, never abort the whole message. + try: + bot = await self._get_download_bot() + except Exception as exc: + logger.error( + "[Telegram] failed to initialize download bot for inbound file %s: %s%s", + filename, + type(exc).__name__, + self._describe_download_cause(exc), + ) + unavailable.append(f"{filename} (download failed)") + continue + if bot is None or not file_id: logger.error("[Telegram] cannot download inbound file: %s", filename) unavailable.append(f"{filename} (download unavailable)") @@ -487,8 +512,14 @@ class TelegramChannel(Channel): resolved_size, content = await self._run_on_telegram_loop(self._download_inbound_file(bot, file_id)) except Exception as exc: # Exception strings from HTTP clients can contain request URLs. - # Log only the class name so a Bot API token can never leak. - logger.error("[Telegram] failed to download inbound file %s: %s", filename, type(exc).__name__) + # Log only class names so a Bot API token can never leak, but + # include the cause chain to distinguish timeout / reset / TLS. + logger.error( + "[Telegram] failed to download inbound file %s: %s%s", + filename, + type(exc).__name__, + self._describe_download_cause(exc), + ) unavailable.append(f"{filename} (download failed)") continue @@ -520,6 +551,69 @@ class TelegramChannel(Channel): # -- helpers ----------------------------------------------------------- + def _describe_download_cause(self, exc: BaseException) -> str: + """Build a token-safe suffix describing a download failure's cause. + + HTTP client exception strings can embed request URLs. Both Bot API URL + forms carry the token in the path — the file download URL + (``/file/bot/…``) and the method URLs (``/bot/getMe``, + ``/bot/getFile``) — so the configured token is removed first and + any remaining Bot API URL is collapsed before the cause chain is logged. + The cause class and message are kept to distinguish timeout / reset / TLS. + """ + cause = exc.__cause__ or exc.__context__ + if cause is None: + return "" + token = self.config.get("bot_token", "") + cause_msg = str(cause) + if token: + cause_msg = cause_msg.replace(token, "[redacted]") + cause_msg = re.sub(r"api\.telegram\.org/bot[^/\s]+/", "api.telegram.org/bot[redacted]/", cause_msg) + cause_msg = re.sub(r"api\.telegram\.org/file/\S+", "api.telegram.org/file/[redacted]", cause_msg) + cause_msg = cause_msg[:200] + return f" caused_by={type(cause).__name__}:{cause_msg}" + + async def _get_download_bot(self) -> Any: + """Return a PTB Bot whose httpx client is bound to the Telegram loop. + + The Application's Bot is constructed on the manager/main loop, so its + shared connection pool is bound there; opening a fresh download + connection from the Telegram loop fails with "bound to a different + event loop". This dedicated Bot is created and initialized on the + Telegram loop, keeping its pool on the loop that performs downloads. + """ + if self._download_bot is not None: + return self._download_bot + telegram_loop = self._tg_loop + if telegram_loop is None: + return None + + async def _init() -> Any: + if self._download_bot is not None: + return self._download_bot + async with self._download_bot_lock: + if self._download_bot is not None: + return self._download_bot + from telegram import Bot + + bot = Bot(token=self.config.get("bot_token", "")) + try: + await bot.initialize() + except BaseException: + # initialize() performs a network getMe and may have left + # partially-open HTTPX clients. Close them (no-op if the + # request objects never initialized) before the error + # propagates, and never cache a bot that failed to start. + try: + await bot.shutdown() + except BaseException: + logger.debug("[Telegram] failed to close partially-initialized download bot", exc_info=True) + raise + self._download_bot = bot + return bot + + return await self._run_on_telegram_loop(_init()) + async def _download_inbound_file(self, bot: Any, file_id: str) -> tuple[int | None, bytearray | None]: """Fetch one file entirely on the event loop that owns PTB's HTTP client.""" telegram_file = await bot.get_file(file_id) @@ -553,6 +647,32 @@ class TelegramChannel(Channel): if still_pending: logger.warning("[Telegram] %d inbound file download task(s) did not cancel promptly", len(still_pending)) + async def _shutdown_telegram_resources(self) -> None: + """Cancel in-flight PTB bridge work and close the download Bot. + + Runs on the Telegram loop (scheduled from ``stop()``) and must finish + before that loop is stopped, so the download Bot's HTTPX clients are + shut down on the loop that owns them. + """ + await self._cancel_telegram_bridge_tasks() + await self._shutdown_download_bot() + + async def _shutdown_download_bot(self) -> None: + """Close the download Bot's HTTPX clients on the Telegram loop. + + python-telegram-bot 22.7 exposes no ``Bot.session``; the HTTPX clients + are closed by ``Bot.shutdown()``, which is a no-op if the bot was never + initialized. Must run on the loop the bot was initialized on. + """ + bot = self._download_bot + if bot is None: + return + try: + await bot.shutdown() + except Exception: + logger.debug("[Telegram] failed to shut down download bot", exc_info=True) + self._download_bot = None + async def _run_on_telegram_loop(self, coroutine: Coroutine[Any, Any, Any]) -> Any: """Await a PTB coroutine without using its HTTP client across event loops.""" telegram_loop = self._tg_loop diff --git a/backend/app/gateway/routers/uploads.py b/backend/app/gateway/routers/uploads.py index 20211ca12..3e152e272 100644 --- a/backend/app/gateway/routers/uploads.py +++ b/backend/app/gateway/routers/uploads.py @@ -24,6 +24,7 @@ from deerflow.uploads.manager import ( UPLOAD_STAGING_SUFFIX, PathTraversalError, UnsafeUploadPathError, + apply_upload_sandbox_permits, claim_unique_filename, delete_file_safe, enrich_file_listing, @@ -123,16 +124,11 @@ def _make_file_sandbox_writable(file_path: os.PathLike[str] | str) -> None: In AIO sandbox mode, the gateway writes the authoritative host-side file first, then the sandbox runtime may rewrite the same mounted path. Granting world-writable access here prevents permission mismatches between the - gateway user and the sandbox runtime user. + gateway user and the sandbox runtime user. Delegates to the shared + apply_upload_sandbox_permits helper so the change stays bound to the + validated upload inode (O_NOFOLLOW + fchmod). """ - file_stat = os.lstat(file_path) - if stat.S_ISLNK(file_stat.st_mode): - logger.warning("Skipping sandbox chmod for symlinked upload path: %s", file_path) - return - - writable_mode = stat.S_IMODE(file_stat.st_mode) | stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH | stat.S_IRGRP | stat.S_IROTH - chmod_kwargs = {"follow_symlinks": False} if os.chmod in os.supports_follow_symlinks else {} - os.chmod(file_path, writable_mode, **chmod_kwargs) + apply_upload_sandbox_permits(file_path, stat.S_IWUSR | stat.S_IWGRP | stat.S_IWOTH | stat.S_IRGRP | stat.S_IROTH) def _make_file_sandbox_readable(file_path: os.PathLike[str] | str) -> None: @@ -142,16 +138,10 @@ def _make_file_sandbox_readable(file_path: os.PathLike[str] | str) -> None: permissions, then bind-mounts the host directory into the container. The sandbox process inside the container runs as a non-root user and cannot read those files without group/other read bits. This function adds - ``S_IRGRP | S_IROTH`` so the sandbox can read the uploaded content. + ``S_IRGRP | S_IROTH`` so the sandbox can read the uploaded content, via the + shared apply_upload_sandbox_permits helper (O_NOFOLLOW + fchmod). """ - file_stat = os.lstat(file_path) - if stat.S_ISLNK(file_stat.st_mode): - logger.warning("Skipping sandbox chmod for symlinked upload path: %s", file_path) - return - - readable_mode = stat.S_IMODE(file_stat.st_mode) | stat.S_IRGRP | stat.S_IROTH - chmod_kwargs = {"follow_symlinks": False} if os.chmod in os.supports_follow_symlinks else {} - os.chmod(file_path, readable_mode, **chmod_kwargs) + apply_upload_sandbox_permits(file_path, stat.S_IRGRP | stat.S_IROTH) def _uses_thread_data_mounts(sandbox_provider: SandboxProvider) -> bool: diff --git a/backend/packages/harness/deerflow/uploads/manager.py b/backend/packages/harness/deerflow/uploads/manager.py index ac8fad171..3204c87ab 100644 --- a/backend/packages/harness/deerflow/uploads/manager.py +++ b/backend/packages/harness/deerflow/uploads/manager.py @@ -333,6 +333,73 @@ def copy_upload_file_no_symlink(base_dir: Path, filename: str, src: Path) -> Pat return dest +def apply_upload_sandbox_permits(file_path: os.PathLike[str] | str, extra_mode_bits: int) -> None: + """Apply sandbox permission bits to an upload, bound to its validated inode. + + The gateway writes uploads as root with ``0o600``. In AIO/Docker sandbox mode + the sandbox runs as a non-root user on the bind-mounted path, so it needs + extra group/other (and, for the writable variant, write) bits. + + The change is applied with ``os.fchmod`` on a descriptor opened with + ``O_NOFOLLOW`` (and ``O_NONBLOCK`` where available) and validated as a + regular file via ``os.fstat``. That binds the permission change to the exact + inode that was validated instead of re-resolving the pathname, so a sandbox + process that swaps the upload for a symlink after validation cannot redirect + the change to a target outside the uploads directory. ``O_NONBLOCK`` stops a + swapped-in FIFO from blocking the open before the type check. On platforms + without ``O_NOFOLLOW``/``os.fchmod`` (Windows) the ``os.chmod`` path (with + the lstat symlink guard) is retained. A path that disappears or becomes a + symlink during validation is skipped; other permission errors propagate so + callers do not report an upload the sandbox still cannot access. + """ + try: + file_stat = os.lstat(file_path) + except (FileNotFoundError, NotADirectoryError): + return + if stat.S_ISLNK(file_stat.st_mode): + return + + if hasattr(os, "O_NOFOLLOW") and hasattr(os, "fchmod"): + open_flags = os.O_RDONLY | os.O_NOFOLLOW + if hasattr(os, "O_NONBLOCK"): + # The uploads directory is sandbox-writable, so the sandbox can swap + # the just-written file for a FIFO before this open. Without + # O_NONBLOCK an O_RDONLY open on a FIFO blocks in the kernel waiting + # for a writer (before the fstat regular-file check below), hanging + # ingestion and occupying a Gateway file-IO executor thread that + # coroutine cancellation cannot interrupt. O_NONBLOCK returns + # immediately; the S_ISREG check then skips the non-regular inode. + open_flags |= os.O_NONBLOCK + try: + fd = os.open(file_path, open_flags) + except OSError as exc: + # The path disappeared, stopped resolving, or became a symlink + # after lstat. Leave permissions untouched for these expected + # replacement races, but surface operational failures such as + # EACCES so callers cannot report an unreadable upload as ready. + if exc.errno in {errno.ENOENT, errno.ENOTDIR, errno.ELOOP}: + return + raise + try: + opened = os.fstat(fd) + if not stat.S_ISREG(opened.st_mode): + return + os.fchmod(fd, stat.S_IMODE(opened.st_mode) | extra_mode_bits) + finally: + os.close(fd) + return + + # Windows / platforms without O_NOFOLLOW + fchmod: retain the lstat-guarded + # chmod fallback. Expected replacement races are no-ops; permission errors + # must still reach the caller. + try: + os.chmod(file_path, stat.S_IMODE(file_stat.st_mode) | extra_mode_bits) + except OSError as exc: + if exc.errno in {errno.ENOENT, errno.ENOTDIR, errno.ELOOP}: + return + raise + + def list_files_in_dir(directory: Path) -> dict: """List files (not directories) in *directory*. diff --git a/backend/tests/test_channel_file_attachments.py b/backend/tests/test_channel_file_attachments.py index 1bd2f7a76..94cff8e9a 100644 --- a/backend/tests/test_channel_file_attachments.py +++ b/backend/tests/test_channel_file_attachments.py @@ -4,9 +4,11 @@ from __future__ import annotations import asyncio import os +import stat from pathlib import Path from unittest.mock import MagicMock, patch +import pytest from support.symlinks import symlink_or_skip from app.channels.base import Channel @@ -404,6 +406,170 @@ class TestInboundFileIngestion: assert (uploads_dir / "victim_1.txt").read_bytes() == b"new attachment data" +# --------------------------------------------------------------------------- +# Inbound file sandbox-readability tests +# --------------------------------------------------------------------------- + + +class TestInboundFileSandboxPerms: + def test_make_inbound_file_sandbox_readable_sets_group_other_read(self, tmp_path): + from app.channels.manager import _make_inbound_file_sandbox_readable + + f = tmp_path / "photo.jpg" + f.write_bytes(b"\x89PNG data") + os.chmod(f, 0o600) + + _make_inbound_file_sandbox_readable(f) + + mode = stat.S_IMODE(os.stat(f).st_mode) + assert mode == 0o644 + assert mode & stat.S_IRGRP + assert mode & stat.S_IROTH + + def test_make_inbound_file_sandbox_readable_preserves_owner_bits(self, tmp_path): + from app.channels.manager import _make_inbound_file_sandbox_readable + + f = tmp_path / "report.pdf" + f.write_bytes(b"pdf") + os.chmod(f, 0o640) + + _make_inbound_file_sandbox_readable(f) + + mode = stat.S_IMODE(os.stat(f).st_mode) + # Owner rw and the existing group read are preserved; only the missing + # other-read bit is added and no write is granted to group/other. + assert mode == 0o644 + assert not (mode & stat.S_IWOTH) + + def test_make_inbound_file_sandbox_readable_skips_symlink(self, tmp_path): + from support.symlinks import symlink_or_skip + + from app.channels.manager import _make_inbound_file_sandbox_readable + + target = tmp_path / "target.txt" + target.write_text("secret") + os.chmod(target, 0o600) + link = tmp_path / "link.txt" + symlink_or_skip(link, target) + + # Must not raise and must leave the symlink target's mode untouched. + _make_inbound_file_sandbox_readable(link) + + assert stat.S_IMODE(os.stat(target).st_mode) == 0o600 + + def test_make_inbound_file_sandbox_readable_missing_noop(self, tmp_path): + from app.channels.manager import _make_inbound_file_sandbox_readable + + # Best-effort helper: a missing file is a silent no-op. + _make_inbound_file_sandbox_readable(tmp_path / "does-not-exist.txt") + + def test_make_inbound_file_sandbox_readable_swap_after_lstat_does_not_follow_symlink(self, tmp_path, monkeypatch): + from app.channels.manager import _make_inbound_file_sandbox_readable + + # An unrelated target OUTSIDE the uploads dir with restrictive perms. + target = tmp_path / "outside" / "secret.txt" + target.parent.mkdir() + target.write_text("secret") + os.chmod(target, 0o600) + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + upload = uploads_dir / "report.txt" + upload.write_bytes(b"upload") # a regular file at lstat time + + # Select the symlink-following chmod fallback: a platform whose os.chmod + # is not in os.supports_follow_symlinks degrades to a following chmod. + monkeypatch.setattr(os, "supports_follow_symlinks", frozenset(), raising=False) + + real_lstat = os.lstat + + def racing_lstat(path, *args, **kwargs): + st = real_lstat(path) + # The sandbox swaps the just-validated regular file for a symlink to + # the out-of-dir target immediately after lstat (the TOCTOU race). + if path == upload: + os.unlink(path) + os.symlink(target, path) + return st + + monkeypatch.setattr(os, "lstat", racing_lstat) + + _make_inbound_file_sandbox_readable(upload) + + # The permission change must stay bound to the upload's validated inode + # and must not follow the swapped-in symlink to a target outside the + # uploads dir (the old following-chmod fallback would chmod it 0600->0644). + assert stat.S_IMODE(os.lstat(target).st_mode) == 0o600 + + @pytest.mark.skipif(not (hasattr(os, "mkfifo") and hasattr(os, "O_NOFOLLOW")), reason="POSIX-only: mkfifo + O_NOFOLLOW") + def test_make_inbound_file_sandbox_readable_swap_after_lstat_does_not_block_on_fifo(self, tmp_path, monkeypatch): + import threading + + from app.channels.manager import _make_inbound_file_sandbox_readable + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + upload = uploads_dir / "report.txt" + upload.write_bytes(b"upload") # a regular file at lstat time + + real_lstat = os.lstat + + def racing_lstat(path, *args, **kwargs): + st = real_lstat(path) + # The sandbox swaps the just-validated regular file for a FIFO + # immediately after lstat. Without O_NONBLOCK the read-only open + # below blocks in the kernel waiting for a writer, before the + # S_ISREG type check can skip the non-regular inode. + if path == upload: + os.unlink(path) + os.mkfifo(path) + return st + + monkeypatch.setattr(os, "lstat", racing_lstat) + + completed = threading.Event() + + def call(): + _make_inbound_file_sandbox_readable(upload) + completed.set() + + worker = threading.Thread(target=call, daemon=True) + worker.start() + finished = completed.wait(5) + if not finished: + # Unblock the blocked open so the leaked thread can exit, then fail. + unblocker = os.open(upload, os.O_WRONLY | os.O_NONBLOCK) + os.close(unblocker) + worker.join(timeout=5) + pytest.fail("apply_upload_sandbox_permits blocked on a swapped-in FIFO (missing O_NONBLOCK)") + + # Returned promptly and left the FIFO untouched (no chmod on a FIFO). + assert stat.S_ISFIFO(os.lstat(upload).st_mode) + + def test_ingest_inbound_files_makes_file_sandbox_readable(self, tmp_path): + from app.channels import manager + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + msg = InboundMessage( + channel_name="telegram", + chat_id="chat-1", + user_id="user-1", + text="see attachment", + files=[{"type": "image", "filename": "photo.jpg", "_content": b"\x89PNG data"}], + ) + + with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir): + _run(manager._ingest_inbound_files("thread-1", msg)) + + dest = uploads_dir / "photo.jpg" + mode = stat.S_IMODE(os.stat(dest).st_mode) + # The 0o600 root-written upload is made group/other readable so the + # non-root sandbox process can read it. + assert mode & stat.S_IRGRP + assert mode & stat.S_IROTH + + # --------------------------------------------------------------------------- # Channel base class _on_outbound with attachments # --------------------------------------------------------------------------- diff --git a/backend/tests/test_channels.py b/backend/tests/test_channels.py index 193619721..5f64c63b3 100644 --- a/backend/tests/test_channels.py +++ b/backend/tests/test_channels.py @@ -9406,7 +9406,7 @@ class TestTelegramInboundMessages: downloaded = bytearray(b"data") telegram_file = SimpleNamespace(file_size=4, download_as_bytearray=AsyncMock(return_value=downloaded)) bot = SimpleNamespace(get_file=AsyncMock(return_value=telegram_file)) - ch._application = SimpleNamespace(bot=bot) + ch._download_bot = bot msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9475,7 +9475,7 @@ class TestTelegramInboundMessages: return LoopBoundFile() ch._tg_loop = telegram_loop - ch._application = SimpleNamespace(bot=LoopBoundBot()) + ch._download_bot = LoopBoundBot() msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9534,7 +9534,7 @@ class TestTelegramInboundMessages: ch._tg_loop = telegram_loop ch._thread = loop_thread ch._running = True - ch._application = SimpleNamespace(bot=LoopBoundBot()) + ch._download_bot = LoopBoundBot() msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9619,7 +9619,7 @@ class TestTelegramInboundMessages: stopped_loop = asyncio.new_event_loop() bot = SimpleNamespace(get_file=AsyncMock()) ch._tg_loop = stopped_loop - ch._application = SimpleNamespace(bot=bot) + ch._download_bot = bot msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9647,7 +9647,7 @@ class TestTelegramInboundMessages: bus = MessageBus() ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) bot = SimpleNamespace(get_file=AsyncMock()) - ch._application = SimpleNamespace(bot=bot) + ch._download_bot = bot msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9681,7 +9681,7 @@ class TestTelegramInboundMessages: bus = MessageBus() ch = telegram.TelegramChannel(bus=bus, config={"bot_token": "test-token"}) telegram_file = SimpleNamespace(file_size=2, download_as_bytearray=AsyncMock(return_value=bytearray(b"four"))) - ch._application = SimpleNamespace(bot=SimpleNamespace(get_file=AsyncMock(return_value=telegram_file))) + ch._download_bot = SimpleNamespace(get_file=AsyncMock(return_value=telegram_file)) msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9707,7 +9707,7 @@ class TestTelegramInboundMessages: ch = telegram.TelegramChannel(bus=bus, config={"bot_token": "test-token"}) download = AsyncMock(return_value=bytearray(b"four")) telegram_file = SimpleNamespace(file_size=4, download_as_bytearray=download) - ch._application = SimpleNamespace(bot=SimpleNamespace(get_file=AsyncMock(return_value=telegram_file))) + ch._download_bot = SimpleNamespace(get_file=AsyncMock(return_value=telegram_file)) msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9732,7 +9732,7 @@ class TestTelegramInboundMessages: bus = MessageBus() ch = telegram.TelegramChannel(bus=bus, config={"bot_token": "test-token"}) telegram_file = SimpleNamespace(file_size=4, download_as_bytearray=AsyncMock(return_value=bytearray(b"four"))) - ch._application = SimpleNamespace(bot=SimpleNamespace(get_file=AsyncMock(return_value=telegram_file))) + ch._download_bot = SimpleNamespace(get_file=AsyncMock(return_value=telegram_file)) msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9754,7 +9754,7 @@ class TestTelegramInboundMessages: async def go(): bus = MessageBus() ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) - ch._application = SimpleNamespace(bot=SimpleNamespace(get_file=AsyncMock(side_effect=RuntimeError("GET https://api.telegram.org/bottest-token/getFile failed")))) + ch._download_bot = SimpleNamespace(get_file=AsyncMock(side_effect=RuntimeError("GET https://api.telegram.org/bottest-token/getFile failed"))) msg = InboundMessage( channel_name="telegram", chat_id="100", @@ -9948,6 +9948,354 @@ class TestTelegramInboundMessages: _run(go()) + def test_get_download_bot_returns_none_without_telegram_loop(self): + from app.channels.telegram import TelegramChannel + + async def go(): + bus = MessageBus() + ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + assert ch._tg_loop is None + + result = await ch._get_download_bot() + + assert result is None + assert ch._download_bot is None + + _run(go()) + + def test_get_download_bot_creates_loop_bound_bot_and_caches_it(self): + from app.channels import telegram as telegram_module + + async def go(): + bus = MessageBus() + ch = telegram_module.TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + telegram_loop = asyncio.new_event_loop() + loop_started: Future[None] = Future() + + def run_telegram_loop(): + asyncio.set_event_loop(telegram_loop) + telegram_loop.call_soon(loop_started.set_result, None) + telegram_loop.run_forever() + + loop_thread = threading.Thread(target=run_telegram_loop, daemon=True) + loop_thread.start() + try: + loop_started.result(timeout=2) + ch._tg_loop = telegram_loop + + created_tokens: list[str] = [] + init_loops: list[asyncio.AbstractEventLoop] = [] + + class FakeBot: + def __init__(self, token: str) -> None: + created_tokens.append(token) + + async def initialize(self) -> None: + init_loops.append(asyncio.get_running_loop()) + + with patch("telegram.Bot", FakeBot): + first = await ch._get_download_bot() + second = await ch._get_download_bot() + + # One Bot, built with the configured token and initialized on the + # Telegram loop; the second call reuses the cached instance. + assert first is second + assert first is ch._download_bot + assert created_tokens == ["test-token"] + assert init_loops == [telegram_loop] + finally: + if telegram_loop.is_running(): + telegram_loop.call_soon_threadsafe(telegram_loop.stop) + await asyncio.to_thread(loop_thread.join, 2) + if loop_thread.is_alive(): + pytest.fail("Telegram test event loop did not stop") + telegram_loop.close() + + _run(go()) + + def test_receive_file_uses_download_bot_not_application_bot(self): + from app.channels.telegram import TelegramChannel + + async def go(): + bus = MessageBus() + ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + # The Application's Bot is main-loop-bound; the download must never + # route through it. + app_bot = SimpleNamespace(get_file=AsyncMock()) + download_bot = SimpleNamespace( + get_file=AsyncMock( + return_value=SimpleNamespace( + file_size=2, + download_as_bytearray=AsyncMock(return_value=bytearray(b"data")), + ) + ) + ) + ch._application = SimpleNamespace(bot=app_bot) + ch._download_bot = download_bot + msg = InboundMessage( + channel_name="telegram", + chat_id="100", + user_id="42", + text="caption", + files=[{"type": "file", "file_id": "document-id", "filename": "report.pdf", "size": 2}], + ) + + result = await ch.receive_file(msg, "thread-1") + + download_bot.get_file.assert_awaited_once_with("document-id") + app_bot.get_file.assert_not_awaited() + assert result.files[0]["_content"] == b"data" + + _run(go()) + + def test_stop_shuts_down_download_bot_on_telegram_loop(self): + """Regression: stop() must close the download Bot's HTTPX clients. + + python-telegram-bot 22.7 has no ``Bot.session``; the old code called + ``bot.session.close()`` (always an AttributeError, suppressed), leaving + both request clients open. shutdown() now runs on the Telegram loop. + Uses the real Bot so the client-close assertion is meaningful. + """ + import httpx + from telegram import Bot + from telegram.request import HTTPXRequest + + from app.channels.telegram import TelegramChannel + + def ok_get_me(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": True, "result": {"id": 1, "is_bot": True, "first_name": "b", "username": "b"}}) + + async def go(): + bus = MessageBus() + ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + telegram_loop = asyncio.new_event_loop() + loop_started: Future[None] = Future() + + def run_telegram_loop(): + asyncio.set_event_loop(telegram_loop) + telegram_loop.call_soon(loop_started.set_result, None) + telegram_loop.run_forever() + + loop_thread = threading.Thread(target=run_telegram_loop, daemon=True) + loop_thread.start() + try: + loop_started.result(timeout=2) + ch._tg_loop = telegram_loop + ch._thread = loop_thread + ch._running = True + + # A real, initialized Bot (offline via MockTransport) whose + # HTTPX clients are bound to the Telegram loop; both open. + real_bot = Bot(token="test-token", request=HTTPXRequest(httpx_kwargs={"transport": httpx.MockTransport(ok_get_me)})) + await ch._run_on_telegram_loop(real_bot.initialize()) + ch._download_bot = real_bot + assert real_bot._request[0]._client.is_closed is False + assert real_bot._request[1]._client.is_closed is False + + await ch.stop() + + # Shut down on the Telegram loop (before it stops): both HTTPX + # clients are closed and the reference is cleared. + assert ch._download_bot is None + assert real_bot._request[0]._client.is_closed is True + assert real_bot._request[1]._client.is_closed is True + finally: + if telegram_loop.is_running(): + telegram_loop.call_soon_threadsafe(telegram_loop.stop) + await asyncio.to_thread(loop_thread.join, 2) + if loop_thread.is_alive(): + pytest.fail("Telegram test event loop did not stop") + telegram_loop.close() + + _run(go()) + + def test_get_download_bot_cleans_up_on_init_failure(self): + """A bot whose getMe fails is never cached and its clients are closed.""" + import httpx + from telegram import Bot + from telegram.error import TimedOut + from telegram.request import HTTPXRequest + + from app.channels import telegram as telegram_module + + async def go(): + bus = MessageBus() + ch = telegram_module.TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + telegram_loop = asyncio.new_event_loop() + loop_started: Future[None] = Future() + + def run_telegram_loop(): + asyncio.set_event_loop(telegram_loop) + telegram_loop.call_soon(loop_started.set_result, None) + telegram_loop.run_forever() + + loop_thread = threading.Thread(target=run_telegram_loop, daemon=True) + loop_thread.start() + try: + loop_started.result(timeout=2) + ch._tg_loop = telegram_loop + + def timeout_handler(request: httpx.Request) -> httpx.Response: + raise httpx.ReadTimeout("read timed out") + + real_bot = Bot(token="test-token", request=HTTPXRequest(httpx_kwargs={"transport": httpx.MockTransport(timeout_handler)})) + + with patch("telegram.Bot", lambda token: real_bot): + with pytest.raises(TimedOut): + await ch._get_download_bot() + + # A bot that failed to initialize is never cached, and its + # partially-open HTTPX clients are closed. + assert ch._download_bot is None + assert real_bot._request[0]._client.is_closed is True + assert real_bot._request[1]._client.is_closed is True + finally: + if telegram_loop.is_running(): + telegram_loop.call_soon_threadsafe(telegram_loop.stop) + await asyncio.to_thread(loop_thread.join, 2) + if loop_thread.is_alive(): + pytest.fail("Telegram test event loop did not stop") + telegram_loop.close() + + _run(go()) + + def test_receive_file_init_timeout_does_not_abort_message(self): + """A first-time getMe timeout is contained to one attachment. + + The old code awaited _get_download_bot() outside the per-attachment + handler, so a TimedOut escaped receive_file() and the manager aborted + the whole message. Now the caption is preserved and only the + attachment is reported unavailable. + """ + import httpx + from telegram import Bot + from telegram.request import HTTPXRequest + + from app.channels import telegram as telegram_module + + async def go(): + bus = MessageBus() + ch = telegram_module.TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + telegram_loop = asyncio.new_event_loop() + loop_started: Future[None] = Future() + + def run_telegram_loop(): + asyncio.set_event_loop(telegram_loop) + telegram_loop.call_soon(loop_started.set_result, None) + telegram_loop.run_forever() + + loop_thread = threading.Thread(target=run_telegram_loop, daemon=True) + loop_thread.start() + try: + loop_started.result(timeout=2) + ch._tg_loop = telegram_loop + + def timeout_handler(request: httpx.Request) -> httpx.Response: + raise httpx.ReadTimeout("read timed out") + + real_bot = Bot(token="test-token", request=HTTPXRequest(httpx_kwargs={"transport": httpx.MockTransport(timeout_handler)})) + msg = InboundMessage( + channel_name="telegram", + chat_id="100", + user_id="42", + text="caption", + files=[{"type": "file", "file_id": "document-id", "filename": "report.pdf", "size": 10}], + ) + + with patch("telegram.Bot", lambda token: real_bot): + result = await ch.receive_file(msg, "thread-1") + + # No exception escapes; the caption survives and the attachment + # is reported unavailable. + assert result.files == [] + assert result.text.startswith("caption") + assert "report.pdf" in result.text + assert "download failed" in result.text + # And the partially-initialized bot was cleaned up, not cached. + assert ch._download_bot is None + assert real_bot._request[1]._client.is_closed is True + finally: + if telegram_loop.is_running(): + telegram_loop.call_soon_threadsafe(telegram_loop.stop) + await asyncio.to_thread(loop_thread.join, 2) + if loop_thread.is_alive(): + pytest.fail("Telegram test event loop did not stop") + telegram_loop.close() + + _run(go()) + + def test_receive_file_download_failure_logs_cause_chain_without_token(self, caplog): + from app.channels.telegram import TelegramChannel + + async def go(): + bus = MessageBus() + ch = TelegramChannel(bus=bus, config={"bot_token": "test-token"}) + + def boom(file_id: str) -> None: + raise RuntimeError("download aborted") from ConnectionResetError("GET https://api.telegram.org/file/bottest-token/ABC123/photo.jpg") + + ch._download_bot = SimpleNamespace(get_file=AsyncMock(side_effect=boom)) + msg = InboundMessage( + channel_name="telegram", + chat_id="100", + user_id="42", + text="caption", + files=[{"type": "file", "file_id": "document-id", "filename": "report.pdf", "size": 10}], + ) + + with caplog.at_level(logging.ERROR): + result = await ch.receive_file(msg, "thread-1") + + assert result.files == [] + assert "report.pdf" in result.text + # The cause chain is surfaced for diagnosis... + assert "caused_by=" in caplog.text + assert "ConnectionResetError" in caplog.text + # ...with the token-bearing Bot API file URL masked (host kept for + # diagnosis; the exact redaction marker varies by which pass handled + # it — our own vs the global UrlRedactionFilter). + assert "api.telegram.org" in caplog.text + assert "ABC123" not in caplog.text + assert "test-token" not in caplog.text + + _run(go()) + + def test_describe_download_cause_redacts_bot_api_urls(self): + """Both Bot API URL forms carry the token; neither may publish it. + + PTB builds the file download URL as ``/file/bot/…`` and the + method URLs (getMe, getFile) as ``/bot/`` — the token is + in the path in every form. The helper redacts the configured token and + collapses any remaining Bot API URL so a chained HTTP exception cannot + leak it, while keeping the method name for diagnosis. + """ + from app.channels.telegram import TelegramChannel + + ch = TelegramChannel(bus=MessageBus(), config={"bot_token": "test-token"}) + + def with_cause(cause: BaseException) -> BaseException: + exc = RuntimeError("x") + exc.__cause__ = cause + return exc + + file_exc = with_cause(ConnectionResetError("GET https://api.telegram.org/file/bottest-token/ABC123/photo.jpg")) + getme_exc = with_cause(ConnectionResetError("GET https://api.telegram.org/bottest-token/getMe")) + getfile_exc = with_cause(ConnectionResetError("GET https://api.telegram.org/bottest-token/getFile")) + + for exc in (file_exc, getme_exc, getfile_exc): + cause = ch._describe_download_cause(exc) + assert "test-token" not in cause + assert "bottest-token" not in cause + + # The file URL is fully collapsed; the method URLs keep their method + # name for diagnosis with the token redacted. + assert "api.telegram.org/file/[redacted]" in ch._describe_download_cause(file_exc) + assert "bot[redacted]/getMe" in ch._describe_download_cause(getme_exc) + assert "bot[redacted]/getFile" in ch._describe_download_cause(getfile_exc) + + # A cause with no URL passes through untouched. + assert ch._describe_download_cause(with_cause(TimeoutError("read timed out"))) == " caused_by=TimeoutError:read timed out" + class TestTelegramProcessingOrder: """Ensure 'working on it...' is sent before inbound is published.""" diff --git a/backend/tests/test_uploads_manager.py b/backend/tests/test_uploads_manager.py index d92500eb5..eb3c27889 100644 --- a/backend/tests/test_uploads_manager.py +++ b/backend/tests/test_uploads_manager.py @@ -11,6 +11,7 @@ import pytest from deerflow.uploads.manager import ( PathTraversalError, UnsafeUploadPathError, + apply_upload_sandbox_permits, claim_unique_filename, cleanup_stale_upload_staging_files, copy_upload_file_no_symlink, @@ -21,6 +22,31 @@ from deerflow.uploads.manager import ( write_upload_file_no_symlink, ) + +@pytest.mark.skipif(not (hasattr(os, "O_NOFOLLOW") and hasattr(os, "fchmod")), reason="POSIX-only: O_NOFOLLOW + fchmod") +def test_apply_upload_sandbox_permits_propagates_permission_errors(tmp_path): + upload = tmp_path / "attachment.bin" + upload.write_bytes(b"attachment") + upload.chmod(0o600) + + with patch.object(os, "fchmod", side_effect=PermissionError("permission denied")): + with pytest.raises(PermissionError, match="permission denied"): + apply_upload_sandbox_permits(upload, stat.S_IRGRP | stat.S_IROTH) + + assert stat.S_IMODE(upload.stat().st_mode) == 0o600 + + +def test_apply_upload_sandbox_permits_fallback_propagates_permission_errors(tmp_path, monkeypatch): + upload = tmp_path / "attachment.bin" + upload.write_bytes(b"attachment") + upload.chmod(0o600) + monkeypatch.delattr(os, "O_NOFOLLOW", raising=False) + + with patch.object(os, "chmod", side_effect=PermissionError("permission denied")): + with pytest.raises(PermissionError, match="permission denied"): + apply_upload_sandbox_permits(upload, stat.S_IRGRP | stat.S_IROTH) + + # --------------------------------------------------------------------------- # normalize_filename # ---------------------------------------------------------------------------