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.
This commit is contained in:
parent
b9fd28ca3b
commit
bbb0c316c5
3 changed files with 279 additions and 1 deletions
|
|
@ -2,6 +2,7 @@
|
||||||
|
|
||||||
__version__ = "0.0.1"
|
__version__ = "0.0.1"
|
||||||
|
|
||||||
|
from erldistpy.epmd import EpmdError, EpmdInfo, lookup
|
||||||
from erldistpy.etf import (
|
from erldistpy.etf import (
|
||||||
Atom,
|
Atom,
|
||||||
Pid,
|
Pid,
|
||||||
|
|
@ -10,4 +11,14 @@ from erldistpy.etf import (
|
||||||
encode,
|
encode,
|
||||||
)
|
)
|
||||||
|
|
||||||
__all__ = ["Atom", "Pid", "Reference", "decode", "encode", "__version__"]
|
__all__ = [
|
||||||
|
"Atom",
|
||||||
|
"EpmdError",
|
||||||
|
"EpmdInfo",
|
||||||
|
"Pid",
|
||||||
|
"Reference",
|
||||||
|
"__version__",
|
||||||
|
"decode",
|
||||||
|
"encode",
|
||||||
|
"lookup",
|
||||||
|
]
|
||||||
|
|
|
||||||
122
erldistpy/epmd.py
Normal file
122
erldistpy/epmd.py
Normal file
|
|
@ -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)
|
||||||
145
tests/test_epmd.py
Normal file
145
tests/test_epmd.py
Normal file
|
|
@ -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
|
||||||
Loading…
Add table
Add a link
Reference in a new issue