From d46c8d49c6036a0c738ad72efa4c527d327ea657 Mon Sep 17 00:00:00 2001 From: Hampus Date: Tue, 8 Sep 2026 14:51:34 +0200 Subject: [PATCH] fix(svc): authenticate to nats with the configured token (#2581) --- deploy/self-hosting/docker-compose.yml | 1 + fluxer_docs/src/content/docs/operator/configuration.mdx | 2 +- fluxer_gifs/src/main.rs | 3 ++- fluxer_messages/src/main.rs | 3 ++- fluxer_snowflakes/src/main.rs | 3 ++- fluxer_svc/src/config.rs | 4 ++++ fluxer_svc/src/lib.rs | 4 +++- fluxer_svc/src/router.rs | 1 + fluxer_svc/src/shard.rs | 1 + fluxer_svc/src/transport.rs | 7 ++++++- fluxer_unfurl/src/main.rs | 3 ++- fluxer_users/src/main.rs | 3 ++- 12 files changed, 27 insertions(+), 8 deletions(-) diff --git a/deploy/self-hosting/docker-compose.yml b/deploy/self-hosting/docker-compose.yml index 1edb78ce0..2bb82639a 100644 --- a/deploy/self-hosting/docker-compose.yml +++ b/deploy/self-hosting/docker-compose.yml @@ -26,6 +26,7 @@ x-fluxer-env: &fluxer-env FLUXER_KV_URL: redis://valkey:6379/0 FLUXER_NATS_URL: nats://nats:4222 FLUXER_NATS_JETSTREAM_URL: nats://nats:4222 + FLUXER_NATS_AUTH_TOKEN: ${FLUXER_NATS_AUTH_TOKEN:-} FLUXER_SVC_NATS_URL: nats://nats:4222 FLUXER_SVC_SHARD_COUNT: "1" diff --git a/fluxer_docs/src/content/docs/operator/configuration.mdx b/fluxer_docs/src/content/docs/operator/configuration.mdx index 9b70e2eb5..02749c1de 100644 --- a/fluxer_docs/src/content/docs/operator/configuration.mdx +++ b/fluxer_docs/src/content/docs/operator/configuration.mdx @@ -581,7 +581,7 @@ Default `nats://127.0.0.1:4222`. The JetStream address. Read by `api` and `worke #### `FLUXER_NATS_AUTH_TOKEN` -Default empty. NATS authentication. Read by `api`, `worker`, and `gateway`. The shipped NATS runs without authentication. +Default empty. NATS authentication. Read by `api`, `worker`, `gateway`, and the five internal services. The shipped NATS runs without authentication, and Compose forwards this name to every container that connects to it. #### `FLUXER_SVC_NATS_URL` diff --git a/fluxer_gifs/src/main.rs b/fluxer_gifs/src/main.rs index 3d2666d6d..9f100a6ac 100644 --- a/fluxer_gifs/src/main.rs +++ b/fluxer_gifs/src/main.rs @@ -15,7 +15,8 @@ use shard_impl::GifsShard; async fn main() -> anyhow::Result<()> { fluxer_svc::init_tracing(); let config = ServiceConfig::from_env()?; - let transport = NatsTransport::connect(&config.nats_url).await?; + let transport = + NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?; tracing::info!( service = config.service_name, diff --git a/fluxer_messages/src/main.rs b/fluxer_messages/src/main.rs index 3682074e8..d92a44d0b 100644 --- a/fluxer_messages/src/main.rs +++ b/fluxer_messages/src/main.rs @@ -9,7 +9,8 @@ use fluxer_svc::transport::NatsTransport; async fn main() -> anyhow::Result<()> { fluxer_svc::init_tracing(); let config = ServiceConfig::from_env()?; - let transport = NatsTransport::connect(&config.nats_url).await?; + let transport = + NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?; tracing::info!( service = config.service_name, diff --git a/fluxer_snowflakes/src/main.rs b/fluxer_snowflakes/src/main.rs index 7c15bbb0f..890d8a61b 100644 --- a/fluxer_snowflakes/src/main.rs +++ b/fluxer_snowflakes/src/main.rs @@ -14,7 +14,8 @@ use types::SERVICE_NAME; async fn main() -> anyhow::Result<()> { fluxer_svc::init_tracing(); let config = ServiceConfig::from_env()?; - let transport = NatsTransport::connect(&config.nats_url).await?; + let transport = + NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?; tracing::info!( service = SERVICE_NAME, mode = ?config.mode, diff --git a/fluxer_svc/src/config.rs b/fluxer_svc/src/config.rs index a8d01dced..038a18072 100644 --- a/fluxer_svc/src/config.rs +++ b/fluxer_svc/src/config.rs @@ -17,6 +17,7 @@ pub struct ServiceConfig { pub shard_count: u32, pub listen_addr: SocketAddr, pub nats_url: String, + pub nats_auth_token: Option, pub cache_max_entries: u64, pub cache_ttl: Duration, pub cache_hard_ttl: Duration, @@ -105,6 +106,8 @@ impl ServiceConfig { let nats_url = optional_from(&get, "FLUXER_SVC_NATS_URL") .unwrap_or_else(|| "nats://127.0.0.1:4222".to_owned()); + let nats_auth_token = optional_from(&get, "FLUXER_NATS_AUTH_TOKEN"); + let cache_ttl_ms = optional_from(&get, "FLUXER_SVC_CACHE_TTL_MS") .map(|v| v.parse::()) .transpose()? @@ -165,6 +168,7 @@ impl ServiceConfig { shard_count, listen_addr: format!("{listen_host}:{listen_port}").parse()?, nats_url, + nats_auth_token, cache_max_entries: optional_from(&get, "FLUXER_SVC_CACHE_MAX_ENTRIES") .map(|v| v.parse::()) .transpose()? diff --git a/fluxer_svc/src/lib.rs b/fluxer_svc/src/lib.rs index f1610b7e5..123ff13a5 100644 --- a/fluxer_svc/src/lib.rs +++ b/fluxer_svc/src/lib.rs @@ -33,7 +33,9 @@ where { init_tracing(); let config = config::ServiceConfig::from_env()?; - let transport = transport::NatsTransport::connect(&config.nats_url).await?; + let transport = + transport::NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()) + .await?; tracing::info!( service = config.service_name, mode = ?config.mode, diff --git a/fluxer_svc/src/router.rs b/fluxer_svc/src/router.rs index 7bf8127a1..82f7e3d4e 100644 --- a/fluxer_svc/src/router.rs +++ b/fluxer_svc/src/router.rs @@ -868,6 +868,7 @@ mod tests { shard_count: 1, listen_addr: "127.0.0.1:0".parse().unwrap(), nats_url: "memory".to_owned(), + nats_auth_token: None, cache_max_entries: 100, cache_ttl: Duration::from_secs(30), cache_hard_ttl: Duration::from_secs(600), diff --git a/fluxer_svc/src/shard.rs b/fluxer_svc/src/shard.rs index b3efdcd5d..405f55bf7 100644 --- a/fluxer_svc/src/shard.rs +++ b/fluxer_svc/src/shard.rs @@ -417,6 +417,7 @@ mod tests { shard_count: 1, listen_addr: "127.0.0.1:0".parse().unwrap(), nats_url: "memory".to_owned(), + nats_auth_token: None, cache_max_entries: 100, cache_ttl: Duration::from_secs(30), cache_hard_ttl: Duration::from_secs(600), diff --git a/fluxer_svc/src/transport.rs b/fluxer_svc/src/transport.rs index 567dc2994..4b9611f0b 100644 --- a/fluxer_svc/src/transport.rs +++ b/fluxer_svc/src/transport.rs @@ -77,7 +77,7 @@ pub struct NatsMessage { } impl NatsTransport { - pub async fn connect(url: &str) -> anyhow::Result { + pub async fn connect(url: &str, auth_token: Option<&str>) -> anyhow::Result { let reconnect_notify = Arc::new(Notify::new()); let event_notify = reconnect_notify.clone(); @@ -112,6 +112,11 @@ impl NatsTransport { } }); + let options = match auth_token { + Some(token) => options.token(token.to_owned()), + None => options, + }; + let client = options .subscription_capacity(NATS_SUBSCRIPTION_CAPACITY) .connect(url) diff --git a/fluxer_unfurl/src/main.rs b/fluxer_unfurl/src/main.rs index 84771ccf9..550470a1b 100644 --- a/fluxer_unfurl/src/main.rs +++ b/fluxer_unfurl/src/main.rs @@ -27,7 +27,8 @@ use shard_impl::UnfurlShard; async fn main() -> anyhow::Result<()> { fluxer_svc::init_tracing(); let config = ServiceConfig::from_env()?; - let transport = NatsTransport::connect(&config.nats_url).await?; + let transport = + NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?; tracing::info!( service = config.service_name, diff --git a/fluxer_users/src/main.rs b/fluxer_users/src/main.rs index 85adc465b..fe644ae7f 100644 --- a/fluxer_users/src/main.rs +++ b/fluxer_users/src/main.rs @@ -13,7 +13,8 @@ use shard_impl::UsersShard; async fn main() -> anyhow::Result<()> { fluxer_svc::init_tracing(); let config = ServiceConfig::from_env()?; - let transport = NatsTransport::connect(&config.nats_url).await?; + let transport = + NatsTransport::connect(&config.nats_url, config.nats_auth_token.as_deref()).await?; tracing::info!( service = config.service_name,