Bremen
/
emma-hermes
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(gateway/platforms): migrate all Weixin fallbacks to get_secret() for consistent profile-scoped resolution 

Per egilewski's security review, WEIXIN_BASE_URL and WEIXIN_CDN_BASE_URL
were still resolved from process-global environment variables, leaving
mixed-scope bypasses in multiplex mode.

Changed files:
- gateway/platforms/weixin.py: Added get_secret import, replaced os.getenv()
  with get_secret() for WEIXIN_ACCOUNT_ID, WEIXIN_TOKEN, WEIXIN_BASE_URL,
  WEIXIN_CDN_BASE_URL in WeixinAdapter.__init__() and send_weixin_direct()
- tools/send_message_tool.py: Added get_secret import, replaced os.getenv()
  with get_secret() for all WEIXIN_* fallbacks in _handle_send()

All runtime Weixin send paths now resolve both credentials and endpoint
configuration from the same profile-scoped source.
fix/verification-admin-route-recovery
liuhao1024 2026-07-08 06:21:33 +08:00 committed by Teknium
parent 8fc989b416
commit 6160a80253
2 changed files with 14 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
gateway/platforms/weixin.py
View File

@ -68,6 +68,7 @@ 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
ILINK_BASE_URL = "https://ilinkai.weixin.qq.com"
WEIXIN_CDN_BASE_URL = "https://novac2c.cdn.weixin.qq.com/c2c"
@ -1159,11 +1160,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 os.getenv("WEIXIN_ACCOUNT_ID", "")).strip()
self._token = str(config.token or extra.get("token") or os.getenv("WEIXIN_TOKEN", "")).strip()
self._base_url = str(extra.get("base_url") or os.getenv("WEIXIN_BASE_URL", ILINK_BASE_URL)).strip().rstrip("/")
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._cdn_base_url = str(
extra.get("cdn_base_url") or os.getenv("WEIXIN_CDN_BASE_URL", WEIXIN_CDN_BASE_URL)
extra.get("cdn_base_url") or get_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")
@ -2292,10 +2293,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 os.getenv("WEIXIN_ACCOUNT_ID", "")).strip()
base_url = str(extra.get("base_url") or os.getenv("WEIXIN_BASE_URL", ILINK_BASE_URL)).strip().rstrip("/")
cdn_base_url = str(extra.get("cdn_base_url") or os.getenv("WEIXIN_CDN_BASE_URL", WEIXIN_CDN_BASE_URL)).strip().rstrip("/")
resolved_token = str(token or extra.get("token") or os.getenv("WEIXIN_TOKEN", "")).strip()
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()
if not resolved_token:
return {"error": "Weixin token missing. Configure WEIXIN_TOKEN or platforms.weixin.token."}
if not account_id:

9
tools/send_message_tool.py
View File

@ -15,6 +15,7 @@ import time
from email.utils import formatdate
from agent.redact import redact_sensitive_text
from agent.secret_scope import get_secret
logger = logging.getLogger(__name__)
@ -409,8 +410,8 @@ def _handle_send(args):
# Weixin can be configured purely via .env; synthesize a pconfig so
# send_message and cron delivery work without a gateway.yaml entry.
if platform_name == "weixin":
wx_token = os.getenv("WEIXIN_TOKEN", "").strip()
wx_account = os.getenv("WEIXIN_ACCOUNT_ID", "").strip()
wx_token = get_secret("WEIXIN_TOKEN", "").strip()
wx_account = get_secret("WEIXIN_ACCOUNT_ID", "").strip()
if wx_token and wx_account:
from gateway.config import PlatformConfig
pconfig = PlatformConfig(
@ -418,8 +419,8 @@ def _handle_send(args):
token=wx_token,
extra={
"account_id": wx_account,
"base_url": os.getenv("WEIXIN_BASE_URL", "").strip(),
"cdn_base_url": os.getenv("WEIXIN_CDN_BASE_URL", "").strip(),
"base_url": get_secret("WEIXIN_BASE_URL", "").strip(),
"cdn_base_url": get_secret("WEIXIN_CDN_BASE_URL", "").strip(),
},
)
else: