Merge commit 'fe889a7' into feat/triple-c-agent

container/triple-c-agent/Cargo.lock

@@ -1002,6 +1002,8 @@ dependencies = [
  "base64 0.22.1",
  "bytes",
  "futures-util",
+ "hyper",
+ "hyper-util",
  "libc",
  "reqwest",
  "serde",

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"

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/<agent pid>/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 <version> listening on <addr>`, 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
 ```
 

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<Arc<AppState>>, 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<AppState>) {
             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<AppState>) {
     // it records the exit code either way.
     let (ev_tx, mut ev_rx) = mpsc::channel::<Event>(64);
     let (in_tx, in_rx) = mpsc::unbounded_channel::<Input>();
+    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<AppState>) {
         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<AppState>) {
     // 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<pty::PtyMaster>,
     mut input: mpsc::UnboundedReceiver<Input>,
+    hangup: oneshot::Receiver<()>,
     attach_stdin: bool,
     tx: &mpsc::Sender<Event>,
 ) -> 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);
+                }
+            }
         }
     }
 }

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/<pid>/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<AppState>) -> 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**

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/<agent pid>/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

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::<u64>(),
+            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::<u64>(),
+        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 {

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<AppState>, port: u16) {

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/<pid>/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;
+    }
+}

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<Duration, ()> {
+    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();