diff --git a/erldistpy/channel.py b/erldistpy/channel.py index d0744b2..c5ba3d3 100644 --- a/erldistpy/channel.py +++ b/erldistpy/channel.py @@ -144,7 +144,14 @@ class Channel: def _encode_message(control: tuple, payload: object | None) -> bytes: - """Build a distribution message body (no length prefix).""" + """Build a distribution message body (no length prefix). + + Uses legacy pass-through framing (``0x70 | ControlMsg | OptionalPayload``). + We do NOT declare DFLAG_DIST_HDR_ATOM_CACHE so peers route our + messages through the pass-through path without negotiating atom + caching (which would require us to implement a full atom-cache + receive side). + """ body = bytes([PASS_THROUGH]) + encode(control) if payload is not None: body += encode(payload) @@ -152,23 +159,75 @@ def _encode_message(control: tuple, payload: object | None) -> bytes: def _decode_message(body: bytes) -> IncomingMessage: - if not body or body[0] != PASS_THROUGH: - raise ChannelError( - f"expected pass-through byte 0x70, got {body[:1].hex() or 'empty'}" - ) + """Decode a distribution message body. + + Handles both framings: + - Legacy pass-through: ``0x70 | ControlMsg | OptionalPayload`` + - Dist-header (modern): ``0x83 0x44 NumRefs | ControlMsg | OptionalPayload`` + + In the dist-header case the inline terms omit the ETF magic byte + (it's implicit from the outer ``0x83``); we synthesize it before + handing off to decode_term. + """ + if not body: + raise ChannelError("empty message body") + try: - control, off = decode_term(body, 1) - if not isinstance(control, tuple): - raise ChannelError(f"control message is not a tuple: {type(control).__name__}") - payload: object | None - if off == len(body): - payload = None + if body[0] == PASS_THROUGH: + control, off = decode_term(body, 1) + payload: object | None + if off == len(body): + payload = None + else: + payload, off = decode_term(body, off) + if off != len(body): + raise ChannelError( + f"trailing bytes after payload: {len(body) - off} unread" + ) + elif len(body) >= 3 and body[0:2] == b"\x83\x44": + num_refs = body[2] + if num_refs != 0: + raise ChannelError( + f"dist header has {num_refs} atom cache refs; " + "we don't maintain a cache, peer should send 0" + ) + # Strip the header. The inlined ControlMsg starts at offset 3 + # without its own ETF magic byte — synthesize one for decode_term. + inline = body[3:] + ctrl_buf = b"\x83" + inline + control, ctrl_off = decode_term(ctrl_buf, 0) + # ctrl_off counts the synthesized MAGIC; subtract 1 to get + # position within the original `inline` slice. + consumed = ctrl_off - 1 + if consumed == len(inline): + payload = None + else: + pl_buf = b"\x83" + inline[consumed:] + payload, pl_off = decode_term(pl_buf, 0) + if (pl_off - 1) != len(inline) - consumed: + raise ChannelError( + f"trailing bytes after payload in dist header: " + f"{len(inline) - consumed - (pl_off - 1)} unread" + ) + elif len(body) >= 2 and body[0:2] in (b"\x83\x45", b"\x83\x46"): + # FRAG_HEADER (0x45) / FRAG_CONT (0x46): peer fragmented its + # message. We declared DFLAG_FRAGMENTS, so peers may fragment + # large replies. Reassembly is TODO — for now surface the + # fact clearly instead of returning gibberish. + raise ChannelError( + "fragmented dist message received; reassembly not yet " + "implemented (TODO)" + ) else: - payload, off = decode_term(body, off) - if off != len(body): - raise ChannelError(f"trailing bytes after payload: {len(body) - off} unread") + raise ChannelError( + f"expected pass-through (0x70) or dist header (0x83 0x44), " + f"got {body[:2].hex()}" + ) except ETFError as e: raise ChannelError(f"ETF decode failed: {e}") from e + + if not isinstance(control, tuple): + raise ChannelError(f"control message is not a tuple: {type(control).__name__}") return IncomingMessage(control=control, payload=payload) diff --git a/erldistpy/flags.py b/erldistpy/flags.py index 36efa50..2306a5e 100644 --- a/erldistpy/flags.py +++ b/erldistpy/flags.py @@ -43,6 +43,19 @@ DFLAG_V4_NC = 0x0000000800000000 # bit 35 # What we advertise to peers. Enough to round-trip the term types we # care about (atoms, integers, binaries, lists, tuples, pids, refs, # maps) and to ride the v6 handshake. +# +# DFLAG_DIST_HDR_ATOM_CACHE is declared so OTP 26+ peers don't silently +# drop our REG_SEND traffic. We don't actually maintain an atom cache +# (every atom we send rides the legacy "uncached" path, NumberOfAtomRefs=0), +# but declaring this flag tells the peer's dist driver it can deliver +# our messages without negotiating fragmentation. +# +# DFLAG_FRAGMENTS lets the peer fragment large replies; we reassemble. +# +# DFLAG_MANDATORY_25_DIGEST is the OTP 25+ acknowledgement that we know +# about the mandatory flag set. Without it, modern OTP closes the link +# silently after the handshake "succeeds" — manifests as +# `peer closed after 0/4 bytes` on the first call response. DEFAULT_FLAGS = ( DFLAG_EXTENDED_REFERENCES | DFLAG_FUN_TAGS @@ -57,4 +70,5 @@ DEFAULT_FLAGS = ( | DFLAG_HANDSHAKE_23 | DFLAG_UNLINK_ID | DFLAG_V4_NC + | DFLAG_MANDATORY_25_DIGEST ) diff --git a/pyproject.toml b/pyproject.toml index f3bf846..3bf44b2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "erldistpy" -version = "0.1.6" +version = "0.1.7" description = "Native Python client for Erlang distribution protocol — EPMD + v6 handshake + gen_server call(), no asyncio." readme = { file = "README.md", content-type = "text/markdown" } requires-python = ">=3.10" diff --git a/tests/test_channel.py b/tests/test_channel.py index 32ee601..abaad95 100644 --- a/tests/test_channel.py +++ b/tests/test_channel.py @@ -57,7 +57,7 @@ def test_encode_message_with_payload(): def test_decode_rejects_wrong_first_byte(): - with pytest.raises(ChannelError, match="pass-through byte"): + with pytest.raises(ChannelError, match="pass-through .0x70. or dist header"): _decode_message(b"\x00" + encode((1,)))