feat: scoped cf-access-token header for internal endpoints

cli/src/access.rs

@@ -0,0 +1,57 @@
+// Copyright (c) 2023-2024 Cloudflare, Inc.
+// Licensed under the BSD-3-Clause license found in the LICENSE file or at https://opensource.org/licenses/BSD-3-Clause
+
+//! Cloudflare Access support for testing gated endpoints.
+//!
+//! When the `_MACH_CF_ACCESS_TOKEN` environment variable is set, mach attaches a
+//! `cf-access-token` header to its test requests so that Cloudflare
+//! Access-gated hosts can be reached. The token is only ever attached to hosts
+//! on the allowlist below so that it is never leaked to an unrelated origin
+//! returned by a remote configuration.
+
+use anyhow::Context;
+use http::{HeaderMap, HeaderValue};
+use nq_core::ScopedHeaders;
+
+/// Environment variable holding the Cloudflare Access token.
+const ACCESS_TOKEN_ENV: &str = "_MACH_CF_ACCESS_TOKEN";
+
+/// Header used to carry the Cloudflare Access token.
+const ACCESS_TOKEN_HEADER: &str = "cf-access-token";
+
+/// Domain suffixes the access token may be sent to. Matching is done with
+/// strict label boundaries by [`ScopedHeaders`].
+const ALLOWED_HOST_SUFFIXES: &[&str] = &["speed.cloudflare.com"];
+
+/// Build the scoped headers carrying the Cloudflare Access token, if the
+/// `_MACH_CF_ACCESS_TOKEN` environment variable is set to a non-empty value.
+///
+/// Returns `Ok(None)` when the variable is unset or blank. Returns an error
+/// when the token is not a valid header value, so a misconfigured token fails
+/// fast at startup rather than silently sending no header.
+pub fn cf_access_scoped_headers() -> anyhow::Result<Option<ScopedHeaders>> {
+    let Ok(token) = std::env::var(ACCESS_TOKEN_ENV) else {
+        return Ok(None);
+    };
+
+    let token = token.trim();
+    if token.is_empty() {
+        return Ok(None);
+    }
+
+    let mut value = HeaderValue::from_str(token)
+        .with_context(|| format!("{ACCESS_TOKEN_ENV} is not a valid header value"))?;
+    // Mark the value sensitive so it is redacted (printed as `Sensitive`) in
+    // request/response debug logs.
+    value.set_sensitive(true);
+
+    let mut headers = HeaderMap::new();
+    headers.insert(ACCESS_TOKEN_HEADER, value);
+
+    let allowed_host_suffixes = ALLOWED_HOST_SUFFIXES
+        .iter()
+        .map(|s| s.to_string())
+        .collect();
+
+    Ok(Some(ScopedHeaders::new(headers, allowed_host_suffixes)))
+}

cli/src/latency.rs

@@ -22,6 +22,7 @@ pub async fn run(url: String, runs: usize) -> anyhow::Result<()> {
     let result = run_test(&LatencyConfig {
         url: url.parse()?,
         runs,
+        scoped_headers: crate::access::cf_access_scoped_headers()?,
     })
     .await?;
 

cli/src/main.rs

@@ -1,6 +1,7 @@
 // Copyright (c) 2023-2024 Cloudflare, Inc.
 // Licensed under the BSD-3-Clause license found in the LICENSE file or at https://opensource.org/licenses/BSD-3-Clause
 
+mod access;
 mod aim_report;
 pub(crate) mod args;
 mod latency;

cli/src/packet_loss.rs

@@ -26,6 +26,7 @@ pub async fn run(args: PacketLossArgs) -> anyhow::Result<()> {
         response_wait_time: Duration::from_millis(args.response_wait_time_ms),
         download_url: args.download_url.parse()?,
         upload_url: args.upload_url.parse()?,
+        scoped_headers: crate::access::cf_access_scoped_headers()?,
     };
 
     info!("fetching TURN server credentials");
@@ -58,10 +59,13 @@ async fn fetch_turn_server_creds(
     let mut headers = HeaderMap::new();
     headers.append(hyper::header::HOST, HeaderValue::from_str(host)?);
 
-    let response = Client::default()
+    let client = Client::default()
         .new_connection(ConnectionType::h1())
         .method("GET")
         .headers(headers)
+        .scoped_headers(config.scoped_headers.clone());
+
+    let response = client
         .send(
             request_url.to_string().parse()?,
             http_body_util::Empty::new(),

cli/src/rpm.rs

@@ -25,10 +25,12 @@ use crate::util::pretty_secs_to_ms;
 pub async fn run(cli_config: RpmArgs) -> anyhow::Result<()> {
     info!("running responsiveness test");
 
+    let scoped_headers = crate::access::cf_access_scoped_headers()?;
+
     let rpm_urls = match cli_config.config.clone() {
         Some(endpoint) => {
             info!("fetching configuration from {endpoint}");
-            let urls = get_rpm_config(endpoint).await?.urls;
+            let urls = get_rpm_config(endpoint, scoped_headers.clone()).await?.urls;
             info!("retrieved configuration urls: {urls:?}");
 
             urls
@@ -53,6 +55,7 @@ pub async fn run(cli_config: RpmArgs) -> anyhow::Result<()> {
     let rtt_result = crate::latency::run_test(&LatencyConfig {
         url: rpm_urls.small_https_download_url.parse()?,
         runs: 20,
+        scoped_headers: scoped_headers.clone(),
     })
     .await?;
     info!(
@@ -79,6 +82,7 @@ pub async fn run(cli_config: RpmArgs) -> anyhow::Result<()> {
         max_loaded_connections: cli_config.max_loaded_connections,
         conn_type: ConnectionType::H2,
         determine_load_only: false,
+        scoped_headers,
     };
 
     info!("running download test");
@@ -156,17 +160,23 @@ pub struct RpmUrls {
     https_upload_url: String,
 }
 
-pub async fn get_rpm_config(config_url: String) -> anyhow::Result<RpmServerConfig> {
+pub async fn get_rpm_config(
+    config_url: String,
+    scoped_headers: Option<nq_core::ScopedHeaders>,
+) -> anyhow::Result<RpmServerConfig> {
     let shutdown = CancellationToken::new();
     let time = Arc::new(TokioTime::new());
     let network = Arc::new(TokioNetwork::new(
         Arc::clone(&time) as Arc<dyn Time>,
         shutdown.clone(),
     ));
 
-    let response = Client::default()
+    let client = Client::default()
         .new_connection(ConnectionType::H2)
         .method("GET")
+        .scoped_headers(scoped_headers);
+
+    let response = client
         .send(
             config_url.parse().context("parsing rpm config url")?,
             http_body_util::Empty::new(),

cli/src/saturate.rs

@@ -30,6 +30,7 @@ pub async fn run(cli_config: SaturateArgs) -> anyhow::Result<()> {
         determine_load_only: true,
         conn_type: cli_config.conn_type.into(),
         test_duration: Duration::from_secs(cli_config.duration),
+        scoped_headers: crate::access::cf_access_scoped_headers()?,
         ..Default::default()
     };
 

cli/src/up_down.rs

@@ -26,8 +26,11 @@ pub async fn download(args: DownloadArgs) -> anyhow::Result<()> {
     let conn_type = args.conn_type.into();
     info!("downloading: {}", args.url);
 
-    let inflight_body = ThroughputClient::download()
+    let client = ThroughputClient::download()
         .new_connection(conn_type)
+        .scoped_headers(crate::access::cf_access_scoped_headers()?);
+
+    let inflight_body = client
         .send(
             args.url.parse().context("parsing download url")?,
             Arc::clone(&network),
@@ -107,9 +110,12 @@ pub async fn upload(args: UploadArgs) -> anyhow::Result<()> {
         info!("added header: {key}");
     }
 
-    let inflight_body = ThroughputClient::upload(bytes)
+    let client = ThroughputClient::upload(bytes)
         .new_connection(conn_type)
         .headers(headers)
+        .scoped_headers(crate::access::cf_access_scoped_headers()?);
+
+    let inflight_body = client
         .send(
             args.url.parse()?,
             Arc::clone(&network),

crates/nq-core/src/client.rs

@@ -19,7 +19,7 @@ use tokio_util::sync::CancellationToken;
 use tracing::{Instrument, debug, error, info, trace};
 
 use crate::{
-    ConnectionType, EstablishedConnection, Network, OneshotResult, Time, Timestamp,
+    ConnectionType, EstablishedConnection, Network, OneshotResult, ScopedHeaders, Time, Timestamp,
     body::{BodyEvent, CountingBody, InflightBody, NqBody, UploadBody, empty},
     oneshot_result,
 };
@@ -49,6 +49,7 @@ pub struct ThroughputClient {
     connection: Option<Arc<RwLock<EstablishedConnection>>>,
     new_connection_type: Option<ConnectionType>,
     headers: Option<HeaderMap>,
+    scoped_headers: Option<ScopedHeaders>,
     direction: Direction,
 }
 
@@ -59,6 +60,7 @@ impl ThroughputClient {
             connection: None,
             new_connection_type: None,
             headers: None,
+            scoped_headers: None,
             direction: Direction::Down,
         }
     }
@@ -69,6 +71,7 @@ impl ThroughputClient {
             connection: None,
             new_connection_type: None,
             headers: None,
+            scoped_headers: None,
             direction: Direction::Up(size),
         }
     }
@@ -91,6 +94,13 @@ impl ThroughputClient {
         self
     }
 
+    /// Set headers that are only attached when the request's host matches the
+    /// scope's allowlist. Headers set via [`Self::headers`] take precedence.
+    pub fn scoped_headers(mut self, scoped_headers: Option<ScopedHeaders>) -> Self {
+        self.scoped_headers = scoped_headers;
+        self
+    }
+
     /// Execute a download or upload request against the given [`Uri`].
     // #[tracing::instrument(skip(self, network, time, shutdown))]
     pub fn send(
@@ -148,6 +158,10 @@ impl ThroughputClient {
             }
         };
 
+        if let Some(scoped_headers) = self.scoped_headers.take() {
+            scoped_headers.apply(&uri, &mut headers);
+        }
+
         let mut request = http::Request::builder()
             .method(method)
             .uri(uri)
@@ -355,6 +369,7 @@ pub struct Client {
     connection: Option<Arc<RwLock<EstablishedConnection>>>,
     new_connection_type: Option<ConnectionType>,
     headers: Option<HeaderMap>,
+    scoped_headers: Option<ScopedHeaders>,
     method: Option<String>,
 }
 
@@ -377,6 +392,13 @@ impl Client {
         self
     }
 
+    /// Set headers that are only attached when the request's host matches the
+    /// scope's allowlist. Headers set via [`Self::headers`] take precedence.
+    pub fn scoped_headers(mut self, scoped_headers: Option<ScopedHeaders>) -> Self {
+        self.scoped_headers = scoped_headers;
+        self
+    }
+
     /// Set the method used by the client.
     pub fn method(mut self, method: &str) -> Self {
         self.method = Some(method.to_string());
@@ -411,6 +433,10 @@ impl Client {
 
         let method: http::Method = self.method.as_deref().unwrap_or("GET").parse()?;
 
+        if let Some(scoped_headers) = self.scoped_headers {
+            scoped_headers.apply(&uri, &mut headers);
+        }
+
         let mut request = http::Request::builder()
             .method(method)
             .uri(uri)

crates/nq-core/src/lib.rs

@@ -13,6 +13,7 @@ mod body;
 pub mod client;
 mod connection;
 mod network;
+mod scoped_headers;
 mod time;
 mod upgraded;
 mod util;
@@ -21,6 +22,7 @@ pub use crate::{
     body::{BodyEvent, CountingBody, NqBody},
     connection::{ConnectionManager, ConnectionTiming, ConnectionType, EstablishedConnection},
     network::Network,
+    scoped_headers::ScopedHeaders,
     time::{Time, Timestamp, TokioTime},
     upgraded::ConnectUpgraded,
     util::{OneshotResult, ResponseFuture, oneshot_result},