environmentd: allow widening CORS allowlist via system variable

src/adapter/src/coord.rs

@@ -1251,6 +1251,9 @@ pub struct Config {
     pub aws_privatelink_availability_zones: Option<Vec<String>>,
     pub connection_context: ConnectionContext,
     pub connection_limit_callback: Box<dyn Fn(u64, u64) -> () + Send + Sync + 'static>,
+    /// Called with the raw value of the `http_additional_cors_allowed_origins`
+    /// system variable whenever it changes, and at least once at startup.
+    pub additional_cors_origins_callback: Box<dyn Fn(String) -> () + Send + Sync + 'static>,
     pub webhook_concurrency_limit: WebhookConcurrencyLimiter,
     pub http_host_name: Option<String>,
     pub tracing_handle: TracingHandle,
@@ -4654,6 +4657,7 @@ pub fn serve(
         aws_privatelink_availability_zones,
         connection_context,
         connection_limit_callback,
+        additional_cors_origins_callback,
         remote_system_parameters,
         webhook_concurrency_limit,
         http_host_name,
@@ -4957,6 +4961,17 @@ pub fn serve(
             connection_limit_callback,
         );
 
+        // Same pattern for the additional CORS origins, which the HTTP
+        // servers consult on every origin check.
+        catalog.system_config_mut().register_callback(
+            &mz_sql::session::vars::HTTP_ADDITIONAL_CORS_ALLOWED_ORIGINS,
+            Arc::new(move |system_vars: &SystemVars| {
+                (additional_cors_origins_callback)(
+                    system_vars.http_additional_cors_allowed_origins(),
+                );
+            }),
+        );
+
         let (group_commit_tx, group_commit_rx) = appends::notifier();
 
         let parent_span = tracing::Span::current();

src/environmentd/src/environmentd/main.rs

@@ -799,8 +799,7 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> {
         }
         allowed_origins
     };
-    let cors_allowed_origin = mz_http_util::build_cors_allowed_origin(&allowed_origins);
-    let cors_allowed_origin_list = allowed_origins.clone();
+    let cors_allowed_origin_list = allowed_origins;
 
     // Configure controller.
     let entered = info_span!("environmentd::configure_controller").entered();
@@ -1107,7 +1106,6 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> {
                 external_login_password_mz_system: args.external_login_password_mz_system,
                 frontegg,
                 frontegg_oauth_issuer_url,
-                cors_allowed_origin,
                 cors_allowed_origin_list,
                 egress_addresses: args.announce_egress_address,
                 http_host_name: args.http_host_name,

src/environmentd/src/http.rs

@@ -58,7 +58,7 @@ use std::collections::BTreeMap;
 use std::fmt::Debug;
 use std::net::{IpAddr, SocketAddr};
 use std::pin::Pin;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 use std::time::{Duration, SystemTime};
 
 use anyhow::Context;
@@ -158,10 +158,14 @@ pub struct HttpConfig {
     pub frontegg: Option<mz_frontegg_auth::Authenticator>,
     pub oidc_rx: Delayed<mz_authenticator::GenericOidcAuthenticator>,
     pub adapter_client_rx: Shared<Receiver<Client>>,
-    pub allowed_origin: AllowOrigin,
-    /// Raw list of allowed CORS origins, used by the MCP endpoints for
-    /// server-side Origin validation to defend against DNS rebinding.
+    /// Base allowlist of CORS origins, fixed at startup. The
+    /// `http_additional_cors_allowed_origins` system variable can widen the
+    /// allowlist at runtime, but entries in this list can never be removed
+    /// through it.
     pub allowed_origin_list: Vec<HeaderValue>,
+    /// Origins from the `http_additional_cors_allowed_origins` system
+    /// variable. Updated by the coordinator whenever the variable changes.
+    pub additional_allowed_origins: Arc<RwLock<Vec<HeaderValue>>>,
     pub active_connection_counter: ConnectionCounter,
     pub helm_chart_version: Option<String>,
     /// Externally-visible host name for this environment (without scheme).
@@ -214,6 +218,34 @@ pub struct WebhookState {
     dyncfgs: Arc<ConfigSet>,
 }
 
+/// The CORS origin allowlist: a base list fixed at startup, plus any origins
+/// from the `http_additional_cors_allowed_origins` system variable, which is
+/// read dynamically on every check. The system variable only widens the
+/// allowlist, base entries can never be removed through it.
+#[derive(Clone, Debug)]
+pub struct AllowedOrigins {
+    base: Arc<Vec<HeaderValue>>,
+    additional: Arc<RwLock<Vec<HeaderValue>>>,
+}
+
+impl AllowedOrigins {
+    fn is_allowed(&self, origin: &HeaderValue) -> bool {
+        if mz_http_util::origin_is_allowed(origin, &self.base) {
+            return true;
+        }
+        let additional = self
+            .additional
+            .read()
+            .expect("additional CORS origins lock poisoned");
+        mz_http_util::origin_is_allowed(origin, &additional)
+    }
+
+    fn to_allow_origin(&self) -> AllowOrigin {
+        let this = self.clone();
+        AllowOrigin::predicate(move |origin, _| this.is_allowed(origin))
+    }
+}
+
 #[derive(Clone, Debug)]
 struct HelmChartVersion(Option<String>);
 
@@ -232,8 +264,8 @@ impl HttpServer {
             frontegg,
             oidc_rx,
             adapter_client_rx,
-            allowed_origin,
             allowed_origin_list,
+            additional_allowed_origins,
             active_connection_counter,
             helm_chart_version,
             http_host_name,
@@ -252,6 +284,11 @@ impl HttpServer {
         let tls_enabled = tls.is_some();
         let webhook_cache = WebhookAppenderCache::new();
 
+        let allowed_origins = AllowedOrigins {
+            base: Arc::new(allowed_origin_list),
+            additional: additional_allowed_origins,
+        };
+
         // Compute OAuth discovery once per listener so the Bearer challenge
         // and the discovery handler always agree, and the middleware doesn't
         // re-derive (and re-allocate the Frontegg issuer) on each request.
@@ -612,7 +649,6 @@ impl HttpServer {
             // DNS rebinding attack the browser considers the request
             // same-origin, so no preflight fires and CORS enforcement is
             // bypassed.
-            let mcp_allowed_origins = Arc::new(allowed_origin_list.clone());
             mcp_router = mcp_router
                 .layer(auth_middleware.clone())
                 .layer(Extension(oauth_metadata::McpOAuthConfig {
@@ -622,12 +658,12 @@ impl HttpServer {
                 .layer(Extension(adapter_client_rx.clone()))
                 .layer(Extension(active_connection_counter.clone()))
                 .layer(Extension(HelmChartVersion(helm_chart_version.clone())))
-                .layer(Extension(mcp_allowed_origins))
+                .layer(Extension(allowed_origins.clone()))
                 .layer(Extension(mcp_metrics))
                 .layer(
                     CorsLayer::new()
                         .allow_methods(Method::POST)
-                        .allow_origin(allowed_origin.clone())
+                        .allow_origin(allowed_origins.to_allow_origin())
                         .allow_headers([AUTHORIZATION, CONTENT_TYPE]),
                 );
             router = router.merge(mcp_router);
@@ -647,7 +683,7 @@ impl HttpServer {
                         HeaderName::from_static("x-materialize-version"),
                     ])
                     .allow_methods(Any)
-                    .allow_origin(allowed_origin)
+                    .allow_origin(allowed_origins.to_allow_origin())
                     .expose_headers(Any)
                     .max_age(Duration::from_secs(60) * 60),
             );

src/environmentd/src/http/mcp.rs

@@ -23,13 +23,11 @@
 //!
 //! Data products are discovered via `mz_internal.mz_mcp_data_products` system view.
 
-use std::sync::Arc;
-
 use anyhow::anyhow;
 use axum::Extension;
 use axum::Json;
 use axum::response::IntoResponse;
-use http::{HeaderMap, HeaderValue, StatusCode};
+use http::{HeaderMap, StatusCode};
 use mz_adapter_types::dyncfgs::{
     ENABLE_MCP_AGENT, ENABLE_MCP_AGENT_QUERY_TOOL, ENABLE_MCP_AGENT_READ_DATA_PRODUCT_TOOL,
     ENABLE_MCP_DEVELOPER, ENABLE_MCP_DEVELOPER_QUERY_TOOL, MCP_MAX_RESPONSE_SIZE,
@@ -48,9 +46,9 @@ use serde_json::json;
 use thiserror::Error;
 use tracing::{debug, warn};
 
-use crate::http::AuthedClient;
 use crate::http::mcp_metrics::{McpMetrics, ToolCallGuard};
 use crate::http::sql::{SqlRequest, SqlResponse, SqlResult, execute_request};
+use crate::http::{AllowedOrigins, AuthedClient};
 
 // To add a new tool: add entry to tools/list, add handler function, add dispatch case.
 
@@ -413,7 +411,7 @@ pub async fn handle_mcp_method_not_allowed() -> impl IntoResponse {
 /// Agent endpoint: exposes user data products.
 pub async fn handle_mcp_agent(
     headers: HeaderMap,
-    Extension(allowed_origins): Extension<Arc<Vec<HeaderValue>>>,
+    Extension(allowed_origins): Extension<AllowedOrigins>,
     Extension(metrics): Extension<McpMetrics>,
     client: AuthedClient,
     Json(body): Json<McpRequest>,
@@ -429,7 +427,7 @@ pub async fn handle_mcp_agent(
 /// Developer endpoint: exposes system catalog (mz_*) only.
 pub async fn handle_mcp_developer(
     headers: HeaderMap,
-    Extension(allowed_origins): Extension<Arc<Vec<HeaderValue>>>,
+    Extension(allowed_origins): Extension<AllowedOrigins>,
     Extension(metrics): Extension<McpMetrics>,
     client: AuthedClient,
     Json(body): Json<McpRequest>,
@@ -452,10 +450,10 @@ pub async fn handle_mcp_developer(
 /// attacker arranges same-origin DNS rebinding (no preflight fires).
 fn validate_origin(
     headers: &HeaderMap,
-    allowed: &[HeaderValue],
+    allowed: &AllowedOrigins,
 ) -> Option<axum::response::Response> {
     let origin = headers.get(http::header::ORIGIN)?;
-    if mz_http_util::origin_is_allowed(origin, allowed) {
+    if allowed.is_allowed(origin) {
         return None;
     }
     warn!(
@@ -1539,6 +1537,10 @@ fn validate_system_catalog_query(sql: &str) -> Result<(), McpRequestError> {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
+    use http::HeaderValue;
+
     use super::*;
     use crate::http::sql::{Description, SqlError};
 
@@ -1617,28 +1619,43 @@ mod tests {
 
     /// The DNS-rebinding defense: a disallowed `Origin` is rejected with 403,
     /// an allowed one passes, and a missing one passes (non-browser clients).
+    /// Origins added via the `http_additional_cors_allowed_origins` system
+    /// variable are picked up without rebuilding the allowlist.
     #[mz_ore::test]
     fn test_validate_origin() {
-        let allowed = [HeaderValue::from_static("https://good.example")];
+        use std::sync::RwLock;
+
+        let additional: Arc<RwLock<Vec<HeaderValue>>> = Default::default();
+        let good = HeaderValue::from_static("https://good.example");
+        let allowed = AllowedOrigins {
+            base: Arc::new(vec![good.clone()]),
+            additional: Arc::clone(&additional),
+        };
 
         assert!(validate_origin(&HeaderMap::new(), &allowed).is_none());
 
         let mut ok = HeaderMap::new();
-        ok.insert(http::header::ORIGIN, allowed[0].clone());
+        ok.insert(http::header::ORIGIN, good);
         assert!(validate_origin(&ok, &allowed).is_none());
 
-        let mut bad = HeaderMap::new();
-        bad.insert(
+        let mut extra = HeaderMap::new();
+        extra.insert(
             http::header::ORIGIN,
             HeaderValue::from_static("https://evil.example"),
         );
-        let rejected = validate_origin(&bad, &allowed);
+        let rejected = validate_origin(&extra, &allowed);
         assert_eq!(
             rejected
                 .expect("disallowed origin must be rejected")
                 .status(),
             StatusCode::FORBIDDEN,
         );
+
+        // Adding the origin via the shared handle (as the coordinator does
+        // when the system variable changes) allows it dynamically.
+        *additional.write().expect("lock poisoned") =
+            vec![HeaderValue::from_static("https://evil.example")];
+        assert!(validate_origin(&extra, &allowed).is_none());
     }
 
     /// The two constructors set the JSON-RPC version and put the payload in

src/environmentd/src/lib.rs

@@ -69,7 +69,6 @@ use mz_server_core::{
 use mz_sql::catalog::EnvironmentId;
 use mz_sql::session::vars::{Value, VarInput};
 use tokio::sync::oneshot;
-use tower_http::cors::AllowOrigin;
 use tracing::{Instrument, info, info_span};
 
 use crate::deployment::preflight::{PreflightInput, PreflightOutput};
@@ -111,13 +110,11 @@ pub struct Config {
     pub frontegg: Option<FronteggAuthenticator>,
     /// Frontegg workspace URL advertised in MCP OAuth discovery.
     pub frontegg_oauth_issuer_url: Option<String>,
-    /// Origins for which cross-origin resource sharing (CORS) for HTTP requests
-    /// is permitted.
-    pub cors_allowed_origin: AllowOrigin,
-    /// Raw list of allowed CORS origins. Retained alongside `cors_allowed_origin`
-    /// (which is the computed predicate) so that endpoints like MCP can perform
-    /// server-side Origin validation to defend against DNS rebinding attacks
-    /// (where same-origin requests bypass CORS enforcement).
+    /// Base allowlist of origins for which cross-origin resource sharing
+    /// (CORS) for HTTP requests is permitted. The
+    /// `http_additional_cors_allowed_origins` system variable can widen the
+    /// allowlist at runtime, but entries in this list can never be removed
+    /// through it.
     pub cors_allowed_origin_list: Vec<HeaderValue>,
     /// Public IP addresses which the cloud environment has configured for
     /// egress.
@@ -396,6 +393,12 @@ impl Listeners {
         let mcp_metrics = http::mcp_metrics::McpMetrics::register_into(&metrics_registry);
         let oauth_metadata_metrics =
             http::oauth_metadata::OauthMetadataMetrics::register_into(&metrics_registry);
+        // Shared handle for origins from the
+        // `http_additional_cors_allowed_origins` system variable. The HTTP
+        // servers read it on every origin check, the coordinator updates it
+        // via `additional_cors_origins_callback` whenever the variable
+        // changes.
+        let additional_cors_origins: Arc<std::sync::RwLock<Vec<HeaderValue>>> = Default::default();
         let mut http_listener_handles = BTreeMap::new();
         for (name, listener) in self.http {
             let authenticator_kind = listener.config.authenticator_kind();
@@ -416,8 +419,8 @@ impl Listeners {
                 authenticator_kind,
                 frontegg: config.frontegg.clone(),
                 oidc_rx: authenticator_oidc_rx.clone(),
-                allowed_origin: config.cors_allowed_origin.clone(),
                 allowed_origin_list: config.cors_allowed_origin_list.clone(),
+                additional_allowed_origins: Arc::clone(&additional_cors_origins),
                 concurrent_webhook_req: webhook_concurrency_limit.semaphore(),
                 dyncfgs: Arc::clone(&config.system_dyncfgs),
                 metrics: metrics.clone(),
@@ -749,10 +752,17 @@ impl Listeners {
             connection_limiter.update_limit(limit);
             connection_limiter.update_superuser_reserved(superuser_reserved);
         });
+        let additional_cors_origins_callback = Box::new(move |origins: String| {
+            let parsed = mz_http_util::parse_origin_list(&origins);
+            *additional_cors_origins
+                .write()
+                .expect("additional CORS origins lock poisoned") = parsed;
+        });
 
         let (adapter_handle, adapter_client) = mz_adapter::serve(mz_adapter::Config {
             connection_context: config.controller.connection_context.clone(),
             connection_limit_callback,
+            additional_cors_origins_callback,
             controller_config: config.controller,
             controller_envd_epoch: envd_epoch,
             storage: adapter_storage,

src/environmentd/src/test_util.rs

@@ -83,7 +83,6 @@ use tokio::runtime::Runtime;
 use tokio_postgres::config::{Host, SslMode};
 use tokio_postgres::{AsyncMessage, Client};
 use tokio_stream::wrappers::TcpListenerStream;
-use tower_http::cors::AllowOrigin;
 use tracing::Level;
 use tracing_capture::SharedStorage;
 use tracing_subscriber::EnvFilter;
@@ -879,7 +878,6 @@ impl Listeners {
                 metrics_registry: metrics_registry.clone(),
                 now: config.now,
                 environment_id: config.environment_id,
-                cors_allowed_origin: AllowOrigin::list([]),
                 cors_allowed_origin_list: Vec::new(),
                 cluster_replica_sizes: ClusterReplicaSizeMap::for_tests(),
                 bootstrap_default_cluster_replica_size: config.default_cluster_replica_size,