Skip to main content

max / makenotwork

Dedup ct_eq + sando live_log into ops-core; put sando errors on JSON - ops-core: new daemon module holds the byte-identical constant-time bearer-token compare (ct_eq); both daemons now call it. - sando: drop the hand-rolled live_log.rs for ops_core::live_log with a GateLogChunk callback (behavior-identical; parity verified). - sando: error responses now use bento's leak-safe JSON envelope. Db/Other return a generic "internal server error" instead of to_string(), so SQL and anyhow detail no longer reach the client; client errors keep their message. New tests pin both the envelope and the no-leak property. require_bearer, recover_orphaned_running, and the executor-map builders stay per-daemon: they have genuinely diverged (peer logging, different schemas, different key types), so folding them would be a behavior change, not a mechanical dedup. metrics init/render is moot — sando has none. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-23 14:11 UTC
Commit: 91e82ecde8134ab4ff7ea005e0f2cb382c5fed4d
Parent: 5df81b8
8 files changed, +134 insertions, -246 deletions
M bento/daemon/src/routes.rs +1 -22
@@ -67,7 +67,7 @@ async fn require_bearer(
67 67 .get(axum::http::header::AUTHORIZATION)
68 68 .and_then(|h| h.to_str().ok())
69 69 .and_then(|h| h.strip_prefix("Bearer "))
70 - .is_some_and(|t| ct_eq(t, expected));
70 + .is_some_and(|t| ops_core::daemon::ct_eq(t, expected));
71 71 if ok {
72 72 next.run(req).await
73 73 } else {
@@ -80,19 +80,6 @@ async fn require_bearer(
80 80 }
81 81 }
82 82
83 - /// Constant-time string compare (length may leak; bytes do not short-circuit).
84 - fn ct_eq(a: &str, b: &str) -> bool {
85 - let (a, b) = (a.as_bytes(), b.as_bytes());
86 - if a.len() != b.len() {
87 - return false;
88 - }
89 - let mut diff = 0u8;
90 - for (x, y) in a.iter().zip(b) {
91 - diff |= x ^ y;
92 - }
93 - diff == 0
94 - }
95 -
96 83 #[derive(Serialize)]
97 84 struct StateView {
98 85 build: Option<BuildView>,
@@ -590,14 +577,6 @@ repo = "{}"
590 577 assert_eq!(resp.status(), StatusCode::OK);
591 578 }
592 579
593 - #[test]
594 - fn ct_eq_matches_only_identical_strings() {
595 - assert!(ct_eq("abc", "abc"));
596 - assert!(!ct_eq("abc", "abd"));
597 - assert!(!ct_eq("abc", "ab"));
598 - assert!(ct_eq("", ""));
599 - }
600 -
601 580 #[tokio::test]
602 581 async fn step_log_rejects_traversal() {
603 582 let tmp = tempfile::tempdir().unwrap();
M sando/daemon/src/error.rs +63 -6
@@ -18,13 +18,31 @@ pub enum Error {
18 18
19 19 impl IntoResponse for Error {
20 20 fn into_response(self) -> Response {
21 - let status = match &self {
22 - Error::NotFound => StatusCode::NOT_FOUND,
23 - Error::BadRequest(_) => StatusCode::BAD_REQUEST,
24 - Error::GateBlocked(_) => StatusCode::CONFLICT,
25 - _ => StatusCode::INTERNAL_SERVER_ERROR,
21 + // Client errors carry their (already user-facing) message; server errors
22 + // are logged in full but return a generic body so internal detail (SQL
23 + // text, anyhow chains, file paths) never leaks to the client. Every
24 + // response is the JSON envelope `{"error": <message>}`, matching Bento
25 + // and the 400/404/409/500 shape the endpoints already speak.
26 + let (status, message) = match &self {
27 + Error::NotFound => (StatusCode::NOT_FOUND, self.to_string()),
28 + Error::BadRequest(_) => (StatusCode::BAD_REQUEST, self.to_string()),
29 + Error::GateBlocked(_) => (StatusCode::CONFLICT, self.to_string()),
30 + Error::Db(e) => {
31 + tracing::error!(error = %e, "internal db error");
32 + (
33 + StatusCode::INTERNAL_SERVER_ERROR,
34 + "internal server error".to_string(),
35 + )
36 + }
37 + Error::Other(e) => {
38 + tracing::error!(error = format!("{e:#}"), "internal error");
39 + (
40 + StatusCode::INTERNAL_SERVER_ERROR,
41 + "internal server error".to_string(),
42 + )
43 + }
26 44 };
27 - (status, self.to_string()).into_response()
45 + (status, axum::Json(serde_json::json!({ "error": message }))).into_response()
28 46 }
29 47 }
30 48
@@ -58,3 +76,42 @@ where
58 76 Ok(TypedBody(value))
59 77 }
60 78 }
79 +
80 + #[cfg(test)]
81 + mod tests {
82 + use super::*;
83 + use axum::response::IntoResponse;
84 + use http_body_util::BodyExt;
85 +
86 + async fn parts(err: Error) -> (StatusCode, serde_json::Value) {
87 + let resp = err.into_response();
88 + let status = resp.status();
89 + let bytes = resp.into_body().collect().await.unwrap().to_bytes();
90 + (status, serde_json::from_slice(&bytes).unwrap())
91 + }
92 +
93 + #[tokio::test]
94 + async fn client_errors_keep_their_message_in_a_json_envelope() {
95 + let (status, body) = parts(Error::BadRequest("no version specified".into())).await;
96 + assert_eq!(status, StatusCode::BAD_REQUEST);
97 + assert_eq!(body["error"], "bad request: no version specified");
98 +
99 + let (status, body) = parts(Error::GateBlocked("boot_smoke red".into())).await;
100 + assert_eq!(status, StatusCode::CONFLICT);
101 + assert_eq!(body["error"], "gate not satisfied: boot_smoke red");
102 + }
103 +
104 + /// The leak fix: a server error is a generic body, never the underlying SQL
105 + /// text or anyhow chain (which used to reach the client via `to_string()`).
106 + #[tokio::test]
107 + async fn server_errors_do_not_leak_internal_detail() {
108 + let secret = "no such column: super_secret_internal_table.token";
109 + let (status, body) = parts(Error::Other(anyhow::anyhow!(secret))).await;
110 + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
111 + assert_eq!(body["error"], "internal server error");
112 + assert!(
113 + !body.to_string().contains("super_secret"),
114 + "internal detail must not reach the client: {body}"
115 + );
116 + }
117 + }
M sando/daemon/src/gates.rs +23 -4
@@ -7,17 +7,36 @@ use crate::classify;
7 7 use crate::config::Config;
8 8 use crate::domain::{GateKind, GateRunId, TierId, Version};
9 9 use crate::events::{self, Event, EventTx};
10 - use crate::live_log::LiveLog;
11 10 use crate::outcome::{GateBlocker, GateFailure, GateOutcome, LogRef, PassNote};
12 11 use crate::topology::Gate;
13 12 use anyhow::{Context, Result};
14 13 use chrono::Utc;
14 + use ops_core::live_log::LiveLog;
15 + use ops_core::remote::LogSink; // brings `LiveLog::write_chunk` (the sink trait) into scope
15 16 use sqlx::SqlitePool;
16 17 use std::path::PathBuf;
17 18 use std::sync::Arc;
18 19 use tokio::io::AsyncReadExt;
19 20 use tokio::process::Command;
20 21
22 + /// The gate live-log callback: emit each chunk as a `GateLogChunk` event so the
23 + /// TUI sees the tail stream in real time. `ops_core::live_log::LiveLog` owns the
24 + /// disk append and the per-run sequence counter; this closure is the one
25 + /// tool-specific bit (Sando previously carried a whole `live_log.rs` copy that
26 + /// hardcoded exactly this emit).
27 + fn gate_chunk_cb(events: EventTx, run_id: GateRunId) -> ops_core::live_log::ChunkCallback {
28 + Box::new(move |seq, text| {
29 + events::emit(
30 + &events,
31 + Event::GateLogChunk {
32 + run_id,
33 + seq,
34 + text: text.to_owned(),
35 + },
36 + );
37 + })
38 + }
39 +
21 40 pub struct GateCtx {
22 41 pub pool: SqlitePool,
23 42 pub cfg: Arc<Config>,
@@ -1564,7 +1583,7 @@ async fn boot_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
1564 1583 Err(e) => {
1565 1584 // Spawn failures get a one-off log line via LiveLog so the
1566 1585 // on-disk file still exists for `GET /logs/...`.
1567 - let mut log = LiveLog::open(ctx.events.clone(), run_id, log_path).await;
1586 + let mut log = LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await;
1568 1587 log.write_chunk(format!("spawn: {e}\n").as_bytes()).await;
1569 1588 log.close().await;
1570 1589 return Ok(GateOutcome::failed(GateFailure::SpawnFailed {
@@ -1580,7 +1599,7 @@ async fn boot_smoke(ctx: &GateCtx, run_id: GateRunId) -> Result<GateOutcome> {
1580 1599 // stream for post-mortem reads. The drainers exit when their pipe
1581 1600 // closes — which happens when the child exits naturally or after kill.
1582 1601 let log = std::sync::Arc::new(tokio::sync::Mutex::new(
1583 - LiveLog::open(ctx.events.clone(), run_id, log_path).await,
1602 + LiveLog::open(log_path, gate_chunk_cb(ctx.events.clone(), run_id)).await,
1584 1603 ));
1585 1604 let stdout_task = tokio::spawn(stream_into_log(child.stdout.take(), log.clone()));
1586 1605 let stderr_task = tokio::spawn(stream_into_log(child.stderr.take(), log.clone()));
@@ -1797,7 +1816,7 @@ async fn stream_child_to_live_log(
1797 1816 log_path: PathBuf,
1798 1817 ) -> Result<(Vec<u8>, Vec<u8>, std::process::ExitStatus)> {
1799 1818 let log = std::sync::Arc::new(tokio::sync::Mutex::new(
1800 - LiveLog::open(events, run_id, log_path).await,
1819 + LiveLog::open(log_path, gate_chunk_cb(events, run_id)).await,
1801 1820 ));
1802 1821 let stdout_task = tokio::spawn(stream_into_log(child.stdout.take(), log.clone()));
1803 1822 let stderr_task = tokio::spawn(stream_into_log(child.stderr.take(), log.clone()));
M sando/daemon/src/lib.rs -1
@@ -24,7 +24,6 @@ pub mod error;
24 24 pub mod events;
25 25 pub mod gates;
26 26 pub mod git;
27 - pub mod live_log;
28 27 pub mod outcome;
29 28 pub mod routes;
30 29 pub mod runs;
D sando/daemon/src/live_log.rs -189
@@ -1,189 +0,0 @@
1 - //! Live gate-log sink.
2 - //!
3 - //! Wraps the on-disk per-run log file with chunk broadcasting. Gate
4 - //! runners (`cargo_test`, `boot_smoke`) push their stdout/stderr through
5 - //! `LiveLog::write_chunk` as it arrives; the sink:
6 - //! 1. appends to the on-disk file at `<logs_root>/<version>/<gate>.log`
7 - //! so the post-mortem `GET /logs/...` route still has the full byte
8 - //! stream;
9 - //! 2. broadcasts a `GateLogChunk` event with a UTF-8-lossy slice of the
10 - //! same bytes, so the TUI sees the tail in real time.
11 - //!
12 - //! Chunks reflect tokio read boundaries — they are NOT line-aligned.
13 - //! Consumers that want lines must reassemble; the on-disk log preserves
14 - //! the exact byte stream for that purpose.
15 -
16 - use crate::domain::GateRunId;
17 - use crate::events::{self, Event, EventTx};
18 - use std::path::{Path, PathBuf};
19 - use tokio::fs::File;
20 - use tokio::io::AsyncWriteExt;
21 -
22 - pub struct LiveLog {
23 - file: Option<File>,
24 - /// Kept for diagnostic logging when file IO is unavailable.
25 - path: PathBuf,
26 - events: EventTx,
27 - run_id: GateRunId,
28 - seq: u32,
29 - }
30 -
31 - impl LiveLog {
32 - /// Open the log file for append-streaming. Creates parent directories
33 - /// as needed. If the file can't be opened, the sink degrades to
34 - /// "broadcast only" — chunks still go out as `GateLogChunk` events;
35 - /// the missing on-disk log is logged as a warning. This matches the
36 - /// pre-step-6 invariant: a broken log dir doesn't turn a passing gate
37 - /// red.
38 - pub async fn open(events: EventTx, run_id: GateRunId, path: PathBuf) -> Self {
39 - let file = open_for_append(&path).await;
40 - Self {
41 - file,
42 - path,
43 - events,
44 - run_id,
45 - seq: 0,
46 - }
47 - }
48 -
49 - /// Append `bytes` to the on-disk log and broadcast a `GateLogChunk`.
50 - /// The broadcast goes out even if the disk write fails — operators
51 - /// watching live still see the chunk.
52 - pub async fn write_chunk(&mut self, bytes: &[u8]) {
53 - if bytes.is_empty() {
54 - return;
55 - }
56 - if let Some(f) = self.file.as_mut()
57 - && let Err(e) = f.write_all(bytes).await
58 - {
59 - tracing::warn!(error = %e, path = %self.path.display(), "live log write failed");
60 - self.file = None;
61 - }
62 - let text = String::from_utf8_lossy(bytes).into_owned();
63 - events::emit(
64 - &self.events,
65 - Event::GateLogChunk {
66 - run_id: self.run_id,
67 - seq: self.seq,
68 - text,
69 - },
70 - );
71 - self.seq = self.seq.saturating_add(1);
72 - }
73 -
74 - /// Flush and close the file. Best-effort: errors are logged.
75 - pub async fn close(mut self) {
76 - if let Some(mut f) = self.file.take()
77 - && let Err(e) = f.flush().await
78 - {
79 - tracing::warn!(error = %e, path = %self.path.display(), "live log flush failed");
80 - }
81 - }
82 -
83 - pub fn run_id(&self) -> GateRunId {
84 - self.run_id
85 - }
86 - pub fn chunks_emitted(&self) -> u32 {
87 - self.seq
88 - }
89 - }
90 -
91 - /// Lets a Sando `LiveLog` be used directly as an [`ops_exec::LogSink`], so a
92 - /// gate or deploy run through `executor.run_streaming` streams into the same
93 - /// on-disk file + `GateLogChunk` broadcast the bespoke runner used. Delegates
94 - /// to the inherent [`LiveLog::write_chunk`] (inherent methods take precedence,
95 - /// so this is not recursive).
96 - #[async_trait::async_trait]
97 - impl ops_exec::LogSink for LiveLog {
98 - async fn write_chunk(&mut self, bytes: &[u8]) {
99 - self.write_chunk(bytes).await;
100 - }
101 - }
102 -
103 - async fn open_for_append(path: &Path) -> Option<File> {
104 - if let Some(parent) = path.parent()
105 - && let Err(e) = tokio::fs::create_dir_all(parent).await
106 - {
107 - tracing::warn!(error = %e, dir = %parent.display(), "could not create gate log dir");
108 - return None;
109 - }
110 - match tokio::fs::OpenOptions::new()
111 - .create(true)
112 - .append(true)
113 - .open(path)
114 - .await
115 - {
116 - Ok(f) => Some(f),
117 - Err(e) => {
118 - tracing::warn!(error = %e, path = %path.display(), "could not open gate log file");
119 - None
120 - }
121 - }
122 - }
123 -
124 - #[cfg(test)]
125 - mod tests {
126 - use super::*;
127 - use crate::events::EventEnvelope;
128 -
129 - #[tokio::test]
130 - async fn write_chunk_emits_event_and_appends_to_file() {
131 - let dir = tempfile::tempdir().unwrap();
132 - let path = dir.path().join("nested/test.log");
133 - let events = events::channel();
134 - let mut rx = events.subscribe_logs();
135 - let mut log = LiveLog::open(events.clone(), GateRunId(7), path.clone()).await;
136 - log.write_chunk(b"hello ").await;
137 - log.write_chunk(b"world\n").await;
138 - log.close().await;
139 -
140 - let on_disk = tokio::fs::read_to_string(&path).await.unwrap();
141 - assert_eq!(on_disk, "hello world\n");
142 -
143 - let mut chunks: Vec<String> = Vec::new();
144 - while let Ok(env) = rx.try_recv() {
145 - if let Event::GateLogChunk { run_id, seq, text } = env.event {
146 - assert_eq!(run_id, GateRunId(7));
147 - assert_eq!(seq as usize, chunks.len());
148 - chunks.push(text);
149 - }
150 - }
151 - assert_eq!(chunks, vec!["hello ".to_string(), "world\n".to_string()]);
152 - }
153 -
154 - #[tokio::test]
155 - async fn write_chunk_emits_even_when_file_cannot_be_opened() {
156 - // Point at a path whose parent is a regular file — create_dir_all
157 - // will fail. The broadcast must still fire.
158 - let dir = tempfile::tempdir().unwrap();
159 - let blocker = dir.path().join("blocker");
160 - tokio::fs::write(&blocker, b"i am a file, not a dir")
161 - .await
162 - .unwrap();
163 - let path = blocker.join("inside.log"); // parent is a file
164 - let events = events::channel();
165 - let mut rx = events.subscribe_logs();
166 - let mut log = LiveLog::open(events.clone(), GateRunId(1), path).await;
167 - log.write_chunk(b"streamed despite no file\n").await;
168 - log.close().await;
169 -
170 - let env: EventEnvelope = rx.try_recv().unwrap();
171 - match env.event {
172 - Event::GateLogChunk { text, .. } => assert_eq!(text, "streamed despite no file\n"),
173 - _ => panic!("expected GateLogChunk"),
174 - }
175 - }
176 -
177 - #[tokio::test]
178 - async fn empty_chunk_is_noop() {
179 - let dir = tempfile::tempdir().unwrap();
180 - let path = dir.path().join("empty.log");
181 - let events = events::channel();
182 - let mut rx = events.subscribe_logs();
183 - let mut log = LiveLog::open(events.clone(), GateRunId(9), path).await;
184 - log.write_chunk(b"").await;
185 - assert_eq!(log.chunks_emitted(), 0);
186 - assert!(rx.try_recv().is_err());
187 - log.close().await;
188 - }
189 - }
M sando/daemon/src/routes/mod.rs +1 -24
@@ -61,7 +61,7 @@ async fn require_bearer(
61 61 .get(axum::http::header::AUTHORIZATION)
62 62 .and_then(|h| h.to_str().ok())
63 63 .and_then(|h| h.strip_prefix("Bearer "))
64 - .is_some_and(|t| ct_eq(t, expected));
64 + .is_some_and(|t| ops_core::daemon::ct_eq(t, expected));
65 65 if ok {
66 66 // Attribute mutations (POST) to a caller. Reads are GET and poll every
67 67 // few seconds, so logging them would drown the journal; the audit value
@@ -88,20 +88,6 @@ async fn require_bearer(
88 88 }
89 89 }
90 90
91 - /// Constant-time string compare. Length is allowed to leak (a token's length is
92 - /// not the secret); the byte comparison does not short-circuit.
93 - fn ct_eq(a: &str, b: &str) -> bool {
94 - let (a, b) = (a.as_bytes(), b.as_bytes());
95 - if a.len() != b.len() {
96 - return false;
97 - }
98 - let mut diff = 0u8;
99 - for (x, y) in a.iter().zip(b) {
100 - diff |= x ^ y;
101 - }
102 - diff == 0
103 - }
104 -
105 91 #[derive(Serialize)]
106 92 pub(crate) struct StateView {
107 93 /// The running sandod's own package version. Lets a self-update caller
@@ -2556,15 +2542,6 @@ mod tests {
2556 2542 assert_eq!(resp.status(), StatusCode::NOT_FOUND);
2557 2543 }
2558 2544
2559 - #[test]
2560 - fn ct_eq_matches_only_identical_strings() {
2561 - assert!(ct_eq("abc", "abc"));
2562 - assert!(!ct_eq("abc", "abd"));
2563 - assert!(!ct_eq("abc", "ab"));
2564 - assert!(!ct_eq("", "x"));
2565 - assert!(ct_eq("", ""));
2566 - }
2567 -
2568 2545 /// Path-traversal guard: a `..` segment must not escape logs_root.
2569 2546 /// axum's `{name}` param already rejects a literal `/` in the value, but
2570 2547 /// `..` as a whole segment is structurally valid and must be blocked at
A shared/ops-core/src/daemon.rs +43
@@ -0,0 +1,43 @@
1 + //! Shared HTTP-daemon helpers for the operator tools (Sando, Bento).
2 + //!
3 + //! Both daemons expose a small axum service guarded by a single bearer token,
4 + //! and both compared that token the same way — a constant-time byte compare, to
5 + //! the byte, test included. That one belongs here so there is exactly one copy.
6 + //!
7 + //! Not everything the two daemons appear to share actually is: `require_bearer`
8 + //! diverged (Sando attributes each authenticated POST to its peer; Bento leaves
9 + //! reads open and logs nothing), and the `Error` → response mappers carry
10 + //! different variants and leak postures (Bento answers a leak-safe JSON
11 + //! envelope; Sando returns `to_string()` text). Folding those is a behavior
12 + //! change, not a mechanical dedup, so each daemon keeps its own — only the
13 + //! genuinely identical primitive lives here.
14 +
15 + /// Constant-time string compare for bearer tokens. The length may leak (a
16 + /// token's length is not the secret); the byte comparison does not
17 + /// short-circuit, so a matching prefix is not distinguishable by timing.
18 + #[must_use]
19 + pub fn ct_eq(a: &str, b: &str) -> bool {
20 + let (a, b) = (a.as_bytes(), b.as_bytes());
21 + if a.len() != b.len() {
22 + return false;
23 + }
24 + let mut diff = 0u8;
25 + for (x, y) in a.iter().zip(b) {
26 + diff |= x ^ y;
27 + }
28 + diff == 0
29 + }
30 +
31 + #[cfg(test)]
32 + mod tests {
33 + use super::*;
34 +
35 + #[test]
36 + fn ct_eq_matches_only_identical_strings() {
37 + assert!(ct_eq("abc", "abc"));
38 + assert!(!ct_eq("abc", "abd"));
39 + assert!(!ct_eq("abc", "ab"));
40 + assert!(!ct_eq("", "x"));
41 + assert!(ct_eq("", ""));
42 + }
43 + }
M shared/ops-core/src/lib.rs +3
@@ -18,11 +18,14 @@
18 18 //! - [`live_log`] — a disk-append + callback live-log sink that implements
19 19 //! `LogSink`; parameterized over a chunk callback so it is not tied to any
20 20 //! tool's `Event` enum.
21 + //! - [`daemon`] — small HTTP-daemon helpers both tools share verbatim (the
22 + //! constant-time bearer-token compare).
21 23 //! - [`sqlite`] — connection helper (each tool runs its own `sqlx::migrate!`).
22 24 //!
23 25 //! Design: `ops-exec-overview` in the shared code wiki (`~/Code/_private/wiki/`).
24 26 //! <!-- wiki: ops-exec-overview -->
25 27
28 + pub mod daemon;
26 29 pub mod eventbus;
27 30 pub mod live_log;
28 31 pub mod sqlite;