feat(push): allow listed hosts to resolve to private addresses (#3228)

deploy/self-hosting/.env.example

@@ -219,6 +219,9 @@ FLUXER_VAPID_PRIVATE_KEY=CHANGE_ME
 #FLUXER_PUSH_SERVICE_FCM_BASE_URL=https://fcm.googleapis.com
 #FLUXER_PUSH_SERVICE_MANAGED_RELAY_HOSTS=push.fluxer.com
 #FLUXER_PUSH_SERVICE_OWN_RELAY_HOSTS=
+# Push hosts on your own network, such as a ntfy server, that may resolve to
+# private addresses. Comma separated.
+#FLUXER_PUSH_SERVICE_PRIVATE_HOSTS=ntfy.example.com
 #FLUXER_PUSH_SERVICE_RELAY_CONSENT_ACCEPTED=false
 
 # Direct mobile push through your own APNs and FCM credentials, off by default.

deploy/self-hosting/docker-compose.yml

@@ -615,6 +615,7 @@ services:
       FLUXER_PUSH_SERVICE_FCM_BASE_URL: ${FLUXER_PUSH_SERVICE_FCM_BASE_URL:-}
       FLUXER_PUSH_SERVICE_MANAGED_RELAY_HOSTS: ${FLUXER_PUSH_SERVICE_MANAGED_RELAY_HOSTS:-}
       FLUXER_PUSH_SERVICE_OWN_RELAY_HOSTS: ${FLUXER_PUSH_SERVICE_OWN_RELAY_HOSTS:-}
+      FLUXER_PUSH_SERVICE_PRIVATE_HOSTS: ${FLUXER_PUSH_SERVICE_PRIVATE_HOSTS:-}
       FLUXER_PUSH_SERVICE_RELAY_CONSENT_ACCEPTED: ${FLUXER_PUSH_SERVICE_RELAY_CONSENT_ACCEPTED:-}
       FLUXER_PUSH_APNS_DEFAULT_ENVIRONMENT: ${FLUXER_PUSH_APNS_DEFAULT_ENVIRONMENT:-}
       FLUXER_PUSH_FCM_ENABLED: ${FLUXER_PUSH_FCM_ENABLED:-}

fluxer_docs/src/content/docs/operator/configuration.mdx

@@ -928,6 +928,10 @@ Default `push.fluxer.com`. Hostnames, comma separated, of the Fluxer-run push re
 
 Default empty. Hostnames, comma separated, of a push relay this instance runs itself. `push` delivers a browser push endpoint on one of them straight through its own APNs or FCM credentials, with no HTTP hop. Compose forwards it from `.env`.
 
+#### `FLUXER_PUSH_SERVICE_PRIVATE_HOSTS`
+
+Default empty. Hostnames, comma separated, that a push endpoint may use even when they resolve to private, loopback, or other non-public addresses, such as a ntfy server on your own network. `push` refuses every other such endpoint and logs `blocked_address`. Compose forwards it from `.env`.
+
 `push` also runs as a relay under `--mode relay`, which the stack does not use. Only that mode reads `FLUXER_PUSH_RELAY_MAX_CONCURRENT`, default `1024`, `FLUXER_PUSH_RELAY_MAX_BODY_BYTES`, default `2816`, `FLUXER_PUSH_RELAY_TRUSTED_PROXY_HOPS`, default `1`, and `FLUXER_PUSH_RELAY_SOURCE_BUCKET_ENABLED`, default `false`. Compose does not forward them, and [Names the stack has no use for](#names-the-stack-has-no-use-for) lists them.
 
 ## Payments

fluxer_push/src/config.rs

@@ -185,6 +185,7 @@ pub struct DeliveryConfig {
     pub apns: Option<ApnsConfig>,
     pub fcm: Option<FcmConfig>,
     pub own_relay_hosts: Vec<String>,
+    pub private_hosts: Vec<String>,
     pub managed_relay_hosts: Vec<String>,
     pub relay_consent_accepted: bool,
 }
@@ -266,7 +267,8 @@ impl DeliveryConfig {
             vapid: vapid_config(&env)?,
             apns: apns_config(&env)?,
             fcm: fcm_config(&env)?,
-            own_relay_hosts: own_relay_hosts(&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",
@@ -277,8 +279,8 @@ impl DeliveryConfig {
     }
 }
 
-fn own_relay_hosts(env: &Env) -> Vec<String> {
-    env.get("FLUXER_PUSH_SERVICE_OWN_RELAY_HOSTS")
+fn host_list(env: &Env, key: &str) -> Vec<String> {
+    env.get(key)
         .unwrap_or_default()
         .split(',')
         .map(|host| host.trim().to_ascii_lowercase())

fluxer_push/src/providers/web_push.rs

@@ -87,7 +87,7 @@ struct Record {
 }
 
 fn seal(state: &AppState, sub: &Subscription, envelope: &Value) -> Result<Record, SendOutcome> {
-    if !endpoint_is_allowed(&sub.endpoint) {
+    if !endpoint_is_allowed(&sub.endpoint, &state.cfg.private_hosts) {
         return Err(SendOutcome::permanent("endpoint_rejected"));
     }
     let (Some(p256dh), Some(auth)) = (sub.p256dh_key.as_deref(), sub.auth_key.as_deref()) else {
@@ -223,7 +223,7 @@ fn classify(status: u16) -> SendOutcome {
     }
 }
 
-fn endpoint_is_allowed(endpoint: &str) -> bool {
+fn endpoint_is_allowed(endpoint: &str, private_hosts: &[String]) -> bool {
     let Ok(url) = Url::parse(endpoint) else {
         return false;
     };
@@ -237,7 +237,9 @@ fn endpoint_is_allowed(endpoint: &str) -> bool {
         return false;
     }
     match url.host() {
-        Some(Host::Domain(host)) => is_public_hostname(host),
+        Some(Host::Domain(host)) => {
+            resolver::is_private_host(host, private_hosts) || is_public_hostname(host)
+        }
         Some(Host::Ipv4(ip)) => !resolver::is_blocked(IpAddr::V4(ip)),
         Some(Host::Ipv6(ip)) => !resolver::is_blocked(IpAddr::V6(ip)),
         None => false,

fluxer_push/src/resolver.rs

@@ -2,10 +2,13 @@
 
 use reqwest::dns::{Addrs, Name, Resolve, Resolving};
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use std::sync::Arc;
 use tokio::net::lookup_host;
 
 type ResolveError = Box<dyn std::error::Error + Send + Sync>;
 
+pub const BLOCKED_ADDRESS_ERROR: &str = "host resolved into blocked address space";
+
 const BLOCKED_V4: &[(Ipv4Addr, u32)] = &[
     (Ipv4Addr::new(0, 0, 0, 0), 8),
     (Ipv4Addr::new(10, 0, 0, 0), 8),
@@ -36,26 +39,42 @@ const BLOCKED_V6: &[(Ipv6Addr, u32)] = &[
 const NAT64_PREFIX: [u8; 4] = [0x00, 0x64, 0xff, 0x9b];
 const SIXTOFOUR_PREFIX: [u8; 2] = [0x20, 0x02];
 
-pub struct PublicOnlyResolver;
+pub struct PublicOnlyResolver {
+    private_hosts: Arc<[String]>,
+}
+
+impl PublicOnlyResolver {
+    pub fn new(private_hosts: &[String]) -> Self {
+        Self {
+            private_hosts: private_hosts.into(),
+        }
+    }
+}
 
 impl Resolve for PublicOnlyResolver {
     fn resolve(&self, name: Name) -> Resolving {
         let host = name.as_str().to_owned();
-        Box::pin(async move { resolve_public(&host).await })
+        let allow_private = is_private_host(&host, &self.private_hosts);
+        Box::pin(async move {
+            let resolved: Vec<SocketAddr> = lookup_host((host.as_str(), 0)).await?.collect();
+            screen(resolved, allow_private)
+        })
     }
 }
 
-async fn resolve_public(host: &str) -> Result<Addrs, ResolveError> {
-    let resolved: Vec<SocketAddr> = lookup_host((host, 0)).await?.collect();
-    screen(resolved)
+pub fn is_private_host(host: &str, private_hosts: &[String]) -> bool {
+    let host = host.strip_suffix('.').unwrap_or(host);
+    private_hosts
+        .iter()
+        .any(|private| private.eq_ignore_ascii_case(host))
 }
 
-fn screen(resolved: Vec<SocketAddr>) -> Result<Addrs, ResolveError> {
+fn screen(resolved: Vec<SocketAddr>, allow_private: bool) -> Result<Addrs, ResolveError> {
     if resolved.is_empty() {
         return Err("host resolved to no addresses".into());
     }
-    if resolved.iter().any(|addr| is_blocked(addr.ip())) {
-        return Err("host resolved into blocked address space".into());
+    if !allow_private && resolved.iter().any(|addr| is_blocked(addr.ip())) {
+        return Err(BLOCKED_ADDRESS_ERROR.into());
     }
     Ok(Box::new(resolved.into_iter()))
 }
@@ -122,3 +141,26 @@ fn embedded_v4(ip: Ipv6Addr) -> Option<Ipv4Addr> {
     }
     None
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn lan_address() -> Vec<SocketAddr> {
+        vec![SocketAddr::from(([192, 168, 1, 20], 0))]
+    }
+
+    #[test]
+    fn a_private_address_is_refused_by_default() {
+        let error = screen(lan_address(), false).err().unwrap();
+        assert_eq!(error.to_string(), BLOCKED_ADDRESS_ERROR);
+    }
+
+    #[test]
+    fn a_listed_private_host_may_resolve_to_a_private_address() {
+        let hosts = vec!["ntfy.example.com".to_owned()];
+        assert!(is_private_host("NTFY.example.com.", &hosts));
+        assert!(!is_private_host("other.example.com", &hosts));
+        assert!(screen(lan_address(), true).is_ok());
+    }
+}

fluxer_push/src/server.rs

@@ -110,7 +110,7 @@ impl AppState {
             rpc: RpcClient::new(&cfg.rpc, http.clone(), Arc::clone(&metrics)),
             relay_consent,
             sidecar: Arc::new(Sidecar::new(Arc::clone(&metrics))),
-            web_push_http: vendor::web_push_http_client()?,
+            web_push_http: vendor::web_push_http_client(&cfg.private_hosts)?,
             apns_http: vendor::apns_http_client()?,
             tokens: TokenCache::new(),
             draining: watch::Sender::new(false),

fluxer_push/src/vendor.rs

@@ -2,7 +2,7 @@
 
 use crate::config::{ApnsConfig, FcmConfig, ProviderEnvironment};
 use crate::metrics::Metrics;
-use crate::resolver::PublicOnlyResolver;
+use crate::resolver::{BLOCKED_ADDRESS_ERROR, PublicOnlyResolver};
 use crate::tokens::{TokenCache, TokenError};
 use reqwest::header::{AUTHORIZATION, CONTENT_TYPE};
 use reqwest::redirect::Policy;
@@ -30,10 +30,10 @@ pub fn http_client() -> reqwest::Result<reqwest::Client> {
         .build()
 }
 
-pub fn web_push_http_client() -> reqwest::Result<reqwest::Client> {
+pub fn web_push_http_client(private_hosts: &[String]) -> reqwest::Result<reqwest::Client> {
     reqwest::Client::builder()
         .redirect(Policy::none())
-        .dns_resolver(PublicOnlyResolver)
+        .dns_resolver(PublicOnlyResolver::new(private_hosts))
         .connect_timeout(CONNECT_TIMEOUT)
         .timeout(HTTP_TIMEOUT)
         .build()
@@ -65,6 +65,7 @@ pub enum VendorOutcome {
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum Unreachable {
+    BlockedAddress,
     Dns,
     Transport,
 }
@@ -72,17 +73,20 @@ pub enum Unreachable {
 impl Unreachable {
     pub fn label(self) -> &'static str {
         match self {
+            Self::BlockedAddress => "blocked_address",
             Self::Dns => "dns",
             Self::Transport => "transport",
         }
     }
 
     pub fn is_permanent(self) -> bool {
-        matches!(self, Self::Dns)
+        matches!(self, Self::BlockedAddress | Self::Dns)
     }
 
     pub fn of(error: &reqwest::Error) -> Self {
-        if names_no_host(error) {
+        if source_mentions(error, BLOCKED_ADDRESS_ERROR) {
+            Self::BlockedAddress
+        } else if source_mentions(error, DNS_ERROR_MARKER) {
             Self::Dns
         } else {
             Self::Transport
@@ -90,10 +94,10 @@ impl Unreachable {
     }
 }
 
-fn names_no_host(error: &reqwest::Error) -> bool {
+fn source_mentions(error: &reqwest::Error, marker: &str) -> bool {
     let mut current = std::error::Error::source(error);
     while let Some(error) = current {
-        if error.to_string().contains(DNS_ERROR_MARKER) {
+        if error.to_string().contains(marker) {
             return true;
         }
         current = error.source();
@@ -320,6 +324,29 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn a_host_in_blocked_address_space_is_named_as_such() {
+        let error = web_push_http_client(&[])
+            .expect("the http client builds")
+            .post("https://localhost/push")
+            .send()
+            .await
+            .expect_err("the request cannot complete");
+        assert_eq!(Unreachable::of(&error), Unreachable::BlockedAddress);
+        assert!(Unreachable::of(&error).is_permanent());
+    }
+
+    #[tokio::test]
+    async fn a_listed_private_host_gets_through_to_connect() {
+        let error = web_push_http_client(&["localhost".to_owned()])
+            .expect("the http client builds")
+            .post("https://localhost:1/push")
+            .send()
+            .await
+            .expect_err("the request cannot complete");
+        assert_eq!(Unreachable::of(&error), Unreachable::Transport);
+    }
+
     #[tokio::test]
     async fn a_refused_connection_stays_retryable() {
         let error = error_for("http://127.0.0.1:1/").await;