// SPDX-License-Identifier: AGPL-3.0-or-later use fluxer_common::config::{parse_bool, parse_number}; use anyhow::Context as _; use fluxer_common::secret::SecretString; use serde::Deserialize; use std::net::{IpAddr, SocketAddr}; pub const RPC_AUTH_HEADER: &str = "x-fluxer-rpc-auth"; pub const DEFAULT_APP_ID: &str = "stable"; const RPC_PATH: &str = "/internal/rpc"; const DEFAULT_HOST: &str = "0.0.0.0"; const DEFAULT_DELIVERY_PORT: u16 = 8126; const DEFAULT_RELAY_PORT: u16 = 8127; const DEFAULT_NATS_URL: &str = "nats://127.0.0.1:4222"; const DEFAULT_QUEUE_CAPACITY: usize = 10_000; const DEFAULT_SEND_CONCURRENCY: usize = 256; const DEFAULT_RELAY_MAX_CONCURRENT: usize = 1_024; const DEFAULT_RELAY_MAX_BODY_BYTES: usize = 2_816; const DEFAULT_TRUSTED_PROXY_HOPS: usize = 1; const DEFAULT_DEVICE_TOKEN_BUCKET_ENTRIES: usize = 1_000_000; const DEFAULT_DEVICE_TOKEN_BUCKET_PER_MINUTE: u32 = 60; const DEFAULT_DEVICE_TOKEN_BUCKET_BURST: u32 = 20; const DEFAULT_SOURCE_BUCKET_ENTRIES: usize = 100_000; const DEFAULT_SOURCE_BUCKET_PER_MINUTE: u32 = 600; const DEFAULT_SOURCE_BUCKET_BURST: u32 = 200; const DEFAULT_FCM_TOKEN_URI: &str = "https://oauth2.googleapis.com/token"; const DEFAULT_FCM_BASE_URL: &str = "https://fcm.googleapis.com"; const DEFAULT_CLIENT_IP_HEADER_NAME: &str = "x-forwarded-for"; const APNS_PRODUCTION_BASE_URL: &str = "https://api.push.apple.com"; const APNS_DEVELOPMENT_BASE_URL: &str = "https://api.sandbox.push.apple.com"; const DEFAULT_MANAGED_RELAY_HOST: &str = "push.fluxer.com"; const VOIP_TOPIC_SUFFIX: &str = ".voip"; #[derive(Clone, Copy, Debug, Default, Eq, PartialEq, clap::ValueEnum)] pub enum Mode { #[default] Delivery, Relay, } impl Mode { pub fn default_port(self) -> u16 { match self { Self::Delivery => DEFAULT_DELIVERY_PORT, Self::Relay => DEFAULT_RELAY_PORT, } } } #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum ProviderEnvironment { Production, Development, } impl ProviderEnvironment { pub fn label(self) -> &'static str { match self { Self::Production => "production", Self::Development => "development", } } pub fn from_label(raw: &str) -> Option { match raw { "production" => Some(Self::Production), "development" => Some(Self::Development), _ => None, } } } #[derive(Clone, Debug)] pub struct ProviderApp { pub app_id: String, pub topic: Option, pub voip_topic: Option, pub environment: Option, pub project_id: Option, } #[derive(Debug)] pub struct NatsConfig { pub url: String, pub auth_token: Option, } #[derive(Debug)] pub struct RpcConfig { pub url: String, pub auth_token: SecretString, } #[derive(Debug)] pub struct VapidConfig { pub email: String, pub public_key: String, pub private_key: SecretString, } #[derive(Debug)] pub struct ApnsConfig { pub team_id: String, pub key_id: String, pub private_key: SecretString, pub default_environment: ProviderEnvironment, pub apps: Vec, pub base_url_override: Option, } impl ApnsConfig { pub fn topic_for(&self, app_id: &str, environment: ProviderEnvironment) -> Option<&str> { self.topic_by(app_id, environment, |app| app.topic.as_deref()) } pub fn voip_topic_for(&self, app_id: &str, environment: ProviderEnvironment) -> Option<&str> { self.topic_by(app_id, environment, |app| app.voip_topic.as_deref()) } fn topic_by( &self, app_id: &str, environment: ProviderEnvironment, topic: fn(&ProviderApp) -> Option<&str>, ) -> Option<&str> { let listed = |app: &&ProviderApp| app.app_id == app_id && topic(app).is_some(); self.apps .iter() .find(|app| listed(app) && app.environment == Some(environment)) .or_else(|| self.apps.iter().find(listed)) .and_then(topic) } pub fn base_url(&self, environment: ProviderEnvironment) -> &str { if let Some(base_url) = self.base_url_override.as_deref() { return base_url; } match environment { ProviderEnvironment::Production => APNS_PRODUCTION_BASE_URL, ProviderEnvironment::Development => APNS_DEVELOPMENT_BASE_URL, } } } #[derive(Debug)] pub struct FcmConfig { pub project_id: String, pub client_email: String, pub private_key: SecretString, pub token_uri: String, pub apps: Vec, pub base_url: String, } impl FcmConfig { pub fn project_id_for(&self, app_id: &str) -> &str { self.apps .iter() .find(|app| app.app_id == app_id && app.project_id.is_some()) .and_then(|app| app.project_id.as_deref()) .unwrap_or(self.project_id.as_str()) } pub fn listed_project_id(&self, app_id: &str) -> Option<&str> { self.apps .iter() .find(|app| app.app_id == app_id) .map(|app| { app.project_id .as_deref() .unwrap_or(self.project_id.as_str()) }) } } #[derive(Debug)] pub struct DeliveryConfig { pub bind_addr: SocketAddr, pub nats: NatsConfig, pub rpc: RpcConfig, pub queue_capacity: usize, pub send_concurrency: usize, pub vapid: VapidConfig, pub apns: Option, pub fcm: Option, pub own_relay_hosts: Vec, pub private_hosts: Vec, pub managed_relay_hosts: Vec, pub relay_consent_accepted: bool, } #[derive(Clone, Copy, Debug)] pub struct BucketConfig { pub entries: usize, pub per_minute: u32, pub burst: u32, } #[derive(Debug)] pub struct RelayConfig { pub bind_addr: SocketAddr, pub max_concurrent: usize, pub max_body_bytes: usize, pub trust_client_ip_header: bool, pub client_ip_header_name: String, pub trusted_proxy_hops: usize, pub device_token_bucket: BucketConfig, pub source_bucket: Option, pub apns: Option, pub fcm: Option, } #[derive(Debug)] pub enum Config { Delivery(Box), Relay(Box), } impl Config { pub fn load_from_iter(mode: Mode, vars: I) -> anyhow::Result where I: IntoIterator, K: Into, V: Into, { Ok(match mode { Mode::Delivery => Self::Delivery(Box::new(DeliveryConfig::load_from_iter(vars)?)), Mode::Relay => Self::Relay(Box::new(RelayConfig::load_from_iter(vars)?)), }) } pub fn bind_addr_mut(&mut self) -> &mut SocketAddr { match self { Self::Delivery(cfg) => &mut cfg.bind_addr, Self::Relay(cfg) => &mut cfg.bind_addr, } } } impl DeliveryConfig { pub fn load_from_iter(vars: I) -> anyhow::Result where I: IntoIterator, K: Into, V: Into, { let env = Env::from_iter(vars); Ok(Self { bind_addr: bind_addr(&env, Mode::Delivery)?, nats: nats_config(&env), rpc: rpc_config(&env)?, queue_capacity: parse_number( "FLUXER_PUSH_SERVICE_QUEUE_CAPACITY", env.get("FLUXER_PUSH_SERVICE_QUEUE_CAPACITY"), DEFAULT_QUEUE_CAPACITY, 1, 1_000_000, )?, send_concurrency: parse_number( "FLUXER_PUSH_SERVICE_SEND_CONCURRENCY", env.get("FLUXER_PUSH_SERVICE_SEND_CONCURRENCY"), DEFAULT_SEND_CONCURRENCY, 1, 65_536, )?, vapid: vapid_config(&env)?, apns: apns_config(&env)?, fcm: fcm_config(&env)?, own_relay_hosts: host_list(&env, "FLUXER_PUSH_SERVICE_OWN_RELAY_HOSTS"), private_hosts: host_list(&env, "FLUXER_PUSH_SERVICE_PRIVATE_HOSTS"), managed_relay_hosts: managed_relay_hosts(&env), relay_consent_accepted: parse_bool( "FLUXER_PUSH_SERVICE_RELAY_CONSENT_ACCEPTED", env.get("FLUXER_PUSH_SERVICE_RELAY_CONSENT_ACCEPTED"), )? .unwrap_or(false), }) } } fn host_list(env: &Env, key: &str) -> Vec { env.get(key) .unwrap_or_default() .split(',') .map(|host| host.trim().to_ascii_lowercase()) .filter(|host| !host.is_empty()) .collect() } fn managed_relay_hosts(env: &Env) -> Vec { let Some(raw) = env.get("FLUXER_PUSH_SERVICE_MANAGED_RELAY_HOSTS") else { return vec![DEFAULT_MANAGED_RELAY_HOST.to_owned()]; }; raw.split(',') .map(|host| host.trim().to_ascii_lowercase()) .filter(|host| !host.is_empty()) .collect() } impl RelayConfig { pub fn load_from_iter(vars: I) -> anyhow::Result where I: IntoIterator, K: Into, V: Into, { let env = Env::from_iter(vars); let cfg = Self { bind_addr: bind_addr(&env, Mode::Relay)?, max_concurrent: parse_number( "FLUXER_PUSH_RELAY_MAX_CONCURRENT", env.get("FLUXER_PUSH_RELAY_MAX_CONCURRENT"), DEFAULT_RELAY_MAX_CONCURRENT, 1, 1_000_000, )?, max_body_bytes: parse_number( "FLUXER_PUSH_RELAY_MAX_BODY_BYTES", env.get("FLUXER_PUSH_RELAY_MAX_BODY_BYTES"), DEFAULT_RELAY_MAX_BODY_BYTES, 1, 1_048_576, )?, trust_client_ip_header: parse_bool( "FLUXER_TRUST_CLIENT_IP_HEADER", env.get("FLUXER_TRUST_CLIENT_IP_HEADER"), )? .unwrap_or(false), client_ip_header_name: env .get("FLUXER_CLIENT_IP_HEADER_NAME") .unwrap_or(DEFAULT_CLIENT_IP_HEADER_NAME) .to_ascii_lowercase(), trusted_proxy_hops: parse_number( "FLUXER_PUSH_RELAY_TRUSTED_PROXY_HOPS", env.get("FLUXER_PUSH_RELAY_TRUSTED_PROXY_HOPS"), DEFAULT_TRUSTED_PROXY_HOPS, 0, 8, )?, device_token_bucket: bucket_config( &env, "FLUXER_PUSH_RELAY_TOKEN_BUCKET", BucketConfig { entries: DEFAULT_DEVICE_TOKEN_BUCKET_ENTRIES, per_minute: DEFAULT_DEVICE_TOKEN_BUCKET_PER_MINUTE, burst: DEFAULT_DEVICE_TOKEN_BUCKET_BURST, }, )?, source_bucket: source_bucket_config(&env)?, apns: apns_config(&env)?, fcm: fcm_config(&env)?, }; for (var_name, apps) in [ ( "FLUXER_PUSH_APNS_APPS", cfg.apns.as_ref().map(|apns| &apns.apps), ), ( "FLUXER_PUSH_FCM_APPS", cfg.fcm.as_ref().map(|fcm| &fcm.apps), ), ] { anyhow::ensure!( apps.is_none_or(|apps| !apps.is_empty()), "{var_name} must list at least one app" ); } Ok(cfg) } } fn bind_addr(env: &Env, mode: Mode) -> anyhow::Result { let host = env.get("FLUXER_PUSH_SERVICE_HOST").unwrap_or(DEFAULT_HOST); let ip = host .parse::() .with_context(|| format!("FLUXER_PUSH_SERVICE_HOST is not an IP address: {host}"))?; let port = parse_number( "FLUXER_PUSH_SERVICE_PORT", env.get("FLUXER_PUSH_SERVICE_PORT"), mode.default_port(), 1, u16::MAX, )?; Ok(SocketAddr::new(ip, port)) } fn nats_config(env: &Env) -> NatsConfig { NatsConfig { url: env .get("FLUXER_SVC_NATS_URL") .unwrap_or(DEFAULT_NATS_URL) .to_owned(), auth_token: env .get("FLUXER_NATS_AUTH_TOKEN") .map(|token| SecretString::new(token.to_owned())), } } fn rpc_config(env: &Env) -> anyhow::Result { let api_endpoint = env .require("FLUXER_INTERNAL_API_ENDPOINT")? .trim_end_matches('/') .to_owned(); anyhow::ensure!( !api_endpoint.is_empty(), "FLUXER_INTERNAL_API_ENDPOINT must not be only slashes" ); Ok(RpcConfig { url: format!("{api_endpoint}{RPC_PATH}"), auth_token: SecretString::new(env.require("FLUXER_GATEWAY_RPC_AUTH_TOKEN")?.to_owned()), }) } fn vapid_config(env: &Env) -> anyhow::Result { Ok(VapidConfig { email: env.require("FLUXER_VAPID_EMAIL")?.to_owned(), public_key: env.require("FLUXER_VAPID_PUBLIC_KEY")?.to_owned(), private_key: SecretString::new(env.require("FLUXER_VAPID_PRIVATE_KEY")?.to_owned()), }) } fn bucket_config(env: &Env, prefix: &str, defaults: BucketConfig) -> anyhow::Result { let entries_var = format!("{prefix}_ENTRIES"); let per_minute_var = format!("{prefix}_PER_MINUTE"); let burst_var = format!("{prefix}_BURST"); Ok(BucketConfig { entries: parse_number( &entries_var, env.get(&entries_var), defaults.entries, 1, 100_000_000, )?, per_minute: parse_number( &per_minute_var, env.get(&per_minute_var), defaults.per_minute, 1, 1_000_000, )?, burst: parse_number( &burst_var, env.get(&burst_var), defaults.burst, 1, 1_000_000, )?, }) } fn source_bucket_config(env: &Env) -> anyhow::Result> { if !parse_bool( "FLUXER_PUSH_RELAY_SOURCE_BUCKET_ENABLED", env.get("FLUXER_PUSH_RELAY_SOURCE_BUCKET_ENABLED"), )? .unwrap_or(false) { return Ok(None); } bucket_config( env, "FLUXER_PUSH_RELAY_SOURCE_BUCKET", BucketConfig { entries: DEFAULT_SOURCE_BUCKET_ENTRIES, per_minute: DEFAULT_SOURCE_BUCKET_PER_MINUTE, burst: DEFAULT_SOURCE_BUCKET_BURST, }, ) .map(Some) } fn apns_config(env: &Env) -> anyhow::Result> { if !parse_bool( "FLUXER_PUSH_APNS_ENABLED", env.get("FLUXER_PUSH_APNS_ENABLED"), )? .unwrap_or(false) { return Ok(None); } let default_environment = match env.get("FLUXER_PUSH_APNS_DEFAULT_ENVIRONMENT") { Some(raw) => parse_environment("FLUXER_PUSH_APNS_DEFAULT_ENVIRONMENT", raw)?, None => ProviderEnvironment::Production, }; Ok(Some(ApnsConfig { team_id: required_field( env.get("FLUXER_PUSH_APNS_TEAM_ID"), "FLUXER_PUSH_APNS_TEAM_ID", "APNs push", )? .to_owned(), key_id: required_field( env.get("FLUXER_PUSH_APNS_KEY_ID"), "FLUXER_PUSH_APNS_KEY_ID", "APNs push", )? .to_owned(), private_key: SecretString::new(read_key( env, "FLUXER_PUSH_APNS_PRIVATE_KEY", "FLUXER_PUSH_APNS_PRIVATE_KEY_PATH", "APNs push", )?), default_environment, apps: parse_apps("FLUXER_PUSH_APNS_APPS", env.get("FLUXER_PUSH_APNS_APPS"))?, base_url_override: env .get("FLUXER_PUSH_SERVICE_APNS_BASE_URL") .map(ToOwned::to_owned), })) } fn fcm_config(env: &Env) -> anyhow::Result> { if !parse_bool( "FLUXER_PUSH_FCM_ENABLED", env.get("FLUXER_PUSH_FCM_ENABLED"), )? .unwrap_or(false) { return Ok(None); } let service_account = match env.get("FLUXER_PUSH_FCM_SERVICE_ACCOUNT_JSON_PATH") { Some(path) => Some(read_service_account(path)?), None => None, }; let project_id = env .get("FLUXER_PUSH_FCM_PROJECT_ID") .map(ToOwned::to_owned) .or_else(|| service_account.as_ref().and_then(|a| a.project_id.clone())); let client_email = env .get("FLUXER_PUSH_FCM_CLIENT_EMAIL") .map(ToOwned::to_owned) .or_else(|| { service_account .as_ref() .and_then(|a| a.client_email.clone()) }); let private_key = match env.get("FLUXER_PUSH_FCM_PRIVATE_KEY") { Some(key) => key.to_owned(), None => match env.get("FLUXER_PUSH_FCM_PRIVATE_KEY_PATH") { Some(path) => read_file("FLUXER_PUSH_FCM_PRIVATE_KEY_PATH", path)?, None => service_account .as_ref() .and_then(|a| a.private_key.clone()) .ok_or_else(|| { anyhow::anyhow!( "FLUXER_PUSH_FCM_PRIVATE_KEY or FLUXER_PUSH_FCM_PRIVATE_KEY_PATH is required when FCM push is enabled" ) })?, }, }; Ok(Some(FcmConfig { project_id: required_field( project_id.as_deref(), "FLUXER_PUSH_FCM_PROJECT_ID", "FCM push", )? .to_owned(), client_email: required_field( client_email.as_deref(), "FLUXER_PUSH_FCM_CLIENT_EMAIL", "FCM push", )? .to_owned(), private_key: SecretString::new(private_key), token_uri: env .get("FLUXER_PUSH_FCM_TOKEN_URI") .unwrap_or(DEFAULT_FCM_TOKEN_URI) .to_owned(), apps: parse_apps("FLUXER_PUSH_FCM_APPS", env.get("FLUXER_PUSH_FCM_APPS"))?, base_url: env .get("FLUXER_PUSH_SERVICE_FCM_BASE_URL") .unwrap_or(DEFAULT_FCM_BASE_URL) .to_owned(), })) } #[derive(Default, Deserialize)] struct ServiceAccount { project_id: Option, client_email: Option, private_key: Option, } fn read_service_account(path: &str) -> anyhow::Result { let raw = read_file("FLUXER_PUSH_FCM_SERVICE_ACCOUNT_JSON_PATH", path)?; serde_json::from_str(&raw).with_context(|| { format!("FLUXER_PUSH_FCM_SERVICE_ACCOUNT_JSON_PATH is not a service account JSON: {path}") }) } fn read_key(env: &Env, key_var: &str, path_var: &str, provider: &str) -> anyhow::Result { if let Some(key) = env.get(key_var) { return Ok(key.to_owned()); } match env.get(path_var) { Some(path) => read_file(path_var, path), None => Err(anyhow::anyhow!( "{key_var} or {path_var} is required when {provider} is enabled" )), } } fn read_file(var_name: &str, path: &str) -> anyhow::Result { std::fs::read_to_string(path).with_context(|| format!("{var_name} could not be read: {path}")) } fn required_field<'a>( value: Option<&'a str>, var_name: &str, provider: &str, ) -> anyhow::Result<&'a str> { value.ok_or_else(|| anyhow::anyhow!("{var_name} is required when {provider} is configured")) } #[derive(Deserialize)] struct RawProviderApp { app_id: Option, topic: Option, environment: Option, project_id: Option, } fn parse_apps(var_name: &str, raw: Option<&str>) -> anyhow::Result> { let Some(raw) = raw else { return Ok(Vec::new()); }; let entries: Vec = serde_json::from_str(raw) .with_context(|| format!("{var_name} must be a JSON array of push app objects"))?; entries .into_iter() .map(|entry| { let app_id = entry .app_id .filter(|app_id| !app_id.trim().is_empty()) .ok_or_else(|| anyhow::anyhow!("{var_name} contains an entry with no app_id"))?; let environment = match entry.environment.as_deref() { Some(value) => Some(parse_environment(var_name, value)?), None => None, }; let topic = entry .topic .map(|topic| topic.trim().to_owned()) .filter(|topic| !topic.is_empty()); if let Some(topic) = topic.as_deref() { anyhow::ensure!( !topic.ends_with(VOIP_TOPIC_SUFFIX), "{var_name} lists a topic that already ends in {VOIP_TOPIC_SUFFIX}: {topic}" ); } Ok(ProviderApp { app_id, voip_topic: topic .as_deref() .map(|topic| format!("{topic}{VOIP_TOPIC_SUFFIX}")), topic, environment, project_id: entry .project_id .filter(|project_id| !project_id.trim().is_empty()), }) }) .collect() } fn parse_environment(var_name: &str, raw: &str) -> anyhow::Result { ProviderEnvironment::from_label(&raw.trim().to_ascii_lowercase()) .ok_or_else(|| anyhow::anyhow!("{var_name} must be one of: production, development")) } struct Env(Vec<(String, String)>); impl Env { fn from_iter(vars: I) -> Self where I: IntoIterator, K: Into, V: Into, { Self( vars.into_iter() .map(|(key, value)| (key.into(), value.into())) .collect(), ) } fn get(&self, key: &str) -> Option<&str> { self.0 .iter() .find_map(|(name, value)| (name == key).then_some(value.trim())) .filter(|value| !value.is_empty()) } fn require(&self, key: &str) -> anyhow::Result<&str> { self.get(key) .ok_or_else(|| anyhow::anyhow!("{key} is required")) } }