Skip to main content

max / makenotwork

12.5 KB · 288 lines History Blame Raw
1 //! `AgentRpc` — the Executor transport that talks to a remote `ops-agent`.
2 //!
3 //! Used for the macOS in-session sign step (the agent runs in the Aqua security
4 //! session where codesign can use the Developer ID key) and, later, the
5 //! resident observe plane. Non-signing, Linux/Windows work stays on
6 //! [`crate::SshExec`].
7 //!
8 //! Trust is the tailnet: the connection rides WireGuard, and the agent
9 //! independently re-checks the caller's identity via `whois`. This client also
10 //! enforces its own grant caller-side (the first half of double enforcement).
11
12 use crate::capability::{CapabilityDenied, CapabilitySet};
13 use crate::executor::{Executor, SyncOpts};
14 use crate::remote::{LogSink, RunOutput};
15 use crate::step::{Action, ObserveKind, Step};
16 use crate::wire::{Frame, HealthResponse, RunRequest};
17 use anyhow::{Context, Result};
18 use async_trait::async_trait;
19 use futures_util::StreamExt;
20 use std::path::Path;
21 use std::time::Duration;
22
23 /// A handle to one `ops-agent`, scoped to a caller-side capability set.
24 pub struct AgentRpc {
25 base_url: String,
26 host_label: String,
27 caps: CapabilitySet,
28 client: reqwest::Client,
29 }
30
31 /// How long to wait for the TCP connect before calling an agent unreachable.
32 ///
33 /// Applies to every request, including `/run` and `/pull`: it bounds only
34 /// establishing the connection, never a live response body, so a multi-minute
35 /// build or a large artifact download is unaffected. The agent is one tailnet
36 /// hop away, so a connect that has not landed in this long is not going to.
37 const CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
38
39 /// Whole-request cap for the short control calls (`/health`), which return a
40 /// small JSON body immediately.
41 ///
42 /// Deliberately NOT applied to the client as a whole: `reqwest`'s request
43 /// timeout runs until the response *body* has finished, so a client-level
44 /// timeout would sever `/run` mid-build and truncate `/pull` on a large DMG.
45 /// Those two are unbounded by design — a build legitimately runs for minutes —
46 /// and their liveness is covered by `CONNECT_TIMEOUT` plus the exit frame that
47 /// `run_streaming` requires before it will report success.
48 const HEALTH_TIMEOUT: Duration = Duration::from_secs(10);
49
50 impl AgentRpc {
51 /// `base_url` is e.g. `http://mbp.tailnet:8765` (the agent listens on the
52 /// tailnet interface). `host_label` is used only for audit messages.
53 pub fn new(
54 base_url: impl Into<String>,
55 host_label: impl Into<String>,
56 caps: CapabilitySet,
57 ) -> Self {
58 // A default-built client has no timeouts whatsoever, which left an agent
59 // that accepts TCP and then never answers (a wedged process, or a path
60 // that blackholes after the handshake) hanging a release indefinitely
61 // with no output. An unreachable host only ever failed because the OS
62 // gave up on the connect (~30s), which was luck, not design.
63 let client = reqwest::Client::builder()
64 .connect_timeout(CONNECT_TIMEOUT)
65 .build()
66 .expect("reqwest client with a connect timeout");
67 Self {
68 base_url: base_url.into(),
69 host_label: host_label.into(),
70 caps,
71 client,
72 }
73 }
74
75 /// `GET /health` — liveness plus the agent's own declared grant.
76 pub async fn health(&self) -> Result<HealthResponse> {
77 let resp = self
78 .client
79 .get(format!("{}/health", self.base_url))
80 .timeout(HEALTH_TIMEOUT)
81 .send()
82 .await
83 .with_context(|| {
84 format!(
85 "GET /health ({}): no answer within {}s — the agent is unreachable, or it is \
86 accepting connections without answering them",
87 self.host_label,
88 HEALTH_TIMEOUT.as_secs()
89 )
90 })?
91 .error_for_status()
92 .context("agent /health status")?;
93 resp.json().await.context("decoding /health")
94 }
95 }
96
97 /// Reconstruct an `ExitStatus` from the raw exit code the agent reports, which
98 /// is a process exit code (0..=255); `.success()` and the non-zero distinction —
99 /// all a recipe branches on — are preserved exactly. A negative/out-of-range
100 /// code (e.g. the agent's `-1` "terminated by signal" sentinel) collapses into
101 /// the low byte, so it reads back as non-zero but not its original value.
102 ///
103 /// Shared with the ssh path, which recovers a code the same way from its
104 /// [`crate::remote::RcSentinel`].
105 use crate::remote::exit_status_from_code as exit_status;
106
107 #[async_trait]
108 impl Executor for AgentRpc {
109 async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput> {
110 // Caller-side enforcement (half 1 of 2). The agent re-checks against
111 // its own grant (half 2) — this just fails fast before a round-trip.
112 if !self.caps.permits(&step.action) {
113 return Err(CapabilityDenied::new(&self.host_label, &step.action).into());
114 }
115
116 let resp = self
117 .client
118 .post(format!("{}/run", self.base_url))
119 .json(&RunRequest { step: step.clone() })
120 .send()
121 .await
122 .context("POST /run")?;
123 if resp.status() == reqwest::StatusCode::FORBIDDEN {
124 let body = resp.text().await.unwrap_or_default();
125 anyhow::bail!("agent denied /run: {}", body.trim());
126 }
127 let resp = resp.error_for_status().context("agent /run status")?;
128
129 let mut stream = resp.bytes_stream();
130 let mut buf: Vec<u8> = Vec::new();
131 let mut captured: Vec<u8> = Vec::new();
132 let mut exit_code: Option<i32> = None;
133
134 // A single NDJSON frame must fit in this much; a stream that sends this
135 // many bytes with no newline is malformed (and would otherwise grow the
136 // buffer unbounded — an OOM vector against the daemon).
137 const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
138 while let Some(chunk) = stream.next().await {
139 let chunk = chunk.context("reading /run stream")?;
140 buf.extend_from_slice(&chunk);
141 while let Some(nl) = buf.iter().position(|&b| b == b'\n') {
142 let line: Vec<u8> = buf.drain(..=nl).collect();
143 let line = &line[..line.len() - 1];
144 if line.is_empty() {
145 continue;
146 }
147 let frame: Frame = serde_json::from_slice(line).with_context(|| {
148 format!("decoding agent frame: {}", String::from_utf8_lossy(line))
149 })?;
150 match frame {
151 Frame::Chunk { text } => {
152 crate::remote::push_bounded(
153 &mut captured,
154 text.as_bytes(),
155 crate::remote::OUTPUT_TAIL_CAP,
156 );
157 sink.write_chunk(text.as_bytes()).await;
158 }
159 Frame::Exit { code } => exit_code = Some(code),
160 Frame::Error { message } => anyhow::bail!("agent error: {message}"),
161 // Forward-compatible: a newer agent's unknown frame is ignored.
162 Frame::Unknown => {}
163 }
164 }
165 // `buf` now holds at most one incomplete trailing line — bound it.
166 anyhow::ensure!(
167 buf.len() <= MAX_FRAME_BYTES,
168 "agent /run frame exceeded {MAX_FRAME_BYTES} bytes without a newline",
169 );
170 }
171
172 let code = exit_code.context("agent closed /run stream without an exit frame")?;
173 Ok(RunOutput {
174 status: exit_status(code),
175 stdout: captured,
176 stderr: Vec::new(),
177 })
178 }
179
180 async fn pull_file(&self, remote: &Path, local: &Path, _opts: &SyncOpts) -> Result<()> {
181 // Caller-side enforcement (half 1 of 2), same as `run_streaming`. `/pull`
182 // is observe-plane: it needs the `artifact` grant. Failing fast here
183 // matters more than elsewhere — a pull is the *last* step of a release,
184 // so an ungranted caller would otherwise learn it after a full build and
185 // an Apple notary round trip.
186 let action = Action::Observe(ObserveKind::Artifact);
187 if !self.caps.permits(&action) {
188 return Err(CapabilityDenied::new(&self.host_label, &action).into());
189 }
190
191 let resp = self
192 .client
193 .get(format!("{}/pull", self.base_url))
194 .query(&[("path", remote.to_string_lossy().as_ref())])
195 .send()
196 .await
197 .context("GET /pull")?;
198 // Surface the agent's reason (ungranted, /pull disabled, outside
199 // pull_root); `error_for_status` alone would reduce it to a bare 403.
200 if resp.status() == reqwest::StatusCode::FORBIDDEN {
201 let body = resp.text().await.unwrap_or_default();
202 anyhow::bail!("agent denied /pull: {}", body.trim());
203 }
204 let resp = resp.error_for_status().context("agent /pull status")?;
205 // Stream the body straight to disk in chunks — a signed .app/.dmg can be
206 // hundreds of MB, so never buffer the whole artifact in the daemon's heap
207 // (the agent's /pull already streams; this is the matching client half).
208 if let Some(parent) = local.parent() {
209 tokio::fs::create_dir_all(parent).await.ok();
210 }
211 let mut file = tokio::fs::File::create(local)
212 .await
213 .context("creating pulled artifact")?;
214 let mut stream = resp.bytes_stream();
215 while let Some(chunk) = stream.next().await {
216 let chunk = chunk.context("reading /pull body")?;
217 tokio::io::AsyncWriteExt::write_all(&mut file, &chunk)
218 .await
219 .context("writing pulled artifact")?;
220 }
221 tokio::io::AsyncWriteExt::flush(&mut file)
222 .await
223 .context("flushing pulled artifact")?;
224 Ok(())
225 }
226
227 async fn pull_dir(&self, _remote: &Path, _local: &Path, _opts: &SyncOpts) -> Result<()> {
228 // `/pull` serves exactly one file per request (a streamed HTTP body);
229 // there is no directory form, and inventing one would mean walking the
230 // host's tree over the agent's exec surface. Artifact retrieval is
231 // per-file by design — collect a directory with SshExec/rsync.
232 anyhow::bail!(
233 "AgentRpc::pull_dir is unsupported by design; the agent serves one file per \
234 /pull — use pull_file, or SshExec/rsync for a directory"
235 )
236 }
237
238 async fn pull_glob(
239 &self,
240 _remote_glob: &str,
241 _local_dir: &Path,
242 _opts: &SyncOpts,
243 ) -> Result<()> {
244 // `/pull` takes one concrete path and has no expansion surface — and
245 // giving it one would mean the agent globbing its own filesystem on a
246 // caller's behalf. Artifact collection over the agent is per-file.
247 anyhow::bail!(
248 "AgentRpc::pull_glob is unsupported by design; the agent serves one concrete \
249 path per /pull — resolve the glob on the host and use pull_file, or collect \
250 with SshExec/rsync"
251 )
252 }
253
254 async fn push_dir(&self, _local: &Path, _remote: &Path, _opts: &SyncOpts) -> Result<()> {
255 // The agent transport is for in-session *execution*. Bulk data movement
256 // onto the host uses SshExec/rsync or a `git pull` step inside the
257 // recipe — keeping the agent's surface small (one open port, exec only).
258 anyhow::bail!(
259 "AgentRpc::push_dir is unsupported by design; move source/data with SshExec/rsync \
260 or a `git pull` step, not the agent"
261 )
262 }
263
264 async fn preflight(&self) -> Result<()> {
265 let health = self.health().await.with_context(|| {
266 format!(
267 "agent /health failed for `{}` — is ops-agent running in the Aqua session on the build host?",
268 self.host_label,
269 )
270 })?;
271 // Detect a skewed agent here rather than mid-stream. version 0 = a legacy
272 // agent that predates the field (HealthResponse.version defaults to 0);
273 // tolerate it, but refuse a future major we don't speak.
274 anyhow::ensure!(
275 health.version == 0 || health.version == crate::wire::PROTOCOL_VERSION,
276 "agent `{}` speaks wire protocol v{}, but this client expects v{} — upgrade the mismatched side",
277 self.host_label,
278 health.version,
279 crate::wire::PROTOCOL_VERSION,
280 );
281 Ok(())
282 }
283
284 fn capabilities(&self) -> &CapabilitySet {
285 &self.caps
286 }
287 }
288