autoconnect_settings/
lib.rs

1mod app_state;
2
3extern crate slog;
4#[macro_use]
5extern crate slog_scope;
6extern crate serde_derive;
7
8// Specify "unused_imports" to satisfy clippy.
9#[allow(unused_imports)]
10use std::env;
11use std::{io, net::ToSocketAddrs, time::Duration};
12
13use config::{Config, ConfigError, Environment, File};
14use fernet::Fernet;
15use lazy_static::lazy_static;
16use serde::{Deserialize, Deserializer};
17
18use autopush_common::util::deserialize_u32_to_duration;
19// Specify "unused_imports" to satisfy clippy.
20#[allow(unused_imports)]
21use serde_json::json;
22
23pub use app_state::AppState;
24
25pub const ENV_PREFIX: &str = "autoconnect";
26
27lazy_static! {
28    static ref HOSTNAME: String = mozsvc_common::get_hostname()
29        .expect("Couldn't get_hostname")
30        .into_string()
31        .expect("Couldn't convert get_hostname");
32    static ref RESOLVED_HOSTNAME: String = resolve_ip(&HOSTNAME)
33        .unwrap_or_else(|_| panic!("Failed to resolve hostname: {}", *HOSTNAME));
34}
35
36/// Resolve a hostname to its IP if possible
37fn resolve_ip(hostname: &str) -> io::Result<String> {
38    Ok((hostname, 0)
39        .to_socket_addrs()?
40        .next()
41        .map_or_else(|| hostname.to_owned(), |addr| addr.ip().to_string()))
42}
43
44/// Indicate whether the port should be included for the given scheme
45fn include_port(scheme: &str, port: u16) -> bool {
46    !((scheme == "http" && port == 80) || (scheme == "https" && port == 443))
47}
48
49/// The Applications settings, read from CLI, Environment or settings file, for the
50/// autoconnect application. These are later converted to
51/// [autoconnect::autoconnect-settings::AppState].
52#[derive(Clone, Debug, Deserialize)]
53#[serde(default)]
54pub struct Settings {
55    /// The application port to listen on
56    pub port: u16,
57    /// The DNS specified name of the application host to used for internal routing
58    pub hostname: Option<String>,
59    /// The override hostname to use for internal routing (NOTE: requires `hostname` to be set)
60    pub resolve_hostname: bool,
61    /// The internal webpush routing port
62    pub router_port: u16,
63    /// The DNS name to use for internal routing
64    pub router_hostname: Option<String>,
65    /// The server based ping interval (also used for Broadcast sends)
66    #[serde(deserialize_with = "deserialize_f64_to_duration")]
67    pub auto_ping_interval: Duration,
68    /// How long to wait for a response Pong before being timed out and connection drop
69    #[serde(deserialize_with = "deserialize_f64_to_duration")]
70    pub auto_ping_timeout: Duration,
71    /// How long to wait for the initial connection handshake.
72    #[serde(deserialize_with = "deserialize_u32_to_duration")]
73    pub open_handshake_timeout: Duration,
74    /// The URL scheme (http/https) for the endpoint URL
75    pub endpoint_scheme: String,
76    /// The host url for the endpoint URL (differs from `hostname` and `resolve_hostname`)
77    pub endpoint_hostname: String,
78    /// The optional port override for the endpoint URL
79    pub endpoint_port: u16,
80    /// The seed key to use for endpoint encryption (deprecated: use crypto_keys instead)
81    #[deprecated(since = "1.84.1", note = "Use `crypto_keys` instead")]
82    #[serde(default)]
83    pub crypto_key: String,
84    /// The cryptographic keys to use for endpoint encryption. Format: [key1,key2,...].
85    /// If not set, a random key is generated. Supports multiple keys for rotation.
86    #[serde(default)]
87    pub crypto_keys: String,
88    /// The host name to send recorded metrics
89    pub statsd_host: Option<String>,
90    /// The port number to send recorded metrics
91    pub statsd_port: u16,
92    /// The root label to apply to metrics.
93    pub statsd_label: String,
94    /// Whether to disable Sentry error reporting
95    pub disable_sentry: bool,
96    /// The DSN to connect to the storage engine (Used to select between storage systems)
97    pub db_dsn: Option<String>,
98    /// JSON set of specific database settings (See data storage engines)
99    pub db_settings: String,
100    /// Server endpoint to pull Broadcast ID change values (Sent in Pings)
101    pub megaphone_api_url: Option<String>,
102    /// Broadcast token for authentication (deprecated, no longer used)
103    pub megaphone_api_token: Option<String>,
104    /// How often to poll the server for new data
105    #[serde(deserialize_with = "deserialize_u32_to_duration")]
106    pub megaphone_poll_interval: Duration,
107    /// Use human readable (simplified, non-JSON)
108    pub human_logs: bool,
109    /// Number of log records to buffer before dropping them. Records are dropped
110    /// when the buffer is full, and each drop emits an ERROR level overflow
111    /// report, so this should comfortably exceed the peak logging rate. 0 uses
112    /// the default.
113    pub log_chan_size: usize,
114    /// Maximum allowed number of backlogged messages. Exceeding this number will
115    /// trigger a user reset because the user may have been offline way too long.
116    pub msg_limit: u32,
117    /// Maximum number of buffered notifications per client before backpressure
118    /// is applied on the notification channel.
119    pub client_channel_capacity: usize,
120    /// Sets the maximum number of concurrent connections per actix-web worker.
121    ///
122    /// All socket listeners will stop accepting connections when this limit is
123    /// reached for each worker.
124    pub actix_max_connections: Option<usize>,
125    /// Sets number of actix-web workers to start (per bind address).
126    ///
127    /// By default, the number of available physical CPUs is used as the worker count.
128    pub actix_workers: Option<usize>,
129    /// Maximum idle connections per host in the HTTP connection pool.
130    pub pool_max_idle_per_host: usize,
131    /// Idle connection timeout in seconds.
132    pub pool_idle_timeout_secs: u64,
133    #[cfg(feature = "reliable_report")]
134    /// The DNS for the reliability data store. This is normally a Redis compatible
135    /// storage system. See [Connection Parameters](https://docs.rs/redis/latest/redis/#connection-parameters)
136    /// for details.
137    pub reliability_dsn: Option<String>,
138    #[cfg(feature = "reliable_report")]
139    /// Max number of retries for retries for Redis transactions
140    pub reliability_retry_count: usize,
141}
142// Did you update the documentation in `docs/src/config_options.md`?
143
144impl Default for Settings {
145    fn default() -> Self {
146        Self {
147            port: 8080,
148            hostname: None,
149            resolve_hostname: false,
150            router_port: 8081,
151            router_hostname: None,
152            auto_ping_interval: Duration::from_secs(300),
153            auto_ping_timeout: Duration::from_secs(4),
154            open_handshake_timeout: Duration::from_secs(5),
155            endpoint_scheme: "http".to_owned(),
156            endpoint_hostname: "localhost".to_owned(),
157            endpoint_port: 8082,
158            crypto_key: String::new(),
159            crypto_keys: format!("[{}]", Fernet::generate_key()),
160            statsd_host: Some("localhost".to_owned()),
161            // Matches the legacy value
162            statsd_label: "autoconnect".to_owned(),
163            statsd_port: 8125,
164            disable_sentry: false,
165            db_dsn: None,
166            db_settings: "".to_owned(),
167            megaphone_api_url: None,
168            megaphone_api_token: None,
169            megaphone_poll_interval: Duration::from_secs(30),
170            human_logs: false,
171            log_chan_size: autopush_common::logging::DEFAULT_LOG_CHAN_SIZE,
172            msg_limit: 150,
173            client_channel_capacity: 128,
174            actix_max_connections: None,
175            actix_workers: None,
176            pool_max_idle_per_host: 10,
177            pool_idle_timeout_secs: 30,
178            #[cfg(feature = "reliable_report")]
179            reliability_dsn: None,
180            #[cfg(feature = "reliable_report")]
181            reliability_retry_count: autopush_common::redis_util::MAX_TRANSACTION_LOOP,
182        }
183    }
184}
185
186impl Settings {
187    /// Load the settings from the config files in order first then the environment.
188    pub fn with_env_and_config_files(filenames: &[String]) -> Result<Self, ConfigError> {
189        let mut s = Config::builder();
190
191        // Merge the configs from the files
192        for filename in filenames {
193            s = s.add_source(File::with_name(filename));
194        }
195
196        // Merge the environment overrides
197        s = s.add_source(Environment::with_prefix(&ENV_PREFIX.to_uppercase()).separator("__"));
198
199        let built = s.build()?;
200        let mut settings = built.try_deserialize::<Settings>()?;
201        settings.normalize_crypto_keys();
202        settings.validate()?;
203        Ok(settings)
204    }
205
206    pub fn router_url(&self) -> String {
207        let router_scheme = "http";
208        let url = format!(
209            "{}://{}",
210            router_scheme,
211            self.router_hostname
212                .as_ref()
213                .map_or_else(|| self.get_hostname(), String::clone),
214        );
215        if include_port(router_scheme, self.router_port) {
216            format!("{}:{}", url, self.router_port)
217        } else {
218            url
219        }
220    }
221
222    pub fn endpoint_url(&self) -> String {
223        let url = format!("{}://{}", self.endpoint_scheme, self.endpoint_hostname,);
224        if include_port(&self.endpoint_scheme, self.endpoint_port) {
225            format!("{}:{}", url, self.endpoint_port)
226        } else {
227            url
228        }
229    }
230
231    fn get_hostname(&self) -> String {
232        if let Some(ref hostname) = self.hostname {
233            if self.resolve_hostname {
234                resolve_ip(hostname)
235                    .unwrap_or_else(|_| panic!("Failed to resolve provided hostname: {hostname}"))
236            } else {
237                hostname.clone()
238            }
239        } else if self.resolve_hostname {
240            RESOLVED_HOSTNAME.clone()
241        } else {
242            HOSTNAME.clone()
243        }
244    }
245
246    pub fn validate(&self) -> Result<(), ConfigError> {
247        let non_zero = |val: Duration, name| {
248            if val.is_zero() {
249                return Err(ConfigError::Message(format!(
250                    "Invalid {ENV_PREFIX}_{name}: cannot be 0"
251                )));
252            }
253            Ok(())
254        };
255        non_zero(self.megaphone_poll_interval, "MEGAPHONE_POLL_INTERVAL")?;
256        non_zero(self.auto_ping_interval, "AUTO_PING_INTERVAL")?;
257        non_zero(self.auto_ping_timeout, "AUTO_PING_TIMEOUT")?;
258        Ok(())
259    }
260
261    /// Normalize crypto key settings: prefer crypto_keys, fall back to crypto_key
262    pub fn normalize_crypto_keys(&mut self) {
263        if self.crypto_keys.is_empty() && !self.crypto_key.is_empty() {
264            warn!("AUTOCONNECT__CRYPTO_KEY is deprecated; use AUTOCONNECT__CRYPTO_KEYS instead");
265            self.crypto_keys = self.crypto_key.clone();
266        } else if self.crypto_keys.is_empty() {
267            self.crypto_keys = format!("[{}]", Fernet::generate_key());
268        }
269    }
270
271    pub fn test_settings() -> Self {
272        // Provide test settings based on enabled features.
273        // semi-hack to satisfy clippy --all --all-features
274        if cfg!(feature = "bigtable") {
275            let host = env::var("BIGTABLE_EMULATOR_HOST").unwrap_or("localhost:8086".to_owned());
276            let db_dsn = Some(format!("grpc://{}", host));
277            // BigTable DB_SETTINGS.
278            let db_settings = json!({
279                "table_name":"projects/test/instances/test/tables/autopush",
280                "message_family":"message",
281                "router_family":"router",
282                "message_topic_family":"message_topic",
283            })
284            .to_string();
285            return Self {
286                db_dsn,
287                db_settings,
288                ..Default::default()
289            };
290        }
291        if cfg!(feature = "redis") {
292            let host = env::var("REDIS_HOST").unwrap_or("localhost:6379".to_owned());
293            let db_dsn = Some(format!("redis://{}", host));
294            let db_settings = "".to_string();
295            return Self {
296                db_dsn,
297                db_settings,
298                ..Default::default()
299            };
300        }
301        if cfg!(feature = "postgres") {
302            let host = env::var("POSTGRES_HOST").unwrap_or("localhost:5432".to_owned());
303            let db_dsn = Some(format!("postgres://{}", host));
304            let db_settings = "".to_string();
305            return Self {
306                db_dsn,
307                db_settings,
308                ..Default::default()
309            };
310        }
311        Self::default()
312    }
313}
314
315fn deserialize_f64_to_duration<'de, D>(deserializer: D) -> Result<Duration, D::Error>
316where
317    D: Deserializer<'de>,
318{
319    let seconds: f64 = Deserialize::deserialize(deserializer)?;
320    Ok(Duration::new(
321        seconds as u64,
322        (seconds.fract() * 1_000_000_000.0) as u32,
323    ))
324}
325
326#[cfg(test)]
327mod tests {
328    use super::*;
329    #[cfg(feature = "unsafe")]
330    use slog_scope::trace;
331
332    #[test]
333    fn test_normalize_crypto_keys_prefers_new_field() {
334        let test_key = "[mqCGb8D-N7mqx6iWJov9wm70Us6kA9veeXdb8QUuzLQ=]";
335        let mut settings = Settings {
336            crypto_key: "[old-key-mqCGb8D-N7mqx6iWJov9wm70Us6kA9veeXdb8QUuzLQ=]".to_string(),
337            crypto_keys: test_key.to_string(),
338            ..Default::default()
339        };
340        settings.normalize_crypto_keys();
341        // crypto_keys should be unchanged when already set
342        assert_eq!(settings.crypto_keys, test_key);
343    }
344
345    #[test]
346    fn test_normalize_crypto_keys_falls_back_to_old_field() {
347        let test_key = "[mqCGb8D-N7mqx6iWJov9wm70Us6kA9veeXdb8QUuzLQ=]";
348        let mut settings = Settings {
349            crypto_key: test_key.to_string(),
350            crypto_keys: String::new(),
351            ..Default::default()
352        };
353        settings.normalize_crypto_keys();
354        // Should copy crypto_key to crypto_keys when crypto_keys is empty
355        assert_eq!(settings.crypto_keys, test_key);
356    }
357
358    #[test]
359    fn test_router_url() {
360        let mut settings = Settings {
361            router_hostname: Some("testname".to_string()),
362            router_port: 80,
363            ..Default::default()
364        };
365        let url = settings.router_url();
366        assert_eq!("http://testname", url);
367
368        settings.router_port = 8080;
369        let url = settings.router_url();
370        assert_eq!("http://testname:8080", url);
371    }
372
373    #[test]
374    fn test_endpoint_url() {
375        let mut settings = Settings {
376            endpoint_hostname: "testname".to_string(),
377            endpoint_port: 80,
378            endpoint_scheme: "http".to_string(),
379            ..Default::default()
380        };
381        let url = settings.endpoint_url();
382        assert_eq!("http://testname", url);
383
384        settings.endpoint_port = 8080;
385        let url = settings.endpoint_url();
386        assert_eq!("http://testname:8080", url);
387
388        settings.endpoint_port = 443;
389        settings.endpoint_scheme = "https".to_string();
390        let url = settings.endpoint_url();
391        assert_eq!("https://testname", url);
392
393        settings.endpoint_port = 8080;
394        let url = settings.endpoint_url();
395        assert_eq!("https://testname:8080", url);
396    }
397
398    // The following test is commented out due to the recent change in rust that makes `env::set_var` unsafe
399    #[cfg(all(test, feature = "unsafe"))]
400    #[test]
401    fn test_default_settings() {
402        // Test that the Config works the way we expect it to.
403        use std::env;
404        let port = format!("{ENV_PREFIX}__PORT").to_uppercase();
405        let msg_limit = format!("{ENV_PREFIX}__MSG_LIMIT").to_uppercase();
406        let fernet = format!("{ENV_PREFIX}__CRYPTO_KEYS").to_uppercase();
407
408        let v1 = env::var(&port);
409        let v2 = env::var(&msg_limit);
410        let v3 = env::var(&fernet);
411        unsafe {
412            env::set_var(&port, "9123");
413            env::set_var(&msg_limit, "123");
414            env::set_var(&fernet, "[mqCGb8D-N7mqx6iWJov9wm70Us6kA9veeXdb8QUuzLQ=]");
415        }
416        let settings = Settings::with_env_and_config_files(&Vec::new()).unwrap();
417        assert_eq!(settings.endpoint_hostname, "localhost".to_owned());
418        assert_eq!(&settings.port, &9123);
419        assert_eq!(&settings.msg_limit, &123);
420        assert_eq!(
421            &settings.crypto_keys,
422            "[mqCGb8D-N7mqx6iWJov9wm70Us6kA9veeXdb8QUuzLQ=]"
423        );
424
425        assert_eq!(settings.open_handshake_timeout, Duration::from_secs(5));
426
427        // reset (just in case)
428        if let Ok(p) = v1 {
429            trace!("Resetting {}", &port);
430            // TODO: Audit that the environment access only happens in single-threaded code.
431            unsafe { env::set_var(&port, p) };
432        } else {
433            // TODO: Audit that the environment access only happens in single-threaded code.
434            unsafe { env::remove_var(&port) };
435        }
436        if let Ok(p) = v2 {
437            trace!("Resetting {}", msg_limit);
438            // TODO: Audit that the environment access only happens in single-threaded code.
439            unsafe { env::set_var(&msg_limit, p) };
440        } else {
441            // TODO: Audit that the environment access only happens in single-threaded code.
442            unsafe { env::remove_var(&msg_limit) };
443        }
444        // reset fernet var
445        if let Ok(p) = v3 {
446            trace!("Resetting {}", &fernet);
447            unsafe { env::set_var(&fernet, p) };
448        } else {
449            unsafe { env::remove_var(&fernet) };
450        }
451        // TODO: Audit that the environment access only happens in single-threaded code.
452        unsafe { env::remove_var(&fernet) };
453    }
454}
autoconnect_settings/lib.rs

autoconnect_settings/
lib.rs
