fix(weixin): Slack-pattern secret scoping for WEIXIN_* credential reads

The adapter's __init__ and send_weixin_direct read WEIXIN_ACCOUNT_ID/ TOKEN/BASE_URL/CDN_BASE_URL via bare get_secret, which raises UnscopedSecretError when the DEFAULT profile's adapter constructs or sends unscoped under multiplexing (corrects the direction of #66073 / #68854, which tried to solve this by borrowing os.environ on every read — a cross-profile leak). Add a module-level _wx_secret helper following the established Slack SLACK_APP_TOKEN pattern (#59739) and WhatsApp's _get_wsecret: a SCOPED miss returns the default (the scope is authoritative — no environ borrow), while an UNSCOPED read under multiplex falls back to os.environ, which is the default profile's own value. Regression tests cover both directions: scoped construction reads the scope's value and a scoped miss yields empty (no borrow); unscoped construction falls back to os.environ instead of raising.
2026-08-02 00:50:20 -07:00
parent 3fb066c372
commit f7efe3d766
2 changed files with 130 additions and 9 deletions
@@ -68,7 +68,25 @@ from gateway.platforms.base import (
 )
 from hermes_constants import get_hermes_home
 from utils import atomic_json_write
-from agent.secret_scope import get_secret
+from agent.secret_scope import UnscopedSecretError, get_secret
+
+
+def _wx_secret(name: str, default: Optional[str] = None) -> Optional[str]:
+    """Scope-aware WEIXIN_* read with the default-profile startup fallback.
+
+    Secondary profiles construct their adapters under
+    ``_profile_runtime_scope`` — the scope is authoritative and a scoped miss
+    returns ``default`` (no cross-profile borrow from ``os.environ``). The
+    DEFAULT profile's adapter constructs and sends *unscoped* under
+    multiplexing, where a bare ``get_secret`` would raise
+    ``UnscopedSecretError`` and crash its Weixin path; there ``os.environ`` is
+    that profile's own value, so fall back to it. Same pattern as the Slack
+    ``SLACK_APP_TOKEN`` read (#59739) and WhatsApp's ``_get_wsecret``.
+    """
+    try:
+        return get_secret(name, default)
+    except UnscopedSecretError:
+        return os.getenv(name, default)

 ILINK_BASE_URL = "https://ilinkai.weixin.qq.com"
 WEIXIN_CDN_BASE_URL = "https://novac2c.cdn.weixin.qq.com/c2c"
@@ -1175,11 +1193,11 @@ class WeixinAdapter(BasePlatformAdapter):
        self._poll_task: Optional[asyncio.Task] = None
        self._dedup = MessageDeduplicator(ttl_seconds=MESSAGE_DEDUP_TTL_SECONDS)

-        self._account_id = str(extra.get("account_id") or get_secret("WEIXIN_ACCOUNT_ID", "")).strip()
-        self._token = str(config.token or extra.get("token") or get_secret("WEIXIN_TOKEN", "")).strip()
-        self._base_url = str(extra.get("base_url") or get_secret("WEIXIN_BASE_URL", ILINK_BASE_URL)).strip().rstrip("/")
+        self._account_id = str(extra.get("account_id") or _wx_secret("WEIXIN_ACCOUNT_ID", "")).strip()
+        self._token = str(config.token or extra.get("token") or _wx_secret("WEIXIN_TOKEN", "")).strip()
+        self._base_url = str(extra.get("base_url") or _wx_secret("WEIXIN_BASE_URL", ILINK_BASE_URL)).strip().rstrip("/")
        self._cdn_base_url = str(
-            extra.get("cdn_base_url") or get_secret("WEIXIN_CDN_BASE_URL", WEIXIN_CDN_BASE_URL)
+            extra.get("cdn_base_url") or _wx_secret("WEIXIN_CDN_BASE_URL", WEIXIN_CDN_BASE_URL)
        ).strip().rstrip("/")
        self._send_chunk_delay_seconds = float(
            extra.get("send_chunk_delay_seconds") or os.getenv("WEIXIN_SEND_CHUNK_DELAY_SECONDS", "1.5")
@@ -2313,10 +2331,10 @@ async def send_weixin_direct(

    This bypasses the long-poll adapter lifecycle and uses the raw API directly.
    """
-    account_id = str(extra.get("account_id") or get_secret("WEIXIN_ACCOUNT_ID", "")).strip()
-    base_url = str(extra.get("base_url") or get_secret("WEIXIN_BASE_URL", ILINK_BASE_URL)).strip().rstrip("/")
-    cdn_base_url = str(extra.get("cdn_base_url") or get_secret("WEIXIN_CDN_BASE_URL", WEIXIN_CDN_BASE_URL)).strip().rstrip("/")
-    resolved_token = str(token or extra.get("token") or get_secret("WEIXIN_TOKEN", "")).strip()
+    account_id = str(extra.get("account_id") or _wx_secret("WEIXIN_ACCOUNT_ID", "")).strip()
+    base_url = str(extra.get("base_url") or _wx_secret("WEIXIN_BASE_URL", ILINK_BASE_URL)).strip().rstrip("/")
+    cdn_base_url = str(extra.get("cdn_base_url") or _wx_secret("WEIXIN_CDN_BASE_URL", WEIXIN_CDN_BASE_URL)).strip().rstrip("/")
+    resolved_token = str(token or extra.get("token") or _wx_secret("WEIXIN_TOKEN", "")).strip()
    if not resolved_token:
        return {"error": "Weixin token missing. Configure WEIXIN_TOKEN or platforms.weixin.token."}
    if not account_id:
@@ -0,0 +1,103 @@
+"""Weixin adapter secret-scope regression tests.
+
+The adapter's WEIXIN_* credential reads must follow the Slack pattern
+(#59739): under multiplexing a SCOPED miss is authoritative (no borrow from
+``os.environ`` — that would be a cross-profile leak), while an UNSCOPED read
+(default-profile startup/send path) falls back to ``os.environ``, which is
+that profile's own value, instead of raising ``UnscopedSecretError``.
+"""
+
+import pytest
+
+from agent import secret_scope
+from gateway.config import PlatformConfig
+from gateway.platforms.weixin import WeixinAdapter, _wx_secret
+
+
+@pytest.fixture()
+def multiplex_on():
+    previous = secret_scope.is_multiplex_active()
+    secret_scope.set_multiplex_active(True)
+    try:
+        yield
+    finally:
+        secret_scope.set_multiplex_active(previous)
+
+
+class TestWxSecretHelper:
+    def test_scoped_read_uses_scope_value(self, multiplex_on, monkeypatch):
+        monkeypatch.setenv("WEIXIN_TOKEN", "default-profile-token")
+        token = secret_scope.set_secret_scope({"WEIXIN_TOKEN": "scoped-token"})
+        try:
+            assert _wx_secret("WEIXIN_TOKEN") == "scoped-token"
+        finally:
+            secret_scope.reset_secret_scope(token)
+
+    def test_scoped_miss_does_not_borrow_environ(self, multiplex_on, monkeypatch):
+        """A secondary profile without WEIXIN_TOKEN must NOT inherit the
+        default profile's process-env token."""
+        monkeypatch.setenv("WEIXIN_TOKEN", "default-profile-token")
+        token = secret_scope.set_secret_scope({"OTHER_KEY": "x"})
+        try:
+            assert _wx_secret("WEIXIN_TOKEN", "") == ""
+            assert _wx_secret("WEIXIN_TOKEN") is None
+        finally:
+            secret_scope.reset_secret_scope(token)
+
+    def test_unscoped_read_falls_back_to_environ(self, multiplex_on, monkeypatch):
+        """The default profile's adapter runs unscoped under multiplexing;
+        os.environ is its own value — fall back instead of raising."""
+        monkeypatch.setenv("WEIXIN_TOKEN", "default-profile-token")
+        token = secret_scope.set_secret_scope(None)
+        try:
+            assert _wx_secret("WEIXIN_TOKEN") == "default-profile-token"
+        finally:
+            secret_scope.reset_secret_scope(token)
+
+
+class TestWeixinAdapterConstructionScope:
+    def test_multiplex_scoped_construction_reads_scope_not_environ(
+        self, multiplex_on, monkeypatch
+    ):
+        monkeypatch.setenv("WEIXIN_ACCOUNT_ID", "env-account")
+        monkeypatch.setenv("WEIXIN_TOKEN", "env-token")
+        token = secret_scope.set_secret_scope(
+            {
+                "WEIXIN_ACCOUNT_ID": "scoped-account",
+                "WEIXIN_TOKEN": "scoped-token",
+            }
+        )
+        try:
+            adapter = WeixinAdapter(PlatformConfig(enabled=True))
+        finally:
+            secret_scope.reset_secret_scope(token)
+        assert adapter._account_id == "scoped-account"
+        assert adapter._token == "scoped-token"
+
+    def test_multiplex_scoped_miss_yields_empty_not_environ_borrow(
+        self, multiplex_on, monkeypatch
+    ):
+        monkeypatch.setenv("WEIXIN_ACCOUNT_ID", "env-account")
+        monkeypatch.setenv("WEIXIN_TOKEN", "env-token")
+        token = secret_scope.set_secret_scope({"SOMETHING_ELSE": "x"})
+        try:
+            adapter = WeixinAdapter(PlatformConfig(enabled=True))
+        finally:
+            secret_scope.reset_secret_scope(token)
+        assert adapter._account_id == ""
+        assert adapter._token == ""
+
+    def test_multiplex_unscoped_construction_falls_back_to_environ(
+        self, multiplex_on, monkeypatch
+    ):
+        """Regression for the bare get_secret reads: default-profile adapter
+        construction under multiplexing must not raise UnscopedSecretError."""
+        monkeypatch.setenv("WEIXIN_ACCOUNT_ID", "env-account")
+        monkeypatch.setenv("WEIXIN_TOKEN", "env-token")
+        token = secret_scope.set_secret_scope(None)
+        try:
+            adapter = WeixinAdapter(PlatformConfig(enabled=True))
+        finally:
+            secret_scope.reset_secret_scope(token)
+        assert adapter._account_id == "env-account"
+        assert adapter._token == "env-token"