fix(code-exec): bind execute_code tool socket to a per-session RPC token

The execute_code sandbox exposed its tool-call RPC (AF_UNIX socket and
remote file-poll transports) without any caller check, so any local
process that could reach the socket / rpc dir could dispatch
terminal-capable tool calls through the parent. Mint a per-session
HERMES_RPC_TOKEN, pass it to the sandboxed child, and require a
timing-safe match on every request in both _rpc_server_loop and
_rpc_poll_loop. Empty/missing/wrong token fails closed.

Salvaged from #44073 (per-session RPC token). Added timing-safe
secrets.compare_digest comparison and fail-closed regression tests.

Co-authored-by: Hermes Agent <agent@nousresearch.com>

tests/tools/test_code_execution.py

@@ -17,6 +17,7 @@ import pytest
 
 import json
 import os
+import socket
 import time
 
 os.environ["TERMINAL_ENV"] = "local"
@@ -1006,5 +1007,131 @@ for i in range(15000):
             self.assertIn("total", output)
 
 
+class TestRpcTokenAuthorization(unittest.TestCase):
+    """The per-session RPC token must gate socket dispatch (fail-closed).
+
+    Regression coverage for the execute_code tool-socket hardening: a
+    request without the matching HERMES_RPC_TOKEN must be rejected before
+    the tool is dispatched, while a request carrying the correct token
+    round-trips normally.
+    """
+
+    def _drive_server(self, rpc_token, requests):
+        """Run _rpc_server_loop against a real AF_UNIX socketpair.
+
+        Sends each dict in *requests* as a newline-delimited JSON message
+        and returns the list of decoded JSON responses.
+        """
+        from tools.code_execution_tool import _rpc_server_loop
+
+        # socketpair gives us a connected client end and a "server" end we
+        # can hand to accept() by wrapping it in a tiny listener shim.
+        srv, cli = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
+
+        class _OneShotListener:
+            """Minimal object exposing the .accept()/.settimeout() the loop uses."""
+
+            def __init__(self, conn):
+                self._conn = conn
+                self._served = False
+
+            def settimeout(self, _t):
+                pass
+
+            def accept(self):
+                if self._served:
+                    raise socket.timeout()
+                self._served = True
+                return self._conn, ("peer", 0)
+
+        listener = _OneShotListener(srv)
+        stop_event = threading.Event()
+        tool_call_log = []
+        tool_call_counter = [0]
+
+        def _run():
+            with patch(
+                "model_tools.handle_function_call",
+                side_effect=_mock_handle_function_call,
+            ):
+                _rpc_server_loop(
+                    listener,
+                    "test-task",
+                    tool_call_log,
+                    tool_call_counter,
+                    max_tool_calls=10,
+                    allowed_tools=frozenset({"terminal"}),
+                    stop_event=stop_event,
+                    rpc_token=rpc_token,
+                )
+
+        t = threading.Thread(target=_run, daemon=True)
+        t.start()
+
+        responses = []
+        try:
+            for req in requests:
+                cli.sendall((json.dumps(req) + "\n").encode())
+            cli.settimeout(5)
+            buf = b""
+            while len(responses) < len(requests):
+                chunk = cli.recv(65536)
+                if not chunk:
+                    break
+                buf += chunk
+                while b"\n" in buf:
+                    line, buf = buf.split(b"\n", 1)
+                    line = line.strip()
+                    if line:
+                        responses.append(json.loads(line.decode()))
+        finally:
+            stop_event.set()
+            cli.close()
+            srv.close()
+            t.join(timeout=5)
+        return responses
+
+    def test_missing_token_rejected(self):
+        """A request with no token is rejected as Unauthorized."""
+        resp = self._drive_server(
+            "secret-token", [{"tool": "terminal", "args": {"command": "echo hi"}}]
+        )
+        self.assertEqual(len(resp), 1)
+        self.assertIn("Unauthorized", resp[0].get("error", ""))
+
+    def test_wrong_token_rejected(self):
+        """A request with a mismatched token is rejected as Unauthorized."""
+        resp = self._drive_server(
+            "secret-token",
+            [{"tool": "terminal", "args": {"command": "echo hi"}, "token": "nope"}],
+        )
+        self.assertEqual(len(resp), 1)
+        self.assertIn("Unauthorized", resp[0].get("error", ""))
+
+    def test_matching_token_dispatched(self):
+        """A request carrying the correct token round-trips to the tool."""
+        resp = self._drive_server(
+            "secret-token",
+            [{"tool": "terminal", "args": {"command": "echo hi"}, "token": "secret-token"}],
+        )
+        self.assertEqual(len(resp), 1)
+        self.assertNotIn("Unauthorized", json.dumps(resp[0]))
+        self.assertIn("mock output for: echo hi", json.dumps(resp[0]))
+
+    def test_empty_server_token_fails_closed(self):
+        """An empty server-side token rejects everything (fail-closed)."""
+        resp = self._drive_server(
+            "", [{"tool": "terminal", "args": {"command": "echo hi"}, "token": ""}]
+        )
+        self.assertEqual(len(resp), 1)
+        self.assertIn("Unauthorized", resp[0].get("error", ""))
+
+    def test_generated_module_sends_token(self):
+        """The generated hermes_tools module reads HERMES_RPC_TOKEN and sends it."""
+        src = generate_hermes_tools_module(["terminal"], transport="uds")
+        self.assertIn("HERMES_RPC_TOKEN", src)
+        self.assertIn('"token"', src)
+
+
 if __name__ == "__main__":
     unittest.main()