Add triple-c-agent: the in-container agent for the v1 wire protocol
A static musl binary (agent/, standalone crate) implementing docs/superpowers/specs/2026-11-13-triple-c-agent-protocol.md: one-shot, detached and attached execs (pipes or pty), exec status and resize, file upload, archive download, TCP tunnels and activity counters, all behind a constant-time bearer token. Matches docker exec: same user/initgroups, working directory, env merge (PATH/TERM defaults, HOME/USER/LOGNAME from passwd), 126/127 for commands that cannot start, 128+n for signals, the exact OUTPUT_LIMIT message, and a process group per exec so the limit kill and disconnect SIGHUP reach children. Builds statically for x86_64 and aarch64 musl with no C toolchain (aarch64 via rust-lld); commands in agent/README.md. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
shadowdaoandClaude Opus 5.5 committed
1 parent
086abd5713
commit
5d03bea544
14 files changed
+4197
No files matched your search
@@ -1,6 +1,7 @@
|
||||
node_modules/
|
||||
app/dist/
|
||||
app/src-tauri/target/
|
||||
agent/target/
|
||||
# Written by build.rs (tauri-build AppManifest); gen/schemas/acl-manifests.json is the
|
||||
# tracked, reviewable form of the same information.
|
||||
app/src-tauri/permissions/autogenerated/
|
||||
|
||||
Generated
+1299
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,44 @@
|
||||
[package]
|
||||
name = "triple-c-agent"
|
||||
version = "0.1.0"
|
||||
edition = "2021"
|
||||
description = "In-container agent: docker-exec parity over authenticated HTTP/WebSocket"
|
||||
publish = false
|
||||
|
||||
# Deliberately not part of any workspace: this binary is built for
|
||||
# *-unknown-linux-musl and baked into the container image, on its own cadence.
|
||||
# The empty table stops cargo from walking up and adopting it into one.
|
||||
[workspace]
|
||||
|
||||
[lib]
|
||||
name = "triple_c_agent"
|
||||
path = "src/lib.rs"
|
||||
|
||||
[[bin]]
|
||||
name = "triple-c-agent"
|
||||
path = "src/main.rs"
|
||||
|
||||
# Pure-Rust only — no TLS, no openssl, nothing with a C build step — so the
|
||||
# binary links statically for both musl targets without a C cross toolchain.
|
||||
[dependencies]
|
||||
axum = { version = "0.8", default-features = false, features = ["http1", "tokio", "json", "query", "ws"] }
|
||||
base64 = "0.22"
|
||||
bytes = "1"
|
||||
futures-util = { version = "0.3", default-features = false, features = ["std", "sink"] }
|
||||
libc = "0.2"
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_json = "1"
|
||||
tar = { version = "0.4", default-features = false }
|
||||
tokio = { version = "1", features = ["rt-multi-thread", "macros", "net", "process", "io-util", "sync", "time", "fs"] }
|
||||
|
||||
[dev-dependencies]
|
||||
reqwest = { version = "0.12", default-features = false, features = ["json"] }
|
||||
tempfile = "3"
|
||||
tokio-tungstenite = { version = "0.29", default-features = false, features = ["connect"] }
|
||||
|
||||
[profile.release]
|
||||
opt-level = "s"
|
||||
lto = true
|
||||
codegen-units = 1
|
||||
strip = true
|
||||
panic = "abort"
|
||||
@@ -0,0 +1,67 @@
|
||||
# triple-c-agent
|
||||
|
||||
The in-container agent: a small static binary that does over authenticated HTTP/WebSocket what
|
||||
`host::ContainerTarget` does through the Docker API. It provides one-shot, attached and detached
|
||||
execs, resize, exit status, file upload, archive download and TCP tunnels. The wire protocol is
|
||||
`docs/superpowers/specs/2026-11-13-triple-c-agent-protocol.md`, and that spec is the contract.
|
||||
This crate implements it and does not extend it.
|
||||
|
||||
It is a standalone crate: not part of any workspace, with its own `Cargo.lock`.
|
||||
|
||||
## Running
|
||||
|
||||
| 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_ADDR` | Listen address. Default `127.0.0.1:47900`. |
|
||||
|
||||
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.
|
||||
|
||||
It must run **as root** (see CLAUDE.md, "Never hand the container's lifetime to a process the
|
||||
`claude` user owns"). Execs then switch to the requested user. A non-root agent can only run execs
|
||||
as itself and answers `403` for any other user. The tests rely on that.
|
||||
|
||||
## Tests
|
||||
|
||||
```bash
|
||||
cd agent
|
||||
cargo test # everything, as the current user
|
||||
cargo clippy --all-targets
|
||||
# The root-only test (user switch + chown), as root:
|
||||
sudo -E env "PATH=$PATH" CARGO_HOME="$HOME/.cargo" RUSTUP_HOME="$HOME/.rustup" \
|
||||
cargo test --test protocol -- --include-ignored
|
||||
sudo chown -R "$(id -u):$(id -g)" target # sudo leaves root-owned build files behind
|
||||
```
|
||||
|
||||
The integration tests (`tests/protocol.rs`) start the real server in-process on `127.0.0.1:0` and
|
||||
drive every endpoint over real HTTP and WebSocket.
|
||||
|
||||
## Static release builds
|
||||
|
||||
Both targets build from an x86_64 Linux host with **no C cross toolchain**: every dependency is
|
||||
pure Rust, and Rust's musl targets ship a self-contained CRT. aarch64 links with the `rust-lld`
|
||||
that comes with the toolchain. Verified with rustc 1.98.0:
|
||||
|
||||
```bash
|
||||
rustup target add x86_64-unknown-linux-musl aarch64-unknown-linux-musl
|
||||
|
||||
cd agent
|
||||
cargo build --release --target x86_64-unknown-linux-musl
|
||||
CARGO_TARGET_AARCH64_UNKNOWN_LINUX_MUSL_LINKER=rust-lld \
|
||||
cargo build --release --target aarch64-unknown-linux-musl
|
||||
```
|
||||
|
||||
Output:
|
||||
|
||||
- `target/x86_64-unknown-linux-musl/release/triple-c-agent`: `static-pie linked`, about 1.3 MB.
|
||||
- `target/aarch64-unknown-linux-musl/release/triple-c-agent`: `statically linked`, about 1.2 MB.
|
||||
|
||||
Check a build with `file <binary>`. It must say `statically linked` or `static-pie linked`, and never
|
||||
`dynamically linked` or `interpreter`.
|
||||
|
||||
Do not add a dependency that needs a C compiler (`ring`, `openssl-sys`, anything using `cc`). The
|
||||
aarch64 build would then need a cross C toolchain or `cargo-zigbuild`. TLS is out of scope for v1
|
||||
anyway, because the transport is the SSM channel or loopback.
|
||||
@@ -0,0 +1,320 @@
|
||||
//! `GET /v1/attach`: an exec with its stdio carried over a WebSocket.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
|
||||
use axum::extract::State;
|
||||
use axum::response::Response;
|
||||
use bytes::Bytes;
|
||||
use serde::Deserialize;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use crate::pty;
|
||||
use crate::spawn::{self, ExecSpec, Io, PrepareError};
|
||||
use crate::{AppState, CountGuard};
|
||||
|
||||
/// How long the client has to send its spec after the upgrade.
|
||||
const SPEC_TIMEOUT: Duration = Duration::from_secs(30);
|
||||
|
||||
/// After a tty exec's own process exits, how long to keep draining the pty
|
||||
/// for output still in flight. Waiting for EOF alone is not enough: a
|
||||
/// background job that kept the terminal open would hold the attach (and its
|
||||
/// exit frame) forever, while Docker ends the stream when the exec exits.
|
||||
const TTY_DRAIN_IDLE: Duration = Duration::from_millis(100);
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct AttachRequest {
|
||||
#[serde(flatten)]
|
||||
spec: ExecSpec,
|
||||
#[serde(default)]
|
||||
attach_stdin: bool,
|
||||
#[serde(default)]
|
||||
cols: Option<u16>,
|
||||
#[serde(default)]
|
||||
rows: Option<u16>,
|
||||
}
|
||||
|
||||
/// What the pump sends to the socket.
|
||||
enum Event {
|
||||
Out(u8, Bytes),
|
||||
Exit(i32),
|
||||
}
|
||||
|
||||
/// What the socket sends to the process.
|
||||
enum Input {
|
||||
Data(Bytes),
|
||||
}
|
||||
|
||||
pub async fn upgrade(State(st): State<Arc<AppState>>, ws: WebSocketUpgrade) -> Response {
|
||||
ws.on_upgrade(move |socket| session(socket, st))
|
||||
}
|
||||
|
||||
fn text(v: serde_json::Value) -> Message {
|
||||
Message::Text(v.to_string().into())
|
||||
}
|
||||
|
||||
async fn fail(mut socket: WebSocket, msg: String) {
|
||||
let _ = socket.send(text(serde_json::json!({ "type": "error", "error": msg }))).await;
|
||||
let _ = socket.send(Message::Close(None)).await;
|
||||
}
|
||||
|
||||
fn frame(channel: u8, data: &[u8]) -> Message {
|
||||
let mut v = Vec::with_capacity(data.len() + 1);
|
||||
v.push(channel);
|
||||
v.extend_from_slice(data);
|
||||
Message::Binary(v.into())
|
||||
}
|
||||
|
||||
async fn session(mut socket: WebSocket, st: Arc<AppState>) {
|
||||
// 1. The spec, as one text frame.
|
||||
let first = match tokio::time::timeout(SPEC_TIMEOUT, socket.recv()).await {
|
||||
Ok(Some(Ok(Message::Text(t)))) => t,
|
||||
Ok(Some(Ok(_))) => return fail(socket, "the first frame must be a text ExecSpec".into()).await,
|
||||
_ => return,
|
||||
};
|
||||
let req: AttachRequest = match serde_json::from_str(first.as_str()) {
|
||||
Ok(r) => r,
|
||||
Err(e) => return fail(socket, format!("invalid attach request: {e}")).await,
|
||||
};
|
||||
let tty = req.spec.tty;
|
||||
|
||||
let prepared = match spawn::prepare(&st.config, &req.spec, tty) {
|
||||
Ok(p) => p,
|
||||
Err(PrepareError::Refused(_, m)) => return fail(socket, m).await,
|
||||
Err(PrepareError::CannotStart { code, message }) => {
|
||||
return not_started(socket, &st, code, message).await
|
||||
}
|
||||
};
|
||||
|
||||
// 2. Start it. A tty exec gets the pty slave on all three fds; stdin
|
||||
// stays the pty even with `attach_stdin: false` (the client's bytes are
|
||||
// just ignored), because `stty`, `read -s` and friends need a terminal on
|
||||
// fd 0 — the same as a Docker tty exec without `-i`.
|
||||
let (child, master) = if tty {
|
||||
let (master, slave) = match pty::open(req.cols.unwrap_or(80), req.rows.unwrap_or(24)) {
|
||||
Ok(p) => p,
|
||||
Err(e) => return fail(socket, format!("cannot open a pty: {e}")).await,
|
||||
};
|
||||
let fds = (|| Ok::<_, std::io::Error>((slave.try_clone()?, slave.try_clone()?, slave)))();
|
||||
let (a, b, c) = match fds {
|
||||
Ok(f) => f,
|
||||
Err(e) => return fail(socket, format!("cannot open a pty: {e}")).await,
|
||||
};
|
||||
// The slave copies are moved into the Command and closed when this
|
||||
// returns, so the master sees EOF once the *child's* copies close.
|
||||
match prepared.spawn(Io::Fd(a), Io::Fd(b), Io::Fd(c), true) {
|
||||
Ok(child) => (child, Some(Arc::new(master))),
|
||||
Err(PrepareError::CannotStart { code, message }) => {
|
||||
return not_started(socket, &st, code, message).await
|
||||
}
|
||||
Err(PrepareError::Refused(_, m)) => return fail(socket, m).await,
|
||||
}
|
||||
} else {
|
||||
let stdin = if req.attach_stdin { Io::Piped } else { Io::Null };
|
||||
match prepared.spawn(stdin, Io::Piped, Io::Piped, false) {
|
||||
Ok(child) => (child, None),
|
||||
Err(PrepareError::CannotStart { code, message }) => {
|
||||
return not_started(socket, &st, code, message).await
|
||||
}
|
||||
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 })))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
// Gone before it began; still let the process run to an exit code.
|
||||
}
|
||||
|
||||
// 3. The pump owns the child and its output for the process's whole life,
|
||||
// independent of the socket: if the client leaves, it keeps draining (and
|
||||
// discarding) output so the process is never wedged on a full pipe, and
|
||||
// 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 guard = CountGuard::execs(&st);
|
||||
{
|
||||
let st = st.clone();
|
||||
let exec_id = exec_id.clone();
|
||||
let master = master.clone();
|
||||
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,
|
||||
None => pump_pipes(child, in_rx, &ev_tx).await,
|
||||
};
|
||||
st.execs.finish(&exec_id, code);
|
||||
let _ = ev_tx.send(Event::Exit(code)).await;
|
||||
});
|
||||
}
|
||||
|
||||
// 4. Shuttle between socket and pump. Unbounded stdin on purpose: a child
|
||||
// that is not reading stdin must not stop us forwarding its output, or a
|
||||
// heavy two-way stream deadlocks. The client is authenticated; how much it
|
||||
// queues is its own business, as with Docker's hijacked connection.
|
||||
// The loop only `break`s when the client has gone; the normal end (exit
|
||||
// frame sent) `return`s from inside it.
|
||||
let mut in_tx = Some(in_tx);
|
||||
loop {
|
||||
tokio::select! {
|
||||
ev = ev_rx.recv() => match ev {
|
||||
Some(Event::Out(ch, data)) => {
|
||||
st.activity.touch();
|
||||
if socket.send(frame(ch, &data)).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
Some(Event::Exit(code)) => {
|
||||
let _ = socket.send(text(serde_json::json!({ "type": "exit", "code": code }))).await;
|
||||
let _ = socket.send(Message::Close(None)).await;
|
||||
return;
|
||||
}
|
||||
None => return,
|
||||
},
|
||||
msg = socket.recv() => match msg {
|
||||
Some(Ok(Message::Binary(data))) => {
|
||||
st.activity.touch();
|
||||
if req.attach_stdin {
|
||||
if let Some(tx) = &in_tx {
|
||||
let _ = tx.send(Input::Data(data));
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(Ok(Message::Text(t))) => {
|
||||
let is_close = serde_json::from_str::<serde_json::Value>(t.as_str())
|
||||
.ok()
|
||||
.is_some_and(|v| v["type"] == "close_stdin");
|
||||
if is_close {
|
||||
// Dropping the sender ends the writer, which drops the
|
||||
// child's stdin: EOF, like Docker's CloseWrite.
|
||||
in_tx = None;
|
||||
}
|
||||
}
|
||||
Some(Ok(Message::Ping(_) | Message::Pong(_))) => {}
|
||||
Some(Ok(Message::Close(_))) | Some(Err(_)) | None => break,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// 5. The client went away, as a dropped `docker exec` does. A tty exec
|
||||
// gets SIGHUP on its process group — what a terminal hangup delivers. A
|
||||
// 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.
|
||||
drop(in_tx);
|
||||
if tty {
|
||||
spawn::signal_group(pid, libc::SIGHUP);
|
||||
}
|
||||
}
|
||||
|
||||
/// The command could not start: register it as finished, and tell the client
|
||||
/// the way a Docker attach would show it — started, the reason on stderr,
|
||||
/// then the exit code.
|
||||
async fn not_started(mut socket: WebSocket, st: &Arc<AppState>, code: i32, message: String) {
|
||||
let id = st.execs.insert_finished(code);
|
||||
let _ = socket.send(text(serde_json::json!({ "type": "started", "exec_id": id }))).await;
|
||||
let _ = socket.send(frame(2, format!("{message}\n").as_bytes())).await;
|
||||
let _ = socket.send(text(serde_json::json!({ "type": "exit", "code": code }))).await;
|
||||
let _ = socket.send(Message::Close(None)).await;
|
||||
}
|
||||
|
||||
/// Send output unless the socket side has gone; output after that is drained
|
||||
/// and discarded so the process never blocks on a full pipe.
|
||||
async fn emit(tx: &mpsc::Sender<Event>, ch: u8, data: &[u8]) {
|
||||
let _ = tx.send(Event::Out(ch, Bytes::copy_from_slice(data))).await;
|
||||
}
|
||||
|
||||
async fn pump_pipes(
|
||||
mut child: tokio::process::Child,
|
||||
mut input: mpsc::UnboundedReceiver<Input>,
|
||||
tx: &mpsc::Sender<Event>,
|
||||
) -> i32 {
|
||||
let mut stdout = child.stdout.take().expect("piped stdout");
|
||||
let mut stderr = child.stderr.take().expect("piped stderr");
|
||||
if let Some(mut stdin) = child.stdin.take() {
|
||||
tokio::spawn(async move {
|
||||
while let Some(Input::Data(d)) = input.recv().await {
|
||||
if stdin.write_all(&d).await.is_err() {
|
||||
break;
|
||||
}
|
||||
let _ = stdin.flush().await;
|
||||
}
|
||||
// `stdin` drops here: EOF for the child.
|
||||
});
|
||||
}
|
||||
let (mut ob, mut eb) = (vec![0u8; 16 * 1024], vec![0u8; 16 * 1024]);
|
||||
let (mut out_open, mut err_open) = (true, true);
|
||||
let mut code: Option<i32> = None;
|
||||
// The exit code is taken as soon as the process exits (so `GET
|
||||
// /v1/exec/{id}` is not held up by a grandchild still holding a pipe), but
|
||||
// the exit *frame* waits for both streams to end: "all output has been
|
||||
// sent" first, as the spec requires.
|
||||
while out_open || err_open || code.is_none() {
|
||||
tokio::select! {
|
||||
r = stdout.read(&mut ob), if out_open => match r {
|
||||
Ok(n) if n > 0 => emit(tx, 1, &ob[..n]).await,
|
||||
_ => out_open = false,
|
||||
},
|
||||
r = stderr.read(&mut eb), if err_open => match r {
|
||||
Ok(n) if n > 0 => emit(tx, 2, &eb[..n]).await,
|
||||
_ => err_open = false,
|
||||
},
|
||||
s = child.wait(), if code.is_none() => {
|
||||
code = Some(s.map(spawn::exit_code).unwrap_or(255));
|
||||
}
|
||||
}
|
||||
}
|
||||
code.unwrap_or(255)
|
||||
}
|
||||
|
||||
async fn pump_tty(
|
||||
mut child: tokio::process::Child,
|
||||
master: Arc<pty::PtyMaster>,
|
||||
mut input: mpsc::UnboundedReceiver<Input>,
|
||||
attach_stdin: bool,
|
||||
tx: &mpsc::Sender<Event>,
|
||||
) -> i32 {
|
||||
if attach_stdin {
|
||||
let master = master.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(Input::Data(d)) = input.recv().await {
|
||||
if master.write_all(&d).await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
// A tty has no EOF to give: closing stdin on a terminal is not
|
||||
// a thing, and the pty must stay open for output. Docker's
|
||||
// CloseWrite on a tty exec is likewise not an EOF to the process.
|
||||
});
|
||||
}
|
||||
let mut buf = vec![0u8; 16 * 1024];
|
||||
let mut code: Option<i32> = None;
|
||||
loop {
|
||||
if let Some(c) = code {
|
||||
match tokio::time::timeout(TTY_DRAIN_IDLE, master.read(&mut buf)).await {
|
||||
Ok(Ok(n)) if n > 0 => emit(tx, 1, &buf[..n]).await,
|
||||
_ => return c,
|
||||
}
|
||||
continue;
|
||||
}
|
||||
tokio::select! {
|
||||
r = master.read(&mut buf) => match r {
|
||||
// Everything is channel 1 on a tty: Docker's `Console` stream
|
||||
// has no stderr, and the app maps it to stdout.
|
||||
Ok(n) if n > 0 => emit(tx, 1, &buf[..n]).await,
|
||||
_ => {
|
||||
let s = child.wait().await;
|
||||
return s.map(spawn::exit_code).unwrap_or(255);
|
||||
}
|
||||
},
|
||||
s = child.wait() => {
|
||||
code = Some(s.map(spawn::exit_code).unwrap_or(255));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,280 @@
|
||||
//! One-shot and detached execs, exec status, and resize.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use axum::extract::{Path, State};
|
||||
use axum::http::StatusCode;
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::Json;
|
||||
use bytes::Bytes;
|
||||
use base64::Engine;
|
||||
use serde::Deserialize;
|
||||
use tokio::io::AsyncReadExt;
|
||||
|
||||
use crate::pty::PtyMaster;
|
||||
use crate::spawn::{self, ExecSpec, Io, PrepareError};
|
||||
use crate::{error, AppState, CountGuard, MAX_ONESHOT_OUTPUT};
|
||||
|
||||
/// How long a finished exec stays answerable. The spec promises "at least 10
|
||||
/// minutes": the app polls `GET /v1/exec/{id}` after an attach ends, and a
|
||||
/// slow poll (a laptop waking from sleep) must not turn a real exit code into
|
||||
/// an unknown id.
|
||||
const KEEP_FINISHED: Duration = Duration::from_secs(10 * 60);
|
||||
|
||||
pub struct ExecEntry {
|
||||
pub running: bool,
|
||||
pub exit_code: Option<i32>,
|
||||
finished_at: Option<Instant>,
|
||||
/// Present only for tty execs that are still running — what `resize`
|
||||
/// needs. Dropped on exit so the terminal is released.
|
||||
pub pty: Option<Arc<PtyMaster>>,
|
||||
}
|
||||
|
||||
/// Exec ids → status, for `detached` and `attach` execs.
|
||||
#[derive(Default)]
|
||||
pub struct Registry {
|
||||
map: Mutex<HashMap<String, ExecEntry>>,
|
||||
}
|
||||
|
||||
impl Registry {
|
||||
/// Register a running exec and return its id. Expired entries are pruned
|
||||
/// on the way in, which bounds the map by the insert rate × 10 minutes.
|
||||
pub fn insert_running(&self, pty: Option<Arc<PtyMaster>>) -> String {
|
||||
self.insert(ExecEntry {
|
||||
running: true,
|
||||
exit_code: None,
|
||||
finished_at: None,
|
||||
pty,
|
||||
})
|
||||
}
|
||||
|
||||
/// Register an exec that never started (126/127), so its id answers like
|
||||
/// any finished exec — which is how Docker presents it.
|
||||
pub fn insert_finished(&self, code: i32) -> String {
|
||||
self.insert(ExecEntry {
|
||||
running: false,
|
||||
exit_code: Some(code),
|
||||
finished_at: Some(Instant::now()),
|
||||
pty: None,
|
||||
})
|
||||
}
|
||||
|
||||
fn insert(&self, entry: ExecEntry) -> String {
|
||||
let id = new_id();
|
||||
let mut map = self.map.lock().unwrap();
|
||||
map.retain(|_, e| e.finished_at.is_none_or(|t| t.elapsed() < KEEP_FINISHED));
|
||||
map.insert(id.clone(), entry);
|
||||
id
|
||||
}
|
||||
|
||||
pub fn finish(&self, id: &str, code: i32) {
|
||||
if let Some(e) = self.map.lock().unwrap().get_mut(id) {
|
||||
e.running = false;
|
||||
e.exit_code = Some(code);
|
||||
e.finished_at = Some(Instant::now());
|
||||
e.pty = None;
|
||||
}
|
||||
}
|
||||
|
||||
fn with<R>(&self, id: &str, f: impl FnOnce(&ExecEntry) -> R) -> Option<R> {
|
||||
self.map.lock().unwrap().get(id).map(f)
|
||||
}
|
||||
}
|
||||
|
||||
/// 64 hex characters from the kernel's CSPRNG — Docker's exec id shape. Ids
|
||||
/// are not secrets (the token is the boundary), but guessable ones would let
|
||||
/// one client poke at another's execs for no reason.
|
||||
fn new_id() -> String {
|
||||
let mut bytes = [0u8; 32];
|
||||
let mut filled = 0;
|
||||
while filled < bytes.len() {
|
||||
let n = unsafe {
|
||||
libc::getrandom(bytes[filled..].as_mut_ptr().cast(), bytes.len() - filled, 0)
|
||||
};
|
||||
if n > 0 {
|
||||
filled += n as usize;
|
||||
} else if std::io::Error::last_os_error().kind() != std::io::ErrorKind::Interrupted {
|
||||
// getrandom(2) cannot fail on a supported kernel once the pool is
|
||||
// initialised; panicking beats handing out predictable ids.
|
||||
panic!("getrandom failed: {}", std::io::Error::last_os_error());
|
||||
}
|
||||
}
|
||||
bytes.iter().map(|b| format!("{b:02x}")).collect()
|
||||
}
|
||||
|
||||
/// Map a spec refusal onto an HTTP error response.
|
||||
pub fn refusal(status: StatusCode, msg: String) -> Response {
|
||||
error(status, msg)
|
||||
}
|
||||
|
||||
/// Parse a JSON body without insisting on `Content-Type: application/json`.
|
||||
/// The spec says bodies are JSON and says nothing about the header, so a
|
||||
/// client that omits it (curl `-d`, a hand-rolled hyper request) must not get
|
||||
/// a 415 for a body that is perfectly good JSON.
|
||||
fn parse_json<T: serde::de::DeserializeOwned>(body: &[u8]) -> Result<T, String> {
|
||||
serde_json::from_slice(body).map_err(|e| format!("invalid request body: {e}"))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct OneshotRequest {
|
||||
#[serde(flatten)]
|
||||
spec: ExecSpec,
|
||||
#[serde(default)]
|
||||
limit: Option<usize>,
|
||||
}
|
||||
|
||||
/// `POST /v1/exec`.
|
||||
pub async fn oneshot(
|
||||
State(st): State<Arc<AppState>>,
|
||||
body: Bytes,
|
||||
) -> Response {
|
||||
let req: OneshotRequest = match parse_json(&body) {
|
||||
Ok(r) => r,
|
||||
Err(e) => return error(StatusCode::BAD_REQUEST, e),
|
||||
};
|
||||
let limit = req.limit.unwrap_or(MAX_ONESHOT_OUTPUT);
|
||||
// A one-shot never has a tty (spec), whatever the spec's `tty` says.
|
||||
let prepared = spawn::prepare(&st.config, &req.spec, false);
|
||||
let child = prepared.and_then(|p| p.spawn(Io::Null, Io::Piped, Io::Piped, false));
|
||||
let mut child = match child {
|
||||
Ok(c) => c,
|
||||
Err(PrepareError::Refused(s, m)) => return refusal(s, m),
|
||||
Err(PrepareError::CannotStart { code, message }) => {
|
||||
// Docker's shape for an exec that could not start: a finished exec
|
||||
// with 126/127 and the reason on stderr — not an API error.
|
||||
return oneshot_response(&[(false, format!("{message}\n").into_bytes())], code);
|
||||
}
|
||||
};
|
||||
let _active = CountGuard::execs(&st);
|
||||
let pid = child.id();
|
||||
let mut stdout = child.stdout.take().expect("piped stdout");
|
||||
let mut stderr = child.stderr.take().expect("piped stderr");
|
||||
|
||||
// (is_stdout, bytes) in arrival order; adjacent reads from the same
|
||||
// stream are merged, which changes nothing a client can rebuild.
|
||||
let mut chunks: Vec<(bool, Vec<u8>)> = Vec::new();
|
||||
let mut total = 0usize;
|
||||
let (mut out_open, mut err_open) = (true, true);
|
||||
let mut ob = vec![0u8; 16 * 1024];
|
||||
let mut eb = vec![0u8; 16 * 1024];
|
||||
while out_open || err_open {
|
||||
let (is_out, n) = tokio::select! {
|
||||
r = stdout.read(&mut ob), if out_open => (true, r.unwrap_or(0)),
|
||||
r = stderr.read(&mut eb), if err_open => (false, r.unwrap_or(0)),
|
||||
};
|
||||
if n == 0 {
|
||||
if is_out { out_open = false } else { err_open = false }
|
||||
continue;
|
||||
}
|
||||
// The same test as `docker/exec.rs`'s `push_capped`: refuse the chunk
|
||||
// that would take the total past `limit`, so exactly `limit` bytes is
|
||||
// still a success on both transports.
|
||||
if total + n > limit {
|
||||
// Kill the whole process group, not just the pid: a pipeline or a
|
||||
// backgrounded writer would otherwise keep running with nobody
|
||||
// reading. Reaped in the background so the response is not held
|
||||
// hostage by a process stuck in uninterruptible sleep.
|
||||
spawn::signal_group(pid, libc::SIGKILL);
|
||||
tokio::spawn(async move {
|
||||
let _ = child.wait().await;
|
||||
});
|
||||
// Byte-for-byte the Docker path's message, so the client can pass
|
||||
// it through and `OUTPUT_LIMIT_MARKER` checks keep matching.
|
||||
return error(
|
||||
StatusCode::PAYLOAD_TOO_LARGE,
|
||||
format!("OUTPUT_LIMIT: Command output exceeded {limit} bytes and was abandoned"),
|
||||
);
|
||||
}
|
||||
total += n;
|
||||
let data = if is_out { &ob[..n] } else { &eb[..n] };
|
||||
match chunks.last_mut() {
|
||||
Some((s, buf)) if *s == is_out => buf.extend_from_slice(data),
|
||||
_ => chunks.push((is_out, data.to_vec())),
|
||||
}
|
||||
}
|
||||
let code = match child.wait().await {
|
||||
Ok(status) => spawn::exit_code(status),
|
||||
Err(e) => return error(StatusCode::INTERNAL_SERVER_ERROR, format!("wait failed: {e}")),
|
||||
};
|
||||
oneshot_response(&chunks, code)
|
||||
}
|
||||
|
||||
fn oneshot_response(chunks: &[(bool, Vec<u8>)], code: i32) -> Response {
|
||||
let b64 = base64::engine::general_purpose::STANDARD;
|
||||
let chunks: Vec<_> = chunks
|
||||
.iter()
|
||||
.map(|(is_out, d)| serde_json::json!({ "s": if *is_out { "out" } else { "err" }, "d": b64.encode(d) }))
|
||||
.collect();
|
||||
Json(serde_json::json!({ "chunks": chunks, "exit_code": code })).into_response()
|
||||
}
|
||||
|
||||
/// `POST /v1/exec/detached`: stdin, stdout and stderr all `/dev/null`.
|
||||
pub async fn detached(
|
||||
State(st): State<Arc<AppState>>,
|
||||
body: Bytes,
|
||||
) -> Response {
|
||||
let spec: ExecSpec = match parse_json(&body) {
|
||||
Ok(r) => r,
|
||||
Err(e) => return error(StatusCode::BAD_REQUEST, e),
|
||||
};
|
||||
let child = spawn::prepare(&st.config, &spec, false)
|
||||
.and_then(|p| p.spawn(Io::Null, Io::Null, Io::Null, false));
|
||||
let id = match child {
|
||||
Ok(mut child) => {
|
||||
let id = st.execs.insert_running(None);
|
||||
let guard = CountGuard::execs(&st);
|
||||
let (st2, id2) = (st.clone(), id.clone());
|
||||
tokio::spawn(async move {
|
||||
let _active = guard;
|
||||
let code = child.wait().await.map(spawn::exit_code).unwrap_or(255);
|
||||
st2.execs.finish(&id2, code);
|
||||
});
|
||||
id
|
||||
}
|
||||
Err(PrepareError::Refused(s, m)) => return refusal(s, m),
|
||||
// Docker accepts a detached exec of a missing binary and reports the
|
||||
// failure as its exit code; the output has nowhere to go.
|
||||
Err(PrepareError::CannotStart { code, .. }) => st.execs.insert_finished(code),
|
||||
};
|
||||
Json(serde_json::json!({ "exec_id": id })).into_response()
|
||||
}
|
||||
|
||||
/// `GET /v1/exec/{id}`.
|
||||
pub async fn status(State(st): State<Arc<AppState>>, Path(id): Path<String>) -> Response {
|
||||
match st.execs.with(&id, |e| (e.running, e.exit_code)) {
|
||||
Some((running, code)) => {
|
||||
Json(serde_json::json!({ "running": running, "exit_code": code })).into_response()
|
||||
}
|
||||
None => error(StatusCode::NOT_FOUND, format!("no such exec: {id}")),
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct ResizeRequest {
|
||||
cols: u16,
|
||||
rows: u16,
|
||||
}
|
||||
|
||||
/// `POST /v1/exec/{id}/resize`: `204`, or `409` for anything that is not a
|
||||
/// running tty exec.
|
||||
pub async fn resize(
|
||||
State(st): State<Arc<AppState>>,
|
||||
Path(id): Path<String>,
|
||||
body: Bytes,
|
||||
) -> Response {
|
||||
let req: ResizeRequest = match parse_json(&body) {
|
||||
Ok(r) => r,
|
||||
Err(e) => return error(StatusCode::BAD_REQUEST, e),
|
||||
};
|
||||
let pty = match st.execs.with(&id, |e| e.pty.clone()) {
|
||||
None => return error(StatusCode::NOT_FOUND, format!("no such exec: {id}")),
|
||||
Some(None) => return error(StatusCode::CONFLICT, "exec is not a running tty exec"),
|
||||
Some(Some(p)) => p,
|
||||
};
|
||||
match pty.resize(req.cols, req.rows) {
|
||||
Ok(()) => StatusCode::NO_CONTENT.into_response(),
|
||||
Err(e) => error(StatusCode::INTERNAL_SERVER_ERROR, format!("resize failed: {e}")),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,256 @@
|
||||
//! `PUT /v1/files` and `GET /v1/archive`: the agent's side of the app's file
|
||||
//! upload and Docker's archive download.
|
||||
|
||||
use std::io::{self, Write};
|
||||
use std::os::fd::AsRawFd;
|
||||
use std::os::unix::fs::OpenOptionsExt;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
|
||||
use axum::body::Body;
|
||||
use axum::extract::rejection::QueryRejection;
|
||||
use axum::extract::{Query, State};
|
||||
use axum::http::{header, StatusCode};
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::Json;
|
||||
use bytes::Bytes;
|
||||
use futures_util::StreamExt;
|
||||
use serde::Deserialize;
|
||||
use tokio::io::AsyncWriteExt;
|
||||
use tokio::sync::mpsc;
|
||||
|
||||
use crate::{error, AppState};
|
||||
|
||||
fn query_rejection(r: QueryRejection) -> Response {
|
||||
error(StatusCode::BAD_REQUEST, format!("invalid query: {}", r.body_text()))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct PutQuery {
|
||||
dir: String,
|
||||
name: String,
|
||||
/// Octal, e.g. `644`. Defaults to 0644, the mode the app's uploads use.
|
||||
#[serde(default)]
|
||||
mode: Option<String>,
|
||||
#[serde(default)]
|
||||
uid: Option<u32>,
|
||||
#[serde(default)]
|
||||
gid: Option<u32>,
|
||||
}
|
||||
|
||||
/// A name the spec allows: one path component, nothing that walks.
|
||||
fn valid_name(name: &str) -> bool {
|
||||
!name.is_empty() && name != "." && name != ".." && !name.contains('/') && !name.contains('\0')
|
||||
}
|
||||
|
||||
/// `PUT /v1/files?dir=&name=&mode=&uid=&gid=`.
|
||||
///
|
||||
/// Mirrors what Docker's archive extractor does with the app's single-file
|
||||
/// tar (`build_single_file_tar`): an existing file is unlinked and a new one
|
||||
/// created — never written through, so a hard link or a file another process
|
||||
/// holds open keeps its old content — then given `mode` and `uid:gid`, mtime
|
||||
/// now. `dir` is not created: the app has already checked it, and silently
|
||||
/// making directories is how a typo becomes a stray tree.
|
||||
pub async fn put_file(
|
||||
State(st): State<Arc<AppState>>,
|
||||
q: Result<Query<PutQuery>, QueryRejection>,
|
||||
body: Body,
|
||||
) -> Response {
|
||||
let Query(q) = match q {
|
||||
Ok(q) => q,
|
||||
Err(r) => return query_rejection(r),
|
||||
};
|
||||
if !q.dir.starts_with('/') {
|
||||
return error(StatusCode::BAD_REQUEST, "dir must be an absolute path");
|
||||
}
|
||||
if !valid_name(&q.name) {
|
||||
return error(StatusCode::BAD_REQUEST, "name must be a single path component");
|
||||
}
|
||||
let mode = match q.mode.as_deref() {
|
||||
None => 0o644,
|
||||
Some(m) => match u32::from_str_radix(m, 8) {
|
||||
Ok(m) if m <= 0o7777 => m,
|
||||
_ => return error(StatusCode::BAD_REQUEST, format!("invalid octal mode: {m}")),
|
||||
},
|
||||
};
|
||||
let (euid, egid) = unsafe { (libc::geteuid(), libc::getegid()) };
|
||||
let uid = q.uid.unwrap_or(euid);
|
||||
let gid = q.gid.unwrap_or(egid);
|
||||
// Same rule as execs: a non-root agent (tests) can only act as itself.
|
||||
// Checked before touching the filesystem, so a refusal changes nothing.
|
||||
if euid != 0 && (uid != euid || gid != egid) {
|
||||
return error(StatusCode::FORBIDDEN, "the agent is not root and cannot chown to another user");
|
||||
}
|
||||
match std::fs::metadata(&q.dir) {
|
||||
Ok(m) if m.is_dir() => {}
|
||||
Ok(_) => return error(StatusCode::BAD_REQUEST, format!("{} is not a directory", q.dir)),
|
||||
Err(e) if e.kind() == io::ErrorKind::NotFound => {
|
||||
return error(StatusCode::NOT_FOUND, format!("{} does not exist", q.dir))
|
||||
}
|
||||
Err(e) => return error(StatusCode::INTERNAL_SERVER_ERROR, format!("{}: {e}", q.dir)),
|
||||
}
|
||||
let path = PathBuf::from(&q.dir).join(&q.name);
|
||||
let shown = format!("{}/{}", q.dir.trim_end_matches('/'), q.name);
|
||||
|
||||
// Unlink, then create — tar extraction's replace. A directory in the way
|
||||
// is refused rather than removed recursively.
|
||||
match std::fs::symlink_metadata(&path) {
|
||||
Ok(m) if m.is_dir() => {
|
||||
return error(StatusCode::CONFLICT, format!("{shown} is a directory"));
|
||||
}
|
||||
Ok(_) => {
|
||||
if let Err(e) = std::fs::remove_file(&path) {
|
||||
return error(StatusCode::INTERNAL_SERVER_ERROR, format!("cannot replace {shown}: {e}"));
|
||||
}
|
||||
}
|
||||
Err(_) => {}
|
||||
}
|
||||
let file = std::fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.mode(0o600)
|
||||
.custom_flags(libc::O_NOFOLLOW)
|
||||
.open(&path);
|
||||
let file = match file {
|
||||
Ok(f) => f,
|
||||
Err(e) => return error(StatusCode::INTERNAL_SERVER_ERROR, format!("cannot create {shown}: {e}")),
|
||||
};
|
||||
// Owner before mode: chown clears setuid/setgid bits, so the mode has to
|
||||
// land last to survive. fchmod rather than the open mode so the umask
|
||||
// does not get a say — tar extraction as root restores modes exactly.
|
||||
let fd = file.as_raw_fd();
|
||||
let set = unsafe {
|
||||
(euid != 0 || libc::fchown(fd, uid, gid) == 0) && libc::fchmod(fd, mode as libc::mode_t) == 0
|
||||
};
|
||||
if !set {
|
||||
let e = io::Error::last_os_error();
|
||||
let _ = std::fs::remove_file(&path);
|
||||
return error(StatusCode::INTERNAL_SERVER_ERROR, format!("cannot set owner/mode on {shown}: {e}"));
|
||||
}
|
||||
|
||||
let mut file = tokio::fs::File::from_std(file);
|
||||
let mut stream = body.into_data_stream();
|
||||
let mut written: u64 = 0;
|
||||
let fail = |status, msg: String| {
|
||||
let _ = std::fs::remove_file(&path);
|
||||
error(status, msg)
|
||||
};
|
||||
while let Some(chunk) = stream.next().await {
|
||||
let chunk = match chunk {
|
||||
Ok(c) => c,
|
||||
Err(e) => return fail(StatusCode::BAD_REQUEST, format!("upload interrupted: {e}")),
|
||||
};
|
||||
written += chunk.len() as u64;
|
||||
if written > st.config.max_file_bytes {
|
||||
return fail(
|
||||
StatusCode::PAYLOAD_TOO_LARGE,
|
||||
format!("File too large to upload (limit {} MB)", st.config.max_file_bytes / (1024 * 1024)),
|
||||
);
|
||||
}
|
||||
if let Err(e) = file.write_all(&chunk).await {
|
||||
return fail(StatusCode::INTERNAL_SERVER_ERROR, format!("cannot write {shown}: {e}"));
|
||||
}
|
||||
}
|
||||
if let Err(e) = file.flush().await {
|
||||
return fail(StatusCode::INTERNAL_SERVER_ERROR, format!("cannot write {shown}: {e}"));
|
||||
}
|
||||
Json(serde_json::json!({ "path": shown })).into_response()
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct ArchiveQuery {
|
||||
path: String,
|
||||
}
|
||||
|
||||
/// `GET /v1/archive?path=`: a streamed tar whose root member is
|
||||
/// `basename(path)`, recursing into directories — the shape of Docker's
|
||||
/// `GET /containers/{id}/archive`, which is what the app's download parses.
|
||||
///
|
||||
/// Symlinks are archived as links, not followed, at the root as well as
|
||||
/// inside: Docker does the same, and following them would let a link in the
|
||||
/// tree pull in files from anywhere in the container.
|
||||
pub async fn archive(q: Result<Query<ArchiveQuery>, QueryRejection>) -> Response {
|
||||
let Query(q) = match q {
|
||||
Ok(q) => q,
|
||||
Err(r) => return query_rejection(r),
|
||||
};
|
||||
if !q.path.starts_with('/') {
|
||||
return error(StatusCode::BAD_REQUEST, "path must be absolute");
|
||||
}
|
||||
let path = PathBuf::from(&q.path);
|
||||
let meta = match std::fs::symlink_metadata(&path) {
|
||||
Ok(m) => m,
|
||||
Err(e) if e.kind() == io::ErrorKind::NotFound => {
|
||||
return error(StatusCode::NOT_FOUND, format!("{} does not exist", q.path))
|
||||
}
|
||||
Err(e) => return error(StatusCode::INTERNAL_SERVER_ERROR, format!("{}: {e}", q.path)),
|
||||
};
|
||||
let root_name = path
|
||||
.file_name()
|
||||
.map(PathBuf::from)
|
||||
.unwrap_or_else(|| PathBuf::from("."));
|
||||
|
||||
// tar is a synchronous writer, so it runs on a blocking thread feeding a
|
||||
// small channel. When the client disconnects, the body (and receiver) is
|
||||
// dropped, the next send fails, and the walk stops — "dropping the
|
||||
// connection cancels the read".
|
||||
let (tx, mut rx) = mpsc::channel::<io::Result<Bytes>>(8);
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let mut w = ChannelWriter { tx: tx.clone(), buf: Vec::with_capacity(CHUNK) };
|
||||
let res = (|| {
|
||||
let mut b = tar::Builder::new(&mut w);
|
||||
b.follow_symlinks(false);
|
||||
b.mode(tar::HeaderMode::Complete);
|
||||
if meta.is_dir() {
|
||||
b.append_dir_all(&root_name, &path)?;
|
||||
} else {
|
||||
b.append_path_with_name(&path, &root_name)?;
|
||||
}
|
||||
b.finish()
|
||||
})();
|
||||
let res = res.and_then(|()| w.flush());
|
||||
if let Err(e) = res {
|
||||
// Surface as a body error: the client sees a truncated, failed
|
||||
// download rather than a short archive that looks complete.
|
||||
let _ = tx.blocking_send(Err(e));
|
||||
}
|
||||
});
|
||||
let stream = futures_util::stream::poll_fn(move |cx| rx.poll_recv(cx));
|
||||
Response::builder()
|
||||
.header(header::CONTENT_TYPE, "application/x-tar")
|
||||
.body(Body::from_stream(stream))
|
||||
.unwrap_or_else(|_| error(StatusCode::INTERNAL_SERVER_ERROR, "response build failed"))
|
||||
}
|
||||
|
||||
const CHUNK: usize = 64 * 1024;
|
||||
|
||||
struct ChannelWriter {
|
||||
tx: mpsc::Sender<io::Result<Bytes>>,
|
||||
buf: Vec<u8>,
|
||||
}
|
||||
|
||||
impl ChannelWriter {
|
||||
fn send(&mut self) -> io::Result<()> {
|
||||
if self.buf.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
let data = Bytes::from(std::mem::replace(&mut self.buf, Vec::with_capacity(CHUNK)));
|
||||
self.tx
|
||||
.blocking_send(Ok(data))
|
||||
.map_err(|_| io::Error::new(io::ErrorKind::BrokenPipe, "client went away"))
|
||||
}
|
||||
}
|
||||
|
||||
impl Write for ChannelWriter {
|
||||
fn write(&mut self, data: &[u8]) -> io::Result<usize> {
|
||||
self.buf.extend_from_slice(data);
|
||||
if self.buf.len() >= CHUNK {
|
||||
self.send()?;
|
||||
}
|
||||
Ok(data.len())
|
||||
}
|
||||
fn flush(&mut self) -> io::Result<()> {
|
||||
self.send()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
//! `triple-c-agent`: everything `host::ContainerTarget` does through the Docker
|
||||
//! API, done from inside the container over authenticated HTTP/WebSocket.
|
||||
//!
|
||||
//! The contract is `docs/superpowers/specs/2026-11-13-triple-c-agent-protocol.md`.
|
||||
//! **Docker parity is the requirement**: the same `ExecSpec` must behave the
|
||||
//! same here as through `docker exec`, because the app's call sites were
|
||||
//! written against Docker and are not going to grow a second set of rules.
|
||||
//! Each place this code reproduces a Docker behaviour says so where it does it.
|
||||
//!
|
||||
//! The library exists so the integration tests can run the real server
|
||||
//! in-process on an ephemeral port; `main.rs` only reads the environment.
|
||||
|
||||
mod attach;
|
||||
mod exec;
|
||||
mod files;
|
||||
mod pty;
|
||||
mod spawn;
|
||||
mod tunnel;
|
||||
mod users;
|
||||
|
||||
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 axum::extract::{Request, State};
|
||||
use axum::http::{header, StatusCode};
|
||||
use axum::middleware::{self, Next};
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::routing::{get, post, put};
|
||||
use axum::{Json, Router};
|
||||
|
||||
/// The wire protocol version reported by `/v1/health`.
|
||||
pub const PROTOCOL_VERSION: u32 = 1;
|
||||
|
||||
/// Upload ceiling for `PUT /v1/files`: the app's `MAX_DROP_BYTES`
|
||||
/// (`docker/exec.rs`). The app checks it before sending, so this only has to
|
||||
/// stop a body that lies about itself.
|
||||
pub const MAX_DROP_BYTES: u64 = 256 * 1024 * 1024;
|
||||
|
||||
/// The default one-shot budget when a request omits `limit` — the app's
|
||||
/// `MAX_ONESHOT_OUTPUT`. Every real client call sends its own.
|
||||
pub const MAX_ONESHOT_OUTPUT: usize = 8 * 1024 * 1024;
|
||||
|
||||
/// The shortest token the agent will accept. There is no unauthenticated mode,
|
||||
/// 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.
|
||||
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";
|
||||
|
||||
/// Everything the server needs, captured once at startup.
|
||||
#[derive(Clone)]
|
||||
pub struct Config {
|
||||
/// The bearer token every request must carry.
|
||||
pub token: String,
|
||||
/// Where an exec with `working_dir: null` runs: the agent's own cwd at
|
||||
/// startup, which is the container's `WorkingDir` — what `docker exec`
|
||||
/// without `-w` uses.
|
||||
pub default_cwd: PathBuf,
|
||||
/// The env every exec starts from: the agent's own at startup, minus the
|
||||
/// token. That is the container's configured env, which is what a
|
||||
/// `docker exec` inherits.
|
||||
pub base_env: Vec<(OsString, OsString)>,
|
||||
/// `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,
|
||||
}
|
||||
|
||||
impl Config {
|
||||
/// Capture the process's own cwd and env. Call after the token has been
|
||||
/// removed from the environment; the token is filtered here as well, so a
|
||||
/// caller that forgets cannot leak it into execs.
|
||||
pub fn from_process(token: String) -> std::io::Result<Config> {
|
||||
Ok(Config {
|
||||
token,
|
||||
default_cwd: std::env::current_dir()?,
|
||||
base_env: std::env::vars_os().filter(|(k, _)| k != TOKEN_ENV).collect(),
|
||||
max_file_bytes: MAX_DROP_BYTES,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Read the token as the spec says: the env var first, then the file. Refuses
|
||||
/// anything empty or shorter than [`MIN_TOKEN_BYTES`] — the agent has no
|
||||
/// unauthenticated mode to fall back to.
|
||||
///
|
||||
/// Surrounding whitespace is trimmed: a token file written by `echo` or a
|
||||
/// secrets manager usually ends in a newline, and a trailing `\n` the client
|
||||
/// never sends would be a token nobody can present.
|
||||
pub fn load_token(env_token: Option<OsString>, file: Option<PathBuf>) -> Result<String, String> {
|
||||
let (raw, source) = match env_token.filter(|t| !t.is_empty()) {
|
||||
Some(t) => (
|
||||
t.into_string()
|
||||
.map_err(|_| format!("{TOKEN_ENV} is not valid UTF-8"))?,
|
||||
TOKEN_ENV.to_string(),
|
||||
),
|
||||
None => match file {
|
||||
Some(path) => (
|
||||
std::fs::read_to_string(&path)
|
||||
.map_err(|e| format!("cannot read {TOKEN_FILE_ENV} ({}): {e}", path.display()))?,
|
||||
format!("{TOKEN_FILE_ENV} ({})", path.display()),
|
||||
),
|
||||
None => {
|
||||
return Err(format!(
|
||||
"no token: set {TOKEN_ENV} or {TOKEN_FILE_ENV}; refusing to start unauthenticated"
|
||||
))
|
||||
}
|
||||
},
|
||||
};
|
||||
let token = raw.trim().to_string();
|
||||
if token.len() < MIN_TOKEN_BYTES {
|
||||
return Err(format!(
|
||||
"token from {source} is {} bytes; at least {MIN_TOKEN_BYTES} are required",
|
||||
token.len()
|
||||
));
|
||||
}
|
||||
Ok(token)
|
||||
}
|
||||
|
||||
/// Shared server state.
|
||||
pub(crate) struct AppState {
|
||||
pub config: Config,
|
||||
pub execs: exec::Registry,
|
||||
pub activity: Activity,
|
||||
}
|
||||
|
||||
/// What `/v1/activity` reports; the input to idle auto-stop.
|
||||
pub(crate) struct Activity {
|
||||
last_unix: AtomicI64,
|
||||
pub execs: AtomicI64,
|
||||
pub tunnels: AtomicI64,
|
||||
}
|
||||
|
||||
impl Activity {
|
||||
pub fn touch(&self) {
|
||||
self.last_unix.store(now_unix(), Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
/// Increments a counter for as long as it lives, so an exec or tunnel that
|
||||
/// ends by any path — including a panic or a dropped task — stops counting.
|
||||
pub(crate) struct CountGuard(Arc<AppState>, fn(&Activity) -> &AtomicI64);
|
||||
|
||||
impl CountGuard {
|
||||
pub fn execs(st: &Arc<AppState>) -> CountGuard {
|
||||
st.activity.execs.fetch_add(1, Ordering::Relaxed);
|
||||
CountGuard(st.clone(), |a| &a.execs)
|
||||
}
|
||||
pub fn tunnels(st: &Arc<AppState>) -> CountGuard {
|
||||
st.activity.tunnels.fetch_add(1, Ordering::Relaxed);
|
||||
CountGuard(st.clone(), |a| &a.tunnels)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for CountGuard {
|
||||
fn drop(&mut self) {
|
||||
(self.1)(&self.0.activity).fetch_sub(1, Ordering::Relaxed);
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn now_unix() -> i64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map(|d| d.as_secs() as i64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
/// The spec's error shape: `{"error": "<message>"}` with a JSON content type.
|
||||
pub(crate) fn error(status: StatusCode, msg: impl Into<String>) -> Response {
|
||||
(status, Json(serde_json::json!({ "error": msg.into() }))).into_response()
|
||||
}
|
||||
|
||||
/// Every route, behind the token gate.
|
||||
fn router(state: Arc<AppState>) -> Router {
|
||||
Router::new()
|
||||
.route("/v1/health", get(health))
|
||||
.route("/v1/activity", get(activity))
|
||||
.route("/v1/exec", post(exec::oneshot))
|
||||
.route("/v1/exec/detached", post(exec::detached))
|
||||
.route("/v1/exec/{id}", get(exec::status))
|
||||
.route("/v1/exec/{id}/resize", post(exec::resize))
|
||||
.route("/v1/attach", get(attach::upgrade))
|
||||
.route(
|
||||
"/v1/files",
|
||||
// The handler streams the body and enforces its own ceiling, so
|
||||
// axum's 2 MB default must not cut it off first.
|
||||
put(files::put_file).layer(axum::extract::DefaultBodyLimit::disable()),
|
||||
)
|
||||
.route("/v1/archive", get(files::archive))
|
||||
.route("/v1/tunnel", get(tunnel::upgrade))
|
||||
.fallback(|| async { error(StatusCode::NOT_FOUND, "no such endpoint") })
|
||||
// `layer`, not `route_layer`: the fallback must be behind the token
|
||||
// too, or an unauthenticated caller could map the API by 404 vs 401.
|
||||
.layer(middleware::from_fn_with_state(state.clone(), auth))
|
||||
.with_state(state)
|
||||
}
|
||||
|
||||
/// Serve the protocol on `listener` until the listener fails.
|
||||
pub async fn serve(listener: tokio::net::TcpListener, config: Config) -> std::io::Result<()> {
|
||||
let state = Arc::new(AppState {
|
||||
config,
|
||||
execs: exec::Registry::default(),
|
||||
activity: Activity {
|
||||
last_unix: AtomicI64::new(now_unix()),
|
||||
execs: AtomicI64::new(0),
|
||||
tunnels: AtomicI64::new(0),
|
||||
},
|
||||
});
|
||||
axum::serve(listener, router(state)).await
|
||||
}
|
||||
|
||||
/// Bearer-token gate on every request. A failure is `401` with an **empty**
|
||||
/// body (spec): nothing about the API is disclosed to a caller without the
|
||||
/// token.
|
||||
async fn auth(State(st): State<Arc<AppState>>, req: Request, next: Next) -> Response {
|
||||
let presented = req
|
||||
.headers()
|
||||
.get(header::AUTHORIZATION)
|
||||
.and_then(|v| v.as_bytes().strip_prefix(b"Bearer "));
|
||||
let ok = presented.is_some_and(|p| constant_time_eq(p, st.config.token.as_bytes()));
|
||||
if !ok {
|
||||
return StatusCode::UNAUTHORIZED.into_response();
|
||||
}
|
||||
// `health` and `activity` are the idle monitor's own probes; counting them
|
||||
// would mean an idle container could never look idle.
|
||||
let path = req.uri().path();
|
||||
if path != "/v1/health" && path != "/v1/activity" {
|
||||
st.activity.touch();
|
||||
}
|
||||
next.run(req).await
|
||||
}
|
||||
|
||||
/// Compare without an early exit, so response timing says nothing about how
|
||||
/// many leading bytes of a guess were right. The length is not secret (the
|
||||
/// minimum is public and tokens are generated at a fixed size), but it is
|
||||
/// folded in rather than short-circuited anyway.
|
||||
fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
|
||||
let mut diff: usize = a.len() ^ b.len();
|
||||
for i in 0..a.len().max(b.len()) {
|
||||
let x = a.get(i).copied().unwrap_or(0);
|
||||
let y = b.get(i).copied().unwrap_or(0);
|
||||
diff |= (x ^ y) as usize;
|
||||
}
|
||||
diff == 0
|
||||
}
|
||||
|
||||
async fn health() -> Response {
|
||||
Json(serde_json::json!({
|
||||
"agent": "triple-c-agent",
|
||||
"version": env!("CARGO_PKG_VERSION"),
|
||||
"protocol": PROTOCOL_VERSION,
|
||||
}))
|
||||
.into_response()
|
||||
}
|
||||
|
||||
async fn activity(State(st): State<Arc<AppState>>) -> Response {
|
||||
Json(serde_json::json!({
|
||||
"last_activity_unix": st.activity.last_unix.load(Ordering::Relaxed),
|
||||
"active_execs": st.activity.execs.load(Ordering::Relaxed),
|
||||
"active_tunnels": st.activity.tunnels.load(Ordering::Relaxed),
|
||||
}))
|
||||
.into_response()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn constant_time_eq_matches_plain_equality() {
|
||||
assert!(constant_time_eq(b"abc", b"abc"));
|
||||
assert!(!constant_time_eq(b"abc", b"abd"));
|
||||
assert!(!constant_time_eq(b"abc", b"abcd"));
|
||||
assert!(!constant_time_eq(b"", b"a"));
|
||||
// A length difference of exactly 256 must not wrap to "equal length".
|
||||
assert!(!constant_time_eq(&[0u8; 256], b""));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_token_never_reaches_the_exec_base_env() {
|
||||
// `main` removes the var before capturing; this is the second guard,
|
||||
// for a caller that forgets.
|
||||
std::env::set_var(TOKEN_ENV, "x".repeat(40));
|
||||
let cfg = Config::from_process("t".into()).unwrap();
|
||||
std::env::remove_var(TOKEN_ENV);
|
||||
assert!(cfg.base_env.iter().all(|(k, _)| k != TOKEN_ENV));
|
||||
assert!(cfg.base_env.iter().any(|(k, _)| k == "PATH"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn token_rules() {
|
||||
let long = "x".repeat(32);
|
||||
assert_eq!(load_token(Some(long.clone().into()), None).unwrap(), long);
|
||||
assert!(load_token(Some("short".into()), None).is_err());
|
||||
assert!(load_token(None, None).is_err());
|
||||
// An empty env var falls through to the file.
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let f = dir.path().join("tok");
|
||||
std::fs::write(&f, format!("{long}\n")).unwrap();
|
||||
assert_eq!(load_token(Some("".into()), Some(f.clone())).unwrap(), long);
|
||||
std::fs::write(&f, "\n").unwrap();
|
||||
assert!(load_token(None, Some(f)).is_err());
|
||||
assert!(load_token(None, Some(dir.path().join("missing"))).is_err());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
//! `triple-c-agent` entry point: read the environment, drop the token from
|
||||
//! it, bind, serve. Everything else is in the library.
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::process::ExitCode;
|
||||
|
||||
use triple_c_agent::{load_token, serve, Config, ADDR_ENV, DEFAULT_ADDR, TOKEN_ENV, TOKEN_FILE_ENV};
|
||||
|
||||
fn main() -> ExitCode {
|
||||
let token = match load_token(
|
||||
std::env::var_os(TOKEN_ENV),
|
||||
std::env::var_os(TOKEN_FILE_ENV).map(PathBuf::from),
|
||||
) {
|
||||
Ok(t) => t,
|
||||
Err(e) => {
|
||||
eprintln!("triple-c-agent: {e}");
|
||||
return ExitCode::from(2);
|
||||
}
|
||||
};
|
||||
// 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.
|
||||
std::env::remove_var(TOKEN_ENV);
|
||||
|
||||
// Loopback by default. A non-loopback address is only ever bound when
|
||||
// someone wrote it into TRIPLEC_AGENT_ADDR on purpose.
|
||||
let addr = std::env::var(ADDR_ENV).unwrap_or_else(|_| DEFAULT_ADDR.to_string());
|
||||
let config = match Config::from_process(token) {
|
||||
Ok(c) => c,
|
||||
Err(e) => {
|
||||
eprintln!("triple-c-agent: cannot read startup state: {e}");
|
||||
return ExitCode::from(2);
|
||||
}
|
||||
};
|
||||
|
||||
let rt = match tokio::runtime::Builder::new_multi_thread().enable_all().build() {
|
||||
Ok(rt) => rt,
|
||||
Err(e) => {
|
||||
eprintln!("triple-c-agent: cannot start runtime: {e}");
|
||||
return ExitCode::from(2);
|
||||
}
|
||||
};
|
||||
rt.block_on(async move {
|
||||
let listener = match tokio::net::TcpListener::bind(&addr).await {
|
||||
Ok(l) => l,
|
||||
Err(e) => {
|
||||
eprintln!("triple-c-agent: cannot bind {addr}: {e}");
|
||||
return ExitCode::from(2);
|
||||
}
|
||||
};
|
||||
let shown = listener.local_addr().map(|a| a.to_string()).unwrap_or(addr);
|
||||
// One line, never the token.
|
||||
eprintln!("triple-c-agent {} listening on {shown}", env!("CARGO_PKG_VERSION"));
|
||||
match serve(listener, config).await {
|
||||
Ok(()) => ExitCode::SUCCESS,
|
||||
Err(e) => {
|
||||
eprintln!("triple-c-agent: server failed: {e}");
|
||||
ExitCode::FAILURE
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
//! Pseudo-terminals for `tty: true` execs.
|
||||
|
||||
use std::ffi::CStr;
|
||||
use std::io;
|
||||
use std::os::fd::{AsRawFd, FromRawFd, OwnedFd};
|
||||
|
||||
use tokio::io::unix::AsyncFd;
|
||||
|
||||
/// The agent's end of a pty. Shared (via `Arc`) by the output pump, the stdin
|
||||
/// writer and the exec registry, which needs it for `resize`.
|
||||
pub struct PtyMaster {
|
||||
fd: AsyncFd<OwnedFd>,
|
||||
}
|
||||
|
||||
/// Open a pty sized `cols`×`rows`, returning the master and the slave the
|
||||
/// child gets as stdin/stdout/stderr.
|
||||
///
|
||||
/// Both ends are opened `O_CLOEXEC` *atomically* (`posix_openpt`, then `open`
|
||||
/// on the slave path) rather than with `openpty` and a later `FD_CLOEXEC`:
|
||||
/// execs are spawned concurrently from other threads, and a master fd leaked
|
||||
/// into an unrelated child would keep this terminal open after its own process
|
||||
/// exits — the attach would never see EOF.
|
||||
pub fn open(cols: u16, rows: u16) -> io::Result<(PtyMaster, OwnedFd)> {
|
||||
unsafe {
|
||||
let m = libc::posix_openpt(libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC);
|
||||
if m < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
let master = OwnedFd::from_raw_fd(m);
|
||||
if libc::grantpt(m) != 0 || libc::unlockpt(m) != 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
let mut name = [0 as libc::c_char; 128];
|
||||
let rc = libc::ptsname_r(m, name.as_mut_ptr(), name.len());
|
||||
if rc != 0 {
|
||||
return Err(io::Error::from_raw_os_error(rc));
|
||||
}
|
||||
let s = libc::open(
|
||||
CStr::from_ptr(name.as_ptr()).as_ptr(),
|
||||
libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC,
|
||||
);
|
||||
if s < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
let slave = OwnedFd::from_raw_fd(s);
|
||||
set_size(m, cols, rows)?;
|
||||
let flags = libc::fcntl(m, libc::F_GETFL);
|
||||
if flags < 0 || libc::fcntl(m, libc::F_SETFL, flags | libc::O_NONBLOCK) < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
// Sound: `master` is an `OwnedFd`, open for as long as the AsyncFd
|
||||
// owns it, and `as_raw_fd` always returns that same descriptor.
|
||||
Ok((PtyMaster { fd: AsyncFd::register(master)? }, slave))
|
||||
}
|
||||
}
|
||||
|
||||
fn set_size(fd: libc::c_int, cols: u16, rows: u16) -> io::Result<()> {
|
||||
let ws = libc::winsize {
|
||||
ws_row: rows,
|
||||
ws_col: cols,
|
||||
ws_xpixel: 0,
|
||||
ws_ypixel: 0,
|
||||
};
|
||||
if unsafe { libc::ioctl(fd, libc::TIOCSWINSZ, &ws) } != 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
impl PtyMaster {
|
||||
/// `TIOCSWINSZ` on the master; the kernel delivers `SIGWINCH` to the
|
||||
/// terminal's foreground process group, as Docker's resize does.
|
||||
pub fn resize(&self, cols: u16, rows: u16) -> io::Result<()> {
|
||||
set_size(self.fd.get_ref().as_raw_fd(), cols, rows)
|
||||
}
|
||||
|
||||
/// Read output. `Ok(0)` is end of stream: Linux reports "every slave fd is
|
||||
/// closed" on the master as `EIO`, which is folded into EOF here.
|
||||
pub async fn read(&self, buf: &mut [u8]) -> io::Result<usize> {
|
||||
loop {
|
||||
let mut guard = self.fd.readable().await?;
|
||||
let res = guard.try_io(|inner| {
|
||||
let n = unsafe {
|
||||
libc::read(inner.get_ref().as_raw_fd(), buf.as_mut_ptr().cast(), buf.len())
|
||||
};
|
||||
if n < 0 {
|
||||
Err(io::Error::last_os_error())
|
||||
} else {
|
||||
Ok(n as usize)
|
||||
}
|
||||
});
|
||||
match res {
|
||||
Ok(Ok(n)) => return Ok(n),
|
||||
Ok(Err(e)) if e.raw_os_error() == Some(libc::EIO) => return Ok(0),
|
||||
Ok(Err(e)) if e.kind() == io::ErrorKind::Interrupted => continue,
|
||||
Ok(Err(e)) => return Err(e),
|
||||
Err(_would_block) => continue,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn write_all(&self, mut data: &[u8]) -> io::Result<()> {
|
||||
while !data.is_empty() {
|
||||
let mut guard = self.fd.writable().await?;
|
||||
let res = guard.try_io(|inner| {
|
||||
let n = unsafe {
|
||||
libc::write(inner.get_ref().as_raw_fd(), data.as_ptr().cast(), data.len())
|
||||
};
|
||||
if n < 0 {
|
||||
Err(io::Error::last_os_error())
|
||||
} else {
|
||||
Ok(n as usize)
|
||||
}
|
||||
});
|
||||
match res {
|
||||
Ok(Ok(n)) => data = &data[n..],
|
||||
Ok(Err(e)) if e.kind() == io::ErrorKind::Interrupted => continue,
|
||||
Ok(Err(e)) => return Err(e),
|
||||
Err(_would_block) => continue,
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,323 @@
|
||||
//! Turning an `ExecSpec` into a running process the way `docker exec` would.
|
||||
|
||||
use std::ffi::{CString, OsStr, OsString};
|
||||
use std::io;
|
||||
use std::os::unix::ffi::OsStrExt;
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::os::unix::process::{CommandExt, ExitStatusExt};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::process::{ExitStatus, Stdio};
|
||||
|
||||
use axum::http::StatusCode;
|
||||
use serde::Deserialize;
|
||||
|
||||
use crate::users::{self, UserInfo};
|
||||
use crate::Config;
|
||||
|
||||
/// The spec's ExecSpec. `user` may be omitted, meaning the agent's own user —
|
||||
/// as an exec with no `User` runs as the container's configured user (root).
|
||||
#[derive(Debug, Clone, Deserialize)]
|
||||
pub struct ExecSpec {
|
||||
pub cmd: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub user: Option<String>,
|
||||
#[serde(default)]
|
||||
pub working_dir: Option<String>,
|
||||
#[serde(default)]
|
||||
pub env: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub tty: bool,
|
||||
}
|
||||
|
||||
/// Docker's default `PATH` when the container env has none
|
||||
/// (`CreateDaemonEnvironment` → `oci.DefaultPathEnv`).
|
||||
const DEFAULT_PATH: &str = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin";
|
||||
|
||||
/// An exec ready to start.
|
||||
pub struct Prepared {
|
||||
program: PathBuf,
|
||||
argv: Vec<String>,
|
||||
env: Vec<(OsString, OsString)>,
|
||||
cwd: CString,
|
||||
user: UserInfo,
|
||||
/// Only when running as root. A non-root agent can only ever run as
|
||||
/// itself (anything else is refused with 403 before we get here), and
|
||||
/// calling `setgroups` unprivileged would fail the exec.
|
||||
switch_ids: bool,
|
||||
}
|
||||
|
||||
/// Why a spec could not become a process.
|
||||
pub enum PrepareError {
|
||||
/// An HTTP-level refusal (bad request, wrong user for a non-root agent).
|
||||
Refused(StatusCode, String),
|
||||
/// The command cannot start. Docker reports this as a *finished* exec with
|
||||
/// 126/127 and the reason on stderr, not as an API error, so callers turn
|
||||
/// it into exactly that.
|
||||
CannotStart { code: i32, message: String },
|
||||
}
|
||||
|
||||
impl PrepareError {
|
||||
fn cannot_start(code: i32, message: String) -> Self {
|
||||
PrepareError::CannotStart { code, message }
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate a spec and resolve user, env, working directory and program.
|
||||
pub fn prepare(cfg: &Config, spec: &ExecSpec, tty: bool) -> Result<Prepared, PrepareError> {
|
||||
if spec.cmd.is_empty() || spec.cmd[0].is_empty() {
|
||||
return Err(PrepareError::Refused(StatusCode::BAD_REQUEST, "cmd must be non-empty".into()));
|
||||
}
|
||||
if spec.cmd.iter().any(|a| a.contains('\0')) {
|
||||
return Err(PrepareError::Refused(StatusCode::BAD_REQUEST, "cmd contains a NUL byte".into()));
|
||||
}
|
||||
|
||||
let me = users::current();
|
||||
let is_root = unsafe { libc::geteuid() } == 0;
|
||||
let user = match spec.user.as_deref().filter(|u| !u.is_empty()) {
|
||||
Some(name) => users::lookup(name).ok_or_else(|| {
|
||||
// runc's wording, which is what `docker exec -u nobody-here` prints.
|
||||
PrepareError::cannot_start(
|
||||
126,
|
||||
format!("unable to find user {name}: no matching entries in passwd file"),
|
||||
)
|
||||
})?,
|
||||
None => me.clone().ok_or_else(|| {
|
||||
PrepareError::Refused(
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
"the agent's own user has no passwd entry".into(),
|
||||
)
|
||||
})?,
|
||||
};
|
||||
if !is_root && user.uid != unsafe { libc::geteuid() } {
|
||||
return Err(PrepareError::Refused(
|
||||
StatusCode::FORBIDDEN,
|
||||
format!("the agent is not root and cannot run as {}", user.name),
|
||||
));
|
||||
}
|
||||
|
||||
let env = merge_env(cfg, spec, &user, tty);
|
||||
|
||||
// `working_dir: null` is `docker exec` without `-w`: the container's
|
||||
// WorkingDir, which is the agent's cwd at startup. Checked here so a bad
|
||||
// directory is a clean 126 with runc's message rather than an errno that
|
||||
// happens to collide with "no such binary".
|
||||
let cwd = match &spec.working_dir {
|
||||
Some(d) if !d.is_empty() => PathBuf::from(d),
|
||||
_ => cfg.default_cwd.clone(),
|
||||
};
|
||||
if !cwd.is_dir() {
|
||||
return Err(PrepareError::cannot_start(
|
||||
126,
|
||||
format!(
|
||||
"chdir to cwd (\"{}\") set in config.json failed: no such file or directory",
|
||||
cwd.display()
|
||||
),
|
||||
));
|
||||
}
|
||||
|
||||
let program = resolve_program(&spec.cmd[0], &env, &cwd).ok_or_else(|| {
|
||||
// runc's text for a missing binary; Docker reports it with 127 the
|
||||
// way a shell does, and the app's callers check that code.
|
||||
PrepareError::cannot_start(
|
||||
127,
|
||||
format!(
|
||||
"OCI runtime exec failed: exec failed: unable to start container process: exec: \"{}\": executable file not found in $PATH: unknown",
|
||||
spec.cmd[0]
|
||||
),
|
||||
)
|
||||
})?;
|
||||
|
||||
Ok(Prepared {
|
||||
program,
|
||||
argv: spec.cmd.clone(),
|
||||
env,
|
||||
cwd: CString::new(cwd.as_os_str().as_bytes()).map_err(|_| {
|
||||
PrepareError::Refused(StatusCode::BAD_REQUEST, "working_dir contains a NUL byte".into())
|
||||
})?,
|
||||
switch_ids: is_root,
|
||||
user,
|
||||
})
|
||||
}
|
||||
|
||||
/// Build the exec's environment.
|
||||
///
|
||||
/// Order, matching Docker's `ReplaceOrAppendEnvValues(CreateDaemonEnvironment(…), spec.Env)`:
|
||||
/// the base env; `PATH` defaulted if absent and `TERM=xterm` for a tty if
|
||||
/// absent (both Docker daemon defaults); then the spec's entries, later ones
|
||||
/// winning. An entry with no `=` *removes* that variable, as Docker does.
|
||||
/// Finally `HOME`/`USER`/`LOGNAME` come from passwd unless the **spec** set
|
||||
/// them — the agent's own `HOME=/root` must not leak into a `claude` exec.
|
||||
fn merge_env(cfg: &Config, spec: &ExecSpec, user: &UserInfo, tty: bool) -> Vec<(OsString, OsString)> {
|
||||
let mut env: Vec<(OsString, OsString)> = cfg.base_env.clone();
|
||||
fn set(env: &mut Vec<(OsString, OsString)>, k: &OsStr, v: &OsStr) {
|
||||
match env.iter_mut().find(|(ek, _)| ek == k) {
|
||||
Some(slot) => slot.1 = v.to_os_string(),
|
||||
None => env.push((k.to_os_string(), v.to_os_string())),
|
||||
}
|
||||
}
|
||||
if !env.iter().any(|(k, _)| k == "PATH") {
|
||||
set(&mut env, OsStr::new("PATH"), OsStr::new(DEFAULT_PATH));
|
||||
}
|
||||
if tty && !env.iter().any(|(k, _)| k == "TERM") {
|
||||
set(&mut env, OsStr::new("TERM"), OsStr::new("xterm"));
|
||||
}
|
||||
let mut spec_set: Vec<&str> = Vec::new();
|
||||
for entry in &spec.env {
|
||||
match entry.split_once('=') {
|
||||
Some((k, v)) => {
|
||||
set(&mut env, OsStr::new(k), OsStr::new(v));
|
||||
spec_set.push(k);
|
||||
}
|
||||
None => {
|
||||
env.retain(|(k, _)| k != entry.as_str());
|
||||
spec_set.push(entry);
|
||||
}
|
||||
}
|
||||
}
|
||||
for (k, v) in [("HOME", &user.home), ("USER", &user.name), ("LOGNAME", &user.name)] {
|
||||
if !spec_set.contains(&k) {
|
||||
set(&mut env, OsStr::new(k), OsStr::new(v));
|
||||
}
|
||||
}
|
||||
env.retain(|(k, v)| {
|
||||
!k.is_empty() && !k.as_bytes().contains(&b'=') && !k.as_bytes().contains(&0) && !v.as_bytes().contains(&0)
|
||||
});
|
||||
env
|
||||
}
|
||||
|
||||
/// `cmd[0]` through the **merged** env's `PATH` (spec), like runc's
|
||||
/// `LookPath`. A name containing `/` is used as given (relative to the
|
||||
/// working directory, as it will be after `chdir`).
|
||||
fn resolve_program(cmd0: &str, env: &[(OsString, OsString)], cwd: &Path) -> Option<PathBuf> {
|
||||
if cmd0.contains('/') {
|
||||
return Some(PathBuf::from(cmd0));
|
||||
}
|
||||
let path = env.iter().find(|(k, _)| k == "PATH").map(|(_, v)| v.clone())?;
|
||||
for dir in std::env::split_paths(&path) {
|
||||
let dir = if dir.as_os_str().is_empty() { cwd.to_path_buf() } else { cwd.join(dir) };
|
||||
let candidate = dir.join(cmd0);
|
||||
if let Ok(meta) = std::fs::metadata(&candidate) {
|
||||
if meta.is_file() && meta.permissions().mode() & 0o111 != 0 {
|
||||
return Some(candidate);
|
||||
}
|
||||
}
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
/// What the child's stdio should be.
|
||||
pub enum Io {
|
||||
Null,
|
||||
Piped,
|
||||
Fd(std::os::fd::OwnedFd),
|
||||
}
|
||||
|
||||
impl Io {
|
||||
fn stdio(self) -> Stdio {
|
||||
match self {
|
||||
Io::Null => Stdio::null(),
|
||||
Io::Piped => Stdio::piped(),
|
||||
Io::Fd(fd) => Stdio::from(fd),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Prepared {
|
||||
/// Spawn. `controlling_tty` makes stdin (which must be a pty slave) the
|
||||
/// session's controlling terminal.
|
||||
///
|
||||
/// Every exec runs in **its own session, hence its own process group**
|
||||
/// (`setsid`), so the output-limit kill and the disconnect `SIGHUP` reach
|
||||
/// everything it started, not just the first pid.
|
||||
pub fn spawn(
|
||||
&self,
|
||||
stdin: Io,
|
||||
stdout: Io,
|
||||
stderr: Io,
|
||||
controlling_tty: bool,
|
||||
) -> Result<tokio::process::Child, PrepareError> {
|
||||
let mut cmd = std::process::Command::new(&self.program);
|
||||
cmd.arg0(&self.argv[0])
|
||||
.args(&self.argv[1..])
|
||||
.env_clear()
|
||||
.envs(self.env.iter().map(|(k, v)| (k, v)))
|
||||
.stdin(stdin.stdio())
|
||||
.stdout(stdout.stdio())
|
||||
.stderr(stderr.stdio());
|
||||
|
||||
let switch = self.switch_ids;
|
||||
let (uid, gid) = (self.user.uid, self.user.gid);
|
||||
let groups = self.user.groups.clone();
|
||||
let cwd = self.cwd.clone();
|
||||
// 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,
|
||||
// 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
|
||||
// after the switch so a directory the user cannot enter fails the way
|
||||
// it does under runc, rather than succeeding as root.
|
||||
unsafe {
|
||||
cmd.pre_exec(move || {
|
||||
if libc::setsid() < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
if controlling_tty && libc::ioctl(0, libc::TIOCSCTTY, 0) < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
if switch {
|
||||
if libc::setgid(gid) < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
if libc::setgroups(groups.len() as _, groups.as_ptr()) < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
if libc::setuid(uid) < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
}
|
||||
if libc::chdir(cwd.as_ptr()) < 0 {
|
||||
return Err(io::Error::last_os_error());
|
||||
}
|
||||
Ok(())
|
||||
});
|
||||
}
|
||||
// Never `kill_on_drop`: a dropped `docker exec` stream does not kill
|
||||
// the process, and neither does losing our handle to it.
|
||||
tokio::process::Command::from(cmd).spawn().map_err(|e| {
|
||||
// ENOENT here means a relative or explicit path that was not
|
||||
// there (PATH lookups were checked already): 127 like a shell.
|
||||
// Anything else — EACCES, ENOEXEC, a failed setuid — is 126,
|
||||
// "found but could not run".
|
||||
let code = if e.kind() == io::ErrorKind::NotFound { 127 } else { 126 };
|
||||
PrepareError::cannot_start(
|
||||
code,
|
||||
format!(
|
||||
"OCI runtime exec failed: exec failed: unable to start container process: exec: \"{}\": {e}: unknown",
|
||||
self.argv[0]
|
||||
),
|
||||
)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// 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 {
|
||||
match (status.code(), status.signal()) {
|
||||
(Some(c), _) => c,
|
||||
(None, Some(s)) => 128 + s,
|
||||
(None, None) => 255,
|
||||
}
|
||||
}
|
||||
|
||||
/// Signal a whole exec: its pid is its process-group id (it called `setsid`).
|
||||
pub fn signal_group(pid: Option<u32>, sig: libc::c_int) {
|
||||
if let Some(pid) = pid.filter(|&p| p > 0) {
|
||||
unsafe {
|
||||
libc::kill(-(pid as libc::pid_t), sig);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
//! `GET /v1/tunnel?port=`: a WebSocket piped to `127.0.0.1:<port>` inside the
|
||||
//! container — what the auth bridge and browser view do with `socat` through
|
||||
//! the Docker API today.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use axum::extract::rejection::QueryRejection;
|
||||
use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade};
|
||||
use axum::extract::{Query, State};
|
||||
use axum::http::StatusCode;
|
||||
use axum::response::Response;
|
||||
use serde::Deserialize;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpStream;
|
||||
|
||||
use crate::{error, AppState, CountGuard};
|
||||
|
||||
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct TunnelQuery {
|
||||
port: u16,
|
||||
}
|
||||
|
||||
pub async fn upgrade(
|
||||
State(st): State<Arc<AppState>>,
|
||||
q: Result<Query<TunnelQuery>, QueryRejection>,
|
||||
ws: WebSocketUpgrade,
|
||||
) -> Response {
|
||||
let port = match q {
|
||||
Ok(Query(q)) if q.port != 0 => q.port,
|
||||
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))
|
||||
}
|
||||
|
||||
async fn session(mut socket: WebSocket, st: Arc<AppState>, port: u16) {
|
||||
// Loopback only, by construction: the tunnel reaches in-container
|
||||
// listeners, never the network the container can see.
|
||||
let tcp = match tokio::time::timeout(CONNECT_TIMEOUT, TcpStream::connect(("127.0.0.1", port))).await {
|
||||
Ok(Ok(s)) => s,
|
||||
Ok(Err(e)) => return fail(socket, format!("cannot connect to 127.0.0.1:{port}: {e}")).await,
|
||||
Err(_) => return fail(socket, format!("timed out connecting to 127.0.0.1:{port}")).await,
|
||||
};
|
||||
let _ = tcp.set_nodelay(true);
|
||||
let _active = CountGuard::tunnels(&st);
|
||||
let (mut rd, mut wr) = tcp.into_split();
|
||||
let mut buf = vec![0u8; 32 * 1024];
|
||||
// When either side ends, the other is closed (spec): TCP EOF sends a
|
||||
// WebSocket close; a WebSocket close drops the TCP connection.
|
||||
loop {
|
||||
tokio::select! {
|
||||
r = rd.read(&mut buf) => match r {
|
||||
Ok(n) if n > 0 => {
|
||||
st.activity.touch();
|
||||
if socket.send(Message::Binary(buf[..n].to_vec().into())).await.is_err() {
|
||||
return;
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
let _ = socket.send(Message::Close(None)).await;
|
||||
return;
|
||||
}
|
||||
},
|
||||
m = socket.recv() => match m {
|
||||
Some(Ok(Message::Binary(d))) => {
|
||||
st.activity.touch();
|
||||
if wr.write_all(&d).await.is_err() {
|
||||
let _ = socket.send(Message::Close(None)).await;
|
||||
return;
|
||||
}
|
||||
}
|
||||
Some(Ok(Message::Text(_) | Message::Ping(_) | Message::Pong(_))) => {}
|
||||
Some(Ok(Message::Close(_))) | Some(Err(_)) | None => return,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn fail(mut socket: WebSocket, msg: String) {
|
||||
let _ = socket
|
||||
.send(Message::Text(serde_json::json!({ "type": "error", "error": msg }).to_string().into()))
|
||||
.await;
|
||||
let _ = socket.send(Message::Close(None)).await;
|
||||
}
|
||||
@@ -0,0 +1,131 @@
|
||||
//! User resolution through `/etc/passwd` and `/etc/group`.
|
||||
//!
|
||||
//! The static musl build has no NSS, so `getpwnam_r` and `getgrouplist` read
|
||||
//! the container's files directly — which is exactly the database `docker exec
|
||||
//! --user` resolves against (runc parses `/etc/passwd` itself), so the two
|
||||
//! cannot disagree about who a name is.
|
||||
|
||||
use std::ffi::{CStr, CString};
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct UserInfo {
|
||||
pub name: String,
|
||||
pub uid: libc::uid_t,
|
||||
pub gid: libc::gid_t,
|
||||
pub home: String,
|
||||
/// Supplementary groups, primary included — what `initgroups` would set.
|
||||
/// Resolved here, in the parent, because `pre_exec` runs between `fork`
|
||||
/// and `exec` in a multi-threaded process where reading files and
|
||||
/// allocating are not safe; the child only calls `setgroups` on this list.
|
||||
pub groups: Vec<libc::gid_t>,
|
||||
}
|
||||
|
||||
/// Resolve a user name (or, as Docker also allows, a numeric uid).
|
||||
///
|
||||
/// A numeric uid with no passwd entry still resolves, with gid 0 and HOME `/`,
|
||||
/// because that is what runc does for `--user 1234` in an image without that
|
||||
/// user.
|
||||
pub fn lookup(name: &str) -> Option<UserInfo> {
|
||||
let cname = CString::new(name).ok()?;
|
||||
if let Some(u) = getpw(|pwd, buf, len, res| unsafe {
|
||||
libc::getpwnam_r(cname.as_ptr(), pwd, buf, len, res)
|
||||
}) {
|
||||
return Some(u);
|
||||
}
|
||||
let uid: libc::uid_t = name.parse().ok()?;
|
||||
if let Some(u) = getpw(|pwd, buf, len, res| unsafe { libc::getpwuid_r(uid, pwd, buf, len, res) }) {
|
||||
return Some(u);
|
||||
}
|
||||
Some(UserInfo {
|
||||
name: name.to_string(),
|
||||
uid,
|
||||
gid: 0,
|
||||
home: "/".to_string(),
|
||||
groups: vec![0],
|
||||
})
|
||||
}
|
||||
|
||||
/// The user the agent itself runs as.
|
||||
pub fn current() -> Option<UserInfo> {
|
||||
let uid = unsafe { libc::geteuid() };
|
||||
getpw(|pwd, buf, len, res| unsafe { libc::getpwuid_r(uid, pwd, buf, len, res) })
|
||||
}
|
||||
|
||||
fn getpw(
|
||||
call: impl Fn(*mut libc::passwd, *mut libc::c_char, libc::size_t, *mut *mut libc::passwd) -> libc::c_int,
|
||||
) -> Option<UserInfo> {
|
||||
let mut size = 1024usize;
|
||||
loop {
|
||||
let mut pwd: libc::passwd = unsafe { std::mem::zeroed() };
|
||||
let mut buf = vec![0 as libc::c_char; size];
|
||||
let mut res: *mut libc::passwd = std::ptr::null_mut();
|
||||
let rc = call(&mut pwd, buf.as_mut_ptr(), buf.len(), &mut res);
|
||||
if rc == libc::ERANGE && size < (1 << 20) {
|
||||
size *= 4;
|
||||
continue;
|
||||
}
|
||||
if rc != 0 || res.is_null() {
|
||||
return None;
|
||||
}
|
||||
let name = unsafe { CStr::from_ptr(pwd.pw_name) }.to_string_lossy().into_owned();
|
||||
let home = unsafe { CStr::from_ptr(pwd.pw_dir) }.to_string_lossy().into_owned();
|
||||
let groups = group_list(&name, pwd.pw_gid);
|
||||
return Some(UserInfo {
|
||||
name,
|
||||
uid: pwd.pw_uid,
|
||||
gid: pwd.pw_gid,
|
||||
home,
|
||||
groups,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
fn group_list(name: &str, gid: libc::gid_t) -> Vec<libc::gid_t> {
|
||||
let Ok(cname) = CString::new(name) else {
|
||||
return vec![gid];
|
||||
};
|
||||
let mut n: libc::c_int = 32;
|
||||
loop {
|
||||
let mut groups = vec![0 as libc::gid_t; n as usize];
|
||||
let mut count = n;
|
||||
let rc = unsafe { libc::getgrouplist(cname.as_ptr(), gid, groups.as_mut_ptr(), &mut count) };
|
||||
if rc >= 0 {
|
||||
groups.truncate(count.max(0) as usize);
|
||||
return groups;
|
||||
}
|
||||
// -1 means "too small" and `count` now holds the size needed (glibc
|
||||
// and musl both). Guard against a libc that does not update it.
|
||||
if count <= n {
|
||||
if n >= 65536 {
|
||||
return vec![gid];
|
||||
}
|
||||
n *= 2;
|
||||
} else {
|
||||
n = count;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn resolves_the_current_user_by_name_and_uid() {
|
||||
let me = current().expect("current user has a passwd entry");
|
||||
let by_name = lookup(&me.name).unwrap();
|
||||
assert_eq!(by_name.uid, me.uid);
|
||||
assert!(by_name.groups.contains(&me.gid));
|
||||
let by_uid = lookup(&me.uid.to_string()).unwrap();
|
||||
assert_eq!(by_uid.name, me.name);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_names_do_not_resolve() {
|
||||
assert!(lookup("no-such-user-triple-c").is_none());
|
||||
assert!(lookup("bad\0name").is_none());
|
||||
// A bare uid nobody owns still resolves, as runc does.
|
||||
let u = lookup("4000123").unwrap();
|
||||
assert_eq!((u.uid, u.gid, u.home.as_str()), (4000123, 0, "/"));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,892 @@
|
||||
//! End-to-end tests of the wire protocol
|
||||
//! (`docs/superpowers/specs/2026-11-13-triple-c-agent-protocol.md`): the real
|
||||
//! server, in-process on `127.0.0.1:0`, driven over real HTTP and WebSocket,
|
||||
//! running as the current (non-root) user.
|
||||
|
||||
use std::ffi::OsString;
|
||||
use std::net::SocketAddr;
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use base64::Engine;
|
||||
use futures_util::{SinkExt, StreamExt};
|
||||
use serde_json::{json, Value};
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
|
||||
use tokio_tungstenite::tungstenite::Message;
|
||||
use triple_c_agent::{serve, Config};
|
||||
|
||||
const TOKEN: &str = "test-token-0123456789abcdef0123456789abcdef";
|
||||
|
||||
type Ws = tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
|
||||
|
||||
struct Agent {
|
||||
addr: SocketAddr,
|
||||
http: reqwest::Client,
|
||||
/// The agent's startup cwd — where `working_dir: null` runs.
|
||||
cwd: tempfile::TempDir,
|
||||
/// Scratch space for the test's own files.
|
||||
tmp: tempfile::TempDir,
|
||||
user: String,
|
||||
}
|
||||
|
||||
fn current_user() -> String {
|
||||
let out = std::process::Command::new("id").arg("-un").output().unwrap();
|
||||
String::from_utf8(out.stdout).unwrap().trim().to_string()
|
||||
}
|
||||
|
||||
fn passwd_home(user: &str) -> String {
|
||||
std::fs::read_to_string("/etc/passwd")
|
||||
.unwrap()
|
||||
.lines()
|
||||
.map(|l| l.split(':').collect::<Vec<_>>())
|
||||
.find(|f| f[0] == user)
|
||||
.map(|f| f[5].to_string())
|
||||
.expect("current user is in /etc/passwd")
|
||||
}
|
||||
|
||||
fn is_root() -> bool {
|
||||
unsafe { libc::geteuid() == 0 }
|
||||
}
|
||||
|
||||
async fn start() -> Agent {
|
||||
start_with(|_| {}).await
|
||||
}
|
||||
|
||||
async fn start_with(tweak: impl FnOnce(&mut Config)) -> Agent {
|
||||
let cwd = tempfile::tempdir().unwrap();
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let mut 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()),
|
||||
("FOO".into(), "base".into()),
|
||||
("BASE_ONLY".into(), "base".into()),
|
||||
("GONE".into(), "present".into()),
|
||||
// The agent runs as root with HOME=/root in production; passwd
|
||||
// must win over the base env for the exec's user.
|
||||
(OsString::from("HOME"), OsString::from("/agent-home-must-not-leak")),
|
||||
],
|
||||
max_file_bytes: 1024,
|
||||
};
|
||||
tweak(&mut config);
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
tokio::spawn(serve(listener, config));
|
||||
Agent { addr, http: reqwest::Client::new(), cwd, tmp, user: current_user() }
|
||||
}
|
||||
|
||||
impl Agent {
|
||||
fn url(&self, path: &str) -> String {
|
||||
format!("http://{}{}", self.addr, path)
|
||||
}
|
||||
|
||||
async fn get(&self, path: &str) -> (u16, Value) {
|
||||
let r = self.http.get(self.url(path)).bearer_auth(TOKEN).send().await.unwrap();
|
||||
let status = r.status().as_u16();
|
||||
(status, r.json().await.unwrap_or(Value::Null))
|
||||
}
|
||||
|
||||
async fn post(&self, path: &str, body: Value) -> (u16, Value) {
|
||||
let r = self.http.post(self.url(path)).bearer_auth(TOKEN).json(&body).send().await.unwrap();
|
||||
let status = r.status().as_u16();
|
||||
(status, r.json().await.unwrap_or(Value::Null))
|
||||
}
|
||||
|
||||
/// One-shot as the current user, defaulting the rest of the spec.
|
||||
async fn exec(&self, mut body: Value) -> (u16, Value) {
|
||||
if body.get("user").is_none() {
|
||||
body["user"] = json!(self.user);
|
||||
}
|
||||
self.post("/v1/exec", body).await
|
||||
}
|
||||
|
||||
async fn ws(&self, path: &str) -> Ws {
|
||||
let mut req = format!("ws://{}{}", self.addr, path).into_client_request().unwrap();
|
||||
req.headers_mut()
|
||||
.insert("Authorization", format!("Bearer {TOKEN}").parse().unwrap());
|
||||
tokio_tungstenite::connect_async(req).await.unwrap().0
|
||||
}
|
||||
|
||||
/// Open an attach and return the socket and the exec id.
|
||||
async fn attach(&self, mut spec: Value) -> (Ws, String) {
|
||||
if spec.get("user").is_none() {
|
||||
spec["user"] = json!(self.user);
|
||||
}
|
||||
let mut ws = self.ws("/v1/attach").await;
|
||||
ws.send(Message::Text(spec.to_string().into())).await.unwrap();
|
||||
let started = next_text(&mut ws).await;
|
||||
assert_eq!(started["type"], "started", "{started}");
|
||||
(ws, started["exec_id"].as_str().unwrap().to_string())
|
||||
}
|
||||
|
||||
async fn wait_finished(&self, id: &str) -> Value {
|
||||
let deadline = Instant::now() + Duration::from_secs(10);
|
||||
loop {
|
||||
let (s, v) = self.get(&format!("/v1/exec/{id}")).await;
|
||||
assert_eq!(s, 200, "{v}");
|
||||
if v["running"] == false {
|
||||
return v;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "exec {id} never finished");
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn next_text(ws: &mut Ws) -> Value {
|
||||
loop {
|
||||
match tokio::time::timeout(Duration::from_secs(10), ws.next()).await.expect("timed out") {
|
||||
Some(Ok(Message::Text(t))) => return serde_json::from_str(t.as_str()).unwrap(),
|
||||
Some(Ok(Message::Ping(_) | Message::Pong(_))) => continue,
|
||||
other => panic!("expected a text frame, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Everything until the exit frame: (stdout bytes, stderr bytes, exit code).
|
||||
async fn collect_until_exit(ws: &mut Ws) -> (Vec<u8>, Vec<u8>, i64) {
|
||||
let (mut out, mut err) = (Vec::new(), Vec::new());
|
||||
loop {
|
||||
match tokio::time::timeout(Duration::from_secs(10), ws.next()).await.expect("timed out") {
|
||||
Some(Ok(Message::Binary(b))) => match b[0] {
|
||||
1 => out.extend_from_slice(&b[1..]),
|
||||
2 => err.extend_from_slice(&b[1..]),
|
||||
c => panic!("unknown channel {c}"),
|
||||
},
|
||||
Some(Ok(Message::Text(t))) => {
|
||||
let v: Value = serde_json::from_str(t.as_str()).unwrap();
|
||||
assert_eq!(v["type"], "exit", "{v}");
|
||||
return (out, err, v["code"].as_i64().unwrap());
|
||||
}
|
||||
other => panic!("unexpected {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Read stdout frames until `needle` has appeared.
|
||||
async fn read_until(ws: &mut Ws, acc: &mut Vec<u8>, needle: &str) {
|
||||
while !String::from_utf8_lossy(acc).contains(needle) {
|
||||
match tokio::time::timeout(Duration::from_secs(10), ws.next()).await.expect("timed out") {
|
||||
Some(Ok(Message::Binary(b))) => {
|
||||
assert_eq!(b[0], 1, "tty output must all be channel 1");
|
||||
acc.extend_from_slice(&b[1..]);
|
||||
}
|
||||
other => panic!("unexpected {other:?} waiting for {needle:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// (combined, stdout, stderr, [(stream, text)]) — what the client rebuilds.
|
||||
fn decode(v: &Value) -> (String, String, String, Vec<(String, String)>) {
|
||||
let b64 = base64::engine::general_purpose::STANDARD;
|
||||
let (mut all, mut out, mut err, mut list) = (String::new(), String::new(), String::new(), vec![]);
|
||||
for c in v["chunks"].as_array().unwrap() {
|
||||
let s = c["s"].as_str().unwrap().to_string();
|
||||
let d = String::from_utf8(b64.decode(c["d"].as_str().unwrap()).unwrap()).unwrap();
|
||||
all.push_str(&d);
|
||||
if s == "out" { out.push_str(&d) } else { err.push_str(&d) }
|
||||
list.push((s, d));
|
||||
}
|
||||
(all, out, err, list)
|
||||
}
|
||||
|
||||
fn sh(script: &str) -> Value {
|
||||
json!(["sh", "-c", script])
|
||||
}
|
||||
|
||||
fn process_gone(pid: i32) -> bool {
|
||||
match std::fs::read_to_string(format!("/proc/{pid}/stat")) {
|
||||
Err(_) => true,
|
||||
// A zombie is dead; it is only waiting for whoever reaps orphans here.
|
||||
Ok(s) => s.rsplit(')').next().unwrap_or("").trim_start().starts_with('Z'),
|
||||
}
|
||||
}
|
||||
|
||||
// ── auth / health / errors ─────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_or_wrong_token_is_401_with_an_empty_body() {
|
||||
let a = start().await;
|
||||
for auth in [None, Some("Bearer nope"), Some(&*format!("Basic {TOKEN}")), Some(&*format!("Bearer {TOKEN}x"))] {
|
||||
let mut req = a.http.get(a.url("/v1/health"));
|
||||
if let Some(h) = auth {
|
||||
req = req.header("Authorization", h);
|
||||
}
|
||||
let r = req.send().await.unwrap();
|
||||
assert_eq!(r.status().as_u16(), 401, "{auth:?}");
|
||||
assert!(r.bytes().await.unwrap().is_empty());
|
||||
}
|
||||
// Unknown endpoints are behind the token too.
|
||||
let r = a.http.get(a.url("/v1/nope")).send().await.unwrap();
|
||||
assert_eq!(r.status().as_u16(), 401);
|
||||
// And so is a WebSocket upgrade.
|
||||
let req = format!("ws://{}/v1/attach", a.addr).into_client_request().unwrap();
|
||||
assert!(tokio_tungstenite::connect_async(req).await.is_err());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn health_reports_agent_version_and_protocol() {
|
||||
let a = start().await;
|
||||
let (s, v) = a.get("/v1/health").await;
|
||||
assert_eq!(s, 200);
|
||||
assert_eq!(v["agent"], "triple-c-agent");
|
||||
assert_eq!(v["version"], env!("CARGO_PKG_VERSION"));
|
||||
assert_eq!(v["protocol"], 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn bad_requests_are_400_with_a_json_error() {
|
||||
let a = start().await;
|
||||
let r = a
|
||||
.http
|
||||
.post(a.url("/v1/exec"))
|
||||
.bearer_auth(TOKEN)
|
||||
.header("content-type", "application/json")
|
||||
.body("{not json")
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(r.status().as_u16(), 400);
|
||||
assert_eq!(r.headers()["content-type"], "application/json");
|
||||
assert!(r.json::<Value>().await.unwrap()["error"].is_string());
|
||||
|
||||
let (s, v) = a.exec(json!({ "cmd": [] })).await;
|
||||
assert_eq!(s, 400);
|
||||
assert!(v["error"].as_str().unwrap().contains("non-empty"));
|
||||
|
||||
// Valid JSON without a Content-Type header is still accepted: the spec
|
||||
// says the body is JSON and says nothing about the header.
|
||||
let r = a
|
||||
.http
|
||||
.post(a.url("/v1/exec"))
|
||||
.bearer_auth(TOKEN)
|
||||
.body(json!({ "cmd": ["true"], "user": a.user }).to_string())
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(r.status().as_u16(), 200);
|
||||
}
|
||||
|
||||
// ── one-shot ───────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn oneshot_chunks_keep_arrival_order_across_streams() {
|
||||
let a = start().await;
|
||||
let (s, v) = a
|
||||
.exec(json!({ "cmd": sh("printf a; sleep 0.2; printf b >&2; sleep 0.2; printf c; sleep 0.2; printf d >&2") }))
|
||||
.await;
|
||||
assert_eq!(s, 200, "{v}");
|
||||
let (all, out, err, list) = decode(&v);
|
||||
assert_eq!(all, "abcd");
|
||||
assert_eq!(out, "ac");
|
||||
assert_eq!(err, "bd");
|
||||
let streams: Vec<_> = list.iter().map(|(s, _)| s.as_str()).collect();
|
||||
assert_eq!(streams, ["out", "err", "out", "err"]);
|
||||
assert_eq!(v["exit_code"], 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn oneshot_exit_codes_match_docker() {
|
||||
let a = start().await;
|
||||
let (_, v) = a.exec(json!({ "cmd": sh("exit 3") })).await;
|
||||
assert_eq!(v["exit_code"], 3);
|
||||
// Killed by a signal: 128 + n, as Docker and a shell report it.
|
||||
let (_, v) = a.exec(json!({ "cmd": sh("kill -KILL $$") })).await;
|
||||
assert_eq!(v["exit_code"], 137);
|
||||
let (_, v) = a.exec(json!({ "cmd": sh("kill -TERM $$") })).await;
|
||||
assert_eq!(v["exit_code"], 143);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_command_that_cannot_start_is_a_finished_exec_not_an_api_error() {
|
||||
let a = start().await;
|
||||
let (s, v) = a.exec(json!({ "cmd": ["definitely-not-a-binary-xyz"] })).await;
|
||||
assert_eq!(s, 200);
|
||||
assert_eq!(v["exit_code"], 127);
|
||||
let (_, _, err, list) = decode(&v);
|
||||
assert_eq!(list.len(), 1);
|
||||
assert!(err.contains("executable file not found in $PATH"), "{err}");
|
||||
|
||||
// Present but not executable: 126.
|
||||
let f = a.tmp.path().join("not-exec");
|
||||
std::fs::write(&f, "#!/bin/sh\n").unwrap();
|
||||
std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o644)).unwrap();
|
||||
let (_, v) = a.exec(json!({ "cmd": [f] })).await;
|
||||
assert_eq!(v["exit_code"], 126);
|
||||
|
||||
// An explicit path that does not exist: 127.
|
||||
let (_, v) = a.exec(json!({ "cmd": [a.tmp.path().join("missing")] })).await;
|
||||
assert_eq!(v["exit_code"], 127);
|
||||
|
||||
// No such user: 126 with runc's text.
|
||||
let (s, v) = a.exec(json!({ "cmd": ["true"], "user": "no-such-user-triple-c" })).await;
|
||||
assert_eq!(s, 200);
|
||||
assert_eq!(v["exit_code"], 126);
|
||||
assert!(decode(&v).2.contains("unable to find user no-such-user-triple-c"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_non_root_agent_refuses_other_users_with_403() {
|
||||
if is_root() {
|
||||
eprintln!("skipped: running as root");
|
||||
return;
|
||||
}
|
||||
let a = start().await;
|
||||
let (s, v) = a.exec(json!({ "cmd": ["true"], "user": "root" })).await;
|
||||
assert_eq!(s, 403, "{v}");
|
||||
assert!(v["error"].is_string());
|
||||
// The agent's own user, by name and by uid, works.
|
||||
let (s, _) = a.exec(json!({ "cmd": ["true"] })).await;
|
||||
assert_eq!(s, 200);
|
||||
let uid = unsafe { libc::geteuid() }.to_string();
|
||||
let (s, v) = a.exec(json!({ "cmd": ["id", "-u"], "user": uid })).await;
|
||||
assert_eq!(s, 200);
|
||||
assert_eq!(decode(&v).1.trim(), uid);
|
||||
// Omitting the user means the agent's own.
|
||||
let (s, v) = a.post("/v1/exec", json!({ "cmd": ["id", "-un"] })).await;
|
||||
assert_eq!(s, 200);
|
||||
assert_eq!(decode(&v).1.trim(), a.user);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn env_merges_over_the_base_and_sets_home_user_logname_from_passwd() {
|
||||
let a = start().await;
|
||||
let script = r#"printf '%s|%s|%s|%s|%s|%s|%s' "$FOO" "$BAR" "$BASE_ONLY" "$HOME" "$USER" "$LOGNAME" "${GONE-unset}""#;
|
||||
let (_, v) = a
|
||||
.exec(json!({ "cmd": sh(script), "env": ["FOO=spec", "BAR=1", "BAR=2", "GONE"] }))
|
||||
.await;
|
||||
let home = passwd_home(&a.user);
|
||||
assert_eq!(decode(&v).1, format!("spec|2|base|{home}|{u}|{u}|unset", u = a.user));
|
||||
|
||||
// The spec's own HOME/USER/LOGNAME win over passwd.
|
||||
let (_, v) = a
|
||||
.exec(json!({ "cmd": sh(r#"printf '%s|%s|%s' "$HOME" "$USER" "$LOGNAME""#), "env": ["HOME=/custom", "USER=u", "LOGNAME=l"] }))
|
||||
.await;
|
||||
assert_eq!(decode(&v).1, "/custom|u|l");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cmd0_resolves_through_the_merged_path() {
|
||||
let a = start().await;
|
||||
let bin = a.tmp.path().join("bin");
|
||||
std::fs::create_dir(&bin).unwrap();
|
||||
let tool = bin.join("mytool");
|
||||
std::fs::write(&tool, "#!/bin/sh\necho \"mytool ran as $0\"\n").unwrap();
|
||||
std::fs::set_permissions(&tool, std::fs::Permissions::from_mode(0o755)).unwrap();
|
||||
|
||||
let (_, v) = a.exec(json!({ "cmd": ["mytool"] })).await;
|
||||
assert_eq!(v["exit_code"], 127, "not on the base PATH");
|
||||
let path = format!("PATH={}:/usr/bin:/bin", bin.display());
|
||||
let (_, v) = a.exec(json!({ "cmd": ["mytool"], "env": [path] })).await;
|
||||
assert_eq!(v["exit_code"], 0, "{v}");
|
||||
assert!(decode(&v).1.starts_with("mytool ran as"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn working_dir_null_is_the_agent_cwd_and_a_set_one_is_used() {
|
||||
let a = start().await;
|
||||
let (_, v) = a.exec(json!({ "cmd": ["pwd"], "working_dir": null })).await;
|
||||
assert_eq!(Path::new(decode(&v).1.trim()), a.cwd.path().canonicalize().unwrap());
|
||||
let (_, v) = a.exec(json!({ "cmd": ["pwd"], "working_dir": a.tmp.path() })).await;
|
||||
assert_eq!(Path::new(decode(&v).1.trim()), a.tmp.path().canonicalize().unwrap());
|
||||
let (s, v) = a.exec(json!({ "cmd": ["pwd"], "working_dir": "/no/such/dir" })).await;
|
||||
assert_eq!(s, 200);
|
||||
assert_eq!(v["exit_code"], 126);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn output_over_the_limit_is_413_and_kills_the_process_group() {
|
||||
let a = start().await;
|
||||
let pidfile = a.tmp.path().join("bg.pid");
|
||||
// A backgrounded sleeper in the same process group, then a flood.
|
||||
let script = format!("sleep 30 & echo $! > {}; exec head -c 100000 /dev/zero", pidfile.display());
|
||||
let (s, v) = a.exec(json!({ "cmd": sh(&script), "limit": 1000 })).await;
|
||||
assert_eq!(s, 413, "{v}");
|
||||
assert_eq!(
|
||||
v["error"],
|
||||
"OUTPUT_LIMIT: Command output exceeded 1000 bytes and was abandoned"
|
||||
);
|
||||
assert!(v["error"].as_str().unwrap().starts_with("OUTPUT_LIMIT:"));
|
||||
let pid: i32 = std::fs::read_to_string(&pidfile).unwrap().trim().parse().unwrap();
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
while !process_gone(pid) {
|
||||
assert!(Instant::now() < deadline, "background sleeper {pid} survived the limit kill");
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
|
||||
// Exactly at the limit is still a success, as with `push_capped`.
|
||||
let (s, v) = a.exec(json!({ "cmd": sh("printf 12345"), "limit": 5 })).await;
|
||||
assert_eq!(s, 200);
|
||||
assert_eq!(decode(&v).0, "12345");
|
||||
let (s, _) = a.exec(json!({ "cmd": sh("printf 123456"), "limit": 5 })).await;
|
||||
assert_eq!(s, 413);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn oneshot_stdin_is_dev_null() {
|
||||
let a = start().await;
|
||||
let (_, v) = a.exec(json!({ "cmd": ["cat"] })).await;
|
||||
assert_eq!(v["exit_code"], 0);
|
||||
assert_eq!(decode(&v).0, "");
|
||||
}
|
||||
|
||||
// ── detached / status / resize ─────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn detached_exec_can_be_polled_to_its_exit_code() {
|
||||
let a = start().await;
|
||||
let marker = a.tmp.path().join("ran");
|
||||
let script = format!("sleep 0.3; touch {}; exit 7", marker.display());
|
||||
let (s, v) = a.post("/v1/exec/detached", json!({ "cmd": sh(&script), "user": a.user })).await;
|
||||
assert_eq!(s, 200);
|
||||
let id = v["exec_id"].as_str().unwrap().to_string();
|
||||
let (_, v) = a.get(&format!("/v1/exec/{id}")).await;
|
||||
assert_eq!(v, json!({ "running": true, "exit_code": null }));
|
||||
let v = a.wait_finished(&id).await;
|
||||
assert_eq!(v, json!({ "running": false, "exit_code": 7 }));
|
||||
assert!(marker.exists());
|
||||
|
||||
// Not a tty: resize is a conflict.
|
||||
let (s, _) = a.post(&format!("/v1/exec/{id}/resize"), json!({ "cols": 80, "rows": 24 })).await;
|
||||
assert_eq!(s, 409);
|
||||
|
||||
// A missing binary still yields an id that reports 127.
|
||||
let (_, v) = a.post("/v1/exec/detached", json!({ "cmd": ["no-such-bin-xyz"], "user": a.user })).await;
|
||||
let v = a.wait_finished(v["exec_id"].as_str().unwrap()).await;
|
||||
assert_eq!(v["exit_code"], 127);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn unknown_exec_ids_are_404() {
|
||||
let a = start().await;
|
||||
let (s, _) = a.get("/v1/exec/0000").await;
|
||||
assert_eq!(s, 404);
|
||||
let (s, _) = a.post("/v1/exec/0000/resize", json!({ "cols": 1, "rows": 1 })).await;
|
||||
assert_eq!(s, 404);
|
||||
}
|
||||
|
||||
// ── attach ─────────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn attach_pipes_stdin_and_close_stdin_gives_eof_and_an_exit_frame() {
|
||||
let a = start().await;
|
||||
let (mut ws, id) = a.attach(json!({ "cmd": ["cat"], "attach_stdin": true })).await;
|
||||
ws.send(Message::Binary(b"hello\n".to_vec().into())).await.unwrap();
|
||||
let mut acc = Vec::new();
|
||||
match ws.next().await.unwrap().unwrap() {
|
||||
Message::Binary(b) => {
|
||||
assert_eq!(b[0], 1);
|
||||
acc.extend_from_slice(&b[1..]);
|
||||
}
|
||||
m => panic!("{m:?}"),
|
||||
}
|
||||
assert_eq!(acc, b"hello\n");
|
||||
ws.send(Message::Text(json!({ "type": "close_stdin" }).to_string().into())).await.unwrap();
|
||||
let (out, err, code) = collect_until_exit(&mut ws).await;
|
||||
assert!(out.is_empty() && err.is_empty());
|
||||
assert_eq!(code, 0);
|
||||
assert_eq!(a.wait_finished(&id).await, json!({ "running": false, "exit_code": 0 }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn attach_without_tty_keeps_stderr_on_channel_2() {
|
||||
let a = start().await;
|
||||
let (mut ws, id) = a
|
||||
.attach(json!({ "cmd": sh("echo out; echo oops >&2; exit 4"), "attach_stdin": false }))
|
||||
.await;
|
||||
let (out, err, code) = collect_until_exit(&mut ws).await;
|
||||
assert_eq!(out, b"out\n");
|
||||
assert_eq!(err, b"oops\n");
|
||||
assert_eq!(code, 4);
|
||||
assert_eq!(a.wait_finished(&id).await["exit_code"], 4);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn attach_without_stdin_gives_the_process_dev_null() {
|
||||
let a = start().await;
|
||||
let (mut ws, _) = a.attach(json!({ "cmd": ["cat"], "attach_stdin": false })).await;
|
||||
// Ignored: stdin is /dev/null, so cat exits on its own.
|
||||
let _ = ws.send(Message::Binary(b"ignored".to_vec().into())).await;
|
||||
let (out, _, code) = collect_until_exit(&mut ws).await;
|
||||
assert!(out.is_empty());
|
||||
assert_eq!(code, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn attach_tty_puts_everything_on_channel_1_and_resize_reaches_stty() {
|
||||
let a = start().await;
|
||||
let (mut ws, id) = a
|
||||
.attach(json!({
|
||||
"cmd": sh("stty size; echo to-stderr >&2; read x; stty size; echo \"term=$TERM\""),
|
||||
"tty": true, "attach_stdin": true, "cols": 80, "rows": 24,
|
||||
}))
|
||||
.await;
|
||||
let mut acc = Vec::new();
|
||||
read_until(&mut ws, &mut acc, "24 80").await;
|
||||
read_until(&mut ws, &mut acc, "to-stderr").await;
|
||||
let (s, _) = a.post(&format!("/v1/exec/{id}/resize"), json!({ "cols": 100, "rows": 40 })).await;
|
||||
assert_eq!(s, 204);
|
||||
ws.send(Message::Binary(b"\n".to_vec().into())).await.unwrap();
|
||||
let (out, err, code) = collect_until_exit(&mut ws).await;
|
||||
acc.extend_from_slice(&out);
|
||||
let text = String::from_utf8_lossy(&acc);
|
||||
assert!(err.is_empty(), "a tty has no channel 2");
|
||||
assert!(text.contains("40 100"), "{text}");
|
||||
// Docker's default TERM for a tty exec when the env has none.
|
||||
assert!(text.contains("term=xterm"), "{text}");
|
||||
assert_eq!(code, 0);
|
||||
// Finished: no longer resizable.
|
||||
let (s, _) = a.post(&format!("/v1/exec/{id}/resize"), json!({ "cols": 1, "rows": 1 })).await;
|
||||
assert_eq!(s, 409);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn attach_reports_a_missing_binary_as_started_stderr_and_127() {
|
||||
let a = start().await;
|
||||
let (mut ws, id) = a.attach(json!({ "cmd": ["no-such-bin-xyz"], "attach_stdin": false })).await;
|
||||
let (out, err, code) = collect_until_exit(&mut ws).await;
|
||||
assert!(out.is_empty());
|
||||
assert!(String::from_utf8_lossy(&err).contains("executable file not found"));
|
||||
assert_eq!(code, 127);
|
||||
assert_eq!(a.wait_finished(&id).await["exit_code"], 127);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn attach_refusals_are_an_error_frame() {
|
||||
let a = start().await;
|
||||
let mut ws = a.ws("/v1/attach").await;
|
||||
ws.send(Message::Text(json!({ "cmd": [] }).to_string().into())).await.unwrap();
|
||||
let v = next_text(&mut ws).await;
|
||||
assert_eq!(v["type"], "error");
|
||||
if !is_root() {
|
||||
let mut ws = a.ws("/v1/attach").await;
|
||||
ws.send(Message::Text(json!({ "cmd": ["true"], "user": "root" }).to_string().into())).await.unwrap();
|
||||
assert_eq!(next_text(&mut ws).await["type"], "error");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_dropped_non_tty_attach_closes_stdin_but_does_not_kill() {
|
||||
let a = start().await;
|
||||
let out = a.tmp.path().join("got");
|
||||
// cat sees EOF when the client goes away, then the shell carries on.
|
||||
let script = format!("cat > {o}; sleep 0.3; echo after >> {o}", o = out.display());
|
||||
let (mut ws, id) = a.attach(json!({ "cmd": sh(&script), "attach_stdin": true })).await;
|
||||
ws.send(Message::Binary(b"abc".to_vec().into())).await.unwrap();
|
||||
tokio::time::sleep(Duration::from_millis(200)).await;
|
||||
drop(ws);
|
||||
let v = a.wait_finished(&id).await;
|
||||
assert_eq!(v["exit_code"], 0);
|
||||
assert_eq!(std::fs::read_to_string(&out).unwrap(), "abcafter\n");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_dropped_tty_attach_hangs_up_the_process_group() {
|
||||
let a = start().await;
|
||||
let (mut ws, id) = a
|
||||
.attach(json!({ "cmd": sh("echo ready; sleep 30"), "tty": true, "attach_stdin": true }))
|
||||
.await;
|
||||
let mut acc = Vec::new();
|
||||
read_until(&mut ws, &mut acc, "ready").await;
|
||||
drop(ws);
|
||||
let v = a.wait_finished(&id).await;
|
||||
// SIGHUP: 128 + 1.
|
||||
assert_eq!(v["exit_code"], 129);
|
||||
}
|
||||
|
||||
// ── files ──────────────────────────────────────────────────────────────────
|
||||
|
||||
async fn put(a: &Agent, query: &[(&str, String)], body: Vec<u8>) -> (u16, Value) {
|
||||
let r = a
|
||||
.http
|
||||
.put(a.url("/v1/files"))
|
||||
.query(query)
|
||||
.bearer_auth(TOKEN)
|
||||
.body(body)
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
let s = r.status().as_u16();
|
||||
(s, r.json().await.unwrap_or(Value::Null))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn put_file_writes_content_and_mode_and_replaces_by_unlinking() {
|
||||
let a = start().await;
|
||||
let dir = a.tmp.path().to_str().unwrap().to_string();
|
||||
let target = a.tmp.path().join("f.txt");
|
||||
// An existing read-only file with a second hard link: replacement must
|
||||
// unlink and create, leaving the other link's content alone.
|
||||
std::fs::write(&target, "old").unwrap();
|
||||
std::fs::hard_link(&target, a.tmp.path().join("other-link")).unwrap();
|
||||
std::fs::set_permissions(&target, std::fs::Permissions::from_mode(0o400)).unwrap();
|
||||
|
||||
let (s, v) = put(
|
||||
&a,
|
||||
&[("dir", dir.clone()), ("name", "f.txt".into()), ("mode", "640".into())],
|
||||
b"new content".to_vec(),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(s, 200, "{v}");
|
||||
assert_eq!(v["path"], format!("{dir}/f.txt"));
|
||||
assert_eq!(std::fs::read(&target).unwrap(), b"new content");
|
||||
assert_eq!(std::fs::metadata(&target).unwrap().permissions().mode() & 0o7777, 0o640);
|
||||
assert_eq!(std::fs::read(a.tmp.path().join("other-link")).unwrap(), b"old");
|
||||
|
||||
// mode is exact, not filtered through the umask.
|
||||
let (s, _) = put(&a, &[("dir", dir.clone()), ("name", "x".into()), ("mode", "777".into())], vec![]).await;
|
||||
assert_eq!(s, 200);
|
||||
assert_eq!(std::fs::metadata(a.tmp.path().join("x")).unwrap().permissions().mode() & 0o7777, 0o777);
|
||||
|
||||
// Own uid/gid are accepted when not root.
|
||||
let (uid, gid) = unsafe { (libc::geteuid(), libc::getegid()) };
|
||||
let (s, _) = put(
|
||||
&a,
|
||||
&[("dir", dir.clone()), ("name", "own".into()), ("uid", uid.to_string()), ("gid", gid.to_string())],
|
||||
b"x".to_vec(),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(s, 200);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn put_file_refusals() {
|
||||
let a = start().await;
|
||||
let dir = a.tmp.path().to_str().unwrap().to_string();
|
||||
for name in ["a/b", "..", ".", ""] {
|
||||
let (s, _) = put(&a, &[("dir", dir.clone()), ("name", name.into())], b"x".to_vec()).await;
|
||||
assert_eq!(s, 400, "name {name:?}");
|
||||
}
|
||||
let (s, _) = put(&a, &[("dir", "relative".into()), ("name", "n".into())], vec![]).await;
|
||||
assert_eq!(s, 400);
|
||||
// The directory is not created.
|
||||
let missing = a.tmp.path().join("missing");
|
||||
let (s, _) = put(&a, &[("dir", missing.to_str().unwrap().into()), ("name", "n".into())], vec![]).await;
|
||||
assert_eq!(s, 404);
|
||||
assert!(!missing.exists());
|
||||
// Over the ceiling (1 KiB in tests, MAX_DROP_BYTES in production): 413
|
||||
// and nothing left behind.
|
||||
let (s, _) = put(&a, &[("dir", dir.clone()), ("name", "big".into())], vec![0u8; 1025]).await;
|
||||
assert_eq!(s, 413);
|
||||
assert!(!a.tmp.path().join("big").exists());
|
||||
let (s, _) = put(&a, &[("dir", dir.clone()), ("name", "fits".into())], vec![0u8; 1024]).await;
|
||||
assert_eq!(s, 200);
|
||||
if !is_root() {
|
||||
let (s, _) = put(&a, &[("dir", dir.clone()), ("name", "o".into()), ("uid", "0".into())], vec![]).await;
|
||||
assert_eq!(s, 403);
|
||||
assert!(!a.tmp.path().join("o").exists());
|
||||
}
|
||||
}
|
||||
|
||||
// ── archive ────────────────────────────────────────────────────────────────
|
||||
|
||||
async fn archive(a: &Agent, path: &Path) -> (u16, Vec<u8>) {
|
||||
let r = a
|
||||
.http
|
||||
.get(a.url("/v1/archive"))
|
||||
.query(&[("path", path.to_str().unwrap())])
|
||||
.bearer_auth(TOKEN)
|
||||
.send()
|
||||
.await
|
||||
.unwrap();
|
||||
let s = r.status().as_u16();
|
||||
if s == 200 {
|
||||
assert_eq!(r.headers()["content-type"], "application/x-tar");
|
||||
}
|
||||
(s, r.bytes().await.unwrap().to_vec())
|
||||
}
|
||||
|
||||
fn entries(tar_bytes: &[u8]) -> Vec<(PathBuf, tar::EntryType, Vec<u8>)> {
|
||||
let mut ar = tar::Archive::new(tar_bytes);
|
||||
ar.entries()
|
||||
.unwrap()
|
||||
.map(|e| {
|
||||
let mut e = e.unwrap();
|
||||
let p = e.path().unwrap().into_owned();
|
||||
let t = e.header().entry_type();
|
||||
let mut d = Vec::new();
|
||||
std::io::Read::read_to_end(&mut e, &mut d).unwrap();
|
||||
(p, t, d)
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn archive_of_a_file_has_its_basename_as_the_only_member() {
|
||||
let a = start().await;
|
||||
let f = a.tmp.path().join("hello.txt");
|
||||
std::fs::write(&f, "hi there").unwrap();
|
||||
std::fs::set_permissions(&f, std::fs::Permissions::from_mode(0o640)).unwrap();
|
||||
let (s, body) = archive(&a, &f).await;
|
||||
assert_eq!(s, 200);
|
||||
let mut ar = tar::Archive::new(&body[..]);
|
||||
let mut list = ar.entries().unwrap();
|
||||
let mut e = list.next().unwrap().unwrap();
|
||||
assert_eq!(e.path().unwrap(), Path::new("hello.txt"));
|
||||
assert_eq!(e.header().mode().unwrap() & 0o7777, 0o640);
|
||||
let mut d = String::new();
|
||||
std::io::Read::read_to_string(&mut e, &mut d).unwrap();
|
||||
assert_eq!(d, "hi there");
|
||||
drop(e);
|
||||
assert!(list.next().is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn archive_of_a_directory_recurses_under_its_basename_and_keeps_links() {
|
||||
let a = start().await;
|
||||
let d = a.tmp.path().join("tree");
|
||||
std::fs::create_dir_all(d.join("sub")).unwrap();
|
||||
std::fs::write(d.join("a.txt"), "A").unwrap();
|
||||
std::fs::write(d.join("sub/b.txt"), "B").unwrap();
|
||||
std::os::unix::fs::symlink("/etc/passwd", d.join("link")).unwrap();
|
||||
// A trailing slash names the same directory.
|
||||
let (s, body) = archive(&a, &PathBuf::from(format!("{}/", d.display()))).await;
|
||||
assert_eq!(s, 200);
|
||||
let list = entries(&body);
|
||||
let names: Vec<_> = list.iter().map(|(p, _, _)| p.to_str().unwrap().trim_end_matches('/').to_string()).collect();
|
||||
for want in ["tree", "tree/a.txt", "tree/sub", "tree/sub/b.txt", "tree/link"] {
|
||||
assert!(names.contains(&want.to_string()), "{want} missing from {names:?}");
|
||||
}
|
||||
assert_eq!(list[0].0.to_str().unwrap().trim_end_matches('/'), "tree", "root member first");
|
||||
let (_, t, _) = list.iter().find(|(p, _, _)| p.ends_with("link")).unwrap();
|
||||
assert!(t.is_symlink(), "symlinks are archived, not followed");
|
||||
let (_, _, data) = list.iter().find(|(p, _, _)| p.ends_with("sub/b.txt")).unwrap();
|
||||
assert_eq!(data, b"B");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn archive_of_a_missing_path_is_404() {
|
||||
let a = start().await;
|
||||
let (s, _) = archive(&a, &a.tmp.path().join("nope")).await;
|
||||
assert_eq!(s, 404);
|
||||
}
|
||||
|
||||
// ── tunnel ─────────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn tunnel_pipes_bytes_both_ways_and_closes_with_the_tcp_side() {
|
||||
let a = start().await;
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let port = listener.local_addr().unwrap().port();
|
||||
tokio::spawn(async move {
|
||||
// Echo one message with a prefix, then hang up.
|
||||
let (mut s, _) = listener.accept().await.unwrap();
|
||||
let mut buf = [0u8; 64];
|
||||
let n = s.read(&mut buf).await.unwrap();
|
||||
s.write_all(b"echo:").await.unwrap();
|
||||
s.write_all(&buf[..n]).await.unwrap();
|
||||
});
|
||||
let mut ws = a.ws(&format!("/v1/tunnel?port={port}")).await;
|
||||
ws.send(Message::Binary(b"ping".to_vec().into())).await.unwrap();
|
||||
let mut got = Vec::new();
|
||||
loop {
|
||||
match tokio::time::timeout(Duration::from_secs(5), ws.next()).await.unwrap() {
|
||||
Some(Ok(Message::Binary(b))) => got.extend_from_slice(&b),
|
||||
Some(Ok(Message::Close(_))) | None => break,
|
||||
other => panic!("{other:?}"),
|
||||
}
|
||||
}
|
||||
assert_eq!(got, b"echo:ping");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn tunnel_to_a_closed_port_sends_an_error_frame() {
|
||||
let a = start().await;
|
||||
// Bind and drop to get a port that is very likely closed.
|
||||
let port = {
|
||||
let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
|
||||
l.local_addr().unwrap().port()
|
||||
};
|
||||
let mut ws = a.ws(&format!("/v1/tunnel?port={port}")).await;
|
||||
let v = next_text(&mut ws).await;
|
||||
assert_eq!(v["type"], "error");
|
||||
assert!(v["error"].as_str().unwrap().contains(&port.to_string()));
|
||||
let (s, _) = a.get("/v1/tunnel?port=nope").await;
|
||||
assert_eq!(s, 400);
|
||||
}
|
||||
|
||||
// ── activity ───────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
async fn activity_counts_execs_and_tunnels_and_ignores_its_own_probes() {
|
||||
let a = start().await;
|
||||
let (_, v0) = a.get("/v1/activity").await;
|
||||
assert_eq!(v0["active_execs"], 0);
|
||||
assert_eq!(v0["active_tunnels"], 0);
|
||||
let t0 = v0["last_activity_unix"].as_i64().unwrap();
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(1100)).await;
|
||||
a.get("/v1/health").await;
|
||||
a.get("/v1/activity").await;
|
||||
let (_, v) = a.get("/v1/activity").await;
|
||||
assert_eq!(v["last_activity_unix"].as_i64().unwrap(), t0, "health/activity are not activity");
|
||||
|
||||
let (_, d) = a.post("/v1/exec/detached", json!({ "cmd": ["sleep", "1"], "user": a.user })).await;
|
||||
let (_, v) = a.get("/v1/activity").await;
|
||||
assert!(v["last_activity_unix"].as_i64().unwrap() > t0);
|
||||
assert_eq!(v["active_execs"], 1);
|
||||
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let port = listener.local_addr().unwrap().port();
|
||||
tokio::spawn(async move {
|
||||
let (mut s, _) = listener.accept().await.unwrap();
|
||||
let mut b = [0u8; 8];
|
||||
let _ = s.read(&mut b).await;
|
||||
});
|
||||
let ws = a.ws(&format!("/v1/tunnel?port={port}")).await;
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
loop {
|
||||
let (_, v) = a.get("/v1/activity").await;
|
||||
if v["active_tunnels"] == 1 {
|
||||
break;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "tunnel never counted");
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
drop(ws);
|
||||
a.wait_finished(d["exec_id"].as_str().unwrap()).await;
|
||||
let deadline = Instant::now() + Duration::from_secs(5);
|
||||
loop {
|
||||
let (_, v) = a.get("/v1/activity").await;
|
||||
if v["active_tunnels"] == 0 && v["active_execs"] == 0 {
|
||||
break;
|
||||
}
|
||||
assert!(Instant::now() < deadline, "counters never returned to zero: {v}");
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
}
|
||||
|
||||
// ── root only ──────────────────────────────────────────────────────────────
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "needs root: switches to another user (nobody) and chowns an upload"]
|
||||
async fn as_root_execs_switch_user_and_uploads_chown() {
|
||||
assert!(is_root(), "run with sudo -E cargo test -- --ignored");
|
||||
let a = start().await;
|
||||
let (s, v) = a
|
||||
.post("/v1/exec", json!({ "cmd": sh("id -u; id -g; echo $HOME; echo $USER"), "user": "nobody" }))
|
||||
.await;
|
||||
assert_eq!(s, 200, "{v}");
|
||||
let out = decode(&v).1;
|
||||
let lines: Vec<_> = out.lines().collect();
|
||||
assert_eq!(lines[0], "65534");
|
||||
assert_eq!(lines[3], "nobody");
|
||||
assert_eq!(lines[2], passwd_home("nobody"));
|
||||
|
||||
let dir = a.tmp.path().to_str().unwrap().to_string();
|
||||
let (s, _) = put(
|
||||
&a,
|
||||
&[("dir", dir), ("name", "owned".into()), ("uid", "65534".into()), ("gid", "65534".into()), ("mode", "4755".into())],
|
||||
b"x".to_vec(),
|
||||
)
|
||||
.await;
|
||||
assert_eq!(s, 200);
|
||||
let m = std::fs::metadata(a.tmp.path().join("owned")).unwrap();
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
assert_eq!((m.uid(), m.gid()), (65534, 65534));
|
||||
// Mode applied after chown, so the setuid bit survives.
|
||||
assert_eq!(m.mode() & 0o7777, 0o4755);
|
||||
}
|
||||
Reference in new issue
Block a user