From bbb0c316c5bc4544e3a0e47ce8e97677e7189c58 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 16 Jun 2026 11:00:00 -0400 Subject: [PATCH] phase 2: EPMD client Synchronous TCP client for Erlang Port Mapper Daemon. One request type (PORT_PLEASE2_REQ, tag 122), one response type (PORT2_RESP, tag 119). Returns EpmdInfo dataclass or None if the node is not registered. Tests run two layers: - Unit tests against recorded byte streams captured from a real EPMD answering for `erl -sname testnode` and for an unregistered name. - Integration tests spawn `erl -sname erldistpy_itest` in a fixture and verify lookup() returns the live port; skipped if erl or EPMD are absent. 10 new tests, 68 total green, lint clean. --- erldistpy/__init__.py | 13 +++- erldistpy/epmd.py | 122 +++++++++++++++++++++++++++++++++++ tests/test_epmd.py | 145 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 279 insertions(+), 1 deletion(-) create mode 100644 erldistpy/epmd.py create mode 100644 tests/test_epmd.py diff --git a/erldistpy/__init__.py b/erldistpy/__init__.py index cad57ff..df56f78 100644 --- a/erldistpy/__init__.py +++ b/erldistpy/__init__.py @@ -2,6 +2,7 @@ __version__ = "0.0.1" +from erldistpy.epmd import EpmdError, EpmdInfo, lookup from erldistpy.etf import ( Atom, Pid, @@ -10,4 +11,14 @@ from erldistpy.etf import ( encode, ) -__all__ = ["Atom", "Pid", "Reference", "decode", "encode", "__version__"] +__all__ = [ + "Atom", + "EpmdError", + "EpmdInfo", + "Pid", + "Reference", + "__version__", + "decode", + "encode", + "lookup", +] diff --git a/erldistpy/epmd.py b/erldistpy/epmd.py new file mode 100644 index 0000000..621ec3b --- /dev/null +++ b/erldistpy/epmd.py @@ -0,0 +1,122 @@ +"""EPMD — Erlang Port Mapper Daemon client. + +EPMD maps node names to TCP ports. Before opening a distribution +connection we ask the host's EPMD ``where does node X live?``. The wire +is tiny: one request type (``PORT_PLEASE2_REQ``), one response type +(``PORT2_RESP``). + +Spec: https://www.erlang.org/doc/apps/erts/erl_dist_protocol.html#epmd-protocol +""" + +from __future__ import annotations + +import socket +import struct +from dataclasses import dataclass + +DEFAULT_PORT = 4369 + +PORT_PLEASE2_REQ = 122 +PORT2_RESP = 119 + +# Node types reported by EPMD +NODE_TYPE_NORMAL = 77 # M — visible Erlang node +NODE_TYPE_HIDDEN = 72 # H — hidden (does not appear in nodes()) + + +class EpmdError(RuntimeError): + """EPMD I/O or protocol error.""" + + +@dataclass(frozen=True) +class EpmdInfo: + """What EPMD told us about a node.""" + + name: str + port: int + node_type: int # 77 normal, 72 hidden + protocol: int # 0 = tcp/ipv4 + highest_version: int # distribution version we can speak + lowest_version: int + extra: bytes = b"" + + +def lookup( + node_name: str, + host: str = "localhost", + port: int = DEFAULT_PORT, + timeout: float = 5.0, +) -> EpmdInfo | None: + """Ask EPMD where ``node_name`` is listening on ``host``. + + Returns ``None`` if EPMD says the node is not registered. Raises + :class:`EpmdError` on socket failures or malformed responses. + + ``node_name`` is the short part only (e.g. ``"testnode"``, not + ``"testnode@host"``). + """ + if "@" in node_name: + raise ValueError(f"pass the short name only, not {node_name!r}") + name_bytes = node_name.encode("utf-8") + req = bytes([PORT_PLEASE2_REQ]) + name_bytes + framed = struct.pack(">H", len(req)) + req + try: + with socket.create_connection((host, port), timeout=timeout) as s: + s.sendall(framed) + data = _recv_all(s) + except OSError as e: + raise EpmdError(f"epmd at {host}:{port} unreachable: {e}") from e + return parse_port2_resp(data) + + +def parse_port2_resp(data: bytes) -> EpmdInfo | None: + """Parse a raw ``PORT2_RESP`` payload. Exposed for tests.""" + if len(data) < 2: + raise EpmdError(f"PORT2_RESP too short: {len(data)} bytes") + if data[0] != PORT2_RESP: + raise EpmdError(f"unexpected response tag: {data[0]} (want {PORT2_RESP})") + result = data[1] + if result != 0: + # Non-zero result means "no such node" or an EPMD-side error. + # The wire only carries the byte; there's no error string. + return None + if len(data) < 12: + raise EpmdError(f"PORT2_RESP OK payload truncated: {len(data)} bytes") + port, ntype, proto, hi, lo, nlen = struct.unpack(">HBBHHH", data[2:12]) + name_end = 12 + nlen + if len(data) < name_end + 2: + raise EpmdError("PORT2_RESP node name truncated") + name = data[12:name_end].decode("utf-8") + (elen,) = struct.unpack(">H", data[name_end:name_end + 2]) + extra_start = name_end + 2 + if len(data) < extra_start + elen: + raise EpmdError("PORT2_RESP extra truncated") + extra = data[extra_start:extra_start + elen] + return EpmdInfo( + name=name, + port=port, + node_type=ntype, + protocol=proto, + highest_version=hi, + lowest_version=lo, + extra=extra, + ) + + +def build_port_please2_req(node_name: str) -> bytes: + """Build a framed ``PORT_PLEASE2_REQ``. Exposed for tests.""" + if "@" in node_name: + raise ValueError(f"pass the short name only, not {node_name!r}") + body = bytes([PORT_PLEASE2_REQ]) + node_name.encode("utf-8") + return struct.pack(">H", len(body)) + body + + +def _recv_all(sock: socket.socket) -> bytes: + """Drain the socket. EPMD closes the connection after one response.""" + chunks = [] + while True: + chunk = sock.recv(4096) + if not chunk: + break + chunks.append(chunk) + return b"".join(chunks) diff --git a/tests/test_epmd.py b/tests/test_epmd.py new file mode 100644 index 0000000..ee5b80e --- /dev/null +++ b/tests/test_epmd.py @@ -0,0 +1,145 @@ +"""EPMD client tests. + +Unit tests run against recorded byte streams captured from a real EPMD +talking to a real named Erlang node (``erl -sname testnode``). The +integration test only runs if an EPMD is reachable on localhost and a +node is registered under the name we ask for. +""" + +from __future__ import annotations + +import shutil +import socket +import subprocess +import time + +import pytest + +from erldistpy.epmd import ( + DEFAULT_PORT, + EpmdError, + EpmdInfo, + build_port_please2_req, + lookup, + parse_port2_resp, +) + +# Captured from a real EPMD answering for `erl -sname testnode`. +# port = 36205, node_type = 77 (normal), proto = 0 (tcp/ipv4), +# hi_ver = 6, lo_ver = 5, name = "testnode", extra = empty. +PORT2_RESP_OK = bytes.fromhex("77008d6d4d00000600050008746573746e6f64650000") + +# Captured from EPMD when the asked-for node is not registered. +PORT2_RESP_NOT_FOUND = bytes.fromhex("7701") + + +def test_build_request_known_shape(): + req = build_port_please2_req("testnode") + # 2-byte length prefix (BE) + tag 122 + name + assert req[:2] == b"\x00\x09" # 9 bytes follow + assert req[2] == 122 # PORT_PLEASE2_REQ + assert req[3:] == b"testnode" + + +def test_build_request_rejects_fully_qualified_name(): + with pytest.raises(ValueError): + build_port_please2_req("testnode@somehost") + + +def test_parse_port2_resp_ok(): + info = parse_port2_resp(PORT2_RESP_OK) + assert info == EpmdInfo( + name="testnode", + port=36205, + node_type=77, + protocol=0, + highest_version=6, + lowest_version=5, + extra=b"", + ) + + +def test_parse_port2_resp_not_found_returns_none(): + assert parse_port2_resp(PORT2_RESP_NOT_FOUND) is None + + +def test_parse_port2_resp_wrong_tag(): + with pytest.raises(EpmdError, match="unexpected response tag"): + parse_port2_resp(b"\x00\x00") + + +def test_parse_port2_resp_truncated(): + with pytest.raises(EpmdError, match="truncated|too short"): + parse_port2_resp(b"\x77\x00\x00") + + +def test_lookup_rejects_fully_qualified_name(): + with pytest.raises(ValueError): + lookup("testnode@somehost") + + +def test_lookup_unreachable_host(): + # Reserved port unlikely to be open + with pytest.raises(EpmdError, match="unreachable"): + lookup("any", host="127.0.0.1", port=1, timeout=0.5) + + +# -------------------------------------------------------------------------- +# Integration — requires a live EPMD + a named Erlang node +# -------------------------------------------------------------------------- + + +def _epmd_running() -> bool: + try: + with socket.create_connection(("localhost", DEFAULT_PORT), timeout=0.5): + return True + except OSError: + return False + + +@pytest.fixture(scope="module") +def named_erlang_node(): + """Spawn `erl -sname testnode`, register with EPMD, tear down after.""" + if not shutil.which("erl"): + pytest.skip("erl not installed") + if not _epmd_running(): + pytest.skip("EPMD not running on localhost") + + proc = subprocess.Popen( + ["erl", "-sname", "erldistpy_itest", "-noshell", "-eval", "timer:sleep(infinity)."], + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + # Wait for the node to register with EPMD + for _ in range(40): + time.sleep(0.1) + info = lookup("erldistpy_itest", timeout=0.5) + if info is not None: + break + else: + proc.terminate() + proc.wait(timeout=2) + pytest.skip("erl node failed to register with EPMD") + + yield "erldistpy_itest" + + proc.terminate() + try: + proc.wait(timeout=2) + except subprocess.TimeoutExpired: + proc.kill() + + +def test_live_lookup_returns_info(named_erlang_node): + info = lookup(named_erlang_node) + assert info is not None + assert info.name == named_erlang_node + assert info.port > 0 + assert info.node_type == 77 + assert info.protocol == 0 + # OTP 23+ speaks at least version 6 + assert info.highest_version >= 5 + + +def test_live_lookup_unknown_node_returns_none(named_erlang_node): + assert lookup("definitely_not_a_real_node_qzxw") is None