Add bearer token authorization check

crates/api-iceberg-rest/src/auth.rs

@@ -0,0 +1,39 @@
+use axum::{
+    extract::{Request, State},
+    http::StatusCode,
+    middleware::Next,
+    response::{IntoResponse, Response},
+};
+
+use crate::state::State as AppState;
+
+/// Middleware to validate Bearer token authorization for Iceberg REST API
+///
+/// If a bearer token is configured in the state, validates incoming requests
+/// against it. Requests without proper authorization return 401 Unauthorized.
+/// If no bearer token is configured, all requests are allowed through.
+pub async fn require_auth(State(state): State<AppState>, req: Request, next: Next) -> Response {
+    // If no bearer token is configured, allow all requests
+    let Some(configured_token) = &state.bearer_token else {
+        return next.run(req).await;
+    };
+
+    // Extract the bearer token from the Authorization header
+    let auth_header = req.headers().get(axum::http::header::AUTHORIZATION);
+
+    let provided_token = if let Some(auth_header) = auth_header {
+        if let Ok(auth_str) = auth_header.to_str() {
+            auth_str.strip_prefix("Bearer ").map(|s| s.to_string())
+        } else {
+            None
+        }
+    } else {
+        None
+    };
+
+    match provided_token {
+        Some(token) if &token == configured_token => next.run(req).await,
+        Some(_) => (StatusCode::UNAUTHORIZED, "Invalid bearer token").into_response(),
+        None => (StatusCode::UNAUTHORIZED, "Missing Authorization header").into_response(),
+    }
+}

crates/api-iceberg-rest/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod auth;
 pub mod error;
 pub mod handlers;
 pub mod router;

crates/api-iceberg-rest/src/state.rs

@@ -12,11 +12,20 @@ pub struct Config {
 pub struct State {
     pub metastore: Arc<dyn Metastore + Send + Sync>,
     pub config: Arc<Config>,
+    pub bearer_token: Option<String>,
 }
 
 impl State {
     // You can add helper methods for state initialization if needed
-    pub fn new(metastore: Arc<dyn Metastore + Send + Sync>, config: Arc<Config>) -> Self {
-        Self { metastore, config }
+    pub fn new(
+        metastore: Arc<dyn Metastore + Send + Sync>,
+        config: Arc<Config>,
+        bearer_token: Option<String>,
+    ) -> Self {
+        Self {
+            metastore,
+            config,
+            bearer_token,
+        }
     }
 }

crates/embucketd/src/cli.rs

@@ -241,6 +241,14 @@ pub struct CliOpts {
     )]
     jwt_secret: Option<String>,
 
+    #[arg(
+        long,
+        env = "ICEBERG_REST_BEARER_TOKEN",
+        hide_env_values = true,
+        help = "Bearer token for Iceberg REST API authorization"
+    )]
+    pub iceberg_rest_bearer_token: Option<String>,
+
     #[arg(
         long,
         env = "AUTH_DEMO_USER",

crates/embucketd/src/main.rs

@@ -5,6 +5,7 @@ pub(crate) mod cli;
 pub(crate) mod helpers;
 pub(crate) mod layers;
 
+use api_iceberg_rest::auth::require_auth as iceberg_require_auth;
 use api_iceberg_rest::router::create_router as create_iceberg_router;
 use api_iceberg_rest::state::Config as IcebergConfig;
 use api_iceberg_rest::state::State as IcebergAppState;
@@ -258,10 +259,17 @@ async fn async_main(
         .with_state(snowflake_state.clone())
         .layer(compression_layer);
     let snowflake_router = snowflake_router.merge(snowflake_auth_router);
-    let iceberg_router = create_iceberg_router().with_state(IcebergAppState {
+    let iceberg_state = IcebergAppState {
         metastore: metastore.clone(),
         config: Arc::new(iceberg_config),
-    });
+        bearer_token: opts.iceberg_rest_bearer_token.clone(),
+    };
+    let iceberg_router = create_iceberg_router()
+        .with_state(iceberg_state.clone())
+        .layer(middleware::from_fn_with_state(
+            iceberg_state,
+            iceberg_require_auth,
+        ));
 
     // --- OpenAPI specs ---
     let mut spec = ApiDoc::openapi();