From 26800d124560192ee7595a9a042363e80d509fac Mon Sep 17 00:00:00 2001 From: hataa <79907651+hata33@users.noreply.github.com> Date: Wed, 16 Sep 2026 16:33:47 +0800 Subject: [PATCH] fix(channels): stream-cap and validate WeChat/WeCom inbound media downloads, fixes #5223 (#5225) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(channels): stream-cap and validate WeChat/WeCom inbound media downloads, fixes #5223 * fix(channels): address WeCom APPID, decompression, and log-sanitization review findings (#5223) Round-4 review follow-ups on the inbound-media download cap: - The COS bucket numeric suffix is the owner's Tencent Cloud APPID and bucket names are user-chosen, so any Tencent Cloud account could register a matching ww-aibot-img-* bucket and pass the shape gate. The built-in rule now admits only the APPID observed in Tencent's published aibot callback examples (1258476243), across regions; any other account (including a future WeCom rotation) goes through channels.wecom.allowed_media_hosts. - aiter_bytes() transparently decodes Content-Encoding, and the decoder allocates the full decompressed body before the byte cap sees a chunk (an ~8 KB gzip wire chunk decoding to 8 MiB reproduces it). Both URL readers now send Accept-Encoding: identity, refuse a response with a residual Content-Encoding before reading, and iterate aiter_raw(). - httpx.HTTPStatusError formats the signed URL (path + query credentials) into its message, so _ingest_inbound_files' reader-failure branch logs a sanitized summary (class + status) instead of logger.exception, and the WeChat extract paths catch httpx.HTTPError so the polling loop's per-message logger.exception can never render a media URL. Every change ships with a red/green regression: the reviewer's 403 mock-transport repro asserted against caplog.text (fully formatted logs), the reviewer's different-APPID bucket host, and gzip bombs driven through real httpx mock transports in both readers. Docs (channels AGENTS.md, README, config.example.yaml) updated for the APPID pinning and encoding gate. * docs(channels): document why the inbound-media cap is 50 MB, not WeCom's 100 MB ceiling * fix(logging): redact URLs in httpx request logs down to scheme + host, fixes #5223 httpx emits 'HTTP Request: GET ' at the Gateway's INFO level before any response handling runs, so even successful signed-media downloads leaked their credentials. HttpxUrlQueryRedactionFilter (installed by configure_logging) rewrites those records in place — path and query become /, method/status/duration observability is preserved — which also keeps Telegram's token-bearing Bot API paths out of the logs. Reader-level regression tests run at production INFO level with a real MockTransport, success paths included. * fix(logging): blank userinfo credentials in httpx request-log redaction * fix(logging): redact authority-only URLs and cover urllib3 redirect logs Two follow-ups from the review plus one extrapolation of the same class: - rest is now optional in _URL_REDACT_RE, so an authority-only URL (scheme://user:pass@host, no path) is rewritten too — userinfo had nowhere else to hide and previously passed through verbatim. A bare credential-free origin still passes through unchanged. - Renamed to UrlRedactionFilter / install_url_log_redaction and attached to the urllib3 logger as well: urllib3 logs 'Redirecting -> ' at INFO with full URLs on both sides, the same leak class on a different library logger. No gateway path today both uses requests and redirects a signed URL, but the class stays closed instead of dormant. - Unit tests now build records with the real httpx 0.28.1 format string ('HTTP Request: %s %s "%s %d %s"', 5 args) and httpx.URL args, per the nit, instead of a synthetic shape httpx never emits. * fix(logging): install URL redaction at handler level so propagated records are covered A logging.Filter on a logger only runs for records emitted through that exact logger — child loggers neither inherit it nor trigger it on propagation — so the previous attachment to the bare urllib3 logger was dead code: urllib3 emits Redirecting via urllib3.poolmanager at INFO and urllib3.connectionpool at DEBUG. The filter is now attached to every root handler (mirroring _install_trace_filter, which already iterates root handlers; handler-level filters see propagated records) in addition to the httpx logger (httpx emits via the bare name, and emission-point coverage survives handlers added later). The wiring is pinned by tests that emit through the real urllib3 child loggers — a mutation removing the handler-level install turns them red. Comments, docstrings, and AGENTS.md now state the actual emitter names and levels. * fix(logging): redact urllib3 DEBUG request lines, whose split shape evaded the URL regex urllib3's per-request line (connectionpool.py:545 on 2.7.0) renders as `scheme://host:port "METHOD /path?query HTTP/x.x" status len` — the authority ends at a space so _URL_REDACT_RE's bare-origin early return applies, and the quoted origin-form target has no scheme, so neither half was rewritten. UrlRedactionFilter now runs a dedicated request-line shape first (collapsing the target to /, keeping scheme+host+method+ version), then the absolute-URL pass. Regressions pin the exact format string both at unit level and through the real urllib3.connectionpool DEBUG emit path; AGENTS.md wording now names both covered DEBUG shapes. * fix(logging): redact urllib3 retry lines and linearize scheme scanning Closes the two open review threads on the inbound-media log hardening: Retry/redirect targets: urllib3 logs the request target with no scheme in five shapes the generic absolute-URL pass cannot see - `Retry: ` (connectionpool.py:954 DEBUG), `Incremented Retry for (url='')` (util/retry.py:545 DEBUG, absolute on the redirect path), `Retrying (...) after connection broken by '': ` (connectionpool.py:869 WARNING, above the INFO root), and origin-form halves of both Redirecting emitters (poolmanager.py:500 INFO / connectionpool.py:922 DEBUG). Each gets a rewrite anchored to the exact urllib3 format, collapsing the target to /; the generic pass's rest now stops at quote characters so a quoted URL keeps its closing punctuation (previously the absolute-form increment line was mangled), and the request-line method class accepts any case. The emitter enumeration in channels AGENTS.md is closed against the installed urllib3 2.7.0 source. Quadratic scanning: both scheme-bearing patterns start with a character class, so re.sub retried every suffix of a long token - 64K paths cost ~1.8s and URL-free 64K error bodies ~3.1s per record, synchronously in every root handler. The two passes are now driven from literal "://" occurrences: _scheme_starts walks back over the scheme charset to each run's first letter and the pattern is attempted only there, reproducing re.sub's leftmost-non-overlapping result in linear time (256K path: 5.6ms; worst adversarial shapes <= 28ms). Long-input regressions pin the URL-bearing and URL-free cases with mutation-verified bounds, plus nested-scheme and digit-headed-run equivalence cases. Validation: tests/test_logging_config.py 12/12; scheme-pass equivalence against the old re.sub pipeline verified by two independent 30k+ case fuzz runs; full-suite A/B against HEAD shows zero tests that pass on HEAD and fail with this diff. * fix(logging): boundary-aware quote stops and whole-message Redirecting anchor Two follow-ups on the urllib3 redaction shapes: Embedded quotes: `rest` treated ANY quote as a closing mark, so a URL with an apostrophe in the path kept everything after it verbatim (`https://h/path'quoted'?token=Q` rendered the credential suffix in full) while the class docstring claimed path/query/fragment are replaced. A quote now closes `rest` only at a boundary - followed by whitespace, a closing parenthesis, or end of string - so urllib3's Incremented Retry (url='...') scaffolding keeps its ') closer while an embedded quote stays consumed. The increment line's url capture gets the same rule narrowed to its fixed ')' closer. Redirecting anchoring: the origin-half pass matched `(? <=-> )/path` as a substring, and an `-> /path` arrow is not urllib3-owned shape - the sandbox provider's actionable mount error (`sandbox.mounts entry -> /mnt/knowledge ignored: ...`) had its container path rewritten to /, failing test_setup_path_mappings_logs_actionable_error_for_missing_host_path on CI (backend-unit-tests shard 3). The pass is now anchored to the whole `Redirecting -> ` message, which is exactly urllib3's record; origin slots collapse, absolute slots stay for the generic pass. Regression tests pin the embedded-quote shapes and the sandbox error's byte-for-byte passthrough; both mutations verified red. Validation: tests/test_logging_config.py 14/14; the CI-failing sandbox test green locally; every test file asserting redaction/arrow log content passes (attachments, support bundle, run metadata, skill secrets, ragflow, skillscan, sandbox provider); full offline backend suite 14084 passed / 164 failed with the failure set matching this machine's documented Windows-environment baseline (NTFS chmod/symlink, docker/lark/langfuse absences) - no failure involves redaction output. * fix(logging): redact redirects with spaced locations * fix(logging): grammar-complete Redirecting anchor; neutral WeChat guard labels Round-13 P3 (Redirecting anchor strictness): the whole-message anchor kept the ^Redirecting prefix (the urllib3-owned literal that stops the sandbox false positive) but required BOTH slots whitespace-free, so a Location header with an interior space voided the pass and leaked the origin-form request target in the first slot - redirect_location is the raw header string and interior spaces are legal field syntax. The tail is now loose (\S.*$) and the first slot gets the same grammar treatment (\S.*?): the recursive urlopen frame passes the previous raw Location as its url, so t1 can carry interior spaces too, lazy-split at the first arrow the way the line is constructed. A space-carrying slot collapses whole when it starts with /; the sandbox mount error keeps passing through untouched. Round-14 nit (None conflation): _download_cdn_bytes returns None for two reasons (in-flight cap abort, Content-Encoding refusal) but both image and file callers labeled it "exceeds size limit (N bytes)" - contradicting the accurate encoding line right above it, and reporting the plaintext limit for a ciphertext-cap decision. Callers now log a neutral "skipped by download guard" line (the manager reader callers' shape); the accurate reason stays inside the download function. The same sweep also logs _stage_downloaded_file's silent None (no state dir configured), which made an attachment vanish with no log line at all. Also anchors the emitter-enumeration closure to its urllib3 version: the closure reopens if an upgrade changes these format strings, so the comment now says so explicitly. Validation: logging 15/15 and attachments 60/62 (the two pre-existing Windows symlink-privilege failures documented in the PR body); three mutations verified red (old wording, strict t1, silent staging None); ruff clean. Full offline suite run before push (per round-11 lesson). --------- Co-authored-by: Willem Jiang --- README.md | 8 + backend/app/channels/AGENTS.md | 1 + backend/app/channels/manager.py | 202 ++++- backend/app/channels/wechat.py | 194 ++++- backend/app/channels/wecom.py | 29 + .../harness/deerflow/logging_config.py | 266 +++++++ .../blocking_io/test_wechat_channel_state.py | 4 +- .../tests/test_channel_file_attachments.py | 585 +++++++++++++++ backend/tests/test_logging_config.py | 693 ++++++++++++++++++ backend/tests/test_wechat_channel.py | 357 ++++++++- config.example.yaml | 9 + 11 files changed, 2308 insertions(+), 40 deletions(-) diff --git a/README.md b/README.md index 1ac921d6d..79be25490 100644 --- a/README.md +++ b/README.md @@ -615,6 +615,11 @@ channels: enabled: true bot_id: $WECOM_BOT_ID bot_secret: $WECOM_BOT_SECRET + # Optional: extra host suffixes inbound media downloads may come from, in + # addition to the built-in qq.com family and WeCom's official COS media + # host (ww-aibot-img-1258476243..myqcloud.com); add one here if + # WeCom rotates to a new COS account or media goes through a proxy + allowed_media_hosts: [] slack: enabled: true @@ -644,6 +649,9 @@ channels: max_outbound_image_bytes: 20971520 max_inbound_file_bytes: 52428800 max_outbound_file_bytes: 52428800 + # Inbound media downloads stream with the caps above and are restricted to + # these host suffixes (plus *.qq.com and the cdn_base_url host by default) + allowed_media_hosts: [] # Optional: per-channel / per-user session settings session: diff --git a/backend/app/channels/AGENTS.md b/backend/app/channels/AGENTS.md index e3fb5fe55..9f79bae71 100644 --- a/backend/app/channels/AGENTS.md +++ b/backend/app/channels/AGENTS.md @@ -24,6 +24,7 @@ Bridges external messaging platforms (Feishu, Slack, Telegram, Discord, DingTalk - For GitHub, the webhook router verifies the delivery then calls `fanout_event(bus, ...)`; matching agent bindings publish one `InboundMessage` each instead of a long-polling channel worker. - Telegram photo/document updates use the largest photo size or document metadata, preserve `message.caption`, enforce the hosted Bot API's 20,000,000-byte download ceiling before and after download, and never expose the token-bearing Bot API file URL. Downloaded bytes cross the adapter/manager boundary only through `message_bus.INBOUND_FILE_CONTENT_KEY`; the manager consumes that transient field before persisting safe upload metadata. - Feishu/Lark inbound image/file downloads read at most 20,000,001 bytes and reject anything above 20,000,000 bytes before persistence or sandbox sync. Oversize, unsafe-path, and path-resolution failures rewrite only that attachment placeholder to `Failed to obtain the [type]` so later attachments in the same message can still load. + - WeChat iLink inbound image/file downloads stream with an in-flight cap derived from `channels.wechat.max_inbound_image_bytes` / `max_inbound_file_bytes` (20 MB / 50 MB defaults) that aborts before the full read, and the payload-supplied `media.full_url` must pass a scheme + dot-boundary host-suffix allowlist (`channels.wechat.allowed_media_hosts`, defaulting to the `qq.com` family plus the configured `cdn_base_url` host) before anything is fetched. The limits are PLAINTEXT limits while the stream measures CIPHERTEXT, so the in-flight cap is the PKCS#7-padded size of exactly-limit plaintext (`WechatChannel._stream_cap_for`); the exact post-decryption check stays as the authority. There is deliberately **no URL re-fetch fallback**: `_read_wechat_inbound_file` (manager) only reads the locally staged `path` — the channel always stages one before publishing — so nothing fetches a WeChat URL outside the channel-side gate. WeCom media URLs (frame-supplied `url`, no per-channel size config) get the same style of host gate — `qq.com` suffixes plus the COS shape `ww-aibot-img-.cos..myqcloud.com` WeCom serves signed links from, where the numeric suffix must be one of the verified WeCom-owned Tencent Cloud APPIDs (`_WECOM_MEDIA_COS_APPIDS`, currently only the `1258476243` observed in Tencent's published callback examples) because bucket names are user-chosen and any Tencent Cloud account can register a lookalike `ww-aibot-img-*` bucket; other accounts go through operator suffixes from `channels.wecom.allowed_media_hosts` resolved from the live channel at read time — and the manager-level `MAX_INBOUND_URL_FILE_BYTES` (50 MB) streaming cap. Both URL downloads and `_download_cdn_bytes` request `Accept-Encoding: identity`, refuse a response with any residual `Content-Encoding` before reading, and iterate `aiter_raw()`: httpx's transparent decoding would allocate the full decompressed body before the byte cap sees a chunk. No URL-based inbound-media log line may contain any part of the URL (signed links carry credentials in path and query): failure labels use the attachment filename or host only (`manager._inbound_file_label`), and reader failures log a sanitized exception summary — class name plus HTTP status (`manager._reader_error_summary`, `wechat._media_download_error_summary`), never the URL-bearing message or traceback. The rule extends to the HTTP client libraries' own records: httpx emits `HTTP Request: GET ` at INFO via the bare `httpx` logger, and urllib3 — enumerated against the installed source; every remaining emitter logs host:port only (connection establishment/reset) or one absolute URL (the header-parse warning) that the generic pattern rewrites — renders a request target in five shapes, each needing a dedicated redaction pattern because the generic absolute-URL pass cannot see a target with no scheme: `Redirecting -> ` at INFO via `urllib3.poolmanager` and DEBUG via `urllib3.connectionpool` (either slot may be an origin-form relative Location, which pool/connection callers pass or the Location header carries; the shape keeps the `^Redirecting ` prefix anchor because `-> /path` arrows also appear in non-URL logs — sandbox mount mappings — and must pass through untouched, while both slots consume through end-of-line: the raw Location header may carry interior spaces, and the recursive urlopen frame can pass it on as the next request target), the per-request `scheme://host:port "METHOD /path?query HTTP/x.x"` DEBUG line, `Retry: ` DEBUG and `Retrying (…) after connection broken by '': ` **WARNING** — above the INFO root, so reachable at the default log level — via `urllib3.connectionpool`, and `Incremented Retry for (url='')` DEBUG via `urllib3.util.retry` (origin-form on the request path, absolute on the redirect path). The generic pass's `rest` treats a quote as a closing mark only at a boundary — followed by whitespace, a closing parenthesis, or end of string — so urllib3's `Incremented Retry for (url='…')` keeps its `')` scaffolding while a quote embedded in a URL stays consumed and redacted. Logger filters only see records emitted through that exact logger — child loggers neither inherit them nor trigger them on propagation — so `UrlRedactionFilter` is attached to the `httpx` logger AND to every root handler by `configure_logging` (handler-level filters see propagated records), which also keeps Telegram's token-bearing Bot API paths (`api.telegram.org/bot/...`, emitted by python-telegram-bot's httpx client) out of the logs. 2. `ChannelManager._dispatch_loop()` consumes from queue 3. For user-owned channel connections, incoming messages carry `connection_id`, `owner_user_id`, and `workspace_id`; `owner_user_id` becomes the DeerFlow run `user_id`, while the raw platform user id remains `channel_user_id`. The Gateway accepts `channel_user_id` only from an internally authenticated channel caller's top-level `body.context`, clears it from both free-form `body.config` sections, and writes it into runtime context only (never `configurable`, which is checkpointed). `bash_tool` exposes it to sandbox commands as the fixed env var `DEERFLOW_CHANNEL_USER_ID` — via a shell-quoted command-string prefix, NOT the `execute_command(env=...)` channel, which is reserved for request-scoped secrets and would switch `AioSandbox` onto the `bash.exec` path (image >= 1.9.3, fresh session per call). Per-call injection keeps group-chat identity correct (one thread/sandbox, many senders) **without depending on the AIO shell's session semantics**: every IM-channel command carries an explicit `export VAR=; ` (valid id) or `unset VAR; ` (empty / non-str / over the 256-char cap). The AIO no-env path reuses a persistent shell session (the reason for the class lock, #1433), so a bare command could otherwise resolve a stale id an earlier sender exported; the `unset` closes the window the length/type guard would open (a dropped id would inherit the previous sender's value). Non-IM runs (no `channel_user_id` in context) are left untouched. Not injected on the Windows local sandbox (its PowerShell/cmd.exe fallback has no `export`/`unset`). Propagates across `task` delegation: `task_tool` captures the dispatching turn's id and the subagent executor forwards it into the subagent's runtime context, same as the guardrail attribution fields. The runtime-context value is authorization-grade at the Gateway/guardrail boundary, but the exported shell variable remains informational because any bash command can overwrite its own environment; skills must not treat the shell variable itself as authenticated identity. Tests: `tests/test_gateway_services.py`, `tests/test_channel_user_id_env.py` 4. For chat: look up/create thread through Gateway's LangGraph-compatible API diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 193dcd371..c8096fb4f 100644 --- a/backend/app/channels/manager.py +++ b/backend/app/channels/manager.py @@ -14,7 +14,7 @@ from dataclasses import dataclass from html import escape from pathlib import Path from typing import Any -from urllib.parse import quote +from urllib.parse import quote, urlparse import httpx from fastapi import HTTPException @@ -149,6 +149,37 @@ CHANNEL_CAPABILITIES = { InboundFileReader = Callable[[dict[str, Any], httpx.AsyncClient], Awaitable[bytes | None]] +# Cap for URL-based inbound attachments fetched by the generic reader (WeCom +# media today; the WeChat reader is path-only by design — see +# _read_wechat_inbound_file). The bytes are buffered in memory before being +# persisted, so an oversized attachment must be refused before it is fully +# read, not after — mirrors DingTalkChannel._download_by_code. 50 MB is a +# deliberate default, not the platform ceiling: published WeCom callback +# examples document files up to 100 MB, but the whole file is buffered (and +# decrypt_file allocates a second copy), so the bound matches the sibling +# channels' inbound caps (DingTalk's identically-sized 50 MB, WeChat's +# max_inbound_file_bytes) and halves worst-case per-message buffering; a +# legit-but-oversized file drops with a host-labeled warning naming the limit. +MAX_INBOUND_URL_FILE_BYTES = 50 * 1024 * 1024 + +# WeCom inbound media URLs come from the platform's WS frames (wecom.py passes +# ``payload.get("url")`` straight through), the same untrusted-input shape as +# WeChat's ``full_url``; the fetch is therefore gated to platform-owned hosts +# before streaming, mirroring WechatChannel._is_allowed_media_url. Two +# families: qq.com hosts, and the temporary signed COS links WeCom actually +# serves media from — ``ww-aibot-img-.cos..myqcloud.com`` +# (published callback examples; valid ~5 minutes). The COS numeric suffix is +# the owner's Tencent Cloud APPID and the bucket name is user-chosen, so any +# Tencent Cloud account could register a matching ``ww-aibot-img-*`` bucket: +# the shape alone proves nothing about ownership. Only the APPID observed in +# Tencent's published aibot callback examples (1258476243) is trusted by +# default; media from any other account — including a future WeCom rotation +# to a new APPID — goes through the operator suffix list +# ``channels.wecom.allowed_media_hosts``. +WECOM_ALLOWED_MEDIA_HOST_SUFFIXES = ("qq.com",) +_WECOM_MEDIA_COS_APPIDS = frozenset({"1258476243"}) +_WECOM_MEDIA_COS_HOST_RE = re.compile(r"^ww-aibot-img-(?P\d+)\.cos\.[a-z0-9-]+\.myqcloud\.com$") + _METADATA_DROP_KEYS = frozenset({"raw_message", "ref_msg"}) @@ -169,12 +200,149 @@ async def _read_http_inbound_file(file_info: dict[str, Any], client: httpx.Async if not isinstance(url, str) or not url: return None - resp = await client.get(url) - resp.raise_for_status() - return resp.content + chunks: list[bytes] = [] + total = 0 + # The transfer must stay undecoded: aiter_bytes() transparently decodes + # Content-Encoding, and the decoder allocates the whole decompressed body + # before yielding a single chunk — a compressed response from an admitted + # host would blow past the cap exactly like the unbounded read this + # reader exists to prevent. Identity is requested up front, any residual + # encoding is refused before reading, and aiter_raw() never decodes. + async with client.stream("GET", url, headers={"Accept-Encoding": "identity"}) as response: + response.raise_for_status() + encoding = (response.headers.get("content-encoding") or "").strip().lower() + if encoding and encoding != "identity": + logger.warning( + "[Manager] inbound file response uses Content-Encoding %r, dropping before decode: %s", + encoding, + _inbound_file_label(file_info, url), + ) + return None + async for chunk in response.aiter_raw(): + total += len(chunk) + if total > MAX_INBOUND_URL_FILE_BYTES: + logger.warning( + "[Manager] inbound file exceeds %d bytes download limit, dropping: %s", + MAX_INBOUND_URL_FILE_BYTES, + _inbound_file_label(file_info, url), + ) + return None + chunks.append(chunk) + return b"".join(chunks) + + +def _url_host(url: str) -> str: + """Best-effort host extraction for logging; never raises, never logs the URL. + + Media URLs can carry access tokens in their query strings, so only the + host is surfaced in drop warnings. + """ + try: + return (urlparse(url).hostname or "").lower() + except ValueError: + return "" + + +def _inbound_file_label(file_info: dict[str, Any], url: str | None = None, idx: int | None = None) -> str: + """Sanitized logging label for one inbound attachment: filename, else URL host. + + Signed media URLs carry credentials in both the path and the query string + (WeCom COS links: ``/path?sign=...&q-signature=...``), so no part of the + URL itself may reach the logs — only the hostname. Filenames are webhook + supplied, so whitespace is collapsed and length capped to keep a crafted + name from forging log lines (mirrors dingtalk._display_filename). + """ + filename = file_info.get("filename") + if isinstance(filename, str) and filename.strip(): + return re.sub(r"\s+", " ", filename).strip()[:80] + source = url if isinstance(url, str) else None + if source is None: + for key in ("url", "full_url"): + value = file_info.get(key) + if isinstance(value, str) and value.strip(): + source = value + break + if source: + host = _url_host(source) + if host: + return f"host={host}" + return f"#{idx}" if idx is not None else "" + + +def _reader_error_summary(exc: BaseException) -> str: + """Sanitized exception summary for inbound-media reader failures. + + httpx exceptions format the full request URL into their message — + ``HTTPStatusError`` includes the path and query, i.e. the signed download + credentials — and rendering the traceback (``logger.exception``) would + reproduce them verbatim, so only the class name and explicitly safe + fields ever reach the logs. + """ + summary = type(exc).__name__ + if isinstance(exc, httpx.HTTPStatusError): + summary = f"{summary} ({exc.response.status_code})" + return summary + + +def _is_allowed_wecom_media_url(url: str, extra_suffixes: frozenset[str] | tuple[str, ...] | list[str] = ()) -> bool: + """Platform-owned-host gate for WeCom inbound media fetches. + + Matching semantics mirror ``WechatChannel._is_allowed_media_url``: http/https + only, and ``notqq.com`` / ``qq.com.evil.io`` never match a ``qq.com`` suffix. + The COS shape is matched exactly (see ``_WECOM_MEDIA_COS_HOST_RE``) AND its + numeric suffix must be one of the verified WeCom-owned APPIDs + (``_WECOM_MEDIA_COS_APPIDS``) — the suffix is a Tencent Cloud account + APPID and bucket names are user-chosen, so the shape alone would admit + any account that registers a lookalike bucket. Operator-supplied + ``channels.wecom.allowed_media_hosts`` suffixes are merged in on top of the + hard-coded families (see ``_wecom_extra_media_host_suffixes``). + """ + try: + parsed = urlparse(url) + except ValueError: + return False + if parsed.scheme not in ("http", "https"): + return False + host = (parsed.hostname or "").lower() + if not host: + return False + if any(host == suffix or host.endswith(f".{suffix}") for suffix in (*WECOM_ALLOWED_MEDIA_HOST_SUFFIXES, *extra_suffixes)): + return True + cos_match = _WECOM_MEDIA_COS_HOST_RE.fullmatch(host) + return cos_match is not None and cos_match.group("appid") in _WECOM_MEDIA_COS_APPIDS + + +def _wecom_extra_media_host_suffixes() -> frozenset[str]: + """Operator host suffixes from the live WeCom channel, if one is running. + + The registry reader is module-level while ``channels.wecom.allowed_media_hosts`` + is per-channel config, so the extras are resolved per call from the running + channel instance — the same service-reach-in ``_channel_supports_streaming`` + already uses. Empty when no WeCom channel is up (direct calls, most tests), + leaving only the strict built-in families; a strict default plus an operator + escape hatch means a platform URL-shape change never requires widening the + hard-coded pattern for every deployment. + """ + try: + from app.channels.service import get_channel_service + + service = get_channel_service() + channel = service.get_channel("wecom") if service is not None else None + except Exception: + return frozenset() + return frozenset(getattr(channel, "allowed_media_host_suffixes", ()) or ()) async def _read_wecom_inbound_file(file_info: dict[str, Any], client: httpx.AsyncClient) -> bytes | None: + url = file_info.get("url") + if isinstance(url, str) and url and not _is_allowed_wecom_media_url(url, _wecom_extra_media_host_suffixes()): + logger.warning( + "[Manager] WeCom inbound media URL host is not allowed, dropping file=%s host=%s", + _inbound_file_label(file_info, url), + _url_host(url), + ) + return None + data = await _read_http_inbound_file(file_info, client) if data is None: return None @@ -201,10 +369,14 @@ async def _read_wechat_inbound_file(file_info: dict[str, Any], client: httpx.Asy logger.exception("[Manager] failed to read WeChat inbound file from local path: %s", raw_path) return None - full_url = file_info.get("full_url") - if isinstance(full_url, str) and full_url.strip(): - return await _read_http_inbound_file({"url": full_url}, client) - + # No re-fetch fallback: the only producer (WechatChannel) always stages a + # local ``path`` and drops the attachment when staging fails, so a file + # dict without one has no legitimate fetch source. Fetching ``full_url`` + # here would be the one ungated Gateway-host fetch left in the WeChat + # path — the channel-side ``channels.wechat.allowed_media_hosts`` gate + # cannot reach this module-level reader, and re-implementing it with + # divergent rules would drop media the operator explicitly allowed. + logger.debug("[Manager] WeChat inbound file has no staged local path, skipping") return None @@ -948,11 +1120,15 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: else: try: data = await file_reader(f, client) - except Exception: - logger.exception( - "[Manager] failed to read inbound file: channel=%s, file=%s", + except Exception as exc: + # Sanitized on purpose: the URL-bearing exception message + # and traceback must not reach the logs (see + # _reader_error_summary). + logger.warning( + "[Manager] failed to read inbound file: channel=%s, file=%s, error=%s", msg.channel_name, - f.get("url") or filename or idx, + _inbound_file_label(f, idx=idx), + _reader_error_summary(exc), ) continue @@ -960,7 +1136,7 @@ async def _ingest_inbound_files(thread_id: str, msg: InboundMessage, *, user_id: logger.warning( "[Manager] inbound file reader returned no data: channel=%s, file=%s", msg.channel_name, - f.get("url") or filename or idx, + _inbound_file_label(f, idx=idx), ) continue diff --git a/backend/app/channels/wechat.py b/backend/app/channels/wechat.py index 6a5d4b2e1..1836f8791 100644 --- a/backend/app/channels/wechat.py +++ b/backend/app/channels/wechat.py @@ -17,7 +17,7 @@ from collections.abc import Mapping from enum import IntEnum from pathlib import Path from typing import Any -from urllib.parse import quote +from urllib.parse import quote, urlparse import httpx from cryptography.hazmat.primitives import padding @@ -116,6 +116,33 @@ def _encode_outbound_media_aes_key(aes_key: bytes) -> str: return base64.b64encode(aes_key.hex().encode("utf-8")).decode("utf-8") +def _media_url_host(url: str) -> str: + """Best-effort host extraction for logging; never raises, never logs the URL. + + CDN URLs can carry access tokens in their query strings, so only the host + is surfaced in skip warnings. + """ + try: + return (urlparse(url).hostname or "").lower() + except ValueError: + return "" + + +def _media_download_error_summary(exc: BaseException) -> str: + """Sanitized exception summary for inbound-media download failures. + + httpx exceptions format the full request URL into their message — + ``HTTPStatusError`` includes the path and query, i.e. the CDN download + credentials — so only the class name and explicitly safe fields are ever + surfaced; the raw exception must not reach a ``logger.exception`` site + (the polling loop's per-message handler would render its traceback). + """ + summary = type(exc).__name__ + if isinstance(exc, httpx.HTTPStatusError): + summary = f"{summary} ({exc.response.status_code})" + return summary + + def _detect_image_extension_and_mime(content: bytes) -> tuple[str, str] | None: if content.startswith(b"\x89PNG\r\n\x1a\n"): return ".png", "image/png" @@ -138,6 +165,8 @@ class WechatChannel(Channel): - ``qrcode_login_enabled``: (optional) Allow first-time QR bootstrap when ``bot_token`` is missing. - ``base_url``: (optional) iLink API base URL. - ``allowed_users``: (optional) List of allowed iLink user IDs. Empty = allow all. + - ``allowed_media_hosts``: (optional) Extra host suffixes inbound media URLs may + be downloaded from, in addition to the platform CDN defaults. Default: ``qq.com``. - ``polling_timeout``: (optional) Long-poll timeout in seconds. Default: 35. - ``state_dir``: (optional) Directory used to persist the long-poll cursor. """ @@ -154,6 +183,7 @@ class WechatChannel(Channel): DEFAULT_CONFIG_TIMEOUT = 10.0 DEFAULT_CDN_TIMEOUT = 30.0 DEFAULT_IMAGE_DOWNLOAD_DIRNAME = "downloads" + DEFAULT_ALLOWED_MEDIA_HOST_SUFFIXES = ("qq.com",) DEFAULT_MAX_IMAGE_BYTES = 20 * 1024 * 1024 DEFAULT_MAX_OUTBOUND_IMAGE_BYTES = 20 * 1024 * 1024 DEFAULT_MAX_INBOUND_FILE_BYTES = 50 * 1024 * 1024 @@ -243,6 +273,7 @@ class WechatChannel(Channel): self._max_inbound_file_bytes = self._coerce_int(config.get("max_inbound_file_bytes"), self.DEFAULT_MAX_INBOUND_FILE_BYTES) self._max_outbound_file_bytes = self._coerce_int(config.get("max_outbound_file_bytes"), self.DEFAULT_MAX_OUTBOUND_FILE_BYTES) self._allowed_file_extensions = self._coerce_str_set(config.get("allowed_file_extensions"), self.DEFAULT_ALLOWED_FILE_EXTENSIONS) + self._allowed_media_hosts = self._coerce_host_suffixes(config.get("allowed_media_hosts")) self._allowed_users: set[str] = {str(uid).strip() for uid in config.get("allowed_users", []) if str(uid).strip()} self._bot_token = str(config.get("bot_token") or "").strip() self._ilink_bot_id = str(config.get("ilink_bot_id") or "").strip() or None @@ -952,11 +983,60 @@ class WechatChannel(Channel): payload["no_need_thumb"] = True return payload - async def _download_cdn_bytes(self, url: str, *, timeout: float | None = None) -> bytes: + @staticmethod + def _stream_cap_for(plaintext_limit: int) -> int | None: + """Translate a plaintext size limit into the ciphertext stream cap. + + ``max_inbound_image_bytes`` / ``max_inbound_file_bytes`` bound the + DECRYPTED payload, but ``_download_cdn_bytes`` measures what is still + encrypted — AES-128-ECB with PKCS#7 padding, up to one 16-byte block + larger. Capping the stream at the plaintext limit would reject a + boundary-sized valid attachment purely for its padding, so the cap is + the padded size of exactly-limit plaintext. ``None``/non-positive + limits keep the stream uncapped, matching the ``> 0`` checks. + """ + if plaintext_limit <= 0: + return None + return _encrypted_size_for_aes_128_ecb(plaintext_limit) + + async def _download_cdn_bytes(self, url: str, *, timeout: float | None = None, max_bytes: int | None = None) -> bytes | None: + """Stream one media download, aborting in flight once it exceeds *max_bytes*. + + The bytes are buffered in memory before being decrypted and persisted, + so an oversized attachment must be refused before it is fully read, not + after (mirrors ``DingTalkChannel._download_by_code``). The transfer is + kept undecoded — identity requested, unexpected Content-Encoding + refused before reading, ``aiter_raw`` used — because the transparent + decoder allocates the full decompressed body before yielding, which + would blow past the cap for a compressed response. Returns ``None`` + when the download was aborted by the cap or rejected for its + encoding; other HTTP-level failures raise for the caller's + per-message error handling. + """ client = await self._ensure_client() - response = await client.get(url, timeout=timeout or self.DEFAULT_CDN_TIMEOUT) - response.raise_for_status() - return response.content + chunks: list[bytes] = [] + total = 0 + async with client.stream( + "GET", + url, + timeout=timeout or self.DEFAULT_CDN_TIMEOUT, + headers={"Accept-Encoding": "identity"}, + ) as response: + response.raise_for_status() + encoding = (response.headers.get("content-encoding") or "").strip().lower() + if encoding and encoding != "identity": + logger.warning( + "[WeChat] inbound media response uses Content-Encoding %r, aborting before decode", + encoding, + ) + return None + async for chunk in response.aiter_raw(): + total += len(chunk) + if max_bytes is not None and max_bytes > 0 and total > max_bytes: + logger.warning("[WeChat] inbound media download exceeds %d bytes, aborting before full read", max_bytes) + return None + chunks.append(chunk) + return b"".join(chunks) async def _upload_cdn_bytes( self, @@ -1061,6 +1141,9 @@ class WechatChannel(Channel): if not full_url: logger.warning("[WeChat] inbound image missing full_url, skipping message_id=%s", message_id) return None + if not self._is_allowed_media_url(full_url): + logger.warning("[WeChat] inbound image URL host is not allowed, skipping message_id=%s host=%s", message_id, _media_url_host(full_url)) + return None aes_key = self._resolve_media_aes_key(item, image_item, media) if not aes_key: @@ -1071,7 +1154,31 @@ class WechatChannel(Channel): ) return None - encrypted = await self._download_cdn_bytes(full_url) + # The configured limit bounds the PLAINTEXT, but the stream caps the + # CIPHERTEXT, which PKCS#7 padding makes up to a full block larger — a + # boundary-sized valid attachment must not be rejected for its padding. + # The exact post-decryption check below remains the authority. + try: + encrypted = await self._download_cdn_bytes(full_url, max_bytes=self._stream_cap_for(self._max_inbound_image_bytes)) + except httpx.HTTPError as exc: + # The URL-bearing exception must not escape to the polling loop's + # logger.exception; the attachment is dropped and the message + # continues, same as the other skip paths above. + logger.warning( + "[WeChat] inbound image download failed, skipping message_id=%s host=%s error=%s", + message_id, + _media_url_host(full_url), + _media_download_error_summary(exc), + ) + return None + if encrypted is None: + # Neutral on purpose: None covers both the in-flight cap abort + # and the Content-Encoding refusal, and _download_cdn_bytes has + # already logged the accurate reason for either — asserting a + # size limit here would contradict the encoding line (the + # manager's reader callers use the same neutral shape). + logger.warning("[WeChat] inbound image skipped by download guard, message_id=%s", message_id) + return None decrypted = _decrypt_aes_128_ecb(encrypted, aes_key) if self._max_inbound_image_bytes > 0 and len(decrypted) > self._max_inbound_image_bytes: logger.warning("[WeChat] inbound image exceeds size limit (%d bytes), skipping message_id=%s", len(decrypted), message_id) @@ -1109,6 +1216,9 @@ class WechatChannel(Channel): if not full_url: logger.warning("[WeChat] inbound file missing full_url, skipping message_id=%s", message_id) return None + if not self._is_allowed_media_url(full_url): + logger.warning("[WeChat] inbound file URL host is not allowed, skipping message_id=%s host=%s", message_id, _media_url_host(full_url)) + return None aes_key = self._resolve_media_aes_key(item, file_item, media) if not aes_key: @@ -1125,7 +1235,25 @@ class WechatChannel(Channel): logger.warning("[WeChat] inbound file type blocked, skipping message_id=%s filename=%s", message_id, filename) return None - encrypted = await self._download_cdn_bytes(full_url) + # Plaintext limit vs ciphertext cap: see the image path above. + try: + encrypted = await self._download_cdn_bytes(full_url, max_bytes=self._stream_cap_for(self._max_inbound_file_bytes)) + except httpx.HTTPError as exc: + # See the image path: the URL-bearing exception must not escape + # to the polling loop's logger.exception. + logger.warning( + "[WeChat] inbound file download failed, skipping message_id=%s host=%s error=%s", + message_id, + _media_url_host(full_url), + _media_download_error_summary(exc), + ) + return None + if encrypted is None: + # Same neutral shape as the image path: the accurate reason (cap + # abort vs Content-Encoding refusal) is logged inside + # _download_cdn_bytes; asserting one here can contradict it. + logger.warning("[WeChat] inbound file skipped by download guard, message_id=%s", message_id) + return None decrypted = _decrypt_aes_128_ecb(encrypted, aes_key) if self._max_inbound_file_bytes > 0 and len(decrypted) > self._max_inbound_file_bytes: logger.warning("[WeChat] inbound file exceeds size limit (%d bytes), skipping message_id=%s", len(decrypted), message_id) @@ -1149,6 +1277,9 @@ class WechatChannel(Channel): def _stage_downloaded_file(self, filename: str, content: bytes) -> Path | None: download_dir = self._download_dir() if download_dir is None: + # Silent None here made an attachment vanish with no log line — + # the same observability gap as a mislabeled skip reason. + logger.warning("[WeChat] no state directory configured, dropping staged inbound media file %s", filename) return None try: download_dir.mkdir(parents=True, exist_ok=True) @@ -1477,3 +1608,52 @@ class WechatChannel(Channel): return set(default) normalized = {str(item).strip().lower() if str(item).strip().startswith(".") else f".{str(item).strip().lower()}" for item in value if str(item).strip()} return normalized or set(default) + + def _coerce_host_suffixes(self, value: Any) -> frozenset[str]: + """Resolve the inbound-media host allowlist: operator suffixes plus defaults. + + The configured ``cdn_base_url`` host is always admitted so a custom CDN + endpoint keeps working without touching ``allowed_media_hosts``. + Entries are host suffixes; a leading ``*.`` (a natural DNS habit) is + stripped so ``*.example.com`` behaves exactly like ``example.com`` + instead of silently never matching. + """ + if isinstance(value, str): + values: list[Any] = [value] + elif isinstance(value, (list, tuple, set, frozenset)): + values = list(value) + else: + values = [] + suffixes: set[str] = set() + for item in values: + text = str(item).strip().lower().lstrip(".") + if text.startswith("*."): + text = text[2:] + if text: + suffixes.add(text) + suffixes.update(self.DEFAULT_ALLOWED_MEDIA_HOST_SUFFIXES) + cdn_host = urlparse(self._cdn_base_url).hostname + if cdn_host: + suffixes.add(cdn_host.strip().lower().lstrip(".")) + return frozenset(suffixes) + + def _is_allowed_media_url(self, url: str) -> bool: + """Gate inbound media fetches to the platform CDN domains. + + ``full_url`` is message-payload data relayed by the platform; like the + DingTalk channel's ``download_code``, it is treated as untrusted input. + The fetch runs on the Gateway host network, so an unrestricted URL + would let a crafted message point it at loopback/private services. + Matching is dot-boundary aware, so ``notqq.com`` or + ``qq.com.evil.io`` never match a ``qq.com`` suffix. + """ + try: + parsed = urlparse(url) + except ValueError: + return False + if parsed.scheme not in ("http", "https"): + return False + host = (parsed.hostname or "").lower() + if not host: + return False + return any(host == suffix or host.endswith(f".{suffix}") for suffix in self._allowed_media_hosts) diff --git a/backend/app/channels/wecom.py b/backend/app/channels/wecom.py index abb97ad24..154b232d4 100644 --- a/backend/app/channels/wecom.py +++ b/backend/app/channels/wecom.py @@ -106,6 +106,35 @@ class WeComChannel(Channel): self._ws_send_lock_users: dict[str, int] = {} self._ws_send_locks_guard = asyncio.Lock() self._working_message = "Working on it..." + raw_hosts = config.get("allowed_media_hosts") + if isinstance(raw_hosts, str): + host_values: list[Any] = [raw_hosts] + elif isinstance(raw_hosts, (list, tuple, set, frozenset)): + host_values = list(raw_hosts) + else: + host_values = [] + # Host suffixes; a leading ``*.`` is stripped so ``*.example.com`` and + # ``example.com`` behave identically (mirrors WechatChannel._coerce_host_suffixes). + normalized_hosts: set[str] = set() + for host in host_values: + text = str(host).strip().lower().lstrip(".") + if text.startswith("*."): + text = text[2:] + if text: + normalized_hosts.add(text) + self._allowed_media_host_suffixes = frozenset(normalized_hosts) + + @property + def allowed_media_host_suffixes(self) -> frozenset[str]: + """Operator host suffixes the manager-side inbound-media gate merges in. + + ``channels.wecom.allowed_media_hosts``: extra host suffixes inbound + media URLs may be downloaded from, on top of the built-in ``qq.com`` + family and the pinned WeCom COS bucket shape. Gives deployments that + proxy or mirror WeCom media an escape hatch that does not require + widening the hard-coded pattern for everyone. + """ + return self._allowed_media_host_suffixes @property def supports_streaming(self) -> bool: diff --git a/backend/packages/harness/deerflow/logging_config.py b/backend/packages/harness/deerflow/logging_config.py index 164b62d21..8790eec3c 100644 --- a/backend/packages/harness/deerflow/logging_config.py +++ b/backend/packages/harness/deerflow/logging_config.py @@ -4,6 +4,8 @@ from __future__ import annotations import json import logging +import re +from collections.abc import Iterator from datetime import UTC, datetime from typing import Any @@ -15,6 +17,265 @@ DEFAULT_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - %(message)s" TRACE_TEXT_LOG_FORMAT = "%(asctime)s - %(name)s - %(levelname)s - [trace_id=%(trace_id)s] - %(message)s" _TRACE_FILTER_NAME = "deerflow_trace_context_filter" +# httpx logs ``HTTP Request: GET HTTP/x.x `` at +# INFO before any response handling runs, and urllib3 logs +# ``Redirecting -> `` at INFO when a redirect is followed. Inbound +# media URLs are signed — the credentials live in the query string, and the +# repo-wide inbound-media rule is that no part of a media URL beyond its host +# may reach the logs — so even successful downloads would leak unless the +# record itself is rewritten. The authority is split so userinfo (basic-auth +# ``user:pass@`` credentials, accepted by httpx for MCP/extension/community- +# tool endpoints) is blanked too, not just the path and query. ``rest`` is +# optional so an authority-only URL (``scheme://user:pass@host`` — no path) +# is still rewritten; a bare credential-free origin passes through as-is. +# ``rest`` treats a quote as a closing mark only at a boundary — followed by +# whitespace, a closing parenthesis, or end of string — so URLs wrapped in +# surrounding prose or a format's own quoting (urllib3's +# ``Incremented Retry for (url='…')``) keep their closing punctuation, while +# a quote EMBEDDED in the URL (``/path'quoted'?token=…``) is consumed and +# everything after it stays redacted. +_URL_REDACT_RE = re.compile(r"(?P[a-zA-Z][a-zA-Z0-9+.-]*://)(?P[^/?#\s@]*@)?(?P[^/?#\s]+)(?P[/?#](?:[^\s'\"]|['\"](?!$|\s|\)))*)?") + +# urllib3's per-request DEBUG line (connectionpool.py:545 on urllib3 2.7.0) +# splits the URL across the format string: +# ``'%s://%s:%s "%s %s %s" %s %s'`` renders as +# ``scheme://host:port "GET /private/x?token=y HTTP/1.1" 200 None`` — the +# authority and the signed origin-form target are two separate args. The +# absolute-URL regex above cannot see either half: the authority is followed +# by a space (so ``rest`` never matches and the bare-origin early return +# applies) and the quoted target has no scheme. The request line therefore +# gets its own shape — authority immediately followed by a quoted +# ``METHOD target HTTP/x.x`` line — rewritten to scheme + host with the +# target collapsed to ``/``. The method class is case-tolerant: +# HTTP methods are case-sensitive tokens, and callers may pass lowercase +# custom methods through to urllib3. +_URLLIB3_REQUEST_LINE_RE = re.compile(r'(?P[a-zA-Z][a-zA-Z0-9+.-]*://)(?P[^/?#\s"@]*@)?(?P[^/?#\s"]+) "(?P[A-Za-z]+) (?P/[^"\s]*) (?PHTTP/[0-9.]+)"') + +# urllib3's retry sites log the request target with NO scheme and NO request- +# line scaffolding, so neither pattern above can see it (installed 2.7.0): +# - ``Retry: %s`` — connectionpool.py:954, DEBUG, origin-form target. +# - ``Incremented Retry for (url='%s'): %r`` — util/retry.py:545, DEBUG via +# the ``urllib3.util.retry`` logger; the target is origin-form on the +# request path and absolute on the redirect path (poolmanager resolves the +# Location before retrying). The url capture applies the same boundary +# idea as ``rest``, narrowed to this format's fixed ``')`` closer: an +# embedded quote is consumed, the closing quote is the one directly +# followed by ``)``. Absolute targets are left for the generic absolute- +# URL pass — its ``rest`` stops at the closing quote — while origin-form +# targets collapse to ``/``. +# - ``Retrying (%r) after connection broken by '%r': %s`` — +# connectionpool.py:869, **WARNING**, so it passes the Gateway's INFO root +# without DEBUG being enabled; the greedy prefix groups pin the split to +# the final ``': `` so an error repr containing quotes cannot shift it. +_URLLIB3_RETRY_TARGET_RE = re.compile(r"^Retry: (?P/\S+)$") +_URLLIB3_INCREMENT_RETRY_RE = re.compile(r"Incremented Retry for \(url='(?P(?:[^']|'(?!\)))*)'\)") +_URLLIB3_RETRYING_RE = re.compile(r"^(?PRetrying \(.*\) after connection broken by .*'): (?P/\S+)$") + +# urllib3's ``Redirecting %s -> %s`` (poolmanager.py:500 at INFO, +# connectionpool.py:922 at DEBUG) can carry an origin-form target in either +# slot: connectionpool passes the origin-form request target, and the Location +# header may itself be a relative reference (RFC 9110 allows it). The generic +# absolute-URL pass only sees scheme-bearing halves, so origin-form slots +# collapse to ``/`` here; absolute slots are left for that pass. +# The pattern keeps the ``^Redirecting `` prefix anchor — the urllib3-owned +# literal — because an ``-> /path`` arrow is not urllib3-owned shape: +# non-URL logs render it too (sandbox mount mappings log +# ``sandbox.mounts entry -> ``), and a substring +# match rewrote the container path in that actionable error (CI round 11). +# The tail is deliberately loose: ``redirect_location`` is the raw Location +# header string, and interior spaces are legal field syntax a misbehaving +# server can emit — a whitespace-strict tail would void the pass entirely +# and leak the origin-form request target in the first slot (round 13). A +# space-carrying second slot collapses whole when it starts with ``/``. The +# first slot gets the same grammar treatment: the recursive urlopen frame +# passes the previous raw Location as its url, so t1 can carry interior +# spaces too — it is lazy, splitting at the FIRST `` -> `` the way the +# line was constructed left to right. +_URLLIB3_REDIRECTING_ORIGIN_RE = re.compile(r"^Redirecting (?P\S.*?) -> (?P\S.*)$") + +# The two scheme-bearing patterns start with a character class, so re.sub +# retries the match at every position of a long token — a letter run with no +# ``://`` makes each attempt walk to the end of the run, which is quadratic +# overall (a 64K-character path or error body costs seconds per record, and +# this filter runs synchronously in every root handler). Both are instead +# driven from the literal ``://`` occurrences: each one walks back over the +# scheme charset to the first letter of its maximal run, and the pattern is +# attempted only there. A regex match can only start at such a position, and +# if it fails at the run's first letter it fails identically at every other +# letter of the run (the scheme group is the only part that differs), so this +# reproduces re.sub's leftmost-non-overlapping result in linear time. The +# span-skip below is safe for the same reason: a match of either pattern +# always ends at a character outside the scheme charset (whitespace, quote, +# ``/``, ``?``, ``#``, or end of string — never a letter/digit/``+``/``-``/ +# ``.``), so a scheme run — and with it a candidate start — can never +# straddle a previous match's end; re.sub likewise never re-enters a +# consumed span. +_SCHEME_TAIL_CHARS = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+.-") +_SCHEME_HEAD_CHARS = frozenset("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz") + + +def _scheme_starts(message: str) -> Iterator[int]: + pos = message.find("://") + while pos != -1: + start = pos + while start > 0 and message[start - 1] in _SCHEME_TAIL_CHARS: + start -= 1 + # Leading non-letters (digits, +, -, .) are valid scheme TAIL + # characters but cannot start the scheme, so the match starts at the + # run's first letter; a run with none cannot start a match at all. + while start < pos and message[start] not in _SCHEME_HEAD_CHARS: + start += 1 + if start < pos: + yield start + pos = message.find("://", pos + 1) + + +def _redact_scheme_bearing(pattern: re.Pattern[str], rewrite, message: str) -> str: + parts: list[str] = [] + last = 0 + for start in _scheme_starts(message): + if start < last: # inside the span of the previous match + continue + match = pattern.match(message, start) + if match is None: + continue + parts.append(message[last:start]) + parts.append(rewrite(match)) + last = match.end() + if not parts: + return message + parts.append(message[last:]) + return "".join(parts) + + +class UrlRedactionFilter(logging.Filter): + """Redact URLs in httpx/urllib3 request log records down to scheme + host. + + Path, query, fragment, and any userinfo credentials in the authority are + replaced; the host (and port) stay for operator debuggability. urllib3 + splits or disassembles the URL across several of its own log formats, and + the generic absolute-URL pattern only sees a scheme-bearing URL in one + piece, so each remaining shape gets its own rewrite anchored to the + exact urllib3 format — ``^``-anchored for whole-message lines, literal- + prefix-anchored otherwise: the + per-request ``scheme://host:port "METHOD target HTTP/x.x"`` line, the + retry lines that log a bare origin-form target (``Retry: ``, + ``Incremented Retry for (url='')``, ``Retrying (…) after + connection broken by '…': ``), and origin-form halves of + ``Redirecting -> ``. The record is rewritten in place + (``msg`` set to the redacted formatted message, ``args`` cleared) so + every downstream handler and formatter — text or JSON — sees the same + redacted line, while the method/status/error observability is preserved. + The scheme-bearing patterns are attempted only at ``://``-anchored + scheme starts, so filtering a record costs linear time in its message + length. A URL is rewritten only when it carries something to hide + (userinfo, path, query, or fragment); a bare credential-free origin and + records without any URL pass through untouched. + """ + + def filter(self, record: logging.LogRecord) -> bool: + message = record.getMessage() + + def _redact(match: re.Match[str]) -> str: + if not (match.group("userinfo") or match.group("rest")): + return match.group(0) # bare origin: nothing to redact + userinfo = "@" if match.group("userinfo") else "" + return match.group("scheme") + userinfo + match.group("host") + "/" + + def _redact_request_line(match: re.Match[str]) -> str: + userinfo = "@" if match.group("userinfo") else "" + return match.group("scheme") + userinfo + match.group("host") + ' "' + match.group("method") + " / " + match.group("version") + '"' + + def _redact_increment(match: re.Match[str]) -> str: + url = match.group("url") + if _URL_REDACT_RE.fullmatch(url): + # Absolute target (redirect path): the generic absolute-URL + # pass rewrites it, and its rest stops at the closing quote. + return match.group(0) + return "Incremented Retry for (url='/')" + + def _redact_retry_target(match: re.Match[str]) -> str: + return "Retry: /" + + def _redact_retrying(match: re.Match[str]) -> str: + return match.group("head") + ": /" + + def _redact_redirecting_origin(match: re.Match[str]) -> str: + # Origin-form slots collapse; absolute slots stay for the generic + # absolute-URL pass (which runs after this one). + def _slot(target: str) -> str: + return "/" if target.startswith("/") else target + + return "Redirecting " + _slot(match.group("t1")) + " -> " + _slot(match.group("t2")) + + # The urllib3 shape passes run before the absolute-URL pass: their + # rewrites either leave scheme-bearing text for that pass to handle + # or collapse the target before it could interact with surrounding + # punctuation, while the reverse order would already have rewritten + # an authority into shapes the urllib3 patterns no longer match. The + # two scheme-bearing passes scan from ``://`` occurrences (see + # _scheme_starts) instead of re.sub, so long letter runs in any + # record — URL paths or URL-free error bodies — stay linear-time. + redacted = _redact_scheme_bearing(_URLLIB3_REQUEST_LINE_RE, _redact_request_line, message) + redacted = _URLLIB3_INCREMENT_RETRY_RE.sub(_redact_increment, redacted) + redacted = _URLLIB3_RETRY_TARGET_RE.sub(_redact_retry_target, redacted) + redacted = _URLLIB3_RETRYING_RE.sub(_redact_retrying, redacted) + redacted = _URLLIB3_REDIRECTING_ORIGIN_RE.sub(_redact_redirecting_origin, redacted) + redacted = _redact_scheme_bearing(_URL_REDACT_RE, _redact, redacted) + if redacted != message: + record.msg = redacted + record.args = None + return True + + +# The filter class is generic over the formatted message, so it serves any +# HTTP client library whose records embed full URLs. Where it must be +# ATTACHED differs per library, because a logging.Filter on a logger only +# runs for records emitted through that exact logger — it is not inherited +# by child loggers and never sees propagated records: +# - httpx emits via the bare ``httpx`` logger, so a logger filter works. +# - urllib3 emits via children (``urllib3.poolmanager`` logs +# ``Redirecting -> `` at INFO, ``urllib3.connectionpool`` logs +# redirect lines, the per-request authority/quoted-target line, and the +# ``Retry:``/``Retrying`` lines at DEBUG/WARNING, and ``urllib3.util.retry`` +# logs ``Incremented Retry for (url=…)`` at DEBUG), so a filter on bare +# ``urllib3`` is dead code. Handler-level filters DO see propagated +# records, so the filter is also attached to every root handler — covering +# urllib3 and any future library without knowing its logger names. +# The enumeration of urllib3's URL-bearing lines is closed against the +# installed source (2.7.0): every other emitter logs host:port only +# (connection establishment/reset) or an absolute URL in one piece +# (``connection.py``'s header-parse warning), which the generic absolute-URL +# pass rewrites without a dedicated shape. The closure is version-anchored: +# a urllib3 upgrade can change these format strings and silently reopen it — +# re-run the emitter enumeration when bumping the dependency. +_REDACTED_LOGGER_NAMES = ("httpx",) + + +def _has_url_redaction_filter(handler: logging.Handler) -> bool: + return any(isinstance(item, UrlRedactionFilter) for item in handler.filters) + + +def _install_url_redaction_filter(handler: logging.Handler) -> None: + if not _has_url_redaction_filter(handler): + handler.addFilter(UrlRedactionFilter()) + + +def install_url_log_redaction() -> None: + """Attach URL redaction to the ``httpx`` logger and to every root handler. + + The httpx logger filter covers records at their emission point (httpx + logs via the bare ``httpx`` name); the root-handler filters cover + propagated records from libraries that emit through child loggers, such + as urllib3's ``urllib3.poolmanager`` / ``urllib3.connectionpool``. + """ + for logger_name in _REDACTED_LOGGER_NAMES: + target = logging.getLogger(logger_name) + if not any(isinstance(item, UrlRedactionFilter) for item in target.filters): + target.addFilter(UrlRedactionFilter()) + for handler in logging.root.handlers: + _install_url_redaction_filter(handler) + class TraceContextFilter(logging.Filter): """Inject the current request trace id into every log record.""" @@ -92,12 +353,17 @@ def configure_logging(config: object) -> None: only the additional ``trace_id`` field. """ _ensure_root_handler() + install_url_log_redaction() logging_config = getattr(config, "logging", None) enhance = getattr(logging_config, "enhance", None) enhanced = bool(getattr(enhance, "enabled", False)) for handler in logging.root.handlers: + _install_url_redaction_filter(handler) + # URL redaction is level-agnostic and applies whether or not the + # trace enhancement is on; handler filters see propagated records + # from child loggers (urllib3 et al.), which logger filters cannot. if enhanced: _install_trace_filter(handler) handler.setFormatter(_trace_formatter(getattr(enhance, "format", "text"))) diff --git a/backend/tests/blocking_io/test_wechat_channel_state.py b/backend/tests/blocking_io/test_wechat_channel_state.py index b1f966b77..ff8236a3f 100644 --- a/backend/tests/blocking_io/test_wechat_channel_state.py +++ b/backend/tests/blocking_io/test_wechat_channel_state.py @@ -71,7 +71,7 @@ async def test_wechat_inbound_file_staging_does_not_block_event_loop(tmp_path: P aes_key = b"1234567890abcdef" encrypted = _encrypt_aes_128_ecb(plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -85,7 +85,7 @@ async def test_wechat_inbound_file_staging_does_not_block_event_loop(tmp_path: P "item_list": [ { "type": 2, - "image_item": {"aeskey": aes_key.hex(), "media": {"full_url": "https://cdn.example/image.bin"}}, + "image_item": {"aeskey": aes_key.hex(), "media": {"full_url": "https://cdn.weixin.qq.com/image.bin"}}, } ], } diff --git a/backend/tests/test_channel_file_attachments.py b/backend/tests/test_channel_file_attachments.py index b7f15ab8e..1bd2f7a76 100644 --- a/backend/tests/test_channel_file_attachments.py +++ b/backend/tests/test_channel_file_attachments.py @@ -614,3 +614,588 @@ class TestManagerArtifactResolution: result = _format_artifact_text(["/mnt/user-data/outputs/a.txt", "/mnt/user-data/outputs/b.txt"]) assert "a.txt" in result assert "b.txt" in result + + +# --------------------------------------------------------------------------- +# URL-based inbound file reader (WeCom media / WeChat full_url fallback) +# --------------------------------------------------------------------------- + + +class _FakeStreamResponse: + def __init__(self, chunks: list[bytes], headers: dict[str, str] | None = None): + self._chunks = chunks + self.headers = headers or {} + + def raise_for_status(self) -> None: + return None + + # Deliberately no aiter_bytes: the reader must consume undecoded bytes, so + # an accidental switch back to the decoding iterator fails loudly here. + async def aiter_raw(self): + for chunk in self._chunks: + yield chunk + + +class _FakeStreamContext: + def __init__(self, response: _FakeStreamResponse): + self._response = response + + async def __aenter__(self) -> _FakeStreamResponse: + return self._response + + async def __aexit__(self, *exc_info) -> bool: + return False + + +class _FakeStreamingClient: + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + + def stream(self, _method: str, _url: str, **_kwargs) -> _FakeStreamContext: + return _FakeStreamContext(_FakeStreamResponse(self._chunks)) + + +class TestHttpInboundFileReader: + def test_joins_streamed_chunks_under_the_cap(self): + from app.channels.manager import _read_http_inbound_file + + client = _FakeStreamingClient([b"ab", b"cd"]) + + result = _run(_read_http_inbound_file({"url": "https://cdn.example/x", "filename": "x.bin"}, client)) # type: ignore[arg-type] + + assert result == b"abcd" + + def test_aborts_when_stream_exceeds_the_cap(self, monkeypatch): + from app.channels import manager + + monkeypatch.setattr(manager, "MAX_INBOUND_URL_FILE_BYTES", 5) + client = _FakeStreamingClient([b"abc", b"def"]) # 6 bytes total vs a 5-byte cap + + result = _run(manager._read_http_inbound_file({"url": "https://cdn.example/x?token=1", "filename": "x.bin"}, client)) # type: ignore[arg-type] + + assert result is None + + def test_missing_url_returns_none(self): + from app.channels.manager import _read_http_inbound_file + + assert _run(_read_http_inbound_file({"filename": "x.bin"}, _FakeStreamingClient([]))) is None # type: ignore[arg-type] + + def test_compressed_response_is_rejected_before_decode(self, caplog): + """aiter_bytes() would transparently decode Content-Encoding, and the + decoder allocates the whole decompressed body before yielding a chunk + — a gzip bomb from an admitted host would bypass the byte cap (one + ~8 KB wire chunk decoding to 8 MiB, reproduced here). The reader must + request identity and refuse any residual encoding before reading.""" + import gzip as _gzip + import logging as _logging + + import httpx + + from app.channels import manager + + class _AsyncChunks(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + + async def __aiter__(self): + for chunk in self._chunks: + yield chunk + + seen_requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(request) + return httpx.Response( + 200, + headers={"Content-Encoding": "gzip"}, + stream=_AsyncChunks([_gzip.compress(b"\x00" * (8 * 1024 * 1024))]), + ) + + async def go(): + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + return await manager._read_http_inbound_file({"url": "https://cdn.example/x", "filename": "x.bin"}, client) + finally: + await client.aclose() + + with caplog.at_level(_logging.WARNING, logger="app.channels.manager"): + result = _run(go()) + + assert result is None + assert "Content-Encoding" in caplog.text + # Identity is requested up front so a compliant server never compresses. + assert seen_requests[0].headers.get("accept-encoding") == "identity" + + def test_identity_response_round_trips(self): + """An unencoded response still streams and joins as before (real httpx transport).""" + import httpx + + from app.channels.manager import _read_http_inbound_file + + class _AsyncChunks(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + + async def __aiter__(self): + for chunk in self._chunks: + yield chunk + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, stream=_AsyncChunks([b"ab"])) + + async def go(): + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + return await _read_http_inbound_file({"url": "https://cdn.example/x", "filename": "x.bin"}, client) + finally: + await client.aclose() + + assert _run(go()) == b"ab" + + +class _RecordingStreamClient(_FakeStreamingClient): + def __init__(self, chunks: list[bytes]): + super().__init__(chunks) + self.streamed_urls: list[str] = [] + + def stream(self, method: str, url: str, **kwargs) -> _FakeStreamContext: + self.streamed_urls.append(url) + return super().stream(method, url, **kwargs) + + +class TestWecomMediaUrlGate: + def test_disallowed_host_rejected_before_fetch(self): + from app.channels.manager import _read_wecom_inbound_file + + client = _RecordingStreamClient([b"secret-internal-response"]) + + result = _run( + _read_wecom_inbound_file( + {"url": "http://169.254.169.254/latest/meta-data", "filename": "x.bin"}, + client, # type: ignore[arg-type] + ) + ) + + assert result is None + assert client.streamed_urls == [] + + def test_qq_host_passes_through_to_fetch(self): + from app.channels.manager import _read_wecom_inbound_file + + client = _RecordingStreamClient([b"ab"]) + + result = _run( + _read_wecom_inbound_file( + {"url": "https://cdn.work.weixin.qq.com/media/x?token=1", "filename": "x.bin", "aeskey": None}, + client, # type: ignore[arg-type] + ) + ) + + assert result == b"ab" + assert client.streamed_urls == ["https://cdn.work.weixin.qq.com/media/x?token=1"] + + def test_lookalike_suffix_never_matches(self): + from app.channels.manager import _is_allowed_wecom_media_url + + assert not _is_allowed_wecom_media_url("https://notqq.com/x") + assert not _is_allowed_wecom_media_url("https://qq.com.evil.io/x") + assert not _is_allowed_wecom_media_url("file:///etc/passwd") + assert _is_allowed_wecom_media_url("https://cdn.weixin.qq.com/x") + + def test_realistic_wecom_cos_host_passes_gate(self): + """WeCom serves media from signed COS links shaped ww-aibot-img-.cos..myqcloud.com. + + Published callback examples (go-sphere/wecom-bot-api API.md) use exactly + this shape for both image.url and file.url; a bare qq.com allowlist + would drop every normal WeCom attachment. + """ + from app.channels.manager import _is_allowed_wecom_media_url, _read_wecom_inbound_file + + realistic = "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/BHoPdA3/7571665296904772241?sign=q-sign-algorithm%3Dsha1&q-signature=QuerySecret" + assert _is_allowed_wecom_media_url(realistic) + + client = _RecordingStreamClient([b"ab"]) + result = _run( + _read_wecom_inbound_file( + {"url": realistic, "aeskey": None}, + client, # type: ignore[arg-type] + ) + ) + assert result == b"ab" + assert client.streamed_urls == [realistic] + + def test_arbitrary_cos_bucket_never_matches(self): + """The COS numeric suffix is the owner's Tencent Cloud APPID and bucket + names are user-chosen, so only the APPID observed in Tencent's + published aibot callback examples (1258476243) matches — not a custom + bucket name, a different account's APPID, or a lookalike domain.""" + from app.channels.manager import _is_allowed_wecom_media_url + + assert not _is_allowed_wecom_media_url("https://my-own-bucket.cos.ap-guangzhou.myqcloud.com/x?sign=1") + assert not _is_allowed_wecom_media_url("https://attacker-prefix-1.cos.ap-guangzhou.myqcloud.com/x?sign=1") + assert not _is_allowed_wecom_media_url("https://ww-aibot-img-evil.cos.ap-guangzhou.myqcloud.com/x?sign=1") + assert not _is_allowed_wecom_media_url("https://ww-aibot-img-123.cos.ap-guangzhou.myqcloud.com.evil.io/x") + # Same bucket prefix, different Tencent Cloud account (reviewer repro): + # any account can register a "ww-aibot-img-*" bucket of its own. + assert not _is_allowed_wecom_media_url("https://ww-aibot-img-1250000000.cos.ap-guangzhou.myqcloud.com/x?sign=1") + assert not _is_allowed_wecom_media_url("https://ww-aibot-img-0.cos.ap-shanghai.myqcloud.com/x") + # The verified APPID is trusted across regions. + assert _is_allowed_wecom_media_url("https://ww-aibot-img-1258476243.cos.ap-shanghai.myqcloud.com/x") + + def test_oversize_rejection_does_not_log_url_credentials(self, monkeypatch, caplog): + """Signed media URLs carry credentials in path and query; only the host may be logged.""" + import logging as _logging + + from app.channels import manager + + monkeypatch.setattr(manager, "MAX_INBOUND_URL_FILE_BYTES", 4) + url = "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/private/BearerSecret?token=QuerySecret" + client = _FakeStreamingClient([b"abcdef"]) + + with caplog.at_level(_logging.WARNING, logger="app.channels.manager"): + result = _run(manager._read_http_inbound_file({"url": url}, client)) # type: ignore[arg-type] + + assert result is None + logged = "\n".join(record.getMessage() for record in caplog.records) + assert "BearerSecret" not in logged + assert "QuerySecret" not in logged + assert "/private/" not in logged + assert "ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com" in logged # host label is present + + def test_inbound_file_label_never_contains_url(self): + from app.channels.manager import _inbound_file_label + + # Whitespace in a webhook-supplied filename is collapsed and capped. + assert _inbound_file_label({"filename": "a\nb " + "x" * 100}) == "a b " + "x" * 76 + # No filename: host-only label from whichever URL field is present. + assert _inbound_file_label({"url": "https://h.example/p?token=t"}) == "host=h.example" + assert _inbound_file_label({"full_url": "https://h2.example/p?token=t"}) == "host=h2.example" + # Nothing usable at all: positional label. + assert _inbound_file_label({}, idx=3) == "#3" + assert _inbound_file_label({}) == "" + + def test_wecom_reader_success_does_not_leak_url_at_info(self, caplog): + """Successful signed-media downloads must not leak credentials at the Gateway's INFO level. + + httpx emits ``HTTP Request: GET `` at INFO before the reader + sees the response; the filter installed by ``configure_logging`` + rewrites those records down to scheme + host. Real transport logging + path via MockTransport, success included — not just failure branches. + """ + import logging as _logging + + import httpx + + from app.channels import manager + from deerflow.logging_config import UrlRedactionFilter, install_url_log_redaction + + class _AsyncBody(httpx.AsyncByteStream): + async def __aiter__(self): + yield b"ok" + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, stream=_AsyncBody()) + + install_url_log_redaction() + url = "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/private/BearerSecret?token=QuerySecret" + + async def go(): + client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + try: + return await manager._read_wecom_inbound_file({"url": url, "aeskey": None}, client) + finally: + await client.aclose() + + with caplog.at_level(_logging.INFO): + result = _run(go()) + + assert result == b"ok" + formatted = "\n".join(record.getMessage() for record in caplog.records) + request_lines = [line for line in formatted.splitlines() if "HTTP Request" in line] + assert request_lines, "the request record itself must survive redaction (not suppression)" + assert any(isinstance(f, UrlRedactionFilter) for f in _logging.getLogger("httpx").filters) + assert "BearerSecret" not in formatted + assert "QuerySecret" not in formatted + assert "/private/" not in formatted + assert any("ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/" in line for line in request_lines) + + def test_wechat_download_success_does_not_leak_url_at_info(self, caplog): + """WechatChannel._download_cdn_bytes hits the same httpx INFO path on success.""" + import logging as _logging + + import httpx + + from app.channels.wechat import WechatChannel + from deerflow.logging_config import install_url_log_redaction + + class _AsyncBody(httpx.AsyncByteStream): + async def __aiter__(self): + yield b"ok" + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, stream=_AsyncBody()) + + install_url_log_redaction() + channel = WechatChannel(MessageBus(), config={"bot_token": "test-token"}) + channel._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) # type: ignore[assignment] + + async def go(): + try: + return await channel._download_cdn_bytes("https://cdn.weixin.qq.com/private/BearerSecret?token=QuerySecret") + finally: + await channel._client.aclose() + + with caplog.at_level(_logging.INFO): + result = _run(go()) + + assert result == b"ok" + formatted = "\n".join(record.getMessage() for record in caplog.records) + assert "BearerSecret" not in formatted + assert "QuerySecret" not in formatted + assert "cdn.weixin.qq.com/" in formatted + + def test_outer_warning_branches_do_not_log_url_credentials(self, tmp_path, caplog): + """The _ingest_inbound_files reader-exception and no-data branches use host-only labels. + + A reader that raises and one that returns None for a url-bearing file + dict without a filename are the real shapes that reach these branches + (WeCom gate drop, cap abort, download failure). + """ + import logging as _logging + + from app.channels import manager + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + secret_url = "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/private/BearerSecret?token=QuerySecret" + + async def raising_reader(_file_info, _client): + raise RuntimeError("reader boom") + + async def none_reader(_file_info, _client): + return None + + with caplog.at_level(_logging.WARNING, logger="app.channels.manager"): + for reader in (raising_reader, none_reader): + msg = InboundMessage( + channel_name="test-channel", + chat_id="chat-1", + user_id="user-1", + text="see attachment", + files=[{"url": secret_url}], + ) + with ( + patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir), + patch.dict(manager.INBOUND_FILE_READERS, {"test-channel": reader}, clear=False), + ): + assert _run(manager._ingest_inbound_files("thread-1", msg)) == [] + + logged = "\n".join(record.getMessage() for record in caplog.records) + assert "BearerSecret" not in logged + assert "QuerySecret" not in logged + assert "/private/" not in logged + assert "host=ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com" in logged + + def test_http_failure_does_not_log_url_credentials(self, tmp_path, monkeypatch, caplog): + """A real HTTP failure must not leak the signed URL through the + exception traceback. + + httpx.HTTPStatusError formats the full request URL (path + query, i.e. + the download credentials) into its message; logger.exception would + render it verbatim. Reproduces the reviewer's mock-transport 403 + through the real _ingest_inbound_files() and asserts against the + fully formatted logs (caplog.text includes rendered tracebacks). + """ + import logging as _logging + + import httpx + + from app.channels import manager + + uploads_dir = tmp_path / "uploads" + uploads_dir.mkdir() + secret_url = "https://ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com/private/BearerSecret?token=QuerySecret" + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(403) + + real_async_client = httpx.AsyncClient + + def patched_async_client(**kwargs): + kwargs["transport"] = httpx.MockTransport(handler) + return real_async_client(**kwargs) + + monkeypatch.setattr(manager.httpx, "AsyncClient", patched_async_client) + + msg = InboundMessage( + channel_name="wecom", + chat_id="chat-1", + user_id="user-1", + text="see attachment", + files=[{"url": secret_url}], + ) + + with caplog.at_level(_logging.WARNING, logger="app.channels.manager"): + with patch("deerflow.uploads.manager.ensure_uploads_dir", return_value=uploads_dir): + assert _run(manager._ingest_inbound_files("thread-1", msg)) == [] + + assert "BearerSecret" not in caplog.text + assert "QuerySecret" not in caplog.text + assert "/private/" not in caplog.text + # The operator still sees what failed and for which host. + assert "HTTPStatusError (403)" in caplog.text + assert "ww-aibot-img-1258476243.cos.ap-guangzhou.myqcloud.com" in caplog.text + + def test_operator_media_host_suffixes_extend_the_wecom_gate(self): + from app.channels.manager import _is_allowed_wecom_media_url + + proxied = "https://media.mirror.example/wecom/x?sign=1" + assert not _is_allowed_wecom_media_url(proxied) + assert _is_allowed_wecom_media_url(proxied, extra_suffixes=frozenset({"mirror.example"})) + # Extras widen only by dot-boundary suffix, same as the built-ins. + assert not _is_allowed_wecom_media_url("https://mirror.example.evil.io/x", extra_suffixes=frozenset({"mirror.example"})) + + def test_wecom_reader_resolves_operator_suffixes_from_live_channel(self, monkeypatch): + """The registry reader merges channels.wecom.allowed_media_hosts from the running channel.""" + from types import SimpleNamespace as _NS + + from app.channels import manager + from app.channels import service as service_module + + proxied = "https://media.mirror.example/wecom/x?sign=1" + client = _RecordingStreamClient([b"ab"]) + + fake_service = _NS(get_channel=lambda name: _NS(allowed_media_host_suffixes=frozenset({"mirror.example"})) if name == "wecom" else None) + monkeypatch.setattr(service_module, "get_channel_service", lambda: fake_service) + + result = _run(manager._read_wecom_inbound_file({"url": proxied, "aeskey": None}, client)) # type: ignore[arg-type] + assert result == b"ab" + assert client.streamed_urls == [proxied] + + # Without the operator suffix (no live channel -> strict defaults only) the same URL is dropped pre-fetch. + monkeypatch.setattr(service_module, "get_channel_service", lambda: None) + strict_client = _RecordingStreamClient([b"ab"]) + assert _run(manager._read_wecom_inbound_file({"url": proxied, "aeskey": None}, strict_client)) is None # type: ignore[arg-type] + assert strict_client.streamed_urls == [] + + def test_wecom_channel_parses_operator_media_hosts(self): + from app.channels.wecom import WeComChannel + + channel = WeComChannel( + MessageBus(), + config={"bot_id": "b", "bot_secret": "s", "allowed_media_hosts": [" .Mirror.Example ", "cdn2.example", "*.wild.example", ""]}, + ) + assert channel.allowed_media_host_suffixes == frozenset({"mirror.example", "cdn2.example", "wild.example"}) + + assert WeComChannel(MessageBus(), config={"bot_id": "b", "bot_secret": "s"}).allowed_media_host_suffixes == frozenset() + + def test_wechat_fallback_without_staged_path_never_fetches(self): + """A WeChat file dict with only full_url has no re-fetch path: the URL gate is channel-side.""" + from app.channels.manager import _read_wechat_inbound_file + + client = _RecordingStreamClient([b"secret"]) + + result = _run( + _read_wechat_inbound_file( + {"url": None, "full_url": "https://cdn.weixin.qq.com/image.bin"}, + client, # type: ignore[arg-type] + ) + ) + + assert result is None + assert client.streamed_urls == [] + + +class TestWeChatDownloadGuardLabels: + """Round-14 nit: ``_download_cdn_bytes`` returns None for two reasons + (in-flight cap abort, Content-Encoding refusal), and both used to be + labeled "exceeds size limit (N bytes)" by the callers — a contradictory + pair on the encoding path, where the transfer may be tiny. The callers + now log a neutral guard line; the accurate reason stays inside the + download function (same shape as the manager reader callers).""" + + def _channel_with(self, handler): + import base64 + + import httpx + + from app.channels.wechat import WechatChannel + + channel = WechatChannel(MessageBus(), config={"bot_token": "test-token"}) + channel._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) # type: ignore[assignment] + self._aes_key = base64.b64encode(b"\x01" * 16).decode() + return channel + + @staticmethod + def _image_item(aes_key: str) -> dict: + return {"image_item": {"media": {"full_url": "https://cdn.weixin.qq.com/private/photo.bin", "aes_key": aes_key}}} + + def test_encoding_refusal_logs_neutral_guard_line(self, caplog): + import gzip as _gzip + import logging as _logging + + import httpx + + class _AsyncChunks(httpx.AsyncByteStream): + async def __aiter__(self): + yield _gzip.compress(b"\x00" * 1024) + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, headers={"Content-Encoding": "gzip"}, stream=_AsyncChunks()) + + channel = self._channel_with(handler) + channel._max_inbound_image_bytes = 1024 * 1024 + + with caplog.at_level(_logging.WARNING, logger="app.channels.wechat"): + result = _run(channel._extract_image_file(self._image_item(self._aes_key), message_id="m1", index=0)) + + assert result is None + formatted = "\n".join(record.getMessage() for record in caplog.records) + # Accurate reason from the download function... + assert "Content-Encoding" in formatted + # ...and a neutral caller line that no longer asserts a size limit + # the transfer never hit. + assert "skipped by download guard" in formatted + assert "exceeds size limit" not in formatted + + def test_cap_abort_logs_neutral_guard_line(self, caplog): + import logging as _logging + + import httpx + + class _AsyncChunks(httpx.AsyncByteStream): + async def __aiter__(self): + yield b"\x00" * 8 + yield b"\x00" * 8 + yield b"\x00" * 8 + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, stream=_AsyncChunks()) + + channel = self._channel_with(handler) + channel._max_inbound_image_bytes = 8 # stream cap = padded(8) = 16 < 24 + + with caplog.at_level(_logging.WARNING, logger="app.channels.wechat"): + result = _run(channel._extract_image_file(self._image_item(self._aes_key), message_id="m2", index=0)) + + assert result is None + formatted = "\n".join(record.getMessage() for record in caplog.records) + # The function's accurate in-flight abort line carries the reason... + assert "aborting before full read" in formatted + # ...and the caller stays neutral instead of asserting the plaintext + # limit for a ciphertext cap decision. + assert "skipped by download guard" in formatted + assert "exceeds size limit" not in formatted + + def test_staging_without_state_dir_is_logged_not_silent(self, caplog): + import logging as _logging + + from app.channels.wechat import WechatChannel + + channel = WechatChannel(MessageBus(), config={"bot_token": "test-token"}) + assert channel._download_dir() is None + + with caplog.at_level(_logging.WARNING, logger="app.channels.wechat"): + assert channel._stage_downloaded_file("photo.bin", b"x") is None + + assert "no state directory configured" in caplog.text diff --git a/backend/tests/test_logging_config.py b/backend/tests/test_logging_config.py index 48ee1df0c..8ac354755 100644 --- a/backend/tests/test_logging_config.py +++ b/backend/tests/test_logging_config.py @@ -2,6 +2,8 @@ import io import logging from types import SimpleNamespace +import httpx + from deerflow.logging_config import TraceContextFilter, configure_logging from deerflow.trace_context import request_trace_context @@ -38,3 +40,694 @@ def test_configure_logging_enhanced_text_includes_trace_id() -> None: finally: root.handlers = old_handlers root.setLevel(old_level) + + +# The installed httpx (0.28.1) emits this exact record from _client.py: +# logger.info('HTTP Request: %s %s "%s %d %s"', method, url, version, status, reason) +_HTTPX_REQUEST_FORMAT = 'HTTP Request: %s %s "%s %d %s"' + + +def _httpx_record(url: str, method: str = "GET", status: int = 200) -> logging.LogRecord: + return logging.LogRecord( + "httpx", + logging.INFO, + __file__, + 1, + _HTTPX_REQUEST_FORMAT, + (method, httpx.URL(url), "HTTP/1.1", status, "OK"), + None, + ) + + +def test_url_redaction_filter_rewrites_request_records() -> None: + from deerflow.logging_config import UrlRedactionFilter + + filt = UrlRedactionFilter() + + # Records are built with the real httpx format string and httpx.URL args + # (verified against httpx/_client.py on the installed version), so the + # unit test pins the production format. Redaction runs on the formatted + # message and clears args; path AND query disappear — only scheme + host + # may remain (the repo-wide inbound-media log rule). + record = _httpx_record("https://host.example/private/BearerSecret?token=QuerySecret") + assert filt.filter(record) is True + formatted = record.getMessage() + assert "host.example/" in formatted + assert "BearerSecret" not in formatted + assert "token=" not in formatted + assert "GET" in formatted and "200" in formatted # observability preserved + + # Same class of leak, different secret location: the Telegram Bot API + # carries the bot token in the PATH (api.telegram.org/bot/method), + # and python-telegram-bot's HTTPXRequest rides the same httpx logger — + # redacting down to scheme + host is what keeps telegram.py's promise + # that the token-bearing URL never reaches the logs. + telegram = _httpx_record("https://api.telegram.org/bot123456:AAE-token-secret/sendMessage", method="POST") + assert filt.filter(telegram) is True + telegram_formatted = telegram.getMessage() + assert "api.telegram.org/" in telegram_formatted + assert "AAE-token-secret" not in telegram_formatted + assert "bot123456" not in telegram_formatted + + # Userinfo credentials in the authority (basic-auth style endpoints that + # httpx accepts, e.g. MCP/extension proxies) must be blanked too — the + # authority is split so only @ survives in front of the host. + userinfo = _httpx_record("https://user:token123@internal-proxy.corp:8080/v1/secret-endpoint") + assert filt.filter(userinfo) is True + userinfo_formatted = userinfo.getMessage() + assert "https://@internal-proxy.corp:8080/" in userinfo_formatted + assert "token123" not in userinfo_formatted + assert "user:" not in userinfo_formatted + + # Authority-ONLY URL (no path/query): rest is optional in the regex, so a + # userinfo credential with nowhere else to hide is still blanked. + authority_only = _httpx_record("https://user:tok@internal-proxy.corp") + assert filt.filter(authority_only) is True + authority_formatted = authority_only.getMessage() + assert "https://@internal-proxy.corp" in authority_formatted + assert "tok" not in authority_formatted.replace("", "") + + # A bare credential-free origin without path/query is left as-is. + bare = _httpx_record("https://host.example") + assert filt.filter(bare) is True + assert "https://host.example" in bare.getMessage() + assert "" not in bare.getMessage() + + # Records without a URL pass through untouched (message + args kept). + plain = logging.LogRecord("httpx", logging.INFO, __file__, 1, "keep %s", ("this",), None) + assert filt.filter(plain) is True + assert plain.getMessage() == "keep this" + + +def test_url_redaction_filter_covers_urllib3_redirect_records() -> None: + """Real-emitter wiring: urllib3 logs through CHILD loggers, and a filter on + the bare ``urllib3`` logger never sees propagated records (logger filters + are not inherited). Verified against the installed urllib3 2.7.0: + ``urllib3.poolmanager`` logs ``Redirecting %s -> %s`` at INFO + (poolmanager.py:500) and ``urllib3.connectionpool`` logs the same shape at + DEBUG (connectionpool.py:922). Both must come out redacted through the + real emit path with configure_logging's handler-level installation.""" + from deerflow.logging_config import configure_logging + + root = logging.getLogger() + old_handlers = root.handlers[:] + old_level = root.level + stream = io.StringIO() + handler = logging.StreamHandler(stream) + + try: + root.handlers = [handler] + root.setLevel(logging.DEBUG) + configure_logging(SimpleNamespace(log_level="debug", logging=SimpleNamespace(enhance=SimpleNamespace(enabled=False, format="text")))) + + logging.getLogger("urllib3.poolmanager").info( + "Redirecting %s -> %s", + "https://cdn.example/private/BearerSecret?token=QuerySecret", + "https://mirror.example/private/BearerSecret?sig=OtherSecret", + ) + logging.getLogger("urllib3.connectionpool").debug( + "Redirecting %s -> %s", + "https://cdn.example/private/BearerSecret?token=QuerySecret", + "https://mirror.example/private/BearerSecret?sig=OtherSecret", + ) + + out = stream.getvalue() + assert "BearerSecret" not in out + assert "QuerySecret" not in out + assert "OtherSecret" not in out + assert out.count("cdn.example/") == 2 + assert out.count("mirror.example/") == 2 + finally: + root.handlers = old_handlers + root.setLevel(old_level) + + +def test_url_redaction_filter_covers_urllib3_request_line_records() -> None: + """urllib3's per-request DEBUG line splits the URL across its format + string, so the absolute-URL regex alone cannot catch it. The installed + urllib3 (2.7.0) emits this exact record from + HTTPConnectionPool._make_request (connectionpool.py:545): + log.debug('%s://%s:%s "%s %s %s" %s %s', scheme, host, port, method, + url, response.version_string, response.status, + response.length_remaining) — the authority ends at a space (bare-origin + early return) and the quoted origin-form target has no scheme, which is + why the request line needs its own redaction shape.""" + from deerflow.logging_config import UrlRedactionFilter + + format_string = '%s://%s:%s "%s %s %s" %s %s' + filt = UrlRedactionFilter() + + def _record(target: str, version: str = "HTTP/1.1", method: str = "GET") -> logging.LogRecord: + return logging.LogRecord( + "urllib3.connectionpool", + logging.DEBUG, + __file__, + 1, + format_string, + ("https", "cdn.example", 443, method, target, version, 200, None), + None, + ) + + # The reviewer's repro shape: host:port, then a quoted request line whose + # origin-form target carries the signed path+query. The rewrite keeps + # scheme + host + method + version for observability and collapses the + # target to /. + record = _record("/private/BearerSecret?token=QuerySecret") + assert filt.filter(record) is True + formatted = record.getMessage() + assert formatted == 'https://cdn.example:443 "GET / HTTP/1.1" 200 None' + assert "BearerSecret" not in formatted + assert "token=" not in formatted + + # A target with no query still hides the path: the inbound-media rule is + # host-only visibility, not query-only. + path_only = _record("/private/photo.jpg") + assert filt.filter(path_only) is True + assert '"GET / HTTP/1.1"' in path_only.getMessage() + assert "photo.jpg" not in path_only.getMessage() + + # HTTP/2 responses keep version_string in the quoted line; the shape must + # still match and rewrite. + http2 = _record("/private/BearerSecret?token=QuerySecret", version="HTTP/2") + assert filt.filter(http2) is True + assert '"GET / HTTP/2"' in http2.getMessage() + assert "BearerSecret" not in http2.getMessage() + + +def test_url_redaction_filter_covers_urllib3_request_line_through_real_emit() -> None: + """Real-emitter wiring for the per-request line: urllib3 logs through the + ``urllib3.connectionpool`` child logger at DEBUG, so only the + handler-level filters installed by configure_logging can rewrite the + record. Emits with the connectionpool.py:545 format string at root + DEBUG.""" + from deerflow.logging_config import configure_logging + + root = logging.getLogger() + old_handlers = root.handlers[:] + old_level = root.level + stream = io.StringIO() + handler = logging.StreamHandler(stream) + + try: + root.handlers = [handler] + root.setLevel(logging.DEBUG) + configure_logging(SimpleNamespace(log_level="debug", logging=SimpleNamespace(enhance=SimpleNamespace(enabled=False, format="text")))) + + logging.getLogger("urllib3.connectionpool").debug( + '%s://%s:%s "%s %s %s" %s %s', + "https", + "cdn.example", + 443, + "GET", + "/private/BearerSecret?token=QuerySecret", + "HTTP/1.1", + 200, + None, + ) + + out = stream.getvalue() + assert "BearerSecret" not in out + assert "QuerySecret" not in out + assert 'https://cdn.example:443 "GET / HTTP/1.1"' in out + assert "200" in out # status observability preserved + finally: + root.handlers = old_handlers + root.setLevel(old_level) + + +def test_url_redaction_filter_covers_urllib3_retry_lines() -> None: + """urllib3's retry sites log the request target with no scheme and no + quoting, so neither the absolute-URL nor the quoted request-line pattern + can see it. Verified against the installed urllib3 (2.7.0): + ``Retry: %s`` (connectionpool.py:954, DEBUG), + ``Incremented Retry for (url='%s'): %r`` (util/retry.py:545, DEBUG — + origin-form target on the request path), and + ``Retrying (%r) after connection broken by '%r': %s`` + (connectionpool.py:869, WARNING — above the Gateway's INFO root). Each + collapses the target to ``/`` while keeping the surrounding + format (status counts, error text) for observability.""" + from deerflow.logging_config import UrlRedactionFilter + + filt = UrlRedactionFilter() + + def _record(name: str, fmt: str, args: tuple) -> logging.LogRecord: + return logging.LogRecord(name, logging.DEBUG, __file__, 1, fmt, args, None) + + # connectionpool.py:954 — bare origin-form target after "Retry: ". + retry_target = _record("urllib3.connectionpool", "Retry: %s", ("/private/BearerSecret?token=QuerySecret",)) + assert filt.filter(retry_target) is True + assert retry_target.getMessage() == "Retry: /" + + # An absolute target on the same line is the generic pass's job. + retry_absolute = _record("urllib3.connectionpool", "Retry: %s", ("https://cdn.example/private/BearerSecret?token=QuerySecret",)) + assert filt.filter(retry_absolute) is True + assert retry_absolute.getMessage() == "Retry: https://cdn.example/" + + # util/retry.py:545 — origin-form target inside the quoted url slot; the + # ``'): `` closer must survive the rewrite byte-for-byte. + increment = _record( + "urllib3.util.retry", + "Incremented Retry for (url='%s'): %r", + ("/private/BearerSecret?token=QuerySecret", "Retry(total=1, connect=2, read=None, redirect=None, status=None, other=None, allowed_methods=None)"), + ) + assert filt.filter(increment) is True + assert increment.getMessage() == "Incremented Retry for (url='/'): 'Retry(total=1, connect=2, read=None, redirect=None, status=None, other=None, allowed_methods=None)'" + + # The same line on the redirect path carries an ABSOLUTE target. The + # round-8 review repro: the generic pass's ``rest`` swallowed the + # ``'): `` closer and mangled the line — ``rest`` now stops at quotes, so + # the absolute URL is rewritten in place with the closer intact. + increment_absolute = _record( + "urllib3.util.retry", + "Incremented Retry for (url='%s'): %r", + ("https://cdn.example/private/BearerSecret?token=QuerySecret", "Retry(total=1, connect=2)"), + ) + assert filt.filter(increment_absolute) is True + assert increment_absolute.getMessage() == "Incremented Retry for (url='https://cdn.example/'): 'Retry(total=1, connect=2)'" + + # Userinfo in the absolute increment target is blanked like everywhere + # else, with the closing quote intact. + increment_userinfo = _record( + "urllib3.util.retry", + "Incremented Retry for (url='%s'): %r", + ("https://user:tok@internal-proxy.corp:8080/private/BearerSecret", "Retry(total=1)"), + ) + assert filt.filter(increment_userinfo) is True + assert increment_userinfo.getMessage() == "Incremented Retry for (url='https://@internal-proxy.corp:8080/'): 'Retry(total=1)'" + + # connectionpool.py:869 — WARNING level, so it passes an INFO root. The + # error repr keeps its own quotes; the greedy split still pins the target + # to the final ``': `` and the error text survives for observability. + # Args are real objects, matching how urlopen calls the site. + import urllib3 + + retries_obj = urllib3.Retry(total=2, redirect=0) + timeout_err = urllib3.exceptions.ReadTimeoutError(None, None, "Read timed out.") + retrying = _record( + "urllib3.connectionpool", + "Retrying (%r) after connection broken by '%r': %s", + (retries_obj, timeout_err, "/private/BearerSecret?token=QuerySecret"), + ) + assert filt.filter(retrying) is True + # Exact line equality: only the target changed; retry state and error + # text (observability) survive verbatim. + assert retrying.getMessage() == f"Retrying ({retries_obj!r}) after connection broken by '{timeout_err!r}': /" + assert "BearerSecret" not in retrying.getMessage() + assert "Read timed out" in retrying.getMessage() + + # Redirecting (poolmanager.py:500 INFO / connectionpool.py:922 DEBUG): + # either slot may be an origin-form target — connectionpool passes the + # origin-form request target, and a relative Location header has no + # scheme. Origin-form slots collapse; absolute slots are the generic + # pass's job. + redirect_origin = logging.LogRecord("urllib3.connectionpool", logging.DEBUG, __file__, 1, "Redirecting %s -> %s", ("/private/BearerSecret?token=QuerySecret", "/other/BearerSecret?sig=OtherSecret"), None) + assert filt.filter(redirect_origin) is True + assert redirect_origin.getMessage() == "Redirecting / -> /" + + redirect_mixed = logging.LogRecord("urllib3.poolmanager", logging.INFO, __file__, 1, "Redirecting %s -> %s", ("/private/BearerSecret?token=QuerySecret", "https://mirror.example/other?sig=OtherSecret"), None) + assert filt.filter(redirect_mixed) is True + assert redirect_mixed.getMessage() == "Redirecting / -> https://mirror.example/" + + # A raw Location field from a misbehaving server can contain interior + # whitespace. It must not disable the source-target redaction. + redirect_spaced_location = logging.LogRecord( + "urllib3.connectionpool", + logging.DEBUG, + __file__, + 1, + "Redirecting %s -> %s", + ("/private/BearerSecret?token=QuerySecret", "/bad location"), + None, + ) + assert filt.filter(redirect_spaced_location) is True + assert redirect_spaced_location.getMessage() == "Redirecting / -> /" + + # Lowercase custom methods ride the same request-line shape (methods are + # case-sensitive tokens; callers may pass any case). + lowercase = logging.LogRecord( + "urllib3.connectionpool", + logging.DEBUG, + __file__, + 1, + '%s://%s:%s "%s %s %s" %s %s', + ("https", "cdn.example", 443, "patch", "/private/BearerSecret?token=QuerySecret", "HTTP/1.1", 200, None), + None, + ) + assert filt.filter(lowercase) is True + assert lowercase.getMessage() == 'https://cdn.example:443 "patch / HTTP/1.1" 200 None' + + # Adjacent non-URL lines must pass through untouched: the shapes are + # anchored to the exact urllib3 formats, not to the word "Retry". + plain_retry = _record("some.other.lib", "Retry: attempt scheduled soon", ()) + assert filt.filter(plain_retry) is True + assert plain_retry.getMessage() == "Retry: attempt scheduled soon" + plain_conn = _record("urllib3.connectionpool", "Starting new HTTP connection (%d): %s:%s", (1, "cdn.example", 443)) + assert filt.filter(plain_conn) is True + assert plain_conn.getMessage() == "Starting new HTTP connection (1): cdn.example:443" + + +def test_url_redaction_filter_covers_urllib3_retry_lines_through_real_emit() -> None: + """Real-emitter wiring for the retry lines: the increment line is produced + by actually calling ``Retry.increment`` (retry.py:545 logs through the + ``urllib3.util.retry`` child logger at DEBUG), and the connectionpool + shapes are emitted through the real child logger with the exact installed + format strings. Only the handler-level filters installed by + configure_logging can rewrite propagated records.""" + import urllib3 + from urllib3.util.retry import Retry + + from deerflow.logging_config import configure_logging + + root = logging.getLogger() + old_handlers = root.handlers[:] + old_level = root.level + stream = io.StringIO() + handler = logging.StreamHandler(stream) + + try: + root.handlers = [handler] + root.setLevel(logging.DEBUG) + configure_logging(SimpleNamespace(log_level="debug", logging=SimpleNamespace(enhance=SimpleNamespace(enabled=False, format="text")))) + + # retry.py:545 via the library's own code path (no-args increment + # takes the generic-response branch and logs without raising). + Retry(total=2).increment(method="GET", url="/private/BearerSecret?token=QuerySecret") + + logging.getLogger("urllib3.connectionpool").debug("Retry: %s", "/private/BearerSecret?token=QuerySecret") + logging.getLogger("urllib3.connectionpool").warning( + "Retrying (%r) after connection broken by '%r': %s", + Retry(total=2, redirect=0), + urllib3.exceptions.ReadTimeoutError(None, None, "Read timed out."), + "/private/BearerSecret?token=QuerySecret", + ) + logging.getLogger("urllib3.connectionpool").debug( + "Redirecting %s -> %s", + "/private/BearerSecret?token=QuerySecret", + "https://mirror.example/other/BearerSecret?sig=OtherSecret", + ) + + out = stream.getvalue() + assert "BearerSecret" not in out + assert "QuerySecret" not in out + assert "OtherSecret" not in out + # Redacted, not suppressed: every line still renders with its shape + # and the parts that carry no URL (retry state, error text, hosts). + assert "Incremented Retry for (url='/')" in out + assert "Retry: /" in out + assert "after connection broken by" in out and "Read timed out" in out and "': /" in out + assert "Redirecting / -> https://mirror.example/" in out + finally: + root.handlers = old_handlers + root.setLevel(old_level) + + +def test_configure_logging_installs_url_redaction_on_httpx_logger_and_root_handlers() -> None: + from deerflow.logging_config import UrlRedactionFilter, _has_url_redaction_filter, configure_logging, install_url_log_redaction + + httpx_logger = logging.getLogger("httpx") + root = logging.getLogger() + old_filters = httpx_logger.filters[:] + old_handlers = root.handlers[:] + handler = logging.StreamHandler(io.StringIO()) + + try: + root.handlers = [handler] + httpx_logger.filters = [f for f in old_filters if not isinstance(f, UrlRedactionFilter)] + install_url_log_redaction() + install_url_log_redaction() # idempotent + assert sum(isinstance(f, UrlRedactionFilter) for f in httpx_logger.filters) == 1 + assert all(_has_url_redaction_filter(h) for h in root.handlers) + + # Handlers added later are covered by the configure_logging loop, not + # by the one-shot installer. + late = logging.StreamHandler(io.StringIO()) + root.handlers.append(late) + configure_logging(SimpleNamespace(log_level="info", logging=SimpleNamespace(enhance=SimpleNamespace(enabled=False, format="text")))) + assert _has_url_redaction_filter(late) + assert _has_url_redaction_filter(root.handlers[0]) + finally: + httpx_logger.filters = old_filters + root.handlers = old_handlers + + +def test_url_redaction_filter_long_input_stays_linear_time() -> None: + """Long-input regression (round-9 review finding): both scheme-bearing + patterns start with a character class, so re.sub-style scanning retries + every suffix of a long token — the reviewer measured ~1.79 s for a 64K + path and ~3.10 s for a URL-free 64K error body, per filter call, and the + filter runs synchronously in every root handler. The scheme passes are + driven from "://" occurrences instead, so the same inputs cost + milliseconds. The bound is generous (the quadratic path at 256K would + take tens of seconds) to stay robust on slow CI runners, while still + going red against any regression to per-position rescanning.""" + import time + + from deerflow.logging_config import UrlRedactionFilter + + filt = UrlRedactionFilter() + + # Ordinary HTTPX URL whose path is a 256K letter run. httpx.URL rejects + # URLs this long, so the record is built directly with the URL as a + # plain string arg — the rendered message shape is identical. + record = logging.LogRecord( + "httpx", + logging.INFO, + __file__, + 1, + _HTTPX_REQUEST_FORMAT, + ("GET", "https://cdn.weixin.qq.com/private/" + "A" * 262144, "HTTP/1.1", 200, "OK"), + None, + ) + started = time.perf_counter() + assert filt.filter(record) is True + elapsed_url = time.perf_counter() - started + assert elapsed_url < 5.0, f"URL-bearing record took {elapsed_url:.2f}s" + formatted = record.getMessage() + assert "cdn.weixin.qq.com/" in formatted # still redacted, and + assert "A" * 64 not in formatted # the long path itself did not survive + + # A URL-free 64K letter error body through the real wiring: the filter + # runs in the root handler, and the message must pass through verbatim. + root = logging.getLogger() + old_handlers = root.handlers[:] + old_level = root.level + stream = io.StringIO() + handler = logging.StreamHandler(stream) + try: + root.handlers = [handler] + root.setLevel(logging.INFO) + configure_logging(SimpleNamespace(log_level="info", logging=SimpleNamespace(enhance=SimpleNamespace(enabled=False, format="text")))) + body = "E" * 65536 + started = time.perf_counter() + logging.getLogger("some.error.reporter").error("payload too large: %s", body) + elapsed_plain = time.perf_counter() - started + assert elapsed_plain < 5.0, f"URL-free record took {elapsed_plain:.2f}s" + assert stream.getvalue().endswith("payload too large: " + body + "\n") + finally: + root.handlers = old_handlers + root.setLevel(old_level) + + +def test_url_redaction_filter_nested_scheme_in_path_keeps_both_passes() -> None: + """The scheme-bearing passes run per "://" start, so a scheme-shaped + target NESTED inside another URL's path still gets its own pass attempt: + the outer absolute URL is rewritten first (its rest swallows the inner + scheme text), and the inner request-line shape — if the path is followed + by urllib3 quoting — is rewritten by the request-line pass that ran + before it. Pins the leftmost-non-overlapping equivalence with the old + two-pass re.sub behavior on overlapping candidates.""" + from deerflow.logging_config import UrlRedactionFilter + + filt = UrlRedactionFilter() + record = logging.LogRecord( + "httpx", + logging.INFO, + __file__, + 1, + "fetch failed for %s and %s", + ("https://gateway.example/redirect?to=https://evil.example/sink", 'https://evil.example "GET /private/BearerSecret?token=QuerySecret HTTP/1.1" 200 None'), + None, + ) + assert filt.filter(record) is True + formatted = record.getMessage() + assert "BearerSecret" not in formatted + assert "token=" not in formatted + assert "gateway.example/" in formatted + # The nested request line kept its quoted target redacted too. + assert 'https://evil.example "GET / HTTP/1.1"' in formatted + + +def test_url_redaction_filter_scheme_start_skips_non_letter_run_head() -> None: + """The linear scan walks back over the full scheme charset (letters, + digits, +, -, .) but a regex match can only start at the run's first + LETTER — digits are valid scheme tail characters, never the head. A + digit glued in front of a URL shifts the match start past it, exactly + like re.sub's leftmost scan; a run with no letter at all ("123://x") + cannot start any match and passes through with nothing rewritten.""" + from deerflow.logging_config import UrlRedactionFilter + + filt = UrlRedactionFilter() + glued = logging.LogRecord("httpx", logging.INFO, __file__, 1, "fetch %s", ("9https://host.example/private/x?token=QuerySecret",), None) + assert filt.filter(glued) is True + assert glued.getMessage() == "fetch 9https://host.example/" + + digits_only = logging.LogRecord("httpx", logging.INFO, __file__, 1, "fetch %s", ("123://host.example/private/x?token=QuerySecret",), None) + assert filt.filter(digits_only) is True + # "123" is not a scheme head, so no scheme starts at this "://" — the + # text is left as-is by the scheme passes (no URL rewrite, no signal + # loss; the record itself is not a valid URL shape). + assert digits_only.getMessage() == "fetch 123://host.example/private/x?token=QuerySecret" + + +def test_url_redaction_filter_embedded_quotes_stay_inside_rest() -> None: + """Boundary rule for the rest quote-stop (round-10 residual): a quote is + a closing mark only when whitespace, ``)``, or end of string follows — + the shapes that actually close a quoted URL (urllib3's + ``Incremented Retry for (url='…')`` scaffolding, surrounding prose). A + quote EMBEDDED in the URL itself (``/path'quoted'?token=…``) must be + consumed so the whole path+query stays redacted; the earlier + quote-stop-at-any-quote behavior kept the suffix after the quote + verbatim.""" + from deerflow.logging_config import UrlRedactionFilter + + filt = UrlRedactionFilter() + + # The review repro, through the real httpx record shape: httpx.URL keeps + # an apostrophe raw, so the embedded quote is present verbatim in the + # rendered message. rest must consume it — the credential-bearing suffix + # does not survive. (The httpx format quotes "version status reason" + # together.) + embedded = _httpx_record("https://host.example/path'quoted'?token=QuerySecret") + assert filt.filter(embedded) is True + assert embedded.getMessage() == 'HTTP Request: GET https://host.example/ "HTTP/1.1 200 OK"' + assert "quoted" not in embedded.getMessage() + assert "token=" not in embedded.getMessage() + + # A raw embedded DOUBLE quote (httpx.URL would percent-encode %22, so + # this rides a plain string arg — MCP/extension riders may log + # pre-rendered URLs). + embedded_double = logging.LogRecord( + "httpx", + logging.INFO, + __file__, + 1, + _HTTPX_REQUEST_FORMAT, + ("GET", 'https://host.example/pa"th?token=QuerySecret', "HTTP/1.1", 200, "OK"), + None, + ) + assert filt.filter(embedded_double) is True + assert embedded_double.getMessage() == 'HTTP Request: GET https://host.example/ "HTTP/1.1 200 OK"' + assert "token=" not in embedded_double.getMessage() + + # Boundary quotes still close rest: prose quoting keeps its punctuation. + prose_double = logging.LogRecord("some.lib", logging.INFO, __file__, 1, "see %s in the docs", ('"https://host.example/private/x?tok=1"',), None) + assert filt.filter(prose_double) is True + assert prose_double.getMessage() == 'see "https://host.example/" in the docs' + prose_single = logging.LogRecord("some.lib", logging.INFO, __file__, 1, "see %s please", ("'https://host.example/private/x?tok=1'",), None) + assert filt.filter(prose_single) is True + assert prose_single.getMessage() == "see 'https://host.example/' please" + + # The increment line's url capture applies the same rule with its fixed + # ``')`` closer: an embedded quote inside the target no longer truncates + # the capture, so the whole origin-form target collapses. + increment = logging.LogRecord( + "urllib3.util.retry", + logging.DEBUG, + __file__, + 1, + "Incremented Retry for (url='%s'): %r", + ("/a'b?tok=QuerySecret", "Retry(total=1)"), + None, + ) + assert filt.filter(increment) is True + assert increment.getMessage() == "Incremented Retry for (url='/'): 'Retry(total=1)'" + assert "tok=" not in increment.getMessage() + + +def test_url_redaction_filter_leaves_arrow_paths_in_other_logs_alone() -> None: + """The Redirecting origin pass is anchored to the WHOLE ``Redirecting + -> `` message because an ``-> /path`` arrow is not urllib3-owned + shape: the sandbox provider's actionable mount error renders + ``sandbox.mounts entry -> `` and a substring + match rewrote the container path to ``/``, breaking the error's + instructions (backend-unit-tests shard 3 on CI, round 11).""" + from deerflow.logging_config import UrlRedactionFilter + + filt = UrlRedactionFilter() + sandbox_error = ( + "sandbox.mounts entry /srv/deer-flow/knowledge -> /mnt/knowledge ignored: host_path " + "/srv/deer-flow/knowledge does not exist from the perspective of the gateway process. " + "In Docker deployments (make up / docker-compose), this path must also be bind-mounted " + "into the gateway container — add a matching volume entry under services.gateway.volumes " + "in docker/docker-compose.yaml (and use the in-container path here), or run in local mode " + "(make dev) where the gateway sees the host filesystem directly." + ) + record = logging.LogRecord("deerflow.sandbox.local.local_sandbox_provider", logging.ERROR, "provider.py", 1, "%s", (sandbox_error,), None) + assert filt.filter(record) is True + assert record.getMessage() == sandbox_error # byte-for-byte passthrough + assert "/mnt/knowledge" in record.getMessage() + assert "" not in record.getMessage() + + +def test_url_redaction_filter_redirecting_survives_spacey_location() -> None: + """Round-13 P3: the Redirecting anchor keeps the ``^Redirecting `` prefix + (the urllib3-owned literal that stops the sandbox false positive) but the + tail must be loose — ``redirect_location`` is the raw Location header + string, and interior spaces are legal field syntax a misbehaving server + can emit. A whitespace-strict tail voided the pass entirely and leaked + the origin-form request target in the first slot; a space-carrying + second slot now collapses whole.""" + from deerflow.logging_config import UrlRedactionFilter + + filt = UrlRedactionFilter() + + # The reviewer's repro: both credentials must go, shape kept. + spacey = logging.LogRecord( + "urllib3.connectionpool", + logging.DEBUG, + __file__, + 1, + "Redirecting %s -> %s", + ("/private/BearerSecret?token=QuerySecret", "/bad location"), + None, + ) + assert filt.filter(spacey) is True + assert spacey.getMessage() == "Redirecting / -> /" + assert "BearerSecret" not in spacey.getMessage() + + # The FIRST slot gets the same grammar treatment: the recursive urlopen + # frame passes the previous raw Location as its url, so t1 can carry + # interior spaces too. + spacey_t1 = logging.LogRecord( + "urllib3.connectionpool", + logging.DEBUG, + __file__, + 1, + "Redirecting %s -> %s", + ("/bad target?token=QuerySecret", "/private/x"), + None, + ) + assert filt.filter(spacey_t1) is True + assert spacey_t1.getMessage() == "Redirecting / -> /" + assert "QuerySecret" not in spacey_t1.getMessage() + + # An absolute Location with an interior space stays whole for the + # generic absolute-URL pass (which stops its rest at whitespace). + spacey_absolute = logging.LogRecord( + "urllib3.connectionpool", + logging.DEBUG, + __file__, + 1, + "Redirecting %s -> %s", + ("/private/BearerSecret?token=QuerySecret", "https://mirror.example/other page?sig=OtherSecret"), + None, + ) + assert filt.filter(spacey_absolute) is True + assert spacey_absolute.getMessage() == "Redirecting / -> https://mirror.example/ page?sig=OtherSecret" + assert "BearerSecret" not in spacey_absolute.getMessage() + + # The sandbox arrow false positive stays excluded: the prefix anchor, + # not a strict tail, is what keeps non-Redirecting messages untouched. + sandbox = logging.LogRecord("deerflow.sandbox.local.local_sandbox_provider", logging.ERROR, "p.py", 1, "sandbox.mounts entry /srv/knowledge -> /mnt/knowledge ignored: missing", (), None) + assert filt.filter(sandbox) is True + assert sandbox.getMessage() == "sandbox.mounts entry /srv/knowledge -> /mnt/knowledge ignored: missing" diff --git a/backend/tests/test_wechat_channel.py b/backend/tests/test_wechat_channel.py index 451221dd4..e63dd736b 100644 --- a/backend/tests/test_wechat_channel.py +++ b/backend/tests/test_wechat_channel.py @@ -169,7 +169,7 @@ def test_handle_update_downloads_inbound_image(monkeypatch, tmp_path: Path): channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) encrypted = channel.__class__.__dict__["_extract_image_file"].__globals__["_encrypt_aes_128_ecb"](plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -185,7 +185,7 @@ def test_handle_update_downloads_inbound_image(monkeypatch, tmp_path: Path): "type": 2, "image_item": { "aeskey": aes_key.hex(), - "media": {"full_url": "https://cdn.example/image.bin"}, + "media": {"full_url": "https://cdn.weixin.qq.com/image.bin"}, }, } ], @@ -224,7 +224,7 @@ def test_handle_update_downloads_inbound_png_with_png_extension(monkeypatch, tmp channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) encrypted = channel.__class__.__dict__["_extract_image_file"].__globals__["_encrypt_aes_128_ecb"](plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -240,7 +240,7 @@ def test_handle_update_downloads_inbound_png_with_png_extension(monkeypatch, tmp "type": 2, "image_item": { "aeskey": aes_key.hex(), - "media": {"full_url": "https://cdn.example/image.bin"}, + "media": {"full_url": "https://cdn.weixin.qq.com/image.bin"}, }, } ], @@ -272,7 +272,7 @@ def test_handle_update_preserves_text_and_ref_msg_with_image(monkeypatch, tmp_pa channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) encrypted = channel.__class__.__dict__["_extract_image_file"].__globals__["_encrypt_aes_128_ecb"](plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -290,7 +290,7 @@ def test_handle_update_preserves_text_and_ref_msg_with_image(monkeypatch, tmp_pa "ref_msg": {"title": "quoted", "message_item": {"type": 1}}, "image_item": { "aeskey": aes_key.hex(), - "media": {"full_url": "https://cdn.example/image2.bin"}, + "media": {"full_url": "https://cdn.weixin.qq.com/image2.bin"}, }, }, ], @@ -985,7 +985,7 @@ def test_handle_update_downloads_inbound_file(monkeypatch, tmp_path: Path): channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) encrypted = channel.__class__.__dict__["_extract_file_item"].__globals__["_encrypt_aes_128_ecb"](plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -1002,7 +1002,7 @@ def test_handle_update_downloads_inbound_file(monkeypatch, tmp_path: Path): "file_item": { "file_name": "report.pdf", "aeskey": aes_key.hex(), - "media": {"full_url": "https://cdn.example/report.bin"}, + "media": {"full_url": "https://cdn.weixin.qq.com/report.bin"}, }, } ], @@ -1040,7 +1040,7 @@ def test_handle_update_downloads_inbound_file_with_media_aeskey_hex(monkeypatch, channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) encrypted = channel.__class__.__dict__["_extract_file_item"].__globals__["_encrypt_aes_128_ecb"](plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -1057,7 +1057,7 @@ def test_handle_update_downloads_inbound_file_with_media_aeskey_hex(monkeypatch, "file_item": { "file_name": "report.pdf", "media": { - "full_url": "https://cdn.example/report.bin", + "full_url": "https://cdn.weixin.qq.com/report.bin", "aeskey": aes_key.hex(), }, }, @@ -1091,7 +1091,7 @@ def test_handle_update_downloads_inbound_file_with_unpadded_item_aes_key(monkeyp channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) encrypted = channel.__class__.__dict__["_extract_file_item"].__globals__["_encrypt_aes_128_ecb"](plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -1108,7 +1108,7 @@ def test_handle_update_downloads_inbound_file_with_unpadded_item_aes_key(monkeyp "aesKey": encoded_key, "file_item": { "file_name": "report.pdf", - "media": {"full_url": "https://cdn.example/report.bin"}, + "media": {"full_url": "https://cdn.weixin.qq.com/report.bin"}, }, } ], @@ -1140,7 +1140,7 @@ def test_handle_update_downloads_inbound_file_with_media_aes_key_base64_of_hex(m channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) encrypted = channel.__class__.__dict__["_extract_file_item"].__globals__["_encrypt_aes_128_ecb"](plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -1157,7 +1157,7 @@ def test_handle_update_downloads_inbound_file_with_media_aes_key_base64_of_hex(m "file_item": { "file_name": "report.pdf", "media": { - "full_url": "https://cdn.example/report.bin", + "full_url": "https://cdn.weixin.qq.com/report.bin", "aes_key": encoded_hex_key, }, }, @@ -1190,7 +1190,7 @@ def test_handle_update_skips_disallowed_inbound_file(monkeypatch, tmp_path: Path channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) encrypted = channel.__class__.__dict__["_extract_file_item"].__globals__["_encrypt_aes_128_ecb"](plaintext, aes_key) - async def _fake_download(_url: str, *, timeout: float | None = None): + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None): return encrypted channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -1207,7 +1207,7 @@ def test_handle_update_skips_disallowed_inbound_file(monkeypatch, tmp_path: Path "file_item": { "file_name": "malware.exe", "aeskey": aes_key.hex(), - "media": {"full_url": "https://cdn.example/bad.bin"}, + "media": {"full_url": "https://cdn.weixin.qq.com/bad.bin"}, }, } ], @@ -1322,7 +1322,7 @@ def test_poll_loop_one_bad_message_does_not_permanently_lose_its_siblings(monkey "type": 2, "image_item": { "aeskey": aes_key.hex(), - "media": {"full_url": "https://cdn.example/corrupt-attachment.bin"}, + "media": {"full_url": "https://cdn.weixin.qq.com/corrupt-attachment.bin"}, }, } ], @@ -1356,7 +1356,7 @@ def test_poll_loop_one_bad_message_does_not_permanently_lose_its_siblings(monkey config={"bot_token": "test-token", "state_dir": str(state_dir), "polling_retry_delay": 0.001}, ) - async def _fake_download(_url: str, *, timeout: float | None = None) -> bytes: + async def _fake_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None) -> bytes: return non_block_aligned_ciphertext channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] @@ -1553,3 +1553,324 @@ def test_save_auth_state_chmod_failure_is_logged_not_warned(tmp_path: Path, capl messages = [record.getMessage() for record in caplog.records] assert any("unable to chmod auth state" in message for message in messages) assert not any("failed to persist auth state" in message for message in messages) + + +# --------------------------------------------------------------------------- +# Inbound media download cap + destination allowlist +# --------------------------------------------------------------------------- + + +def test_is_allowed_media_url_suffix_boundaries(): + from app.channels.wechat import WechatChannel + + channel = WechatChannel(MessageBus(), config={"bot_token": "test-token"}) + + # Platform CDN defaults (plus the configured cdn_base_url host) are allowed. + assert channel._is_allowed_media_url("https://novac2c.cdn.weixin.qq.com/c2c/x?token=1") + assert channel._is_allowed_media_url("https://cdn.weixin.qq.com/image.bin") + # Dot-boundary suffix matching: lookalike hosts never match. + assert not channel._is_allowed_media_url("https://notqq.com/image.bin") + assert not channel._is_allowed_media_url("https://qq.com.evil.io/image.bin") + assert not channel._is_allowed_media_url("https://cdn.example/image.bin") + # Loopback/private targets and non-HTTP schemes are rejected outright. + assert not channel._is_allowed_media_url("http://127.0.0.1:8001/api/user") + assert not channel._is_allowed_media_url("http://169.254.169.254/latest/meta-data") + assert not channel._is_allowed_media_url("file:///etc/passwd") + + # Operator suffixes extend the allowlist; the configured cdn_base_url host + # is admitted automatically so a custom CDN endpoint keeps working. A + # leading ``*.`` (DNS habit) normalizes to the bare suffix. + custom = WechatChannel( + MessageBus(), + config={ + "bot_token": "test-token", + "cdn_base_url": "https://media.internal.example/c2c", + "allowed_media_hosts": ["cdn.example", "*.wild.example"], + }, + ) + assert custom._is_allowed_media_url("https://media.internal.example/c2c/x") + assert custom._is_allowed_media_url("https://a.cdn.example/image.bin") + assert custom._is_allowed_media_url("https://cdn.weixin.qq.com/image.bin") + assert custom._is_allowed_media_url("https://b.wild.example/image.bin") + + +def test_handle_update_skips_media_from_disallowed_host(tmp_path: Path): + from app.channels.wechat import WechatChannel + + async def go(): + bus = MessageBus() + published = [] + + async def capture(msg): + published.append(msg) + + bus.publish_inbound = capture # type: ignore[method-assign] + + channel = WechatChannel(bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) + + async def _forbidden_download(_url: str, *, timeout: float | None = None, max_bytes: int | None = None) -> bytes: + raise AssertionError("download must not be attempted for a disallowed media host") + + channel._download_cdn_bytes = _forbidden_download # type: ignore[method-assign] + + await channel._handle_update( + { + "message_type": 1, + "message_id": 201, + "from_user_id": "wx-user-1", + "context_token": "ctx-evil-1", + "item_list": [ + { + "type": 2, + "image_item": { + "aeskey": b"1234567890abcdef".hex(), + "media": {"full_url": "https://evil.example/image.bin"}, + }, + } + ], + } + ) + + # The image is dropped and, with no text either, nothing is published. + assert published == [] + downloads_dir = tmp_path / "downloads" + assert not downloads_dir.exists() or not list(downloads_dir.iterdir()) + + _run(go()) + + +def test_handle_update_http_download_failure_is_sanitized_not_raised(tmp_path: Path, caplog): + """An HTTP failure during the media download drops that attachment with a + sanitized log instead of escaping to the polling loop's logger.exception. + + httpx.HTTPStatusError formats the signed URL (path + query credentials) + into its message, so letting it propagate would render the URL in the + per-message traceback. Reproduced with a real 403 mock transport. + """ + import httpx + + from app.channels.wechat import WechatChannel + + async def go(): + bus = MessageBus() + published = [] + + async def capture(msg): + published.append(msg) + + bus.publish_inbound = capture # type: ignore[method-assign] + + channel = WechatChannel(bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) + channel._client = httpx.AsyncClient( # type: ignore[assignment] + transport=httpx.MockTransport(lambda _request: httpx.Response(403)) + ) + try: + await channel._handle_update( + { + "message_type": 1, + "message_id": 202, + "from_user_id": "wx-user-1", + "context_token": "ctx-403-1", + "item_list": [ + { + "type": 2, + "image_item": { + "aeskey": b"1234567890abcdef".hex(), + "media": {"full_url": "https://cdn.weixin.qq.com/private/BearerSecret?token=QuerySecret"}, + }, + } + ], + } + ) + finally: + await channel._client.aclose() + + # The image is dropped and, with no text either, nothing is published. + assert published == [] + + with caplog.at_level(logging.WARNING, logger="app.channels.wechat"): + _run(go()) + + assert "BearerSecret" not in caplog.text + assert "QuerySecret" not in caplog.text + assert "/private/" not in caplog.text + # The operator still sees what failed and for which host. + assert "HTTPStatusError (403)" in caplog.text + assert "cdn.weixin.qq.com" in caplog.text + + +class _FakeStreamResponse: + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + self.headers: dict[str, str] = {} + + def raise_for_status(self) -> None: + return None + + # Deliberately no aiter_bytes: the reader must consume undecoded bytes, so + # an accidental switch back to the decoding iterator fails loudly here. + async def aiter_raw(self): + for chunk in self._chunks: + yield chunk + + +class _FakeStreamContext: + def __init__(self, response: _FakeStreamResponse): + self._response = response + + async def __aenter__(self) -> _FakeStreamResponse: + return self._response + + async def __aexit__(self, *exc_info) -> bool: + return False + + +class _FakeStreamingClient: + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + + def stream(self, _method: str, _url: str, timeout: float | None = None, **_kwargs) -> _FakeStreamContext: + return _FakeStreamContext(_FakeStreamResponse(self._chunks)) + + +def test_download_cdn_bytes_aborts_when_stream_exceeds_cap(): + from app.channels.wechat import WechatChannel + + async def go(): + channel = WechatChannel(MessageBus(), config={"bot_token": "test-token"}) + channel._client = _FakeStreamingClient([b"abc", b"def"]) # type: ignore[assignment] + + # 6 bytes total vs a 5-byte cap: aborted mid-stream before full read. + assert await channel._download_cdn_bytes("https://cdn.weixin.qq.com/x", max_bytes=5) is None + # At/under the cap and with the cap disabled the chunks are joined as before. + assert await channel._download_cdn_bytes("https://cdn.weixin.qq.com/x", max_bytes=6) == b"abcdef" + assert await channel._download_cdn_bytes("https://cdn.weixin.qq.com/x", max_bytes=None) == b"abcdef" + assert await channel._download_cdn_bytes("https://cdn.weixin.qq.com/x", max_bytes=0) == b"abcdef" + + _run(go()) + + +def test_download_cdn_bytes_rejects_compressed_response_before_decode(caplog): + """aiter_bytes() would transparently decode Content-Encoding, allocating the + whole decompressed body before the cap sees a byte — an ~8 KB gzip wire + chunk decoding to 8 MiB (reproduced here) bypasses max_bytes entirely. The + download must request identity and refuse a residual encoding before + reading, even with the cap disabled.""" + import gzip + + import httpx + + from app.channels.wechat import WechatChannel + + class _AsyncChunks(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + + async def __aiter__(self): + for chunk in self._chunks: + yield chunk + + seen_requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(request) + return httpx.Response( + 200, + headers={"Content-Encoding": "gzip"}, + stream=_AsyncChunks([gzip.compress(b"\x00" * (8 * 1024 * 1024))]), + ) + + async def go(): + channel = WechatChannel(MessageBus(), config={"bot_token": "test-token"}) + channel._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) # type: ignore[assignment] + try: + # Cap disabled: the old aiter_bytes() code path would happily + # return the 8 MiB decompressed payload here. + assert await channel._download_cdn_bytes("https://cdn.weixin.qq.com/x", max_bytes=None) is None + # With a cap in place the rejection still happens for the encoding, + # before any decode/size accounting. + assert await channel._download_cdn_bytes("https://cdn.weixin.qq.com/x", max_bytes=1024 * 1024) is None + finally: + await channel._client.aclose() + + with caplog.at_level(logging.WARNING, logger="app.channels.wechat"): + _run(go()) + + assert "Content-Encoding" in caplog.text + assert "exceeds" not in caplog.text # rejected for the encoding, not the size + assert seen_requests[0].headers.get("accept-encoding") == "identity" + + +def test_download_cdn_bytes_streams_identity_response(): + """An unencoded response still round-trips through a real httpx transport.""" + import httpx + + from app.channels.wechat import WechatChannel + + class _AsyncChunks(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]): + self._chunks = chunks + + async def __aiter__(self): + for chunk in self._chunks: + yield chunk + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, stream=_AsyncChunks([b"abcdef"])) + + async def go(): + channel = WechatChannel(MessageBus(), config={"bot_token": "test-token"}) + channel._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) # type: ignore[assignment] + try: + assert await channel._download_cdn_bytes("https://cdn.weixin.qq.com/x") == b"abcdef" + finally: + await channel._client.aclose() + + _run(go()) + + +def test_boundary_sized_inbound_image_survives_pkcs7_padding(tmp_path: Path): + """A valid attachment whose plaintext is exactly the configured limit must load. + + The limit bounds plaintext, but the stream measures ciphertext — AES-128-ECB + with PKCS#7 pads 32 plaintext bytes to 48 — so the stream cap must be the + padded size of exactly-limit plaintext, not the plaintext limit itself. + Exercises the real streaming download (no _download_cdn_bytes stub). + """ + from app.channels.wechat import WechatChannel, _encrypt_aes_128_ecb + + async def go(): + channel = WechatChannel( + MessageBus(), + config={"bot_token": "test-token", "state_dir": str(tmp_path), "max_inbound_image_bytes": 32}, + ) + aes_key = b"1234567890abcdef" + plaintext = b"\x89PNG\r\n\x1a\n" + b"x" * 24 # exactly 32 bytes, valid PNG magic + encrypted = _encrypt_aes_128_ecb(plaintext, aes_key) # 48 bytes > 32-byte limit + assert len(encrypted) == 48 + channel._client = _FakeStreamingClient([encrypted]) # type: ignore[assignment] + + assert channel._stream_cap_for(32) == 48 + assert channel._stream_cap_for(0) is None + + files = await channel._extract_inbound_files( + { + "message_id": 301, + "item_list": [ + { + "type": 2, + "image_item": { + "aeskey": aes_key.hex(), + "media": {"full_url": "https://cdn.weixin.qq.com/image.bin"}, + }, + } + ], + } + ) + + assert len(files) == 1 + assert files[0]["type"] == "image" + assert files[0]["size"] == 32 + assert Path(files[0]["path"]).read_bytes() == plaintext + + _run(go()) diff --git a/config.example.yaml b/config.example.yaml index f36c3f3c3..140935510 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -2636,6 +2636,10 @@ run_ownership: # # Optional: sent as SKRouteTag header when provided # route_tag: "" # allowed_users: [] # empty = allow all +# # Optional: extra host suffixes inbound media downloads may come from, +# # in addition to the platform CDN defaults (*.qq.com and the cdn_base_url host). +# # Entries are suffixes: "example.com" and "*.example.com" are equivalent. +# allowed_media_hosts: [] # # Optional: timing values must be positive finite seconds # polling_timeout: 35 # polling_retry_delay: 5 @@ -2675,6 +2679,11 @@ run_ownership: # enabled: false # bot_id: $WECOM_BOT_ID # bot_secret: $WECOM_BOT_SECRET +# # Optional: extra host suffixes inbound media downloads may come from, in +# # addition to the built-in qq.com family and WeCom's official COS media +# # host (ww-aibot-img-1258476243..myqcloud.com). Add a suffix here +# # if WeCom rotates to a new COS account or media is routed through a proxy. +# allowed_media_hosts: [] # # dingtalk: # enabled: false