swarm-logs-auth: check agent tokens on the log store's query route

A new `auth_request` responder on the log store's host. nginx sends it
every `/select/logsql/` request whose `Authorization` is
`Bearer swarm-agent.<agent>.<secret>`; it answers 200 when the secret is
the one stored at `swarm/agents/<agent>/queue` (`swarm-agent-verify`),
401 otherwise. Every other bearer still goes to authelia, so agents
still presenting the hive's shared OIDC client keep working.

`auth_request` takes a fixed URI, so the split is a `map` on
`$http_authorization` plus an internal location that rewrites to the
map's pick. Checked against nginx 1.30.5 with stub upstreams: an agent
token only the stub validator accepts gets 200, a token only the stub
authelia accepts gets 200, every wrong one 401, and the internal
locations 404 from outside.

The responder serves a unix socket in its own traverse-only runtime
directory and reads the store under its own leaf
(`swarm-logs-auth`), whose policy is `swarm/agents/+/queue`, read-only.
glue-bao-tls mints the leaf where the store runs; elsewhere the
operator places it, and eval refuses a log store host without one.

Part of #4459.

swarm-logs-auth/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "swarm-logs-auth"
+version.workspace = true
+edition.workspace = true
+
+[[bin]]
+name = "swarm-logs-auth"
+path = "src/main.rs"
+
+[dependencies]
+anyhow.workspace = true
+axum.workspace = true
+clap.workspace = true
+swarm-agent-verify.workspace = true
+# The agent-token spelling, and the rustls provider the store client needs.
+swarm-queue-client.workspace = true
+swarm-secret-client.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+tracing-subscriber.workspace = true
+
+[lints]
+workspace = true

swarm-logs-auth/src/main.rs

@@ -0,0 +1,160 @@
+//! The log store gateway's `auth_request` responder for agents.
+//!
+//! nginx sends it every `/select/logsql/` request whose `Authorization` is
+//! `Bearer swarm-agent.<agent>.<secret>` (`swarm-victorialogs.nix`). It answers
+//! 200 when the secret is the one stored at `swarm/agents/<agent>/queue`, and
+//! 401 otherwise, store failures included: `auth_request` turns any other
+//! status into a 500 for the caller.
+//!
+//! It reads the store under its own certificate, whose policy reaches every
+//! agent's `queue` secret and nothing else. It serves a unix socket whose
+//! directory only the gateway's nginx may enter.
+
+use std::os::unix::fs::PermissionsExt as _;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::Context as _;
+use axum::extract::State;
+use axum::http::{HeaderMap, StatusCode, header};
+use clap::Parser;
+use swarm_agent_verify::CredentialSource;
+use swarm_queue_client::agent_token::parse_agent_token;
+
+/// How long one store lookup may take before the request is denied. Well
+/// under nginx's 60s `proxy_read_timeout`, so a slow store reads as a 401
+/// rather than a gateway timeout.
+const LOOKUP_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[derive(Parser)]
+#[command(about = "auth_request responder: checks an agent's queue secret against the store")]
+struct Args {
+    /// Unix socket to serve on.
+    #[arg(long)]
+    socket: PathBuf,
+    /// Cert-auth role this responder logs in to the store as.
+    #[arg(long, default_value = "swarm-logs-auth")]
+    store_cert_role: String,
+}
+
+/// The answer for one subrequest's `Authorization` header.
+async fn decide(source: &impl CredentialSource, authorization: Option<&str>) -> StatusCode {
+    let Some(token) = authorization.and_then(|v| v.strip_prefix("Bearer ")) else {
+        tracing::warn!("no bearer token; denying");
+        return StatusCode::UNAUTHORIZED;
+    };
+    let token = match parse_agent_token(token) {
+        Some(Ok(token)) => token,
+        Some(Err(e)) => {
+            tracing::warn!(error = %e, "malformed agent token; denying");
+            return StatusCode::UNAUTHORIZED;
+        }
+        None => {
+            tracing::warn!("not an agent token; denying");
+            return StatusCode::UNAUTHORIZED;
+        }
+    };
+    if swarm_agent_verify::verify(source, &token, LOOKUP_TIMEOUT).await {
+        tracing::info!(agent = token.agent, "verified");
+        StatusCode::OK
+    } else {
+        StatusCode::UNAUTHORIZED
+    }
+}
+
+async fn handle(
+    State(store): State<Arc<swarm_agent_verify::Store>>,
+    headers: HeaderMap,
+) -> StatusCode {
+    let authorization = headers
+        .get(header::AUTHORIZATION)
+        .and_then(|v| v.to_str().ok());
+    decide(store.as_ref(), authorization).await
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    swarm_queue_client::install_crypto_provider();
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::try_from_default_env()
+                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
+        )
+        // stdout is journald, which keeps ANSI escapes as raw bytes.
+        .with_ansi(false)
+        .init();
+    let args = Args::parse();
+
+    let settings = swarm_secret_client::client::Settings::from_env()
+        .context("reading the secret store's address and this responder's identity")?;
+    let store = Arc::new(swarm_agent_verify::Store::new(
+        settings,
+        args.store_cert_role,
+    ));
+
+    // `RuntimeDirectoryPreserve=yes` can leave the previous run's socket; the
+    // directory is this unit's alone, so unlinking it is safe.
+    if let Err(e) = std::fs::remove_file(&args.socket)
+        && e.kind() != std::io::ErrorKind::NotFound
+    {
+        return Err(e)
+            .with_context(|| format!("clearing stale socket at {}", args.socket.display()));
+    }
+    let listener = tokio::net::UnixListener::bind(&args.socket)
+        .with_context(|| format!("binding {}", args.socket.display()))?;
+    // nginx is another user and `connect(2)` needs write. The directory is
+    // the access control, not this mode.
+    std::fs::set_permissions(&args.socket, std::fs::Permissions::from_mode(0o666))
+        .with_context(|| format!("chmod {}", args.socket.display()))?;
+    tracing::info!(socket = %args.socket.display(), "swarm-logs-auth listening");
+
+    let app = axum::Router::new().fallback(handle).with_state(store);
+    axum::serve(listener, app).await.context("serving")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use swarm_secret_client::queue::AgentCredential;
+
+    /// A store holding atlas's credential.
+    struct Atlas;
+
+    impl CredentialSource for Atlas {
+        async fn lookup(&self, agent: &str) -> anyhow::Result<Option<AgentCredential>> {
+            Ok((agent == "atlas").then(|| AgentCredential {
+                value: "Ab9_-zSECRET".to_owned(),
+                agent: "atlas".to_owned(),
+                minted_at: None,
+            }))
+        }
+    }
+
+    #[tokio::test]
+    async fn the_stored_secret_is_200() {
+        assert_eq!(
+            decide(&Atlas, Some("Bearer swarm-agent.atlas.Ab9_-zSECRET")).await,
+            StatusCode::OK
+        );
+    }
+
+    #[tokio::test]
+    async fn everything_else_is_401() {
+        for authorization in [
+            None,
+            Some("swarm-agent.atlas.Ab9_-zSECRET"),
+            Some("Basic swarm-agent.atlas.Ab9_-zSECRET"),
+            Some("Bearer swarm-agent.atlas.Ab9_-zSECREX"),
+            Some("Bearer swarm-agent.argus.Ab9_-zSECRET"),
+            Some("Bearer swarm-agent.atlas"),
+            Some("Bearer authelia_at_abc.def"),
+        ] {
+            assert_eq!(
+                decide(&Atlas, authorization).await,
+                StatusCode::UNAUTHORIZED,
+                "{authorization:?}"
+            );
+        }
+    }
+}