Skip to main content

max / makenotwork

pom: cap peer response bodies and re-project mesh into a fixed schema Unbounded peer JSON: the heartbeat loop buffered peer responses with a bare resp.json(), so a hostile or MITM'd peer could return a multi-GB body that the 10s timeout doesn't bound (fuzz-2026-07-06). read_capped_json enforces the same 10 MB ceiling the health probe uses (content-length pre-check + post-read cap). Mesh poisoning: /api/mesh re-served each peer's cached status_data verbatim, so a compromised peer injected arbitrary structure into mesh consumers. reproject_peer_status extracts only the known fields (instance/targets/peers) into a fixed schema — the values remain the peer's to report, but the shape is ours, and terminal rendering already scrubs the values. Transport note: forcing https:// peers is out of scope here — pom's own API binds plain HTTP and the mesh runs over tailnet (WireGuard encrypts the hop); requiring TLS would break the deployment without a fronting proxy. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-07 16:06 UTC
Commit: 0f242432abe4283e9e6b739ca9dadf198ec53c8e
Parent: a74a90a
2 files changed, +108 insertions, -3 deletions
M pom/src/api.rs +81 -1
@@ -593,7 +593,13 @@ async fn mesh_view(
593 593 instances.insert(instance.name.clone(), self_entry);
594 594
595 595 for (name, status_data, fallback) in peer_entries {
596 - instances.insert(name, status_data.unwrap_or(fallback));
596 + // Re-project the peer's cached status into a FIXED schema instead of
597 + // re-serving its raw JSON verbatim: a compromised/MITM'd peer otherwise
598 + // injects arbitrary structure into /api/mesh consumers (mesh poisoning,
599 + // fuzz-2026-07-06 #6). Values are still the peer's to report; the shape is
600 + // ours. Terminal rendering additionally scrubs the values (display::scrub).
601 + let entry = status_data.as_ref().map(reproject_peer_status).unwrap_or(fallback);
602 + instances.insert(name, entry);
597 603 }
598 604
599 605 Ok(Json(serde_json::json!({
@@ -601,6 +607,53 @@ async fn mesh_view(
601 607 })))
602 608 }
603 609
610 + /// Extract only the known fields from an untrusted peer's cached status blob,
611 + /// dropping any attacker-injected extra structure. Preserves exactly the paths
612 + /// the mesh consumers (`display::format_mesh`, the dashboard) read.
613 + fn reproject_peer_status(raw: &serde_json::Value) -> serde_json::Value {
614 + let instance = raw.get("instance").map(|i| {
615 + serde_json::json!({
616 + "id": i.get("id").and_then(|v| v.as_str()),
617 + "name": i.get("name").and_then(|v| v.as_str()),
618 + "version": i.get("version").and_then(|v| v.as_str()),
619 + "started_at": i.get("started_at").and_then(|v| v.as_str()),
620 + "targets": i.get("targets")
621 + .and_then(|v| v.as_array())
622 + .map(|a| a.iter().filter_map(|t| t.as_str()).collect::<Vec<_>>())
623 + .unwrap_or_default(),
624 + })
625 + });
626 +
627 + let project_map = |key: &str, f: &dyn Fn(&serde_json::Value) -> serde_json::Value| {
628 + let mut out = serde_json::Map::new();
629 + if let Some(obj) = raw.get(key).and_then(|v| v.as_object()) {
630 + for (k, v) in obj {
631 + out.insert(k.clone(), f(v));
632 + }
633 + }
634 + serde_json::Value::Object(out)
635 + };
636 +
637 + let targets = project_map("targets", &|t| {
638 + serde_json::json!({
639 + "label": t.get("label").and_then(|v| v.as_str()),
640 + "status": t.get("status").and_then(|v| v.as_str()),
641 + "response_time_ms": t.get("response_time_ms").and_then(|v| v.as_i64()),
642 + "checked_at": t.get("checked_at").and_then(|v| v.as_str()),
643 + })
644 + });
645 +
646 + let peers = project_map("peers", &|p| {
647 + serde_json::json!({
648 + "status": p.get("status").and_then(|v| v.as_str()),
649 + "last_seen": p.get("last_seen").and_then(|v| v.as_str()),
650 + "latency_ms": p.get("latency_ms").and_then(|v| v.as_u64()),
651 + })
652 + });
653 +
654 + serde_json::json!({ "instance": instance, "targets": targets, "peers": peers })
655 + }
656 +
604 657 // --- Trends endpoint ---
605 658
606 659 #[derive(Serialize)]
@@ -811,6 +864,33 @@ mod tests {
811 864 std::net::IpAddr::V4(std::net::Ipv4Addr::new(10, 0, 0, n))
812 865 }
813 866
867 + #[test]
868 + fn reproject_drops_injected_structure() {
869 + // #6: a compromised peer stuffs extra structure into its status blob.
870 + let hostile = serde_json::json!({
871 + "instance": { "id": "abc", "version": "9.9", "evil_field": {"x": 1} },
872 + "targets": { "mnw": { "status": "operational", "response_time_ms": 5, "evil": "inject" } },
873 + "peers": { "p2": { "status": "up", "latency_ms": 3 } },
874 + "top_level_injection": [1, 2, 3]
875 + });
876 + let clean = reproject_peer_status(&hostile);
877 +
878 + // Only the fixed top-level keys survive.
879 + let obj = clean.as_object().unwrap();
880 + let mut keys: Vec<&String> = obj.keys().collect();
881 + keys.sort();
882 + assert_eq!(keys, vec!["instance", "peers", "targets"]);
883 + assert!(clean.get("top_level_injection").is_none());
884 +
885 + // Known values are preserved; injected sibling keys are gone.
886 + assert_eq!(clean["instance"]["id"], "abc");
887 + assert_eq!(clean["instance"]["version"], "9.9");
888 + assert!(clean["instance"].get("evil_field").is_none());
889 + assert_eq!(clean["targets"]["mnw"]["status"], "operational");
890 + assert!(clean["targets"]["mnw"].get("evil").is_none());
891 + assert_eq!(clean["peers"]["p2"]["latency_ms"], 3);
892 + }
893 +
814 894 /// Build a GET request carrying a `ConnectInfo<SocketAddr>` extension, which
815 895 /// the real server injects via `into_make_service_with_connect_info` but
816 896 /// `oneshot` does not — the rate-limit layer extracts it.
M pom/src/peer.rs +27 -2
@@ -11,6 +11,31 @@ use crate::alerts::Alerter;
11 11 use crate::config::PeerConfig;
12 12 use crate::error::{PomError, Result};
13 13
14 + /// Cap on a peer response body (matches the health probe's `MAX_RESPONSE_BYTES`).
15 + /// A hostile or MITM'd peer could otherwise return a multi-GB body that
16 + /// `resp.json()` buffers entirely in RAM — the 10s timeout bounds wall-clock, not
17 + /// bytes (fuzz-2026-07-06 unbounded peer JSON). Holds regardless of transport.
18 + const MAX_PEER_BODY_BYTES: u64 = 10 * 1024 * 1024;
19 +
20 + /// Deserialize a peer response as JSON with a hard body-size cap.
21 + async fn read_capped_json<T: serde::de::DeserializeOwned>(
22 + peer_name: &str,
23 + resp: reqwest::Response,
24 + ) -> Option<T> {
25 + if let Some(len) = resp.content_length()
26 + && len > MAX_PEER_BODY_BYTES
27 + {
28 + tracing::warn!("{peer_name}: peer response declares {len} bytes (cap {MAX_PEER_BODY_BYTES}); dropping");
29 + return None;
30 + }
31 + let bytes = resp.bytes().await.ok()?;
32 + if bytes.len() as u64 > MAX_PEER_BODY_BYTES {
33 + tracing::warn!("{peer_name}: peer response body {} bytes exceeded cap; dropping", bytes.len());
34 + return None;
35 + }
36 + serde_json::from_slice(&bytes).ok()
37 + }
38 +
14 39 // --- Identity ---
15 40
16 41 /// Load or create a persistent instance ID (UUID v4).
@@ -212,7 +237,7 @@ async fn heartbeat_loop(
212 237
213 238 match result.and_then(|r| r.error_for_status()) {
214 239 Ok(response) => {
215 - let info: Option<InstanceInfo> = response.json().await.ok();
240 + let info: Option<InstanceInfo> = read_capped_json(peer_name, response).await;
216 241 handle_heartbeat_success(peer_name, &mesh, &pool, info, latency_ms, &alerter).await;
217 242 }
218 243 Err(_e) => {
@@ -228,7 +253,7 @@ async fn heartbeat_loop(
228 253 let status_result = status_req.send().await;
229 254 match status_result {
230 255 Ok(resp) => {
231 - if let Ok(data) = resp.json::<serde_json::Value>().await {
256 + if let Some(data) = read_capped_json::<serde_json::Value>(peer_name, resp).await {
232 257 let mut state = mesh.write().await;
233 258 if let Some(peer) = state.peers.get_mut(peer_name) {
234 259 peer.status_data = Some(data);