clawft_kernel/workload_runtime/
container_relay.rs

1//! In-container ingest relay (ADR-100 section 4).
2//!
3//! A cog always posts its vectors to `127.0.0.1:80/api/v1/store/ingest`
4//! (the address is compiled into the released binaries). Inside a
5//! container, or an Apple `container` VM, that loopback is the container's
6//! own, so nothing answers unless something listens there. When the host
7//! contract names an ingest upstream (the node's ingest bridge as the
8//! container sees it, e.g. the VM gateway), the adapter builds the image
9//! with a small relay as the entrypoint:
10//!
11//! 1. the relay binds `127.0.0.1:80` while still root (the only reason the
12//!    container starts as root, with every capability dropped except
13//!    `NET_BIND_SERVICE`, `SETUID` and `SETGID`);
14//! 2. it forks the cog, which drops to `nobody` and sets `no_new_privs`
15//!    before `exec`, so the cog runs with no capabilities and cannot regain
16//!    any through a setuid binary (Apple `container` has no
17//!    `--security-opt no-new-privileges`, so the relay sets it itself);
18//! 3. it drops to `nobody` (and `no_new_privs`) itself and forwards each connection, byte for
19//!    byte, to the upstream (bounded connections, bytes and time);
20//! 4. it forwards SIGTERM / SIGINT to the cog and exits with the cog's
21//!    status.
22//!
23//! The relay is a Python script, so a relay image needs `python3` in the
24//! operator-pinned base; the build checks for it and fails loudly.
25
26use std::net::SocketAddr;
27
28use super::container_cmd::{COG_PATH, CONTAINER_USER};
29use super::host_contract::HostContract;
30use super::types::RuntimeError;
31
32/// Where the relay script lives in the image.
33pub const RELAY_PATH: &str = "/weftos-ingest-relay.py";
34/// Build-context file name of the relay script.
35pub const RELAY_FILE: &str = "ingest-relay.py";
36/// The fixed address every cog posts its ingest to.
37pub const COG_INGEST_ADDR: &str = "127.0.0.1:80";
38/// Capabilities a relay container keeps (all others are dropped).
39pub const RELAY_CAPS: [&str; 3] = ["CAP_NET_BIND_SERVICE", "CAP_SETUID", "CAP_SETGID"];
40
41/// The upstream the relay forwards to, if the image needs a relay.
42///
43/// Refuses combinations that cannot work: `network=none` (no route out),
44/// `network=host` (the cog already reaches the node's own `127.0.0.1:80`;
45/// a relay would collide with it), and a loopback upstream (inside the
46/// container, loopback is the container itself).
47pub fn relay_upstream(
48    host: &HostContract,
49    network: Option<&str>,
50) -> Result<Option<SocketAddr>, RuntimeError> {
51    let Some(up) = host.ingest_upstream else {
52        return Ok(None);
53    };
54    let bad = |m: &str| Err(RuntimeError::InvalidConfig(m.into()));
55    match network {
56        Some("none") => bad("an ingest upstream needs a network; network=none has none"),
57        Some("host") => bad(
58            "with network=host the cog reaches the node's 127.0.0.1:80 directly; \
59             leave ingest_upstream unset",
60        ),
61        _ if up.ip().is_loopback() || up.ip().is_unspecified() || up.port() == 0 => bad(
62            "the ingest upstream must be an address the container can reach \
63             (not loopback or unspecified, non-zero port)",
64        ),
65        _ => Ok(Some(up)),
66    }
67}
68
69/// Dockerfile for a cog image fronted by the ingest relay.
70pub fn dockerfile(base_image: &str, upstream: SocketAddr) -> String {
71    let (uid, gid) = CONTAINER_USER.split_once(':').unwrap_or(("65534", "65534"));
72    format!(
73        "FROM {base_image}\n\
74         COPY cog {COG_PATH}\n\
75         COPY {RELAY_FILE} {RELAY_PATH}\n\
76         RUN [\"python3\", \"-B\", \"-c\", \"import ctypes, os, socket, threading\"]\n\
77         USER 0:0\n\
78         WORKDIR /\n\
79         ENTRYPOINT [\"python3\", \"-B\", \"-u\", \"{RELAY_PATH}\", \"--listen\", \"{COG_INGEST_ADDR}\", \
80         \"--upstream\", \"{upstream}\", \"--uid\", \"{uid}\", \"--gid\", \"{gid}\", \"--\", \"{COG_PATH}\"]\n"
81    )
82}
83
84/// The relay script (see the module docs).
85pub const RELAY_SCRIPT: &str = r#"import os, signal, socket, sys, threading
86
87MAX_CONNS = 8
88MAX_BYTES = 2 * 1024 * 1024
89IDLE_SECS = 30
90
91
92def log(msg):
93    sys.stderr.write("[ingest-relay] %s\n" % msg)
94
95
96def addr(s):
97    host, _, port = s.rpartition(":")
98    return host.strip("[]"), int(port)
99
100
101def parse(argv):
102    opts, i = {}, 0
103    while i < len(argv) and argv[i] != "--":
104        opts[argv[i]] = argv[i + 1]
105        i += 2
106    cmd = argv[i + 1:]
107    if not cmd:
108        raise SystemExit("usage: --listen H:P --upstream H:P [--uid N --gid N] -- cmd ...")
109    return (addr(opts["--listen"]), addr(opts["--upstream"]),
110            int(opts.get("--uid", "65534")), int(opts.get("--gid", "65534")), cmd)
111
112
113def drop(uid, gid):
114    if os.geteuid() == 0:
115        os.setgroups([])
116        os.setgid(gid)
117        os.setuid(uid)
118    no_new_privs()
119
120
121def no_new_privs():
122    """prctl(PR_SET_NO_NEW_PRIVS): setuid/setgid bits and file capabilities
123    are ignored from here on (Linux only; a no-op elsewhere)."""
124    if not sys.platform.startswith("linux"):
125        return
126    import ctypes
127    libc = ctypes.CDLL(None, use_errno=True)
128    if libc.prctl(38, 1, 0, 0, 0) != 0:
129        raise OSError(ctypes.get_errno(), "prctl(PR_SET_NO_NEW_PRIVS)")
130
131
132def pump(src, dst):
133    total = 0
134    try:
135        while True:
136            data = src.recv(65536)
137            if not data:
138                break
139            total += len(data)
140            if total > MAX_BYTES:
141                log("connection over %d bytes, closed" % MAX_BYTES)
142                break
143            dst.sendall(data)
144    except OSError:
145        pass
146    for s in (src, dst):
147        try:
148            s.shutdown(socket.SHUT_RDWR)
149        except OSError:
150            pass
151
152
153def serve(conn, upstream, slots):
154    try:
155        try:
156            up = socket.create_connection(upstream, timeout=10)
157        except OSError as e:
158            log("upstream %s:%d: %s" % (upstream[0], upstream[1], e))
159            return
160        conn.settimeout(IDLE_SECS)
161        up.settimeout(IDLE_SECS)
162        back = threading.Thread(target=pump, args=(up, conn), daemon=True)
163        back.start()
164        pump(conn, up)
165        back.join(IDLE_SECS)
166        up.close()
167    finally:
168        conn.close()
169        slots.release()
170
171
172def accept_loop(srv, upstream):
173    slots = threading.BoundedSemaphore(MAX_CONNS)
174    while True:
175        conn, _ = srv.accept()
176        if not slots.acquire(blocking=False):
177            conn.close()
178            continue
179        threading.Thread(target=serve, args=(conn, upstream, slots), daemon=True).start()
180
181
182def main():
183    listen, upstream, uid, gid, cmd = parse(sys.argv[1:])
184    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
185    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
186    srv.bind(listen)
187    srv.listen(MAX_CONNS)
188    pid = os.fork()
189    if pid == 0:
190        try:
191            srv.close()
192            drop(uid, gid)
193            os.execv(cmd[0], cmd)
194        except BaseException as e:
195            log("cog exec failed: %s" % e)
196            os._exit(126)
197    drop(uid, gid)
198    for sig in (signal.SIGTERM, signal.SIGINT):
199        signal.signal(sig, lambda s, _f: os.kill(pid, s))
200    threading.Thread(target=accept_loop, args=(srv, upstream), daemon=True).start()
201    while True:
202        try:
203            _, status = os.waitpid(pid, 0)
204            break
205        except InterruptedError:
206            continue
207    code = os.waitstatus_to_exitcode(status)
208    sys.exit(code if code >= 0 else 128 - code)
209
210
211main()
212"#;
clawft_kernel/workload_runtime/container_relay.rs

clawft_kernel/workload_runtime/
container_relay.rs
