Skip to main content

max / makenotwork

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