feat(rust): port BaseAWSLLM auth (credential resolution + SigV4) to litellm-core as a base provider (#33888)

* feat(rust): add feature-gated Bedrock AWS auth

Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>

* refactor(rust): move Bedrock auth into core

Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>

* fix(rust): fall through caller identity lookup errors

Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>

* test(rust): add live Bedrock proof and CI coverage

Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>

* refactor(rust): share in-memory cache with Bedrock auth

Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>

* fix(rust): preserve web identity credential expiry

Co-Authored-By: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>

---------

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: Ishaan Jaffer <155045088+ishaan-berri@users.noreply.github.com>

.github/workflows/test-rust.yml

@@ -61,5 +61,11 @@ jobs:
       - name: Run Clippy
         run: cargo clippy --workspace --all-targets --locked -- -D warnings
 
+      - name: Run Clippy with Bedrock auth
+        run: cargo clippy -p litellm-core --all-targets --features bedrock-auth --locked -- -D warnings
+
       - name: Run Rust tests
         run: cargo test --workspace --locked
+
+      - name: Run core tests with Bedrock auth
+        run: cargo test -p litellm-core --features bedrock-auth --locked

litellm-rust/crates/core/Cargo.toml

@@ -10,6 +10,25 @@ rand.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
+sha2.workspace = true
+aws-config = { version = "1.9.0", default-features = false, features = ["rustls", "rt-tokio"], optional = true }
+aws-credential-types = { version = "1.3.0", features = ["hardcoded-credentials"], optional = true }
+aws-sdk-sts = { version = "1.108.0", default-features = false, features = ["rustls", "rt-tokio"], optional = true }
+aws-sigv4 = { version = "1.5.1", optional = true }
+aws-types = { version = "1.4.0", optional = true }
+aws-smithy-runtime-api = { version = "1.13.0", optional = true }
+
+[features]
+default = []
+bedrock-auth = [
+    "dep:aws-config",
+    "dep:aws-credential-types",
+    "dep:aws-sdk-sts",
+    "dep:aws-sigv4",
+    "dep:aws-types",
+    "dep:aws-smithy-runtime-api",
+]
 
 [dev-dependencies]
+reqwest.workspace = true
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }

litellm-rust/crates/core/src/caching/in_memory_cache.rs

@@ -0,0 +1,258 @@
+use std::cmp::Reverse;
+use std::collections::{BinaryHeap, HashMap};
+use std::time::{Duration, SystemTime, UNIX_EPOCH};
+
+const DEFAULT_MAX_SIZE_IN_MEMORY: usize = 200;
+const DEFAULT_TTL: Duration = Duration::from_secs(600);
+
+pub struct InMemoryCache<V: Clone> {
+    pub cache_dict: HashMap<String, V>,
+    pub ttl_dict: HashMap<String, Duration>,
+    pub expiration_heap: BinaryHeap<Reverse<(Duration, String)>>,
+    pub max_size_in_memory: usize,
+    pub default_ttl: Duration,
+    now: Box<dyn Fn() -> Duration + Send + Sync>,
+}
+
+impl<V: Clone> Default for InMemoryCache<V> {
+    fn default() -> Self {
+        Self::new(None, None)
+    }
+}
+
+impl<V: Clone> InMemoryCache<V> {
+    pub fn new(max_size_in_memory: Option<usize>, default_ttl: Option<Duration>) -> Self {
+        Self::with_clock(max_size_in_memory, default_ttl, || {
+            SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap_or_default()
+        })
+    }
+
+    pub fn with_clock(
+        max_size_in_memory: Option<usize>,
+        default_ttl: Option<Duration>,
+        now: impl Fn() -> Duration + Send + Sync + 'static,
+    ) -> Self {
+        Self {
+            cache_dict: HashMap::new(),
+            ttl_dict: HashMap::new(),
+            expiration_heap: BinaryHeap::new(),
+            max_size_in_memory: max_size_in_memory.unwrap_or(DEFAULT_MAX_SIZE_IN_MEMORY),
+            default_ttl: default_ttl.unwrap_or(DEFAULT_TTL),
+            now: Box::new(now),
+        }
+    }
+
+    pub fn evict_cache(&mut self) {
+        if self.max_size_in_memory == 0 {
+            return;
+        }
+
+        let current_time = (self.now)();
+        while let Some(Reverse((expiration_time, key))) = self.expiration_heap.peek().cloned() {
+            if self.ttl_dict.get(&key).copied() != Some(expiration_time) {
+                self.expiration_heap.pop();
+            } else if expiration_time <= current_time {
+                self.expiration_heap.pop();
+                self.remove_key(&key);
+            } else {
+                break;
+            }
+        }
+
+        while self.cache_dict.len() >= self.max_size_in_memory {
+            let Some(Reverse((expiration_time, key))) = self.expiration_heap.pop() else {
+                break;
+            };
+            if self.ttl_dict.get(&key).copied() == Some(expiration_time) {
+                self.remove_key(&key);
+            }
+        }
+    }
+
+    pub fn allow_ttl_override(&self, key: &str) -> bool {
+        match self.ttl_dict.get(key).copied() {
+            None => true,
+            Some(expiration_time) => expiration_time < (self.now)(),
+        }
+    }
+
+    pub fn set_cache(&mut self, key: impl Into<String>, value: V, ttl: Option<Duration>) {
+        if self.max_size_in_memory == 0 {
+            return;
+        }
+
+        self.evict_cache();
+        let key = key.into();
+        self.cache_dict.insert(key.clone(), value);
+        if self.allow_ttl_override(&key) {
+            let expiration_time = (self.now)() + ttl.unwrap_or(self.default_ttl);
+            self.ttl_dict.insert(key.clone(), expiration_time);
+            self.expiration_heap.push(Reverse((expiration_time, key)));
+        }
+    }
+
+    // Generic values intentionally omit Python's per-item size check.
+    pub fn get_cache(&mut self, key: &str) -> Option<V> {
+        if self.cache_dict.contains_key(key) {
+            if self.is_key_expired(key) {
+                self.remove_key(key);
+                return None;
+            }
+            return self.cache_dict.get(key).cloned();
+        }
+        None
+    }
+
+    pub fn get_ttl(&self, key: &str) -> Option<Duration> {
+        self.ttl_dict.get(key).copied()
+    }
+
+    pub fn delete_cache(&mut self, key: &str) {
+        self.remove_key(key);
+    }
+
+    pub fn flush_cache(&mut self) {
+        self.cache_dict.clear();
+        self.ttl_dict.clear();
+        self.expiration_heap.clear();
+    }
+
+    fn is_key_expired(&self, key: &str) -> bool {
+        self.ttl_dict
+            .get(key)
+            .is_some_and(|expiration_time| *expiration_time < (self.now)())
+    }
+
+    fn remove_key(&mut self, key: &str) {
+        self.cache_dict.remove(key);
+        self.ttl_dict.remove(key);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    };
+
+    use super::InMemoryCache;
+    use std::time::Duration;
+
+    fn cache(now: Arc<AtomicU64>, max_size: usize, default_ttl: Duration) -> InMemoryCache<String> {
+        InMemoryCache::with_clock(Some(max_size), Some(default_ttl), move || {
+            Duration::from_secs(now.load(Ordering::Relaxed))
+        })
+    }
+
+    #[test]
+    fn ttl_expiry_is_deterministic() {
+        let now = Arc::new(AtomicU64::new(100));
+        let mut cache = cache(now.clone(), 10, Duration::from_secs(60));
+        cache.set_cache("key", "value".to_string(), None);
+        assert_eq!(cache.get_cache("key"), Some("value".to_string()));
+        now.store(161, Ordering::Relaxed);
+        assert_eq!(cache.get_cache("key"), None);
+        assert_eq!(cache.get_ttl("key"), None);
+    }
+
+    #[test]
+    fn default_and_per_set_ttl_are_applied() {
+        let now = Arc::new(AtomicU64::new(100));
+        let mut cache = cache(now.clone(), 10, Duration::from_secs(60));
+        cache.set_cache("default", "value".to_string(), None);
+        cache.set_cache("custom", "value".to_string(), Some(Duration::from_secs(20)));
+        assert_eq!(cache.get_ttl("default"), Some(Duration::from_secs(160)));
+        assert_eq!(cache.get_ttl("custom"), Some(Duration::from_secs(120)));
+    }
+
+    #[test]
+    fn unexpired_entries_do_not_allow_ttl_override() {
+        let now = Arc::new(AtomicU64::new(100));
+        let mut cache = cache(now.clone(), 10, Duration::from_secs(60));
+        cache.set_cache("key", "first".to_string(), Some(Duration::from_secs(20)));
+        cache.set_cache("key", "second".to_string(), Some(Duration::from_secs(80)));
+        assert_eq!(cache.get_cache("key"), Some("second".to_string()));
+        assert_eq!(cache.get_ttl("key"), Some(Duration::from_secs(120)));
+        now.store(121, Ordering::Relaxed);
+        cache.set_cache("key", "third".to_string(), Some(Duration::from_secs(80)));
+        assert_eq!(cache.get_ttl("key"), Some(Duration::from_secs(201)));
+    }
+
+    #[test]
+    fn max_size_evicts_earliest_expiration() {
+        let now = Arc::new(AtomicU64::new(100));
+        let mut cache = cache(now, 2, Duration::from_secs(60));
+        cache.set_cache("early", "value".to_string(), Some(Duration::from_secs(10)));
+        cache.set_cache("late", "value".to_string(), Some(Duration::from_secs(20)));
+        cache.set_cache("new", "value".to_string(), Some(Duration::from_secs(30)));
+        assert_eq!(cache.get_cache("early"), None);
+        assert!(cache.get_cache("late").is_some());
+        assert!(cache.get_cache("new").is_some());
+    }
+
+    #[test]
+    fn expired_entries_are_evicted_before_live_entries() {
+        let now = Arc::new(AtomicU64::new(100));
+        let mut cache = cache(now.clone(), 3, Duration::from_secs(60));
+        cache.set_cache(
+            "expired-one",
+            "value".to_string(),
+            Some(Duration::from_secs(10)),
+        );
+        cache.set_cache(
+            "expired-two",
+            "value".to_string(),
+            Some(Duration::from_secs(20)),
+        );
+        cache.set_cache("live", "value".to_string(), Some(Duration::from_secs(100)));
+        now.store(121, Ordering::Relaxed);
+        cache.set_cache("new", "value".to_string(), Some(Duration::from_secs(100)));
+        assert_eq!(cache.get_cache("expired-one"), None);
+        assert_eq!(cache.get_cache("expired-two"), None);
+        assert!(cache.get_cache("live").is_some());
+        assert!(cache.get_cache("new").is_some());
+    }
+
+    #[test]
+    fn stale_heap_entries_are_skipped() {
+        let now = Arc::new(AtomicU64::new(100));
+        let mut cache = cache(now, 1, Duration::from_secs(60));
+        cache.set_cache(
+            "removed",
+            "value".to_string(),
+            Some(Duration::from_secs(10)),
+        );
+        cache.delete_cache("removed");
+        cache.set_cache("kept", "value".to_string(), Some(Duration::from_secs(20)));
+        cache.set_cache("new", "value".to_string(), Some(Duration::from_secs(30)));
+        assert_eq!(cache.get_cache("removed"), None);
+        assert_eq!(cache.get_cache("kept"), None);
+        assert!(cache.get_cache("new").is_some());
+    }
+
+    #[test]
+    fn delete_and_flush_remove_values_and_ttls() {
+        let now = Arc::new(AtomicU64::new(100));
+        let mut cache = cache(now, 10, Duration::from_secs(60));
+        cache.set_cache("one", "value".to_string(), None);
+        cache.set_cache("two", "value".to_string(), None);
+        cache.delete_cache("one");
+        assert_eq!(cache.get_cache("one"), None);
+        cache.flush_cache();
+        assert!(cache.cache_dict.is_empty());
+        assert!(cache.ttl_dict.is_empty());
+        assert!(cache.expiration_heap.is_empty());
+    }
+
+    #[test]
+    fn zero_max_size_does_not_cache() {
+        let now = Arc::new(AtomicU64::new(100));
+        let mut cache = cache(now, 0, Duration::from_secs(60));
+        cache.set_cache("key", "value".to_string(), None);
+        assert_eq!(cache.get_cache("key"), None);
+        assert!(cache.cache_dict.is_empty());
+    }
+}

litellm-rust/crates/core/src/caching/mod.rs

@@ -0,0 +1 @@
+pub mod in_memory_cache;

litellm-rust/crates/core/src/lib.rs

@@ -1,3 +1,4 @@
+pub mod caching;
 pub mod call_lifecycle;
 pub mod constants;
 pub mod error;

litellm-rust/crates/core/src/providers/bedrock/aws_base.rs

@@ -0,0 +1,724 @@
+use std::collections::BTreeMap;
+use std::sync::{Mutex, OnceLock};
+use std::time::Duration;
+use std::time::{SystemTime, UNIX_EPOCH};
+
+use crate::caching::in_memory_cache::InMemoryCache;
+use crate::error::{CoreError, CoreResult};
+use aws_credential_types::provider::ProvideCredentials;
+use aws_credential_types::Credentials;
+use aws_sigv4::http_request::{
+    sign, SignableBody, SignableRequest, SigningParams, SigningSettings,
+};
+use aws_sigv4::sign::v4;
+use aws_smithy_runtime_api::client::identity::Identity;
+use sha2::{Digest, Sha256};
+
+use super::constants::{
+    AWS_ACCESS_KEY_ID, AWS_EXTERNAL_ID, AWS_PROFILE_NAME, AWS_REGION_NAME, AWS_ROLE_ARN,
+    AWS_ROLE_NAME, AWS_SECRET_ACCESS_KEY, AWS_SESSION_NAME, AWS_SESSION_TOKEN, AWS_STS_ENDPOINT,
+    AWS_WEB_IDENTITY_TOKEN, AWS_WEB_IDENTITY_TOKEN_FILE, BEDROCK_SERVICE,
+    DEFAULT_SESSION_NAME_PREFIX,
+};
+
+const STATIC_CREDENTIALS_TTL: Duration = Duration::from_secs(3600 - 60);
+const AMBIENT_CREDENTIALS_TTL: Duration = Duration::from_secs(600);
+
+static IAM_CREDENTIALS_CACHE: OnceLock<Mutex<InMemoryCache<Credentials>>> = OnceLock::new();
+
+fn credential_cache_ttl(flow: &AwsAuthFlow) -> Option<Duration> {
+    match flow {
+        AwsAuthFlow::StaticKeys { .. } => Some(STATIC_CREDENTIALS_TTL),
+        AwsAuthFlow::DefaultChain => Some(AMBIENT_CREDENTIALS_TTL),
+        AwsAuthFlow::WebIdentity { .. }
+        | AwsAuthFlow::AssumeRole { .. }
+        | AwsAuthFlow::Profile { .. }
+        | AwsAuthFlow::SessionToken { .. } => None,
+    }
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct AwsAuthConfig {
+    pub access_key_id: Option<String>,
+    pub secret_access_key: Option<String>,
+    pub session_token: Option<String>,
+    pub region_name: Option<String>,
+    pub session_name: Option<String>,
+    pub profile_name: Option<String>,
+    pub role_name: Option<String>,
+    pub web_identity_token: Option<String>,
+    pub sts_endpoint: Option<String>,
+    pub external_id: Option<String>,
+}
+
+impl AwsAuthConfig {
+    fn with_environment(self, env_lookup: &dyn Fn(&str) -> Option<String>) -> Self {
+        Self {
+            access_key_id: self.access_key_id.or_else(|| env_lookup(AWS_ACCESS_KEY_ID)),
+            secret_access_key: self
+                .secret_access_key
+                .or_else(|| env_lookup(AWS_SECRET_ACCESS_KEY)),
+            session_token: self.session_token.or_else(|| env_lookup(AWS_SESSION_TOKEN)),
+            region_name: self.region_name.or_else(|| env_lookup(AWS_REGION_NAME)),
+            session_name: self.session_name.or_else(|| env_lookup(AWS_SESSION_NAME)),
+            profile_name: self.profile_name.or_else(|| env_lookup(AWS_PROFILE_NAME)),
+            role_name: self.role_name.or_else(|| env_lookup(AWS_ROLE_NAME)),
+            web_identity_token: self
+                .web_identity_token
+                .or_else(|| env_lookup(AWS_WEB_IDENTITY_TOKEN)),
+            sts_endpoint: self.sts_endpoint.or_else(|| env_lookup(AWS_STS_ENDPOINT)),
+            external_id: self.external_id.or_else(|| env_lookup(AWS_EXTERNAL_ID)),
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum AwsAuthFlow {
+    WebIdentity {
+        token: String,
+        role: String,
+        session_name: String,
+    },
+    AssumeRole {
+        role: String,
+        session_name: Option<String>,
+    },
+    Profile {
+        name: String,
+    },
+    SessionToken {
+        access_key_id: String,
+        secret_access_key: String,
+        session_token: String,
+    },
+    StaticKeys {
+        access_key_id: String,
+        secret_access_key: String,
+        region_name: String,
+    },
+    DefaultChain,
+}
+
+fn cache_key(config: &AwsAuthConfig, flow: &AwsAuthFlow) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(format!("{config:?}:{flow:?}"));
+    format!("{:x}", hasher.finalize())
+}
+
+fn get_cached_credentials(key: &str) -> Option<Credentials> {
+    let cache = IAM_CREDENTIALS_CACHE.get_or_init(|| Mutex::new(InMemoryCache::default()));
+    let mut entries = cache.lock().ok()?;
+    entries.get_cache(key)
+}
+
+fn set_cached_credentials(key: String, credentials: Credentials, ttl: Duration) {
+    let cache = IAM_CREDENTIALS_CACHE.get_or_init(|| Mutex::new(InMemoryCache::default()));
+    if let Ok(mut entries) = cache.lock() {
+        entries.set_cache(key, credentials, Some(ttl));
+    }
+}
+
+fn role_identity(arn: &str) -> Option<(&str, &str, &str)> {
+    let mut parts = arn.splitn(6, ':');
+    let ("arn", partition, _, _, account, resource) = (
+        parts.next()?,
+        parts.next()?,
+        parts.next()?,
+        parts.next()?,
+        parts.next()?,
+        parts.next()?,
+    ) else {
+        return None;
+    };
+    let role = if let Some(role) = resource.strip_prefix("role/") {
+        role.rsplit('/').next()?
+    } else {
+        resource.strip_prefix("assumed-role/")?.split('/').next()?
+    };
+    Some((partition, account, role))
+}
+
+fn same_role_arns(target: &str, caller: &str) -> bool {
+    role_identity(target) == role_identity(caller)
+}
+
+pub fn classify_auth(
+    config: AwsAuthConfig,
+    env_lookup: &dyn Fn(&str) -> Option<String>,
+) -> AwsAuthFlow {
+    let config = config.with_environment(env_lookup);
+    if let (Some(token), Some(role), Some(session_name)) = (
+        config.web_identity_token.clone(),
+        config.role_name.clone(),
+        config.session_name.clone(),
+    ) {
+        return AwsAuthFlow::WebIdentity {
+            token,
+            role,
+            session_name,
+        };
+    }
+    if let Some(role) = config.role_name.clone() {
+        return AwsAuthFlow::AssumeRole {
+            role,
+            session_name: config.session_name.clone(),
+        };
+    }
+    if let Some(name) = config.profile_name {
+        return AwsAuthFlow::Profile { name };
+    }
+    if let (Some(access_key_id), Some(secret_access_key), Some(session_token)) = (
+        config.access_key_id.clone(),
+        config.secret_access_key.clone(),
+        config.session_token,
+    ) {
+        return AwsAuthFlow::SessionToken {
+            access_key_id,
+            secret_access_key,
+            session_token,
+        };
+    }
+    if let (Some(access_key_id), Some(secret_access_key), Some(region_name)) = (
+        config.access_key_id,
+        config.secret_access_key,
+        config.region_name,
+    ) {
+        return AwsAuthFlow::StaticKeys {
+            access_key_id,
+            secret_access_key,
+            region_name,
+        };
+    }
+    AwsAuthFlow::DefaultChain
+}
+
+pub async fn resolve_credentials(
+    config: AwsAuthConfig,
+    env_lookup: &dyn Fn(&str) -> Option<String>,
+) -> CoreResult<Credentials> {
+    let resolved = config.clone().with_environment(env_lookup);
+    let flow = classify_auth(config, env_lookup);
+    match flow {
+        AwsAuthFlow::SessionToken {
+            access_key_id,
+            secret_access_key,
+            session_token,
+        } => Ok(Credentials::new(
+            access_key_id,
+            secret_access_key,
+            Some(session_token),
+            None,
+            "litellm-static-session",
+        )),
+        AwsAuthFlow::StaticKeys {
+            access_key_id,
+            secret_access_key,
+            region_name,
+        } => {
+            let flow = AwsAuthFlow::StaticKeys {
+                access_key_id: access_key_id.clone(),
+                secret_access_key: secret_access_key.clone(),
+                region_name,
+            };
+            let key = cache_key(&resolved, &flow);
+            if let Some(credentials) = get_cached_credentials(&key) {
+                return Ok(credentials);
+            }
+            let credentials = Credentials::new(
+                access_key_id,
+                secret_access_key,
+                None,
+                None,
+                "litellm-static",
+            );
+            set_cached_credentials(
+                key,
+                credentials.clone(),
+                credential_cache_ttl(&flow).unwrap_or(STATIC_CREDENTIALS_TTL),
+            );
+            Ok(credentials)
+        }
+        AwsAuthFlow::Profile { name } => {
+            let provider = aws_config::profile::ProfileFileCredentialsProvider::builder()
+                .profile_name(name)
+                .build();
+            provider.provide_credentials().await.map_err(|error| {
+                CoreError::Auth(format!("AWS profile credentials failed: {error}"))
+            })
+        }
+        AwsAuthFlow::AssumeRole { role, session_name } => {
+            if is_already_running_as_role(&role, &resolved).await? {
+                let ambient_flow = AwsAuthFlow::DefaultChain;
+                let key = cache_key(&resolved, &ambient_flow);
+                if let Some(credentials) = get_cached_credentials(&key) {
+                    return Ok(credentials);
+                }
+                let provider =
+                    aws_config::default_provider::credentials::DefaultCredentialsChain::builder()
+                        .build()
+                        .await;
+                let credentials = provider.provide_credentials().await.map_err(|error| {
+                    CoreError::Auth(format!("AWS default credentials failed: {error}"))
+                })?;
+                set_cached_credentials(
+                    key,
+                    credentials.clone(),
+                    credential_cache_ttl(&ambient_flow).unwrap_or(AMBIENT_CREDENTIALS_TTL),
+                );
+                return Ok(credentials);
+            }
+            let mut loader = aws_config::defaults(aws_config::BehaviorVersion::latest());
+            if let Some(region) = resolved.region_name.clone() {
+                loader = loader.region(aws_types::region::Region::new(region));
+            }
+            if let Some(endpoint) = resolved.sts_endpoint.clone() {
+                loader = loader.endpoint_url(endpoint);
+            }
+            if let (Some(access_key_id), Some(secret_access_key)) =
+                (resolved.access_key_id, resolved.secret_access_key)
+            {
+                loader = loader.credentials_provider(Credentials::new(
+                    access_key_id,
+                    secret_access_key,
+                    resolved.session_token,
+                    None,
+                    "litellm-role-source",
+                ));
+            }
+            let sdk_config = loader.load().await;
+            let builder = aws_config::sts::AssumeRoleProvider::builder(role);
+            let builder = match session_name {
+                Some(name) => builder.session_name(name),
+                None => builder.session_name(default_session_name()),
+            };
+            let builder = match resolved.external_id {
+                Some(id) => builder.external_id(id),
+                None => builder,
+            };
+            let provider = builder.configure(&sdk_config).build().await;
+            provider
+                .provide_credentials()
+                .await
+                .map_err(|error| CoreError::Auth(format!("AWS role credentials failed: {error}")))
+        }
+        AwsAuthFlow::WebIdentity {
+            token,
+            role,
+            session_name,
+        } => {
+            let mut loader = aws_config::defaults(aws_config::BehaviorVersion::latest());
+            if let Some(region) = resolved.region_name {
+                loader = loader.region(aws_types::region::Region::new(region));
+            }
+            if let Some(endpoint) = resolved.sts_endpoint {
+                loader = loader.endpoint_url(endpoint);
+            }
+            let sdk_config = loader.load().await;
+            let client = aws_sdk_sts::Client::new(&sdk_config);
+            let response = client
+                .assume_role_with_web_identity()
+                .role_arn(role)
+                .role_session_name(session_name)
+                .web_identity_token(token)
+                .send()
+                .await
+                .map_err(|error| {
+                    CoreError::Auth(format!("AWS web identity credentials failed: {error}"))
+                })?;
+            let credentials = response.credentials().ok_or_else(|| {
+                CoreError::Auth("AWS web identity response had no credentials".to_string())
+            })?;
+            let expiration = SystemTime::try_from(*credentials.expiration()).map_err(|error| {
+                CoreError::Auth(format!("AWS web identity expiration was invalid: {error}"))
+            })?;
+            Ok(Credentials::new(
+                credentials.access_key_id(),
+                credentials.secret_access_key(),
+                Some(credentials.session_token().to_string()),
+                Some(expiration),
+                "litellm-web-identity",
+            ))
+        }
+        AwsAuthFlow::DefaultChain => {
+            let key = cache_key(&resolved, &AwsAuthFlow::DefaultChain);
+            if let Some(credentials) = get_cached_credentials(&key) {
+                return Ok(credentials);
+            }
+            let provider =
+                aws_config::default_provider::credentials::DefaultCredentialsChain::builder()
+                    .build()
+                    .await;
+            let credentials = provider.provide_credentials().await.map_err(|error| {
+                CoreError::Auth(format!("AWS default credentials failed: {error}"))
+            })?;
+            set_cached_credentials(
+                key,
+                credentials.clone(),
+                credential_cache_ttl(&AwsAuthFlow::DefaultChain).unwrap_or(AMBIENT_CREDENTIALS_TTL),
+            );
+            Ok(credentials)
+        }
+    }
+}
+
+async fn is_already_running_as_role(role: &str, config: &AwsAuthConfig) -> CoreResult<bool> {
+    if role_identity(role).is_none() {
+        return Ok(false);
+    }
+    if let (Ok(current_role), Ok(token_file)) = (
+        std::env::var(AWS_ROLE_ARN),
+        std::env::var(AWS_WEB_IDENTITY_TOKEN_FILE),
+    ) {
+        if !token_file.is_empty() {
+            return Ok(same_role_arns(role, &current_role));
+        }
+    }
+    let mut loader = aws_config::defaults(aws_config::BehaviorVersion::latest());
+    if let Some(region) = config.region_name.clone() {
+        loader = loader.region(aws_types::region::Region::new(region));
+    }
+    if let Some(endpoint) = config.sts_endpoint.clone() {
+        loader = loader.endpoint_url(endpoint);
+    }
+    let sdk_config = loader.load().await;
+    let response = match aws_sdk_sts::Client::new(&sdk_config)
+        .get_caller_identity()
+        .send()
+        .await
+    {
+        Ok(response) => response,
+        Err(_) => return Ok(false),
+    };
+    Ok(response
+        .arn()
+        .is_some_and(|caller| same_role_arns(role, caller)))
+}
+
+fn default_session_name() -> String {
+    let seconds = SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map_or(0, |duration| duration.as_secs());
+    format!("{DEFAULT_SESSION_NAME_PREFIX}-{seconds}")
+}
+
+pub fn sign_bedrock_post(
+    url: &str,
+    body: &[u8],
+    headers: &BTreeMap<String, String>,
+    region: &str,
+    credentials: &Credentials,
+    signing_time: SystemTime,
+) -> CoreResult<BTreeMap<String, String>> {
+    let identity: Identity = credentials.clone().into();
+    let params = v4::SigningParams::builder()
+        .identity(&identity)
+        .region(region)
+        .name(BEDROCK_SERVICE)
+        .time(signing_time)
+        .settings(SigningSettings::default())
+        .build()
+        .map(SigningParams::from)
+        .map_err(|error| CoreError::Auth(format!("AWS signing parameters failed: {error}")))?;
+    let header_refs = headers
+        .iter()
+        .map(|(name, value)| (name.as_str(), value.as_str()));
+    let request = SignableRequest::new("POST", url, header_refs, SignableBody::Bytes(body))
+        .map_err(|error| CoreError::Auth(format!("AWS signable request failed: {error}")))?;
+    let (instructions, _) = sign(request, &params)
+        .map_err(|error| CoreError::Auth(format!("AWS request signing failed: {error}")))?
+        .into_parts();
+    Ok(instructions
+        .headers()
+        .map(|(name, value)| {
+            let normalized_name = match name {
+                "authorization" => "Authorization",
+                "x-amz-date" => "X-Amz-Date",
+                "x-amz-security-token" => "X-Amz-Security-Token",
+                _ => name,
+            };
+            (normalized_name.to_string(), value.to_string())
+        })
+        .collect())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn no_env(_: &str) -> Option<String> {
+        None
+    }
+
+    fn parity_inputs() -> (String, Vec<u8>, BTreeMap<String, String>) {
+        (
+            "https://bedrock-runtime.us-east-1.amazonaws.com/model/amazon.titan-text-express-v1/invoke"
+                .to_string(),
+            br#"{"input":"hello"}"#.to_vec(),
+            BTreeMap::from([("Content-Type".to_string(), "application/json".to_string())]),
+        )
+    }
+
+    #[test]
+    fn classification_preserves_python_precedence() {
+        let config = AwsAuthConfig {
+            access_key_id: Some("ak".into()),
+            secret_access_key: Some("sk".into()),
+            session_token: Some("token".into()),
+            region_name: Some("us-east-1".into()),
+            session_name: Some("session".into()),
+            profile_name: Some("profile".into()),
+            role_name: Some("role".into()),
+            web_identity_token: Some("oidc".into()),
+            ..Default::default()
+        };
+        assert!(matches!(
+            classify_auth(config, &no_env),
+            AwsAuthFlow::WebIdentity { .. }
+        ));
+    }
+
+    #[test]
+    fn classification_covers_fallthroughs() {
+        let env = |key: &str| match key {
+            AWS_PROFILE_NAME => Some("profile".into()),
+            _ => None,
+        };
+        assert!(matches!(
+            classify_auth(AwsAuthConfig::default(), &env),
+            AwsAuthFlow::Profile { .. }
+        ));
+        assert!(matches!(
+            classify_auth(
+                AwsAuthConfig {
+                    access_key_id: Some("ak".into()),
+                    secret_access_key: Some("sk".into()),
+                    session_token: Some("token".into()),
+                    ..Default::default()
+                },
+                &no_env
+            ),
+            AwsAuthFlow::SessionToken { .. }
+        ));
+        assert!(matches!(
+            classify_auth(
+                AwsAuthConfig {
+                    access_key_id: Some("ak".into()),
+                    secret_access_key: Some("sk".into()),
+                    region_name: Some("us-east-1".into()),
+                    ..Default::default()
+                },
+                &no_env
+            ),
+            AwsAuthFlow::StaticKeys { .. }
+        ));
+        assert_eq!(
+            classify_auth(AwsAuthConfig::default(), &no_env),
+            AwsAuthFlow::DefaultChain
+        );
+    }
+
+    #[tokio::test]
+    async fn static_credentials_do_not_use_network() {
+        let credentials = resolve_credentials(
+            AwsAuthConfig {
+                access_key_id: Some("ak".into()),
+                secret_access_key: Some("sk".into()),
+                region_name: Some("us-east-1".into()),
+                ..Default::default()
+            },
+            &no_env,
+        )
+        .await
+        .expect("static credentials");
+        assert_eq!(credentials.access_key_id(), "ak");
+        assert_eq!(credentials.session_token(), None);
+    }
+
+    #[test]
+    fn cache_policy_matches_python_flows() {
+        assert_eq!(
+            credential_cache_ttl(&AwsAuthFlow::StaticKeys {
+                access_key_id: "ak".into(),
+                secret_access_key: "sk".into(),
+                region_name: "us-east-1".into(),
+            }),
+            Some(STATIC_CREDENTIALS_TTL)
+        );
+        assert_eq!(
+            credential_cache_ttl(&AwsAuthFlow::DefaultChain),
+            Some(AMBIENT_CREDENTIALS_TTL)
+        );
+        assert_eq!(
+            credential_cache_ttl(&AwsAuthFlow::SessionToken {
+                access_key_id: "ak".into(),
+                secret_access_key: "sk".into(),
+                session_token: "token".into(),
+            }),
+            None
+        );
+        assert_eq!(
+            credential_cache_ttl(&AwsAuthFlow::Profile {
+                name: "profile".into()
+            }),
+            None
+        );
+        assert_eq!(
+            credential_cache_ttl(&AwsAuthFlow::AssumeRole {
+                role: "arn:aws:iam::123456789012:role/demo".into(),
+                session_name: None,
+            }),
+            None
+        );
+        assert_eq!(
+            credential_cache_ttl(&AwsAuthFlow::WebIdentity {
+                token: "token".into(),
+                role: "arn:aws:iam::123456789012:role/demo".into(),
+                session_name: "session".into(),
+            }),
+            None
+        );
+    }
+
+    #[test]
+    fn cache_round_trip_preserves_credentials() {
+        let key = format!("cache-test-{}", std::process::id());
+        let credentials = Credentials::new("cache-ak", "cache-sk", None, None, "test");
+        set_cached_credentials(key.clone(), credentials.clone(), STATIC_CREDENTIALS_TTL);
+        assert_eq!(
+            get_cached_credentials(&key).map(|value| value.access_key_id().to_string()),
+            Some("cache-ak".to_string())
+        );
+    }
+
+    #[test]
+    fn same_role_comparison_matches_partition_account_and_role() {
+        assert!(same_role_arns(
+            "arn:aws:iam::123456789012:role/path/demo",
+            "arn:aws:sts::123456789012:assumed-role/demo/session"
+        ));
+        assert!(!same_role_arns(
+            "arn:aws:iam::123456789012:role/demo",
+            "arn:aws:iam::999999999999:role/demo"
+        ));
+        assert!(!same_role_arns(
+            "arn:aws:iam::123456789012:role/demo",
+            "arn:aws-cn:iam::123456789012:role/demo"
+        ));
+        assert!(!same_role_arns(
+            "arn:aws:iam::123456789012:user/demo",
+            "arn:aws:iam::123456789012:role/demo"
+        ));
+    }
+
+    #[test]
+    fn signing_matches_botocore_golden_vector() {
+        let (url, body, headers) = parity_inputs();
+        let credentials = Credentials::new(
+            "AKIDEXAMPLE",
+            "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
+            Some("session-token".to_string()),
+            None,
+            "test",
+        );
+        let signed = sign_bedrock_post(
+            &url,
+            &body,
+            &headers,
+            "us-east-1",
+            &credentials,
+            UNIX_EPOCH + std::time::Duration::from_secs(1_704_164_645),
+        )
+        .expect("golden signature");
+        assert_eq!(
+            signed.get("X-Amz-Date").map(String::as_str),
+            Some("20240102T030405Z")
+        );
+        assert_eq!(
+            signed.get("X-Amz-Security-Token").map(String::as_str),
+            Some("session-token")
+        );
+        assert_eq!(
+            signed.get("Authorization").map(String::as_str),
+            Some("AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20240102/us-east-1/bedrock/aws4_request, SignedHeaders=content-type;host;x-amz-date;x-amz-security-token, Signature=55c027ef47527d3ad63f1735f9d099efdbc99f296ff914bd94e727e24ec0e464")
+        );
+    }
+
+    #[test]
+    fn signing_without_session_token_omits_security_header() {
+        let (url, body, headers) = parity_inputs();
+        let credentials = Credentials::new(
+            "AKIDEXAMPLE",
+            "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
+            None,
+            None,
+            "test",
+        );
+        let signed = sign_bedrock_post(
+            &url,
+            &body,
+            &headers,
+            "us-east-1",
+            &credentials,
+            UNIX_EPOCH + std::time::Duration::from_secs(1_704_164_645),
+        )
+        .expect("signature");
+        assert!(!signed.contains_key("X-Amz-Security-Token"));
+    }
+
+    #[ignore]
+    #[tokio::test]
+    async fn live_bedrock_invoke_model_returns_200() -> Result<(), Box<dyn std::error::Error>> {
+        let access_key_id = std::env::var("AWS_BEDROCK_TEST_ACCESS_KEY_ID")?;
+        let secret_access_key = std::env::var("AWS_BEDROCK_TEST_SECRET_ACCESS_KEY")?;
+        let body = br#"{"anthropic_version":"bedrock-2023-05-31","max_tokens":1,"messages":[{"role":"user","content":[{"type":"text","text":"ping"}]}]}"#.to_vec();
+        let headers =
+            BTreeMap::from([("Content-Type".to_string(), "application/json".to_string())]);
+        let credentials = resolve_credentials(
+            AwsAuthConfig {
+                access_key_id: Some(access_key_id),
+                secret_access_key: Some(secret_access_key),
+                region_name: Some("us-west-2".to_string()),
+                ..Default::default()
+            },
+            &no_env,
+        )
+        .await?;
+        let client = reqwest::Client::new();
+        let mut failures = Vec::new();
+
+        for region in ["us-west-2", "us-east-1"] {
+            let url = format!(
+                "https://bedrock-runtime.{region}.amazonaws.com/model/us.anthropic.claude-opus-4-8/invoke"
+            );
+            let signed_headers = sign_bedrock_post(
+                &url,
+                &body,
+                &headers,
+                region,
+                &credentials,
+                SystemTime::now(),
+            )?;
+            let mut request = client.post(&url).body(body.clone());
+            for (name, value) in &headers {
+                request = request.header(name, value);
+            }
+            for (name, value) in signed_headers {
+                request = request.header(name, value);
+            }
+            let response = request.send().await?;
+            let status = response.status();
+            let response_body = response.text().await?;
+            let snippet: String = response_body.chars().take(240).collect();
+            println!("region={region} status={status} response={snippet}");
+            if status == reqwest::StatusCode::OK {
+                return Ok(());
+            }
+            failures.push(format!("{region}: {status} {snippet}"));
+        }
+
+        panic!(
+            "no Bedrock region returned HTTP 200: {}",
+            failures.join("; ")
+        );
+    }
+}

litellm-rust/crates/core/src/providers/bedrock/constants.rs

@@ -0,0 +1,14 @@
+pub const AWS_ACCESS_KEY_ID: &str = "AWS_ACCESS_KEY_ID";
+pub const AWS_SECRET_ACCESS_KEY: &str = "AWS_SECRET_ACCESS_KEY";
+pub const AWS_SESSION_TOKEN: &str = "AWS_SESSION_TOKEN";
+pub const AWS_REGION_NAME: &str = "AWS_REGION_NAME";
+pub const AWS_SESSION_NAME: &str = "AWS_SESSION_NAME";
+pub const AWS_PROFILE_NAME: &str = "AWS_PROFILE_NAME";
+pub const AWS_ROLE_NAME: &str = "AWS_ROLE_NAME";
+pub const AWS_WEB_IDENTITY_TOKEN: &str = "AWS_WEB_IDENTITY_TOKEN";
+pub const AWS_ROLE_ARN: &str = "AWS_ROLE_ARN";
+pub const AWS_WEB_IDENTITY_TOKEN_FILE: &str = "AWS_WEB_IDENTITY_TOKEN_FILE";
+pub const AWS_STS_ENDPOINT: &str = "AWS_STS_ENDPOINT";
+pub const AWS_EXTERNAL_ID: &str = "AWS_EXTERNAL_ID";
+pub const BEDROCK_SERVICE: &str = "bedrock";
+pub const DEFAULT_SESSION_NAME_PREFIX: &str = "litellm-session";

litellm-rust/crates/core/src/providers/bedrock/mod.rs

@@ -0,0 +1,6 @@
+//! User-directed exception: this base provider owns AWS auth I/O for parity
+//! with Python's `BaseAWSLLM`; the broader core purity guidance is reconciled
+//! separately.
+
+pub mod aws_base;
+mod constants;

litellm-rust/crates/core/src/providers/mod.rs

@@ -1,5 +1,7 @@
 pub mod anthropic;
 pub mod azure_ai;
+#[cfg(feature = "bedrock-auth")]
+pub mod bedrock;
 pub mod mistral;
 pub mod openai;
 pub mod vertex_ai;

tests/test_litellm/llms/bedrock/test_base_aws_llm.py

@@ -15,6 +15,7 @@ from typing import Any, Dict, Optional
 from unittest.mock import MagicMock, patch
 
 from botocore.awsrequest import AWSPreparedRequest, AWSRequest
+from botocore.auth import SigV4Auth
 from botocore.credentials import Credentials
 
 import litellm
@@ -768,6 +769,30 @@ def test_get_request_headers_with_sigv4():
         assert result == mock_request.prepare.return_value
 
 
+def test_sigv4_matches_rust_golden_vector():
+    request = AWSRequest(
+        method="POST",
+        url="https://bedrock-runtime.us-east-1.amazonaws.com/model/amazon.titan-text-express-v1/invoke",
+        data=b'{"input":"hello"}',
+        headers={"Content-Type": "application/json"},
+    )
+    credentials = Credentials(
+        "AKIDEXAMPLE",
+        "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY",
+        "session-token",
+    )
+    with patch("botocore.auth.get_current_datetime", return_value=datetime(2024, 1, 2, 3, 4, 5)):
+        SigV4Auth(credentials, "bedrock", "us-east-1").add_auth(request)
+    assert request.headers["X-Amz-Date"] == "20240102T030405Z"
+    assert request.headers["X-Amz-Security-Token"] == "session-token"
+    assert (
+        request.headers["Authorization"]
+        == "AWS4-HMAC-SHA256 Credential=AKIDEXAMPLE/20240102/us-east-1/bedrock/aws4_request, "
+        "SignedHeaders=content-type;host;x-amz-date;x-amz-security-token, "
+        "Signature=55c027ef47527d3ad63f1735f9d099efdbc99f296ff914bd94e727e24ec0e464"
+    )
+
+
 def test_get_request_headers_with_api_key_bearer_token():
     """
     Test that get_request_headers uses the api_key parameter as a bearer token when provided