Skip to main content

max / makenotwork

13.2 KB · 302 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 /// Install ring as the process-wide rustls crypto provider, unless the process
51 /// already installed one. reqwest is compiled `rustls-no-provider` to keep
52 /// aws-lc-rs (a C backend) out of the tree, so without this a client cannot be
53 /// built at all; doing it here keeps the crate usable by a consumer that has
54 /// made no choice, while `install_default`'s no-op-if-set behaviour leaves a
55 /// consumer that has made one alone.
56 fn install_crypto_provider() {
57 static INSTALLED: std::sync::Once = std::sync::Once::new();
58 INSTALLED.call_once(|| {
59 let _ = rustls::crypto::ring::default_provider().install_default();
60 });
61 }
62
63 impl AgentRpc {
64 /// `base_url` is e.g. `http://mbp.tailnet:8765` (the agent listens on the
65 /// tailnet interface). `host_label` is used only for audit messages.
66 pub fn new(
67 base_url: impl Into<String>,
68 host_label: impl Into<String>,
69 caps: CapabilitySet,
70 ) -> Self {
71 // A default-built client has no timeouts whatsoever, which left an agent
72 // that accepts TCP and then never answers (a wedged process, or a path
73 // that blackholes after the handshake) hanging a release indefinitely
74 // with no output. An unreachable host only ever failed because the OS
75 // gave up on the connect (~30s), which was luck, not design.
76 install_crypto_provider();
77 let client = reqwest::Client::builder()
78 .connect_timeout(CONNECT_TIMEOUT)
79 .build()
80 .expect("reqwest client with a connect timeout");
81 Self {
82 base_url: base_url.into(),
83 host_label: host_label.into(),
84 caps,
85 client,
86 }
87 }
88
89 /// `GET /health` — liveness plus the agent's own declared grant.
90 pub async fn health(&self) -> Result<HealthResponse> {
91 let resp = self
92 .client
93 .get(format!("{}/health", self.base_url))
94 .timeout(HEALTH_TIMEOUT)
95 .send()
96 .await
97 .with_context(|| {
98 format!(
99 "GET /health ({}): no answer within {}s — the agent is unreachable, or it is \
100 accepting connections without answering them",
101 self.host_label,
102 HEALTH_TIMEOUT.as_secs()
103 )
104 })?
105 .error_for_status()
106 .context("agent /health status")?;
107 resp.json().await.context("decoding /health")
108 }
109 }
110
111 /// Reconstruct an `ExitStatus` from the raw exit code the agent reports, which
112 /// is a process exit code (0..=255); `.success()` and the non-zero distinction —
113 /// all a recipe branches on — are preserved exactly. A negative/out-of-range
114 /// code (e.g. the agent's `-1` "terminated by signal" sentinel) collapses into
115 /// the low byte, so it reads back as non-zero but not its original value.
116 ///
117 /// Shared with the ssh path, which recovers a code the same way from its
118 /// [`crate::remote::RcSentinel`].
119 use crate::remote::exit_status_from_code as exit_status;
120
121 #[async_trait]
122 impl Executor for AgentRpc {
123 async fn run_streaming(&self, step: &Step, sink: &mut dyn LogSink) -> Result<RunOutput> {
124 // Caller-side enforcement (half 1 of 2). The agent re-checks against
125 // its own grant (half 2) — this just fails fast before a round-trip.
126 if !self.caps.permits(&step.action) {
127 return Err(CapabilityDenied::new(&self.host_label, &step.action).into());
128 }
129
130 let resp = self
131 .client
132 .post(format!("{}/run", self.base_url))
133 .json(&RunRequest { step: step.clone() })
134 .send()
135 .await
136 .context("POST /run")?;
137 if resp.status() == reqwest::StatusCode::FORBIDDEN {
138 let body = resp.text().await.unwrap_or_default();
139 anyhow::bail!("agent denied /run: {}", body.trim());
140 }
141 let resp = resp.error_for_status().context("agent /run status")?;
142
143 let mut stream = resp.bytes_stream();
144 let mut buf: Vec<u8> = Vec::new();
145 let mut captured: Vec<u8> = Vec::new();
146 let mut exit_code: Option<i32> = None;
147
148 // A single NDJSON frame must fit in this much; a stream that sends this
149 // many bytes with no newline is malformed (and would otherwise grow the
150 // buffer unbounded — an OOM vector against the daemon).
151 const MAX_FRAME_BYTES: usize = 16 * 1024 * 1024;
152 while let Some(chunk) = stream.next().await {
153 let chunk = chunk.context("reading /run stream")?;
154 buf.extend_from_slice(&chunk);
155 while let Some(nl) = buf.iter().position(|&b| b == b'\n') {
156 let line: Vec<u8> = buf.drain(..=nl).collect();
157 let line = &line[..line.len() - 1];
158 if line.is_empty() {
159 continue;
160 }
161 let frame: Frame = serde_json::from_slice(line).with_context(|| {
162 format!("decoding agent frame: {}", String::from_utf8_lossy(line))
163 })?;
164 match frame {
165 Frame::Chunk { text } => {
166 crate::remote::push_bounded(
167 &mut captured,
168 text.as_bytes(),
169 crate::remote::OUTPUT_TAIL_CAP,
170 );
171 sink.write_chunk(text.as_bytes()).await;
172 }
173 Frame::Exit { code } => exit_code = Some(code),
174 Frame::Error { message } => anyhow::bail!("agent error: {message}"),
175 // Forward-compatible: a newer agent's unknown frame is ignored.
176 Frame::Unknown => {}
177 }
178 }
179 // `buf` now holds at most one incomplete trailing line — bound it.
180 anyhow::ensure!(
181 buf.len() <= MAX_FRAME_BYTES,
182 "agent /run frame exceeded {MAX_FRAME_BYTES} bytes without a newline",
183 );
184 }
185
186 let code = exit_code.context("agent closed /run stream without an exit frame")?;
187 Ok(RunOutput {
188 status: exit_status(code),
189 stdout: captured,
190 stderr: Vec::new(),
191 })
192 }
193
194 async fn pull_file(&self, remote: &Path, local: &Path, _opts: &SyncOpts) -> Result<()> {
195 // Caller-side enforcement (half 1 of 2), same as `run_streaming`. `/pull`
196 // is observe-plane: it needs the `artifact` grant. Failing fast here
197 // matters more than elsewhere — a pull is the *last* step of a release,
198 // so an ungranted caller would otherwise learn it after a full build and
199 // an Apple notary round trip.
200 let action = Action::Observe(ObserveKind::Artifact);
201 if !self.caps.permits(&action) {
202 return Err(CapabilityDenied::new(&self.host_label, &action).into());
203 }
204
205 let resp = self
206 .client
207 .get(format!("{}/pull", self.base_url))
208 .query(&[("path", remote.to_string_lossy().as_ref())])
209 .send()
210 .await
211 .context("GET /pull")?;
212 // Surface the agent's reason (ungranted, /pull disabled, outside
213 // pull_root); `error_for_status` alone would reduce it to a bare 403.
214 if resp.status() == reqwest::StatusCode::FORBIDDEN {
215 let body = resp.text().await.unwrap_or_default();
216 anyhow::bail!("agent denied /pull: {}", body.trim());
217 }
218 let resp = resp.error_for_status().context("agent /pull status")?;
219 // Stream the body straight to disk in chunks — a signed .app/.dmg can be
220 // hundreds of MB, so never buffer the whole artifact in the daemon's heap
221 // (the agent's /pull already streams; this is the matching client half).
222 if let Some(parent) = local.parent() {
223 tokio::fs::create_dir_all(parent).await.ok();
224 }
225 let mut file = tokio::fs::File::create(local)
226 .await
227 .context("creating pulled artifact")?;
228 let mut stream = resp.bytes_stream();
229 while let Some(chunk) = stream.next().await {
230 let chunk = chunk.context("reading /pull body")?;
231 tokio::io::AsyncWriteExt::write_all(&mut file, &chunk)
232 .await
233 .context("writing pulled artifact")?;
234 }
235 tokio::io::AsyncWriteExt::flush(&mut file)
236 .await
237 .context("flushing pulled artifact")?;
238 Ok(())
239 }
240
241 async fn pull_dir(&self, _remote: &Path, _local: &Path, _opts: &SyncOpts) -> Result<()> {
242 // `/pull` serves exactly one file per request (a streamed HTTP body);
243 // there is no directory form, and inventing one would mean walking the
244 // host's tree over the agent's exec surface. Artifact retrieval is
245 // per-file by design — collect a directory with SshExec/rsync.
246 anyhow::bail!(
247 "AgentRpc::pull_dir is unsupported by design; the agent serves one file per \
248 /pull — use pull_file, or SshExec/rsync for a directory"
249 )
250 }
251
252 async fn pull_glob(
253 &self,
254 _remote_glob: &str,
255 _local_dir: &Path,
256 _opts: &SyncOpts,
257 ) -> Result<()> {
258 // `/pull` takes one concrete path and has no expansion surface — and
259 // giving it one would mean the agent globbing its own filesystem on a
260 // caller's behalf. Artifact collection over the agent is per-file.
261 anyhow::bail!(
262 "AgentRpc::pull_glob is unsupported by design; the agent serves one concrete \
263 path per /pull — resolve the glob on the host and use pull_file, or collect \
264 with SshExec/rsync"
265 )
266 }
267
268 async fn push_dir(&self, _local: &Path, _remote: &Path, _opts: &SyncOpts) -> Result<()> {
269 // The agent transport is for in-session *execution*. Bulk data movement
270 // onto the host uses SshExec/rsync or a `git pull` step inside the
271 // recipe — keeping the agent's surface small (one open port, exec only).
272 anyhow::bail!(
273 "AgentRpc::push_dir is unsupported by design; move source/data with SshExec/rsync \
274 or a `git pull` step, not the agent"
275 )
276 }
277
278 async fn preflight(&self) -> Result<()> {
279 let health = self.health().await.with_context(|| {
280 format!(
281 "agent /health failed for `{}` — is ops-agent running in the Aqua session on the build host?",
282 self.host_label,
283 )
284 })?;
285 // Detect a skewed agent here rather than mid-stream. version 0 = a legacy
286 // agent that predates the field (HealthResponse.version defaults to 0);
287 // tolerate it, but refuse a future major we don't speak.
288 anyhow::ensure!(
289 health.version == 0 || health.version == crate::wire::PROTOCOL_VERSION,
290 "agent `{}` speaks wire protocol v{}, but this client expects v{} — upgrade the mismatched side",
291 self.host_label,
292 health.version,
293 crate::wire::PROTOCOL_VERSION,
294 );
295 Ok(())
296 }
297
298 fn capabilities(&self) -> &CapabilitySet {
299 &self.caps
300 }
301 }
302