Skip to main content

max / makenotwork

Add inbound infra-alert ingestion endpoint POST /api/internal/alerts lets external monitoring agents (PoM, MT) push ops alerts (health failures, TLS expiry, DNS changes) to the operator. Previously there was no inbound path: the health monitor only emailed on this server's own status transitions. Auth is a dedicated ALERTS_INGEST_TOKEN bearer (AlertsAuth), separate from cli_service_token so a leak on an agent host can't reach the CLI internal API. Each accepted alert is persisted to the new admin_alerts table and emailed to ALERT_EMAIL; a mail failure is logged, not returned as 5xx, so an agent won't retry and duplicate the row. Admin-routed only, never creator-facing. kind and severity are typed enums, rejected at deserialization rather than by a runtime membership check. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-17 18:58 UTC
Commit: f49bd61f003b7be225f7a1ac2124f2ea0d270bb0
Parent: bb06d0e
7 files changed, +310 insertions, -0 deletions
A server/migrations/169_admin_alerts.sql +33
@@ -0,0 +1,33 @@
1 + -- Inbound infra-alert ingestion (pom / MT / other monitoring agents -> MNW).
2 + --
3 + -- External monitoring agents (PoM health/TLS/DNS watchers, the Multithreaded
4 + -- forum's own probes) had no way to reach an operator: the health monitor in
5 + -- `monitor.rs` only emails on *this server's* status transitions, and there is
6 + -- no admin notification center. This table is the durable log behind the
7 + -- `POST /api/internal/alerts` ingestion endpoint: each accepted alert is
8 + -- persisted here and also emailed to ALERT_EMAIL. Admin-routed only -- these are
9 + -- ops signals for the operator, never creator-facing.
10 + --
11 + -- `source` names the reporting agent ("pom", "mt", ...). `kind` is the alert
12 + -- class the agent assigns ("health", "tls_expiry", "dns_change") and `severity`
13 + -- the urgency ("critical"/"warning"/"info"); both are closed sets enforced at
14 + -- the application layer. Stored as TEXT (not an enum type) so extending either
15 + -- set is a code change, not a migration.
16 + -- `dedup_key` lets an agent collapse repeated firings of the same condition
17 + -- (e.g. one row per (source, dedup_key) flap) without the server modeling agent
18 + -- state; it is nullable for one-shot alerts.
19 + CREATE TABLE admin_alerts (
20 + id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
21 + source TEXT NOT NULL,
22 + kind TEXT NOT NULL,
23 + severity TEXT NOT NULL,
24 + title TEXT NOT NULL,
25 + body TEXT NOT NULL,
26 + dedup_key TEXT,
27 + details JSONB,
28 + received_at TIMESTAMPTZ NOT NULL DEFAULT now(),
29 + emailed BOOLEAN NOT NULL DEFAULT false
30 + );
31 +
32 + -- The admin log view is "newest first", so index the sort key.
33 + CREATE INDEX idx_admin_alerts_received ON admin_alerts (received_at DESC);
M server/src/auth.rs +37
@@ -455,6 +455,43 @@ impl FromRequestParts<crate::AppState> for ServiceAuth {
455 455 }
456 456 }
457 457
458 + /// Extractor for inbound infra-alert ingestion (PoM / MT monitoring agents →
459 + /// `POST /api/internal/alerts`).
460 + ///
461 + /// Validates `Authorization: Bearer {token}` against
462 + /// `config.integrations.alerts_ingest_token`. Returns 401 if the token is
463 + /// missing/invalid, 503 if the token is not configured. Deliberately keyed on a
464 + /// token separate from `ServiceAuth`'s `cli_service_token`: these agents run on
465 + /// other hosts (astra, prod), so a leak there must not grant CLI internal-API
466 + /// access.
467 + pub struct AlertsAuth;
468 +
469 + impl FromRequestParts<crate::AppState> for AlertsAuth {
470 + type Rejection = AppError;
471 +
472 + async fn from_request_parts(
473 + parts: &mut Parts,
474 + state: &crate::AppState,
475 + ) -> Result<Self, Self::Rejection> {
476 + let expected = state.config.integrations.alerts_ingest_token.as_deref().ok_or_else(|| {
477 + AppError::ServiceUnavailable("Alert ingestion not configured".to_string())
478 + })?;
479 +
480 + let header = parts
481 + .headers
482 + .get("authorization")
483 + .and_then(|v| v.to_str().ok())
484 + .and_then(|v| v.strip_prefix("Bearer "))
485 + .ok_or(AppError::Unauthorized)?;
486 +
487 + if !constant_time_compare(header, expected) {
488 + return Err(AppError::Unauthorized);
489 + }
490 +
491 + Ok(AlertsAuth)
492 + }
493 + }
494 +
458 495 /// The SSH-authenticated user identity behind an internal-API request.
459 496 ///
460 497 /// Sourced from the `X-MNW-Actor` header, a signed assertion the server mints
M server/src/config.rs +22
@@ -133,6 +133,10 @@ pub struct IntegrationsConfig {
133 133 pub internal_shared_secret: Option<String>,
134 134 /// Bearer token authenticating CLI SSH server → MNW internal API calls (>=32 chars).
135 135 pub cli_service_token: Option<String>,
136 + /// Bearer token authenticating inbound infra alerts (PoM/MT → `POST
137 + /// /api/internal/alerts`) (>=32 chars). Distinct from `cli_service_token` so
138 + /// a leak on a monitoring agent can't reach the CLI internal API.
139 + pub alerts_ingest_token: Option<String>,
136 140 }
137 141
138 142 /// Upstream OAuth provider config for delegated login (`SSO_*`).
@@ -445,6 +449,19 @@ impl Config {
445 449 Err(_) => None,
446 450 };
447 451
452 + // Inbound infra-alert ingestion token (PoM/MT monitoring agents →
453 + // `POST /api/internal/alerts`). Same 32-char floor as the other service
454 + // secrets; deliberately separate from CLI_SERVICE_TOKEN.
455 + let alerts_ingest_token = match std::env::var("ALERTS_INGEST_TOKEN") {
456 + Ok(secret) => {
457 + if secret.len() < 32 {
458 + return Err(ConfigError::WeakAlertsIngestToken);
459 + }
460 + Some(secret)
461 + }
462 + Err(_) => None,
463 + };
464 +
448 465 // WAM ticket manager URL (tailnet, e.g. "http://100.x.x.x:7890")
449 466 let wam_url = std::env::var("WAM_URL").ok();
450 467
@@ -500,6 +517,7 @@ impl Config {
500 517 wam_url,
501 518 internal_shared_secret,
502 519 cli_service_token,
520 + alerts_ingest_token,
503 521 },
504 522 })
505 523 }
@@ -744,6 +762,7 @@ impl std::fmt::Debug for Config {
744 762 .field("postmark_inbound_webhook_token", &self.email_webhooks.inbound_webhook_token.as_ref().map(|_| "[REDACTED]"))
745 763 .field("internal_shared_secret", &self.integrations.internal_shared_secret.as_ref().map(|_| "[REDACTED]"))
746 764 .field("cli_service_token", &self.integrations.cli_service_token.as_ref().map(|_| "[REDACTED]"))
765 + .field("alerts_ingest_token", &self.integrations.alerts_ingest_token.as_ref().map(|_| "[REDACTED]"))
747 766 .field("wam_url", &self.integrations.wam_url)
748 767 .field("access_gate", &self.access_gate)
749 768 .field("sso", &self.sso.as_ref().map(|s| &s.provider_url))
@@ -791,6 +810,8 @@ pub enum ConfigError {
791 810 WeakInternalSecret,
792 811 #[error("CLI_SERVICE_TOKEN must be at least 32 characters long")]
793 812 WeakCliServiceToken,
813 + #[error("ALERTS_INGEST_TOKEN must be at least 32 characters long")]
814 + WeakAlertsIngestToken,
794 815 #[error("CDN_BASE_URL is required in production (HOST=0.0.0.0 or HTTPS HOST_URL detected). Without a CDN, cover/download URLs are presigned S3 URLs the storage key-derivation logic does not support. Set CDN_BASE_URL to your CDN origin.")]
795 816 MissingCdnBaseUrl,
796 817 #[error("S3_PUBLIC_BUCKET is required in production when storage is configured. The CDN serves ONLY the public bucket; promoted image content (covers, gallery, item/project images) is copied there. Set S3_PUBLIC_BUCKET to the public, world-readable bucket name.")]
@@ -920,6 +941,7 @@ mod tests {
920 941 wam_url: None,
921 942 internal_shared_secret: None,
922 943 cli_service_token: None,
944 + alerts_ingest_token: None,
923 945 },
924 946 };
925 947 let addr = config.socket_addr();
A server/src/db/admin_alerts.rs +95
@@ -0,0 +1,95 @@
1 + //! Inbound infra-alert log (see migration `169_admin_alerts.sql`).
2 + //!
3 + //! Backs the `POST /api/internal/alerts` ingestion endpoint: external
4 + //! monitoring agents (PoM, MT) push ops alerts here, each persisted as one row
5 + //! and emailed to the operator. Admin-routed only -- never creator-facing.
6 +
7 + use serde::Deserialize;
8 + use sqlx::PgPool;
9 +
10 + /// Alert class an agent may report. Closed set, deserialized straight from the
11 + /// wire (`#[serde(rename_all = "snake_case")]` maps `TlsExpiry` → `"tls_expiry"`
12 + /// etc.), so an unknown value is rejected when the request body is parsed —
13 + /// there is no runtime membership check. Extending the taxonomy is a variant
14 + /// here plus its `as_str`, not a migration (the column is TEXT).
15 + #[derive(Debug, Clone, Copy, Deserialize)]
16 + #[serde(rename_all = "snake_case")]
17 + pub enum AlertKind {
18 + Health,
19 + TlsExpiry,
20 + DnsChange,
21 + }
22 +
23 + impl AlertKind {
24 + /// Canonical wire/storage token. Matches the `serde(rename_all)` mapping.
25 + pub fn as_str(self) -> &'static str {
26 + match self {
27 + Self::Health => "health",
28 + Self::TlsExpiry => "tls_expiry",
29 + Self::DnsChange => "dns_change",
30 + }
31 + }
32 + }
33 +
34 + /// Alert urgency, most to least. Closed set, deserialized like [`AlertKind`].
35 + #[derive(Debug, Clone, Copy, Deserialize)]
36 + #[serde(rename_all = "snake_case")]
37 + pub enum AlertSeverity {
38 + Critical,
39 + Warning,
40 + Info,
41 + }
42 +
43 + impl AlertSeverity {
44 + /// Canonical wire/storage token.
45 + pub fn as_str(self) -> &'static str {
46 + match self {
47 + Self::Critical => "critical",
48 + Self::Warning => "warning",
49 + Self::Info => "info",
50 + }
51 + }
52 + }
53 +
54 + /// A validated inbound alert, ready to persist.
55 + pub struct NewAlert<'a> {
56 + pub source: &'a str,
57 + pub kind: AlertKind,
58 + pub severity: AlertSeverity,
59 + pub title: &'a str,
60 + pub body: &'a str,
61 + pub dedup_key: Option<&'a str>,
62 + pub details: Option<&'a serde_json::Value>,
63 + }
64 +
65 + /// Insert an ingested alert and return its generated id. `emailed` starts false;
66 + /// the handler flips it via [`mark_emailed`] after the notification is sent so a
67 + /// mail failure stays visible in the log.
68 + #[tracing::instrument(skip_all, fields(source = alert.source, kind = alert.kind.as_str(), severity = alert.severity.as_str()))]
69 + pub async fn insert_alert(pool: &PgPool, alert: &NewAlert<'_>) -> Result<i64, sqlx::Error> {
70 + let id: i64 = sqlx::query_scalar(
71 + "INSERT INTO admin_alerts (source, kind, severity, title, body, dedup_key, details)
72 + VALUES ($1, $2, $3, $4, $5, $6, $7)
73 + RETURNING id",
74 + )
75 + .bind(alert.source)
76 + .bind(alert.kind.as_str())
77 + .bind(alert.severity.as_str())
78 + .bind(alert.title)
79 + .bind(alert.body)
80 + .bind(alert.dedup_key)
81 + .bind(alert.details)
82 + .fetch_one(pool)
83 + .await?;
84 + Ok(id)
85 + }
86 +
87 + /// Mark an alert row as successfully emailed to the operator.
88 + #[tracing::instrument(skip_all)]
89 + pub async fn mark_emailed(pool: &PgPool, id: i64) -> Result<(), sqlx::Error> {
90 + sqlx::query("UPDATE admin_alerts SET emailed = true WHERE id = $1")
91 + .bind(id)
92 + .execute(pool)
93 + .await?;
94 + Ok(())
95 + }
M server/src/db/mod.rs +1
@@ -37,6 +37,7 @@ pub(crate) mod totp;
37 37 pub mod passkeys; // pub so the integration test crate can exercise the layer directly
38 38 pub(crate) mod health;
39 39 pub(crate) mod monitor;
40 + pub mod admin_alerts;
40 41 pub mod scanning; // pub so the integration test crate can exercise the quarantine purge/stamp layer directly
41 42 pub mod scan_jobs; // pub so the integration test crate can exercise the reaper/heartbeat layer directly
42 43 pub(crate) mod scan_admin_actions;
A server/src/routes/api/internal/alerts.rs +114
@@ -0,0 +1,114 @@
1 + //! Inbound infra-alert ingestion: PoM / MT monitoring agents → MNW.
2 + //!
3 + //! `POST /api/internal/alerts` accepts an ops alert (health failure, TLS
4 + //! expiry, DNS change, ...) from an external monitoring agent, persists it to
5 + //! `admin_alerts`, and emails the operator (`ALERT_EMAIL`). Admin-routed only:
6 + //! these are operator signals, never creator-facing. Authed by [`AlertsAuth`], a
7 + //! bearer token distinct from the CLI internal API's `ServiceAuth`.
8 +
9 + use axum::{extract::State, response::IntoResponse, Json};
10 + use serde::Deserialize;
11 + use sqlx::PgPool;
12 +
13 + use crate::{
14 + auth::AlertsAuth,
15 + db::{
16 + self,
17 + admin_alerts::{AlertKind, AlertSeverity},
18 + },
19 + email::EmailClient,
20 + error::{AppError, Result},
21 + };
22 +
23 + #[derive(Deserialize)]
24 + pub(super) struct AlertRequest {
25 + /// Reporting agent, e.g. "pom" or "mt".
26 + source: String,
27 + /// Alert class. Unknown values are rejected at deserialization.
28 + kind: AlertKind,
29 + /// Alert urgency. Unknown values are rejected at deserialization.
30 + severity: AlertSeverity,
31 + /// Short human-readable headline.
32 + title: String,
33 + /// Full alert detail.
34 + body: String,
35 + /// Optional agent-side collapse key for repeated firings of one condition.
36 + #[serde(default)]
37 + dedup_key: Option<String>,
38 + /// Optional structured context (JSON object), stored verbatim.
39 + #[serde(default)]
40 + details: Option<serde_json::Value>,
41 + }
42 +
43 + fn require_field(name: &str, value: &str, max: usize) -> Result<()> {
44 + if value.is_empty() {
45 + return Err(AppError::validation(format!("{name} is required")));
46 + }
47 + if value.len() > max {
48 + return Err(AppError::validation(format!("{name} must be at most {max} characters")));
49 + }
50 + Ok(())
51 + }
52 +
53 + /// POST /api/internal/alerts
54 + ///
55 + /// Persist an inbound alert and email the operator. Returns 202 with the row id.
56 + /// A missing `ALERT_EMAIL` is not an error: the alert is still logged (the row's
57 + /// `emailed` stays false so the gap is visible), because dropping the durable
58 + /// record just because mail is unconfigured would defeat the point.
59 + pub(super) async fn ingest_alert(
60 + State(pool): State<PgPool>,
61 + State(email): State<EmailClient>,
62 + _auth: AlertsAuth,
63 + Json(req): Json<AlertRequest>,
64 + ) -> Result<impl IntoResponse> {
65 + // `kind` and `severity` are already validated by the type system: the
66 + // request would not have deserialized if either were outside its enum. Only
67 + // the free-text fields need bounds here.
68 + require_field("source", &req.source, 64)?;
69 + require_field("title", &req.title, 200)?;
70 + require_field("body", &req.body, 5000)?;
71 + if let Some(key) = &req.dedup_key
72 + && key.len() > 200
73 + {
74 + return Err(AppError::validation("dedup_key must be at most 200 characters".to_string()));
75 + }
76 +
77 + let id = db::admin_alerts::insert_alert(
78 + &pool,
79 + &db::admin_alerts::NewAlert {
80 + source: &req.source,
81 + kind: req.kind,
82 + severity: req.severity,
83 + title: &req.title,
84 + body: &req.body,
85 + dedup_key: req.dedup_key.as_deref(),
86 + details: req.details.as_ref(),
87 + },
88 + )
89 + .await?;
90 +
91 + match std::env::var("ALERT_EMAIL").ok() {
92 + Some(to) if !to.is_empty() => {
93 + let subject = format!("[{}/{}] {}", req.source, req.severity.as_str(), req.title);
94 + let mail_body = format!(
95 + "{}\n\nsource: {}\nkind: {}\nseverity: {}\n",
96 + req.body, req.source, req.kind.as_str(), req.severity.as_str()
97 + );
98 + match email.send_alert(&to, &subject, &mail_body).await {
99 + Ok(()) => {
100 + if let Err(e) = db::admin_alerts::mark_emailed(&pool, id).await {
101 + tracing::error!(alert_id = id, error = ?e, "alert emailed but mark_emailed failed");
102 + }
103 + }
104 + // Ingestion succeeded and the row is durable; a mail failure is
105 + // logged, not surfaced as a 5xx to the agent (which would make it
106 + // retry and duplicate the row).
107 + Err(e) => tracing::error!(alert_id = id, error = ?e, "failed to email ingested alert"),
108 + }
109 + }
110 + _ => tracing::warn!(alert_id = id, "ALERT_EMAIL unset; alert logged but not emailed"),
111 + }
112 +
113 + Ok((axum::http::StatusCode::ACCEPTED, Json(serde_json::json!({ "id": id }))))
114 + }
M server/src/routes/api/internal/mod.rs +8
@@ -3,6 +3,7 @@
3 3 //! These endpoints are protected by `ServiceAuth` (Bearer token) and are
4 4 //! called by the CLI SSH server running on the same host.
5 5
6 + mod alerts;
6 7 mod cli_features;
7 8 mod content;
8 9 mod creators;
@@ -75,4 +76,11 @@ pub(super) fn internal_routes() -> CsrfRouter<AppState> {
75 76 // CLI features: custom domains
76 77 .route("/api/internal/creator/domain", with_csrf_skip(INTERNAL_SKIP, get(cli_features::get_domain).post(cli_features::add_domain).delete(cli_features::remove_domain)))
77 78 .route("/api/internal/creator/domain/verify", post_csrf_skip(INTERNAL_SKIP, cli_features::verify_domain))
79 + // Inbound infra alerts (PoM/MT). Bearer-authed via `AlertsAuth` on a
80 + // dedicated token, not `ServiceAuth`; no session, so CSRF is skipped.
81 + .route("/api/internal/alerts", post_csrf_skip(ALERTS_SKIP, alerts::ingest_alert))
78 82 }
83 +
84 + /// The alerts ingestion route is bearer-authed via `AlertsAuth` (dedicated
85 + /// token, no session); CSRF is not applicable.
86 + const ALERTS_SKIP: &str = "internal alerts: AlertsAuth bearer, no session";