From fe889a7f167b0473d1c4cc4866d2e0d1b37ac595 Mon Sep 17 00:00:00 2001 From: Josh Knapp Date: Fri, 13 Nov 2026 12:33:29 -0800 Subject: [PATCH] triple-c-agent: reset exec signal state, chown stdio, bound connections - Execs reset every catchable signal to SIG_DFL and clear the signal mask in pre_exec (raw rt_sigaction/rt_sigprocmask, so libc-reserved RT signals are covered too). The supervisor's `trap '' HUP` no longer leaks into execs, so a dropped tty attach hangs up its session again. - runc's fixStdioPermissions: fds 0-2 are fchown'd to the exec user (except /dev/null and EINVAL/EPERM/EROFS), so /dev/stdout, /dev/stderr and a tty slave are usable by a non-root exec user. - Own accept loop on hyper's http1 builder with a TokioTimer: a 10s header-read timeout and a 256-connection semaphore (wait for a slot, not drop). WebSocket upgrades still served; Config gains header_read_timeout and max_connections. - Attach/tunnel WebSockets accept messages and frames of at most 1 MiB. - The disconnect SIGHUP is sent by the pump, and only while the exec is unreaped, so a reused pgid is never signalled. - Docs: removing TRIPLEC_AGENT_TOKEN does not scrub /proc//environ; recommend TRIPLEC_AGENT_TOKEN_FILE. - Tests: inherited_signals.rs (agent with SIGHUP ignored), slowloris and connection-cap tests, root-only stdio ownership test; fix the stale $USER assertion in the existing root test. Co-Authored-By: Claude Opus 5.5 --- container/triple-c-agent/Cargo.lock | 2 + container/triple-c-agent/Cargo.toml | 4 + container/triple-c-agent/README.md | 21 ++- container/triple-c-agent/src/attach.rs | 32 +++- container/triple-c-agent/src/lib.rs | 88 ++++++++- container/triple-c-agent/src/main.rs | 6 + container/triple-c-agent/src/spawn.rs | 104 ++++++++++- container/triple-c-agent/src/tunnel.rs | 6 +- .../triple-c-agent/tests/inherited_signals.rs | 167 ++++++++++++++++++ container/triple-c-agent/tests/protocol.rs | 116 +++++++++++- 10 files changed, 526 insertions(+), 20 deletions(-) create mode 100644 container/triple-c-agent/tests/inherited_signals.rs diff --git a/container/triple-c-agent/Cargo.lock b/container/triple-c-agent/Cargo.lock index 38195c2..5c15bcd 100644 --- a/container/triple-c-agent/Cargo.lock +++ b/container/triple-c-agent/Cargo.lock @@ -1002,6 +1002,8 @@ dependencies = [ "base64 0.22.1", "bytes", "futures-util", + "hyper", + "hyper-util", "libc", "reqwest", "serde", diff --git a/container/triple-c-agent/Cargo.toml b/container/triple-c-agent/Cargo.toml index bda4905..73a9ac7 100644 --- a/container/triple-c-agent/Cargo.toml +++ b/container/triple-c-agent/Cargo.toml @@ -25,6 +25,10 @@ axum = { version = "0.8", default-features = false, features = ["http1", "tokio" base64 = "0.22" bytes = "1" futures-util = { version = "0.3", default-features = false, features = ["std", "sink"] } +# Already in the tree through axum, with these features; named directly for +# the accept loop in `lib.rs::serve` (header-read timeout needs a hyper timer). +hyper = { version = "1", default-features = false, features = ["http1", "server"] } +hyper-util = { version = "0.1", default-features = false, features = ["tokio", "server", "service"] } libc = "0.2" serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/container/triple-c-agent/README.md b/container/triple-c-agent/README.md index 3f72b76..2c58351 100644 --- a/container/triple-c-agent/README.md +++ b/container/triple-c-agent/README.md @@ -12,10 +12,23 @@ It is a standalone crate: not part of any workspace, with its own `Cargo.lock`. | Variable | Meaning | |---|---| -| `TRIPLEC_AGENT_TOKEN` | Bearer token, at least 32 bytes. Removed from the agent's env before anything is spawned. | -| `TRIPLEC_AGENT_TOKEN_FILE` | Read when the env var is unset or empty. Surrounding whitespace is trimmed. | +| `TRIPLEC_AGENT_TOKEN` | Bearer token, at least 32 bytes. Removed from the agent's env before anything is spawned, so no exec inherits it — but see below. | +| `TRIPLEC_AGENT_TOKEN_FILE` | Read when the env var is unset or empty. Must be owned by the agent's uid and not readable by others. Surrounding whitespace is trimmed. **Preferred.** | | `TRIPLEC_AGENT_ADDR` | Listen address. Default `127.0.0.1:47900`. | +Removing `TRIPLEC_AGENT_TOKEN` only edits the process's libc environment. The kernel keeps the +environment the agent was started with, so `/proc//environ` still shows the token to root +and to any process running as the agent's uid. Use `TRIPLEC_AGENT_TOKEN_FILE`; the entrypoint does. + +Connections are HTTP/1.1. A client gets 10 seconds to send a complete request head, and at most 256 +connections are served at once; past that, new connections wait in the listen backlog for a slot. +WebSocket sessions (attach, tunnel) stop counting once upgraded, and accept messages of at most +1 MiB. + +Execs start with every signal at its default disposition and an empty signal mask, whatever the +agent itself inherited (the entrypoint's supervisor ignores `SIGHUP`), and their stdin/stdout/stderr +are owned by the exec's user unless they are `/dev/null`, as runc arranges for `docker exec`. + Without a token of at least 32 bytes the agent exits with status 2. There is no unauthenticated mode. On startup it logs a single line, `triple-c-agent listening on `, to stderr. The token is never logged. @@ -30,9 +43,9 @@ as itself and answers `403` for any other user. The tests rely on that. cd container/triple-c-agent cargo test # everything, as the current user cargo clippy --all-targets -# The root-only test (user switch + chown), as root: +# The root-only tests (user switch, upload chown, stdio ownership), as root: sudo -E env "PATH=$PATH" CARGO_HOME="$HOME/.cargo" RUSTUP_HOME="$HOME/.rustup" \ - cargo test --test protocol -- --include-ignored + cargo test -- --include-ignored sudo chown -R "$(id -u):$(id -g)" target # sudo leaves root-owned build files behind ``` diff --git a/container/triple-c-agent/src/attach.rs b/container/triple-c-agent/src/attach.rs index 2a0b8a9..ffc2597 100644 --- a/container/triple-c-agent/src/attach.rs +++ b/container/triple-c-agent/src/attach.rs @@ -9,11 +9,11 @@ use axum::response::Response; use bytes::Bytes; use serde::Deserialize; use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::sync::mpsc; +use tokio::sync::{mpsc, oneshot}; use crate::pty; use crate::spawn::{self, ExecSpec, Io, PrepareError}; -use crate::{AppState, CountGuard}; +use crate::{AppState, CountGuard, MAX_WS_MESSAGE}; /// How long the client has to send its spec after the upgrade. const SPEC_TIMEOUT: Duration = Duration::from_secs(30); @@ -48,7 +48,9 @@ enum Input { } pub async fn upgrade(State(st): State>, ws: WebSocketUpgrade) -> Response { - ws.on_upgrade(move |socket| session(socket, st)) + ws.max_message_size(MAX_WS_MESSAGE) + .max_frame_size(MAX_WS_MESSAGE) + .on_upgrade(move |socket| session(socket, st)) } fn text(v: serde_json::Value) -> Message { @@ -121,7 +123,6 @@ async fn session(mut socket: WebSocket, st: Arc) { Err(PrepareError::Refused(_, m)) => return fail(socket, m).await, } }; - let pid = child.id(); let exec_id = st.execs.insert_running(master.clone()); if socket .send(text(serde_json::json!({ "type": "started", "exec_id": exec_id }))) @@ -137,6 +138,7 @@ async fn session(mut socket: WebSocket, st: Arc) { // it records the exit code either way. let (ev_tx, mut ev_rx) = mpsc::channel::(64); let (in_tx, in_rx) = mpsc::unbounded_channel::(); + let (hangup_tx, hangup_rx) = oneshot::channel::<()>(); let guard = CountGuard::execs(&st); { let st = st.clone(); @@ -145,7 +147,7 @@ async fn session(mut socket: WebSocket, st: Arc) { tokio::spawn(async move { let _active = guard; let code = match master { - Some(m) => pump_tty(child, m, in_rx, req.attach_stdin, &ev_tx).await, + Some(m) => pump_tty(child, m, in_rx, hangup_rx, req.attach_stdin, &ev_tx).await, None => pump_pipes(child, in_rx, &ev_tx).await, }; st.execs.finish(&exec_id, code); @@ -206,9 +208,14 @@ async fn session(mut socket: WebSocket, st: Arc) { // non-tty exec only loses stdin and is NOT killed: Docker does not kill // it either, and `gh_login`/`setup-token` clean up after themselves with // their own `pkill`. The pump keeps running and records the exit code. + // + // The pump sends the SIGHUP, not this task: only the pump knows whether it + // has already reaped the process, and once it has, the pid — which is the + // process-group id — may belong to someone else. If the pump is already + // gone the send just fails. drop(in_tx); if tty { - spawn::signal_group(pid, libc::SIGHUP); + let _ = hangup_tx.send(()); } } @@ -276,9 +283,13 @@ async fn pump_tty( mut child: tokio::process::Child, master: Arc, mut input: mpsc::UnboundedReceiver, + hangup: oneshot::Receiver<()>, attach_stdin: bool, tx: &mpsc::Sender, ) -> i32 { + // `Some` until the client has gone (or the session ended without saying + // so), so a completed receiver is never polled again. + let mut hangup = Some(hangup); if attach_stdin { let master = master.clone(); tokio::spawn(async move { @@ -315,6 +326,15 @@ async fn pump_tty( s = child.wait() => { code = Some(s.map(spawn::exit_code).unwrap_or(255)); } + // Only reachable while `code` is `None`: `child.wait()` has not + // returned, so the process is unreaped (alive or a zombie) and its + // pid cannot have been reused for another process group. + r = async { hangup.as_mut().expect("guarded by is_some").await }, if hangup.is_some() => { + hangup = None; + if r.is_ok() { + spawn::signal_group(child.id(), libc::SIGHUP); + } + } } } } diff --git a/container/triple-c-agent/src/lib.rs b/container/triple-c-agent/src/lib.rs index b705157..8ac0728 100644 --- a/container/triple-c-agent/src/lib.rs +++ b/container/triple-c-agent/src/lib.rs @@ -22,7 +22,7 @@ use std::ffi::OsString; use std::path::PathBuf; use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::Arc; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use axum::extract::{Request, State}; use axum::http::{header, StatusCode}; @@ -47,13 +47,30 @@ pub const MAX_ONESHOT_OUTPUT: usize = 8 * 1024 * 1024; /// and a short token is close enough to none that it is refused the same way. pub const MIN_TOKEN_BYTES: usize = 32; -/// Name of the env var carrying the token; removed from the process env after -/// it is read so no exec can inherit it. +/// Name of the env var carrying the token. `main` removes it from the +/// process's environment after reading it, so no exec inherits it — but that +/// only edits libc's copy: the kernel's record of the environment the process +/// was started with (`/proc//environ`, readable by root and by the +/// agent's own uid) still holds it. Prefer [`TOKEN_FILE_ENV`], which is what +/// the entrypoint uses. pub const TOKEN_ENV: &str = "TRIPLEC_AGENT_TOKEN"; pub const TOKEN_FILE_ENV: &str = "TRIPLEC_AGENT_TOKEN_FILE"; pub const ADDR_ENV: &str = "TRIPLEC_AGENT_ADDR"; pub const DEFAULT_ADDR: &str = "127.0.0.1:47900"; +/// How long a client gets to send a complete request head. Until its token +/// has been checked a connection is anonymous, so this is what stops a +/// slowloris from holding sockets open forever. +pub const HEADER_READ_TIMEOUT: Duration = Duration::from_secs(10); + +/// Concurrent HTTP connections. Far above what the app opens; a bound on +/// what an unauthenticated caller can make the agent hold. +pub const MAX_CONNECTIONS: usize = 256; + +/// The largest WebSocket message (and frame) the agent accepts on attach and +/// tunnel sockets. The app sends stdin in frames of at most 64 KiB. +pub(crate) const MAX_WS_MESSAGE: usize = 1024 * 1024; + /// Everything the server needs, captured once at startup. #[derive(Clone)] pub struct Config { @@ -70,6 +87,10 @@ pub struct Config { /// `PUT /v1/files` ceiling. [`MAX_DROP_BYTES`] in production; a field only /// so tests can exercise the refusal without a 256 MiB body. pub max_file_bytes: u64, + /// [`HEADER_READ_TIMEOUT`] in production; a field so tests need not wait. + pub header_read_timeout: Duration, + /// [`MAX_CONNECTIONS`] in production. + pub max_connections: usize, } impl Config { @@ -87,6 +108,8 @@ impl Config { .filter(|(k, _)| !k.to_string_lossy().starts_with("TRIPLEC_AGENT_")) .collect(), max_file_bytes: MAX_DROP_BYTES, + header_read_timeout: HEADER_READ_TIMEOUT, + max_connections: MAX_CONNECTIONS, }) } } @@ -231,8 +254,33 @@ fn router(state: Arc) -> Router { .with_state(state) } -/// Serve the protocol on `listener` until the listener fails. +/// Serve the protocol on `listener`. Runs until the task is dropped: accept +/// errors (a full fd table, a connection reset before accept) are transient +/// and retried, as `axum::serve` does. +/// +/// Our own accept loop rather than `axum::serve`, because that gives hyper no +/// timer, so hyper's header-read timeout never fires and an anonymous +/// connection that trickles (or never sends) its request head is held +/// forever. Here every connection gets [`Config::header_read_timeout`] to +/// deliver a complete head, and at most [`Config::max_connections`] are +/// served at once. HTTP/1.1 only — all the protocol needs — with upgrades. +/// +/// At the cap the loop **waits** for a slot before accepting, rather than +/// accepting and dropping: excess clients queue in the kernel's backlog and +/// are served as soon as a slot frees, which the header timeout guarantees +/// for any slot an idle or slow client holds. Dropping would instead fail the +/// app's own requests outright whenever the cap is reached. +/// +/// A connection's slot is released when hyper is done with it. A WebSocket +/// upgrade ends hyper's part (the socket then belongs to the session task), +/// so attach and tunnel sessions do not count against the cap — and they only +/// exist past the token check. pub async fn serve(listener: tokio::net::TcpListener, config: Config) -> std::io::Result<()> { + use hyper_util::rt::{TokioIo, TokioTimer}; + use hyper_util::service::TowerToHyperService; + + let header_timeout = config.header_read_timeout; + let slots = Arc::new(tokio::sync::Semaphore::new(config.max_connections.max(1))); let state = Arc::new(AppState { config, execs: exec::Registry::default(), @@ -242,7 +290,37 @@ pub async fn serve(listener: tokio::net::TcpListener, config: Config) -> std::io tunnels: AtomicI64::new(0), }, }); - axum::serve(listener, router(state)).await + let app = router(state); + loop { + let permit = slots.clone().acquire_owned().await.expect("the semaphore is never closed"); + let stream = match listener.accept().await { + Ok((stream, _)) => stream, + Err(e) => { + // ECONNABORTED and friends are per-connection; EMFILE/ENFILE + // clear as connections close. Either way, back off and retry. + if !matches!( + e.kind(), + std::io::ErrorKind::ConnectionAborted | std::io::ErrorKind::ConnectionReset + ) { + eprintln!("triple-c-agent: accept failed: {e}"); + tokio::time::sleep(Duration::from_millis(100)).await; + } + continue; + } + }; + let service = TowerToHyperService::new(app.clone()); + tokio::spawn(async move { + let _slot = permit; + let conn = hyper::server::conn::http1::Builder::new() + .timer(TokioTimer::new()) + .header_read_timeout(header_timeout) + .serve_connection(TokioIo::new(stream), service) + .with_upgrades(); + // An error here is the client's (reset, malformed request, header + // timeout); there is nobody to report it to. + let _ = conn.await; + }); + } } /// Bearer-token gate on every request. A failure is `401` with an **empty** diff --git a/container/triple-c-agent/src/main.rs b/container/triple-c-agent/src/main.rs index ea3f613..a08df0b 100644 --- a/container/triple-c-agent/src/main.rs +++ b/container/triple-c-agent/src/main.rs @@ -20,6 +20,12 @@ fn main() -> ExitCode { // Before the runtime starts any thread and before anything is spawned: // removing an env var is only sound while the process is single-threaded, // and every exec's base env is captured from this process's env below. + // + // This keeps the token out of every exec's env, and that is all it does: + // it edits libc's copy of the environment, not the block the kernel + // recorded at exec time, so `/proc//environ` still shows the + // token to root and to anything running as the agent's uid. Use + // TRIPLEC_AGENT_TOKEN_FILE (what the entrypoint does) to avoid that. std::env::remove_var(TOKEN_ENV); // Loopback by default. A non-loopback address is only ever bound when diff --git a/container/triple-c-agent/src/spawn.rs b/container/triple-c-agent/src/spawn.rs index 5f9c08d..b835c7d 100644 --- a/container/triple-c-agent/src/spawn.rs +++ b/container/triple-c-agent/src/spawn.rs @@ -250,11 +250,20 @@ impl Prepared { let (uid, gid) = (self.user.uid, self.user.gid); let groups = self.user.groups.clone(); let cwd = self.cwd.clone(); + let dev_null = dev_null_rdev(); // Runs in the forked child: only async-signal-safe syscalls, nothing // that allocates (everything it touches was built above). // - // Order matters. `setsid` first (new session and process group). Then - // the controlling tty, which needs a session leader. Then gid, + // Order matters. Signal state first: dispositions and mask back to + // the defaults. `docker exec` children start from runc's clean state, + // but ours would otherwise inherit whatever started the agent — the + // entrypoint's supervisor runs it under `trap '' HUP`, and an ignored + // disposition survives `exec`, so every tty session would ignore the + // hangup a client disconnect sends and outlive its client. (Handled + // signals already reset on exec; std clears the mask too, but that + // is std's detail, so we do not rely on it.) Then `setsid` (new + // session and process group). Then the controlling tty, which needs a + // session leader. Then stdio ownership, while still root. Then gid, // supplementary groups (what `initgroups` sets — resolved in the // parent, see `UserInfo::groups`), and uid last, since dropping uid // first would forfeit the right to change the other two. `chdir` comes @@ -262,6 +271,7 @@ impl Prepared { // it does under runc, rather than succeeding as root. unsafe { cmd.pre_exec(move || { + reset_signals()?; if libc::setsid() < 0 { return Err(io::Error::last_os_error()); } @@ -269,6 +279,7 @@ impl Prepared { return Err(io::Error::last_os_error()); } if switch { + fix_stdio_ownership(uid, dev_null)?; if libc::setgid(gid) < 0 { return Err(io::Error::last_os_error()); } @@ -304,6 +315,95 @@ impl Prepared { } } +/// Linux's `_NSIG`: signals are 1..=64. libc does not export it. +const NSIG: libc::c_int = 65; + +/// Every catchable signal to `SIG_DFL` and an empty signal mask — the state a +/// `docker exec` child starts in. For the forked child only. +/// +/// Raw `rt_sigaction`/`rt_sigprocmask` syscalls, not the libc wrappers: libc +/// refuses to touch the real-time signals it reserves for itself (glibc's 32 +/// and 33), and those can arrive ignored too — seen in practice, SigIgn bit +/// 32 set under a test harness. The syscalls also take no libc locks, which +/// matters between fork and exec. An all-zero kernel `struct sigaction` is +/// `SIG_DFL` with no flags, no restorer and an empty mask on every Linux +/// architecture, whatever its field order; its size is 4 words. +/// +/// SIGKILL and SIGSTOP cannot be changed and are skipped. Any failure fails +/// the exec rather than running a child in an unknown signal state. +unsafe fn reset_signals() -> io::Result<()> { + const SIGSET_BYTES: libc::size_t = 8; // the kernel's sigset_t: 64 bits + let dfl = [0u64; 4]; + for sig in 1..NSIG { + if sig == libc::SIGKILL || sig == libc::SIGSTOP { + continue; + } + let r = libc::syscall( + libc::SYS_rt_sigaction, + sig as libc::c_long, + dfl.as_ptr(), + std::ptr::null_mut::(), + SIGSET_BYTES, + ); + if r < 0 { + return Err(io::Error::last_os_error()); + } + } + let empty: u64 = 0; + let r = libc::syscall( + libc::SYS_rt_sigprocmask, + libc::SIG_SETMASK as libc::c_long, + &empty as *const u64, + std::ptr::null_mut::(), + SIGSET_BYTES, + ); + if r < 0 { + return Err(io::Error::last_os_error()); + } + Ok(()) +} + +/// The device number of `/dev/null`, read in the parent (`stat` is not +/// something to do between fork and exec when it can be done before). +/// Falls back to Linux's fixed 1:3. +fn dev_null_rdev() -> libc::dev_t { + use std::os::unix::fs::MetadataExt; + std::fs::metadata("/dev/null") + .map(|m| m.rdev() as libc::dev_t) + .unwrap_or_else(|_| libc::makedev(1, 3)) +} + +/// runc's `fixStdioPermissions`: give fds 0–2 to the exec's user, so it can +/// reopen them — `echo x > /dev/stderr`, `tee /dev/stdout`, and on a tty the +/// slave itself (gpg's pinentry, `screen`, `script`). Without it they stay +/// `root`'s: a pipe is `prw------- root` and a pty slave `root tty`. +/// +/// Like runc: only the uid changes (the gid is left to the mount), `/dev/null` +/// is skipped (chowning it would hand the device to the user), so is an fd +/// the user already owns, and `EINVAL`/`EPERM`/`EROFS` are ignored — an owner +/// outside our user namespace or a read-only mount is not worth failing the +/// exec over. For the forked child only: `fstat` and `fchown` are +/// async-signal-safe. +unsafe fn fix_stdio_ownership(uid: libc::uid_t, dev_null: libc::dev_t) -> io::Result<()> { + for fd in 0..=2 { + let mut st: libc::stat = std::mem::zeroed(); + if libc::fstat(fd, &mut st) < 0 { + return Err(io::Error::last_os_error()); + } + let is_dev_null = st.st_mode & libc::S_IFMT == libc::S_IFCHR && st.st_rdev == dev_null; + if st.st_uid == uid || is_dev_null { + continue; + } + if libc::fchown(fd, uid, libc::gid_t::MAX) < 0 { + let e = io::Error::last_os_error(); + if !matches!(e.raw_os_error(), Some(libc::EINVAL | libc::EPERM | libc::EROFS)) { + return Err(e); + } + } + } + Ok(()) +} + /// Docker's exit code for a process: its status, or 128 + signal when a /// signal killed it (what `docker inspect` and a shell both report). pub fn exit_code(status: ExitStatus) -> i32 { diff --git a/container/triple-c-agent/src/tunnel.rs b/container/triple-c-agent/src/tunnel.rs index 3b40ba0..7c99ae6 100644 --- a/container/triple-c-agent/src/tunnel.rs +++ b/container/triple-c-agent/src/tunnel.rs @@ -14,7 +14,7 @@ use serde::Deserialize; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpStream; -use crate::{error, AppState, CountGuard}; +use crate::{error, AppState, CountGuard, MAX_WS_MESSAGE}; const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); @@ -33,7 +33,9 @@ pub async fn upgrade( Ok(_) => return error(StatusCode::BAD_REQUEST, "port must be 1-65535"), Err(r) => return error(StatusCode::BAD_REQUEST, format!("invalid query: {}", r.body_text())), }; - ws.on_upgrade(move |socket| session(socket, st, port)) + ws.max_message_size(MAX_WS_MESSAGE) + .max_frame_size(MAX_WS_MESSAGE) + .on_upgrade(move |socket| session(socket, st, port)) } async fn session(mut socket: WebSocket, st: Arc, port: u16) { diff --git a/container/triple-c-agent/tests/inherited_signals.rs b/container/triple-c-agent/tests/inherited_signals.rs new file mode 100644 index 0000000..e957a09 --- /dev/null +++ b/container/triple-c-agent/tests/inherited_signals.rs @@ -0,0 +1,167 @@ +//! Execs must start with default signal dispositions and an empty mask, no +//! matter what the agent itself inherited — the clean state a `docker exec` +//! child gets from runc. +//! +//! In production the entrypoint's supervisor runs the agent under +//! `trap '' HUP`, and an ignored disposition survives `exec`. These tests +//! reproduce that by ignoring SIGHUP (and SIGUSR1) in *this* test process, +//! which is where the in-process server spawns from. A disposition is +//! process-wide, which is why this is its own test binary rather than part of +//! `protocol.rs`. + +use std::net::SocketAddr; +use std::sync::Once; +use std::time::{Duration, Instant}; + +use futures_util::{SinkExt, StreamExt}; +use serde_json::{json, Value}; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::Message; +use triple_c_agent::{serve, Config}; + +const TOKEN: &str = "test-token-0123456789abcdef0123456789abcdef"; + +fn ignore_signals_like_the_supervisor() { + static ONCE: Once = Once::new(); + ONCE.call_once(|| unsafe { + libc::signal(libc::SIGHUP, libc::SIG_IGN); + libc::signal(libc::SIGUSR1, libc::SIG_IGN); + }); +} + +/// A `/proc//status` mask field (`SigIgn`, `SigBlk`, …) as a bitmask. +fn mask(status: &str, field: &str) -> u64 { + let line = status + .lines() + .find(|l| l.starts_with(&format!("{field}:"))) + .unwrap_or_else(|| panic!("no {field} in status")); + u64::from_str_radix(line.split_whitespace().nth(1).unwrap(), 16).unwrap() +} + +fn bit(sig: libc::c_int) -> u64 { + 1 << (sig - 1) +} + +struct Agent { + addr: SocketAddr, + http: reqwest::Client, + _cwd: tempfile::TempDir, + user: String, +} + +async fn start() -> Agent { + ignore_signals_like_the_supervisor(); + // The precondition, checked rather than assumed. + let own = std::fs::read_to_string("/proc/self/status").unwrap(); + assert_ne!(mask(&own, "SigIgn") & bit(libc::SIGHUP), 0, "the test process must ignore SIGHUP"); + + let cwd = tempfile::tempdir().unwrap(); + let config = Config { + token: TOKEN.to_string(), + default_cwd: cwd.path().to_path_buf(), + base_env: vec![("PATH".into(), "/usr/local/bin:/usr/bin:/bin".into())], + max_file_bytes: 1024, + header_read_timeout: triple_c_agent::HEADER_READ_TIMEOUT, + max_connections: triple_c_agent::MAX_CONNECTIONS, + }; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(serve(listener, config)); + let user = String::from_utf8(std::process::Command::new("id").arg("-un").output().unwrap().stdout) + .unwrap() + .trim() + .to_string(); + Agent { addr, http: reqwest::Client::new(), _cwd: cwd, user } +} + +impl Agent { + async fn get(&self, path: &str) -> Value { + let r = self.http.get(format!("http://{}{path}", self.addr)).bearer_auth(TOKEN).send().await.unwrap(); + assert_eq!(r.status().as_u16(), 200); + r.json().await.unwrap() + } +} + +fn process_gone(pid: i32) -> bool { + match std::fs::read_to_string(format!("/proc/{pid}/stat")) { + Err(_) => true, + Ok(s) => s.rsplit(')').next().unwrap_or("").trim_start().starts_with('Z'), + } +} + +#[tokio::test] +async fn execs_start_with_no_ignored_or_blocked_signals() { + let a = start().await; + let r = a + .http + .post(format!("http://{}/v1/exec", a.addr)) + .bearer_auth(TOKEN) + .json(&json!({ "cmd": ["cat", "/proc/self/status"], "user": a.user })) + .send() + .await + .unwrap(); + assert_eq!(r.status().as_u16(), 200); + let v: Value = r.json().await.unwrap(); + assert_eq!(v["exit_code"], 0, "{v}"); + use base64::Engine; + let status: String = v["chunks"] + .as_array() + .unwrap() + .iter() + .filter(|c| c["s"] == "out") + .map(|c| { + let b = base64::engine::general_purpose::STANDARD.decode(c["d"].as_str().unwrap()).unwrap(); + String::from_utf8(b).unwrap() + }) + .collect(); + assert_eq!(mask(&status, "SigIgn"), 0, "inherited ignored signals:\n{status}"); + assert_eq!(mask(&status, "SigBlk"), 0, "inherited blocked signals:\n{status}"); +} + +#[tokio::test] +async fn a_dropped_tty_attach_still_hangs_up_when_the_agent_ignores_sighup() { + let a = start().await; + let mut req = format!("ws://{}/v1/attach", a.addr).into_client_request().unwrap(); + req.headers_mut().insert("Authorization", format!("Bearer {TOKEN}").parse().unwrap()); + let mut ws = tokio_tungstenite::connect_async(req).await.unwrap().0; + let spec = json!({ + "cmd": ["sh", "-c", "echo pid=$$; exec sleep 600"], + "user": a.user, "tty": true, "attach_stdin": true, + }); + ws.send(Message::Text(spec.to_string().into())).await.unwrap(); + + let mut out = String::new(); + let pid: i32 = loop { + match tokio::time::timeout(Duration::from_secs(10), ws.next()).await.expect("timed out") { + Some(Ok(Message::Text(_))) => {} // "started" + Some(Ok(Message::Binary(b))) => { + out.push_str(&String::from_utf8_lossy(&b[1..])); + if let Some(rest) = out.split("pid=").nth(1) { + if let Some(line) = rest.split_once(['\r', '\n']) { + break line.0.trim().parse().unwrap(); + } + } + } + other => panic!("unexpected {other:?}"), + } + }; + assert_eq!(a.get("/v1/activity").await["active_execs"], 1); + // Let `exec sleep` happen, so the process we watch is the sleep. + tokio::time::sleep(Duration::from_millis(200)).await; + drop(ws); + + let deadline = Instant::now() + Duration::from_secs(5); + while !process_gone(pid) { + assert!(Instant::now() < deadline, "sleep (pid {pid}) survived the client disconnect"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + let deadline = Instant::now() + Duration::from_secs(5); + loop { + let v = a.get("/v1/activity").await; + if v["active_execs"] == 0 { + break; + } + assert!(Instant::now() < deadline, "active_execs never returned to 0: {v}"); + tokio::time::sleep(Duration::from_millis(50)).await; + } +} diff --git a/container/triple-c-agent/tests/protocol.rs b/container/triple-c-agent/tests/protocol.rs index c2b3151..c640f5d 100644 --- a/container/triple-c-agent/tests/protocol.rs +++ b/container/triple-c-agent/tests/protocol.rs @@ -70,6 +70,8 @@ async fn start_with(tweak: impl FnOnce(&mut Config)) -> Agent { (OsString::from("HOME"), OsString::from("/agent-home-must-not-leak")), ], max_file_bytes: 1024, + header_read_timeout: triple_c_agent::HEADER_READ_TIMEOUT, + max_connections: triple_c_agent::MAX_CONNECTIONS, }; tweak(&mut config); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); @@ -860,8 +862,118 @@ async fn activity_counts_execs_and_tunnels_and_ignores_its_own_probes() { } } +// ── connection limits ────────────────────────────────────────────────────── + +/// Wait for the server to close `s`: `Ok(elapsed)` on EOF or reset, +/// `Err` if it is still open after `within`. +async fn closed_within(s: &mut tokio::net::TcpStream, within: Duration) -> Result { + let t0 = Instant::now(); + let mut buf = [0u8; 1024]; + loop { + match tokio::time::timeout(within.saturating_sub(t0.elapsed()), s.read(&mut buf)).await { + Err(_) => return Err(()), + Ok(Ok(0)) | Ok(Err(_)) => return Ok(t0.elapsed()), + // hyper may answer a timed-out head with a 408 before closing. + Ok(Ok(_)) => continue, + } + } +} + +#[tokio::test] +async fn a_slow_request_head_is_cut_off_by_the_header_timeout() { + let a = start_with(|c| c.header_read_timeout = Duration::from_secs(1)).await; + + // A partial request line, never finished: slowloris. + let mut partial = tokio::net::TcpStream::connect(a.addr).await.unwrap(); + partial.write_all(b"GET /v1/heal").await.unwrap(); + // A connection that sends nothing at all. + let mut silent = tokio::net::TcpStream::connect(a.addr).await.unwrap(); + // A head that trickles in a byte at a time, each well inside the timeout. + let mut trickle = tokio::net::TcpStream::connect(a.addr).await.unwrap(); + let trickler = tokio::spawn(async move { + for b in b"GET /v1/health HTTP/1.1\r\nHost: x\r\nX-Pad: aaaaaaaaaaaaaaaaaaaaaaaa".iter() { + if trickle.write_all(&[*b]).await.is_err() { + break; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + trickle + }); + + let e = closed_within(&mut partial, Duration::from_secs(15)).await.expect("partial head never closed"); + assert!(e < Duration::from_secs(5), "closed only after {e:?}"); + closed_within(&mut silent, Duration::from_secs(15)).await.expect("silent connection never closed"); + let mut trickle = trickler.await.unwrap(); + closed_within(&mut trickle, Duration::from_secs(15)).await.expect("trickling head never closed"); + + // The server is still serving, and a normal request is untouched. + let (s, v) = a.get("/v1/health").await; + assert_eq!(s, 200, "{v}"); +} + +#[tokio::test] +async fn connections_past_the_cap_wait_for_a_slot_rather_than_failing() { + let a = start_with(|c| { + c.max_connections = 2; + c.header_read_timeout = Duration::from_secs(1); + }) + .await; + // Fill both slots with connections that never send anything. + let _hog1 = tokio::net::TcpStream::connect(a.addr).await.unwrap(); + let _hog2 = tokio::net::TcpStream::connect(a.addr).await.unwrap(); + tokio::time::sleep(Duration::from_millis(100)).await; + // A real request queues until the header timeout frees a slot. + let t0 = Instant::now(); + let (s, _) = a.get("/v1/health").await; + assert_eq!(s, 200); + assert!(t0.elapsed() >= Duration::from_millis(500), "was not held back: {:?}", t0.elapsed()); +} + // ── root only ────────────────────────────────────────────────────────────── +#[tokio::test] +#[ignore = "needs root: runs execs as nobody to check stdio is chowned to the exec user"] +async fn as_root_stdio_belongs_to_the_exec_user() { + assert!(is_root(), "run with sudo -E cargo test -- --ignored"); + let a = start().await; + + // One-shot: stdout/stderr are pipes; reopening them by path needs the + // user to own them, as runc arranges. + let (s, v) = a + .post( + "/v1/exec", + json!({ "cmd": sh("echo x > /dev/stdout && echo y > /dev/stderr && echo ok"), "user": "nobody" }), + ) + .await; + assert_eq!(s, 200, "{v}"); + let (_, out, err, _) = decode(&v); + assert_eq!(v["exit_code"], 0, "stdout={out:?} stderr={err:?}"); + assert_eq!(out, "x\nok\n"); + assert_eq!(err, "y\n"); + + // Non-tty attach: the same, over pipes the attach owns. + let (mut ws, _) = a + .attach(json!({ "cmd": sh("echo x > /dev/stdout && echo y > /dev/stderr"), "user": "nobody" })) + .await; + let (out, err, code) = collect_until_exit(&mut ws).await; + assert_eq!((code, out.as_slice(), err.as_slice()), (0, &b"x\n"[..], &b"y\n"[..])); + + // Tty attach: the slave is the user's, not `root tty`. + let (mut ws, _) = a + .attach(json!({ "cmd": sh("stat -c %U $(tty)"), "user": "nobody", "tty": true })) + .await; + let (out, _, code) = collect_until_exit(&mut ws).await; + assert_eq!(code, 0); + assert_eq!(String::from_utf8_lossy(&out).trim(), "nobody"); + + // /dev/null is never handed over. + let (s, v) = a.post("/v1/exec/detached", json!({ "cmd": ["true"], "user": "nobody" })).await; + assert_eq!(s, 200, "{v}"); + a.wait_finished(v["exec_id"].as_str().unwrap()).await; + use std::os::unix::fs::MetadataExt; + assert_eq!(std::fs::metadata("/dev/null").unwrap().uid(), 0); +} + #[tokio::test] #[ignore = "needs root: switches to another user (nobody) and chowns an upload"] async fn as_root_execs_switch_user_and_uploads_chown() { @@ -874,7 +986,9 @@ async fn as_root_execs_switch_user_and_uploads_chown() { let out = decode(&v).1; let lines: Vec<_> = out.lines().collect(); assert_eq!(lines[0], "65534"); - assert_eq!(lines[3], "nobody"); + // runc sets only HOME for the exec's user; USER stays unset unless the + // container env has it (the base env here does not) — see `merge_env`. + assert_eq!(lines[3], ""); assert_eq!(lines[2], passwd_home("nobody")); let dir = a.tmp.path().to_str().unwrap().to_string();