planarian_backend_linux/
probe.rs

1//! `ProbeBackend`: TCP, HTTP, sandboxed exec and free-space probes (§10 probe
2//! contract). A probe's timeout is a deadline for the whole probe: name
3//! resolution, connection, writing and every read count against the one
4//! budget (R06). A batch runs on a bounded pool and returns by its deadline;
5//! a probe still running then is recorded `Abandoned` (Unknown, never
6//! Failed) and abandoned; `Timeout` is a probe past its own deadline. Output is data: it is truncated and logged, never
7//! interpreted as instructions.
8//!
9//! Exec probes run under `systemd-run` with a dedicated dynamic user, no
10//! capabilities, NoNewPrivileges, a CPU quota, a runtime limit and, unless
11//! declared, no network.
12
13use crate::cmd::run_limited;
14use planarian_schema::{ProbeBackend, ProbeDef, ProbeKind, ProbeResult, ProbeSandbox, SideEffects, UnitId};
15use std::io::{Read, Write};
16use std::net::{SocketAddr, TcpStream};
17use std::sync::atomic::{AtomicBool, Ordering};
18use std::sync::{Arc, Mutex, mpsc};
19use std::time::{Duration, Instant};
20
21/// At most this many probes run at once.
22const WORKERS: usize = 16;
23/// How long past the longest probe timeout a batch waits for its stragglers.
24const BATCH_ALLOWANCE: Duration = Duration::from_secs(1);
25
26#[derive(Clone, Copy, Debug, Default)]
27pub struct Probes;
28
29fn io_result(e: std::io::Error) -> ProbeResult {
30    match e.kind() {
31        std::io::ErrorKind::TimedOut | std::io::ErrorKind::WouldBlock => ProbeResult::Timeout,
32        _ => ProbeResult::Fail(e.to_string()),
33    }
34}
35
36/// The time left before `deadline`, or `Timeout` if none is.
37fn left(deadline: Instant) -> Result<Duration, ProbeResult> {
38    match deadline.checked_duration_since(Instant::now()) {
39        Some(d) if d >= Duration::from_millis(1) => Ok(d),
40        _ => Err(ProbeResult::Timeout),
41    }
42}
43
44/// `host:port` to an address within the deadline. A literal address needs
45/// no lookup; a name is resolved by `resolver` (`getent ahosts`) as a
46/// subprocess that is killed at the deadline, so a stalled resolver costs
47/// the probe its budget and no more.
48fn resolve(addr: &str, deadline: Instant, resolver: &[&str]) -> Result<SocketAddr, ProbeResult> {
49    if let Ok(sa) = addr.parse::<SocketAddr>() {
50        return Ok(sa);
51    }
52    let (host, port) = addr.rsplit_once(':').ok_or_else(|| ProbeResult::Error(format!("{addr}: no port")))?;
53    let port: u16 = port.parse().map_err(|_| ProbeResult::Error(format!("{addr}: bad port")))?;
54    let mut argv = resolver.to_vec();
55    argv.push(host);
56    let o = run_limited(&argv, left(deadline)?, 4096);
57    if o.timed_out {
58        return Err(ProbeResult::Timeout);
59    }
60    if !o.ok() {
61        return Err(ProbeResult::Error(format!("{host}: does not resolve")));
62    }
63    let ip = o
64        .stdout
65        .lines()
66        .filter_map(|l| l.split_whitespace().next())
67        .find_map(|t| t.parse::<std::net::IpAddr>().ok())
68        .ok_or_else(|| ProbeResult::Error(format!("{host}: no address")))?;
69    Ok(SocketAddr::new(ip, port))
70}
71
72const RESOLVER: &[&str] = &["getent", "ahosts"];
73
74fn connect(addr: &str, deadline: Instant, resolver: &[&str]) -> Result<TcpStream, ProbeResult> {
75    let sa = resolve(addr, deadline, resolver)?;
76    TcpStream::connect_timeout(&sa, left(deadline)?).map_err(io_result)
77}
78
79pub fn tcp(addr: &str, timeout: Duration) -> ProbeResult {
80    match connect(addr, Instant::now() + timeout, RESOLVER) {
81        Ok(_) => ProbeResult::Pass,
82        Err(r) => r,
83    }
84}
85
86/// `http://host[:port]/path` → (host:port, host, path).
87pub fn parse_url(url: &str) -> Option<(String, String, String)> {
88    let rest = url.strip_prefix("http://")?;
89    let (hostport, path) = match rest.find('/') {
90        Some(i) => (&rest[..i], &rest[i..]),
91        None => (rest, "/"),
92    };
93    let host = hostport.split(':').next()?.to_string();
94    let addr = if hostport.contains(':') { hostport.to_string() } else { format!("{hostport}:80") };
95    Some((addr, host, path.to_string()))
96}
97
98/// The status code from an HTTP/1.x status line.
99pub fn status_of(head: &str) -> Option<u16> {
100    let line = head.lines().next()?;
101    let mut parts = line.split_whitespace();
102    if !parts.next()?.starts_with("HTTP/1.") {
103        return None;
104    }
105    parts.next()?.parse().ok()
106}
107
108pub fn http(url: &str, want: u16, timeout: Duration, bytes_max: usize) -> ProbeResult {
109    http_with(url, want, Instant::now() + timeout, bytes_max, RESOLVER)
110}
111
112fn http_with(url: &str, want: u16, deadline: Instant, bytes_max: usize, resolver: &[&str]) -> ProbeResult {
113    let Some((addr, host, path)) = parse_url(url) else {
114        return ProbeResult::Error(format!("unsupported URL {url:?} (only http:// in Phase A)"));
115    };
116    let mut s = match connect(&addr, deadline, resolver) {
117        Ok(s) => s,
118        Err(r) => return r,
119    };
120    let req =
121        format!("GET {path} HTTP/1.0\r\nHost: {host}\r\nUser-Agent: planarian-probe\r\nConnection: close\r\n\r\n");
122    match left(deadline) {
123        Ok(l) => {
124            let _ = s.set_write_timeout(Some(l));
125        }
126        Err(r) => return r,
127    }
128    if let Err(e) = s.write_all(req.as_bytes()) {
129        return io_result(e);
130    }
131    let mut buf = vec![0u8; bytes_max.clamp(64, 65536)];
132    let mut n = 0;
133    while n < buf.len() {
134        // Each read gets only what is left of the one budget, so a server
135        // that trickles bytes cannot stretch the probe.
136        match left(deadline) {
137            Ok(l) => {
138                let _ = s.set_read_timeout(Some(l));
139            }
140            Err(r) => return r,
141        }
142        match s.read(&mut buf[n..]) {
143            Ok(0) => break,
144            Ok(k) => {
145                n += k;
146                if buf[..n].windows(2).any(|w| w == b"\r\n") {
147                    break;
148                }
149            }
150            Err(e) => return io_result(e),
151        }
152    }
153    match status_of(&String::from_utf8_lossy(&buf[..n])) {
154        Some(code) if code == want => ProbeResult::Pass,
155        Some(code) => ProbeResult::Fail(format!("HTTP {code} (want {want})")),
156        None => ProbeResult::Fail("no HTTP status line".into()),
157    }
158}
159
160/// The `systemd-run` argv for an exec probe in its sandbox.
161pub fn sandbox_argv(argv: &[String], sb: &ProbeSandbox, timeout: Duration) -> Vec<String> {
162    let mut a: Vec<String> =
163        ["systemd-run", "--wait", "--pipe", "--quiet", "--collect", "--service-type=exec"].map(String::from).into();
164    if sb.uid == 0 {
165        a.push("--property=DynamicUser=yes".into());
166    } else {
167        a.push(format!("--uid={}", sb.uid));
168    }
169    a.push("--property=NoNewPrivileges=yes".into());
170    a.push(format!("--property=CapabilityBoundingSet={}", sb.caps.join(" ")));
171    a.push(format!("--property=AmbientCapabilities={}", sb.caps.join(" ")));
172    a.push(format!("--property=RuntimeMaxSec={}", timeout.as_secs().max(1)));
173    a.push(format!(
174        "--property=CPUQuota={}%",
175        (sb.cpu_ms as u64 * 100 / timeout.as_millis().max(1) as u64).clamp(1, 100)
176    ));
177    a.push("--property=ProtectSystem=strict".into());
178    a.push("--property=ProtectHome=yes".into());
179    a.push("--property=PrivateTmp=yes".into());
180    if !sb.network {
181        a.push("--property=PrivateNetwork=yes".into());
182    }
183    a.push("--".into());
184    a.extend(argv.iter().cloned());
185    a
186}
187
188/// P10: what is read is bounded by the probe's `bytes_max` as well as the
189/// sandbox's `max_output`, whichever is smaller.
190fn read_limit(sb: &ProbeSandbox, bytes_max: usize) -> usize {
191    (sb.max_output as usize).min(bytes_max)
192}
193
194pub fn exec(argv: &[String], sb: &ProbeSandbox, timeout: Duration, bytes_max: usize) -> ProbeResult {
195    let full = sandbox_argv(argv, sb, timeout);
196    let refs: Vec<&str> = full.iter().map(String::as_str).collect();
197    let o = run_limited(&refs, timeout + Duration::from_secs(2), read_limit(sb, bytes_max));
198    if o.timed_out {
199        ProbeResult::Timeout
200    } else if o.ok() {
201        ProbeResult::Pass
202    } else if o.status.is_none() {
203        ProbeResult::Error(o.tail())
204    } else {
205        ProbeResult::Fail(o.tail())
206    }
207}
208
209/// Calls `method` (`interface.Member`, no arguments) on `service` at `path` on
210/// the system bus; any reply passes. A missing name or an error reply fails.
211pub fn dbus(service: &str, path: &str, method: &str, timeout: Duration, bytes_max: usize) -> ProbeResult {
212    let Some((iface, member)) = method.rsplit_once('.') else {
213        return ProbeResult::Error(format!("method {method:?} is not interface.Member"));
214    };
215    let secs = format!("--timeout={}ms", timeout.as_millis().max(1));
216    let argv = ["busctl", "--system", "--quiet", secs.as_str(), "call", service, path, iface, member];
217    let o = run_limited(&argv, timeout + Duration::from_secs(2), bytes_max);
218    if o.timed_out {
219        ProbeResult::Timeout
220    } else if o.ok() {
221        ProbeResult::Pass
222    } else if o.status.is_none() {
223        ProbeResult::Error(o.tail())
224    } else {
225        ProbeResult::Fail(o.tail())
226    }
227}
228
229pub fn file_present(path: &str) -> ProbeResult {
230    match std::fs::symlink_metadata(path) {
231        Ok(_) => ProbeResult::Pass,
232        Err(e) if e.kind() == std::io::ErrorKind::NotFound => ProbeResult::Fail(format!("{path}: absent")),
233        Err(e) => ProbeResult::Error(format!("{path}: {e}")),
234    }
235}
236
237pub fn free_space(path: &str, min_bytes: u64, min_permille: u16) -> ProbeResult {
238    match rustix::fs::statvfs(path) {
239        Ok(s) => {
240            let free = s.f_bavail * s.f_frsize;
241            let size = s.f_blocks * s.f_frsize;
242            if free >= min_bytes && (size == 0 || free * 1000 >= size * min_permille as u64) {
243                ProbeResult::Pass
244            } else {
245                ProbeResult::Fail(format!("{} MB free of {} MB on {path}", free >> 20, size >> 20))
246            }
247        }
248        Err(e) => ProbeResult::Error(format!("{path}: {e}")),
249    }
250}
251
252fn run_one(p: &ProbeDef) -> ProbeResult {
253    let timeout = Duration::from_millis(p.timeout.0.max(1));
254    let bytes_max = match p.side_effects {
255        SideEffects::Measurement { bytes_max, .. } => bytes_max as usize,
256        _ => 4096,
257    };
258    match &p.kind {
259        ProbeKind::Tcp { addr } => tcp(addr, timeout),
260        ProbeKind::Http { url, status } => http(url, *status, timeout, bytes_max),
261        ProbeKind::Exec { argv, sandbox } => exec(argv, sandbox, timeout, bytes_max),
262        ProbeKind::DBus { service, path, method } => dbus(service, path, method, timeout, bytes_max),
263        ProbeKind::FilePresent { path } => file_present(path),
264        ProbeKind::FreeSpace { path, min_bytes, min_permille } => free_space(path, *min_bytes, *min_permille),
265        ProbeKind::SystemdActive | ProbeKind::SessionTarget { .. } => {
266            ProbeResult::Error("evaluated by state, not the probe backend".into())
267        }
268    }
269}
270
271/// Run `jobs` on at most `WORKERS` threads and return by the batch deadline
272/// (the longest timeout plus `allowance`): a probe not finished by then, or
273/// not started, is `Abandoned` (§10: recorded Unknown and abandoned), and its
274/// thread is never waited for.
275fn run_batch_with(
276    jobs: &[(UnitId, ProbeDef)],
277    run: fn(&ProbeDef) -> ProbeResult,
278    allowance: Duration,
279) -> Vec<ProbeResult> {
280    let longest = jobs.iter().map(|(_, p)| p.timeout.0.max(1)).max().unwrap_or(0);
281    let deadline = Instant::now() + Duration::from_millis(longest) + allowance;
282    let queue: Arc<Mutex<Vec<(usize, ProbeDef)>>> =
283        Arc::new(Mutex::new(jobs.iter().map(|(_, p)| p.clone()).enumerate().rev().collect()));
284    let over = Arc::new(AtomicBool::new(false));
285    let (tx, rx) = mpsc::channel();
286    for _ in 0..jobs.len().min(WORKERS) {
287        let (queue, over, tx) = (queue.clone(), over.clone(), tx.clone());
288        std::thread::spawn(move || {
289            while !over.load(Ordering::Relaxed) {
290                let Some((i, p)) = queue.lock().ok().and_then(|mut q| q.pop()) else { break };
291                let r = std::panic::catch_unwind(|| run(&p))
292                    .unwrap_or_else(|_| ProbeResult::Error("probe panicked".into()));
293                if tx.send((i, r)).is_err() {
294                    break;
295                }
296            }
297        });
298    }
299    drop(tx);
300    let mut out: Vec<Option<ProbeResult>> = vec![None; jobs.len()];
301    let mut got = 0;
302    while got < jobs.len() {
303        let Some(wait) = deadline.checked_duration_since(Instant::now()) else { break };
304        match rx.recv_timeout(wait) {
305            Ok((i, r)) => {
306                out[i] = Some(r);
307                got += 1;
308            }
309            Err(_) => break,
310        }
311    }
312    over.store(true, Ordering::Relaxed);
313    out.into_iter().map(|r| r.unwrap_or(ProbeResult::Abandoned)).collect()
314}
315
316impl ProbeBackend for Probes {
317    fn run_batch(&self, jobs: &[(UnitId, ProbeDef)]) -> Vec<ProbeResult> {
318        run_batch_with(jobs, run_one, BATCH_ALLOWANCE)
319    }
320}
321
322#[cfg(test)]
323mod tests {
324    /// P10: an exec probe reads no more than its bytes_max, even under a
325    /// larger sandbox max_output; a dbus probe's cap is its bytes_max.
326    #[test]
327    fn every_reader_is_bounded_by_bytes_max() {
328        let sb = ProbeSandbox { uid: 0, caps: vec![], max_output: 65536, cpu_ms: 500, network: false };
329        assert_eq!(read_limit(&sb, 4096), 4096);
330        let small = ProbeSandbox { max_output: 1024, ..sb };
331        assert_eq!(read_limit(&small, 4096), 1024);
332        let o = run_limited(&["sh", "-c", "head -c 100000 /dev/zero"], Duration::from_secs(5), 4096);
333        assert!(o.stdout.len() <= 4096, "{}", o.stdout.len());
334    }
335
336    use super::*;
337    use std::net::TcpListener;
338
339    #[test]
340    fn file_present_passes_and_fails() {
341        let d = std::env::temp_dir().join(format!("planarian-fp-{}", std::process::id()));
342        let path = d.to_str().unwrap();
343        assert!(matches!(file_present(path), ProbeResult::Fail(_)));
344        std::fs::write(&d, b"").unwrap();
345        assert_eq!(file_present(path), ProbeResult::Pass);
346        std::fs::remove_file(&d).unwrap();
347    }
348
349    #[test]
350    fn url_and_status() {
351        assert_eq!(
352            parse_url("http://127.0.0.1:8080/health"),
353            Some(("127.0.0.1:8080".into(), "127.0.0.1".into(), "/health".into()))
354        );
355        assert_eq!(parse_url("http://example"), Some(("example:80".into(), "example".into(), "/".into())));
356        assert_eq!(parse_url("https://x/"), None);
357        assert_eq!(status_of("HTTP/1.1 503 Service Unavailable\r\n"), Some(503));
358        assert_eq!(status_of("garbage"), None);
359    }
360
361    #[test]
362    fn tcp_and_http_against_a_local_server() {
363        let l = TcpListener::bind("127.0.0.1:0").unwrap();
364        let addr = l.local_addr().unwrap();
365        std::thread::spawn(move || {
366            for (i, s) in l.incoming().enumerate() {
367                let mut s = s.unwrap();
368                let mut buf = [0u8; 512];
369                let _ = s.read(&mut buf);
370                let body = if i == 0 { "HTTP/1.1 200 OK\r\n\r\nok" } else { "HTTP/1.1 503 Busy\r\n\r\n" };
371                let _ = s.write_all(body.as_bytes());
372            }
373        });
374        let url = format!("http://{addr}/health");
375        assert_eq!(http(&url, 200, Duration::from_secs(2), 4096), ProbeResult::Pass);
376        assert!(matches!(http(&url, 200, Duration::from_secs(2), 4096), ProbeResult::Fail(m) if m.contains("503")));
377        assert_eq!(tcp(&addr.to_string(), Duration::from_secs(1)), ProbeResult::Pass);
378        // Port 1 is closed in any test environment; a freshly released
379        // ephemeral port is not (another test may have taken it).
380        assert!(matches!(tcp("127.0.0.1:1", Duration::from_secs(1)), ProbeResult::Fail(_)));
381    }
382
383    /// R06: a server that sends one byte just before each read would time
384    /// out still costs the probe no more than its one deadline.
385    #[test]
386    fn trickling_http_has_one_total_deadline() {
387        let l = TcpListener::bind("127.0.0.1:0").unwrap();
388        let addr = l.local_addr().unwrap();
389        std::thread::spawn(move || {
390            for s in l.incoming() {
391                let mut s = s.unwrap();
392                let mut buf = [0u8; 512];
393                let _ = s.read(&mut buf);
394                for _ in 0..100 {
395                    if s.write_all(b"H").is_err() {
396                        break;
397                    }
398                    std::thread::sleep(Duration::from_millis(150));
399                }
400            }
401        });
402        let t = Instant::now();
403        let r = http(&format!("http://{addr}/health"), 200, Duration::from_millis(700), 4096);
404        assert_eq!(r, ProbeResult::Timeout);
405        assert!(t.elapsed() < Duration::from_millis(1200), "took {:?}", t.elapsed());
406    }
407
408    /// R06: name resolution counts against the probe's deadline; a stalled
409    /// resolver is killed at it.
410    #[test]
411    fn a_stalled_resolver_is_inside_the_deadline() {
412        let t = Instant::now();
413        let stalled = ["sh", "-c", "sleep 5", "resolver"];
414        let r =
415            http_with("http://gym.invalid/health", 200, Instant::now() + Duration::from_millis(400), 4096, &stalled);
416        assert_eq!(r, ProbeResult::Timeout);
417        assert!(t.elapsed() < Duration::from_millis(1000), "took {:?}", t.elapsed());
418        // A resolver that answers is used; a literal address needs none.
419        let answers = ["sh", "-c", "echo '127.0.0.1 STREAM localhost'", "resolver"];
420        assert_eq!(
421            resolve("localhost:80", Instant::now() + Duration::from_secs(2), &answers),
422            Ok("127.0.0.1:80".parse().unwrap())
423        );
424        assert_eq!(resolve("10.0.0.1:443", Instant::now(), &stalled), Ok("10.0.0.1:443".parse().unwrap()));
425    }
426
427    fn job(id: &str, timeout_ms: u64) -> (UnitId, ProbeDef) {
428        use planarian_schema::{Cadence, Millis, ProbeCost, ProbeId, ProbeOrigin, ProbeRole};
429        (
430            UnitId::service("gym-web"),
431            ProbeDef {
432                id: ProbeId(id.into()),
433                kind: ProbeKind::FilePresent { path: "/".into() },
434                timeout: Millis(timeout_ms),
435                origin: ProbeOrigin::User,
436                role: ProbeRole::Readiness,
437                cadence: Cadence::EveryTick,
438                cost: ProbeCost::Trivial,
439                side_effects: SideEffects::None,
440                establishes: String::new(),
441            },
442        )
443    }
444
445    /// R06: the batch returns by its deadline; a wedged probe is recorded
446    /// Abandoned (Unknown), not Timeout, and the others keep their results.
447    #[test]
448    fn a_wedged_probe_does_not_hold_the_batch() {
449        fn run(p: &ProbeDef) -> ProbeResult {
450            if p.id.0 == "wedged" {
451                std::thread::sleep(Duration::from_secs(10));
452            }
453            ProbeResult::Pass
454        }
455        let jobs = vec![job("fast", 200), job("wedged", 200), job("also-fast", 200)];
456        let t = Instant::now();
457        let r = run_batch_with(&jobs, run, Duration::from_millis(300));
458        assert!(t.elapsed() < Duration::from_millis(900), "took {:?}", t.elapsed());
459        assert_eq!(r, vec![ProbeResult::Pass, ProbeResult::Abandoned, ProbeResult::Pass]);
460        // More jobs than workers still all run.
461        let many: Vec<_> = (0..WORKERS * 2 + 3).map(|i| job(&format!("p{i}"), 500)).collect();
462        assert!(run_batch_with(&many, run, Duration::from_millis(500)).iter().all(|r| *r == ProbeResult::Pass));
463    }
464
465    /// R06, §10: `Timeout` is a probe past its own deadline; a probe still
466    /// queued behind wedged ones when the batch deadline passes is
467    /// `Abandoned`, not blamed with a `Timeout` it never had.
468    #[test]
469    fn a_straggler_is_abandoned_and_its_own_timeout_stays_timeout() {
470        fn run(p: &ProbeDef) -> ProbeResult {
471            match p.id.0.as_str() {
472                "own-timeout" => ProbeResult::Timeout,
473                id if id.starts_with("wedged") => {
474                    std::thread::sleep(Duration::from_secs(10));
475                    ProbeResult::Pass
476                }
477                _ => ProbeResult::Pass,
478            }
479        }
480        // Every worker wedged, then two that never start.
481        let mut jobs = vec![job("own-timeout", 200)];
482        jobs.extend((0..WORKERS).map(|i| job(&format!("wedged{i}"), 200)));
483        jobs.push(job("queued", 200));
484        let t = Instant::now();
485        let r = run_batch_with(&jobs, run, Duration::from_millis(300));
486        assert!(t.elapsed() < Duration::from_millis(900), "took {:?}", t.elapsed());
487        assert_eq!(r[0], ProbeResult::Timeout, "its own deadline");
488        assert!(r[1..].iter().all(|r| *r == ProbeResult::Abandoned), "{r:?}");
489    }
490
491    #[test]
492    fn sandbox_and_free_space() {
493        let sb = ProbeSandbox { uid: 0, caps: vec![], max_output: 4096, cpu_ms: 500, network: false };
494        let a = sandbox_argv(&["true".into()], &sb, Duration::from_secs(10));
495        assert!(a.contains(&"--property=DynamicUser=yes".to_string()));
496        assert!(a.contains(&"--property=PrivateNetwork=yes".to_string()));
497        assert_eq!(a.last().unwrap(), "true");
498        assert_eq!(free_space("/", 0, 0), ProbeResult::Pass);
499        assert!(matches!(free_space("/", u64::MAX, 0), ProbeResult::Fail(_)));
500    }
501}
planarian_backend_linux/probe.rs

planarian_backend_linux/
probe.rs
