fix(channels): stream-cap and validate WeChat/WeCom inbound media downloads, fixes #5223 (#5225)

* 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 <full URL>' 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 /<redacted>, 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 <url> -> <url>'
  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 /<redacted>, 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: <target>`
(connectionpool.py:954 DEBUG), `Incremented Retry for (url='<target>')`
(util/retry.py:545 DEBUG, absolute on the redirect path), `Retrying (...)
after connection broken by '<err>': <target>` (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 /<redacted>; 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 <host> -> /mnt/knowledge ignored: ...`) had its
container path rewritten to /<redacted>, 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 <t> -> <t>` 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 <willem.jiang@gmail.com>
This commit is contained in:
hataa 2026-09-16 16:33:47 +08:00 committed by GitHub
parent 6bab87aca4
commit 26800d1245
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
11 changed files with 2308 additions and 40 deletions

View File

@ -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.<region>.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:

View File

@ -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-<APPID>.cos.<region>.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 <full URL>` 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 <target> -> <target>` 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: <target>` DEBUG and `Retrying (…) after connection broken by '<error>': <target>` **WARNING** — above the INFO root, so reachable at the default log level — via `urllib3.connectionpool`, and `Incremented Retry for (url='<target>')` 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<token>/...`, 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=<id>; ` (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

View File

@ -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-<APPID>.cos.<region>.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<appid>\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 "<unnamed>"
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

View File

@ -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)

View File

@ -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:

View File

@ -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 <full URL> HTTP/x.x <status> <duration>`` at
# INFO before any response handling runs, and urllib3 logs
# ``Redirecting <url> -> <url>`` 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<scheme>[a-zA-Z][a-zA-Z0-9+.-]*://)(?P<userinfo>[^/?#\s@]*@)?(?P<host>[^/?#\s]+)(?P<rest>[/?#](?:[^\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 ``/<redacted>``. 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<scheme>[a-zA-Z][a-zA-Z0-9+.-]*://)(?P<userinfo>[^/?#\s"@]*@)?(?P<host>[^/?#\s"]+) "(?P<method>[A-Za-z]+) (?P<target>/[^"\s]*) (?P<version>HTTP/[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 ``/<redacted>``.
# - ``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<target>/\S+)$")
_URLLIB3_INCREMENT_RETRY_RE = re.compile(r"Incremented Retry for \(url='(?P<url>(?:[^']|'(?!\)))*)'\)")
_URLLIB3_RETRYING_RE = re.compile(r"^(?P<head>Retrying \(.*\) after connection broken by .*'): (?P<target>/\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 ``/<redacted>`` 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 <host_path> -> <container_path>``), 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<t1>\S.*?) -> (?P<t2>\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: <target>``,
``Incremented Retry for (url='<target>')``, ``Retrying () after
connection broken by '': <target>``), and origin-form halves of
``Redirecting <target> -> <target>``. 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 = "<redacted>@" if match.group("userinfo") else ""
return match.group("scheme") + userinfo + match.group("host") + "/<redacted>"
def _redact_request_line(match: re.Match[str]) -> str:
userinfo = "<redacted>@" if match.group("userinfo") else ""
return match.group("scheme") + userinfo + match.group("host") + ' "' + match.group("method") + " /<redacted> " + 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='/<redacted>')"
def _redact_retry_target(match: re.Match[str]) -> str:
return "Retry: /<redacted>"
def _redact_retrying(match: re.Match[str]) -> str:
return match.group("head") + ": /<redacted>"
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 "/<redacted>" 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 <url> -> <url>`` 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")))

View File

@ -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"}},
}
],
}

View File

@ -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-<id>.cos.<region>.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({}) == "<unnamed>"
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 <full URL>`` 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/<redacted>" 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/<redacted>" 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

View File

@ -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/<redacted>" 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<token>/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/<redacted>" 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 <redacted>@ 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://<redacted>@internal-proxy.corp:8080/<redacted>" 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://<redacted>@internal-proxy.corp" in authority_formatted
assert "tok" not in authority_formatted.replace("<redacted>", "")
# 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 "<redacted>" 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/<redacted>") == 2
assert out.count("mirror.example/<redacted>") == 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 /<redacted>.
record = _record("/private/BearerSecret?token=QuerySecret")
assert filt.filter(record) is True
formatted = record.getMessage()
assert formatted == 'https://cdn.example:443 "GET /<redacted> 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 /<redacted> 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 /<redacted> 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 /<redacted> 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 ``/<redacted>`` 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: /<redacted>"
# 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/<redacted>"
# 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='/<redacted>'): '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/<redacted>'): '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://<redacted>@internal-proxy.corp:8080/<redacted>'): '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}': /<redacted>"
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 /<redacted> -> /<redacted>"
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 /<redacted> -> https://mirror.example/<redacted>"
# 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 /<redacted> -> /<redacted>"
# 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 /<redacted> 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='/<redacted>')" in out
assert "Retry: /<redacted>" in out
assert "after connection broken by" in out and "Read timed out" in out and "': /<redacted>" in out
assert "Redirecting /<redacted> -> https://mirror.example/<redacted>" 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/<redacted>" 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/<redacted>" in formatted
# The nested request line kept its quoted target redacted too.
assert 'https://evil.example "GET /<redacted> 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/<redacted>"
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/<redacted> "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/<redacted> "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/<redacted>" 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/<redacted>' 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='/<redacted>'): '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
<t> -> <t>`` message because an ``-> /path`` arrow is not urllib3-owned
shape: the sandbox provider's actionable mount error renders
``sandbox.mounts entry <host_path> -> <container_path>`` and a substring
match rewrote the container path to ``/<redacted>``, 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 "<redacted>" 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 /<redacted> -> /<redacted>"
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 /<redacted> -> /<redacted>"
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 /<redacted> -> https://mirror.example/<redacted> 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"

View File

@ -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())

View File

@ -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.<region>.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