feat(push): gate relay delivery on operator consent (#2984)

fluxer_push/src/config.rs

@@ -29,6 +29,7 @@ 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)]
@@ -184,6 +185,8 @@ pub struct DeliveryConfig {
     pub apns: Option<ApnsConfig>,
     pub fcm: Option<FcmConfig>,
     pub own_relay_hosts: Vec<String>,
+    pub managed_relay_hosts: Vec<String>,
+    pub relay_consent_accepted: bool,
 }
 
 #[derive(Clone, Copy, Debug)]
@@ -264,6 +267,12 @@ impl DeliveryConfig {
             apns: apns_config(&env)?,
             fcm: fcm_config(&env)?,
             own_relay_hosts: own_relay_hosts(&env),
+            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),
         })
     }
 }
@@ -277,6 +286,16 @@ fn own_relay_hosts(env: &Env) -> Vec<String> {
         .collect()
 }
 
+fn managed_relay_hosts(env: &Env) -> Vec<String> {
+    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<I, K, V>(vars: I) -> anyhow::Result<Self>
     where

fluxer_push/src/providers/mod.rs

@@ -91,16 +91,20 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen
         return SendOutcome::permanent("unsupported_platform");
     };
     let started_ms = now_ms();
-    let direct = in_process_hop(&sub.endpoint, &state.cfg.own_relay_hosts);
-    let outcome = match (route, direct) {
-        (Route::WebPush, Some(hop)) => {
-            let hopped = hop.as_subscription(sub);
-            state.metrics.record_own_relay_shortcut();
-            apns::send(state, &hopped, envelope).await
+    let outcome = if relay_consent_missing(state, &sub.endpoint) {
+        SendOutcome::permanent("relay_consent_required")
+    } else {
+        let direct = in_process_hop(&sub.endpoint, &state.cfg.own_relay_hosts);
+        match (route, direct) {
+            (Route::WebPush, Some(hop)) => {
+                let hopped = hop.as_subscription(sub);
+                state.metrics.record_own_relay_shortcut();
+                apns::send(state, &hopped, envelope).await
+            }
+            (Route::WebPush, None) => web_push::send(state, sub, envelope).await,
+            (Route::LegacyApns, _) => apns::send(state, sub, envelope).await,
+            (Route::LegacyFcm, _) => fcm::send(state, sub, envelope).await,
         }
-        (Route::WebPush, None) => web_push::send(state, sub, envelope).await,
-        (Route::LegacyApns, _) => apns::send(state, sub, envelope).await,
-        (Route::LegacyFcm, _) => fcm::send(state, sub, envelope).await,
     };
     state.metrics.record_send(
         provider_of(platform),
@@ -113,6 +117,19 @@ pub async fn send(state: &AppState, sub: &Subscription, envelope: &Value) -> Sen
     outcome
 }
 
+fn relay_consent_missing(state: &AppState, endpoint: &str) -> bool {
+    !relay_consent_accepted(state)
+        && own_relay::is_managed(endpoint, &state.cfg.managed_relay_hosts)
+}
+
+fn relay_consent_accepted(state: &AppState) -> bool {
+    state.cfg.relay_consent_accepted
+        || state
+            .rollout
+            .snapshot()
+            .is_some_and(|held| held.relay_consent_accepted)
+}
+
 fn in_process_hop(endpoint: &str, hosts: &[String]) -> Option<own_relay::Hop> {
     own_relay::parse(endpoint, hosts).filter(|hop| matches!(hop.leg, own_relay::Leg::Apns))
 }
@@ -172,3 +189,129 @@ mod hop_tests {
         assert!(in_process_hop(&voip, &ours()).is_none());
     }
 }
+
+#[cfg(test)]
+mod consent_tests {
+    use super::*;
+    use crate::config::DeliveryConfig;
+    use crate::server::AppState;
+
+    const TOKEN: &str = "3dbc5a5ef1a1c1666afc26f466e1b3ebaaf4c66d92dddeb0fd1b69c49641d4cd";
+    const NTFY_ENDPOINT: &str = "https://ntfy.sh/upZzH87cT9jJCc?up=1";
+
+    fn managed_endpoint() -> String {
+        format!("https://push.fluxer.com/relay/v1/apns/stable/production/{TOKEN}")
+    }
+
+    fn state(relay_consent_accepted: bool) -> AppState {
+        let cfg = DeliveryConfig::load_from_iter([
+            ("FLUXER_INTERNAL_API_ENDPOINT", "http://127.0.0.1:8080"),
+            ("FLUXER_GATEWAY_RPC_AUTH_TOKEN", "rpc-token"),
+            ("FLUXER_VAPID_EMAIL", "[email protected]"),
+            ("FLUXER_VAPID_PUBLIC_KEY", "public-key"),
+            ("FLUXER_VAPID_PRIVATE_KEY", "private-key"),
+            (
+                "FLUXER_PUSH_SERVICE_RELAY_CONSENT_ACCEPTED",
+                if relay_consent_accepted {
+                    "true"
+                } else {
+                    "false"
+                },
+            ),
+        ])
+        .expect("the delivery config loads");
+        AppState::try_new(cfg).expect("the delivery state builds")
+    }
+
+    fn subscription(endpoint: &str) -> Subscription {
+        Subscription {
+            subscription_id: "sub-1".to_owned(),
+            endpoint: endpoint.to_owned(),
+            p256dh_key: None,
+            auth_key: None,
+            platform: Some("web_push".to_owned()),
+            app_id: None,
+            provider_environment: None,
+        }
+    }
+
+    async fn outcome_of(relay_consent_accepted: bool, endpoint: &str) -> SendOutcome {
+        let state = state(relay_consent_accepted);
+        send(&state, &subscription(endpoint), &Value::Null).await
+    }
+
+    #[tokio::test]
+    async fn a_managed_relay_send_waits_for_the_operator_to_accept_the_notice() {
+        assert_eq!(
+            outcome_of(false, &managed_endpoint()).await,
+            SendOutcome::permanent("relay_consent_required")
+        );
+    }
+
+    #[tokio::test]
+    async fn a_refused_managed_relay_send_keeps_the_registration() {
+        assert!(!outcome_of(false, &managed_endpoint()).await.deletes_token());
+    }
+
+    #[tokio::test]
+    async fn an_accepted_notice_lets_the_managed_relay_send_through() {
+        assert_eq!(
+            outcome_of(true, &managed_endpoint()).await,
+            SendOutcome::permanent("missing_keys")
+        );
+    }
+
+    #[tokio::test]
+    async fn a_unified_push_endpoint_is_sent_whatever_the_operator_accepted() {
+        assert_eq!(
+            outcome_of(false, NTFY_ENDPOINT).await,
+            SendOutcome::permanent("missing_keys")
+        );
+        assert_eq!(
+            outcome_of(true, NTFY_ENDPOINT).await,
+            SendOutcome::permanent("missing_keys")
+        );
+    }
+
+    #[tokio::test]
+    async fn a_notice_accepted_in_the_instance_config_lets_the_send_through() {
+        let state = state(false);
+        state.rollout.update(&serde_json::json!({
+            "enabled": true,
+            "config_version": 1,
+            "relay_consent_accepted": true,
+        }));
+        assert_eq!(
+            send(&state, &subscription(&managed_endpoint()), &Value::Null).await,
+            SendOutcome::permanent("missing_keys")
+        );
+    }
+
+    #[tokio::test]
+    async fn an_instance_config_that_has_not_accepted_still_refuses_the_send() {
+        let state = state(false);
+        state.rollout.update(&serde_json::json!({
+            "enabled": true,
+            "config_version": 1,
+            "relay_consent_accepted": false,
+        }));
+        assert_eq!(
+            send(&state, &subscription(&managed_endpoint()), &Value::Null).await,
+            SendOutcome::permanent("relay_consent_required")
+        );
+    }
+
+    #[tokio::test]
+    async fn a_refused_managed_relay_send_is_still_counted() {
+        let state = state(false);
+        let outcome = send(&state, &subscription(&managed_endpoint()), &Value::Null).await;
+        assert_eq!(outcome, SendOutcome::permanent("relay_consent_required"));
+        let rendered = state.metrics.render();
+        for series in [
+            "fluxer_push_sends_total{provider=\"web_push\",result=\"permanent\"} 1",
+            "fluxer_push_delivery_routes_total{route=\"web_push\",result=\"permanent\"} 1",
+        ] {
+            assert!(rendered.contains(series), "{series} must be recorded");
+        }
+    }
+}

fluxer_push/src/providers/own_relay.rs

@@ -97,6 +97,10 @@ pub fn parse(endpoint: &str, hosts: &[String]) -> Option<Hop> {
     })
 }
 
+pub fn is_managed(endpoint: &str, hosts: &[String]) -> bool {
+    parse(endpoint, hosts).is_some()
+}
+
 fn decode(segment: &str) -> Option<String> {
     percent_encoding::percent_decode_str(segment)
         .decode_utf8()
@@ -185,6 +189,26 @@ mod tests {
         assert!(parse("https://ntfy.sh/upZzH87cT9jJCc?up=1", &ours()).is_none());
     }
 
+    #[test]
+    fn every_managed_relay_leg_is_recognised_and_nothing_else_is() {
+        for path in [
+            format!("apns/canary/production/{TOKEN}"),
+            format!("apns-voip/canary/production/{TOKEN}"),
+            "fcm/canary/dYC_x9gXTjyyrG8_Aw3nUM%3AAPA91bExample".to_owned(),
+        ] {
+            let endpoint = format!("https://push.fluxer.com/relay/v1/{path}");
+            assert!(
+                is_managed(&endpoint, &ours()),
+                "{endpoint} must be a managed relay endpoint"
+            );
+        }
+        assert!(!is_managed("https://ntfy.sh/upZzH87cT9jJCc?up=1", &ours()));
+        assert!(!is_managed(
+            "https://updates.push.services.mozilla.com/wpush/v2/gAAAAA",
+            &ours()
+        ));
+    }
+
     #[test]
     fn a_malformed_relay_path_is_refused() {
         for endpoint in [

fluxer_push/src/rollout.rs

@@ -54,6 +54,7 @@ pub struct RolloutSnapshot {
     pub enabled: bool,
     pub config_version: u64,
     pub rollout_basis_points: u32,
+    pub relay_consent_accepted: bool,
 }
 
 impl RolloutConfig for RolloutSnapshot {
@@ -77,6 +78,7 @@ impl RolloutConfig for RolloutSnapshot {
             enabled: parse_enabled(config)?,
             config_version: parse_config_version(config)?,
             rollout_basis_points,
+            relay_consent_accepted: parse_flag(config, "relay_consent_accepted")?,
         })
     }
 
@@ -98,9 +100,13 @@ impl RolloutConfig for RolloutSnapshot {
 }
 
 pub fn parse_enabled(config: &Value) -> Option<bool> {
-    match config.get("enabled") {
+    parse_flag(config, "enabled")
+}
+
+fn parse_flag(config: &Value, key: &str) -> Option<bool> {
+    match config.get(key) {
         None | Some(Value::Null) => Some(false),
-        Some(Value::Bool(enabled)) => Some(*enabled),
+        Some(Value::Bool(flag)) => Some(*flag),
         Some(_) => None,
     }
 }
@@ -160,7 +166,7 @@ impl<C: RolloutConfig> RolloutStore<C> {
         self.update(config)
     }
 
-    fn update(&self, config: &Value) -> RolloutOutcome {
+    pub(crate) fn update(&self, config: &Value) -> RolloutOutcome {
         let Some(offered) = C::parse(config) else {
             warn!(config = C::NAME, "rollout config rejected as invalid");
             return RolloutOutcome::Rejected;
@@ -288,3 +294,33 @@ fn config_object<'a>(value: &'a Value, message_type: &str) -> Option<&'a Value>
     }
     value.is_object().then_some(value)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::json;
+
+    #[test]
+    fn the_operator_relay_consent_is_read_off_the_instance_config() {
+        let config = json!({
+            "enabled": true,
+            "config_version": 3,
+            "relay_consent_accepted": true,
+        });
+        let snapshot = RolloutSnapshot::parse(&config).expect("the config parses");
+        assert!(snapshot.relay_consent_accepted);
+    }
+
+    #[test]
+    fn an_instance_config_without_the_consent_field_has_not_consented() {
+        let config = json!({"enabled": true, "config_version": 3});
+        let snapshot = RolloutSnapshot::parse(&config).expect("the config parses");
+        assert!(!snapshot.relay_consent_accepted);
+    }
+
+    #[test]
+    fn a_consent_field_that_is_not_a_boolean_is_refused() {
+        let config = json!({"enabled": true, "relay_consent_accepted": "yes"});
+        assert!(RolloutSnapshot::parse(&config).is_none());
+    }
+}