net_runner

1.2.0

Safe OS process execution for Elixir with NIF-based backpressure, zero zombie processes, PTY support, and cgroup isolation.

HexDocs

Current section

6 Versions

Jump to

Readme 6 Versions 1 Dependency 1 Dependant Files Activity

Hide whitespace

Compare versions

5 of 16 files loaded

16 files changed

+852 additions

-211 deletions

	@@ -4,6 +4,116 @@ All notable changes to this project will be documented in this file.
4 4
5 5	This project adheres to [Semantic Versioning](https://semver.org/).
6 6
7	+ ## [Unreleased]
8	+
9	+ Focused code-review pass across the NIF, shepherd, and Elixir layers.
10	+ Correctness-first: closes two real-world race/leak bugs, hardens the
11	+ post-fork child window, and adds an AddressSanitizer + UBSan CI job.
12	+
13	+ ### Fixed
14	+
15	+ - FD leak in `nif_create_fd` when `enif_mutex_create` failed
16	+ — the destructor previously gated `close(fd)` on a non-NULL lock,
17	+ so a failed mutex allocation leaked the file descriptor and armed
18	+ a NULL-deref in any later `nif_close`. The mutex result is checked
19	+ and the dtor now closes the fd unconditionally.
20	+ - Use-after-close race in NIF read/write vs. close/down
21	+ — `nif_read`/`nif_write` copied `res->fd` under the mutex and
22	+ released the lock before the syscall; a concurrent `nif_close` or
23	+ owner-death callback could close the fd before the syscall ran,
24	+ letting the read/write target a recycled fd. The mutex is now held
25	+ across the syscall and the subsequent `enif_select` registration;
26	+ the actual `close()` is deferred to the `io_resource_stop` callback
27	+ so BEAM can drain pending selects before the fd is released.
28	+ - Lost initial stderr chunk in `:consume` mode
29	+ — `kick_stderr_read` in `init/1` sent `{:stderr_data, data}` to
30	+ `self()` but no `handle_info/2` clause matched, so the first (and
31	+ often only) chunk of stderr for fast-exiting processes was silently
32	+ dropped. The missing handler now appends to the stderr buffer and
33	+ drains any remainder.
34	+ - `write_loop` spin on `{:ok, 0}` — if the kernel ever returned
35	+ 0 bytes on a non-empty write, the GenServer would recurse forever
36	+ on the dirty scheduler. Bounded with a 1 ms sleep-retry.
37	+ - Shepherd UDS command framing — the event loop parsed only
38	+ `buf[0]`, discarding any coalesced or tail commands (e.g.
39	+ `CMD_CLOSE_STDIN` followed immediately by `CMD_KILL`). Frames are
40	+ now length-dispatched per opcode with a carry-over buffer across
41	+ `poll()` iterations.
42	+ - Post-fork child stdio and signal safety — replaced `fprintf` /
43	+ `strerror` in the post-fork / pre-exec window with a `write(2)`-
44	+ based `child_fail()` helper (async-signal-safe). Every `dup2`,
45	+ `setsid`, and `TIOCSCTTY` return is now checked; on failure the
46	+ child exits 127 with a diagnostic instead of running with broken
47	+ stdio.
48	+ - `waitpid` after SIGKILL — replaced the unbounded
49	+ `waitpid(child_pid, NULL, 0)` with a bounded WNOHANG loop
50	+ (~3 s cap) so the shepherd cannot hang on a child stuck in
51	+ uninterruptible kernel sleep (D-state).
52	+ - SIGCHLD reap loop — reap all pending children per SIGCHLD
53	+ (`while waitpid(-1, ..., WNOHANG) > 0`) so a coalesced signal
54	+ never leaks zombies.
55	+ - Cgroup / UDS path hardening — validate every `snprintf` return,
56	+ reject too-long UDS paths, set `FD_CLOEXEC` on the PTY master,
57	+ treat user-requested cgroup setup failure as fatal, and replace
58	+ the fixed 100 ms `usleep` in `cgroup_cleanup` with a bounded
59	+ polling `rmdir`.
60	+ - `Stream` consumer crash cleanup — `Stream.resource`'s `after`
61	+ callback is only run on normal termination. A consumer crash
62	+ orphaned the `NetRunner.Process` GenServer and its OS child.
63	+ `NetRunner.Process.start/3` now accepts an `:owner` option that
64	+ monitors the caller; `NetRunner.Stream.stream/3` passes `self()`,
65	+ so a consumer crash SIGKILLs the OS process and stops the
66	+ GenServer.
67	+ - Watcher blocking on `Process.sleep` — the 5 s sleep in
68	+ `handle_info/2` wedged the Watcher unresponsive (including to
69	+ supervisor shutdown). Replaced with `Process.send_after/3` and a
70	+ new `:escalate_to_sigkill` handler.
71	+ - Parked-caller tracking in `Operations` — callers parked on
72	+ EAGAIN are now `Process.monitor/1`-ed; dead callers are pruned on
73	+ `:DOWN` instead of lingering in the pending map until process
74	+ exit.
75	+ - `read_uds_message` race — replaced the `:peek` + full-recv
76	+ pattern (which could time out if the payload arrived a moment
77	+ after the opcode) with an opcode-first read flow and longer
78	+ timeouts.
79	+ - `cmd` / `args` validation — reject non-binary, empty, or
80	+ NUL-containing cmd and args at the spawn boundary. Passing NUL
81	+ bytes through `Port.open`'s `args:` is undefined on the C side.
82	+ - `NetRunner.run/2` error surface — previously pattern-matched
83	+ `{:ok, pid}` from `Proc.start`, raising `MatchError` when
84	+ validation failed. Now returns `{:error, reason}` cleanly.
85	+ - `File.rm` cleanup of UDS socket — tolerate `:enoent`
86	+ (shepherd may have unlinked), propagate other errors.
87	+ - `Signal.resolve` integer range — integer signals outside
88	+ POSIX `1..31` now return `{:error, :unknown_signal}` instead of
89	+ being forwarded to `kill(2)`.
90	+ - `Signal` single source of truth — `Signal.resolve` delegates
91	+ to the NIF for known-atom lookup instead of maintaining a duplicate
92	+ allow-list that drifted from the C side.
93	+ - Daemon drain resilience — drain-task crashes used to match a
94	+ catch-all `:DOWN` handler and silently stop draining; the pipe
95	+ then filled until the child blocked. Narrowed to recognised refs
96	+ with a warning log; `drain_loop` wrapped in `try/rescue/catch` so
97	+ a reader or logger exception cannot take the daemon down through
98	+ the linked Task.
99	+ - `terminate/2` explicitly closes the shepherd `Port` after the
100	+ UDS socket for deterministic teardown order.
101	+
102	+ ### Added
103	+
104	+ - AddressSanitizer + UBSan — opt-in build via `SANITIZE=1 make all`
105	+ or `make asan`. New CI job (`sanitizers`) rebuilds the NIF and
106	+ shepherd with `-fsanitize=address,undefined`, preloads `libasan`,
107	+ and runs the full `mix test`. The publish job depends on it.
108	+ - Stale UDS socket sweep in `test/test_helper.exs` (before and
109	+ after the suite) — stops accumulation from test crashes before
110	+ `cleanup_listener/2` runs.
111	+ - Regression tests for: NUL-byte validation in `cmd` and `args`,
112	+ `Signal.resolve` range + type handling, `:owner` monitor SIGKILL
113	+ path, stderr-only fast-exit stats, binary-with-NUL round-trip, and
114	+ `NetRunner.run` / `NetRunner.stream` returning validation errors
115	+ cleanly.
116	+
7 117	## [1.0.0] - 2026-02-26
8 118
9 119	Initial release.

	@@ -16,7 +16,21 @@ ERL_INTERFACE_LIB_DIR ?= $(shell erl -noshell -eval "io:format(\"~ts\", [code:li
16 16	UNAME_S := $(shell uname -s)
17 17
18 18	CC ?= cc
19	- CFLAGS_BASE = -O2 -Wall -Wextra -Werror -std=c99 -fstack-protector-strong -D_FORTIFY_SOURCE=2
19	+
20	+ # Opt-in sanitizer build. Usage:
21	+ # make clean && SANITIZE=1 make all
22	+ # mix test (from Elixir — the NIF and shepherd are rebuilt with ASan/UBSan)
23	+ #
24	+ # Requires LD_PRELOAD of libasan at runtime on Linux when the BEAM isn't
25	+ # built with sanitizers; see ci.yml for the invocation.
26	+ ifeq ($(SANITIZE),1)
27	+ # _FORTIFY_SOURCE is incompatible with ASan (ASan already intercepts
28	+ # memcpy/etc.). Disable optimisation to -O1 and skip FORTIFY.
29	+ SAN_FLAGS = -fsanitize=address,undefined -fno-omit-frame-pointer -g
30	+ CFLAGS_BASE = -O1 -Wall -Wextra -Werror -std=c99 -fstack-protector-strong $(SAN_FLAGS)
31	+ else
32	+ CFLAGS_BASE = -O2 -Wall -Wextra -Werror -std=c99 -fstack-protector-strong -D_FORTIFY_SOURCE=2
33	+ endif
20 34
21 35	ifeq ($(UNAME_S),Darwin)
22 36	# macOS needs _DARWIN_C_SOURCE for SCM_RIGHTS, CMSG_SPACE, etc.
	@@ -31,6 +45,11 @@ else
31 45	NIF_EXT = .so
32 46	endif
33 47
48	+ ifeq ($(SANITIZE),1)
49	+ NIF_LDFLAGS += $(SAN_FLAGS)
50	+ SHEPHERD_LDFLAGS += $(SAN_FLAGS)
51	+ endif
52	+
34 53	NIF_CFLAGS = $(CFLAGS) -I$(ERTS_INCLUDE_DIR) -I$(C_SRC_DIR) -fPIC
35 54
36 55	# Targets
	@@ -45,10 +64,15 @@ NIF_OBJ = $(C_SRC_DIR)/net_runner_nif.o
45 64
46 65	HEADERS = $(C_SRC_DIR)/protocol.h $(C_SRC_DIR)/utils.h
47 66
48	- .PHONY: all clean
67	+ .PHONY: all clean asan
49 68
50 69	all: $(PRIV_DIR) $(SHEPHERD) $(NIF_LIB)
51 70
71	+ # Convenience: force a sanitizer rebuild. Same as SANITIZE=1 make clean all.
72	+ asan:
73	+ $(MAKE) clean
74	+ $(MAKE) SANITIZE=1 all
75	+
52 76	$(PRIV_DIR):
53 77	mkdir -p $(PRIV_DIR)

	@@ -83,6 +83,45 @@ Enum.to_list(stream)
83 83	NetRunner.run(~w(my_server), kill_timeout: 2000, timeout: 10_000)
84 84	```
85 85
86	+ ## Input Validation and Error Returns
87	+
88	+ `run/2` and `stream/2` return tagged errors for bad input instead of
89	+ crashing. NUL bytes inside `cmd` or `args` are rejected early (they
90	+ are undefined in `argv` on the C side).
91	+
92	+ ```elixir
93	+ # Empty executable
94	+ {:error, {:invalid_cmd, _}} = NetRunner.run([""])
95	+
96	+ # NUL byte in an argument
97	+ {:error, {:invalid_args, _}} = NetRunner.run(["echo", "he\0llo"])
98	+
99	+ # Same behaviour for streaming
100	+ {:error, {:invalid_args, _}} = NetRunner.stream(["echo", "he\0llo"])
101	+
102	+ # Unknown signal atoms come back as tagged errors, not raises
103	+ {:error, :unknown_signal} = NetRunner.Signal.resolve(:sigwhatever)
104	+ {:error, :unknown_signal} = NetRunner.Signal.resolve(99)
105	+ ```
106	+
107	+ ## Working with Binary Output
108	+
109	+ stdout is delivered as a BEAM binary, not a String. It is safe to pass
110	+ bytes containing NUL, high-bit, or anything else through the pipeline.
111	+
112	+ ```elixir
113	+ # NUL bytes round-trip unchanged
114	+ {out, 0} = NetRunner.run(["sh", "-c", ~S\|printf 'a\0b\0c'\|])
115	+ byte_size(out) # => 5
116	+ out == "a\0b\0c" # => true
117	+
118	+ # UTF-8 boundaries straddle chunks fine — just concatenate and then
119	+ # decode.
120	+ "héllo\n" =
121	+ NetRunner.stream!(~w(echo héllo))
122	+ \|> Enum.join()
123	+ ```
124	+
86 125	## Process API
87 126
88 127	For fine-grained control over the OS process lifecycle:
	@@ -164,12 +203,45 @@ Proc.await_exit(pid)
164 203	stats = Proc.stats(pid)
165 204	stats.bytes_in # => 5 (bytes written to stdin)
166 205	stats.bytes_out # => 5 (bytes read from stdout)
206	+ stats.bytes_err # => 0 (bytes read from stderr, :consume mode)
167 207	stats.read_count # => 1 (number of read calls)
168 208	stats.write_count # => 1 (number of write calls)
169 209	stats.duration_ms # => 3 (wall-clock time)
170 210	stats.exit_status # => 0 (exit code)
171 211	```
172 212
213	+ ### Tying an OS process to an owner
214	+
215	+ If the calling process crashes, the OS process it launched should go
216	+ with it. Pass `:owner` to have the Process GenServer monitor a pid;
217	+ on `:DOWN` it SIGKILLs the child and stops cleanly. `NetRunner.stream/2`
218	+ does this automatically with `self()`.
219	+
220	+ ```elixir
221	+ # Spawn a long-lived command tied to the caller
222	+ parent = self()
223	+
224	+ spawn(fn ->
225	+ {:ok, pid} = Proc.start("sleep", ["30"], owner: self())
226	+ send(parent, {:os_pid, Proc.os_pid(pid)})
227	+ exit(:boom) # caller dies → Process SIGKILLs sleep, stops itself
228	+ end)
229	+ ```
230	+
231	+ ### Per-call kill timeout
232	+
233	+ Tune the SIGTERM→SIGKILL escalation window per-process. Useful when a
234	+ command has its own graceful shutdown hook you want to honour, or when
235	+ you need a fast hard-kill.
236	+
237	+ ```elixir
238	+ # Give my_server 10s to drain on SIGTERM before SIGKILL
239	+ {:ok, pid} = Proc.start("my_server", [], kill_timeout: 10_000)
240	+
241	+ # Or make it effectively immediate for tests
242	+ {:ok, pid} = Proc.start("sleep", ["100"], kill_timeout: 100)
243	+ ```
244	+
173 245	## PTY Mode
174 246
175 247	Run commands with a pseudo-terminal for programs that require a TTY. PTY mode is designed for interactive and long-running programs — shells, REPLs, curses apps.
	@@ -268,6 +340,59 @@ Isolate child processes in a cgroup v2 hierarchy for resource control:
268 340
269 341	The shepherd creates the cgroup directory, moves the child into it, and cleans up on exit (kills all processes via `cgroup.kill`, then removes the directory). No-op on macOS.
270 342
343	+ ## Command DSL
344	+
345	+ Bundle an executable, default args, and default options into a reusable
346	+ `%NetRunner.Command{}`. Both `run/2` and `stream/2` accept it, and
347	+ call-site options override the defaults.
348	+
349	+ ```elixir
350	+ alias NetRunner.Command
351	+
352	+ # Inline construction
353	+ cmd = Command.new("curl", ["-sS"], timeout: 30_000)
354	+ {body, 0} = NetRunner.run(cmd, args: ["https://example.com"])
355	+
356	+ # Extend at call time (args append; opts merge with runtime winning)
357	+ listing = Command.new("ls", ["-la"])
358	+ {out, 0} = NetRunner.run(listing, args: ["/tmp"])
359	+
360	+ # `defcommand` in your own module captures a reusable template:
361	+ defmodule MyCmds do
362	+ use NetRunner.Command
363	+
364	+ defcommand :curl, "curl", ["-sS", "--max-time", "30"]
365	+ defcommand :echo, "echo"
366	+ end
367	+
368	+ {out, 0} = NetRunner.run(MyCmds.echo(["hi"]))
369	+ {:ok, stream} = NetRunner.stream(MyCmds.curl(["https://example.com"]))
370	+ ```
371	+
372	+ ## Error Handling Cheatsheet
373	+
374	+ ```elixir
375	+ case NetRunner.run(["my_tool", arg], timeout: 5_000) do
376	+ {output, 0} ->
377	+ {:ok, output}
378	+
379	+ {_partial, status} when status != 0 ->
380	+ {:error, {:nonzero_exit, status}}
381	+
382	+ {:error, :timeout} ->
383	+ {:error, :took_too_long}
384	+
385	+ {:error, {:max_output_exceeded, partial}} ->
386	+ {:error, {:too_much_output, byte_size(partial)}}
387	+
388	+ {:error, {:invalid_cmd, msg}} ->
389	+ {:error, {:bad_cmd, msg}}
390	+
391	+ {:error, {:invalid_args, msg}} ->
392	+ {:error, {:bad_args, msg}}
393	+ end
394	+ ```
395	+
271 396	## Parallel Execution
272 397
273 398	Every NetRunner process is fully independent — no shared state, no singleton bottleneck:

	@@ -38,13 +38,17 @@ static ErlNifResourceType *io_resource_type = NULL;
38 38	static void io_resource_dtor(ErlNifEnv env, void obj) {
39 39	(void)env;
40 40	io_resource_t res = (io_resource_t )obj;
41	+ /* Close fd even if mutex construction failed — otherwise ENOMEM during
42	+ * nif_create_fd would leak the underlying FD. */
41 43	if (res->lock) {
42 44	enif_mutex_lock(res->lock);
43	- if (!res->closed && res->fd >= 0) {
44	- close(res->fd);
45	- res->fd = -1;
46	- res->closed = 1;
47	- }
45	+ }
46	+ if (!res->closed && res->fd >= 0) {
47	+ close(res->fd);
48	+ res->fd = -1;
49	+ res->closed = 1;
50	+ }
51	+ if (res->lock) {
48 52	enif_mutex_unlock(res->lock);
49 53	enif_mutex_destroy(res->lock);
50 54	res->lock = NULL;
	@@ -55,30 +59,36 @@ static void io_resource_stop(ErlNifEnv env, void obj, ErlNifEvent event,
55 59	int is_direct_call) {
56 60	(void)env;
57 61	(void)obj;
58	- (void)event;
59 62	(void)is_direct_call;
60	- /* enif_select stop callback - FD is being deselected */
63	+ /* BEAM guarantees no further use of this event by the NIF is in flight
64	+ * when this callback runs. Safe to close the underlying fd here. */
65	+ if ((int)event >= 0) {
66	+ close((int)event);
67	+ }
61 68	}
62 69
63 70	static void io_resource_down(ErlNifEnv env, void obj, ErlNifPid *pid,
64 71	ErlNifMonitor *mon) {
65	- (void)env;
66 72	(void)pid;
67 73	(void)mon;
68 74	io_resource_t res = (io_resource_t )obj;
69	- /* Owner process died - close the FD */
75	+ int fd_to_stop = -1;
70 76	if (res->lock) {
71 77	enif_mutex_lock(res->lock);
72 78	if (!res->closed && res->fd >= 0) {
73	- enif_select(env, (ErlNifEvent)res->fd, ERL_NIF_SELECT_STOP,
74	- obj, NULL, enif_make_atom(env, "undefined"));
75	- close(res->fd);
79	+ fd_to_stop = res->fd;
76 80	res->fd = -1;
77 81	res->closed = 1;
78 82	}
79 83	res->monitor_active = 0;
80 84	enif_mutex_unlock(res->lock);
81 85	}
86	+ if (fd_to_stop >= 0) {
87	+ /* Hand fd off to the stop callback — it will close it after any
88	+ * in-flight enif_select completes. */
89	+ enif_select(env, (ErlNifEvent)fd_to_stop, ERL_NIF_SELECT_STOP,
90	+ obj, NULL, enif_make_atom(env, "undefined"));
91	+ }
82 92	}
83 93
84 94	static ErlNifResourceTypeInit io_resource_init = {
	@@ -167,6 +177,13 @@ static ERL_NIF_TERM nif_create_fd(ErlNifEnv *env, int argc,
167 177	res->owner = owner;
168 178	res->monitor_active = 0;
169 179
180	+ if (!res->lock) {
181	+ /* Mutex allocation failed — release resource (dtor will close fd) */
182	+ enif_release_resource(res);
183	+ return enif_make_tuple2(env, atom_error,
184	+ MAKE_ATOM(env, "mutex_failed"));
185	+ }
186	+
170 187	/* Monitor the owner process */
171 188	if (enif_monitor_process(env, res, &owner, &res->monitor) == 0) {
172 189	res->monitor_active = 1;
	@@ -199,42 +216,51 @@ static ERL_NIF_TERM nif_read(ErlNifEnv *env, int argc,
199 216	}
200 217	if (max_bytes > 1048576) max_bytes = 1048576; /* Cap at 1MB */
201 218
202	- enif_mutex_lock(res->lock);
203	- if (res->closed) {
204	- enif_mutex_unlock(res->lock);
205	- return enif_make_tuple2(env, atom_error, MAKE_ATOM(env, "closed"));
206	- }
207	- int fd = res->fd;
208	- enif_mutex_unlock(res->lock);
209	-
210 219	ErlNifBinary bin;
211 220	if (!enif_alloc_binary(max_bytes, &bin)) {
212 221	return enif_make_tuple2(env, atom_error, MAKE_ATOM(env, "alloc_failed"));
213 222	}
214 223
224	+ /* Hold the lock across read() + enif_select so that a concurrent
225	+ * nif_close / down callback cannot close the fd mid-syscall. read() is
226	+ * non-blocking (O_NONBLOCK) so the lock is held only briefly. */
227	+ enif_mutex_lock(res->lock);
228	+ if (res->closed \|\| res->fd < 0) {
229	+ enif_mutex_unlock(res->lock);
230	+ enif_release_binary(&bin);
231	+ return enif_make_tuple2(env, atom_error, MAKE_ATOM(env, "closed"));
232	+ }
233	+ int fd = res->fd;
234	+
215 235	ssize_t n = read(fd, bin.data, bin.size);
236	+ int saved_errno = errno;
237	+
216 238	if (n > 0) {
239	+ enif_mutex_unlock(res->lock);
217 240	enif_realloc_binary(&bin, (size_t)n);
218 241	return enif_make_tuple2(env, atom_ok, enif_make_binary(env, &bin));
219	- } else if (n == 0) {
242	+ }
243	+ if (n == 0) {
244	+ enif_mutex_unlock(res->lock);
220 245	enif_release_binary(&bin);
221 246	return atom_eof;
222	- } else {
223	- enif_release_binary(&bin);
224	- if (errno == EAGAIN \|\| errno == EWOULDBLOCK) {
225	- /* Register for select notification */
226	- int sel_ret = enif_select(env, (ErlNifEvent)fd,
227	- ERL_NIF_SELECT_READ, res, NULL,
228	- atom_undefined);
229	- if (sel_ret < 0) {
230	- return enif_make_tuple2(env, atom_error,
231	- MAKE_ATOM(env, "select_failed"));
232	- }
233	- return enif_make_tuple2(env, atom_error, atom_eagain);
234	- }
235	- return enif_make_tuple2(env, atom_error,
236	- MAKE_ATOM(env, errno_to_atom(errno)));
237 247	}
248	+ if (saved_errno == EAGAIN \|\| saved_errno == EWOULDBLOCK) {
249	+ int sel_ret = enif_select(env, (ErlNifEvent)fd,
250	+ ERL_NIF_SELECT_READ, res, NULL,
251	+ atom_undefined);
252	+ enif_mutex_unlock(res->lock);
253	+ enif_release_binary(&bin);
254	+ if (sel_ret < 0) {
255	+ return enif_make_tuple2(env, atom_error,
256	+ MAKE_ATOM(env, "select_failed"));
257	+ }
258	+ return enif_make_tuple2(env, atom_error, atom_eagain);
259	+ }
260	+ enif_mutex_unlock(res->lock);
261	+ enif_release_binary(&bin);
262	+ return enif_make_tuple2(env, atom_error,
263	+ MAKE_ATOM(env, errno_to_atom(saved_errno)));
238 264	}
239 265
240 266	/*
	@@ -261,34 +287,39 @@ static ERL_NIF_TERM nif_write(ErlNifEnv *env, int argc,
261 287	return enif_make_tuple2(env, atom_ok, enif_make_int(env, 0));
262 288	}
263 289
290	+ /* Hold the lock across write() + enif_select so that a concurrent
291	+ * close cannot reap the fd mid-syscall. */
264 292	enif_mutex_lock(res->lock);
265	- if (res->closed) {
293	+ if (res->closed \|\| res->fd < 0) {
266 294	enif_mutex_unlock(res->lock);
267 295	return enif_make_tuple2(env, atom_error, MAKE_ATOM(env, "closed"));
268 296	}
269 297	int fd = res->fd;
270	- enif_mutex_unlock(res->lock);
271 298
272 299	ssize_t n = write(fd, bin.data, bin.size);
300	+ int saved_errno = errno;
301	+
273 302	if (n >= 0) {
303	+ enif_mutex_unlock(res->lock);
274 304	return enif_make_tuple2(env, atom_ok, enif_make_int64(env, (int64_t)n));
275	- } else {
276	- if (errno == EAGAIN \|\| errno == EWOULDBLOCK) {
277	- int sel_ret = enif_select(env, (ErlNifEvent)fd,
278	- ERL_NIF_SELECT_WRITE, res, NULL,
279	- atom_undefined);
280	- if (sel_ret < 0) {
281	- return enif_make_tuple2(env, atom_error,
282	- MAKE_ATOM(env, "select_failed"));
283	- }
284	- return enif_make_tuple2(env, atom_error, atom_eagain);
285	- }
286	- if (errno == EPIPE) {
287	- return enif_make_tuple2(env, atom_error, MAKE_ATOM(env, "epipe"));
288	- }
289	- return enif_make_tuple2(env, atom_error,
290	- MAKE_ATOM(env, errno_to_atom(errno)));
291 305	}
306	+ if (saved_errno == EAGAIN \|\| saved_errno == EWOULDBLOCK) {
307	+ int sel_ret = enif_select(env, (ErlNifEvent)fd,
308	+ ERL_NIF_SELECT_WRITE, res, NULL,
309	+ atom_undefined);
310	+ enif_mutex_unlock(res->lock);
311	+ if (sel_ret < 0) {
312	+ return enif_make_tuple2(env, atom_error,
313	+ MAKE_ATOM(env, "select_failed"));
314	+ }
315	+ return enif_make_tuple2(env, atom_error, atom_eagain);
316	+ }
317	+ enif_mutex_unlock(res->lock);
318	+ if (saved_errno == EPIPE) {
319	+ return enif_make_tuple2(env, atom_error, MAKE_ATOM(env, "epipe"));
320	+ }
321	+ return enif_make_tuple2(env, atom_error,
322	+ MAKE_ATOM(env, errno_to_atom(saved_errno)));
292 323	}
293 324
294 325	/*
	@@ -315,27 +346,19 @@ static ERL_NIF_TERM nif_close(ErlNifEnv *env, int argc,
315 346	res->closed = 1;
316 347	res->fd = -1;
317 348
318	- /* Deregister from enif_select before closing */
319	- enif_select(env, (ErlNifEvent)fd, ERL_NIF_SELECT_STOP, res, NULL,
320	- atom_undefined);
321	-
322 349	if (res->monitor_active) {
323 350	enif_demonitor_process(env, res, &res->monitor);
324 351	res->monitor_active = 0;
325 352	}
326 353
327	- /* Close FD inside critical section to prevent TOCTOU race:
328	- * a concurrent nif_read/nif_write on a dirty scheduler could copy the FD
329	- * under lock then use it after we release the lock but before close(). */
330	- int close_ret = close(fd);
331	- int close_errno = errno;
332	-
333 354	enif_mutex_unlock(res->lock);
334 355
335	- if (close_ret != 0 && close_errno != EINTR) {
336	- return enif_make_tuple2(env, atom_error,
337	- MAKE_ATOM(env, errno_to_atom(close_errno)));
338	- }
356	+ /* Hand fd off to the stop callback — BEAM waits for any in-flight select
357	+ * registration to drain before calling stop, which then close()s the fd.
358	+ * Concurrent nif_read/nif_write serialize on res->lock; once they observe
359	+ * closed==1 they early-out without touching the fd. */
360	+ enif_select(env, (ErlNifEvent)fd, ERL_NIF_SELECT_STOP, res, NULL,
361	+ atom_undefined);
339 362
340 363	return atom_ok;
341 364	}

	@@ -45,6 +45,28 @@
45 45	/* Self-pipe for signal handling */
46 46	static int signal_pipe[2] = {-1, -1};
47 47
48	+ /*
49	+ * Async-signal-safe post-fork failure path. Between fork() and exec*(),
50	+ * POSIX allows only async-signal-safe functions, so we cannot call
51	+ * fprintf / strerror / malloc. This helper uses write(2) on a stack
52	+ * buffer only.
53	+ */
54	+ static void child_fail(const char tag, const char detail) {
55	+ if (tag) {
56	+ size_t i = 0;
57	+ while (i < 64 && tag[i] != '\0') i++;
58	+ (void)!write(STDERR_FILENO, tag, i);
59	+ (void)!write(STDERR_FILENO, ": ", 2);
60	+ }
61	+ if (detail) {
62	+ size_t i = 0;
63	+ while (i < 256 && detail[i] != '\0') i++;
64	+ (void)!write(STDERR_FILENO, detail, i);
65	+ }
66	+ (void)!write(STDERR_FILENO, "\n", 1);
67	+ _exit(127);
68	+ }
69	+
48 70	static void sigchld_handler(int sig) {
49 71	(void)sig;
50 72	int saved_errno = errno;
	@@ -75,7 +97,10 @@ static int send_fds(int uds_fd, int *fds, int nfds) {
75 97
76 98	size_t cmsg_space = CMSG_SPACE((size_t)nfds * sizeof(int));
77 99	char *cmsg_buf = calloc(1, cmsg_space);
78	- if (!cmsg_buf) return -1;
100	+ if (!cmsg_buf) {
101	+ ERROR_LOG("send_fds: calloc(%zu) failed", cmsg_space);
102	+ return -1;
103	+ }
79 104
80 105	struct msghdr msg = {0};
81 106	msg.msg_iov = &iov;
	@@ -85,6 +110,7 @@ static int send_fds(int uds_fd, int *fds, int nfds) {
85 110
86 111	struct cmsghdr *cmsg = CMSG_FIRSTHDR(&msg);
87 112	if (!cmsg) {
113	+ ERROR_LOG("send_fds: CMSG_FIRSTHDR returned NULL");
88 114	free(cmsg_buf);
89 115	return -1;
90 116	}
	@@ -93,9 +119,18 @@ static int send_fds(int uds_fd, int *fds, int nfds) {
93 119	cmsg->cmsg_len = CMSG_LEN((size_t)nfds * sizeof(int));
94 120	memcpy(CMSG_DATA(cmsg), fds, (size_t)nfds * sizeof(int));
95 121
96	- ssize_t ret = sendmsg(uds_fd, &msg, 0);
97	- free(cmsg_buf);
98	- return ret > 0 ? 0 : -1;
122	+ /* Retry on EINTR; treat anything other than a full 1-byte send as error. */
123	+ for (;;) {
124	+ ssize_t ret = sendmsg(uds_fd, &msg, 0);
125	+ if (ret == 1) {
126	+ free(cmsg_buf);
127	+ return 0;
128	+ }
129	+ if (ret < 0 && errno == EINTR) continue;
130	+ ERROR_LOG("sendmsg failed: ret=%zd errno=%s", ret, strerror(errno));
131	+ free(cmsg_buf);
132	+ return -1;
133	+ }
99 134	}
100 135
101 136	/*
	@@ -175,12 +210,19 @@ static int cgroup_setup(pid_t child_pid) {
175 210	char full_path[512];
176 211	char procs_path[576];
177 212
178	- /* Create cgroup directory */
179	- snprintf(full_path, sizeof(full_path), "/sys/fs/cgroup/%s", cgroup_path);
213	+ int n = snprintf(full_path, sizeof(full_path), "/sys/fs/cgroup/%s",
214	+ cgroup_path);
215	+ if (n < 0 \|\| (size_t)n >= sizeof(full_path)) {
216	+ ERROR_LOG("cgroup path too long");
217	+ return -1;
218	+ }
180 219	mkdir(full_path, 0755); /* ignore error if exists */
181 220
182	- /* Move child to cgroup */
183	- snprintf(procs_path, sizeof(procs_path), "%s/cgroup.procs", full_path);
221	+ n = snprintf(procs_path, sizeof(procs_path), "%s/cgroup.procs", full_path);
222	+ if (n < 0 \|\| (size_t)n >= sizeof(procs_path)) {
223	+ ERROR_LOG("cgroup procs path too long");
224	+ return -1;
225	+ }
184 226	FILE *f = fopen(procs_path, "w");
185 227	if (!f) {
186 228	ERROR_LOG("failed to open %s: %s", procs_path, strerror(errno));
	@@ -197,21 +239,26 @@ static void cgroup_cleanup(void) {
197 239	char full_path[512];
198 240	char kill_path[576];
199 241
200	- snprintf(full_path, sizeof(full_path), "/sys/fs/cgroup/%s", cgroup_path);
242	+ int n = snprintf(full_path, sizeof(full_path), "/sys/fs/cgroup/%s",
243	+ cgroup_path);
244	+ if (n < 0 \|\| (size_t)n >= sizeof(full_path)) return;
201 245
202 246	/* Kill all processes in the cgroup via cgroup.kill (cgroup v2) */
203	- snprintf(kill_path, sizeof(kill_path), "%s/cgroup.kill", full_path);
247	+ n = snprintf(kill_path, sizeof(kill_path), "%s/cgroup.kill", full_path);
248	+ if (n < 0 \|\| (size_t)n >= sizeof(kill_path)) return;
204 249	FILE *f = fopen(kill_path, "w");
205 250	if (f) {
206 251	fprintf(f, "1\n");
207 252	fclose(f);
208 253	}
209 254
210	- /* Wait briefly for processes to die */
211	- usleep(100000);
212	-
213	- /* Remove cgroup directory */
214	- rmdir(full_path);
255	+ /* Poll for rmdir success rather than a fixed sleep — the kernel needs
256	+ * a moment to reap the killed processes. Bail after ~1s (10 * 100ms). */
257	+ for (int i = 0; i < 10; i++) {
258	+ if (rmdir(full_path) == 0) return;
259	+ if (errno != EBUSY && errno != ENOTEMPTY) return; /* real error */
260	+ usleep(100000);
261	+ }
215 262	}
216 263	#else
217 264	static int cgroup_setup(pid_t child_pid) {
	@@ -250,17 +297,40 @@ static void kill_child(pid_t child_pid) {
250 297
251 298	/* Escalate to SIGKILL the whole process group */
252 299	kill(-child_pid, SIGKILL);
253	- waitpid(child_pid, NULL, 0);
300	+
301	+ /* Bounded WNOHANG reap loop — avoid hanging forever if the child is
302	+ * stuck in uninterruptible kernel sleep (D-state). After the bound
303	+ * elapses we return anyway; cgroup cleanup + the kernel eventually
304	+ * reap. */
305	+ int sigkill_iters = 30; /* ~3s total at 100ms per iteration */
306	+ for (int i = 0; i < sigkill_iters; i++) {
307	+ pid_t ret = waitpid(child_pid, NULL, WNOHANG);
308	+ if (ret > 0 \|\| (ret < 0 && errno == ECHILD)) break;
309	+ usleep(100000);
310	+ }
254 311
255 312	/* Cleanup cgroup (kills any remaining processes, removes dir) */
256 313	cgroup_cleanup();
257 314	}
258 315
259 316	/*
260	- * Handle a command received from BEAM over UDS.
317	+ * Length of a single framed command given its opcode. Returns 0 if the
318	+ * opcode is unknown (in which case the parser will skip one byte).
319	+ */
320	+ static size_t command_length(uint8_t opcode) {
321	+ switch (opcode) {
322	+ case CMD_KILL: return 2;
323	+ case CMD_CLOSE_STDIN: return 1;
324	+ case CMD_SET_WINSIZE: return 5;
325	+ default: return 0;
326	+ }
327	+ }
328	+
329	+ /*
330	+ * Handle a single command frame.
261 331	*/
262 332	static void handle_command(int uds_fd, pid_t child_pid, int stdin_w,
263	- uint8_t *buf, ssize_t len) {
333	+ uint8_t *buf, size_t len) {
264 334	(void)uds_fd;
265 335	if (len < 1) return;
266 336
	@@ -300,6 +370,36 @@ static void handle_command(int uds_fd, pid_t child_pid, int stdin_w,
300 370	}
301 371	}
302 372
373	+ /*
374	+ * Parse and dispatch all framed commands present in buf. Returns the number
375	+ * of bytes consumed (may be less than len if a tail command is truncated).
376	+ */
377	+ static size_t handle_commands(int uds_fd, pid_t child_pid, int *stdin_w,
378	+ uint8_t *buf, size_t len) {
379	+ size_t off = 0;
380	+ while (off < len) {
381	+ size_t clen = command_length(buf[off]);
382	+ if (clen == 0) {
383	+ /* Unknown opcode: skip one byte to make progress rather than
384	+ * stalling the parser on bad input. */
385	+ DEBUG_LOG("unknown command opcode 0x%02x, skipping", buf[off]);
386	+ off += 1;
387	+ continue;
388	+ }
389	+ if (off + clen > len) {
390	+ /* Partial tail — caller must carry this over to the next read. */
391	+ break;
392	+ }
393	+ uint8_t op = buf[off];
394	+ handle_command(uds_fd, child_pid, *stdin_w, &buf[off], clen);
395	+ if (op == CMD_CLOSE_STDIN) {
396	+ *stdin_w = -1;
397	+ }
398	+ off += clen;
399	+ }
400	+ return off;
401	+ }
402	+
303 403	/*
304 404	* Main event loop using poll().
305 405	*
	@@ -312,6 +412,11 @@ static int event_loop(int uds_fd, pid_t child_pid, int stdin_w) {
312 412	int child_status = -1;
313 413	int child_exited = 0;
314 414
415	+ /* Carry-over buffer for partially-framed commands across reads. Max
416	+ * incoming frame is CMD_SET_WINSIZE (5 bytes); keep a little slack. */
417	+ uint8_t cbuf[64];
418	+ size_t cbuf_used = 0;
419	+
315 420	fds[0].fd = uds_fd;
316 421	fds[0].events = POLLIN;
317 422	fds[1].fd = signal_pipe[0];
	@@ -336,19 +441,25 @@ static int event_loop(int uds_fd, pid_t child_pid, int stdin_w) {
336 441	}
337 442
338 443	if (fds[0].revents & POLLIN) {
339	- uint8_t buf[16];
340	- ssize_t n = read(uds_fd, buf, sizeof(buf));
444	+ ssize_t n = read(uds_fd, cbuf + cbuf_used, sizeof(cbuf) - cbuf_used);
341 445	if (n > 0) {
342	- handle_command(uds_fd, child_pid, stdin_w, buf, n);
343	- /* If CMD_CLOSE_STDIN was handled, mark stdin as closed */
344	- if (buf[0] == CMD_CLOSE_STDIN) {
345	- stdin_w = -1;
446	+ cbuf_used += (size_t)n;
447	+ size_t consumed = handle_commands(uds_fd, child_pid, &stdin_w,
448	+ cbuf, cbuf_used);
449	+ if (consumed > 0 && consumed < cbuf_used) {
450	+ memmove(cbuf, cbuf + consumed, cbuf_used - consumed);
346 451	}
452	+ cbuf_used -= consumed;
347 453	} else if (n == 0) {
348 454	/* BEAM closed the socket */
349 455	DEBUG_LOG("BEAM closed UDS, killing child %d", child_pid);
350 456	kill_child(child_pid);
351 457	return -1;
458	+ } else if (errno != EAGAIN && errno != EWOULDBLOCK &&
459	+ errno != EINTR) {
460	+ ERROR_LOG("read(uds) failed: %s", strerror(errno));
461	+ kill_child(child_pid);
462	+ return -1;
352 463	}
353 464	}
354 465
	@@ -358,10 +469,16 @@ static int event_loop(int uds_fd, pid_t child_pid, int stdin_w) {
358 469	char drain[64];
359 470	while (read(signal_pipe[0], drain, sizeof(drain)) > 0) {}
360 471
361	- /* Reap child */
472	+ /* Reap any and all children that have exited. SIGCHLD is
473	+ * coalesced by the kernel — multiple pending exits can deliver
474	+ * as a single SIGCHLD. Loop until no more reapable children
475	+ * remain. We only flip child_exited when the managed child is
476	+ * reaped; other reapees (should be none today, future-proof)
477	+ * are still cleaned up. */
362 478	int status;
363	- pid_t ret_pid = waitpid(child_pid, &status, WNOHANG);
364	- if (ret_pid > 0) {
479	+ pid_t ret_pid;
480	+ while ((ret_pid = waitpid(-1, &status, WNOHANG)) > 0) {
481	+ if (ret_pid != child_pid) continue;
365 482	child_exited = 1;
366 483	if (WIFEXITED(status)) {
367 484	child_status = WEXITSTATUS(status);
	@@ -468,6 +585,12 @@ int main(int argc, char *argv[]) {
468 585	struct sockaddr_un addr;
469 586	memset(&addr, 0, sizeof(addr));
470 587	addr.sun_family = AF_UNIX;
588	+ if (strlen(uds_path) >= sizeof(addr.sun_path)) {
589	+ fprintf(stderr, "error: uds_path too long (max %zu bytes)\n",
590	+ sizeof(addr.sun_path) - 1);
591	+ close(uds_fd);
592	+ return 1;
593	+ }
471 594	strncpy(addr.sun_path, uds_path, sizeof(addr.sun_path) - 1);
472 595
473 596	if (connect(uds_fd, (struct sockaddr *)&addr, sizeof(addr)) != 0) {
	@@ -506,27 +629,37 @@ int main(int argc, char *argv[]) {
506 629	close(signal_pipe[1]);
507 630	close(master_fd);
508 631
509	- /* Create new session and set controlling terminal */
510	- setsid();
511	- ioctl(slave_fd, TIOCSCTTY, 0);
632	+ /* Create new session (required before acquiring controlling tty) */
633	+ if (setsid() == (pid_t)-1) child_fail("setsid", NULL);
634	+ if (ioctl(slave_fd, TIOCSCTTY, 0) != 0) child_fail("TIOCSCTTY", NULL);
512 635
513	- dup2(slave_fd, STDIN_FILENO);
514	- dup2(slave_fd, STDOUT_FILENO);
515	- dup2(slave_fd, STDERR_FILENO);
636	+ if (dup2(slave_fd, STDIN_FILENO) != STDIN_FILENO \|\|
637	+ dup2(slave_fd, STDOUT_FILENO) != STDOUT_FILENO \|\|
638	+ dup2(slave_fd, STDERR_FILENO) != STDERR_FILENO) {
639	+ child_fail("dup2", NULL);
640	+ }
516 641	if (slave_fd > STDERR_FILENO) close(slave_fd);
517 642
518 643	setpgid(0, 0);
519 644	execvp(cmd, cmd_args);
520	- fprintf(stderr, "execvp failed: %s: %s\n", cmd, strerror(errno));
521	- _exit(127);
645	+ child_fail("execvp", cmd);
522 646	}
523 647
524 648	/* === Shepherd (PTY) === */
525 649	close(slave_fd);
526 650	pty_master_fd = master_fd;
651	+ set_cloexec(master_fd);
527 652
528	- /* Move child to cgroup (Linux only, no-op elsewhere) */
529	- cgroup_setup(child_pid);
653	+ /* Move child to cgroup (Linux only, no-op elsewhere). If the user
654	+ * requested a cgroup path and setup failed, isolation is not
655	+ * available — treat as fatal. */
656	+ if (cgroup_setup(child_pid) != 0) {
657	+ send_error(uds_fd, "cgroup setup failed");
658	+ kill_child(child_pid);
659	+ close(master_fd);
660	+ close(uds_fd);
661	+ return 1;
662	+ }
530 663
531 664	/* Send single master FD to BEAM (used for both read and write) */
532 665	int fds_to_send[1] = {master_fd};
	@@ -611,9 +744,11 @@ int main(int argc, char *argv[]) {
611 744	close(stdout_pipe[0]);
612 745	close(stderr_pipe[0]);
613 746
614	- dup2(stdin_pipe[0], STDIN_FILENO);
615	- dup2(stdout_pipe[1], STDOUT_FILENO);
616	- dup2(stderr_pipe[1], STDERR_FILENO);
747	+ if (dup2(stdin_pipe[0], STDIN_FILENO) != STDIN_FILENO \|\|
748	+ dup2(stdout_pipe[1], STDOUT_FILENO) != STDOUT_FILENO \|\|
749	+ dup2(stderr_pipe[1], STDERR_FILENO) != STDERR_FILENO) {
750	+ child_fail("dup2", NULL);
751	+ }
617 752
618 753	close(stdin_pipe[0]);
619 754	close(stdout_pipe[1]);
	@@ -621,8 +756,7 @@ int main(int argc, char *argv[]) {
621 756
622 757	setpgid(0, 0);
623 758	execvp(cmd, cmd_args);
624	- fprintf(stderr, "execvp failed: %s: %s\n", cmd, strerror(errno));
625	- _exit(127);
759	+ child_fail("execvp", cmd);
626 760	}
627 761
628 762	/* === Shepherd (pipe) === */
	@@ -630,8 +764,18 @@ int main(int argc, char *argv[]) {
630 764	close(stdout_pipe[1]);
631 765	close(stderr_pipe[1]);
632 766
633	- /* Move child to cgroup (Linux only, no-op elsewhere) */
634	- cgroup_setup(child_pid);
767	+ /* Move child to cgroup (Linux only, no-op elsewhere). If the user
768	+ * requested a cgroup path and setup failed, isolation is not
769	+ * available — treat as fatal. */
770	+ if (cgroup_setup(child_pid) != 0) {
771	+ send_error(uds_fd, "cgroup setup failed");
772	+ kill_child(child_pid);
773	+ close(stdin_pipe[1]);
774	+ close(stdout_pipe[0]);
775	+ close(stderr_pipe[0]);
776	+ close(uds_fd);
777	+ return 1;
778	+ }
635 779
636 780	int fds_to_send[3] = {stdin_pipe[1], stdout_pipe[0], stderr_pipe[0]};
637 781	if (send_fds(uds_fd, fds_to_send, 3) != 0) {

Loading more files…

Operator	Description	Example
name:	Match package name. Supports * wildcards and repo/package form	name:phx* or name:hexpm/phoenix
description:	Full-text search of package descriptions	description:auth
depends:	Packages depending on a given package. Supports repo:package form	depends:ecto or depends:hexpm:ecto
build_tool:	Filter by build tool	build_tool:mix
updated_after:	Packages updated after an ISO8601 datetime	updated_after:2025-01-01T00:00:00Z
extra:	Match custom metadata (key,value). Nested keys are separated by commas	extra:license,MIT