From bab6d6b18473877a199eeb8f447acee9c083aac6 Mon Sep 17 00:00:00 2001
From: Andre Merzky <andre@merzky.net>
Date: Tue, 25 Aug 2026 17:19:04 +0200
Subject: [PATCH] broker host: header lookup is case-insensitive, like HTTP

The gateway path goes through starlette, which lowercases header names;
an in-process caller (BrokerCaller) hands its dict to handle_request
verbatim.  A client sending 'Content-Type: application/msgpack' missed
the lowercase lookup, the body fell through to the JSON default, and
json.loads died on the first msgpack byte as invalid utf-8 -- the
task dispatcher's rhapsody-dialect bulk submit hit exactly this from
the DT service's pool-backed engine (first live pool run).

Headers are still not passed into the RequestShim: a client-supplied
x-orbit-src must not surface as a trusted owner on this path.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_016Npyz3Hbnwos12ESsdJ2YU
---
 src/radical/orbit/broker_plugin_host.py     | 10 ++++
 tests/unittests/test_broker_host_headers.py | 56 +++++++++++++++++++++
 2 files changed, 66 insertions(+)
 create mode 100644 tests/unittests/test_broker_host_headers.py

diff --git a/src/radical/orbit/broker_plugin_host.py b/src/radical/orbit/broker_plugin_host.py
index 4ec167b3..6a42364d 100644
--- a/src/radical/orbit/broker_plugin_host.py
+++ b/src/radical/orbit/broker_plugin_host.py
@@ -113,6 +113,16 @@ async def handle_request(self, method: str, path: str,
             raise HTTPException(status_code=404,
                                 detail=f'No route: {method} {path}')
 
+        # HTTP headers are case-insensitive, but only the gateway path goes
+        # through starlette's normalisation -- an in-process caller
+        # (BrokerCaller) hands its dict over verbatim, so 'Content-Type'
+        # would miss the lookup below and a msgpack body would be parsed
+        # as JSON.
+        headers = {k.lower(): v for k, v in (headers or {}).items()}
+
+        # NOTE: headers are deliberately NOT passed into the shim -- this
+        # path serves gateway and in-process callers, where a client-supplied
+        # `x-orbit-src` must not surface as a trusted owner (see RequestShim).
         content_type = headers.get('content-type', 'application/json')
         shim = RequestShim(path_params, query_params, body_bytes, content_type)
 
diff --git a/tests/unittests/test_broker_host_headers.py b/tests/unittests/test_broker_host_headers.py
new file mode 100644
index 00000000..ffff9c12
--- /dev/null
+++ b/tests/unittests/test_broker_host_headers.py
@@ -0,0 +1,56 @@
+"""Header-case handling on the broker plugin host.
+
+HTTP headers are case-insensitive, but only the gateway path goes through
+starlette's normalisation -- an in-process caller (BrokerCaller) hands its
+header dict to ``handle_request`` verbatim.  A client sending
+``Content-Type: application/msgpack`` must still get its body parsed as
+msgpack, not fed to ``json.loads`` (where the first msgpack byte dies as
+invalid utf-8).
+"""
+
+import re
+
+import msgpack
+import pytest
+
+from radical.orbit.broker_plugin_host import BrokerPluginHost
+
+
+def _host_with_echo_route():
+    async def noop_broadcast(*_a, **_k):
+        return None
+
+    host = BrokerPluginHost(plugin_names=[], broadcast_fn=noop_broadcast)
+
+    async def echo(request):
+        return {'echoed': await request.json()}
+
+    host._direct_routes.append(
+        ('POST', re.compile(r'^/echo$'), (), echo))
+    return host
+
+
+@pytest.mark.asyncio
+async def test_msgpack_body_with_capitalized_content_type():
+    host = _host_with_echo_route()
+    payload = {'tasks': [{'uid': 'task.000000', 'pool': 'local'}]}
+
+    resp = await host.handle_request(
+        'POST', '/echo',
+        headers={'Content-Type': 'application/msgpack'},
+        body_bytes=msgpack.packb(payload, use_bin_type=True))
+
+    import json
+    assert json.loads(resp.body)['echoed'] == payload
+
+
+@pytest.mark.asyncio
+async def test_json_body_stays_the_default():
+    host = _host_with_echo_route()
+
+    resp = await host.handle_request(
+        'POST', '/echo', headers={},
+        body_bytes=b'{"a": 1}')
+
+    import json
+    assert json.loads(resp.body)['echoed'] == {'a': 1}