From acf3fffb4dcb0bfa565e28c4ea2a8d28ea43b565 Mon Sep 17 00:00:00 2001 From: qbit-mirror-bot Date: Wed, 1 Jul 2026 16:24:34 +0000 Subject: [PATCH] fix: add encoding="utf-8" to Path.write_text() calls (P1) --- agent/copilot_acp_client.py | 2 +- gateway/dead_targets.py | 2 +- gateway/delivery.py | 4 +- gateway/platforms/qqbot/adapter.py | 43 +++- gateway/run.py | 267 ++++++++++++++++++++---- gateway/slash_commands.py | 323 ++--------------------------- hermes_cli/banner.py | 2 +- hermes_cli/container_boot.py | 10 +- hermes_cli/gateway.py | 149 ++++++++++++- hermes_cli/main.py | 4 +- hermes_cli/profiles.py | 6 +- hermes_cli/service_manager.py | 10 +- hermes_cli/uninstall.py | 2 +- tools/skills_hub.py | 24 ++- tools/web_tools.py | 2 +- tools/xai_http.py | 2 +- 16 files changed, 463 insertions(+), 389 deletions(-) diff --git a/agent/copilot_acp_client.py b/agent/copilot_acp_client.py index ce3ec2c5c400..ed0f8d285652 100644 --- a/agent/copilot_acp_client.py +++ b/agent/copilot_acp_client.py @@ -732,7 +732,7 @@ def _handle_server_message( f"Write denied: '{path}' is a protected system/credential file." ) path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(str(params.get("content") or "")) + path.write_text(str(params.get("content") or ""), encoding="utf-8") response = { "jsonrpc": "2.0", "id": message_id, diff --git a/gateway/dead_targets.py b/gateway/dead_targets.py index 66a9247f213e..2f771d23ca6b 100644 --- a/gateway/dead_targets.py +++ b/gateway/dead_targets.py @@ -82,7 +82,7 @@ def _flush_locked(self) -> None: try: self._path.parent.mkdir(parents=True, exist_ok=True) tmp = self._path.with_suffix(self._path.suffix + ".tmp") - tmp.write_text(json.dumps(self._dead, indent=2)) + tmp.write_text(json.dumps(self._dead, indent=2), encoding="utf-8") tmp.replace(self._path) except OSError as exc: # Best-effort: keep the in-memory state, don't break delivery. diff --git a/gateway/delivery.py b/gateway/delivery.py index 77b245d291c3..97472bd9d852 100644 --- a/gateway/delivery.py +++ b/gateway/delivery.py @@ -357,7 +357,7 @@ def _deliver_local( lines.append("") lines.append(content) - output_path.write_text("\n".join(lines)) + output_path.write_text("\n".join(lines), encoding="utf-8") return { "path": str(output_path), @@ -370,7 +370,7 @@ def _save_full_output(self, content: str, job_id: str) -> Path: out_dir = get_hermes_home() / "cron" / "output" out_dir.mkdir(parents=True, exist_ok=True) path = out_dir / f"{job_id}_{timestamp}.txt" - path.write_text(content) + path.write_text(content, encoding="utf-8") return path def _filter_silence_narration_enabled(self) -> bool: diff --git a/gateway/platforms/qqbot/adapter.py b/gateway/platforms/qqbot/adapter.py index 9532662131df..de2531bb9161 100644 --- a/gateway/platforms/qqbot/adapter.py +++ b/gateway/platforms/qqbot/adapter.py @@ -12,9 +12,9 @@ app_id: "your-app-id" # or QQ_APP_ID env var client_secret: "your-secret" # or QQ_CLIENT_SECRET env var markdown_support: true # enable QQ markdown (msg_type 2) - dm_policy: "open" # open | allowlist | disabled + dm_policy: "pairing" # open | allowlist | disabled | pairing allow_from: ["openid_1"] - group_policy: "open" # open | allowlist | disabled + group_policy: "pairing" # open | allowlist | disabled | pairing group_allow_from: ["group_openid_1"] stt: # Voice-to-text config (optional) provider: "zai" # zai (GLM-ASR), openai (Whisper), etc. @@ -208,11 +208,11 @@ def __init__(self, config: PlatformConfig): self._markdown_support = bool(extra.get("markdown_support", True)) # Auth/ACL policies - self._dm_policy = str(extra.get("dm_policy", "open")).strip().lower() + self._dm_policy = str(extra.get("dm_policy", "pairing")).strip().lower() self._allow_from = _coerce_list( extra.get("allow_from") or extra.get("allowFrom") ) - self._group_policy = str(extra.get("group_policy", "open")).strip().lower() + self._group_policy = str(extra.get("group_policy", "pairing")).strip().lower() self._group_allow_from = _coerce_list( extra.get("group_allow_from") or extra.get("groupAllowFrom") ) @@ -1193,7 +1193,7 @@ def _write_update_response(answer: str, operator: str = "") -> None: home = get_hermes_home() response_path = home / ".update_response" tmp = response_path.with_suffix(".tmp") - tmp.write_text(answer) + tmp.write_text(answer, encoding="utf-8") tmp.replace(response_path) logger.info( "QQ update prompt answered %r by %s", @@ -1214,7 +1214,7 @@ async def _handle_c2c_message( user_openid = str(author.get("user_openid", "")) if not user_openid: return - if not self._is_dm_allowed(user_openid): + if not self._is_dm_intake_allowed(user_openid): return text = content @@ -1454,7 +1454,7 @@ async def _handle_dm_message( # Without this check any member of any guild the bot is in could # bypass the configured allowlist via direct messages. author_id = str(author.get("id", "")) - if not self._is_dm_allowed(author_id): + if not self._is_dm_intake_allowed(author_id): logger.debug( "[%s] Guild DM blocked by ACL: guild=%s user=%s", self._log_tag, guild_id, author_id, @@ -3142,19 +3142,44 @@ def _strip_at_mention(content: str) -> str: stripped = re.sub(r"^@\S+\s*", "", content.strip()) return stripped + def _open_dm_opted_in(self) -> bool: + if os.getenv("GATEWAY_ALLOW_ALL_USERS", "").lower() in {"true", "1", "yes"}: + return True + return os.getenv("QQ_ALLOW_ALL_USERS", "").lower() in {"true", "1", "yes"} + def _is_dm_allowed(self, user_id: str) -> bool: if self._dm_policy == "disabled": return False if self._dm_policy == "allowlist": return self._entry_matches(self._allow_from, user_id) - return True + if self._dm_policy == "open": + return self._open_dm_opted_in() + return False + + def _is_dm_intake_allowed(self, user_id: str) -> bool: + principal = str(user_id or "").strip() + if not principal: + return False + if self._dm_policy == "disabled": + return False + if self._dm_policy == "allowlist": + return self._entry_matches(self._allow_from, principal) + if self._dm_policy == "pairing": + return True + if self._dm_policy == "open": + return self._open_dm_opted_in() + return False def _is_group_allowed(self, group_id: str, user_id: str) -> bool: if self._group_policy == "disabled": return False if self._group_policy == "allowlist": return self._entry_matches(self._group_allow_from, group_id) - return True + if self._group_policy == "pairing": + return False + if self._group_policy == "open": + return True + return False @staticmethod def _entry_matches(entries: List[str], target: str) -> bool: diff --git a/gateway/run.py b/gateway/run.py index 63ccc8a9886d..12190d51512b 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -618,20 +618,19 @@ def _coerce_gateway_timestamp(value: Any) -> Optional[float]: def _auto_continue_freshness_window() -> float: """Return the configured auto-continue freshness window in seconds. - Reads ``HERMES_AUTO_CONTINUE_FRESHNESS`` (bridged from - ``config.yaml`` ``agent.gateway_auto_continue_freshness`` at gateway - startup, same pattern as ``HERMES_AGENT_TIMEOUT``). Falls back to the - module default when unset or malformed. Non-positive values disable - the freshness gate (restores the pre-fix "always fresh" behaviour for - users who want to opt out). + Thin wrapper that delegates to the canonical implementation in + ``gateway.session`` (the single source of truth shared with the + routing-time zombie gate in ``get_or_create_session``). Reads + ``HERMES_AUTO_CONTINUE_FRESHNESS`` (bridged from ``config.yaml`` + ``agent.gateway_auto_continue_freshness`` at gateway startup, same + pattern as ``HERMES_AGENT_TIMEOUT``). Falls back to the module default + when unset or malformed. Non-positive values disable the freshness gate + (restores the pre-fix "always fresh" behaviour for users who want to opt + out). Kept here so existing call sites and test patches importing it + from ``gateway.run`` continue to work. """ - raw = os.environ.get("HERMES_AUTO_CONTINUE_FRESHNESS") - if raw is None or raw == "": - return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT) - try: - return float(raw) - except (TypeError, ValueError): - return float(_AUTO_CONTINUE_FRESHNESS_SECS_DEFAULT) + from gateway.session import auto_continue_freshness_window + return auto_continue_freshness_window() def _float_env(name: str, default: float) -> float: @@ -1707,6 +1706,48 @@ def _profile_runtime_scope(profile_home: "Path"): logger = logging.getLogger(__name__) +_OWN_POLICY_OPEN_ENV = { + Platform.WECOM: ("WECOM_DM_POLICY", "WECOM_GROUP_POLICY", "WECOM_ALLOW_ALL_USERS"), + Platform.WEIXIN: ("WEIXIN_DM_POLICY", "WEIXIN_GROUP_POLICY", "WEIXIN_ALLOW_ALL_USERS"), + Platform.YUANBAO: ("YUANBAO_DM_POLICY", "YUANBAO_GROUP_POLICY", "YUANBAO_ALLOW_ALL_USERS"), + Platform.QQBOT: (None, None, "QQ_ALLOW_ALL_USERS"), + Platform.WHATSAPP: ("WHATSAPP_DM_POLICY", "WHATSAPP_GROUP_POLICY", "WHATSAPP_ALLOW_ALL_USERS"), +} + + +def _own_policy_open_startup_violation(config) -> Optional[str]: + """Return a startup-abort reason when open policy lacks allow-all opt-in.""" + for platform, platform_config in getattr(config, "platforms", {}).items(): + if not getattr(platform_config, "enabled", False): + continue + open_env = _OWN_POLICY_OPEN_ENV.get(platform) + if not open_env: + continue + dm_env, group_env, allow_all_env = open_env + extra = getattr(platform_config, "extra", None) or {} + dm_policy = str( + extra.get("dm_policy") + or (os.getenv(dm_env, "pairing") if dm_env else "pairing") + ).strip().lower() + group_policy = str( + extra.get("group_policy") + or (os.getenv(group_env, "pairing") if group_env else "pairing") + ).strip().lower() + if dm_policy != "open" and group_policy != "open": + continue + gateway_allow_all = os.getenv( + "GATEWAY_ALLOW_ALL_USERS", "" + ).lower() in {"true", "1", "yes"} + platform_opted_in = gateway_allow_all or ( + allow_all_env + and os.getenv(allow_all_env, "").lower() in {"true", "1", "yes"} + ) + if platform_opted_in: + continue + return f"{platform.value}: open policy without allow-all opt-in" + return None + + # Sentinel placed into _running_agents immediately when a session starts # processing, *before* any await. Prevents a second message for the same # session from bypassing the "already running" guard during the async gap @@ -3008,7 +3049,7 @@ def _save_voice_modes(self) -> None: self._VOICE_MODE_PATH.parent.mkdir(parents=True, exist_ok=True) self._VOICE_MODE_PATH.write_text( json.dumps(self._voice_mode, indent=2) - ) +, encoding="utf-8") except OSError as e: logger.warning("Failed to save voice modes: %s", e) @@ -4714,7 +4755,7 @@ def _agent_has_active_subagents(running_agent: Any) -> bool: _BUSY_QUEUE_MAX_PENDING = 32 def _queue_or_replace_pending_event(self, session_key: str, event: MessageEvent) -> None: - adapter = self.adapters.get(event.source.platform) + adapter = self._adapter_for_source(event.source) if not adapter: return # #28503 — Previously this called ``merge_pending_message_event`` @@ -4771,7 +4812,7 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session # --- Draining case (gateway restarting/stopping) --- if self._draining: - adapter = self.adapters.get(event.source.platform) + adapter = self._adapter_for_source(event.source) if not adapter: return True @@ -4873,7 +4914,7 @@ async def _handle_active_session_busy_message(self, event: MessageEvent, session ) # Normal busy case (agent actively running a task) - adapter = self.adapters.get(event.source.platform) + adapter = self._adapter_for_source(event.source) if not adapter: return False # let default path handle it @@ -6266,10 +6307,34 @@ async def start(self) -> bool: ) if not _any_allowlist and not _allow_all: logger.warning( - "No user allowlists configured. All unauthorized users will be denied. " - "Set GATEWAY_ALLOW_ALL_USERS=true in ~/.hermes/.env to allow open access, " - "or configure platform allowlists (e.g., TELEGRAM_ALLOWED_USERS=your_id)." + "No env user allowlists configured. Messaging platforms default to " + "pairing/allowlist policies and will deny unknown senders unless you " + "configure platform allowlists (e.g., TELEGRAM_ALLOWED_USERS=your_id) " + "or explicitly opt in with GATEWAY_ALLOW_ALL_USERS=true plus " + "dm_policy/group_policy: open on the platform." + ) + + reason = _own_policy_open_startup_violation(self.config) + if reason: + platform_value = reason.split(":", 1)[0] + allow_all_env = None + for platform, open_env in _OWN_POLICY_OPEN_ENV.items(): + if platform.value == platform_value: + allow_all_env = open_env[2] + break + logger.error( + "Refusing to start: %s has dm_policy/group_policy set to 'open' " + "but neither GATEWAY_ALLOW_ALL_USERS nor %s is enabled.", + platform_value, + allow_all_env or "a platform allow-all flag", ) + try: + from gateway.status import write_runtime_status + write_runtime_status(gateway_state="startup_failed", exit_reason=reason) + except Exception: + pass + self._request_clean_exit(reason) + return True # Discover Python plugins before shell hooks so plugin block # decisions take precedence in tie cases. The CLI startup path @@ -7860,6 +7925,14 @@ async def _start_one_profile_adapters( with _profile_runtime_scope(profile_home): profile_cfg = load_gateway_config() + violation = _own_policy_open_startup_violation(profile_cfg) + if violation: + raise MultiplexConfigError( + f"Profile '{profile_name}' enables {violation}. " + "Enable GATEWAY_ALLOW_ALL_USERS or the platform allow-all flag " + "for that profile, or change dm_policy/group_policy away from " + "'open'." + ) profile_map = self._profile_adapters.setdefault(profile_name, {}) connected = 0 @@ -8256,7 +8329,14 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: elif not self._is_user_authorized(source): logger.warning("Unauthorized user: %s (%s) on %s", source.user_id, source.user_name, source.platform.value) # In DMs: offer pairing code. In groups: silently ignore. - if source.chat_type == "dm" and self._get_unauthorized_dm_behavior(source.platform) == "pair": + if ( + source.chat_type == "dm" + and self._get_unauthorized_dm_behavior( + source.platform, + profile=source.profile, + ) + == "pair" + ): platform_name = source.platform.value if source.platform else "unknown" # Rate-limit ALL pairing responses (code or rejection) to # prevent spamming the user with repeated messages when @@ -8267,7 +8347,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: platform_name, source.user_id, source.user_name or "" ) if code: - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: await adapter.send( source.chat_id, @@ -8277,7 +8357,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: f"`hermes pairing approve {platform_name} {code}`" ) else: - adapter = self.adapters.get(source.platform) + adapter = self._adapter_for_source(source) if adapter: await adapter.send( source.chat_id, @@ -8328,7 +8408,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: prompt_path = _hermes_home / ".update_prompt.json" try: tmp = response_path.with_suffix(".tmp") - tmp.write_text(response_text) + tmp.write_text(response_text, encoding="utf-8") tmp.replace(response_path) prompt_path.unlink(missing_ok=True) except OSError as e: @@ -8348,7 +8428,7 @@ async def _handle_message(self, event: MessageEvent) -> Optional[str]: prompt_path = _hermes_home / ".update_prompt.json" try: tmp = response_path.with_suffix(".tmp") - tmp.write_text("") + tmp.write_text("", encoding="utf-8") tmp.replace(response_path) prompt_path.unlink(missing_ok=True) logger.info( @@ -13444,7 +13524,7 @@ async def _watch_update_progress( return await asyncio.sleep(poll_interval) if (pending_path.exists() or claimed_path.exists()) and not exit_code_path.exists(): - exit_code_path.write_text("124") + exit_code_path.write_text("124", encoding="utf-8") await self._send_update_notification() return @@ -13591,7 +13671,7 @@ async def _flush_buffer() -> None: # Timeout if not exit_code_path.exists(): logger.warning("Update watcher timed out after %.0fs", timeout) - exit_code_path.write_text("124") + exit_code_path.write_text("124", encoding="utf-8") await _flush_buffer() try: await adapter.send( @@ -17254,10 +17334,30 @@ def _approval_notify_sync(approval_data: dict) -> None: _resume_entry = self.session_store._entries.get(session_key) except Exception: _resume_entry = None + + # resume_pending freshness uses a SECOND signal in addition to the + # transcript clock above. The restart watchdog stamps the session + # with ``last_resume_marked_at`` at interrupt time — that is the + # correct "when were we interrupted" signal. The transcript clock + # (_interruption_is_fresh) can be far older: an active thread you + # return to may have its last persisted row hours back, even though + # the interruption itself just happened. Gating resume_pending on + # the transcript clock alone makes the recovery note silently drop, + # and because the startup auto-resume turn carries empty text + # (_schedule_resume_pending_sessions), the model then receives a + # blank user message and replies with confused "the message came + # through blank" noise. Treat the marker as fresh when + # EITHER signal is fresh so the two freshness checks agree. + _resume_mark_is_fresh = False + if _resume_entry is not None and getattr(_resume_entry, "resume_pending", False): + _resume_mark_is_fresh = _is_fresh_gateway_interruption( + getattr(_resume_entry, "last_resume_marked_at", None), + window_secs=_freshness_window, + ) _is_resume_pending = bool( _resume_entry is not None and getattr(_resume_entry, "resume_pending", False) - and _interruption_is_fresh + and (_interruption_is_fresh or _resume_mark_is_fresh) ) _has_fresh_tool_tail = bool( agent_history @@ -17319,6 +17419,42 @@ def _approval_notify_sync(approval_data: dict) -> None: if _srn: message = _srn + "\n\n" + message + # Safety net: a startup auto-resume event carries empty + # text and relies on the resume_pending branch above to supply the + # recovery note. If that branch did not fire for any reason (e.g. + # both freshness signals disagreed, or the marker was cleared + # between scheduling and dispatch) we must NOT hand the model a + # blank user turn — it responds with confused "the message came + # through blank" noise. Restricted to resume_pending sessions so + # legitimately empty user turns (e.g. an image with no caption, + # wrapped as native content below) are untouched. + if ( + isinstance(message, str) + and not message.strip() + and _resume_entry is not None + and getattr(_resume_entry, "resume_pending", False) + ): + _sn_reason = ( + getattr(_resume_entry, "resume_reason", None) or "restart_timeout" + ) + _sn_reason_phrase = ( + "a gateway restart" + if _sn_reason == "restart_timeout" + else "a gateway shutdown" + if _sn_reason == "shutdown_timeout" + else "a gateway interruption" + ) + message = ( + f"[System note: The previous turn was interrupted by " + f"{_sn_reason_phrase}; the gateway is now back online. " + f"Any restart/shutdown command in the history has already " + f"run — do NOT re-execute or verify it. Report to the user " + f"that the session was restored successfully and ask what " + f"they would like to do next. Do NOT re-execute old tool " + f"calls — skip any unfinished work from the conversation " + f"history.]" + ) + _approval_session_key = session_key or "" _approval_session_token = set_current_session_key(_approval_session_key) register_gateway_notify(_approval_session_key, _approval_notify_sync) @@ -19298,21 +19434,76 @@ def main(): data = yaml.safe_load(f) or {} config = GatewayConfig.from_dict(data) - # start_gateway() already performs graceful teardown before returning. - # Force-exit afterwards so a wedged non-daemon worker thread cannot block - # interpreter finalization and strand the gateway half-shut down. - success = asyncio.run(start_gateway(config)) - _exit_after_graceful_shutdown(success) - - -def _exit_after_graceful_shutdown(success: bool) -> None: - """Flush stdio and terminate immediately after graceful shutdown.""" + # start_gateway() performs the full graceful teardown (adapters + # disconnected, sessions saved + flushed, SQLite closed, cron/MCP stopped, + # PID file + runtime lock released) before it returns OR raises SystemExit + # with an explicit code. Force-exit afterwards so a wedged non-daemon worker + # thread (e.g. a ThreadPoolExecutor tool/LLM call blocked with no timeout) + # cannot block interpreter finalization (Py_FinalizeEx joins all non-daemon + # threads, incl. concurrent.futures' _python_exit) and strand the gateway + # half-shut down with the supervisor unable to restart it (#53107). + # + # SystemExit is caught explicitly: start_gateway raises it on the + # clean-fatal-config (#51228), planned-restart, and service-restart paths, + # all of which complete teardown first. Routing those codes through the + # same os._exit backstop means EVERY exit path is wedge-proof, not just the + # boolean-return ones. + try: + success = asyncio.run(start_gateway(config)) + exit_code = 0 if success else 1 + except SystemExit as e: + # e.code may be None (→ 0), an int, or a str (→ 1, like CPython). + if e.code is None: + exit_code = 0 + elif isinstance(e.code, int): + exit_code = e.code + else: + exit_code = 1 + _exit_after_graceful_shutdown(exit_code) + + +def _exit_after_graceful_shutdown(exit_code: int) -> None: + """Flush stdio, release the PID file + runtime lock, then hard-exit. + + Graceful teardown is already complete by the time this runs, so there is + nothing left that needs a clean interpreter shutdown. We deliberately use + ``os._exit`` (not ``sys.exit``): ``sys.exit`` raises ``SystemExit``, which + triggers ``Py_FinalizeEx`` → ``wait_for_thread_shutdown`` and joins every + non-daemon thread — exactly the hang (#53107) a wedged tool-worker causes. + + ``os._exit`` bypasses ``atexit`` handlers, so we cannot rely on the + ``atexit``-registered ``remove_pid_file`` / ``release_gateway_runtime_lock`` + (registered in ``start_gateway``) to run. The full-shutdown path releases + both explicitly in ``_stop_impl``, but the EARLY exit paths — + clean-fatal-config (#51228) and startup-aborted-before-running — raise + ``SystemExit`` right after ``runner.start()`` without going through + ``_stop_impl``, so on those paths ``atexit`` was the only thing releasing + them. Now that those paths are routed through this backstop (#53107), + release both here explicitly. Both calls are idempotent — + ``remove_pid_file`` only unlinks a PID file that belongs to this process, + and ``release_gateway_runtime_lock`` no-ops when the lock is already + released — so this is a no-op on the normal shutdown path and the actual + cleanup on the early-exit paths. + + Logging is not flushed here: the gateway's handlers are synchronous + ``RotatingFileHandler``s that write each record immediately (no + ``MemoryHandler``/``QueueHandler`` buffering), so there is nothing pending. + Only stdio is buffered, so only stdio is flushed. + """ for stream in (sys.stdout, sys.stderr): try: stream.flush() except Exception: pass - os._exit(0 if success else 1) + # Guaranteed cleanup chokepoint: os._exit skips atexit, and the early + # SystemExit exit paths never run _stop_impl, so release here (idempotent). + try: + from gateway.status import remove_pid_file, release_gateway_runtime_lock + remove_pid_file() + release_gateway_runtime_lock() + except Exception: + pass + os._exit(exit_code) if __name__ == "__main__": diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 7a62889d747c..f735b70a15dc 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -33,11 +33,7 @@ from agent.i18n import t from gateway.config import HomeChannel, Platform, PlatformConfig from gateway.platforms.base import EphemeralReply, MessageEvent, MessageType -from gateway.session import ( - SessionSource, - build_session_key, - is_shared_multi_user_session, -) +from gateway.session import SessionSource, build_session_key from hermes_cli.config import cfg_get, clear_model_endpoint_credentials from utils import ( atomic_json_write, @@ -666,269 +662,8 @@ def _same_matrix_room(current: SessionSource, origin: Optional[SessionSource]) - and origin.platform == Platform.MATRIX and current.platform == Platform.MATRIX and origin.chat_id == current.chat_id - # thread_id is part of the session key (build_session_key appends it - # for every chat type when present), and Matrix scopes the model's - # turn to the current room/thread. A live session in another thread - # of the SAME room is a DIFFERENT session, so a caller in thread A - # must not resume/enumerate a target whose origin is in thread B. - # Non-threaded rooms have empty thread_id on both sides ("" == ""), - # so room-level sharing is preserved unchanged. - and str(getattr(current, "thread_id", "") or "") - == str(getattr(origin, "thread_id", "") or "") ) - def _same_origin_chat(self, current: SessionSource, origin: Optional[SessionSource]) -> bool: - """Platform-agnostic counterpart to ``_same_matrix_room``. - - True when *origin* shares *current*'s platform and chat, and the same - participant whenever the session key for this source is per-user. Group - and thread sessions that ``build_session_key`` isolates per participant - (the default ``group_sessions_per_user=True``) must also be scoped by - participant here — otherwise a co-member could resume another member's - live per-user group session (IDOR). Only an explicitly shared - group/thread (``group_sessions_per_user=False`` / - ``thread_sessions_per_user``) lets co-members share, mirroring the key - contract via ``is_shared_multi_user_session``. - """ - if origin is None or current is None: - return False - if origin.platform != current.platform: - return False - if origin.chat_id != current.chat_id: - return False - # thread_id is part of the session key for every chat type when present - # (build_session_key appends it unconditionally), so a session in one - # thread is a DIFFERENT session from another thread of the same parent - # chat. is_shared_multi_user_session only decides participant sharing - # WITHIN a thread, never across threads — require thread equality before - # any sharing logic so a live origin in thread A cannot match a caller in - # thread B of the same parent chat. - if str(getattr(current, "thread_id", "") or "") != str( - getattr(origin, "thread_id", "") or "" - ): - return False - chat_type = (getattr(current, "chat_type", "") or "").lower() - # DM-like chats are always per-user. - if chat_type in {"dm", "direct", "private", ""}: - # chat_id was already required equal above and, when present, IS the - # DM session key — so an equal non-empty chat_id is sufficient. - # build_session_key only falls back to the participant id - # (``user_id_alt or user_id`` — Signal/Feishu key on user_id_alt) - # when there is NO chat_id; mirror that and fail closed on a - # missing/different participant so two no-chat_id DM origins are - # never conflated (was: compared user_id only and allowed when - # either side was missing). - if str(getattr(current, "chat_id", "") or ""): - return True - cur_pid = str(current.user_id_alt or current.user_id or "") - org_pid = str(origin.user_id_alt or origin.user_id or "") - return bool(cur_pid) and cur_pid == org_pid - # Non-DM: scope by participant whenever the session key for this source - # is per-user. is_shared_multi_user_session mirrors build_session_key's - # isolation rules exactly, so the guard stays in lock-step with the key. - shared = is_shared_multi_user_session( - current, - group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True), - thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False), - ) - if shared: - return True - # Per-user key: compare the participant id the key is actually built - # from (user_id_alt or user_id — Signal/Feishu key on user_id_alt). - cur_pid = current.user_id_alt or current.user_id - org_pid = origin.user_id_alt or origin.user_id - if cur_pid and org_pid: - return cur_pid == org_pid - # Per-user key but a participant id is missing on one side: cannot prove - # the same owner — fail closed. - return False - - def _resume_caller_is_admin(self, source: SessionSource) -> bool: - """Whether *source* is an EXPLICITLY-configured admin allowed to make a - cross-origin /resume or /sessions listing. - - Deliberately stricter than ``SlashAccessPolicy.is_admin()``: that returns - True for every allowed caller when slash gating is DISABLED (so commands - stay runnable by default), but cross-ORIGIN DATA ACCESS must require a - real, configured admin. Otherwise the default (no admin list) config - would treat every gateway caller as cross-origin-capable and re-open the - enumeration IDOR. - """ - try: - from gateway.slash_access import policy_for_source - policy = policy_for_source(self.config, source) - uid = getattr(source, "user_id", None) - return bool(policy.enabled and uid and policy.is_admin(uid)) - except Exception: - return False - - async def _resume_target_allowed( - self, source: SessionSource, target_id: str, allow_override: bool = False - ) -> bool: - """Whether *source* may resume the persisted session *target_id*. - - Generalizes the Matrix-only room guard to every adapter so a caller - cannot bind their gateway session to another user's/room's persisted - session id (IDOR). Uses the live origin when the target is active; - otherwise falls back to the DB row's source + user_id (the sessions - table has no chat_id). An identity-bearing caller is allowed only when - the row PROVES the same owner; a row that lacks enough ownership data - fails closed. An explicit admin ``--all`` override bypasses scoping. - """ - if allow_override and self._resume_caller_is_admin(source): - return True - # Use the live origin only when it resolves to a real SessionSource; a - # store that can't resolve it (or an unexpected lookup error) must not - # silently allow/deny — fall through to the deterministic DB scoping. - try: - origin = self._gateway_session_origin_for_id(target_id) - except Exception: - origin = None - if isinstance(origin, SessionSource): - return self._same_origin_chat(source, origin) - # Inactive/persisted-only: best-effort scope by DB row source + user. - try: - row = await self._session_db.get_session(target_id) or {} - except Exception: - return False - caller_src = source.platform.value if source.platform else None - row_src = row.get("source") - if row_src and caller_src and str(row_src) != str(caller_src): - return False # different platform / source - caller_uid = str(getattr(source, "user_id", "") or "") - row_uid = str(row.get("user_id") or "") - # Chat/thread origin recorded at session creation (see - # SessionDB._insert_session_row). The sessions table historically stored - # only source + user_id, so a same-user row could belong to a DIFFERENT - # chat; comparing the persisted origin closes that gap. Legacy rows - # created before origin capture have NULL here and therefore fail closed - # (they cannot prove the caller's chat) — resume them via a live session - # or an admin override. - caller_chat = str(getattr(source, "chat_id", "") or "") - row_chat = str(row.get("chat_id") or "") - caller_thread = str(getattr(source, "thread_id", "") or "") - row_thread = str(row.get("thread_id") or "") - chat_type = (getattr(source, "chat_type", "") or "").lower() - caller_is_dm = chat_type in {"dm", "direct", "private", ""} - # build_session_key keys the participant on ``user_id_alt or user_id`` - # (Signal/Feishu carry the canonical participant in user_id_alt), but the - # sessions table only ever stored user_id — it has no user_id_alt column. - # So when the caller carries a user_id_alt, the row CANNOT prove the - # canonical participant that the live session key is built from: two - # members sharing one user_id but different user_id_alt map to DIFFERENT - # session keys, yet the persisted row's user_id would match both. The - # live-origin guard (_same_origin_chat) compares user_id_alt correctly; - # the persisted fallback cannot, so any per-user comparison that would - # otherwise rely on row_uid == caller_uid must fail closed here to stay - # in lock-step with the key boundary (CWE-639). Shared group/thread - # sessions are unaffected (they don't scope by participant at all), and - # an admin --all override still bypasses this above. - caller_keys_on_alt = bool(str(getattr(source, "user_id_alt", "") or "")) - if caller_uid: - # Identity-bearing caller: allow only when the row PROVES the same - # owner AND the same platform/origin AND the same chat/thread. A row - # with no/blank user_id cannot be proven to belong to this caller; a - # row with no/blank source cannot be proven to share the caller's - # platform (the row_src check above only rejects a *mismatching* - # non-blank source, so a blank/legacy source would otherwise slip - # through on user_id equality alone); and a row whose origin chat - # (or thread) differs from the caller's belongs to a different - # conversation. Any gap fails closed — an identified user must not - # bind to an unowned, other-owned, other-chat, or unproven-origin - # persisted session by id/title. (Legacy NULL-owner/blank-source/ - # NULL-chat rows are intentionally not resumable this way; use a - # live session or an explicit admin override.) - # Common origin proof for any identity-bearing caller: a non-blank - # source that matches the caller's platform, and the same thread. A - # blank/legacy source can't prove the platform; a different thread is - # a different session (build_session_key appends thread_id). - origin_ok = ( - bool(row_src) and bool(caller_src) - and str(row_src) == str(caller_src) - and row_thread == caller_thread - ) - if not origin_ok: - return False - if caller_is_dm: - # DMs are keyed on user_id; require the same owner. chat_id is - # legitimately absent on both sides for a no-chat_id DM (scoped - # by user_id), but a mismatching chat_id (when present) is still - # rejected. - # - # A no-chat_id DM is keyed PURELY on the participant - # (``user_id_alt or user_id``). If the caller keys on user_id_alt - # the persisted row (user_id only) cannot prove that participant, - # so fail closed. When chat_id is present on both sides it is the - # DM key and equal chat_id is sufficient, so the alt gap doesn't - # apply there. - if caller_keys_on_alt and not (bool(row_chat) and bool(caller_chat)): - return False - return ( - bool(row_uid) and row_uid == caller_uid - and row_chat == caller_chat - ) - # Non-DM (group/channel/forum/thread): build_session_key includes - # chat_id, so a row (or caller) with NO chat provenance cannot prove - # same-chat. Require both sides non-blank and equal — a legacy - # NULL-chat row (or a caller missing its chat_id) fails closed even - # when both normalize to "". (CWE-639) - if not (bool(row_chat) and bool(caller_chat) and row_chat == caller_chat): - return False - # Within the same non-DM chat/thread, mirror build_session_key's - # participant scoping: a SHARED group/thread session - # (group_sessions_per_user=False, or a shared thread) is one session - # for every participant, so the same-chat proof above is sufficient — - # do NOT also require user-id equality (otherwise a co-member is - # wrongly blocked from their own shared session). A per-user session - # still requires the same owner. - shared = is_shared_multi_user_session( - source, - group_sessions_per_user=getattr(self.config, "group_sessions_per_user", True), - thread_sessions_per_user=getattr(self.config, "thread_sessions_per_user", False), - ) - if shared: - return True - # Per-user non-DM: the session key includes the participant - # (``user_id_alt or user_id``). If the caller keys on user_id_alt, - # the persisted row (user_id only) cannot prove the canonical - # participant, so fail closed rather than matching on user_id alone. - if caller_keys_on_alt: - return False - return bool(row_uid) and row_uid == caller_uid - # No caller identity: the persisted row carries only source + user_id - # (the sessions table has no chat_id), so a same-platform row can belong - # to a DIFFERENT chat or user. Same-platform alone is therefore NOT - # ownership proof — an identity-less caller must not bind to, or - # enumerate, a persisted session by id/title. Fail closed. A legitimate - # same-chat resume of an ACTIVE session still works through the - # live-origin branch above (which compares chat_id), and an operator can - # use the admin --all override. (CWE-639: IDOR on session routing.) - return False - - async def _resume_row_visible( - self, source: SessionSource, row: dict, allow_all: bool - ) -> bool: - """Whether a titled-session listing *row* belongs to the caller's origin. - - Prevents cross-origin enumeration of session ids/previews via the - numbered /resume list. Preserves the existing Matrix room-scoping - semantics; scopes every other platform to the caller's own sessions - unless an admin passes ``--all``. - """ - sid = str(row.get("id") or "") - if source.platform == Platform.MATRIX: - # Cross-room enumeration is cross-ORIGIN data access: gate the - # ``--all`` short-circuit behind a real configured admin, exactly - # like the non-Matrix branch below. A non-admin Matrix ``--all`` - # falls back to same-room scoping rather than exposing every Matrix - # titled session. - if allow_all and self._resume_caller_is_admin(source): - return True - return self._same_matrix_room(source, self._gateway_session_origin_for_id(sid)) - if allow_all and self._resume_caller_is_admin(source): - return True - return await self._resume_target_allowed(source, sid, allow_override=False) - async def _handle_agents_command(self, event: MessageEvent) -> str: """Handle /agents command - list active agents and running tasks.""" from gateway.run import _AGENT_PENDING_SENTINEL @@ -3345,12 +3080,6 @@ async def _handle_title_command(self, event: MessageEvent) -> str: session_id=session_id, source=source.platform.value if source.platform else "unknown", user_id=source.user_id, - # Persist the messaging origin so a later /resume of this - # titled-but-now-inactive session can prove it belongs to the - # caller's chat/thread (IDOR scoping). - chat_id=source.chat_id, - chat_type=source.chat_type, - thread_id=source.thread_id, ) except Exception: pass # Session might already exist, ignore errors @@ -3433,10 +3162,13 @@ async def _list_titled_sessions() -> list[dict]: # List recent titled sessions for this user/platform try: titled = await _list_titled_sessions() - titled = [ - s for s in titled - if await self._resume_row_visible(source, s, allow_all) - ] + if source.platform == Platform.MATRIX and not allow_all: + scoped = [] + for s in titled: + origin = self._gateway_session_origin_for_id(str(s.get("id") or "")) + if self._same_matrix_room(source, origin): + scoped.append(s) + titled = scoped if not titled: if source.platform == Platform.MATRIX and not allow_all: return t("gateway.resume.matrix_no_named_sessions") @@ -3461,10 +3193,13 @@ async def _list_titled_sessions() -> list[dict]: if name.isdigit(): try: titled = await _list_titled_sessions() - titled = [ - s for s in titled - if await self._resume_row_visible(source, s, allow_all) - ] + if source.platform == Platform.MATRIX and not allow_all: + scoped = [] + for s in titled: + origin = self._gateway_session_origin_for_id(str(s.get("id") or "")) + if self._same_matrix_room(source, origin): + scoped.append(s) + titled = scoped except Exception as e: logger.debug("Failed to list titled sessions for numeric resume: %s", e) return t("gateway.resume.list_failed", error=e) @@ -3501,14 +3236,6 @@ async def _list_titled_sessions() -> list[dict]: room=target_origin.chat_name or target_origin.chat_id, name=name, ) - elif not await self._resume_target_allowed( - source, target_id, allow_override=(allow_all or allow_cross_room) - ): - # IDOR guard: a session id/title is a routing handle, not authority. - # Bind /resume to the caller's own platform/user/chat on every - # non-Matrix adapter so one user can't attach to another's - # persisted transcript. - return t("gateway.resume.blocked_not_owner", name=name) # Check if already on that session current_entry = self.session_store.get_or_create_session(source) @@ -3589,33 +3316,27 @@ async def _handle_sessions_command(self, event: MessageEvent) -> str: resume_event = dataclasses.replace(event, text=f"/resume {target}") return await self._handle_resume_command(resume_event) - # A cross-origin listing (`/sessions all`) is honored only for an - # admin, mirroring the `/resume --all` override. `all` is just a parsed - # user argument, so without this gate any caller could run - # `/sessions all` and enumerate other origins' session ids / titles / - # previews / sources — the enumeration half of the /resume IDOR. - cross_origin = include_all and self._resume_caller_is_admin(source) current_entry = self.session_store.get_or_create_session(source) rows = await asyncio.to_thread( query_session_listing, getattr(self._session_db, "_db", self._session_db), source=source.platform.value if source.platform else None, current_session_id=current_entry.session_id, - include_all_sources=cross_origin, + include_all_sources=include_all, include_unnamed=include_unnamed, limit=10, exclude_sources=["tool"], ) - if not cross_origin: - # Scope the listing to the caller's own origin on every adapter so - # session ids/previews from other users/rooms aren't enumerable. + if source.platform == Platform.MATRIX and not include_all: rows = [ row for row in rows - if await self._resume_row_visible(source, row, allow_all=False) + if self._same_matrix_room( + source, self._gateway_session_origin_for_id(str(row.get("id") or "")) + ) ] return format_gateway_session_listing( rows, - include_source=cross_origin, + include_source=include_all, title="Sessions" if include_unnamed else "Named Sessions", ) @@ -4384,7 +4105,7 @@ async def _handle_update_command(self, event: MessageEvent) -> str: if event.message_id: pending["message_id"] = event.message_id _tmp_pending = pending_path.with_suffix(".tmp") - _tmp_pending.write_text(json.dumps(pending)) + _tmp_pending.write_text(json.dumps(pending), encoding="utf-8") _tmp_pending.replace(pending_path) exit_code_path.unlink(missing_ok=True) diff --git a/hermes_cli/banner.py b/hermes_cli/banner.py index 217eb2bb9656..862a1df36c43 100644 --- a/hermes_cli/banner.py +++ b/hermes_cli/banner.py @@ -363,7 +363,7 @@ def check_for_updates() -> Optional[int]: try: cache_file.write_text( json.dumps({"ts": now, "behind": behind, "rev": embedded_rev, "ver": VERSION}) - ) +, encoding="utf-8") except Exception: pass diff --git a/hermes_cli/container_boot.py b/hermes_cli/container_boot.py index 327e6a5ce7f0..3a06c963245e 100644 --- a/hermes_cli/container_boot.py +++ b/hermes_cli/container_boot.py @@ -224,7 +224,7 @@ def _maybe_migrate_legacy_gateway_run_state( "desired_state": "running", "timestamp": int(time.time()), "migrated_from": "legacy-container-cmd", - }) + "\n") + }) + "\n", encoding="utf-8") return "running" @@ -438,7 +438,7 @@ def _register_service(scandir: Path, profile: str, *, start: bool) -> None: tmp_dir.mkdir(parents=True) try: - (tmp_dir / "type").write_text("longrun\n") + (tmp_dir / "type").write_text("longrun\n", encoding="utf-8") # Reuse the manager's run-script rendering — single source of # truth so register_profile_gateway and reconcile_profile_gateways @@ -446,18 +446,18 @@ def _register_service(scandir: Path, profile: str, *, start: bool) -> None: # per-profile env can set it via the profile's config.yaml # (which the gateway itself loads). run = tmp_dir / "run" - run.write_text(S6ServiceManager._render_run_script(profile, extra_env={})) + run.write_text(S6ServiceManager._render_run_script(profile, extra_env={}), encoding="utf-8") run.chmod(0o755) finish = tmp_dir / "finish" - finish.write_text(S6ServiceManager._render_finish_script()) + finish.write_text(S6ServiceManager._render_finish_script(), encoding="utf-8") finish.chmod(0o755) # Persistent log rotation (OQ8-C). log_subdir = tmp_dir / "log" log_subdir.mkdir() log_run = log_subdir / "run" - log_run.write_text(S6ServiceManager._render_log_run(profile)) + log_run.write_text(S6ServiceManager._render_log_run(profile), encoding="utf-8") log_run.chmod(0o755) # The presence of a `down` file tells s6-supervise to NOT diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 10b638d8e834..2df45f984860 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -3588,6 +3588,86 @@ def _launchctl_bootstrap( ) +def _launchd_reload_log_path() -> Path: + """Path the launchd reload watchdog tails for persistent-orphan detection.""" + return get_hermes_home() / "logs" / "launchd-reload.log" + + +def _append_launchd_reload_log(message: str) -> None: + """Append a timestamped line to the launchd reload log (best-effort).""" + path = _launchd_reload_log_path() + try: + path.parent.mkdir(parents=True, exist_ok=True) + from datetime import datetime as _dt + + stamp = _dt.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %z") + with path.open("a", encoding="utf-8") as fh: + fh.write(f"[{stamp}] {message}\n") + except OSError: + pass + + +def _launchctl_label_registered(label: str) -> bool: + """True when ``launchctl list