00cff51ce5
All checks were successful
Closes the open broker (anonymous publish to any tenant's corrosion.*).
Per-license isolation via NATS user/password + subject permissions:
each license -> user=license_id, password=HMAC-SHA256(license_id,
NATS_TOKEN_SECRET), scoped to corrosion.{license_id}.> + _INBOX. Backend
uses a privileged internal user.
- Agent (alpha.5): nats_user/nats_password config + env, user_and_password
auth; falls back to token/anonymous (transition-safe)
- Backend: connects with NATS_INTERNAL_USER/PASSWORD when set, else anon
- scripts/generate-nats-auth.mjs: regenerates nats-auth.conf from the
licenses table; NATS_AUTH_STAGE=open keeps a no_auth_user fallback
(verify creds first), =enforce rejects anonymous
- committed nats-auth.conf is the SAFE OPEN default (no secrets); the
host copy carries real users and is not committed
- compose: NATS_INTERNAL_USER/PASSWORD/NATS_TOKEN_SECRET, mount nats-auth.conf
Entirely non-breaking until secrets+config deployed; staged cutover next.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
67 lines
2.4 KiB
Rust
67 lines
2.4 KiB
Rust
//! NATS connection layer.
|
|
//!
|
|
//! Connection parameters follow the production-proven Vigilance profile:
|
|
//! infinite reconnects with capped exponential backoff, 30s pings to detect
|
|
//! zombie TCP in ~60s, and a deep client-side send queue so telemetry buffers
|
|
//! through broker outages instead of erroring.
|
|
|
|
use anyhow::{Context, Result};
|
|
use std::time::Duration;
|
|
|
|
use crate::config::Settings;
|
|
|
|
pub async fn connect(cfg: &Settings) -> Result<async_nats::Client> {
|
|
let (url, force_tls) = normalize_url(&cfg.nats_url);
|
|
|
|
let mut opts = async_nats::ConnectOptions::new()
|
|
.name("corrosion-host-agent")
|
|
.retry_on_initial_connect()
|
|
.max_reconnects(None)
|
|
.ping_interval(Duration::from_secs(30))
|
|
.client_capacity(8192)
|
|
.reconnect_delay_callback(|attempts| {
|
|
Duration::from_millis(std::cmp::min(attempts as u64 * 100, 8_000))
|
|
})
|
|
.event_callback(|event| async move {
|
|
match event {
|
|
async_nats::Event::Disconnected => tracing::warn!("nats disconnected"),
|
|
async_nats::Event::Connected => tracing::info!("nats connected"),
|
|
other => tracing::debug!("nats event: {other}"),
|
|
}
|
|
});
|
|
|
|
if force_tls {
|
|
opts = opts.require_tls(true);
|
|
}
|
|
|
|
// Per-license auth: the broker maps user=license_id, password=derived
|
|
// token to permissions scoped to corrosion.{license_id}.>. Falls back to
|
|
// token-only or anonymous so the agent still works against a broker that
|
|
// hasn't enforced auth yet (transition period).
|
|
if let Some(password) = &cfg.nats_password {
|
|
let user = cfg.nats_user.clone().unwrap_or_else(|| cfg.license_id.clone());
|
|
opts = opts.user_and_password(user, password.clone());
|
|
} else if let Some(token) = &cfg.nats_token {
|
|
opts = opts.token(token.clone());
|
|
}
|
|
|
|
let client = opts
|
|
.connect(&url)
|
|
.await
|
|
.with_context(|| format!("connecting to NATS at {url}"))?;
|
|
|
|
Ok(client)
|
|
}
|
|
|
|
/// Accept `tls://` / `nats+tls://` URL schemes by translating to `nats://` +
|
|
/// an explicit TLS requirement.
|
|
fn normalize_url(raw: &str) -> (String, bool) {
|
|
if let Some(rest) = raw.strip_prefix("tls://") {
|
|
(format!("nats://{rest}"), true)
|
|
} else if let Some(rest) = raw.strip_prefix("nats+tls://") {
|
|
(format!("nats://{rest}"), true)
|
|
} else {
|
|
(raw.to_string(), false)
|
|
}
|
|
}
|