Security fixes and improvements (spiceai#9666)

* fix: update copyright year and enhance API key validation logic

* style: format code for better readability in load and store modules

* Address review comments and fix runtime compile errors

* Address remaining PR review comments

* Refactor Cayenne IDs to use UUIDv7 strings (spiceai#9667)

* Refactor table and partition identifiers to use UUIDv7 strings

- Changed `table_id`, `data_file_id`, `delete_file_id`, and `partition_id` fields in various structs from `i64` to `String` to accommodate UUIDv7 format.
- Updated related methods and implementations to handle the new string-based identifiers.
- Removed `table_uuid` from the `TableMetadata` struct and adjusted SQL schemas accordingly.
- Modified tests and catalog operations to work with the new identifier types.

(cherry picked from commit a36ebe4)

* Fix UUIDv7 ID refactor on trunk

* Refactor test_insert_overwrite to use dynamic table_id for directory paths

* Enhance metadata catalog to handle concurrent table creation and validate UUIDs in commit compaction

* Fix test for oversized total payload in StoreMemoryParams validation

Author: lukekim

Cargo.lock

@@ -1,5 +1,5 @@
 /*
-Copyright 2024-2025 The Spice.ai OSS Authors
+Copyright 2024-2026 The Spice.ai OSS Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -19,6 +19,8 @@ use logos::{Lexer, Logos};
 use std::collections::HashMap;
 use std::sync::Arc;
 
+const MAX_RECURSION_DEPTH: usize = 100;
+
 #[derive(Logos, Debug, PartialEq, Clone)]
 #[logos(skip r"[ \t\n\f]+")] // Skip whitespace
 pub enum Token<'input> {
@@ -111,7 +113,7 @@ impl<'input> Parser<'input> {
     }
 
     pub fn parse(&mut self) -> Result<ArrowDataType, String> {
-        self.parse_data_type()
+        self.parse_data_type_with_depth(0)
     }
 
     fn parse_decimal(&mut self) -> Result<ArrowDataType, String> {
@@ -145,7 +147,13 @@ impl<'input> Parser<'input> {
         })
     }
 
-    fn parse_data_type(&mut self) -> Result<ArrowDataType, String> {
+    fn parse_data_type_with_depth(&mut self, depth: usize) -> Result<ArrowDataType, String> {
+        if depth > MAX_RECURSION_DEPTH {
+            return Err(format!(
+                "Maximum schema recursion depth exceeded ({MAX_RECURSION_DEPTH})"
+            ));
+        }
+
         match self.current.clone() {
             Some(Ok(Token::BigInt)) => {
                 self.advance();
@@ -206,23 +214,23 @@ impl<'input> Parser<'input> {
             Some(Ok(Token::Array)) => {
                 self.advance();
                 self.expect(&Token::LAngle)?;
-                let inner_type = self.parse_data_type()?;
+                let inner_type = self.parse_data_type_with_depth(depth + 1)?;
                 self.expect(&Token::RAngle)?;
                 let field = ArrowField::new("item", inner_type, true);
                 Ok(ArrowDataType::List(Arc::new(field)))
             }
-            Some(Ok(Token::Map)) => self.parse_map(),
-            Some(Ok(Token::Struct)) => self.parse_struct(),
+            Some(Ok(Token::Map)) => self.parse_map_with_depth(depth),
+            Some(Ok(Token::Struct)) => self.parse_struct_with_depth(depth),
             _ => Err(format!("Unexpected token: {:?}", self.current)),
         }
     }
 
-    fn parse_map(&mut self) -> Result<ArrowDataType, String> {
+    fn parse_map_with_depth(&mut self, depth: usize) -> Result<ArrowDataType, String> {
         self.advance();
         self.expect(&Token::LAngle)?;
-        let key_type = self.parse_data_type()?;
+        let key_type = self.parse_data_type_with_depth(depth + 1)?;
         self.expect(&Token::Comma)?;
-        let value_type = self.parse_data_type()?;
+        let value_type = self.parse_data_type_with_depth(depth + 1)?;
         self.expect(&Token::RAngle)?;
         let key_field = Arc::new(ArrowField::new("key", key_type, false));
         let value_field = Arc::new(ArrowField::new("value", value_type, true));
@@ -234,13 +242,13 @@ impl<'input> Parser<'input> {
         Ok(ArrowDataType::Map(entry_struct, false))
     }
 
-    fn parse_struct(&mut self) -> Result<ArrowDataType, String> {
+    fn parse_struct_with_depth(&mut self, depth: usize) -> Result<ArrowDataType, String> {
         self.advance();
         self.expect(&Token::LAngle)?;
         let mut fields = Vec::new();
         if self.current != Some(Ok(Token::RAngle)) {
             loop {
-                let field = self.parse_field()?;
+                let field = self.parse_field_with_depth(depth + 1)?;
                 fields.push(field);
                 if self.current == Some(Ok(Token::Comma)) {
                     self.advance();
@@ -283,7 +291,7 @@ impl<'input> Parser<'input> {
         )
     }
 
-    fn parse_field(&mut self) -> Result<ArrowField, String> {
+    fn parse_field_with_depth(&mut self, depth: usize) -> Result<ArrowField, String> {
         let name = match self.current.clone() {
             Some(Ok(Token::Identifier(name))) => {
                 self.advance();
@@ -297,7 +305,7 @@ impl<'input> Parser<'input> {
             _ => return Err("Expected identifier for field name".to_string()),
         };
         self.expect(&Token::Colon)?;
-        let data_type = self.parse_data_type()?;
+        let data_type = self.parse_data_type_with_depth(depth)?;
         let nullable = if self.current == Some(Ok(Token::Not)) {
             self.advance();
             self.expect(&Token::Null)?;
@@ -422,6 +430,20 @@ mod tests {
         }
     }
 
+    #[test]
+    fn test_parse_rejects_excessive_recursion_depth() {
+        let input = format!(
+            "{}INT{}",
+            "ARRAY<".repeat(MAX_RECURSION_DEPTH + 1),
+            ">".repeat(MAX_RECURSION_DEPTH + 1)
+        );
+
+        let mut parser = Parser::new(&input);
+        let err = parser.parse().expect_err("must reject excessive recursion");
+
+        assert!(err.contains("Maximum schema recursion depth exceeded"));
+    }
+
     #[test]
     fn test_struct_reserved_field_names() {
         let input = "STRUCT<date: STRING, value: INT>";

crates/data_components/src/databricks/sql_warehouse/datatypes.rs

@@ -1,5 +1,5 @@
 /*
-Copyright 2024-2025 The Spice.ai OSS Authors
+Copyright 2024-2026 The Spice.ai OSS Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -28,7 +28,27 @@ pub struct ApiKeyAuth {
 impl ApiKeyAuth {
     #[must_use]
     pub fn new(api_keys: Vec<ApiKey>) -> Self {
-        Self { api_keys }
+        let mut filtered_keys = Vec::with_capacity(api_keys.len());
+        let mut dropped_empty_keys = 0usize;
+
+        for api_key in api_keys {
+            if api_key.is_empty() {
+                dropped_empty_keys += 1;
+            } else {
+                filtered_keys.push(api_key);
+            }
+        }
+
+        if dropped_empty_keys > 0 {
+            tracing::warn!(
+                dropped_empty_keys,
+                "Ignoring empty API key values to prevent insecure authentication bypass"
+            );
+        }
+
+        Self {
+            api_keys: filtered_keys,
+        }
     }
 }
 
@@ -41,6 +61,10 @@ impl HttpAuth for ApiKeyAuth {
             .and_then(|value| value.to_str().ok())
             .unwrap_or_default();
 
+        if api_key.is_empty() {
+            return Ok(AuthVerdict::Deny);
+        }
+
         if let Some(api_key) = self.api_keys.iter().find(|key| *key == api_key) {
             Ok(AuthVerdict::Allow(Arc::new(api_key.clone())))
         } else {
@@ -51,6 +75,10 @@ impl HttpAuth for ApiKeyAuth {
 
 impl FlightBasicAuth for ApiKeyAuth {
     fn validate(&self, _username: &str, password: &str) -> Result<String, Error> {
+        if password.is_empty() {
+            return Err(Error::InvalidCredentials);
+        }
+
         if self.api_keys.iter().any(|key| key == password) {
             Ok(password.to_string())
         } else {
@@ -59,6 +87,10 @@ impl FlightBasicAuth for ApiKeyAuth {
     }
 
     fn is_valid(&self, bearer_token: &str) -> Result<AuthVerdict, Error> {
+        if bearer_token.is_empty() {
+            return Ok(AuthVerdict::Deny);
+        }
+
         if let Some(api_key) = self.api_keys.iter().find(|key| *key == bearer_token) {
             Ok(AuthVerdict::Allow(Arc::new(api_key.clone())))
         } else {
@@ -77,6 +109,10 @@ impl GrpcAuth for ApiKeyAuth {
             return Ok(AuthVerdict::Deny);
         };
 
+        if api_key.is_empty() {
+            return Ok(AuthVerdict::Deny);
+        }
+
         if let Some(api_key) = self.api_keys.iter().find(|key| *key == api_key) {
             Ok(AuthVerdict::Allow(Arc::new(api_key.clone())))
         } else {
@@ -140,4 +176,16 @@ mod tests {
         let result = auth.http_verify(&parts);
         assert!(matches!(result, Ok(AuthVerdict::Allow(_))));
     }
+
+    #[test]
+    fn test_empty_configured_key_is_ignored() {
+        let auth = ApiKeyAuth::new(vec![ApiKey::parse_str(""), ApiKey::parse_str("valid-key")]);
+        let empty_key_parts = create_request_parts(Some(""));
+        let empty_key_result = auth.http_verify(&empty_key_parts);
+        assert!(matches!(empty_key_result, Ok(AuthVerdict::Deny)));
+
+        let valid_key_parts = create_request_parts(Some("valid-key"));
+        let valid_key_result = auth.http_verify(&valid_key_parts);
+        assert!(matches!(valid_key_result, Ok(AuthVerdict::Allow(_))));
+    }
 }

crates/runtime-auth/src/api_key/mod.rs

@@ -1,5 +1,5 @@
 /*
-Copyright 2024-2025 The Spice.ai OSS Authors
+Copyright 2024-2026 The Spice.ai OSS Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

crates/runtime-auth/src/layer/flight.rs

@@ -19,6 +19,7 @@ opentelemetry.workspace = true
 percent-encoding.workspace = true
 regex.workspace = true
 runtime-auth = { path = "../runtime-auth" }
+sha2.workspace = true
 spicepod = { path = "../spicepod" }
 tokio.workspace = true
 tracing.workspace = true

crates/runtime-request-context/Cargo.toml

@@ -1,5 +1,5 @@
 /*
-Copyright 2024-2025 The Spice.ai OSS Authors
+Copyright 2024-2026 The Spice.ai OSS Authors
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ use http::HeaderMap;
 use opentelemetry::KeyValue;
 use regex::Regex;
 use runtime_auth::{AuthPrincipalRef, AuthRequestContext};
+use sha2::{Digest, Sha256};
 use spicepod::component::runtime::UserAgentCollection;
 use std::sync::atomic::Ordering;
 use std::{
@@ -81,6 +82,8 @@ static CLIENT_CACHE_KEY_REGEX: LazyLock<Regex> =
         Err(e) => unreachable!("Unable to compile regexp: {}", e),
     });
 
+const AUTHORIZATION_SCOPE_FINGERPRINT_BYTES: usize = 16;
+
 #[derive(Copy, Clone)]
 pub struct AsyncMarker {
     marker: PhantomData<()>,
@@ -244,6 +247,17 @@ impl RequestContext {
         self.authorization_header.as_deref()
     }
 
+    #[must_use]
+    pub fn scoped_client_supplied_cache_key(&self) -> Option<String> {
+        self.client_supplied_cache_key.as_deref().map(|cache_key| {
+            let auth_scope = self
+                .authorization_header
+                .as_deref()
+                .map_or_else(|| "anonymous".to_string(), authorization_scope_fingerprint);
+            format!("{auth_scope}:{cache_key}")
+        })
+    }
+
     pub fn extension<T>(&self) -> Option<T>
     where
         T: Extension + Clone,
@@ -412,6 +426,12 @@ impl RequestContextBuilder {
         self
     }
 
+    #[must_use]
+    pub fn with_authorization_header(mut self, authorization_header: Option<String>) -> Self {
+        self.authorization_header = authorization_header;
+        self
+    }
+
     #[must_use]
     pub fn baggage_mut(&mut self) -> &mut Vec<KeyValue> {
         &mut self.baggage
@@ -520,6 +540,25 @@ impl RequestContextBuilder {
     }
 }
 
+fn authorization_scope_fingerprint(authorization_header: &str) -> String {
+    let digest = Sha256::digest(authorization_header.as_bytes());
+    let mut fingerprint = String::with_capacity(5 + (AUTHORIZATION_SCOPE_FINGERPRINT_BYTES * 2));
+    fingerprint.push_str("auth:");
+
+    for byte in &digest[..AUTHORIZATION_SCOPE_FINGERPRINT_BYTES] {
+        push_hex_byte(&mut fingerprint, *byte);
+    }
+
+    fingerprint
+}
+
+fn push_hex_byte(buf: &mut String, byte: u8) {
+    const HEX: &[u8; 16] = b"0123456789abcdef";
+
+    buf.push(HEX[usize::from(byte >> 4)] as char);
+    buf.push(HEX[usize::from(byte & 0x0f)] as char);
+}
+
 #[cfg(test)]
 mod tests {
     use http::{HeaderMap, HeaderValue};
@@ -560,4 +599,58 @@ mod tests {
         );
         assert_eq!(ctx_bad_user_key.client_supplied_cache_key, None);
     }
+
+    #[test]
+    fn test_scoped_client_supplied_cache_key_hashes_authorization_header() {
+        let auth_header_1 = "Bearer alice".to_string();
+        let auth_header_2 = "Bearer bob".to_string();
+
+        let ctx1 = RequestContextBuilder::new(Protocol::Internal)
+            .with_cache_control(CacheControl::Cache(CacheKeyType::ClientSupplied))
+            .with_client_supplied_cache_key(Some("shared-key".to_string()))
+            .with_authorization_header(Some(auth_header_1.clone()))
+            .build();
+
+        let ctx2 = RequestContextBuilder::new(Protocol::Internal)
+            .with_cache_control(CacheControl::Cache(CacheKeyType::ClientSupplied))
+            .with_client_supplied_cache_key(Some("shared-key".to_string()))
+            .with_authorization_header(Some(auth_header_1.clone()))
+            .build();
+
+        let ctx3 = RequestContextBuilder::new(Protocol::Internal)
+            .with_cache_control(CacheControl::Cache(CacheKeyType::ClientSupplied))
+            .with_client_supplied_cache_key(Some("shared-key".to_string()))
+            .with_authorization_header(Some(auth_header_2))
+            .build();
+
+        let key1 = ctx1
+            .scoped_client_supplied_cache_key()
+            .expect("scoped key should be present");
+        let key2 = ctx2
+            .scoped_client_supplied_cache_key()
+            .expect("scoped key should be present");
+        let key3 = ctx3
+            .scoped_client_supplied_cache_key()
+            .expect("scoped key should be present");
+
+        assert!(key1.ends_with(":shared-key"));
+        assert!(key1.starts_with("auth:"));
+        assert_eq!(key1.len(), "auth:".len() + 32 + ":shared-key".len());
+        assert_eq!(key1, key2);
+        assert_ne!(key1, key3);
+        assert!(!key1.contains(&auth_header_1));
+    }
+
+    #[test]
+    fn test_scoped_client_supplied_cache_key_defaults_to_anonymous() {
+        let ctx = RequestContextBuilder::new(Protocol::Internal)
+            .with_cache_control(CacheControl::Cache(CacheKeyType::ClientSupplied))
+            .with_client_supplied_cache_key(Some("shared-key".to_string()))
+            .build();
+
+        assert_eq!(
+            ctx.scoped_client_supplied_cache_key().as_deref(),
+            Some("anonymous:shared-key")
+        );
+    }
 }