diff --git a/docs/ROADMAP.md b/docs/ROADMAP.md index fa99f5e..47b2c44 100644 --- a/docs/ROADMAP.md +++ b/docs/ROADMAP.md @@ -30,14 +30,21 @@ EPMD (Erlang Port Mapper Daemon) maps node names to TCP ports. - Tests: recorded byte streams from real EPMD + live integration that spawns its own `erl -sname` and tears it down -## Phase 3 — Distribution handshake +## Phase 3 — Distribution handshake ✅ -- TCP connect to the resolved port -- `send_name` / `recv_status` / `recv_challenge` / `send_challenge_reply` - / `recv_challenge_ack` -- Cookie digest via `erlang:phash2`-equivalent (md5-based per spec) -- Version 6 ("v6") handshake, the modern one Elixir 1.15+ uses -- Tests: handshake against a live Erlang node started in conftest +- `handshake(sock, our_name=..., cookie=...)` drives the v6 dance: + send_name(N) / recv_status(s) / recv_challenge(N) / + send_challenge_reply(r) / recv_challenge_ack(a) +- MD5 cookie digest, cross-checked against `erlang:md5/1` output +- Flag set in `erldistpy/flags.py` advertises OTP 23+ compatibility +- Tests: + - Frame builders/parsers as pure functions + - Cookie digest against an Erlang-computed reference + - Live handshake against `erl -sname ... -setcookie ...` + - Wrong-cookie test confirms peer rejection surfaces correctly + +Newer SHA-256 digest (DFLAG_MANDATORY_25_DIGEST) deferred — landed only +if we hit a peer that requires it. ## Phase 4 — Distribution channel diff --git a/erldistpy/__init__.py b/erldistpy/__init__.py index df56f78..221ab02 100644 --- a/erldistpy/__init__.py +++ b/erldistpy/__init__.py @@ -10,15 +10,19 @@ from erldistpy.etf import ( decode, encode, ) +from erldistpy.handshake import HandshakeError, HandshakeResult, handshake __all__ = [ "Atom", "EpmdError", "EpmdInfo", + "HandshakeError", + "HandshakeResult", "Pid", "Reference", "__version__", "decode", "encode", + "handshake", "lookup", ] diff --git a/erldistpy/flags.py b/erldistpy/flags.py new file mode 100644 index 0000000..36efa50 --- /dev/null +++ b/erldistpy/flags.py @@ -0,0 +1,60 @@ +"""Distribution capability flags. + +Negotiated during the handshake; tell each peer what term types and +protocol features the other supports. We advertise the minimum useful +set for talking to an OTP 23+ Elixir node. + +Source: ``$OTP/lib/kernel/include/dist.hrl`` and +https://www.erlang.org/doc/apps/erts/erl_dist_protocol.html +""" + +from __future__ import annotations + +DFLAG_PUBLISHED = 0x00000001 +DFLAG_ATOM_CACHE = 0x00000002 +DFLAG_EXTENDED_REFERENCES = 0x00000004 +DFLAG_DIST_MONITOR = 0x00000008 +DFLAG_FUN_TAGS = 0x00000010 +DFLAG_DIST_MONITOR_NAME = 0x00000020 +DFLAG_HIDDEN_ATOM_CACHE = 0x00000040 +DFLAG_NEW_FUN_TAGS = 0x00000080 +DFLAG_EXTENDED_PIDS_PORTS = 0x00000100 +DFLAG_EXPORT_PTR_TAG = 0x00000200 +DFLAG_BIT_BINARIES = 0x00000400 +DFLAG_NEW_FLOATS = 0x00000800 +DFLAG_UNICODE_IO = 0x00001000 +DFLAG_DIST_HDR_ATOM_CACHE = 0x00002000 +DFLAG_SMALL_ATOM_TAGS = 0x00004000 +DFLAG_UTF8_ATOMS = 0x00010000 +DFLAG_MAP_TAG = 0x00020000 +DFLAG_BIG_CREATION = 0x00040000 +DFLAG_SEND_SENDER = 0x00080000 +DFLAG_BIG_SEQTRACE_LABELS = 0x00100000 +DFLAG_EXIT_PAYLOAD = 0x00400000 +DFLAG_FRAGMENTS = 0x00800000 +DFLAG_HANDSHAKE_23 = 0x01000000 +DFLAG_UNLINK_ID = 0x02000000 +DFLAG_MANDATORY_25_DIGEST = 0x0000000400000000 # bit 34 +DFLAG_SPAWN = 0x0000000100000000 # bit 32 +DFLAG_NAME_ME = 0x0000000200000000 # bit 33 +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. +DEFAULT_FLAGS = ( + DFLAG_EXTENDED_REFERENCES + | DFLAG_FUN_TAGS + | DFLAG_EXTENDED_PIDS_PORTS + | DFLAG_NEW_FUN_TAGS + | DFLAG_EXPORT_PTR_TAG + | DFLAG_BIT_BINARIES + | DFLAG_NEW_FLOATS + | DFLAG_UTF8_ATOMS + | DFLAG_MAP_TAG + | DFLAG_BIG_CREATION + | DFLAG_HANDSHAKE_23 + | DFLAG_UNLINK_ID + | DFLAG_V4_NC +) diff --git a/erldistpy/handshake.py b/erldistpy/handshake.py new file mode 100644 index 0000000..db634bb --- /dev/null +++ b/erldistpy/handshake.py @@ -0,0 +1,223 @@ +"""Distribution handshake — v6 (DFLAG_HANDSHAKE_23) only. + +Drives a freshly-opened TCP socket through the OTP 23+ handshake dance: + + client server + | send_name ('N') ─────────► | + | ◄─────── recv_status ('s') | + | ◄─── recv_challenge ('N') | + | send_challenge_reply ('r') ►| + | ◄── recv_challenge_ack ('a')| + +All frames during the handshake are wrapped in a 2-byte big-endian +length prefix. After ``handshake()`` returns the socket is authenticated +and ready for the post-handshake data phase (4-byte framing, Phase 4). + +Cookie discipline: ``cookie`` is a string passed in by the caller and +never logged by this module. Treat it like any other secret. + +Spec: https://www.erlang.org/doc/apps/erts/erl_dist_protocol.html#distribution-handshake +""" + +from __future__ import annotations + +import hashlib +import os +import socket +import struct +from dataclasses import dataclass + +from erldistpy.flags import DEFAULT_FLAGS, DFLAG_HANDSHAKE_23 + +TAG_SEND_NAME = 0x4E # 'N' +TAG_RECV_STATUS = 0x73 # 's' +TAG_RECV_CHALLENGE = 0x4E # 'N' (same as send_name; distinguished by direction) +TAG_CHALLENGE_REPLY = 0x72 # 'r' +TAG_CHALLENGE_ACK = 0x61 # 'a' + + +class HandshakeError(RuntimeError): + """Handshake protocol or I/O failure.""" + + +@dataclass(frozen=True) +class HandshakeResult: + peer_name: str + peer_flags: int + peer_creation: int + our_flags: int + our_creation: int + + +def handshake( + sock: socket.socket, + *, + our_name: str, + cookie: str, + our_flags: int = DEFAULT_FLAGS, + our_creation: int | None = None, + timeout: float = 5.0, +) -> HandshakeResult: + """Drive ``sock`` through a v6 distribution handshake. + + ``our_name`` is the full ``name@host`` we are calling as. + ``cookie`` is the shared secret with the peer. + + On success the socket is left open and authenticated. On any + protocol or auth failure raises :class:`HandshakeError`. + """ + if "@" not in our_name: + raise ValueError(f"our_name must look like 'name@host', got {our_name!r}") + if not (our_flags & DFLAG_HANDSHAKE_23): + raise ValueError("our_flags must include DFLAG_HANDSHAKE_23 (v6 handshake)") + if our_creation is None: + our_creation = _random_creation() + + sock.settimeout(timeout) + + # Step 1: send_name + _send_frame(sock, build_send_name(our_name, our_flags, our_creation)) + + # Step 2: recv_status + status_frame = _recv_frame(sock) + status = parse_recv_status(status_frame) + if status not in ("ok", "ok_simultaneous"): + raise HandshakeError(f"peer rejected handshake: status={status!r}") + + # Step 3: recv_challenge + challenge_frame = _recv_frame(sock) + peer_flags, peer_challenge, peer_creation, peer_name = parse_recv_challenge(challenge_frame) + if not (peer_flags & DFLAG_HANDSHAKE_23): + raise HandshakeError("peer does not speak v6 handshake (no DFLAG_HANDSHAKE_23)") + + # Step 4: send_challenge_reply with our challenge + MD5(cookie || peer_challenge) + our_challenge = _random_challenge() + reply_digest = cookie_digest(cookie, peer_challenge) + _send_frame(sock, build_challenge_reply(our_challenge, reply_digest)) + + # Step 5: recv_challenge_ack — peer must answer with MD5(cookie || our_challenge) + ack_frame = _recv_frame(sock) + ack_digest = parse_challenge_ack(ack_frame) + expected = cookie_digest(cookie, our_challenge) + if ack_digest != expected: + raise HandshakeError("cookie digest mismatch — wrong cookie or peer impostor") + + return HandshakeResult( + peer_name=peer_name, + peer_flags=peer_flags, + peer_creation=peer_creation, + our_flags=our_flags, + our_creation=our_creation, + ) + + +# --------------------------------------------------------------------------- +# Frame builders / parsers (pure functions, exposed for tests) +# --------------------------------------------------------------------------- + + +def build_send_name(name: str, flags: int, creation: int) -> bytes: + """Build the v6 ``send_name`` frame body (no length prefix).""" + name_bytes = name.encode("utf-8") + return ( + bytes([TAG_SEND_NAME]) + + struct.pack(">Q", flags) + + struct.pack(">I", creation) + + struct.pack(">H", len(name_bytes)) + + name_bytes + ) + + +def parse_recv_status(frame: bytes) -> str: + if not frame or frame[0] != TAG_RECV_STATUS: + raise HandshakeError( + f"expected recv_status tag {TAG_RECV_STATUS}, got {frame[:1].hex()}" + ) + return frame[1:].decode("latin-1") + + +def parse_recv_challenge(frame: bytes) -> tuple[int, int, int, str]: + """Returns (flags, challenge, creation, name).""" + if not frame or frame[0] != TAG_RECV_CHALLENGE: + raise HandshakeError( + f"expected recv_challenge tag {TAG_RECV_CHALLENGE}, got {frame[:1].hex()}" + ) + if len(frame) < 1 + 8 + 4 + 4 + 2: + raise HandshakeError(f"recv_challenge frame truncated: {len(frame)} bytes") + flags = struct.unpack(">Q", frame[1:9])[0] + challenge = struct.unpack(">I", frame[9:13])[0] + creation = struct.unpack(">I", frame[13:17])[0] + nlen = struct.unpack(">H", frame[17:19])[0] + if len(frame) < 19 + nlen: + raise HandshakeError("recv_challenge name truncated") + name = frame[19:19 + nlen].decode("utf-8") + return flags, challenge, creation, name + + +def build_challenge_reply(challenge: int, digest: bytes) -> bytes: + if len(digest) != 16: + raise ValueError(f"digest must be 16 bytes, got {len(digest)}") + return bytes([TAG_CHALLENGE_REPLY]) + struct.pack(">I", challenge) + digest + + +def parse_challenge_ack(frame: bytes) -> bytes: + if not frame or frame[0] != TAG_CHALLENGE_ACK: + raise HandshakeError( + f"expected challenge_ack tag {TAG_CHALLENGE_ACK}, got {frame[:1].hex()}" + ) + if len(frame) != 1 + 16: + raise HandshakeError(f"challenge_ack frame wrong length: {len(frame)} bytes") + return frame[1:17] + + +def cookie_digest(cookie: str, challenge: int) -> bytes: + """MD5 of cookie concatenated with the base-10 string of the challenge. + + Per the OTP handshake spec: ``md5(Cookie ++ integer_to_list(Challenge))``. + The challenge is treated as an unsigned 32-bit integer. + """ + if challenge < 0: + challenge = challenge & 0xFFFFFFFF + payload = cookie.encode("latin-1") + str(challenge).encode("ascii") + return hashlib.md5(payload).digest() + + +# --------------------------------------------------------------------------- +# Wire helpers (2-byte length-prefixed frames) +# --------------------------------------------------------------------------- + + +def _send_frame(sock: socket.socket, body: bytes) -> None: + if len(body) > 0xFFFF: + raise HandshakeError(f"handshake frame too large: {len(body)} bytes") + sock.sendall(struct.pack(">H", len(body)) + body) + + +def _recv_frame(sock: socket.socket) -> bytes: + hdr = _recv_exact(sock, 2) + (length,) = struct.unpack(">H", hdr) + if length == 0: + return b"" + return _recv_exact(sock, length) + + +def _recv_exact(sock: socket.socket, n: int) -> bytes: + chunks = [] + remaining = n + while remaining > 0: + chunk = sock.recv(remaining) + if not chunk: + raise HandshakeError(f"peer closed connection after {n - remaining}/{n} bytes") + chunks.append(chunk) + remaining -= len(chunk) + return b"".join(chunks) + + +def _random_creation() -> int: + """Non-zero uint32. The spec wants a value that survives node restarts.""" + n = struct.unpack(">I", os.urandom(4))[0] + return n if n else 1 + + +def _random_challenge() -> int: + return struct.unpack(">I", os.urandom(4))[0] diff --git a/tests/test_handshake.py b/tests/test_handshake.py new file mode 100644 index 0000000..73bfa60 --- /dev/null +++ b/tests/test_handshake.py @@ -0,0 +1,243 @@ +"""Distribution handshake tests. + +Unit tests cover the frame builders/parsers and the cookie digest +formula (cross-checked against ``erlang:md5/1`` output). Live tests +spawn an Erlang node with a known cookie and drive a full handshake +end-to-end. +""" + +from __future__ import annotations + +import shutil +import socket +import struct +import subprocess +import time + +import pytest + +from erldistpy.epmd import lookup +from erldistpy.flags import DEFAULT_FLAGS, DFLAG_HANDSHAKE_23 +from erldistpy.handshake import ( + HandshakeError, + build_challenge_reply, + build_send_name, + cookie_digest, + handshake, + parse_challenge_ack, + parse_recv_challenge, + parse_recv_status, +) + +# Reference value computed by `erlang:md5("TESTCOOKIE" ++ integer_to_list(1234567890))` +COOKIE_DIGEST_REF = bytes.fromhex("1f856146c1dd930d59ef38a9eeef21f6") + + +# -------------------------------------------------------------------------- +# Cookie digest +# -------------------------------------------------------------------------- + + +def test_cookie_digest_matches_erlang_reference(): + assert cookie_digest("TESTCOOKIE", 1234567890) == COOKIE_DIGEST_REF + + +def test_cookie_digest_treats_challenge_as_unsigned(): + # The wire challenge is a uint32. -1 should hash like 0xFFFFFFFF. + a = cookie_digest("X", -1) + b = cookie_digest("X", 0xFFFFFFFF) + assert a == b + + +def test_cookie_digest_length_is_16(): + assert len(cookie_digest("anything", 0)) == 16 + + +# -------------------------------------------------------------------------- +# Frame builders / parsers +# -------------------------------------------------------------------------- + + +def test_build_send_name_shape(): + frame = build_send_name("erldistpy@localhost", DEFAULT_FLAGS, creation=42) + assert frame[0] == 0x4E + flags = struct.unpack(">Q", frame[1:9])[0] + creation = struct.unpack(">I", frame[9:13])[0] + nlen = struct.unpack(">H", frame[13:15])[0] + name = frame[15:15 + nlen] + assert flags == DEFAULT_FLAGS + assert creation == 42 + assert name == b"erldistpy@localhost" + + +def test_parse_recv_status_ok(): + assert parse_recv_status(b"sok") == "ok" + + +def test_parse_recv_status_nok(): + assert parse_recv_status(b"snot_allowed") == "not_allowed" + + +def test_parse_recv_status_wrong_tag(): + with pytest.raises(HandshakeError, match="recv_status tag"): + parse_recv_status(b"Xok") + + +def test_parse_recv_challenge_roundtrip(): + name = b"peer@host" + flags = DEFAULT_FLAGS + challenge = 0xDEADBEEF + creation = 7 + frame = ( + bytes([0x4E]) + + struct.pack(">Q", flags) + + struct.pack(">I", challenge) + + struct.pack(">I", creation) + + struct.pack(">H", len(name)) + + name + ) + f, c, cr, n = parse_recv_challenge(frame) + assert (f, c, cr, n) == (flags, challenge, creation, "peer@host") + + +def test_parse_recv_challenge_truncated(): + with pytest.raises(HandshakeError, match="truncated"): + parse_recv_challenge(bytes([0x4E]) + b"\x00" * 5) + + +def test_challenge_reply_roundtrip(): + digest = b"\xAA" * 16 + frame = build_challenge_reply(0x12345678, digest) + assert frame[0] == 0x72 + assert struct.unpack(">I", frame[1:5])[0] == 0x12345678 + assert frame[5:] == digest + + +def test_challenge_reply_rejects_wrong_digest_length(): + with pytest.raises(ValueError): + build_challenge_reply(0, b"\x00" * 15) + + +def test_parse_challenge_ack_ok(): + digest = b"\xBB" * 16 + assert parse_challenge_ack(bytes([0x61]) + digest) == digest + + +def test_parse_challenge_ack_wrong_tag(): + with pytest.raises(HandshakeError, match="challenge_ack tag"): + parse_challenge_ack(bytes([0x60]) + b"\x00" * 16) + + +def test_parse_challenge_ack_wrong_length(): + with pytest.raises(HandshakeError, match="wrong length"): + parse_challenge_ack(bytes([0x61]) + b"\x00" * 15) + + +# -------------------------------------------------------------------------- +# Top-level handshake() validation +# -------------------------------------------------------------------------- + + +def test_handshake_requires_qualified_name(): + sock = socket.socket() + with pytest.raises(ValueError, match="name@host"): + handshake(sock, our_name="justname", cookie="X") + + +def test_handshake_requires_v6_flag(): + sock = socket.socket() + with pytest.raises(ValueError, match="DFLAG_HANDSHAKE_23"): + handshake(sock, our_name="x@h", cookie="X", our_flags=0) + + +# -------------------------------------------------------------------------- +# Live interop — spawn an erl node with a known cookie and shake hands +# -------------------------------------------------------------------------- + + +def _epmd_running() -> bool: + try: + with socket.create_connection(("localhost", 4369), timeout=0.5): + return True + except OSError: + return False + + +@pytest.fixture(scope="module") +def live_erlang_peer(): + """Spawn `erl -sname erldistpy_hs -setcookie ERLDISTPY_HS_COOKIE`. + + Returns (sname, cookie). Skips the suite if erl/EPMD are absent. + """ + if not shutil.which("erl"): + pytest.skip("erl not installed") + if not _epmd_running(): + pytest.skip("EPMD not running on localhost") + + sname = "erldistpy_hs" + cookie = "ERLDISTPY_HS_COOKIE" + proc = subprocess.Popen( + [ + "erl", + "-sname", sname, + "-setcookie", cookie, + "-noshell", + "-eval", "timer:sleep(infinity).", + ], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + for _ in range(40): + time.sleep(0.1) + if lookup(sname, timeout=0.5) is not None: + break + else: + proc.terminate() + proc.wait(timeout=2) + pytest.skip("erl node failed to register with EPMD") + + yield sname, cookie + + proc.terminate() + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + proc.kill() + + +def _connect_to(sname: str) -> socket.socket: + info = lookup(sname) + assert info is not None, f"node {sname} not in EPMD" + return socket.create_connection(("localhost", info.port), timeout=2) + + +def test_live_handshake_succeeds(live_erlang_peer): + sname, cookie = live_erlang_peer + sock = _connect_to(sname) + try: + # Use a distinct sname for our client side + result = handshake( + sock, + our_name=f"erldistpy_client@{socket.gethostname()}", + cookie=cookie, + ) + assert result.peer_name.startswith(sname + "@") + assert result.peer_flags & DFLAG_HANDSHAKE_23 + assert result.peer_creation > 0 + assert result.our_creation > 0 + finally: + sock.close() + + +def test_live_handshake_bad_cookie_fails(live_erlang_peer): + sname, _ = live_erlang_peer + sock = _connect_to(sname) + try: + with pytest.raises(HandshakeError, match="digest mismatch|peer closed"): + handshake( + sock, + our_name=f"erldistpy_client_bad@{socket.gethostname()}", + cookie="WRONG_COOKIE", + ) + finally: + sock.close()