Allow API cors behavior to be configured

Author: mwylde

Cargo.lock

@@ -41,6 +41,7 @@ utoipa-swagger-ui = { version = "7", features = ["axum"] }
 
 serde = { workspace = true }
 serde_json = { workspace = true }
+url = { workspace = true }
 
 # logging
 tracing = { workspace = true }
@@ -68,4 +69,4 @@ postgres = "0.19.5"
 arroyo-types = { path = "../arroyo-types" }
 utoipa = { workspace = true }
 rusqlite = {workspace =  true}
-refinery = { version = "0.8.14", features = ["rusqlite"] }
+refinery = { version = "0.8.14", features = ["rusqlite"] }

crates/arroyo-api/Cargo.toml

@@ -138,7 +138,7 @@ pub async fn start_server(database: DatabaseSource, guard: ShutdownGuard) -> any
     let config = config();
     let addr = SocketAddr::new(config.api.bind_address, config.api.http_port);
 
-    let mut app = rest::create_rest_app(database).layer(
+    let mut app = rest::create_rest_app(database)?.layer(
         CompressionLayer::new().zstd(true).compress_when(
             DefaultPredicate::new()
                 // compression doesn't work for server-sent events

crates/arroyo-api/src/lib.rs

@@ -4,10 +4,10 @@ use axum::{
     routing::{delete, get, patch, post, put},
 };
 
-use http::{HeaderMap, HeaderName, StatusCode, Uri, header};
+use anyhow::{Context, bail};
+use http::{HeaderMap, HeaderName, HeaderValue, StatusCode, Uri, header};
 use rust_embed::RustEmbed;
-use tower_http::cors;
-use tower_http::cors::CorsLayer;
+use tower_http::cors::{AllowHeaders, AllowMethods, AllowOrigin, Any, CorsLayer};
 use utoipa::OpenApi;
 use utoipa_swagger_ui::SwaggerUi;
 
@@ -31,7 +31,7 @@ use crate::pipelines::{
 };
 use crate::rest_utils::not_found;
 use crate::udfs::{create_udf, delete_udf, get_udfs, validate_udf};
-use arroyo_rpc::config::config;
+use arroyo_rpc::config::{CorsConfig, CorsOriginPolicy, config};
 use cornucopia_async::DatabaseSource;
 
 static BASENAME_HEADER: HeaderName = HeaderName::from_static("x-arroyo-basename");
@@ -156,12 +156,54 @@ async fn index_html(headers: HeaderMap) -> Response {
     }
 }
 
-pub fn create_rest_app(database: DatabaseSource) -> Router {
-    // TODO: enable in development only!!!
+fn cors_layer(config: &CorsConfig) -> anyhow::Result<CorsLayer> {
     let cors = CorsLayer::new()
-        .allow_methods(cors::Any)
-        .allow_headers(cors::Any)
-        .allow_origin(cors::Any);
+        .allow_methods(AllowMethods::mirror_request())
+        .allow_headers(AllowHeaders::mirror_request());
+
+    match config.origin_policy {
+        CorsOriginPolicy::Any => {
+            if config.allow_credentials {
+                bail!("CORS credentials cannot be enabled when allowing any origin");
+            }
+            if !config.allowed_origins.is_empty() {
+                bail!("CORS allowed origins must be empty when allowing any origin");
+            }
+
+            Ok(cors.allow_credentials(false).allow_origin(Any))
+        }
+        CorsOriginPolicy::AllowList => {
+            if config.allowed_origins.iter().any(|origin| origin == "*") {
+                bail!("CORS allow-list cannot contain the wildcard origin '*'");
+            }
+
+            let origins = config
+                .allowed_origins
+                .iter()
+                .map(|origin| {
+                    let url = url::Url::parse(origin)
+                        .with_context(|| format!("invalid CORS allowed origin {origin:?}"))?;
+                    if url.origin().ascii_serialization() != *origin {
+                        bail!(
+                            "CORS allowed origin {origin:?} must contain only a canonical scheme, host, and optional port"
+                        );
+                    }
+
+                    origin
+                        .parse::<HeaderValue>()
+                        .with_context(|| format!("invalid CORS allowed origin {origin:?}"))
+                })
+                .collect::<anyhow::Result<Vec<_>>>()?;
+
+            Ok(cors
+                .allow_credentials(config.allow_credentials)
+                .allow_origin(AllowOrigin::list(origins)))
+        }
+    }
+}
+
+pub fn create_rest_app(database: DatabaseSource) -> anyhow::Result<Router> {
+    let cors = cors_layer(&config().api.cors)?;
 
     let api_routes = Router::new()
         .route("/ping", get(ping))
@@ -189,7 +231,7 @@ pub fn create_rest_app(database: DatabaseSource) -> Router {
         .merge(pipeline_and_job_routes())
         .fallback(api_fallback);
 
-    Router::new()
+    Ok(Router::new()
         .merge(
             SwaggerUi::new("/api/v1/swagger-ui")
                 .url("/api/v1/api-docs/openapi.json", ApiDoc::openapi()),
@@ -201,5 +243,5 @@ pub fn create_rest_app(database: DatabaseSource) -> Router {
         )
         .fallback(static_handler)
         .with_state(AppState { database })
-        .layer(cors)
+        .layer(cors))
 }

crates/arroyo-api/src/rest.rs

@@ -28,6 +28,11 @@ checkpoints-to-compact = 4
 bind-address = "0.0.0.0"
 http-port = 5115
 
+[api.cors]
+origin-policy = "any"
+allowed-origins = []
+allow-credentials = false
+
 [controller]
 bind-address = "0.0.0.0"
 rpc-port = 5116

crates/arroyo-rpc/default.toml

@@ -356,6 +356,30 @@ pub struct ApiConfig {
 
     #[serde(default)]
     pub auth_mode: ApiAuthMode,
+
+    #[serde(default)]
+    pub cors: CorsConfig,
+}
+
+#[derive(Debug, Deserialize, Serialize, Clone, Default)]
+#[serde(rename_all = "kebab-case", deny_unknown_fields)]
+pub struct CorsConfig {
+    #[serde(default)]
+    pub origin_policy: CorsOriginPolicy,
+
+    #[serde(default)]
+    pub allowed_origins: Vec<String>,
+
+    #[serde(default)]
+    pub allow_credentials: bool,
+}
+
+#[derive(Debug, Deserialize, Serialize, Copy, Clone, Default, PartialEq, Eq)]
+#[serde(rename_all = "kebab-case")]
+pub enum CorsOriginPolicy {
+    #[default]
+    Any,
+    AllowList,
 }
 
 #[derive(Debug, Deserialize, Serialize, Clone)]