fix(gateway): refuse relay sender fields from a logged-in client and say what the author trusts

bot_relay.deliver accepted from_profile, from_handle and from_connection from any admitted JSON-RPC client. The handler now refuses them with error 4095 when the calling transport carries a browser login identity, since a logged-in browser never relays for another connection. The DeliveryAuthor docstring now says the author is trusted because an admitted client relays it, not because the sender is verified.

tests/tui_gateway/test_bot_relay_methods.py

@@ -16,6 +16,7 @@ import json
 import pytest
 
 import tui_gateway.server as srv
+from hermes_cli.dashboard_auth.ws_tickets import INTERNAL_PROVIDER, INTERNAL_USER_ID
 from tools import bot_relay
 
 
@@ -235,3 +236,62 @@ def test_deliver_child_env_carries_the_envelope_sender_on_every_attempt(home, mo
     assert len(envs) == 2
     assert [json.loads(e[TURN_AUTHOR_ENV]) if TURN_AUTHOR_ENV in e else None for e in envs] == [expected, expected]
     assert all(e["HERMES_RELAY_TEST_MARKER"] == "kept" for e in envs)
+
+
+class _Client:
+    def __init__(self, auth_identity=None):
+        self.auth_identity = auth_identity
+
+    def write(self, obj):
+        return True
+
+    def close(self):
+        return None
+
+
+@pytest.fixture
+def bound_client(monkeypatch):
+    """Bind a fake calling transport for the handler; yields a setter for its ``auth_identity``."""
+    client = _Client()
+    token = srv.bind_transport(client)
+    try:
+        yield client
+    finally:
+        srv.reset_transport(token)
+
+
+SENDER = {"from_profile": "scout", "from_handle": "scout", "from_connection": "cloud-1"}
+SENDER_AUTHOR = {"id": "bot:cloud-1/scout", "name": "scout", "is_bot": True}
+
+
+@pytest.mark.parametrize("identity", [
+    None,
+    {"user_id": INTERNAL_USER_ID, "provider": INTERNAL_PROVIDER},
+], ids=["no identity", "server-internal identity"])
+def test_deliver_accepts_a_sender_from_an_admitted_non_login_client(home, fake_runs, bound_client, identity):
+    """The Desktop and server-internal callers carry no login identity; their sender fields become the author."""
+    from agent.turn_author import TURN_AUTHOR_ENV
+
+    calls, _outcomes = fake_runs
+    bound_client.auth_identity = identity
+
+    _result(srv._methods["bot_relay.deliver"](1, {"profile": "ops", "message": "ping", **SENDER}))
+
+    assert [json.loads(c["env"][TURN_AUTHOR_ENV]) for c in calls] == [SENDER_AUTHOR]
+
+
+def test_deliver_refuses_a_sender_from_a_logged_in_client(home, fake_runs, bound_client):
+    """A browser login never relays for another connection, so its from_* fields are refused before any turn runs.
+    Without sender fields the same client still delivers, unattributed."""
+    from agent.turn_author import TURN_AUTHOR_ENV
+
+    calls, _outcomes = fake_runs
+    bound_client.auth_identity = {"user_id": "alice", "provider": "google"}
+
+    for sender in ({"from_profile": "scout"}, {"from_connection": "cloud-1"}, SENDER):
+        err = srv._methods["bot_relay.deliver"](1, {"profile": "ops", "message": "ping", **sender})
+        assert err["error"]["code"] == 4095
+    assert not calls
+
+    _result(srv._methods["bot_relay.deliver"](2, {"profile": "ops", "message": "ping"}))
+    assert len(calls) == 1 and TURN_AUTHOR_ENV not in calls[0]["env"]

tools/bot_relay.py

@@ -385,8 +385,9 @@ def local_delivery_command(profile: str, query_file: str) -> list[str]:
 
 
 class DeliveryAuthor:
-    """A relayed turn's author as an in-process object. A JSON client cannot build one, so
-    ``prompt.submit`` trusts it the way it trusts a hosted-room callback."""
+    """A relayed turn's author as an in-process object. ``bot_relay.deliver`` builds it from the sender fields
+    an admitted gateway client relays for another connection; nothing verifies the sender itself. A JSON
+    client cannot build one, so ``prompt.submit`` accepts the object and refuses a dict."""
 
     __slots__ = ("author",)
 
@@ -401,9 +402,10 @@ class DeliveryAuthor:
 
 
 def delivery_turn_author(from_profile: Any, from_handle: Any, from_connection: Any = None) -> Optional[dict]:
-    """The author of a relayed DM's recipient turn, built from the envelope's sender fields. A relayed DM always
-    comes from another gateway, so the id carries the Desktop's id for the sender's connection (``local`` included)
-    and only the recipient's own profiles are bare ``bot:<profile>``. None when the envelope names no sender."""
+    """The author of a relayed DM's recipient turn, built from the sender fields as the relaying client reports
+    them. A relayed DM always comes from another gateway, so the id carries the Desktop's id for the sender's
+    connection (``local`` included) and only the recipient's own profiles are bare ``bot:<profile>``. None when
+    the envelope names no sender."""
     from agent.turn_author import bot_author_id
 
     profile = str(from_profile or "").strip()

tui_gateway/methods_bot_relay.py

@@ -90,9 +90,15 @@ def _(rid, params: dict, _root=_relay_root, _run=_run_delivery) -> dict:
             if isinstance(record, dict) and (record.get("profile_home") or None) == want_home
             and _session_live_title(
                 record, _session_lookup_key(record, fallback=live_sid)) == BOT_CHAT_TITLE), "")
-        # The Desktop forwards the envelope's sender. The author labels memory only and grants nothing.
+        # The sender fields are whatever the relaying client says. The author labels memory only and grants nothing.
         from tools.bot_relay import DeliveryAuthor, delivery_env, delivery_turn_author
-        author = delivery_turn_author(params.get("from_profile"), params.get("from_handle"), params.get("from_connection"))
+        from tui_gateway.methods_browser_control import _is_authenticated_identity
+        sender_fields = ("from_profile", "from_handle", "from_connection")
+        # A logged-in browser never relays for another connection; only the Desktop and server-internal callers do.
+        if (any(params.get(k) for k in sender_fields)
+                and _is_authenticated_identity(getattr(current_transport(), "auth_identity", None))):
+            return _err(rid, 4095, "a logged-in client cannot name the sender of a relayed dm")
+        author = delivery_turn_author(*(params.get(k) for k in sender_fields))
         if live_sid:
             # queued=True: a teammate's DM runs as the NEXT turn and never interrupts or steers a
             # turn in flight (the default busy mode does); arrivals queue in order.