From 46c1ca793e177415ef1f489ebfa9b09e09cd3faf Mon Sep 17 00:00:00 2001 From: pierrenode <298902573+pierrenode@users.noreply.github.com> Date: Tue, 18 Aug 2026 04:17:02 +0300 Subject: [PATCH] fix(peer): route hermes peer dm through the credential-redirect guard hermes_cli/subcommands/peer.py's _request() sends the peer's Authorization: Bearer via a raw urllib.request.urlopen() call. Python's default HTTPRedirectHandler preserves all request headers across a 3xx redirect, including Authorization, even when the redirect crosses origins. A compromised peer gateway (or a LAN MITM answering the URL registered with `hermes peer add`) can redirect to an attacker- controlled host and harvest the peer key -- which bots also send autonomously via `hermes peer dm`, per the Bot Mode messaging protocol injected by tools/bot_mode_probe.py. This is the same credential-redirect-leak class this repo has closed repeatedly elsewhere (providers/base.py, hermes_cli/models.py, azure_detect.py, the anthropic adapter, plugins/model-providers/actual) via hermes_cli/urllib_security.py::open_credentialed_url(), which strips non-safelisted headers whenever a redirect crosses origin. peer.py never adopted it. Route _request() through open_credentialed_url() -- a drop-in swap for urlopen() since it accepts the same pre-built Request object. Added a regression test using two real loopback HTTP servers (the peer and a stand-in attacker origin): a peer that 302-redirects every request must not leak the Bearer key to the redirect target. Mutation-verified: reverting the fix makes the new test fail with the key present at the attacker origin. --- hermes_cli/subcommands/peer.py | 8 +++- tests/hermes_cli/test_peer_cmd.py | 70 +++++++++++++++++++++++++++++++ 2 files changed, 77 insertions(+), 1 deletion(-) diff --git a/hermes_cli/subcommands/peer.py b/hermes_cli/subcommands/peer.py index 13d10637a6a0e..ca57c219291a6 100644 --- a/hermes_cli/subcommands/peer.py +++ b/hermes_cli/subcommands/peer.py @@ -77,6 +77,8 @@ def _peer_secret(name: str) -> str: def _request(url: str, key: str, *, method: str = "GET", body: dict | None = None, timeout: int = LIST_TIMEOUT_S) -> dict: + from hermes_cli.urllib_security import open_credentialed_url + data = json.dumps(body).encode("utf-8") if body is not None else None req = urllib.request.Request( url, @@ -88,7 +90,11 @@ def _request(url: str, key: str, *, method: str = "GET", body: dict | None = Non "User-Agent": "hermes-peer-dm", }, ) - with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 — user-registered peer URL + # The peer URL is user-registered (``hermes peer add``); a redirect to a + # different origin must not carry the Authorization: Bearer key with it — + # a compromised/MITM'd peer could otherwise harvest it. open_credentialed_url + # strips non-safelisted headers across a cross-origin redirect. + with open_credentialed_url(req, timeout=timeout) as resp: payload = resp.read().decode("utf-8", "replace") try: parsed = json.loads(payload) diff --git a/tests/hermes_cli/test_peer_cmd.py b/tests/hermes_cli/test_peer_cmd.py index a3e4d15ae37e3..9b33fff8188f4 100644 --- a/tests/hermes_cli/test_peer_cmd.py +++ b/tests/hermes_cli/test_peer_cmd.py @@ -186,3 +186,73 @@ def test_dm_reuses_existing_bot_chat(monkeypatch, capsys, fake_peer_server): assert payload["reply"] == "reply from the other machine" # No new session was created — the existing canonical chat was reused. assert _FakePeer.sessions == ["bc_existing"] + + +# ── cross-origin redirect must not carry the peer's Bearer key ────────────── + + +class _AttackerOrigin(BaseHTTPRequestHandler): + """A second real HTTP server standing in for an attacker-controlled host + a compromised/MITM'd peer could redirect a ``hermes peer dm`` request to.""" + + auth_seen: list = [] + + def do_GET(self): + type(self).auth_seen.append(self.headers.get("Authorization")) + body = json.dumps({"object": "list", "data": []}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def log_message(self, *args): # noqa: D102 — silence test server logging + pass + + +class _RedirectingPeer(BaseHTTPRequestHandler): + """A "peer" that 302-redirects every request to a different origin — + the shape of a compromised peer or a LAN MITM answering ``hermes peer + add``'s registered URL.""" + + redirect_target: str = "" + + def do_GET(self): + self.send_response(302) + self.send_header("Location", type(self).redirect_target + self.path) + self.end_headers() + + def log_message(self, *args): # noqa: D102 — silence test server logging + pass + + +def test_request_strips_bearer_key_across_redirect_origin(): + """``_request`` must not forward the peer's Authorization: Bearer key to + a different origin a redirect points at (compromised peer / LAN MITM) — + the exact class of leak ``open_credentialed_url`` exists to close.""" + _AttackerOrigin.auth_seen = [] + attacker = HTTPServer(("127.0.0.1", 0), _AttackerOrigin) + attacker_thread = threading.Thread(target=attacker.serve_forever, daemon=True) + attacker_thread.start() + + _RedirectingPeer.redirect_target = f"http://127.0.0.1:{attacker.server_port}" + peer = HTTPServer(("127.0.0.1", 0), _RedirectingPeer) + peer_thread = threading.Thread(target=peer.serve_forever, daemon=True) + peer_thread.start() + + try: + # The attacker origin answers with a well-formed (empty) listing, so + # the redirect completes successfully — the request itself is not + # the point of this test, only whether the Bearer key rode along. + result = peer_cmd._request(f"http://127.0.0.1:{peer.server_port}/api/sessions", "top-secret-peer-key") + assert result == {"object": "list", "data": []} + finally: + peer.shutdown() + peer_thread.join(timeout=5) + attacker.shutdown() + attacker_thread.join(timeout=5) + + assert _AttackerOrigin.auth_seen, "redirect target was never reached" + assert all(header is None for header in _AttackerOrigin.auth_seen), ( + f"peer's Bearer key leaked to the redirect target: {_AttackerOrigin.auth_seen}" + )