Skip to main content

max / makenotwork

45.9 KB · 1312 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 });
587
588 let incident = db::get_open_incident(pool, &name)
589 .await
590 .ok()
591 .flatten()
592 .map(|i| crate::status::IncidentView {
593 from_status: i.from_status,
594 to_status: i.to_status,
595 started_at: i.started_at,
596 });
597
598 let whois = db::get_latest_whois_check(pool, &name)
599 .await
600 .ok()
601 .flatten()
602 .map(|w| crate::status::WhoisView {
603 days_remaining: w.days_remaining,
604 checked_at: w.checked_at,
605 error: w.error,
606 });
607
608 let mut backups = Vec::new();
609 if let Some(backup_config) = &target_config.backups {
610 for database in &backup_config.databases {
611 if let Ok(Some(row)) = db::get_latest_backup_check(pool, &name, database).await {
612 backups.push(crate::status::BackupView {
613 database: row.database_name,
614 status: row.status,
615 age_hours: row.age_hours,
616 checked_at: row.checked_at,
617 error: row.error,
618 });
619 }
620 }
621 }
622
623 let scan_pipeline = db::get_latest_scan_pipeline_check(pool, &name)
624 .await
625 .ok()
626 .flatten()
627 .map(|s| crate::status::ScanView {
628 issues: s.issue_list(),
629 status: s.status,
630 checked_at: s.checked_at,
631 error: s.error,
632 });
633
634 let systemd = db::get_latest_systemd_check(pool, &name)
635 .await
636 .ok()
637 .flatten()
638 .map(|s| crate::status::SystemdView {
639 issues: s.issue_list(),
640 status: s.status,
641 checked_at: s.checked_at,
642 error: s.error,
643 });
644
645 let synckit_fleet = db::get_latest_synckit_fleet_check(pool, &name)
646 .await
647 .ok()
648 .flatten()
649 .map(|f| crate::status::SyncKitFleetView {
650 versions: f
651 .version_list()
652 .into_iter()
653 .map(|v| (v.client_version, v.devices))
654 .collect(),
655 devices: f.devices,
656 window_days: f.window_days,
657 checked_at: f.checked_at,
658 error: f.error,
659 });
660
661 // Tests: the latest run plus PoM's staleness verdict, sourced exactly as
662 // build_target_status does (version at test time vs current version).
663 let tests = if let Some(tests_config) = &target_config.tests {
664 let latest_test = db::get_latest_test_run(pool, &name).await.ok().flatten();
665 let current_version = health.as_ref().and_then(|h| h.version.clone());
666 let tested_version = if let Some(test) = &latest_test {
667 db::get_version_at_time(pool, &name, &test.started_at)
668 .await
669 .ok()
670 .flatten()
671 } else {
672 None
673 };
674 let staleness = compute_test_staleness(
675 current_version.as_deref(),
676 tested_version.as_deref(),
677 latest_test.as_ref().map(|t| t.started_at.as_str()),
678 tests_config.staleness_days,
679 );
680 Some(crate::status::TestsView {
681 ran: latest_test.is_some(),
682 passed: latest_test.as_ref().is_some_and(|t| t.passed),
683 total_passed: latest_test.as_ref().and_then(|t| t.summary.total_passed),
684 total_failed: latest_test.as_ref().and_then(|t| t.summary.total_failed),
685 started_at: latest_test.as_ref().map(|t| t.started_at.clone()),
686 stale: staleness.stale,
687 stale_reason: staleness.reason,
688 })
689 } else {
690 None
691 };
692
693 // DNS: one entry per monitored record. Absent config yields no rows and
694 // therefore no condition.
695 let dns = {
696 let rows = db::get_latest_dns_checks(pool, &name)
697 .await
698 .unwrap_or_default();
699 (!rows.is_empty()).then(|| crate::status::DnsView {
700 checked_at: rows.iter().map(|r| r.checked_at.clone()).max(),
701 records: rows
702 .into_iter()
703 .map(|r| crate::status::DnsRecordView {
704 name: r.name,
705 record_type: r.record_type,
706 matches: r.matches,
707 error: r.error,
708 })
709 .collect(),
710 })
711 };
712
713 // CORS: one entry per monitored URL.
714 let cors = {
715 let rows = db::get_latest_cors_checks(pool, &name)
716 .await
717 .unwrap_or_default();
718 (!rows.is_empty()).then(|| crate::status::CorsView {
719 checked_at: rows.iter().map(|r| r.checked_at.clone()).max(),
720 checks: rows
721 .into_iter()
722 .map(|r| crate::status::CorsCheckView {
723 url: r.url,
724 origin: r.origin,
725 passes: r.passes,
726 error: r.error,
727 })
728 .collect(),
729 })
730 };
731
732 targets.push(crate::status::TargetView {
733 name,
734 label: target_config.label.clone(),
735 health_configured: target_config.health.is_some(),
736 health,
737 uptime_24h,
738 latency_avg_ms,
739 tls,
740 incident,
741 whois,
742 backups,
743 scan_pipeline,
744 systemd,
745 synckit_fleet,
746 tests,
747 dns,
748 cors,
749 });
750 }
751
752 targets
753 }
754
755 // Peer endpoints
756
757 /// `GET /api/peer/info`: Returns this instance's identity info.
758 #[instrument(skip_all)]
759 async fn peer_info(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
760 let Some(ref mesh) = state.mesh else {
761 return Err((
762 StatusCode::SERVICE_UNAVAILABLE,
763 Json(serde_json::json!({
764 "error": "peer mesh not enabled"
765 })),
766 ));
767 };
768
769 let mesh_state = mesh.read().await;
770 Ok(Json(
771 serde_json::to_value(&mesh_state.instance).unwrap_or_default(),
772 ))
773 }
774
775 /// `GET /api/peer/status`: This instance's full view: own info + target statuses + peer summaries.
776 #[instrument(skip_all)]
777 async fn peer_status(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
778 let Some(ref mesh) = state.mesh else {
779 return Err((
780 StatusCode::SERVICE_UNAVAILABLE,
781 Json(serde_json::json!({
782 "error": "peer mesh not enabled"
783 })),
784 ));
785 };
786
787 // Collect mesh data under lock, then drop lock before DB queries
788 let (instance, peers) = {
789 let mesh_state = mesh.read().await;
790 let instance = mesh_state.instance.clone();
791 let peers: HashMap<String, serde_json::Value> = mesh_state
792 .peers
793 .iter()
794 .map(|(name, peer)| {
795 (
796 name.clone(),
797 serde_json::json!({
798 "status": peer.status,
799 "last_seen": peer.last_seen,
800 "latency_ms": peer.latency_ms,
801 }),
802 )
803 })
804 .collect();
805 (instance, peers)
806 };
807
808 // Build target statuses (DB queries with no lock held)
809 let mut targets = HashMap::new();
810 for name in state.config.target_names() {
811 if let Some(target_config) = state.config.get_target(&name)
812 && let Ok(Some(latest)) = db::get_latest_health(&state.pool, &name).await
813 {
814 targets.insert(
815 name,
816 serde_json::json!({
817 "label": target_config.label,
818 "status": latest.status.to_string(),
819 "response_time_ms": latest.response_time_ms,
820 "checked_at": latest.checked_at,
821 }),
822 );
823 }
824 }
825
826 Ok(Json(serde_json::json!({
827 "instance": instance,
828 "targets": targets,
829 "peers": peers,
830 })))
831 }
832
833 /// `GET /api/mesh`: Aggregated view: self + each peer's cached status.
834 #[instrument(skip_all)]
835 async fn mesh_view(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
836 let Some(ref mesh) = state.mesh else {
837 return Err((
838 StatusCode::SERVICE_UNAVAILABLE,
839 Json(serde_json::json!({
840 "error": "peer mesh not enabled"
841 })),
842 ));
843 };
844
845 // Collect all mesh data under lock, then drop lock before DB queries
846 let (instance, own_peers_json, peer_entries) = {
847 let mesh_state = mesh.read().await;
848 let instance = mesh_state.instance.clone();
849 let own_peers: HashMap<String, serde_json::Value> = mesh_state
850 .peers
851 .iter()
852 .map(|(name, peer)| {
853 (
854 name.clone(),
855 serde_json::json!({
856 "status": peer.status,
857 "last_seen": peer.last_seen,
858 "latency_ms": peer.latency_ms,
859 }),
860 )
861 })
862 .collect();
863 let peer_entries: Vec<(String, Option<serde_json::Value>, serde_json::Value)> = mesh_state
864 .peers
865 .iter()
866 .map(|(name, peer)| {
867 let fallback = serde_json::json!({
868 "status": peer.status,
869 "last_seen": peer.last_seen,
870 "error": "no status data cached",
871 });
872 (name.clone(), peer.status_data.clone(), fallback)
873 })
874 .collect();
875 (instance, own_peers, peer_entries)
876 };
877
878 // Build target statuses (DB queries with no lock held)
879 let mut targets = HashMap::new();
880 for name in state.config.target_names() {
881 if let Some(target_config) = state.config.get_target(&name)
882 && let Ok(Some(latest)) = db::get_latest_health(&state.pool, &name).await
883 {
884 targets.insert(
885 name,
886 serde_json::json!({
887 "label": target_config.label,
888 "status": latest.status.to_string(),
889 "response_time_ms": latest.response_time_ms,
890 "checked_at": latest.checked_at,
891 }),
892 );
893 }
894 }
895
896 let self_entry = serde_json::json!({
897 "instance": instance,
898 "targets": targets,
899 "peers": own_peers_json,
900 });
901
902 let mut instances = serde_json::Map::new();
903 instances.insert(instance.name.clone(), self_entry);
904
905 for (name, status_data, fallback) in peer_entries {
906 // Re-project the peer's cached status into a FIXED schema instead of
907 // re-serving its raw JSON verbatim: a compromised/MITM'd peer otherwise
908 // injects arbitrary structure into /api/mesh consumers (mesh poisoning,
909 // fuzz-2026-07-06 #6). Values are still the peer's to report; the shape is
910 // ours. Terminal rendering additionally scrubs the values (display::scrub).
911 let entry = status_data.as_ref().map_or(fallback, reproject_peer_status);
912 instances.insert(name, entry);
913 }
914
915 Ok(Json(serde_json::json!({
916 "instances": instances,
917 })))
918 }
919
920 /// Extract only the known fields from an untrusted peer's cached status blob,
921 /// dropping any attacker-injected extra structure. Preserves exactly the paths
922 /// the mesh consumers (`display::format_mesh`, the dashboard) read.
923 fn reproject_peer_status(raw: &serde_json::Value) -> serde_json::Value {
924 let instance = raw.get("instance").map(|i| {
925 serde_json::json!({
926 "id": i.get("id").and_then(|v| v.as_str()),
927 "name": i.get("name").and_then(|v| v.as_str()),
928 "version": i.get("version").and_then(|v| v.as_str()),
929 "started_at": i.get("started_at").and_then(|v| v.as_str()),
930 "targets": i.get("targets")
931 .and_then(|v| v.as_array())
932 .map(|a| a.iter().filter_map(|t| t.as_str()).collect::<Vec<_>>())
933 .unwrap_or_default(),
934 })
935 });
936
937 let project_map = |key: &str, f: &dyn Fn(&serde_json::Value) -> serde_json::Value| {
938 let mut out = serde_json::Map::new();
939 if let Some(obj) = raw.get(key).and_then(|v| v.as_object()) {
940 for (k, v) in obj {
941 out.insert(k.clone(), f(v));
942 }
943 }
944 serde_json::Value::Object(out)
945 };
946
947 let targets = project_map("targets", &|t| {
948 serde_json::json!({
949 "label": t.get("label").and_then(|v| v.as_str()),
950 "status": t.get("status").and_then(|v| v.as_str()),
951 "response_time_ms": t.get("response_time_ms").and_then(serde_json::Value::as_i64),
952 "checked_at": t.get("checked_at").and_then(|v| v.as_str()),
953 })
954 });
955
956 let peers = project_map("peers", &|p| {
957 serde_json::json!({
958 "status": p.get("status").and_then(|v| v.as_str()),
959 "last_seen": p.get("last_seen").and_then(|v| v.as_str()),
960 "latency_ms": p.get("latency_ms").and_then(serde_json::Value::as_u64),
961 })
962 });
963
964 serde_json::json!({ "instance": instance, "targets": targets, "peers": peers })
965 }
966
967 // Trends endpoint
968
969 #[derive(Serialize, serde::Deserialize)]
970 pub struct TrendResponse {
971 /// Target config name this trend data belongs to.
972 pub target: String,
973 /// Requested time window in hours (from query param, default 24).
974 pub window_hours: u64,
975 /// Requested bucket width in minutes (from query param, default 60).
976 pub bucket_minutes: u64,
977 /// Per-bucket latency statistics within the requested window.
978 pub buckets: Vec<LatencyBucket>,
979 /// Aggregate latency statistics across the entire requested window.
980 pub overall: Option<LatencyStats>,
981 /// 7-day baseline latency statistics for drift comparison.
982 pub baseline: Option<LatencyStats>,
983 }
984
985 /// `GET /api/trends/{target}?hours=24&bucket_minutes=60`: latency trend data.
986 #[instrument(skip_all, fields(target = %target))]
987 async fn trends(
988 AxumState(state): AxumState<ApiState>,
989 Path(target): Path<String>,
990 axum::extract::Query(params): axum::extract::Query<TrendQueryParams>,
991 ) -> impl IntoResponse {
992 let Some(_target_config) = state.config.get_target(&target) else {
993 return Err((
994 StatusCode::NOT_FOUND,
995 Json(serde_json::json!({
996 "error": format!("unknown target: {target}")
997 })),
998 ));
999 };
1000
1001 let response = build_trends(
1002 &state.pool,
1003 &target,
1004 params.hours.unwrap_or(24),
1005 params.bucket_minutes.unwrap_or(60),
1006 )
1007 .await;
1008
1009 Ok(Json(response))
1010 }
1011
1012 /// The latency trend for one target: per-bucket stats within the window, the
1013 /// window aggregate, and a 7-day baseline to read it against.
1014 ///
1015 /// Public for the same reason as [`status_payload`]: the MCP tools serve the
1016 /// local instance's answer without requiring a running daemon on this machine.
1017 pub async fn build_trends(
1018 pool: &sqlx::SqlitePool,
1019 target: &str,
1020 hours: u64,
1021 bucket_minutes: u64,
1022 ) -> TrendResponse {
1023 let cutoff = (chrono::Utc::now() - chrono::Duration::hours(hours as i64)).to_rfc3339();
1024 let times = db::get_response_times(pool, target, &cutoff)
1025 .await
1026 .unwrap_or_default();
1027
1028 let operational_times: Vec<i64> = times
1029 .iter()
1030 .filter(|(_, ms)| *ms > 0)
1031 .map(|(_, ms)| *ms)
1032 .collect();
1033 let overall = LatencyStats::from_times(&operational_times);
1034
1035 let operational_data: Vec<(String, i64)> =
1036 times.into_iter().filter(|(_, ms)| *ms > 0).collect();
1037 let buckets = LatencyStats::bucket_by_time(&operational_data, bucket_minutes);
1038
1039 // 7d baseline for reference
1040 let baseline_cutoff = (chrono::Utc::now() - chrono::Duration::hours(168)).to_rfc3339();
1041 let baseline_times = db::get_response_times(pool, target, &baseline_cutoff)
1042 .await
1043 .unwrap_or_default();
1044 let baseline_operational: Vec<i64> = baseline_times
1045 .iter()
1046 .filter(|(_, ms)| *ms > 0)
1047 .map(|(_, ms)| *ms)
1048 .collect();
1049 let baseline = LatencyStats::from_times(&baseline_operational);
1050
1051 TrendResponse {
1052 target: target.to_string(),
1053 window_hours: hours,
1054 bucket_minutes,
1055 buckets,
1056 overall,
1057 baseline,
1058 }
1059 }
1060
1061 /// `GET /api/versions`: what every target is running, and how far behind.
1062 ///
1063 /// The commits-behind column is measured against a checkout on *this* host, so
1064 /// an instance without the repo serves the rest of the row and leaves that one
1065 /// blank. Reading it from a peer therefore answers "what is live there", not
1066 /// "how far behind is it here".
1067 #[instrument(skip_all)]
1068 async fn versions(AxumState(state): AxumState<ApiState>) -> impl IntoResponse {
1069 match crate::versions::collect(&state.pool, &state.config).await {
1070 Ok(rows) => Ok(Json(rows)),
1071 Err(e) => Err((
1072 StatusCode::INTERNAL_SERVER_ERROR,
1073 Json(serde_json::json!({ "error": e.to_string() })),
1074 )),
1075 }
1076 }
1077
1078 #[derive(serde::Deserialize)]
1079 struct TrendQueryParams {
1080 /// Time window to query, in hours. Defaults to 24 if omitted.
1081 hours: Option<u64>,
1082 /// Width of each latency bucket, in minutes. Defaults to 60 if omitted.
1083 bucket_minutes: Option<u64>,
1084 }
1085
1086 #[cfg(test)]
1087 mod tests {
1088 use super::*;
1089 use axum::body::Body;
1090 use axum::http::Request as HttpRequest;
1091 use tower::ServiceExt;
1092
1093 fn test_config(api_token: Option<&str>) -> Config {
1094 let mut config = Config {
1095 serve: crate::config::ServeConfig::default(),
1096 instance: crate::config::InstanceConfig::default(),
1097 targets: HashMap::new(),
1098 peers: HashMap::new(),
1099 alerts: None,
1100 };
1101 config.serve.api_token = api_token.map(std::string::ToString::to_string);
1102 config
1103 }
1104
1105 #[tokio::test]
1106 async fn no_token_configured_allows_all_requests() {
1107 let pool = crate::db::connect_in_memory().await.unwrap();
1108 let app = router(pool, test_config(None), None);
1109
1110 let resp = app
1111 .oneshot(with_connect_info("/api/status", None))
1112 .await
1113 .unwrap();
1114 assert_eq!(resp.status(), StatusCode::OK);
1115 }
1116
1117 #[tokio::test]
1118 async fn valid_token_allows_request() {
1119 let pool = crate::db::connect_in_memory().await.unwrap();
1120 let app = router(pool, test_config(Some("secret123")), None);
1121
1122 let resp = app
1123 .oneshot(with_connect_info("/api/status", Some("Bearer secret123")))
1124 .await
1125 .unwrap();
1126 assert_eq!(resp.status(), StatusCode::OK);
1127 }
1128
1129 #[tokio::test]
1130 async fn dashboard_uses_cookie_not_embedded_token() {
1131 // SERIOUS #4: GET / must NOT ship the api_token in the page, and must set
1132 // an httpOnly session cookie that authenticates the dashboard's /api/* calls.
1133 let pool = crate::db::connect_in_memory().await.unwrap();
1134 let mut config = test_config(Some("supersecret-token"));
1135 config.serve.dashboard = true;
1136 let app = router(pool, config, None);
1137
1138 let req = HttpRequest::builder().uri("/").body(Body::empty()).unwrap();
1139 let resp = app.clone().oneshot(req).await.unwrap();
1140 assert_eq!(resp.status(), StatusCode::OK);
1141
1142 let cookie_hdr = resp
1143 .headers()
1144 .get(axum::http::header::SET_COOKIE)
1145 .expect("dashboard must set a session cookie")
1146 .to_str()
1147 .unwrap()
1148 .to_string();
1149 assert!(cookie_hdr.contains("pom_dash="));
1150 assert!(cookie_hdr.contains("HttpOnly"));
1151 assert!(
1152 !cookie_hdr.contains("supersecret-token"),
1153 "cookie must not be the api_token"
1154 );
1155
1156 let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
1157 .await
1158 .unwrap();
1159 let html = String::from_utf8_lossy(&body);
1160 assert!(
1161 !html.contains("supersecret-token"),
1162 "the api_token must never appear in served HTML"
1163 );
1164
1165 // The issued cookie authenticates an /api/* call without any bearer token.
1166 let dash = cookie_hdr.split(';').next().unwrap().trim().to_string(); // "pom_dash=<value>"
1167 let mut api_req = HttpRequest::builder()
1168 .uri("/api/status")
1169 .header(axum::http::header::COOKIE, dash)
1170 .body(Body::empty())
1171 .unwrap();
1172 api_req
1173 .extensions_mut()
1174 .insert(axum::extract::ConnectInfo(std::net::SocketAddr::from((
1175 [127, 0, 0, 1],
1176 40001,
1177 ))));
1178 let api_resp = app.oneshot(api_req).await.unwrap();
1179 assert_eq!(
1180 api_resp.status(),
1181 StatusCode::OK,
1182 "dashboard cookie must authenticate /api/*"
1183 );
1184 }
1185
1186 #[tokio::test]
1187 async fn wrong_token_returns_401() {
1188 let pool = crate::db::connect_in_memory().await.unwrap();
1189 let app = router(pool, test_config(Some("secret123")), None);
1190
1191 let req = HttpRequest::builder()
1192 .uri("/api/status")
1193 .header("authorization", "Bearer wrong-token")
1194 .body(Body::empty())
1195 .unwrap();
1196 let resp = app.oneshot(req).await.unwrap();
1197 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1198 }
1199
1200 #[tokio::test]
1201 async fn missing_header_returns_401() {
1202 let pool = crate::db::connect_in_memory().await.unwrap();
1203 let app = router(pool, test_config(Some("secret123")), None);
1204
1205 let req = HttpRequest::builder()
1206 .uri("/api/status")
1207 .body(Body::empty())
1208 .unwrap();
1209 let resp = app.oneshot(req).await.unwrap();
1210 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1211 }
1212
1213 #[tokio::test]
1214 async fn malformed_header_returns_401() {
1215 let pool = crate::db::connect_in_memory().await.unwrap();
1216 let app = router(pool, test_config(Some("secret123")), None);
1217
1218 let req = HttpRequest::builder()
1219 .uri("/api/status")
1220 .header("authorization", "Basic dXNlcjpwYXNz")
1221 .body(Body::empty())
1222 .unwrap();
1223 let resp = app.oneshot(req).await.unwrap();
1224 assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
1225 }
1226
1227 fn ip(n: u8) -> std::net::IpAddr {
1228 std::net::IpAddr::V4(std::net::Ipv4Addr::new(10, 0, 0, n))
1229 }
1230
1231 #[test]
1232 fn reproject_drops_injected_structure() {
1233 // #6: a compromised peer stuffs extra structure into its status blob.
1234 let hostile = serde_json::json!({
1235 "instance": { "id": "abc", "version": "9.9", "evil_field": {"x": 1} },
1236 "targets": { "mnw": { "status": "operational", "response_time_ms": 5, "evil": "inject" } },
1237 "peers": { "p2": { "status": "up", "latency_ms": 3 } },
1238 "top_level_injection": [1, 2, 3]
1239 });
1240 let clean = reproject_peer_status(&hostile);
1241
1242 // Only the fixed top-level keys survive.
1243 let obj = clean.as_object().unwrap();
1244 let mut keys: Vec<&String> = obj.keys().collect();
1245 keys.sort();
1246 assert_eq!(keys, vec!["instance", "peers", "targets"]);
1247 assert!(clean.get("top_level_injection").is_none());
1248
1249 // Known values are preserved; injected sibling keys are gone.
1250 assert_eq!(clean["instance"]["id"], "abc");
1251 assert_eq!(clean["instance"]["version"], "9.9");
1252 assert!(clean["instance"].get("evil_field").is_none());
1253 assert_eq!(clean["targets"]["mnw"]["status"], "operational");
1254 assert!(clean["targets"]["mnw"].get("evil").is_none());
1255 assert_eq!(clean["peers"]["p2"]["latency_ms"], 3);
1256 }
1257
1258 /// Build a GET request carrying a `ConnectInfo<SocketAddr>` extension, which
1259 /// the real server injects via `into_make_service_with_connect_info` but
1260 /// `oneshot` does not, the rate-limit layer extracts it.
1261 fn with_connect_info(uri: &str, bearer: Option<&str>) -> HttpRequest<Body> {
1262 let mut b = HttpRequest::builder().uri(uri);
1263 if let Some(h) = bearer {
1264 b = b.header("authorization", h);
1265 }
1266 let mut req = b.body(Body::empty()).unwrap();
1267 req.extensions_mut()
1268 .insert(axum::extract::ConnectInfo(std::net::SocketAddr::from((
1269 [127, 0, 0, 1],
1270 40000,
1271 ))));
1272 req
1273 }
1274
1275 #[test]
1276 fn rate_limiter_allows_within_limit() {
1277 let limiter = PerIpRateLimiter::new(3, std::time::Duration::from_mins(1));
1278 assert!(limiter.try_acquire(ip(1)));
1279 assert!(limiter.try_acquire(ip(1)));
1280 assert!(limiter.try_acquire(ip(1)));
1281 }
1282
1283 #[test]
1284 fn rate_limiter_blocks_over_limit() {
1285 let limiter = PerIpRateLimiter::new(2, std::time::Duration::from_mins(1));
1286 assert!(limiter.try_acquire(ip(1)));
1287 assert!(limiter.try_acquire(ip(1)));
1288 assert!(!limiter.try_acquire(ip(1)));
1289 }
1290
1291 #[test]
1292 fn rate_limiter_isolates_clients_by_ip() {
1293 // SERIOUS #5: one client exhausting its bucket must not affect another.
1294 let limiter = PerIpRateLimiter::new(1, std::time::Duration::from_mins(1));
1295 assert!(limiter.try_acquire(ip(1)));
1296 assert!(
1297 !limiter.try_acquire(ip(1)),
1298 "ip(1) is now over its own limit"
1299 );
1300 assert!(limiter.try_acquire(ip(2)), "ip(2) has its own fresh bucket");
1301 }
1302
1303 #[tokio::test]
1304 async fn rate_limiter_resets_after_window() {
1305 let limiter = PerIpRateLimiter::new(1, std::time::Duration::from_millis(10));
1306 assert!(limiter.try_acquire(ip(1)));
1307 assert!(!limiter.try_acquire(ip(1)));
1308 tokio::time::sleep(std::time::Duration::from_millis(15)).await;
1309 assert!(limiter.try_acquire(ip(1)));
1310 }
1311 }
1312