Skip to main content

max / makenotwork

46.4 KB · 1327 lines History Blame Raw
1 //! HTTP API for serve mode, exposes health check data to consumers like MNW.
2
3 use std::collections::{HashMap, HashSet};
4 use std::sync::Arc;
5
6 use axum::extract::{Path, Request, State as AxumState};
7 use axum::http::StatusCode;
8 use axum::middleware::{self, Next};
9 use axum::response::IntoResponse;
10 use axum::routing::get;
11 use axum::{Json, Router};
12 use serde::Serialize;
13 use tracing::instrument;
14
15 use crate::checks::drift::{compute_test_staleness, detect_test_duration_drift};
16 use crate::config::Config;
17 use crate::db;
18 use crate::peer::SharedMeshState;
19 use crate::types::{HealthSnapshot, LatencyBucket, LatencyStats, TestStaleness};
20
21 /// Fixed-window rate limiter, keyed per client IP.
22 ///
23 /// A single global counter would let one noisy source burn the whole 60/min
24 /// budget for everyone. Auth runs before this layer, so an unauthenticated
25 /// attacker cannot reach the limiter at all, and an authenticated client's
26 /// bucket is isolated to its own IP.
27 #[derive(Clone)]
28 pub struct PerIpRateLimiter {
29 windows: Arc<std::sync::Mutex<HashMap<std::net::IpAddr, (std::time::Instant, u64)>>>,
30 max_per_window: u64,
31 window_duration: std::time::Duration,
32 }
33
34 impl PerIpRateLimiter {
35 pub fn new(max_per_window: u64, window_duration: std::time::Duration) -> Self {
36 Self {
37 windows: Arc::new(std::sync::Mutex::new(HashMap::new())),
38 max_per_window,
39 window_duration,
40 }
41 }
42
43 pub fn try_acquire(&self, ip: std::net::IpAddr) -> bool {
44 let now = std::time::Instant::now();
45 let mut windows = self.windows.lock().unwrap();
46 // Opportunistically evict stale buckets so the map can't grow unbounded
47 // from transient/spoofed source IPs.
48 windows.retain(|_, (start, _)| now.duration_since(*start) <= self.window_duration);
49 let (start, count) = windows.entry(ip).or_insert((now, 0));
50 if now.duration_since(*start) > self.window_duration {
51 *start = now;
52 *count = 1;
53 true
54 } else {
55 *count += 1;
56 *count <= self.max_per_window
57 }
58 }
59 }
60
61 /// Mint a fresh random dashboard session secret (128 bits, hex). Ephemeral: it
62 /// lives only for the process, so a restart invalidates any leaked cookie.
63 pub(crate) fn mint_dashboard_token() -> String {
64 uuid::Uuid::new_v4().simple().to_string()
65 }
66
67 /// Shared state for the API server.
68 #[derive(Clone)]
69 pub struct ApiState {
70 pub pool: sqlx::SqlitePool,
71 pub config: Arc<Config>,
72 pub mesh: Option<SharedMeshState>,
73 pub rate_limiter: PerIpRateLimiter,
74 /// Ephemeral, per-process dashboard session secret. `Some` only when the
75 /// dashboard is enabled. Handed to the browser as an httpOnly cookie (never
76 /// embedded in page JS, never the long-lived `api_token`), and accepted by
77 /// [`require_bearer_token`] for same-origin dashboard `/api/*` calls.
78 pub dashboard_token: Option<Arc<str>>,
79 }
80
81 /// Rate limiting middleware. Returns 429 if the request rate exceeds the limit.
82 /// Runs *inside* the auth layer, so only authenticated requests are counted and
83 /// each client IP has its own bucket.
84 async fn rate_limit(
85 AxumState(state): AxumState<ApiState>,
86 axum::extract::ConnectInfo(peer): axum::extract::ConnectInfo<std::net::SocketAddr>,
87 req: Request,
88 next: Next,
89 ) -> impl IntoResponse {
90 if state.rate_limiter.try_acquire(peer.ip()) {
91 Ok(next.run(req).await)
92 } else {
93 Err((
94 StatusCode::TOO_MANY_REQUESTS,
95 Json(serde_json::json!({
96 "error": "rate limit exceeded"
97 })),
98 ))
99 }
100 }
101
102 /// Read the `pom_dash` session cookie from a request, if present.
103 fn dashboard_cookie(req: &Request) -> Option<String> {
104 let cookies = req
105 .headers()
106 .get(axum::http::header::COOKIE)?
107 .to_str()
108 .ok()?;
109 cookies.split(';').find_map(|kv| {
110 let (k, v) = kv.split_once('=')?;
111 (k.trim() == "pom_dash").then(|| v.trim().to_string())
112 })
113 }
114
115 /// Bearer token authentication middleware.
116 /// If `api_token` is configured, requires `Authorization: Bearer <token>` on every request.
117 /// If no token is configured, all requests pass through.
118 async fn require_bearer_token(
119 AxumState(state): AxumState<ApiState>,
120 req: Request,
121 next: Next,
122 ) -> impl IntoResponse {
123 let expected = state.config.serve.api_token.as_deref();
124 let Some(expected) = expected else {
125 return Ok(next.run(req).await);
126 };
127
128 use subtle::ConstantTimeEq;
129
130 // Accept the same-origin dashboard session cookie (httpOnly, ephemeral) as an
131 // alternative to the bearer token, so the dashboard never has to embed the
132 // long-lived api_token in page JS (fuzz-2026-07-06 SERIOUS #4).
133 if let (Some(dash), Some(cookie)) = (state.dashboard_token.as_deref(), dashboard_cookie(&req))
134 && cookie.as_bytes().ct_eq(dash.as_bytes()).into()
135 {
136 return Ok(next.run(req).await);
137 }
138
139 let auth_header = req
140 .headers()
141 .get("authorization")
142 .and_then(|v| v.to_str().ok());
143 match auth_header {
144 Some(header) if header.starts_with("Bearer ") => {
145 let token = &header[7..];
146 // Constant-time comparison to prevent timing side-channels
147 if token.as_bytes().ct_eq(expected.as_bytes()).into() {
148 Ok(next.run(req).await)
149 } else {
150 Err((
151 StatusCode::UNAUTHORIZED,
152 Json(serde_json::json!({
153 "error": "invalid bearer token"
154 })),
155 ))
156 }
157 }
158 _ => Err((
159 StatusCode::UNAUTHORIZED,
160 Json(serde_json::json!({
161 "error": "missing or malformed Authorization header"
162 })),
163 )),
164 }
165 }
166
167 /// `GET /api/health`: simple health endpoint for PoM itself.
168 /// Not behind auth, allows external monitoring without credentials.
169 #[instrument]
170 async fn self_health() -> impl IntoResponse {
171 Json(serde_json::json!({
172 "status": "operational",
173 "version": env!("CARGO_PKG_VERSION"),
174 }))
175 }
176
177 /// Build the axum router for the PoM API.
178 pub fn router(pool: sqlx::SqlitePool, config: Config, mesh: Option<SharedMeshState>) -> Router {
179 // Ephemeral dashboard session secret, minted once per process when the
180 // dashboard is enabled, never the api_token, never embedded in page JS.
181 let dashboard_token: Option<Arc<str>> = config
182 .serve
183 .dashboard
184 .then(|| Arc::from(crate::api::mint_dashboard_token().as_str()));
185
186 let state = ApiState {
187 pool,
188 config: Arc::new(config),
189 mesh,
190 rate_limiter: PerIpRateLimiter::new(60, std::time::Duration::from_mins(1)),
191 dashboard_token,
192 };
193
194 // Authenticated routes. Order matters: `.layer` wraps outward, so listing
195 // `require_bearer_token` LAST makes it the OUTERMOST layer, auth runs before
196 // rate_limit, so an unauthenticated request is rejected before it can consume
197 // any rate-limit budget (fuzz-2026-07-06 SERIOUS #5).
198 let authenticated = Router::new()
199 .route("/status.json", get(status_json))
200 .route("/api/status", get(status_all))
201 .route("/api/status/{target}", get(status_target))
202 .route("/api/trends/{target}", get(trends))
203 .route("/api/versions", get(versions))
204 .route("/api/peer/info", get(peer_info))
205 .route("/api/peer/status", get(peer_status))
206 .route("/api/mesh", get(mesh_view))
207 .layer(middleware::from_fn_with_state(state.clone(), rate_limit))
208 .layer(middleware::from_fn_with_state(
209 state.clone(),
210 require_bearer_token,
211 ));
212
213 // Public routes (no auth, no rate limit)
214 let public = Router::new().route("/api/health", get(self_health));
215
216 let mut app = public.merge(authenticated);
217
218 if state.config.serve.dashboard {
219 app = app.route("/", get(crate::dashboard::dashboard_handler));
220 }
221
222 app.with_state(state)
223 }
224
225 // Response types
226
227 #[derive(Serialize)]
228 struct StatusResponse {
229 /// Per-target status summaries, keyed by target config name.
230 targets: HashMap<String, TargetStatus>,
231 }
232
233 #[derive(Serialize)]
234 struct TargetStatus {
235 /// Human-readable display label for this target.
236 label: String,
237 /// Most recent health check snapshot. `None` if no checks have been recorded yet.
238 latest: Option<SnapshotJson>,
239 /// Last 10 health check snapshots, most recent first.
240 recent: Vec<SnapshotJson>,
241 /// Uptime percentage over the last 24 hours. `None` if no checks in that window.
242 uptime_24h: Option<f64>,
243 /// Uptime percentage over the last 7 days. `None` if no checks in that window.
244 uptime_7d: Option<f64>,
245 /// Latency statistics over the last 24 hours. Omitted if no operational checks exist.
246 #[serde(skip_serializing_if = "Option::is_none")]
247 latency_24h: Option<LatencyStats>,
248 /// Latest TLS certificate check result. Omitted if TLS monitoring is not configured.
249 #[serde(skip_serializing_if = "Option::is_none")]
250 tls: Option<db::TlsCheckRow>,
251 /// Test staleness assessment. Omitted if test running is not configured for this target.
252 #[serde(skip_serializing_if = "Option::is_none")]
253 test_staleness: Option<TestStaleness>,
254 /// Currently open incident. Omitted if the target is not in an incident state.
255 #[serde(skip_serializing_if = "Option::is_none")]
256 current_incident: Option<db::IncidentRow>,
257 /// Recent resolved and open incidents (up to 10). Omitted if empty.
258 #[serde(skip_serializing_if = "Vec::is_empty")]
259 incidents: Vec<db::IncidentRow>,
260 /// Latest route check results per path. Omitted if empty.
261 #[serde(skip_serializing_if = "Vec::is_empty")]
262 route_status: Vec<RouteStatusJson>,
263 /// Latest DNS check results. Omitted if empty.
264 #[serde(skip_serializing_if = "Vec::is_empty")]
265 dns_status: Vec<DnsStatusJson>,
266 /// Latest WHOIS check result. Omitted if no WHOIS monitoring is configured.
267 #[serde(skip_serializing_if = "Option::is_none")]
268 whois: Option<db::WhoisCheckRow>,
269 /// Test duration drift warning. Omitted if no drift detected or no test config.
270 #[serde(skip_serializing_if = "Option::is_none")]
271 test_duration_drift: Option<String>,
272 }
273
274 #[derive(Serialize)]
275 struct DnsStatusJson {
276 name: String,
277 record_type: String,
278 expected: Vec<String>,
279 actual: Vec<String>,
280 matches: bool,
281 checked_at: String,
282 }
283
284 #[derive(Serialize)]
285 struct RouteStatusJson {
286 path: String,
287 status_code: i64,
288 ok: bool,
289 checked_at: String,
290 response_time_ms: i64,
291 }
292
293 #[derive(Serialize)]
294 struct SnapshotJson {
295 /// Health status as a lowercase string (e.g. "operational", "degraded").
296 status: String,
297 /// Timestamp of the check in RFC 3339 format.
298 checked_at: String,
299 /// Round-trip response time in milliseconds.
300 response_time_ms: i64,
301 /// Structured health details from the endpoint. Omitted when unavailable.
302 #[serde(skip_serializing_if = "Option::is_none")]
303 details: Option<serde_json::Value>,
304 /// Error message if the check failed. Omitted on success.
305 #[serde(skip_serializing_if = "Option::is_none")]
306 error: Option<String>,
307 }
308
309 impl From<HealthSnapshot> for SnapshotJson {
310 fn from(s: HealthSnapshot) -> Self {
311 Self {
312 status: s.status.to_string(),
313 checked_at: s.checked_at,
314 response_time_ms: s.response_time_ms,
315 details: s
316 .details
317 .map(|d| serde_json::to_value(d).unwrap_or_default()),
318 error: s.error,
319 }
320 }
321 }
322
323 /// Build a `TargetStatus` for a single target.
324 #[instrument(skip_all, fields(target = %name))]
325 async fn build_target_status(
326 pool: &sqlx::SqlitePool,
327 name: &str,
328 label: &str,
329 config: &Config,
330 ) -> TargetStatus {
331 let recent = db::get_health_history(pool, Some(name), 10)
332 .await
333 .unwrap_or_default();
334
335 // Extract the version info we need before consuming the snapshots.
336 let latest_version = recent
337 .first()
338 .and_then(|s| s.details.as_ref())
339 .and_then(|d| d.version.clone());
340 let latest = recent.first().cloned().map(SnapshotJson::from);
341 let recent_json: Vec<SnapshotJson> = recent.into_iter().map(SnapshotJson::from).collect();
342
343 let uptime_24h = db::get_uptime_percent(pool, name, 24).await.unwrap_or(None);
344 let uptime_7d = db::get_uptime_percent(pool, name, 168)
345 .await
346 .unwrap_or(None);
347
348 // Compute 24h latency stats from operational checks
349 let latency_24h = {
350 let cutoff = (chrono::Utc::now() - chrono::Duration::hours(24)).to_rfc3339();
351 let times = db::get_response_times(pool, name, &cutoff)
352 .await
353 .unwrap_or_default();
354 let operational_times: Vec<i64> = times
355 .iter()
356 .filter(|(_, ms)| *ms > 0)
357 .map(|(_, ms)| *ms)
358 .collect();
359 LatencyStats::from_times(&operational_times)
360 };
361
362 let tls = db::get_latest_tls_check(pool, name).await.unwrap_or(None);
363
364 // Compute test staleness for targets with test config
365 let test_staleness = if let Some(target_config) = config.get_target(name)
366 && let Some(tests_config) = &target_config.tests
367 {
368 let current_version = latest_version.clone();
369
370 let latest_test = db::get_latest_test_run(pool, name).await.unwrap_or(None);
371
372 let tested_version = if let Some(ref test) = latest_test {
373 db::get_version_at_time(pool, name, &test.started_at)
374 .await
375 .unwrap_or(None)
376 } else {
377 None
378 };
379
380 let staleness = compute_test_staleness(
381 current_version.as_deref(),
382 tested_version.as_deref(),
383 latest_test.as_ref().map(|t| t.started_at.as_str()),
384 tests_config.staleness_days,
385 );
386 Some(staleness)
387 } else {
388 None
389 };
390
391 // Compute test duration drift for targets with test config
392 let test_duration_drift = if config
393 .get_target(name)
394 .and_then(|t| t.tests.as_ref())
395 .is_some()
396 {
397 let durations = db::get_test_durations(pool, name, 13)
398 .await
399 .unwrap_or_default();
400 detect_test_duration_drift(&durations, 10, 3, 1.5)
401 } else {
402 None
403 };
404
405 let current_incident = db::get_open_incident(pool, name).await.unwrap_or(None);
406
407 let incidents = db::get_recent_incidents(pool, name, 10)
408 .await
409 .unwrap_or_default();
410
411 let route_checks = db::get_latest_route_checks(pool, name)
412 .await
413 .unwrap_or_default();
414 let expected_routes: HashSet<&str> = config
415 .get_target(name)
416 .map(|t| {
417 t.expected_routes
418 .iter()
419 .map(std::string::String::as_str)
420 .collect()
421 })
422 .unwrap_or_default();
423 let route_status: Vec<RouteStatusJson> = route_checks
424 .into_iter()
425 .filter(|r| expected_routes.contains(r.path.as_str()))
426 .map(|r| RouteStatusJson {
427 path: r.path,
428 status_code: r.status_code,
429 ok: r.ok,
430 checked_at: r.checked_at,
431 response_time_ms: r.response_time_ms,
432 })
433 .collect();
434
435 let dns_checks = db::get_latest_dns_checks(pool, name)
436 .await
437 .unwrap_or_default();
438 let expected_dns: HashSet<(String, String)> = config
439 .get_target(name)
440 .map(|t| {
441 t.dns
442 .iter()
443 .map(|d| (d.name.clone(), d.record_type.to_string()))
444 .collect()
445 })
446 .unwrap_or_default();
447 let dns_status: Vec<DnsStatusJson> = dns_checks
448 .into_iter()
449 .filter(|r| expected_dns.contains(&(r.name.clone(), r.record_type.clone())))
450 .map(|r| DnsStatusJson {
451 name: r.name,
452 record_type: r.record_type,
453 expected: serde_json::from_str(&r.expected).unwrap_or_default(),
454 actual: serde_json::from_str(&r.actual).unwrap_or_default(),
455 matches: r.matches,
456 checked_at: r.checked_at,
457 })
458 .collect();
459
460 let whois = db::get_latest_whois_check(pool, name).await.unwrap_or(None);
461
462 TargetStatus {
463 label: label.to_string(),
464 latest,
465 recent: recent_json,
466 uptime_24h,
467 uptime_7d,
468 latency_24h,
469 tls,
470 test_staleness,
471 test_duration_drift,
472 current_incident,
473 incidents,
474 route_status,
475 dns_status,
476 whois,
477 }
478 }
479
480 /// `GET /api/status`: JSON summary for all targets.
481 #[instrument(skip_all)]
482 async fn status_all(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
483 let mut targets = HashMap::new();
484
485 for name in state.config.target_names() {
486 if let Some(target_config) = state.config.get_target(&name) {
487 let status =
488 build_target_status(&state.pool, &name, &target_config.label, &state.config).await;
489 targets.insert(name, status);
490 }
491 }
492
493 Json(StatusResponse { targets })
494 }
495
496 /// `GET /api/status/{target}`: JSON summary for a single target.
497 #[instrument(skip_all, fields(target = %target))]
498 async fn status_target(
499 AxumState(state): AxumState<ApiState>,
500 Path(target): Path<String>,
501 ) -> impl IntoResponse {
502 let Some(target_config) = state.config.get_target(&target) else {
503 return Err((
504 StatusCode::NOT_FOUND,
505 Json(serde_json::json!({
506 "error": format!("unknown target: {target}")
507 })),
508 ));
509 };
510
511 let status =
512 build_target_status(&state.pool, &target, &target_config.label, &state.config).await;
513 Ok(Json(status))
514 }
515
516 /// `GET /status.json`: every monitored target restated in the shared
517 /// cross-service payload the release viewer renders. See `crate::status`.
518 #[instrument(skip_all)]
519 async fn status_json(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
520 Json(status_payload(&state.pool, &state.config).await)
521 }
522
523 /// The `/status.json` payload, built straight from the database.
524 ///
525 /// Public because the MCP tools serve the same payload for the local instance
526 /// without going through HTTP: a session asking what is live should not need a
527 /// running `pom serve` on its own machine to read its own database. Reading a
528 /// *remote* instance is what the HTTP path is for.
529 pub async fn status_payload(pool: &sqlx::SqlitePool, config: &Config) -> ops_status::Payload {
530 let targets = build_status_view(pool, config).await;
531 crate::status::payload(&targets, chrono::Utc::now())
532 }
533
534 /// Read each target's current signals out of the database into the pure view the
535 /// payload mapping consumes. The DB lives here; `crate::status::payload` stays a
536 /// pure function of `(targets, now)`.
537 async fn build_status_view(
538 pool: &sqlx::SqlitePool,
539 config: &Config,
540 ) -> Vec<crate::status::TargetView> {
541 let mut targets = Vec::new();
542
543 for name in config.target_names() {
544 let Some(target_config) = config.get_target(&name) else {
545 continue;
546 };
547
548 let health = db::get_latest_health(pool, &name)
549 .await
550 .ok()
551 .flatten()
552 .map(|s| crate::status::HealthView {
553 status: s.status,
554 checked_at: s.checked_at,
555 version: s.details.and_then(|d| d.version),
556 error: s.error,
557 });
558
559 let uptime_24h = db::get_uptime_percent(pool, &name, 24)
560 .await
561 .unwrap_or(None);
562
563 let latency_avg_ms = {
564 let cutoff = (chrono::Utc::now() - chrono::Duration::hours(24)).to_rfc3339();
565 let times = db::get_response_times(pool, &name, &cutoff)
566 .await
567 .unwrap_or_default();
568 let operational: Vec<i64> = times
569 .iter()
570 .filter(|(_, ms)| *ms > 0)
571 .map(|(_, ms)| *ms)
572 .collect();
573 LatencyStats::from_times(&operational).map(|s| s.avg_ms)
574 };
575
576 let tls = db::get_latest_tls_check(pool, &name)
577 .await
578 .ok()
579 .flatten()
580 .map(|r| crate::status::TlsView {
581 valid: r.valid,
582 days_remaining: r.days_remaining,
583 checked_at: r.checked_at,
584 error: r.error,
585 webpki_trusted: r.webpki_trusted,
586 platform_trusted: r.platform_trusted,
587 platform_error: r.platform_error,
588 });
589
590 let incident = db::get_open_incident(pool, &name)
591 .await
592 .ok()
593 .flatten()
594 .map(|i| crate::status::IncidentView {
595 from_status: i.from_status,
596 to_status: i.to_status,
597 started_at: i.started_at,
598 });
599
600 let whois = db::get_latest_whois_check(pool, &name)
601 .await
602 .ok()
603 .flatten()
604 .map(|w| crate::status::WhoisView {
605 days_remaining: w.days_remaining,
606 checked_at: w.checked_at,
607 error: w.error,
608 });
609
610 let mut backups = Vec::new();
611 if let Some(backup_config) = &target_config.backups {
612 for database in &backup_config.databases {
613 if let Ok(Some(row)) = db::get_latest_backup_check(pool, &name, database).await {
614 backups.push(crate::status::BackupView {
615 database: row.database_name,
616 status: row.status,
617 age_hours: row.age_hours,
618 checked_at: row.checked_at,
619 error: row.error,
620 });
621 }
622 }
623 }
624
625 let scan_pipeline = db::get_latest_scan_pipeline_check(pool, &name)
626 .await
627 .ok()
628 .flatten()
629 .map(|s| crate::status::ScanView {
630 issues: s.issue_list(),
631 status: s.status,
632 checked_at: s.checked_at,
633 error: s.error,
634 });
635
636 let systemd = db::get_latest_systemd_check(pool, &name)
637 .await
638 .ok()
639 .flatten()
640 .map(|s| crate::status::SystemdView {
641 issues: s.issue_list(),
642 status: s.status,
643 checked_at: s.checked_at,
644 error: s.error,
645 });
646
647 let ca_bundle = db::get_latest_ca_bundle_check(pool, &name)
648 .await
649 .ok()
650 .flatten()
651 .map(|c| crate::status::CaBundleView {
652 issues: c.issue_list(),
653 status: c.status,
654 checked_at: c.checked_at,
655 error: c.error,
656 });
657
658 let synckit_fleet = db::get_latest_synckit_fleet_check(pool, &name)
659 .await
660 .ok()
661 .flatten()
662 .map(|f| crate::status::SyncKitFleetView {
663 versions: f
664 .version_list()
665 .into_iter()
666 .map(|v| (v.client_version, v.devices))
667 .collect(),
668 devices: f.devices,
669 window_days: f.window_days,
670 checked_at: f.checked_at,
671 error: f.error,
672 });
673
674 // Tests: the latest run plus PoM's staleness verdict, sourced exactly as
675 // build_target_status does (version at test time vs current version).
676 let tests = if let Some(tests_config) = &target_config.tests {
677 let latest_test = db::get_latest_test_run(pool, &name).await.ok().flatten();
678 let current_version = health.as_ref().and_then(|h| h.version.clone());
679 let tested_version = if let Some(test) = &latest_test {
680 db::get_version_at_time(pool, &name, &test.started_at)
681 .await
682 .ok()
683 .flatten()
684 } else {
685 None
686 };
687 let staleness = compute_test_staleness(
688 current_version.as_deref(),
689 tested_version.as_deref(),
690 latest_test.as_ref().map(|t| t.started_at.as_str()),
691 tests_config.staleness_days,
692 );
693 Some(crate::status::TestsView {
694 ran: latest_test.is_some(),
695 passed: latest_test.as_ref().is_some_and(|t| t.passed),
696 total_passed: latest_test.as_ref().and_then(|t| t.summary.total_passed),
697 total_failed: latest_test.as_ref().and_then(|t| t.summary.total_failed),
698 started_at: latest_test.as_ref().map(|t| t.started_at.clone()),
699 stale: staleness.stale,
700 stale_reason: staleness.reason,
701 })
702 } else {
703 None
704 };
705
706 // DNS: one entry per monitored record. Absent config yields no rows and
707 // therefore no condition.
708 let dns = {
709 let rows = db::get_latest_dns_checks(pool, &name)
710 .await
711 .unwrap_or_default();
712 (!rows.is_empty()).then(|| crate::status::DnsView {
713 checked_at: rows.iter().map(|r| r.checked_at.clone()).max(),
714 records: rows
715 .into_iter()
716 .map(|r| crate::status::DnsRecordView {
717 name: r.name,
718 record_type: r.record_type,
719 matches: r.matches,
720 error: r.error,
721 })
722 .collect(),
723 })
724 };
725
726 // CORS: one entry per monitored URL.
727 let cors = {
728 let rows = db::get_latest_cors_checks(pool, &name)
729 .await
730 .unwrap_or_default();
731 (!rows.is_empty()).then(|| crate::status::CorsView {
732 checked_at: rows.iter().map(|r| r.checked_at.clone()).max(),
733 checks: rows
734 .into_iter()
735 .map(|r| crate::status::CorsCheckView {
736 url: r.url,
737 origin: r.origin,
738 passes: r.passes,
739 error: r.error,
740 })
741 .collect(),
742 })
743 };
744
745 targets.push(crate::status::TargetView {
746 name,
747 label: target_config.label.clone(),
748 health_configured: target_config.health.is_some(),
749 health,
750 uptime_24h,
751 latency_avg_ms,
752 tls,
753 incident,
754 whois,
755 backups,
756 scan_pipeline,
757 systemd,
758 ca_bundle,
759 synckit_fleet,
760 tests,
761 dns,
762 cors,
763 });
764 }
765
766 targets
767 }
768
769 // Peer endpoints
770
771 /// `GET /api/peer/info`: Returns this instance's identity info.
772 #[instrument(skip_all)]
773 async fn peer_info(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
774 let Some(ref mesh) = state.mesh else {
775 return Err((
776 StatusCode::SERVICE_UNAVAILABLE,
777 Json(serde_json::json!({
778 "error": "peer mesh not enabled"
779 })),
780 ));
781 };
782
783 let mesh_state = mesh.read().await;
784 Ok(Json(
785 serde_json::to_value(&mesh_state.instance).unwrap_or_default(),
786 ))
787 }
788
789 /// `GET /api/peer/status`: This instance's full view: own info + target statuses + peer summaries.
790 #[instrument(skip_all)]
791 async fn peer_status(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
792 let Some(ref mesh) = state.mesh else {
793 return Err((
794 StatusCode::SERVICE_UNAVAILABLE,
795 Json(serde_json::json!({
796 "error": "peer mesh not enabled"
797 })),
798 ));
799 };
800
801 // Collect mesh data under lock, then drop lock before DB queries
802 let (instance, peers) = {
803 let mesh_state = mesh.read().await;
804 let instance = mesh_state.instance.clone();
805 let peers: HashMap<String, serde_json::Value> = mesh_state
806 .peers
807 .iter()
808 .map(|(name, peer)| {
809 (
810 name.clone(),
811 serde_json::json!({
812 "status": peer.status,
813 "last_seen": peer.last_seen,
814 "latency_ms": peer.latency_ms,
815 }),
816 )
817 })
818 .collect();
819 (instance, peers)
820 };
821
822 // Build target statuses (DB queries with no lock held)
823 let mut targets = HashMap::new();
824 for name in state.config.target_names() {
825 if let Some(target_config) = state.config.get_target(&name)
826 && let Ok(Some(latest)) = db::get_latest_health(&state.pool, &name).await
827 {
828 targets.insert(
829 name,
830 serde_json::json!({
831 "label": target_config.label,
832 "status": latest.status.to_string(),
833 "response_time_ms": latest.response_time_ms,
834 "checked_at": latest.checked_at,
835 }),
836 );
837 }
838 }
839
840 Ok(Json(serde_json::json!({
841 "instance": instance,
842 "targets": targets,
843 "peers": peers,
844 })))
845 }
846
847 /// `GET /api/mesh`: Aggregated view: self + each peer's cached status.
848 #[instrument(skip_all)]
849 async fn mesh_view(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
850 let Some(ref mesh) = state.mesh else {
851 return Err((
852 StatusCode::SERVICE_UNAVAILABLE,
853 Json(serde_json::json!({
854 "error": "peer mesh not enabled"
855 })),
856 ));
857 };
858
859 // Collect all mesh data under lock, then drop lock before DB queries
860 let (instance, own_peers_json, peer_entries) = {
861 let mesh_state = mesh.read().await;
862 let instance = mesh_state.instance.clone();
863 let own_peers: HashMap<String, serde_json::Value> = mesh_state
864 .peers
865 .iter()
866 .map(|(name, peer)| {
867 (
868 name.clone(),
869 serde_json::json!({
870 "status": peer.status,
871 "last_seen": peer.last_seen,
872 "latency_ms": peer.latency_ms,
873 }),
874 )
875 })
876 .collect();
877 let peer_entries: Vec<(String, Option<serde_json::Value>, serde_json::Value)> = mesh_state
878 .peers
879 .iter()
880 .map(|(name, peer)| {
881 let fallback = serde_json::json!({
882 "status": peer.status,
883 "last_seen": peer.last_seen,
884 "error": "no status data cached",
885 });
886 (name.clone(), peer.status_data.clone(), fallback)
887 })
888 .collect();
889 (instance, own_peers, peer_entries)
890 };
891
892 // Build target statuses (DB queries with no lock held)
893 let mut targets = HashMap::new();
894 for name in state.config.target_names() {
895 if let Some(target_config) = state.config.get_target(&name)
896 && let Ok(Some(latest)) = db::get_latest_health(&state.pool, &name).await
897 {
898 targets.insert(
899 name,
900 serde_json::json!({
901 "label": target_config.label,
902 "status": latest.status.to_string(),
903 "response_time_ms": latest.response_time_ms,
904 "checked_at": latest.checked_at,
905 }),
906 );
907 }
908 }
909
910 let self_entry = serde_json::json!({
911 "instance": instance,
912 "targets": targets,
913 "peers": own_peers_json,
914 });
915
916 let mut instances = serde_json::Map::new();
917 instances.insert(instance.name.clone(), self_entry);
918
919 for (name, status_data, fallback) in peer_entries {
920 // Re-project the peer's cached status into a FIXED schema instead of
921 // re-serving its raw JSON verbatim: a compromised/MITM'd peer otherwise
922 // injects arbitrary structure into /api/mesh consumers (mesh poisoning,
923 // fuzz-2026-07-06 #6). Values are still the peer's to report; the shape is
924 // ours. Terminal rendering additionally scrubs the values (display::scrub).
925 let entry = status_data.as_ref().map_or(fallback, reproject_peer_status);
926 instances.insert(name, entry);
927 }
928
929 Ok(Json(serde_json::json!({
930 "instances": instances,
931 })))
932 }
933
934 /// Extract only the known fields from an untrusted peer's cached status blob,
935 /// dropping any attacker-injected extra structure. Preserves exactly the paths
936 /// the mesh consumers (`display::format_mesh`, the dashboard) read.
937 fn reproject_peer_status(raw: &serde_json::Value) -> serde_json::Value {
938 let instance = raw.get("instance").map(|i| {
939 serde_json::json!({
940 "id": i.get("id").and_then(|v| v.as_str()),
941 "name": i.get("name").and_then(|v| v.as_str()),
942 "version": i.get("version").and_then(|v| v.as_str()),
943 "started_at": i.get("started_at").and_then(|v| v.as_str()),
944 "targets": i.get("targets")
945 .and_then(|v| v.as_array())
946 .map(|a| a.iter().filter_map(|t| t.as_str()).collect::<Vec<_>>())
947 .unwrap_or_default(),
948 })
949 });
950
951 let project_map = |key: &str, f: &dyn Fn(&serde_json::Value) -> serde_json::Value| {
952 let mut out = serde_json::Map::new();
953 if let Some(obj) = raw.get(key).and_then(|v| v.as_object()) {
954 for (k, v) in obj {
955 out.insert(k.clone(), f(v));
956 }
957 }
958 serde_json::Value::Object(out)
959 };
960
961 let targets = project_map("targets", &|t| {
962 serde_json::json!({
963 "label": t.get("label").and_then(|v| v.as_str()),
964 "status": t.get("status").and_then(|v| v.as_str()),
965 "response_time_ms": t.get("response_time_ms").and_then(serde_json::Value::as_i64),
966 "checked_at": t.get("checked_at").and_then(|v| v.as_str()),
967 })
968 });
969
970 let peers = project_map("peers", &|p| {
971 serde_json::json!({
972 "status": p.get("status").and_then(|v| v.as_str()),
973 "last_seen": p.get("last_seen").and_then(|v| v.as_str()),
974 "latency_ms": p.get("latency_ms").and_then(serde_json::Value::as_u64),
975 })
976 });
977
978 serde_json::json!({ "instance": instance, "targets": targets, "peers": peers })
979 }
980
981 // Trends endpoint
982
983 #[derive(Serialize, serde::Deserialize)]
984 pub struct TrendResponse {
985 /// Target config name this trend data belongs to.
986 pub target: String,
987 /// Requested time window in hours (from query param, default 24).
988 pub window_hours: u64,
989 /// Requested bucket width in minutes (from query param, default 60).
990 pub bucket_minutes: u64,
991 /// Per-bucket latency statistics within the requested window.
992 pub buckets: Vec<LatencyBucket>,
993 /// Aggregate latency statistics across the entire requested window.
994 pub overall: Option<LatencyStats>,
995 /// 7-day baseline latency statistics for drift comparison.
996 pub baseline: Option<LatencyStats>,
997 }
998
999 /// `GET /api/trends/{target}?hours=24&bucket_minutes=60`: latency trend data.
1000 #[instrument(skip_all, fields(target = %target))]
1001 async fn trends(
1002 AxumState(state): AxumState<ApiState>,
1003 Path(target): Path<String>,
1004 axum::extract::Query(params): axum::extract::Query<TrendQueryParams>,
1005 ) -> impl IntoResponse {
1006 let Some(_target_config) = state.config.get_target(&target) else {
1007 return Err((
1008 StatusCode::NOT_FOUND,
1009 Json(serde_json::json!({
1010 "error": format!("unknown target: {target}")
1011 })),
1012 ));
1013 };
1014
1015 let response = build_trends(
1016 &state.pool,
1017 &target,
1018 params.hours.unwrap_or(24),
1019 params.bucket_minutes.unwrap_or(60),
1020 )
1021 .await;
1022
1023 Ok(Json(response))
1024 }
1025
1026 /// The latency trend for one target: per-bucket stats within the window, the
1027 /// window aggregate, and a 7-day baseline to read it against.
1028 ///
1029 /// Public for the same reason as [`status_payload`]: the MCP tools serve the
1030 /// local instance's answer without requiring a running daemon on this machine.
1031 pub async fn build_trends(
1032 pool: &sqlx::SqlitePool,
1033 target: &str,
1034 hours: u64,
1035 bucket_minutes: u64,
1036 ) -> TrendResponse {
1037 let cutoff = (chrono::Utc::now() - chrono::Duration::hours(hours as i64)).to_rfc3339();
1038 let times = db::get_response_times(pool, target, &cutoff)
1039 .await
1040 .unwrap_or_default();
1041
1042 let operational_times: Vec<i64> = times
1043 .iter()
1044 .filter(|(_, ms)| *ms > 0)
1045 .map(|(_, ms)| *ms)
1046 .collect();
1047 let overall = LatencyStats::from_times(&operational_times);
1048
1049 let operational_data: Vec<(String, i64)> =
1050 times.into_iter().filter(|(_, ms)| *ms > 0).collect();
1051 let buckets = LatencyStats::bucket_by_time(&operational_data, bucket_minutes);
1052
1053 // 7d baseline for reference
1054 let baseline_cutoff = (chrono::Utc::now() - chrono::Duration::hours(168)).to_rfc3339();
1055 let baseline_times = db::get_response_times(pool, target, &baseline_cutoff)
1056 .await
1057 .unwrap_or_default();
1058 let baseline_operational: Vec<i64> = baseline_times
1059 .iter()
1060 .filter(|(_, ms)| *ms > 0)
1061 .map(|(_, ms)| *ms)
1062 .collect();
1063 let baseline = LatencyStats::from_times(&baseline_operational);
1064
1065 TrendResponse {
1066 target: target.to_string(),
1067 window_hours: hours,
1068 bucket_minutes,
1069 buckets,
1070 overall,
1071 baseline,
1072 }
1073 }
1074
1075 /// `GET /api/versions`: what every target is running, and how far behind.
1076 ///
1077 /// The commits-behind column is measured against a checkout on *this* host, so
1078 /// an instance without the repo serves the rest of the row and leaves that one
1079 /// blank. Reading it from a peer therefore answers "what is live there", not
1080 /// "how far behind is it here".
1081 #[instrument(skip_all)]
1082 async fn versions(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
1083 match crate::versions::collect(&state.pool, &state.config).await {
1084 Ok(rows) => Ok(Json(rows)),
1085 Err(e) => Err((
1086 StatusCode::INTERNAL_SERVER_ERROR,
1087 Json(serde_json::json!({ "error": e.to_string() })),
1088 )),
1089 }
1090 }
1091
1092 #[derive(serde::Deserialize)]
1093 struct TrendQueryParams {
1094 /// Time window to query, in hours. Defaults to 24 if omitted.
1095 hours: Option<u64>,
1096 /// Width of each latency bucket, in minutes. Defaults to 60 if omitted.
1097 bucket_minutes: Option<u64>,
1098 }
1099
1100 #[cfg(test)]
1101 mod tests {
1102 use super::*;
1103 use axum::body::Body;
1104 use axum::http::Request as HttpRequest;
1105 use tower::ServiceExt;
1106
1107 fn test_config(api_token: Option<&str>) -> Config {
1108 let mut config = Config {
1109 serve: crate::config::ServeConfig::default(),
1110 instance: crate::config::InstanceConfig::default(),
1111 targets: HashMap::new(),
1112 peers: HashMap::new(),
1113 storage: crate::config::StorageConfig::default(),
1114 alerts: None,
1115 };
1116 config.serve.api_token = api_token.map(std::string::ToString::to_string);
1117 config
1118 }
1119
1120 #[tokio::test]
1121 async fn no_token_configured_allows_all_requests() {
1122 let pool = crate::db::connect_in_memory().await.unwrap();
1123 let app = router(pool, test_config(None), None);
1124
1125 let resp = app
1126 .oneshot(with_connect_info("/api/status", None))
1127 .await
1128 .unwrap();
1129 assert_eq!(resp.status(), StatusCode::OK);
1130 }
1131
1132 #[tokio::test]
1133 async fn valid_token_allows_request() {
1134 let pool = crate::db::connect_in_memory().await.unwrap();
1135 let app = router(pool, test_config(Some("secret123")), None);
1136
1137 let resp = app
1138 .oneshot(with_connect_info("/api/status", Some("Bearer secret123")))
1139 .await
1140 .unwrap();
1141 assert_eq!(resp.status(), StatusCode::OK);
1142 }
1143
1144 #[tokio::test]
1145 async fn dashboard_uses_cookie_not_embedded_token() {
1146 // SERIOUS #4: GET / must NOT ship the api_token in the page, and must set
1147 // an httpOnly session cookie that authenticates the dashboard's /api/* calls.
1148 let pool = crate::db::connect_in_memory().await.unwrap();
1149 let mut config = test_config(Some("supersecret-token"));
1150 config.serve.dashboard = true;
1151 let app = router(pool, config, None);
1152
1153 let req = HttpRequest::builder().uri("/").body(Body::empty()).unwrap();
1154 let resp = app.clone().oneshot(req).await.unwrap();
1155 assert_eq!(resp.status(), StatusCode::OK);
1156
1157 let cookie_hdr = resp
1158 .headers()
1159 .get(axum::http::header::SET_COOKIE)
1160 .expect("dashboard must set a session cookie")
1161 .to_str()
1162 .unwrap()
1163 .to_string();
1164 assert!(cookie_hdr.contains("pom_dash="));
1165 assert!(cookie_hdr.contains("HttpOnly"));
1166 assert!(
1167 !cookie_hdr.contains("supersecret-token"),
1168 "cookie must not be the api_token"
1169 );
1170
1171 let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
1172 .await
1173 .unwrap();
1174 let html = String::from_utf8_lossy(&body);
1175 assert!(
1176 !html.contains("supersecret-token"),
1177 "the api_token must never appear in served HTML"
1178 );
1179
1180 // The issued cookie authenticates an /api/* call without any bearer token.
1181 let dash = cookie_hdr.split(';').next().unwrap().trim().to_string(); // "pom_dash=<value>"
1182 let mut api_req = HttpRequest::builder()
1183 .uri("/api/status")
1184 .header(axum::http::header::COOKIE, dash)
1185 .body(Body::empty())
1186 .unwrap();
1187 api_req
1188 .extensions_mut()
1189 .insert(axum::extract::ConnectInfo(std::net::SocketAddr::from((
1190 [127, 0, 0, 1],
1191 40001,
1192 ))));
1193 let api_resp = app.oneshot(api_req).await.unwrap();
1194 assert_eq!(
1195 api_resp.status(),
1196 StatusCode::OK,
1197 "dashboard cookie must authenticate /api/*"
1198 );
1199 }
1200
1201 #[tokio::test]
1202 async fn wrong_token_returns_401() {
1203 let pool = crate::db::connect_in_memory().await.unwrap();
1204 let app = router(pool, test_config(Some("secret123")), None);
1205
1206 let req = HttpRequest::builder()
1207 .uri("/api/status")
1208 .header("authorization", "Bearer wrong-token")
1209 .body(Body::empty())
1210 .unwrap();
1211 let resp = app.oneshot(req).await.unwrap();
1212 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1213 }
1214
1215 #[tokio::test]
1216 async fn missing_header_returns_401() {
1217 let pool = crate::db::connect_in_memory().await.unwrap();
1218 let app = router(pool, test_config(Some("secret123")), None);
1219
1220 let req = HttpRequest::builder()
1221 .uri("/api/status")
1222 .body(Body::empty())
1223 .unwrap();
1224 let resp = app.oneshot(req).await.unwrap();
1225 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1226 }
1227
1228 #[tokio::test]
1229 async fn malformed_header_returns_401() {
1230 let pool = crate::db::connect_in_memory().await.unwrap();
1231 let app = router(pool, test_config(Some("secret123")), None);
1232
1233 let req = HttpRequest::builder()
1234 .uri("/api/status")
1235 .header("authorization", "Basic dXNlcjpwYXNz")
1236 .body(Body::empty())
1237 .unwrap();
1238 let resp = app.oneshot(req).await.unwrap();
1239 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1240 }
1241
1242 fn ip(n: u8) -> std::net::IpAddr {
1243 std::net::IpAddr::V4(std::net::Ipv4Addr::new(10, 0, 0, n))
1244 }
1245
1246 #[test]
1247 fn reproject_drops_injected_structure() {
1248 // #6: a compromised peer stuffs extra structure into its status blob.
1249 let hostile = serde_json::json!({
1250 "instance": { "id": "abc", "version": "9.9", "evil_field": {"x": 1} },
1251 "targets": { "mnw": { "status": "operational", "response_time_ms": 5, "evil": "inject" } },
1252 "peers": { "p2": { "status": "up", "latency_ms": 3 } },
1253 "top_level_injection": [1, 2, 3]
1254 });
1255 let clean = reproject_peer_status(&hostile);
1256
1257 // Only the fixed top-level keys survive.
1258 let obj = clean.as_object().unwrap();
1259 let mut keys: Vec<&String> = obj.keys().collect();
1260 keys.sort();
1261 assert_eq!(keys, vec!["instance", "peers", "targets"]);
1262 assert!(clean.get("top_level_injection").is_none());
1263
1264 // Known values are preserved; injected sibling keys are gone.
1265 assert_eq!(clean["instance"]["id"], "abc");
1266 assert_eq!(clean["instance"]["version"], "9.9");
1267 assert!(clean["instance"].get("evil_field").is_none());
1268 assert_eq!(clean["targets"]["mnw"]["status"], "operational");
1269 assert!(clean["targets"]["mnw"].get("evil").is_none());
1270 assert_eq!(clean["peers"]["p2"]["latency_ms"], 3);
1271 }
1272
1273 /// Build a GET request carrying a `ConnectInfo<SocketAddr>` extension, which
1274 /// the real server injects via `into_make_service_with_connect_info` but
1275 /// `oneshot` does not, the rate-limit layer extracts it.
1276 fn with_connect_info(uri: &str, bearer: Option<&str>) -> HttpRequest<Body> {
1277 let mut b = HttpRequest::builder().uri(uri);
1278 if let Some(h) = bearer {
1279 b = b.header("authorization", h);
1280 }
1281 let mut req = b.body(Body::empty()).unwrap();
1282 req.extensions_mut()
1283 .insert(axum::extract::ConnectInfo(std::net::SocketAddr::from((
1284 [127, 0, 0, 1],
1285 40000,
1286 ))));
1287 req
1288 }
1289
1290 #[test]
1291 fn rate_limiter_allows_within_limit() {
1292 let limiter = PerIpRateLimiter::new(3, std::time::Duration::from_mins(1));
1293 assert!(limiter.try_acquire(ip(1)));
1294 assert!(limiter.try_acquire(ip(1)));
1295 assert!(limiter.try_acquire(ip(1)));
1296 }
1297
1298 #[test]
1299 fn rate_limiter_blocks_over_limit() {
1300 let limiter = PerIpRateLimiter::new(2, std::time::Duration::from_mins(1));
1301 assert!(limiter.try_acquire(ip(1)));
1302 assert!(limiter.try_acquire(ip(1)));
1303 assert!(!limiter.try_acquire(ip(1)));
1304 }
1305
1306 #[test]
1307 fn rate_limiter_isolates_clients_by_ip() {
1308 // SERIOUS #5: one client exhausting its bucket must not affect another.
1309 let limiter = PerIpRateLimiter::new(1, std::time::Duration::from_mins(1));
1310 assert!(limiter.try_acquire(ip(1)));
1311 assert!(
1312 !limiter.try_acquire(ip(1)),
1313 "ip(1) is now over its own limit"
1314 );
1315 assert!(limiter.try_acquire(ip(2)), "ip(2) has its own fresh bucket");
1316 }
1317
1318 #[tokio::test]
1319 async fn rate_limiter_resets_after_window() {
1320 let limiter = PerIpRateLimiter::new(1, std::time::Duration::from_millis(10));
1321 assert!(limiter.try_acquire(ip(1)));
1322 assert!(!limiter.try_acquire(ip(1)));
1323 tokio::time::sleep(std::time::Duration::from_millis(15)).await;
1324 assert!(limiter.try_acquire(ip(1)));
1325 }
1326 }
1327