python/erldistpy
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: python:main
Branches Tags
python:main
python:0.1.7
python:0.1.6
python:0.1.5
python:0.1.4
python:0.1.3
python:0.1.2
python:0.1.1
python:0.1.0
..
compare: python:0.1.5
Branches Tags
python:main
python:0.1.7
python:0.1.6
python:0.1.5
python:0.1.4
python:0.1.3
python:0.1.2
python:0.1.1
python:0.1.0

No commits in common. "main" and "0.1.5" have entirely different histories.
main ... 0.1.5

9 changed files with 21 additions and 865 deletions
Download patch file Download diff file Expand all files Collapse all files

9
.gitlab-ci.yml
View file

@ -51,8 +51,7 @@ pypi-twine:
- pip install build twine
- python -m build
- twine check dist/*
# Twine 6 prefers Trusted Publishing IF TWINE_USERNAME/TWINE_PASSWORD
# are unset. With both set, classic auth is used directly. If the
# diagnostic above shows MISSING, fix the group var Protected flag
# or mark tags as Protected refs on the python/ group.
- twine upload --non-interactive dist/*
# --trusted-publishing never: twine 6 defaults to attempting OIDC
# when it detects GitLab CI, even if TWINE_USERNAME/TWINE_PASSWORD
# are set. Explicitly disable OIDC so it uses the env vars directly.
- twine upload --non-interactive --trusted-publishing never dist/*

2
erldistpy/__init__.py
View file

@ -1,6 +1,6 @@
"""erldistpy — native Python client for our Erlang distribution protocol."""
__version__ = "0.1.6"
__version__ = "0.1.5"
from erldistpy.channel import Channel, ChannelError, IncomingMessage
from erldistpy.epmd import EpmdError, EpmdInfo, lookup

87
erldistpy/channel.py
View file

@ -144,14 +144,7 @@ class Channel:
def _encode_message(control: tuple, payload: object | None) -> bytes:
"""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).
"""
"""Build a distribution message body (no length prefix)."""
body = bytes([PASS_THROUGH]) + encode(control)
if payload is not None:
body += encode(payload)
@ -159,75 +152,23 @@ def _encode_message(control: tuple, payload: object | None) -> bytes:
def _decode_message(body: bytes) -> IncomingMessage:
"""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")
if not body or body[0] != PASS_THROUGH:
raise ChannelError(
f"expected pass-through byte 0x70, got {body[:1].hex() or 'empty'}"
)
try:
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)"
)
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
else:
raise ChannelError(
f"expected pass-through (0x70) or dist header (0x83 0x44), "
f"got {body[:2].hex()}"
)
payload, off = decode_term(body, off)
if off != len(body):
raise ChannelError(f"trailing bytes after payload: {len(body) - off} unread")
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)

14
erldistpy/flags.py
View file

@ -43,19 +43,6 @@ 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
@ -70,5 +57,4 @@ DEFAULT_FLAGS = (
| DFLAG_HANDSHAKE_23
| DFLAG_UNLINK_ID
| DFLAG_V4_NC
| DFLAG_MANDATORY_25_DIGEST
)

6
erldistpy/tls.py
View file

@ -36,7 +36,6 @@ def make_dist_tls_context(
ca: str,
check_hostname: bool = False,
minimum_version: ssl.TLSVersion = ssl.TLSVersion.TLSv1_2,
maximum_version: ssl.TLSVersion = ssl.TLSVersion.MAXIMUM_SUPPORTED,
) -> ssl.SSLContext:
"""Build an SSLContext for a TLS-dist client.
@ -44,16 +43,11 @@ def make_dist_tls_context(
and requires one). ``check_hostname=False`` because dist nodes are
identified by their cookie + cert chain, not by SNI hostname; flip
on if your CA pins per-node CNs.
``maximum_version`` defaults to the highest version Python's ssl
supports (TLS 1.3 in practice). Pin to TLS 1.2 explicitly only if
you hit interop issues against an older Erlang peer.
"""
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
ctx.check_hostname = check_hostname
ctx.verify_mode = ssl.CERT_REQUIRED
ctx.minimum_version = minimum_version
ctx.maximum_version = maximum_version
ctx.load_cert_chain(certfile=cert, keyfile=key)
ctx.load_verify_locations(cafile=ca)
return ctx

2
pyproject.toml
View file

@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "erldistpy"
version = "0.1.7"
version = "0.1.5"
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"

2
tests/test_channel.py
View file

@ -57,7 +57,7 @@ def test_encode_message_with_payload():
def test_decode_rejects_wrong_first_byte():
with pytest.raises(ChannelError, match="pass-through .0x70. or dist header"):
with pytest.raises(ChannelError, match="pass-through byte"):
_decode_message(b"\x00" + encode((1,)))

181
tests/test_node_mix_release.py
View file

@ -1,181 +0,0 @@
"""Mix release + TLS dist test — the final 'is it the same as prod?' check.
Builds a minimal Elixir Mix release containing one GenServer registered
as Wallet.Bridge, runs it in a docker container with TLS dist enabled
using the production-matching inet_tls.conf shape, then exercises
erldistpy.Node.call against it.
Together with test_node_otp26.py this exhausts every dimension we
identified differs between dev box (OTP 24, plain Erlang) and the
production portal beam (OTP 25, Mix release, TLS dist, Elixir
GenServer). If THIS passes too, the production-only failure is
something specific to the live portal beam state that we can't
reproduce in isolation — most likely an interaction with portal's
other dist connections or its larger supervisor tree.
"""
from __future__ import annotations
import os
import shutil
import socket
import subprocess
import time
from pathlib import Path
import pytest
from erldistpy.etf import Atom
from erldistpy.node import Node
from erldistpy.tls import make_dist_tls_context
RELEASE_NAME = "wallet_bridge_test"
COOKIE = "WALLET_BRIDGE_TEST_COOKIE"
MIX_EXS = """\
defmodule WalletBridgeTest.MixProject do
use Mix.Project
def project, do: [
app: :wallet_bridge_test,
version: "0.1.0",
elixir: "~> 1.16",
deps: [],
releases: [wallet_bridge_test: [include_executables_for: [:unix]]]
]
def application, do: [
extra_applications: [:logger, :ssl, :crypto, :public_key],
mod: {WalletBridgeTest.Application, []}
]
end
"""
LIB_EX = """\
defmodule WalletBridgeTest.Application do
use Application
def start(_, _), do: Supervisor.start_link([Wallet.Bridge], strategy: :one_for_one)
end
defmodule Wallet.Bridge do
use GenServer
require Logger
def start_link(_), do: GenServer.start_link(__MODULE__, %{}, name: __MODULE__)
def init(s), do: {:ok, s}
def handle_call(msg, _from, s), do: {:reply, {:ok, msg}, s}
end
"""
def _docker() -> str | None:
return shutil.which("docker")
def _epmd_node_present(sname: str) -> bool:
try:
result = subprocess.run(
["epmd", "-names"], capture_output=True, text=True, timeout=2
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
return f"name {sname} at port" in result.stdout
@pytest.fixture(scope="module")
def mix_release_peer(tmp_path_factory):
if _docker() is None:
pytest.skip("docker not installed")
if not shutil.which("openssl"):
pytest.skip("openssl not installed")
# Must be under $HOME for snap docker to bind-mount it.
base = Path.home() / ".erldistpy-test"
base.mkdir(exist_ok=True)
workdir = base / f"mix_release_{int(time.time())}"
workdir.mkdir()
(workdir / "lib").mkdir()
(workdir / "mix.exs").write_text(MIX_EXS)
(workdir / "lib" / "wallet_bridge.ex").write_text(LIB_EX)
os.chmod(workdir, 0o755)
# Build the release
build = subprocess.run(
["docker", "run", "--rm",
"-v", f"{workdir}:/app", "-w", "/app",
"elixir:1.16-otp-25",
"sh", "-c", "MIX_ENV=prod mix release wallet_bridge_test --overwrite"],
capture_output=True, timeout=180,
)
if build.returncode != 0:
pytest.skip(f"mix release build failed: {build.stderr.decode()[-500:]}")
# Re-use cert generation from test_node_otp26 (we duplicate-import to avoid
# cross-file fixture coupling; if you have it imported already, prefer that.)
from tests.test_node_otp26 import _gen_certs, _write_ssl_config_portal_match
certs = _gen_certs(workdir)
container_certs = {k: v.replace(str(workdir), "/app") for k, v in certs.items()}
_write_ssl_config_portal_match(workdir, container_certs)
# Tweak: ssl_dist_portal.config references /certs paths; rewrite to /app.
cfg_path = workdir / "ssl_dist_portal.config"
cfg_text = cfg_path.read_text().replace("/certs", "/app/certs")
(workdir / "certs").mkdir(exist_ok=True)
for f in ("ca.pem", "server.pem", "server.key", "client.pem", "client.key"):
if (workdir / f).exists():
shutil.copy(workdir / f, workdir / "certs" / f)
cfg_path.write_text(cfg_text)
container_id_file = workdir / "cid"
cid = subprocess.run(
["docker", "run", "--rm", "-d", "--network", "host",
"-v", f"{workdir}:/app", "-w", "/app",
"-e", "RELEASE_DISTRIBUTION=name",
"-e", f"RELEASE_NODE={RELEASE_NAME}@127.0.0.1",
"-e", f"RELEASE_COOKIE={COOKIE}",
"-e", "ERL_FLAGS=-proto_dist inet_tls -ssl_dist_optfile /app/ssl_dist_portal.config",
"elixir:1.16-otp-25",
f"_build/prod/rel/{RELEASE_NAME}/bin/{RELEASE_NAME}", "start"],
capture_output=True, text=True,
).stdout.strip()
container_id_file.write_text(cid)
deadline = time.monotonic() + 20.0
ready = False
while time.monotonic() < deadline:
time.sleep(0.3)
if _epmd_node_present(RELEASE_NAME):
time.sleep(0.5)
ready = True
break
if not ready:
if cid:
subprocess.run(["docker", "kill", cid], capture_output=True)
pytest.skip(f"Mix release {RELEASE_NAME} did not register with epmd")
yield RELEASE_NAME, COOKIE, certs
if cid:
subprocess.run(["docker", "kill", cid], capture_output=True)
def test_call_against_mix_release_tls(mix_release_peer):
"""erldistpy → real Mix release Wallet.Bridge over TLS dist on OTP 25.
Closest local equivalent of the production failure environment.
If THIS passes, the production failure isn't reproducible without
a live portal beam in a mesh — meaning the bug is interaction-level
(other connections, supervisor state) not protocol-level.
"""
sname, cookie, certs = mix_release_peer
ctx = make_dist_tls_context(
cert=certs["client_cert"], key=certs["client_key"], ca=certs["ca"],
)
with Node(
our_name="probe@127.0.0.1",
peer_name=sname,
peer_host="127.0.0.1",
cookie=cookie,
tls_context=ctx,
) as n:
assert n.tls is True
reply = n.call("Elixir.Wallet.Bridge", (Atom("ping"), 42), timeout=8.0)
# Bridge wraps every call as {:ok, msg} — matches what the production
# Wallet.Bridge does on the happy path (modulo the cammy forward).
assert reply == (Atom("ok"), (Atom("ping"), 42))

583
tests/test_node_otp26.py
View file

@ -1,583 +0,0 @@
"""OTP 26 integration tests via Docker.
Reproduces the production failure mode we hit on portal@unsandbox.com:
erldistpy 0.1.7 handshake "succeeds" against OTP 26 but the peer
silently closes the link on the first REG_SEND, surfacing as
``ChannelError: peer closed after 0/4 bytes`` on Node.call().
The pure-Python tests and the existing live tests all run against the
system Erlang, which on most dev boxes is OTP 24 (Ubuntu 22.04 default).
OTP 24 didn't enforce DFLAG_MANDATORY_25_DIGEST or the OTP-25+
mandatory flag set, so flag-negotiation bugs pass silently. These
tests force the modern protocol by running a peer node out of the
official ``erlang:26`` Docker image.
Tests are skipped automatically if Docker isn't available — they're
intentionally NOT part of the smoke that PRs gate on, because they
require docker pull + ~7s of node boot per module.
"""
from __future__ import annotations
import shutil
import socket
import subprocess
import time
import pytest
from pathlib import Path
from erldistpy.etf import Atom
from erldistpy.node import Node
from erldistpy.tls import make_dist_tls_context
SNAME_OTP26 = "erldistpy_otp26"
COOKIE_OTP26 = "ERLDISTPY_OTP26_COOKIE"
# Same gen_target shape as tests/test_node.py — implements the
# gen_server-style {'$gen_call', {From, Ref}, Request} -> From ! {Ref, Reply}
# protocol by hand. Two operations:
# {ping, X} -> {pong, X}
# {add, A, B} -> {ok, A + B}
# Anything else surfaces as {error, {bad_request, _}}.
#
# We also register a heartbeat after 200ms so tests can poll a known
# pid presence rather than depending on a /tmp marker file (which would
# live inside the container, invisible to the host).
ERL_GEN_BOOT = (
"Handle = fun "
" ({ping, X}) -> {pong, X}; "
" ({add, A, B}) -> {ok, A + B}; "
" (Other) -> {error, {bad_request, Other}} "
"end, "
"Loop = fun(F) -> "
" receive "
" {'$gen_call', {From, Ref}, Request} -> "
" From ! {Ref, Handle(Request)}, F(F); "
" Other -> "
" io:format(\"unexpected: ~p~n\", [Other]), F(F) "
" end "
"end, "
"Pid = spawn(fun() -> Loop(Loop) end), "
"register(gen_target, Pid), "
"timer:sleep(infinity)."
)
def _docker_available() -> bool:
return shutil.which("docker") is not None
def _epmd_node_present(sname: str) -> bool:
"""Returns True if ``sname`` is registered in the host's epmd."""
try:
result = subprocess.run(
["epmd", "-names"],
capture_output=True,
text=True,
timeout=2,
)
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
return f"name {sname} at port" in result.stdout
def _socket_open(host: str, port: int) -> bool:
try:
with socket.create_connection((host, port), timeout=0.5):
return True
except OSError:
return False
@pytest.fixture(scope="module")
def otp26_gen_peer():
if not _docker_available():
pytest.skip("docker not installed")
if not _socket_open("127.0.0.1", 4369):
pytest.skip("EPMD not running on 127.0.0.1:4369 (host)")
container_name = f"erldistpy-otp26-{int(time.time())}"
# --network host shares the host's network namespace (Linux only),
# so the OTP 26 node registers with the host's epmd and listens on
# a host-routable interface. epmd lookups from erldistpy on the
# host find the docker node like any other local Erlang node.
proc = subprocess.Popen(
[
"docker",
"run",
"--rm",
"--name",
container_name,
"--network",
"host",
"erlang:25",
"erl",
"-sname",
SNAME_OTP26,
"-setcookie",
COOKIE_OTP26,
"-noshell",
"-eval",
ERL_GEN_BOOT,
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
deadline = time.monotonic() + 15.0
ready = False
while time.monotonic() < deadline:
time.sleep(0.2)
if _epmd_node_present(SNAME_OTP26):
# Allow another 200ms for `register(gen_target, ...)` to land.
time.sleep(0.2)
ready = True
break
if not ready:
subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
proc.terminate()
proc.wait(timeout=2)
pytest.skip(
f"OTP 26 docker node {SNAME_OTP26} did not register with host epmd"
)
yield SNAME_OTP26, COOKIE_OTP26
subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
proc.terminate()
try:
proc.wait(timeout=2)
except subprocess.TimeoutExpired:
proc.kill()
def test_call_ping_otp26(otp26_gen_peer):
"""The minimal repro of the production failure.
erldistpy 0.1.6 silently dropped this call against OTP 26 portal.
0.1.7 added DFLAG_MANDATORY_25_DIGEST; whether THAT was enough is
exactly what this test answers.
"""
sname, cookie = otp26_gen_peer
with Node(our_name="erldistpy_test_otp26@localhost", peer_name=sname, cookie=cookie) as n:
reply = n.call("gen_target", (Atom("ping"), 42), timeout=5.0)
assert reply == (Atom("pong"), 42)
def test_call_add_otp26(otp26_gen_peer):
sname, cookie = otp26_gen_peer
with Node(our_name="erldistpy_test_otp26_add@localhost", peer_name=sname, cookie=cookie) as n:
reply = n.call("gen_target", (Atom("add"), 5, 7), timeout=5.0)
assert reply == (Atom("ok"), 12)
def test_call_sequence_otp26(otp26_gen_peer):
"""Multiple calls on the same dist link survive net_tick window."""
sname, cookie = otp26_gen_peer
with Node(our_name="erldistpy_test_otp26_seq@localhost", peer_name=sname, cookie=cookie) as n:
assert n.call("gen_target", (Atom("ping"), Atom("a")), timeout=5.0) == (Atom("pong"), Atom("a"))
assert n.call("gen_target", (Atom("ping"), Atom("b")), timeout=5.0) == (Atom("pong"), Atom("b"))
assert n.call("gen_target", (Atom("add"), 1, 2), timeout=5.0) == (Atom("ok"), 3)
# ---------------------------------------------------------------------------
# OTP 26 + TLS dist — the production combo (mTLS over Erlang dist)
# ---------------------------------------------------------------------------
TLS_SNAME = "erldistpy_otp26_tls"
TLS_COOKIE = "ERLDISTPY_OTP26_TLS_COOKIE"
def _gen_certs(workdir: Path) -> dict[str, str]:
"""Generate a CA + leaf cert with a SAN matching localhost. Same shape
as tests/test_tls.py — duplicated locally to keep this file self-
contained (the docker container needs to read from a known path)."""
import os
hostname = socket.gethostname()
ca_cnf = workdir / "ca.cnf"
ca_cnf.write_text(
"[req]\n"
"distinguished_name = dn\n"
"x509_extensions = v3_ca\n"
"prompt = no\n"
"[dn]\n"
"CN = erldistpy-otp26-ca\n"
"[v3_ca]\n"
"basicConstraints = critical,CA:TRUE\n"
"keyUsage = critical,keyCertSign,cRLSign\n"
"subjectKeyIdentifier = hash\n"
)
subprocess.run(
["openssl", "req", "-x509", "-newkey", "rsa:2048", "-nodes",
"-keyout", str(workdir / "ca.key"),
"-out", str(workdir / "ca.pem"),
"-days", "1",
"-config", str(ca_cnf), "-extensions", "v3_ca"],
check=True, capture_output=True,
)
leaf_cnf = workdir / "leaf.cnf"
leaf_cnf.write_text(
"[v3]\n"
"basicConstraints = CA:FALSE\n"
"keyUsage = digitalSignature,keyEncipherment\n"
"extendedKeyUsage = serverAuth,clientAuth\n"
"subjectAltName = @alt\n"
"[alt]\n"
f"DNS.1 = {hostname}\n"
"DNS.2 = localhost\n"
)
for name in ("server", "client"):
subprocess.run(
["openssl", "req", "-newkey", "rsa:2048", "-nodes",
"-keyout", str(workdir / f"{name}.key"),
"-out", str(workdir / f"{name}.csr"),
"-subj", f"/CN={hostname}"],
check=True, capture_output=True,
)
subprocess.run(
["openssl", "x509", "-req",
"-in", str(workdir / f"{name}.csr"),
"-CA", str(workdir / "ca.pem"),
"-CAkey", str(workdir / "ca.key"),
"-CAcreateserial",
"-out", str(workdir / f"{name}.pem"),
"-days", "1",
"-extfile", str(leaf_cnf), "-extensions", "v3"],
check=True, capture_output=True,
)
os.chmod(workdir / f"{name}.key", 0o644) # readable by container's erl uid
return {
"ca": str(workdir / "ca.pem"),
"server_cert": str(workdir / "server.pem"),
"server_key": str(workdir / "server.key"),
"client_cert": str(workdir / "client.pem"),
"client_key": str(workdir / "client.key"),
}
def _write_ssl_config(workdir: Path, certs_in_container: dict[str, str]) -> str:
cfg = workdir / "ssl_dist.config"
body = (
"[{server, "
f'[{{certfile, "{certs_in_container["server_cert"]}"}}, '
f'{{keyfile, "{certs_in_container["server_key"]}"}}, '
f'{{cacertfile, "{certs_in_container["ca"]}"}}, '
"{verify, verify_peer}, "
"{fail_if_no_peer_cert, true}]}, "
"{client, "
f'[{{certfile, "{certs_in_container["server_cert"]}"}}, '
f'{{keyfile, "{certs_in_container["server_key"]}"}}, '
f'{{cacertfile, "{certs_in_container["ca"]}"}}, '
"{verify, verify_peer}]}]."
)
cfg.write_text(body)
return str(cfg)
def _write_ssl_config_portal_match(workdir: Path, certs_in_container: dict[str, str]) -> str:
"""Match production portal's /opt/unsandbox/certs/inet_tls.conf exactly:
permissive verify_fun (accepts bad_cert), TLS 1.2/1.3 only, secure
renegotiate. This is the shape the production failure runs under."""
cfg = workdir / "ssl_dist_portal.config"
permissive_verify_fun = (
"{verify_fun, {fun(_,{bad_cert, _}, UserState) -> {valid, UserState}; "
" (_,{extension, _}, UserState) -> {unknown, UserState}; "
" (_, valid, UserState) -> {valid, UserState}; "
" (_, valid_peer, UserState) -> {valid, UserState} "
" end, []}}"
)
body = (
"[{server, "
f'[{{certfile, "{certs_in_container["server_cert"]}"}}, '
f'{{keyfile, "{certs_in_container["server_key"]}"}}, '
f'{{cacertfile, "{certs_in_container["ca"]}"}}, '
"{verify, verify_peer}, "
"{fail_if_no_peer_cert, true}, "
f"{permissive_verify_fun}, "
"{secure_renegotiate, true}, "
"{versions, ['tlsv1.3', 'tlsv1.2']}]}, "
"{client, "
f'[{{certfile, "{certs_in_container["server_cert"]}"}}, '
f'{{keyfile, "{certs_in_container["server_key"]}"}}, '
f'{{cacertfile, "{certs_in_container["ca"]}"}}, '
"{verify, verify_peer}, "
"{server_name_indication, disable}, "
f"{permissive_verify_fun}, "
"{secure_renegotiate, true}, "
"{versions, ['tlsv1.3', 'tlsv1.2']}]}]."
)
cfg.write_text(body)
return str(cfg)
@pytest.fixture(scope="module")
def otp26_tls_peer(tmp_path_factory):
"""OTP 26 + TLS dist peer in Docker — the production combo.
This is the variant that reproduces (or should reproduce) the
portal@unsandbox.com failure: handshake "succeeds" then the peer
silently closes the link on the first REG_SEND.
"""
if not _docker_available():
pytest.skip("docker not installed")
if not shutil.which("openssl"):
pytest.skip("openssl not installed")
if not _socket_open("127.0.0.1", 4369):
pytest.skip("EPMD not running on host")
# Snap-confined Docker (canonical's snap package) can't bind-mount
# /tmp — use a path under $HOME which the confinement allows. Caller
# could override via ERLDISTPY_TLS_OTP26_WORKDIR.
import os as _os
import tempfile as _tempfile
home_base = _os.environ.get(
"ERLDISTPY_TLS_OTP26_WORKDIR", str(Path.home() / ".erldistpy-test")
)
Path(home_base).mkdir(parents=True, exist_ok=True)
workdir = Path(_tempfile.mkdtemp(prefix="tls_otp26_", dir=home_base))
_os.chmod(workdir, 0o755)
host_certs = _gen_certs(workdir)
# Inside the container the certs land at the same path because we
# mount workdir → /certs:ro and we generated under workdir.
container_certs = {
k: v.replace(str(workdir), "/certs") for k, v in host_certs.items()
}
ssl_config_host = _write_ssl_config(workdir, container_certs)
ssl_config_container = ssl_config_host.replace(str(workdir), "/certs")
container_name = f"erldistpy-otp26-tls-{int(time.time())}"
proc = subprocess.Popen(
[
"docker", "run", "--rm",
"--name", container_name,
"--network", "host",
"-v", f"{workdir}:/certs:ro",
"erlang:25",
"erl",
"-sname", TLS_SNAME,
"-setcookie", TLS_COOKIE,
"-proto_dist", "inet_tls",
"-ssl_dist_optfile", ssl_config_container,
"-noshell",
"-eval", ERL_GEN_BOOT,
],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
deadline = time.monotonic() + 20.0
ready = False
while time.monotonic() < deadline:
time.sleep(0.3)
if _epmd_node_present(TLS_SNAME):
time.sleep(0.3)
ready = True
break
if not ready:
subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
proc.terminate()
proc.wait(timeout=3)
pytest.skip(f"OTP 26 TLS docker node {TLS_SNAME} did not register")
yield TLS_SNAME, TLS_COOKIE, host_certs
subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
proc.terminate()
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
proc.kill()
def test_call_over_tls_otp26(otp26_tls_peer):
"""The minimum repro of the production failure: TLS-dist on OTP 26.
On erldistpy ≤0.1.7 against the prod portal this surfaces as
``ChannelError: peer closed after 0/4 bytes`` on the first call.
"""
sname, cookie, certs = otp26_tls_peer
ctx = make_dist_tls_context(
cert=certs["client_cert"],
key=certs["client_key"],
ca=certs["ca"],
)
with Node(
our_name="erldistpy_test_otp26_tls@localhost",
peer_name=sname,
cookie=cookie,
tls_context=ctx,
) as n:
assert n.tls is True
reply = n.call("gen_target", (Atom("ping"), 7), timeout=5.0)
assert reply == (Atom("pong"), 7)
# ---------------------------------------------------------------------------
# OTP 26 + TLS dist + Elixir GenServer — the exact production combo
# ---------------------------------------------------------------------------
ELIXIR_TLS_SNAME = "erldistpy_otp26_elixir_tls"
ELIXIR_TLS_COOKIE = "ERLDISTPY_ELIXIR_TLS"
# Inline Elixir script: spawn a GenServer registered as `gen_target` that
# answers the same {ping, X} / {add, A, B} calls. This is the SHAPE the
# real Wallet.Bridge uses — GenServer.handle_call/3 routing via the
# `:"$gen_call"` envelope, not a hand-coded receive.
ELIXIR_GENSERVER_BOOT = """
defmodule GenTarget do
use GenServer
def start_link, do: GenServer.start_link(__MODULE__, %{}, name: :gen_target)
def init(state) do
IO.puts("[gen_target] init")
{:ok, state}
end
def handle_call(msg, from, state) do
IO.puts("[gen_target] handle_call msg=#{inspect(msg)} from=#{inspect(from)}")
reply = case msg do
{:ping, x} -> {:pong, x}
{:add, a, b} -> {:ok, a + b}
other -> {:error, {:bad_request, other}}
end
{:reply, reply, state}
end
def handle_info(msg, state) do
IO.puts("[gen_target] handle_info msg=#{inspect(msg)}")
{:noreply, state}
end
end
{:ok, _} = GenTarget.start_link()
IO.puts("[gen_target] registered: #{inspect(Process.whereis(:gen_target))}")
Process.sleep(:infinity)
"""
@pytest.fixture(scope="module")
def otp26_elixir_tls_peer(tmp_path_factory):
"""Elixir GenServer over TLS on OTP 26 — exactly what portal runs.
Difference from otp26_tls_peer: the registered process is a real
Elixir GenServer (uses `gen_server:reply/2` for replies, all the
OTP machinery) instead of a hand-coded receive loop.
"""
if not _docker_available():
pytest.skip("docker not installed")
if not shutil.which("openssl"):
pytest.skip("openssl not installed")
if not _socket_open("127.0.0.1", 4369):
pytest.skip("EPMD not running on host")
import os as _os
import tempfile as _tempfile
home_base = _os.environ.get(
"ERLDISTPY_TLS_OTP26_WORKDIR", str(Path.home() / ".erldistpy-test")
)
Path(home_base).mkdir(parents=True, exist_ok=True)
workdir = Path(_tempfile.mkdtemp(prefix="elixir_tls_otp26_", dir=home_base))
_os.chmod(workdir, 0o755)
host_certs = _gen_certs(workdir)
container_certs = {
k: v.replace(str(workdir), "/certs") for k, v in host_certs.items()
}
# Use portal's exact inet_tls.conf shape (permissive verify_fun,
# TLS 1.2/1.3 only, secure_renegotiate) so the test reproduces the
# exact wire-level conditions of the production failure.
_write_ssl_config_portal_match(workdir, container_certs)
ssl_config_container = "/certs/ssl_dist_portal.config"
boot_script = workdir / "boot.exs"
boot_script.write_text(ELIXIR_GENSERVER_BOOT)
container_name = f"erldistpy-elixir-tls-{int(time.time())}"
container_log = workdir / "container.log"
# Long node names (--name) match production. Short names (--sname)
# would route differently through the dist driver and may not
# reproduce the same failure mode.
long_node_name = f"{ELIXIR_TLS_SNAME}@127.0.0.1"
proc = subprocess.Popen(
[
"docker", "run", "--rm",
"--name", container_name,
"--network", "host",
"-v", f"{workdir}:/certs:ro",
"elixir:1.16-otp-25",
"elixir",
"--name", long_node_name,
"--cookie", ELIXIR_TLS_COOKIE,
"--erl", f"-proto_dist inet_tls -ssl_dist_optfile {ssl_config_container}",
"/certs/boot.exs",
],
stdout=open(container_log, "w"),
stderr=subprocess.STDOUT,
)
deadline = time.monotonic() + 25.0
ready = False
while time.monotonic() < deadline:
time.sleep(0.3)
if _epmd_node_present(ELIXIR_TLS_SNAME):
time.sleep(0.5) # Elixir GenServer needs an extra beat to register
ready = True
break
if not ready:
subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
proc.terminate()
proc.wait(timeout=3)
pytest.skip(f"Elixir TLS docker node {ELIXIR_TLS_SNAME} did not register")
yield ELIXIR_TLS_SNAME, ELIXIR_TLS_COOKIE, host_certs, container_log
subprocess.run(["docker", "rm", "-f", container_name], capture_output=True)
proc.terminate()
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
proc.kill()
def test_call_elixir_genserver_over_tls_otp26(otp26_elixir_tls_peer):
"""Repro of the prod failure: erldistpy calls an Elixir GenServer over TLS.
Difference from test_call_over_tls_otp26: peer is a real
GenServer, not a hand-coded receive. Replies go through
:gen_server.reply/2 which sends via :erlang.send/2 — same path as
Wallet.Bridge in production.
"""
sname, cookie, certs, container_log = otp26_elixir_tls_peer
ctx = make_dist_tls_context(
cert=certs["client_cert"],
key=certs["client_key"],
ca=certs["ca"],
)
try:
with Node(
our_name="erldistpy_test_elixir_tls@localhost",
peer_name=sname,
cookie=cookie,
tls_context=ctx,
) as n:
assert n.tls is True
reply = n.call("gen_target", (Atom("ping"), 7), timeout=5.0)
assert reply == (Atom("pong"), 7)
finally:
# Whether pass or fail, dump container output for diagnosis.
try:
text = container_log.read_text()
print(f"=== container stdout ===\n{text}\n=== end ===")
except FileNotFoundError:
pass