fix(admin-config-tool): unwrap Request reply envelope before decoding as RequestReply

config push crashed on every real controller with a capnp KjException --
xce_reply() wraps every reply in an outer Request{body.replymsg} envelope,
but discovery.py/push.py decoded raw wire bytes directly as a bare
RequestReply. Adds a shared decode_reply() unwrap point used by both call
sites, and fixes the same wrong-shape assumption baked into task 5.6's
and test_cli.py's mock-server fixtures.
This commit is contained in:
Rongsong Shen 2026-07-30 09:43:56 +00:00
parent 46bbdaa9a6
commit 915e6fdf64
7 changed files with 231 additions and 26 deletions

View file

@ -12,13 +12,30 @@ Note: design.md's Dependencies list for this component names only
``xcecommon.wire``, but decoding ``RequestReply`` to inspect the redirect
fields genuinely requires ``xcecommon.schema`` too -- treated as a design
doc omission, not a license to hand-roll capnp decoding here.
Reply envelope (task 6.1 bug fix, live-verified 2026-07-30 against a real
deployed controller): every reply ``xce_reply()`` (``src/lib/request.c``)
sends over the wire -- ``config push``'s included -- is wrapped in an
outer ``Request{opcode: OPCode_reply, body: union{replymsg @4:
RequestReply}}`` envelope (``controller.capnp:645-651``) before being
serialized, never a bare ``RequestReply``. Decoding raw reply bytes
directly as ``RequestReply`` (the previous behavior here and in
``xceadmin.push``) misaligns every pointer offset and crashes with
``capnp.KjException: ... non-list pointer where text was expected`` on the
very first field access. :func:`decode_reply` is the single place this
unwrap happens -- this module owns it (it already owns "understand the
``RequestReply`` wire shape"), and ``xceadmin.push`` imports it rather than
duplicating the unwrap at its own call site.
"""
from __future__ import annotations
import contextlib
import sys
from xcecommon.errors import ConnectionTimeoutError, NoPrimaryError, TooManyRedirectsError
import capnp
from xcecommon.errors import ConnectionTimeoutError, NoPrimaryError, ReplyDecodeError, TooManyRedirectsError
from xcecommon.schema import load_schema
#: "Follow at most 3 redirect hops before reporting an error" (3.2) -- up
@ -26,13 +43,46 @@ from xcecommon.schema import load_schema
_MAX_REDIRECT_HOPS = 3
@contextlib.contextmanager
def decode_reply(reply_bytes: bytes):
"""Decode ``reply_bytes`` as the ``Request{opcode: reply,
body.replymsg: RequestReply}`` envelope every controller reply is
wrapped in on the wire (module docstring), and yield the inner
``RequestReply`` for field access -- e.g. ``with decode_reply(b) as
reply: reply.status``.
Raises :class:`ReplyDecodeError` (Requirement 7.3) if ``reply_bytes``
cannot be decoded as a ``Request`` at all, or decodes but
``body.which()`` is not ``"replymsg"`` (the bytes are not actually a
reply envelope -- this is what the old, wrong bare-``RequestReply``
wire shape now produces: a defaulted ``"nomsg"`` variant instead of a
crash, confirmed empirically). pycapnp's malformed-message errors are
often raised lazily on field access rather than at ``from_bytes()``
time, and a byte length that isn't a multiple of 8 words raises a
plain ``ValueError`` instead of ``capnp.KjException`` -- both are
caught here (verified against every corrupt/truncated/empty/garbage
byte shape) so no raw capnp/decode exception ever escapes this
function.
"""
mod = load_schema()
try:
with mod.Request.from_bytes(reply_bytes) as request:
which = request.body.which()
if which != "replymsg":
reason = f"expected reply envelope (body.replymsg), got body.{which}"
raise ReplyDecodeError(f"invalid reply: {reason}", reason)
yield request.body.replymsg
except (capnp.KjException, ValueError) as exc:
reason = str(exc).strip() or "malformed capnp reply envelope"
raise ReplyDecodeError(f"invalid reply: {reason}", reason) from exc
def _redirect_target(reply_bytes: bytes) -> tuple[str, int] | None:
"""``None`` when ``reply_bytes`` carries no redirect (empty
``redirectHost`` or zero ``redirectPort``) -- both must be non-empty
for this to be a real redirect (3.1).
"""
mod = load_schema()
with mod.RequestReply.from_bytes(reply_bytes) as reply:
with decode_reply(reply_bytes) as reply:
host, port = reply.redirectHost, reply.redirectPort
if host and port:
return host, port

View file

@ -49,7 +49,7 @@ from pathlib import Path
from xcecommon import wire
from xcecommon.errors import ReplyDecodeError
from xcecommon.schema import load_schema
from xceadmin.discovery import follow_redirects
from xceadmin.discovery import decode_reply, follow_redirects
_DEFAULT_PORT = 9991
@ -113,7 +113,7 @@ def push_config(cluster_bin_path: str, controller: str) -> PushResult:
host, port = _split_controller(controller)
_, _, reply_bytes = follow_redirects(host, port, _send_via(request_bytes))
with mod.RequestReply.from_bytes(reply_bytes) as reply:
with decode_reply(reply_bytes) as reply:
if reply.status != 0:
return PushResult(
version_tag=0, acked_count=0, error=f"controller returned status {reply.status}"

View file

@ -9,6 +9,8 @@ from pathlib import Path
import pytest
from xcecommon.schema import load_schema
@dataclass(frozen=True)
class TlsCertPair:
@ -42,3 +44,29 @@ def tls_certs(tmp_path_factory) -> TlsCertPair:
capture_output=True,
)
return TlsCertPair(cert_file=cert_file, key_file=key_file)
@pytest.fixture
def wrap_reply():
"""Factory fixture: wrap a ``RequestReply`` message builder in the
outer ``Request{opcode: reply, body.replymsg: ...}`` envelope every
real controller reply is sent in on the wire (``xce_reply()``,
``src/lib/request.c``; ``controller.capnp:645-651``) -- returns the
serialized envelope bytes.
Task 6.1 bug fix: mock TLS servers across this suite previously sent a
bare ``RequestReply`` on the wire, which matched the old (wrong)
client-side decode assumption. ``xceadmin.discovery.decode_reply`` now
expects the real envelope shape, so every mock server reply must be
wrapped through this helper instead.
"""
def _wrap(reply_msg) -> bytes:
mod = load_schema()
request = mod.Request.new_message()
request.opcode = "reply"
request.seq = reply_msg.seq
request.body.replymsg = reply_msg
return request.to_bytes()
return _wrap

View file

@ -196,7 +196,7 @@ def cluster_bin(tmp_path):
return str(path)
def test_config_push_success_exits_zero(mock_controller, cluster_bin):
def test_config_push_success_exits_zero(mock_controller, cluster_bin, wrap_reply):
from xcecommon import wire
mod = load_schema()
@ -209,7 +209,7 @@ def test_config_push_success_exits_zero(mock_controller, cluster_bin):
upgrade_reply.versionTag = 5
upgrade_reply.ackedCount = 3
upgrade_reply.error = ""
wire.send_frame(conn, reply.to_bytes())
wire.send_frame(conn, wrap_reply(reply))
srv = mock_controller(_handler)
@ -224,7 +224,7 @@ def test_config_push_success_exits_zero(mock_controller, cluster_bin):
assert "ackedCount=3" in result.output
def test_config_push_zero_acked_count_exits_nonzero(mock_controller, cluster_bin):
def test_config_push_zero_acked_count_exits_nonzero(mock_controller, cluster_bin, wrap_reply):
from xcecommon import wire
mod = load_schema()
@ -237,7 +237,7 @@ def test_config_push_zero_acked_count_exits_nonzero(mock_controller, cluster_bin
upgrade_reply.versionTag = 0
upgrade_reply.ackedCount = 0
upgrade_reply.error = "no quorum"
wire.send_frame(conn, reply.to_bytes())
wire.send_frame(conn, wrap_reply(reply))
srv = mock_controller(_handler)

View file

@ -99,7 +99,7 @@ def cluster_bin(tmp_path):
# ---------------------------------------------------------------------------
def test_push_config_successful_commit(mock_controller, cluster_bin):
def test_push_config_successful_commit(mock_controller, cluster_bin, wrap_reply):
mod = load_schema()
received = {}
@ -117,7 +117,7 @@ def test_push_config_successful_commit(mock_controller, cluster_bin):
upgrade_reply.versionTag = 7
upgrade_reply.ackedCount = 3
upgrade_reply.error = ""
wire.send_frame(conn, reply.to_bytes())
wire.send_frame(conn, wrap_reply(reply))
srv = mock_controller(_handler)
@ -133,7 +133,7 @@ def test_push_config_successful_commit(mock_controller, cluster_bin):
assert received["sha256"] == hashlib.sha256(Path(cluster_bin).read_bytes()).digest()
def test_push_config_follows_one_redirect_to_primary(mock_controller, cluster_bin):
def test_push_config_follows_one_redirect_to_primary(mock_controller, cluster_bin, wrap_reply):
mod = load_schema()
def _standby_handler(conn):
@ -144,7 +144,7 @@ def test_push_config_follows_one_redirect_to_primary(mock_controller, cluster_bi
# port filled in once the primary server exists, see below
reply.redirectPort = _standby_handler.primary_port
reply.data.nodata = None
wire.send_frame(conn, reply.to_bytes())
wire.send_frame(conn, wrap_reply(reply))
def _primary_handler(conn):
wire.recv_frame(conn)
@ -154,7 +154,7 @@ def test_push_config_follows_one_redirect_to_primary(mock_controller, cluster_bi
upgrade_reply.versionTag = 9
upgrade_reply.ackedCount = 2
upgrade_reply.error = ""
wire.send_frame(conn, reply.to_bytes())
wire.send_frame(conn, wrap_reply(reply))
primary_srv = mock_controller(_primary_handler)
_standby_handler.primary_port = primary_srv.port
@ -166,7 +166,7 @@ def test_push_config_follows_one_redirect_to_primary(mock_controller, cluster_bi
assert result.acked_count == 2
def test_push_config_non_zero_status_reports_error_not_exception(mock_controller, cluster_bin):
def test_push_config_non_zero_status_reports_error_not_exception(mock_controller, cluster_bin, wrap_reply):
mod = load_schema()
def _handler(conn):
@ -174,7 +174,7 @@ def test_push_config_non_zero_status_reports_error_not_exception(mock_controller
reply = mod.RequestReply.new_message()
reply.status = 1
reply.data.debugmsg = "simulated failure"
wire.send_frame(conn, reply.to_bytes())
wire.send_frame(conn, wrap_reply(reply))
srv = mock_controller(_handler)
@ -184,7 +184,7 @@ def test_push_config_non_zero_status_reports_error_not_exception(mock_controller
assert "status 1" in result.error
def test_push_config_wrong_reply_variant_raises_reply_decode_error(mock_controller, cluster_bin):
def test_push_config_wrong_reply_variant_raises_reply_decode_error(mock_controller, cluster_bin, wrap_reply):
mod = load_schema()
def _handler(conn):
@ -192,7 +192,34 @@ def test_push_config_wrong_reply_variant_raises_reply_decode_error(mock_controll
reply = mod.RequestReply.new_message()
reply.status = 0
reply.data.nodata = None # not upgradeconfigreplymsg
wire.send_frame(conn, reply.to_bytes())
wire.send_frame(conn, wrap_reply(reply))
srv = mock_controller(_handler)
with pytest.raises(ReplyDecodeError):
push.push_config(cluster_bin, f"127.0.0.1:{srv.port}")
def test_push_config_rejects_bare_requestreply_bytes_old_wire_shape_regression(
mock_controller, cluster_bin
):
"""Regression test (task 6.1): a mock server that still sends the OLD,
wrong shape -- a bare RequestReply with no outer Request envelope --
must be cleanly rejected as ReplyDecodeError, not silently accepted
and not an unhandled capnp.KjException. This locks in that the fixed
client only accepts the real controller wire shape.
"""
mod = load_schema()
def _handler(conn):
wire.recv_frame(conn)
reply = mod.RequestReply.new_message()
reply.status = 0
upgrade_reply = reply.data.init("upgradeconfigreplymsg")
upgrade_reply.versionTag = 1
upgrade_reply.ackedCount = 1
upgrade_reply.error = ""
wire.send_frame(conn, reply.to_bytes()) # NOT wrapped -- old shape
srv = mock_controller(_handler)
@ -301,7 +328,7 @@ def test_probe_config_has_no_wire_level_rejection_signal_by_design(
# ---------------------------------------------------------------------------
def test_push_config_propagates_too_many_redirects_end_to_end(mock_controller, cluster_bin):
def test_push_config_propagates_too_many_redirects_end_to_end(mock_controller, cluster_bin, wrap_reply):
"""A real chain of 4 redirecting mock servers -- push_config's own
composition with xceadmin.discovery must surface TooManyRedirectsError,
not silently follow forever or return a bogus result.
@ -317,7 +344,7 @@ def test_push_config_propagates_too_many_redirects_end_to_end(mock_controller, c
reply.redirectHost = "127.0.0.1"
reply.redirectPort = next_port_getter()
reply.data.nodata = None
wire.send_frame(conn, reply.to_bytes())
wire.send_frame(conn, wrap_reply(reply))
return _handler

View file

@ -1,24 +1,35 @@
"""Unit tests for ``xceadmin.discovery`` (design.md, ``xceadmin.discovery``;
Requirements 3.1, 3.2, 3.3, 3.4).
Requirements 3.1, 3.2, 3.3, 3.4, 7.3).
"""
from __future__ import annotations
import pytest
from xcecommon.errors import ConnectionTimeoutError, NoPrimaryError, TooManyRedirectsError
from xcecommon.errors import ConnectionTimeoutError, NoPrimaryError, ReplyDecodeError, TooManyRedirectsError
from xcecommon.schema import load_schema
from xceadmin import discovery
def _reply_bytes(redirect_host: str = "", redirect_port: int = 0) -> bytes:
"""Builds the real ``Request{opcode: reply, body.replymsg:
RequestReply}`` envelope -- the actual wire shape every controller
reply is sent in (``xce_reply()``, ``src/lib/request.c``;
``controller.capnp:645-651``; task 6.1). NOT a bare ``RequestReply`` --
see ``test_redirect_target_rejects_bare_requestreply_bytes_old_wire_shape``
below for what happens when the old, wrong shape is decoded.
"""
mod = load_schema()
msg = mod.RequestReply.new_message()
inner = mod.RequestReply.new_message()
if redirect_host:
msg.redirectHost = redirect_host
inner.redirectHost = redirect_host
if redirect_port:
msg.redirectPort = redirect_port
return msg.to_bytes()
inner.redirectPort = redirect_port
request = mod.Request.new_message()
request.opcode = "reply"
request.seq = 1
request.body.replymsg = inner
return request.to_bytes()
def test_follow_redirects_no_redirect_returns_first_reply_unchanged():
@ -138,3 +149,72 @@ def test_follow_redirects_zero_port_is_not_treated_as_a_redirect():
assert (final_host, final_port) == ("node1", 9991)
assert reply_bytes == reply
# ---------------------------------------------------------------------------
# decode_reply(): Request{body.replymsg} envelope unwrap (task 6.1)
# ---------------------------------------------------------------------------
def test_decode_reply_unwraps_envelope_and_yields_request_reply():
"""The real wire shape -- Request{opcode: reply, body.replymsg:
RequestReply} -- decodes cleanly and yields the inner RequestReply.
"""
reply_bytes = _reply_bytes(redirect_host="primary", redirect_port=9991)
with discovery.decode_reply(reply_bytes) as reply:
assert reply.redirectHost == "primary"
assert reply.redirectPort == 9991
def test_decode_reply_rejects_bare_requestreply_bytes_old_wire_shape():
"""Regression test (task 6.1): the OLD, wrong client assumption was
that a reply is a bare RequestReply with no outer Request envelope.
Feeding decode_reply exactly that (unwrapped) byte shape must not
silently succeed and must not raise a raw capnp exception -- it must
be cleanly rejected as ReplyDecodeError, proving this is no longer
accepted as a valid reply shape.
"""
mod = load_schema()
bare_bytes = mod.RequestReply.new_message().to_bytes()
with pytest.raises(ReplyDecodeError) as exc_info:
with discovery.decode_reply(bare_bytes) as _reply:
pass
assert "replymsg" in exc_info.value.message
def test_decode_reply_wraps_truncated_bytes_as_reply_decode_error():
"""A genuinely corrupt/truncated reply must surface as
ReplyDecodeError (Requirement 7.3), never an unhandled
capnp.KjException or ValueError.
"""
full = _reply_bytes(redirect_host="primary", redirect_port=9991)
truncated = full[: len(full) // 2]
with pytest.raises(ReplyDecodeError):
with discovery.decode_reply(truncated) as _reply:
pass
def test_decode_reply_wraps_non_capnp_bytes_as_reply_decode_error():
"""Bytes that aren't capnp at all (e.g. a length not a multiple of 8
words) raise a plain ValueError inside pycapnp -- confirm that is also
converted to ReplyDecodeError, not left to escape raw.
"""
with pytest.raises(ReplyDecodeError):
with discovery.decode_reply(b"not a capnp message at all") as _reply:
pass
def test_follow_redirects_propagates_reply_decode_error_on_corrupt_reply():
"""follow_redirects' own composition with decode_reply (via
_redirect_target) must surface the same ReplyDecodeError, not crash.
"""
def send_fn(host, port):
return b"\x00" * 4 # too short to be a valid capnp frame
with pytest.raises(ReplyDecodeError):
discovery.follow_redirects("node1", 9991, send_fn)