feat(rust): serve every gateway HTTP setting natively instead of declining to Python

litellm-http now builds the rustls config itself, so one route-neutral place covers roots, the client certificate, ALPN, ssl_ecdh_curve and ssl_security_level. A curve picks the single key exchange group. A cipher string restricts the TLS 1.2 suites it names, and entries rustls cannot express, such as @SECLEVEL=1, are logged once and skipped.

user_url_validation and user_url_allowed_hosts are applied by the media fetcher. Document downloads honor the environment proxy whenever provider calls do, keeping the per-hop address check, and stay on the pinned resolver when no proxy applies.

AIOHTTP_SO_KEEPALIVE, AIOHTTP_TCP_KEEPIDLE, AIOHTTP_TCP_KEEPINTVL, AIOHTTP_TCP_KEEPCNT and AIOHTTP_KEEPALIVE_TIMEOUT map onto the client. A client= argument and a live SSLContext are ignored
This commit is contained in:
Yujong Lee 2026-09-18 19:39:07 -07:00
parent fb41bc3ed6
commit 51010ea486
18 changed files with 944 additions and 235 deletions

View file

@ -2135,10 +2135,14 @@ dependencies = [
name = "litellm-http"
version = "0.1.0"
dependencies = [
"http 1.4.2",
"hyper-util",
"reqwest 0.12.28",
"rstest",
"rustls 0.23.42",
"thiserror 2.0.19",
"tokio",
"webpki-roots",
]
[[package]]

View file

@ -27,6 +27,8 @@ litellm-token-counter = { path = "crates/token-counter" }
litellm-host-python = { path = "crates/host-python" }
bytes = "1"
http = "1"
hyper-util = { version = "0.1.20", default-features = false, features = ["client-proxy"] }
proptest = "1.7.0"
pyo3 = "0.29.2"
pyo3-async-runtimes = { version = "0.29.0", features = ["tokio-runtime"] }
@ -50,6 +52,7 @@ base64 = "0.22"
moka = { version = "0.12.16", features = ["future"] }
strum = { version = "0.28.0", features = ["derive"] }
url = "2.5.8"
webpki-roots = "1"
time = { version = "0.3.53", features = ["parsing"] }
criterion = "0.8.2"
fancy-regex = "0.19.2"

View file

@ -12,7 +12,10 @@ use litellm_llms::{
error::Error as OcrError,
transformation::{LiteLLMOcrResponse, OCR_RESPONSE_MAX_BYTES, OcrTransportConfig},
},
custom_httpx::{llm_http_handler::OcrClient, media::PublicDnsResolver},
custom_httpx::{
llm_http_handler::OcrClient,
media::{PublicDnsResolver, UrlPolicy},
},
};
use rstest::rstest;
use serde_json::{Value, json};
@ -181,7 +184,8 @@ async fn ocr_client_uses_the_injected_http_pool_configuration() {
};
let client = OcrClient::new(
&HttpClientPool::new(Arc::new(PublicDnsResolver)),
&HttpClientConfig::resolve(&settings).unwrap(),
&HttpClientConfig::resolve(&settings).config,
UrlPolicy::default(),
VertexAuth::default(),
)
.unwrap();

View file

@ -6,8 +6,12 @@ license.workspace = true
repository.workspace = true
[dependencies]
http.workspace = true
hyper-util.workspace = true
reqwest.workspace = true
rustls.workspace = true
thiserror.workspace = true
webpki-roots.workspace = true
[dev-dependencies]
rstest.workspace = true

View file

@ -1,12 +1,13 @@
use std::{
net::{IpAddr, Ipv4Addr},
path::{Path, PathBuf},
path::PathBuf,
time::Duration,
};
use crate::{
error::Error,
settings::{HttpSettings, SslVerify},
settings::{HttpSettings, SslVerify, TcpKeepalive},
tls::{self, KeyExchangeGroup, Tls12CipherSuite, Unsupported},
};
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
@ -20,27 +21,38 @@ pub enum Verify {
pub struct HttpClientConfig {
pub verify: Verify,
pub client_certificate: Option<PathBuf>,
pub key_exchange_group: Option<KeyExchangeGroup>,
pub tls12_cipher_suites: Option<Vec<Tls12CipherSuite>>,
pub force_ipv4: bool,
pub http2: bool,
pub user_agent: Option<String>,
pub trust_proxy_env: bool,
pub connect_timeout: Duration,
pub tcp_keepalive: Option<TcpKeepalive>,
pub pool_idle_timeout: Duration,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Resolution {
pub config: HttpClientConfig,
pub unsupported: Vec<Unsupported>,
}
impl HttpClientConfig {
pub fn resolve(settings: &HttpSettings) -> Result<Self, Error> {
if let Some(level) = &settings.ssl_security_level {
return Err(Error::Unsupported {
setting: "ssl_security_level",
reason: format!("OpenSSL cipher string {level:?} has no rustls equivalent"),
});
}
if let Some(curve) = &settings.ssl_ecdh_curve {
return Err(Error::Unsupported {
setting: "ssl_ecdh_curve",
reason: format!("key exchange group {curve:?} is fixed by the rustls provider"),
});
}
pub fn resolve(settings: &HttpSettings) -> Resolution {
let (key_exchange_group, unsupported_curve) = match settings
.ssl_ecdh_curve
.as_deref()
.map(KeyExchangeGroup::from_openssl_name)
{
None => (None, None),
Some(Ok(group)) => (Some(group), None),
Some(Err(unsupported)) => (None, Some(unsupported)),
};
let ciphers = settings
.ssl_security_level
.as_deref()
.map(tls::parse_cipher_string);
let verify = match &settings.ssl_verify {
Some(SslVerify::Disabled) => Verify::Disabled,
Some(SslVerify::CaBundle(path)) => Verify::CaBundle(path.clone()),
@ -49,62 +61,50 @@ impl HttpClientConfig {
.clone()
.map_or(Verify::BuiltInRoots, Verify::CaBundle),
};
Ok(Self {
verify,
client_certificate: settings.ssl_certificate.clone(),
force_ipv4: settings.force_ipv4,
http2: settings.http2,
user_agent: settings.user_agent.clone(),
trust_proxy_env: !settings.ignore_proxy_env
|| settings.trust_proxy_env
|| settings.http2
|| settings.httpx_transport,
connect_timeout: settings.connect_timeout,
})
let (tls12_cipher_suites, unsupported_ciphers) = ciphers
.map_or((None, Vec::new()), |ciphers| {
(ciphers.tls12_cipher_suites, ciphers.unsupported)
});
Resolution {
config: Self {
verify,
client_certificate: settings.ssl_certificate.clone(),
key_exchange_group,
tls12_cipher_suites,
force_ipv4: settings.force_ipv4,
http2: settings.http2,
user_agent: settings.user_agent.clone(),
trust_proxy_env: !settings.ignore_proxy_env
|| settings.trust_proxy_env
|| settings.http2
|| settings.httpx_transport,
connect_timeout: settings.connect_timeout,
tcp_keepalive: settings.tcp_keepalive,
pool_idle_timeout: settings.pool_idle_timeout,
},
unsupported: unsupported_curve
.into_iter()
.chain(unsupported_ciphers)
.collect(),
}
}
pub fn client_builder(&self) -> Result<reqwest::ClientBuilder, Error> {
let base = reqwest::Client::builder().connect_timeout(self.connect_timeout);
let with_roots = match &self.verify {
Verify::Disabled => base.danger_accept_invalid_certs(true),
Verify::BuiltInRoots => base,
Verify::CaBundle(path) => {
let pem = read(path)?;
let certificates =
reqwest::Certificate::from_pem_bundle(&pem).map_err(|error| {
Error::InvalidPem {
path: path.clone(),
message: error.without_url().to_string(),
}
})?;
if certificates.is_empty() {
return Err(Error::InvalidPem {
path: path.clone(),
message: "no certificates found".into(),
});
}
certificates.into_iter().fold(
base.tls_built_in_root_certs(false),
|builder, certificate| builder.add_root_certificate(certificate),
)
}
};
let with_identity = match &self.client_certificate {
None => with_roots,
Some(path) => {
let identity = reqwest::Identity::from_pem(&read(path)?).map_err(|error| {
Error::InvalidPem {
path: path.clone(),
message: error.without_url().to_string(),
}
})?;
with_roots.identity(identity)
}
let base = reqwest::Client::builder()
.use_preconfigured_tls(tls::client_config(self)?)
.connect_timeout(self.connect_timeout)
.pool_idle_timeout(self.pool_idle_timeout);
let with_keepalive = match self.tcp_keepalive {
None => base,
Some(keepalive) => base
.tcp_keepalive(keepalive.idle)
.tcp_keepalive_interval(keepalive.interval)
.tcp_keepalive_retries(keepalive.retries),
};
let with_address = if self.force_ipv4 {
with_identity.local_address(IpAddr::V4(Ipv4Addr::UNSPECIFIED))
with_keepalive.local_address(IpAddr::V4(Ipv4Addr::UNSPECIFIED))
} else {
with_identity
with_keepalive
};
let with_protocol = if self.http2 {
with_address
@ -123,13 +123,6 @@ impl HttpClientConfig {
}
}
fn read(path: &Path) -> Result<Vec<u8>, Error> {
std::fs::read(path).map_err(|error| Error::Read {
path: path.to_path_buf(),
message: error.to_string(),
})
}
#[cfg(test)]
mod tests {
use rstest::rstest;
@ -168,7 +161,7 @@ mod tests {
#[case] settings: HttpSettings,
#[case] expected: Verify,
) {
let config = HttpClientConfig::resolve(&settings).unwrap();
let config = HttpClientConfig::resolve(&settings).config;
assert_eq!(config.verify, expected);
}
@ -179,42 +172,88 @@ mod tests {
..HttpSettings::default()
}
.with_environment(&|name: &str| (name == "SSL_VERIFY").then(|| "true".to_string()));
let config = HttpClientConfig::resolve(&settings).unwrap();
let config = HttpClientConfig::resolve(&settings).config;
assert_eq!(config.verify, Verify::BuiltInRoots);
}
#[rstest]
#[case::x25519("X25519", Some(KeyExchangeGroup::X25519))]
#[case::openssl_p256("prime256v1", Some(KeyExchangeGroup::Secp256r1))]
#[case::p384("secp384r1", Some(KeyExchangeGroup::Secp384r1))]
fn ecdh_curve_selects_the_single_key_exchange_group(
#[case] curve: &str,
#[case] expected: Option<KeyExchangeGroup>,
) {
let settings = HttpSettings {
ssl_ecdh_curve: Some(curve.into()),
..HttpSettings::default()
};
let resolution = HttpClientConfig::resolve(&settings);
assert_eq!(resolution.config.key_exchange_group, expected);
assert_eq!(resolution.unsupported, []);
}
#[test]
fn cipher_strings_are_rejected_rather_than_ignored() {
fn unsupported_ecdh_curve_keeps_the_defaults_and_is_reported() {
let settings = HttpSettings {
ssl_ecdh_curve: Some("secp521r1".into()),
..HttpSettings::default()
};
let resolution = HttpClientConfig::resolve(&settings);
assert_eq!(resolution.config.key_exchange_group, None);
assert_eq!(
resolution.unsupported,
[Unsupported::EcdhCurve("secp521r1".into())]
);
}
#[test]
fn legacy_security_level_keeps_every_suite_and_is_reported_unsupported() {
let settings = HttpSettings {
ssl_security_level: Some("DEFAULT@SECLEVEL=1".into()),
..HttpSettings::default()
};
assert!(matches!(
HttpClientConfig::resolve(&settings),
Err(Error::Unsupported {
setting: "ssl_security_level",
..
})
));
let resolution = HttpClientConfig::resolve(&settings);
assert_eq!(resolution.config.tls12_cipher_suites, None);
assert_eq!(
resolution.unsupported,
[Unsupported::SecurityLevel("@SECLEVEL=1".into())]
);
}
#[test]
fn ecdh_curves_are_rejected_rather_than_ignored() {
fn named_suites_restrict_tls12_and_unsupported_entries_are_reported() {
let settings = HttpSettings {
ssl_ecdh_curve: Some("X25519".into()),
ssl_security_level: Some(
"ECDHE-RSA-AES256-GCM-SHA384:ECDHE-ECDSA-AES128-GCM-SHA256:!aNULL:AES256-SHA@SECLEVEL=2"
.into(),
),
..HttpSettings::default()
};
assert!(matches!(
HttpClientConfig::resolve(&settings),
Err(Error::Unsupported {
setting: "ssl_ecdh_curve",
..
})
));
let resolution = HttpClientConfig::resolve(&settings);
assert_eq!(
resolution.config.tls12_cipher_suites,
Some(vec![
Tls12CipherSuite::EcdheEcdsaAes128Gcm,
Tls12CipherSuite::EcdheRsaAes256Gcm
])
);
assert_eq!(
resolution.unsupported,
[
Unsupported::CipherToken("!aNULL".into()),
Unsupported::CipherToken("AES256-SHA".into())
]
);
}
#[test]
fn connection_settings_carry_over_unchanged() {
let keepalive = TcpKeepalive {
idle: Duration::from_secs(60),
interval: Duration::from_secs(30),
retries: 5,
};
let settings = HttpSettings {
ssl_certificate: Some("/client.pem".into()),
force_ipv4: true,
@ -222,19 +261,25 @@ mod tests {
user_agent: Some("litellm/1.0".into()),
trust_proxy_env: true,
connect_timeout: Duration::from_secs(7),
tcp_keepalive: Some(keepalive),
pool_idle_timeout: Duration::from_secs(45),
..HttpSettings::default()
};
let config = HttpClientConfig::resolve(&settings).unwrap();
let config = HttpClientConfig::resolve(&settings).config;
assert_eq!(
config,
HttpClientConfig {
verify: Verify::BuiltInRoots,
client_certificate: Some("/client.pem".into()),
key_exchange_group: None,
tls12_cipher_suites: None,
force_ipv4: true,
http2: true,
user_agent: Some("litellm/1.0".into()),
trust_proxy_env: true,
connect_timeout: Duration::from_secs(7),
tcp_keepalive: Some(keepalive),
pool_idle_timeout: Duration::from_secs(45),
}
);
}
@ -258,7 +303,7 @@ mod tests {
#[case] settings: HttpSettings,
#[case] expected: bool,
) {
let config = HttpClientConfig::resolve(&settings).unwrap();
let config = HttpClientConfig::resolve(&settings).config;
assert_eq!(config.trust_proxy_env, expected);
}
@ -267,7 +312,7 @@ mod tests {
let path = std::env::temp_dir().join("litellm-http-missing-bundle.pem");
let config = HttpClientConfig {
verify: Verify::CaBundle(path.clone()),
..HttpClientConfig::resolve(&HttpSettings::default()).unwrap()
..HttpClientConfig::resolve(&HttpSettings::default()).config
};
assert!(matches!(
config.client_builder(),
@ -282,7 +327,7 @@ mod tests {
std::fs::write(&path, b"not a certificate").unwrap();
let config = HttpClientConfig {
verify: Verify::CaBundle(path.clone()),
..HttpClientConfig::resolve(&HttpSettings::default()).unwrap()
..HttpClientConfig::resolve(&HttpSettings::default()).config
};
let result = config.client_builder().map(drop);
std::fs::remove_file(&path).unwrap();

View file

@ -2,11 +2,6 @@ use std::path::PathBuf;
#[derive(Clone, Debug, thiserror::Error, PartialEq, Eq)]
pub enum Error {
#[error("{setting} cannot be expressed with rustls: {reason}")]
Unsupported {
setting: &'static str,
reason: String,
},
#[error("could not read {}: {message}", path.display())]
Read { path: PathBuf, message: String },
#[error("{} is not a PEM file: {message}", path.display())]

View file

@ -1,9 +1,13 @@
mod config;
mod error;
mod pool;
mod proxy;
mod settings;
mod tls;
pub use config::{HttpClientConfig, Verify};
pub use config::{HttpClientConfig, Resolution, Verify};
pub use error::Error;
pub use pool::{ClientVariant, HttpClientPool};
pub use settings::{HttpSettings, SslVerify};
pub use proxy::EnvironmentProxies;
pub use settings::{HttpSettings, SslVerify, TcpKeepalive};
pub use tls::{KeyExchangeGroup, Tls12CipherSuite, Unsupported, client_config};

View file

@ -13,6 +13,7 @@ pub enum ClientVariant {
Provider,
NoRedirect,
Media,
UnpinnedMedia,
}
const CLIENT_TTL: Duration = Duration::from_secs(3600);
@ -54,6 +55,10 @@ impl HttpClientPool {
trust_proxy_env: false,
..config.clone()
},
ClientVariant::UnpinnedMedia => HttpClientConfig {
client_certificate: None,
..config.clone()
},
ClientVariant::Provider | ClientVariant::NoRedirect => config.clone(),
};
let key = (effective, variant);
@ -84,7 +89,9 @@ impl HttpClientPool {
) -> reqwest::ClientBuilder {
match variant {
ClientVariant::Provider => builder,
ClientVariant::NoRedirect => builder.redirect(reqwest::redirect::Policy::none()),
ClientVariant::NoRedirect | ClientVariant::UnpinnedMedia => {
builder.redirect(reqwest::redirect::Policy::none())
}
ClientVariant::Media => builder
.redirect(reqwest::redirect::Policy::none())
.dns_resolver2(Arc::clone(&self.media_resolver)),
@ -125,7 +132,7 @@ mod tests {
fn config(user_agent: &str) -> HttpClientConfig {
HttpClientConfig {
user_agent: Some(user_agent.into()),
..HttpClientConfig::resolve(&HttpSettings::default()).unwrap()
..HttpClientConfig::resolve(&HttpSettings::default()).config
}
}
@ -233,6 +240,10 @@ mod tests {
.is_err()
);
assert!(pool.client(&with_identity, ClientVariant::Media).is_ok());
assert!(
pool.client(&with_identity, ClientVariant::UnpinnedMedia)
.is_ok()
);
}
#[test]
@ -277,6 +288,20 @@ mod tests {
assert_eq!(response.headers()["location"], "/elsewhere");
}
#[tokio::test]
async fn unpinned_media_variant_uses_the_system_resolver_and_returns_redirects() {
let (address, _, _) = serve("HTTP/1.1 302 Found").await;
let pool = HttpClientPool::new(Arc::new(FixedResolver(([192, 0, 2, 1], 80).into())));
let response = get(
&pool,
&config("a"),
ClientVariant::UnpinnedMedia,
&format!("http://localhost:{}/doc", address.port()),
)
.await;
assert_eq!(response.status(), 302);
}
#[tokio::test]
async fn media_variant_resolves_through_the_injected_resolver() {
let (address, _, requests) = serve("HTTP/1.1 204 No Content").await;

View file

@ -0,0 +1,15 @@
use hyper_util::client::proxy::matcher::Matcher;
pub struct EnvironmentProxies(Matcher);
impl EnvironmentProxies {
pub fn from_environment() -> Self {
Self(Matcher::from_system())
}
pub fn apply_to(&self, url: &reqwest::Url) -> bool {
url.as_str()
.parse::<http::Uri>()
.is_ok_and(|uri| self.0.intercept(&uri).is_some())
}
}

View file

@ -20,6 +20,13 @@ impl SslVerify {
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct TcpKeepalive {
pub idle: Duration,
pub interval: Duration,
pub retries: u32,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct HttpSettings {
pub ssl_verify: Option<SslVerify>,
@ -34,6 +41,8 @@ pub struct HttpSettings {
pub trust_proxy_env: bool,
pub ignore_proxy_env: bool,
pub connect_timeout: Duration,
pub tcp_keepalive: Option<TcpKeepalive>,
pub pool_idle_timeout: Duration,
}
impl Default for HttpSettings {
@ -51,6 +60,8 @@ impl Default for HttpSettings {
trust_proxy_env: false,
ignore_proxy_env: false,
connect_timeout: Duration::from_secs(10),
tcp_keepalive: None,
pool_idle_timeout: Duration::from_secs(120),
}
}
}
@ -59,6 +70,10 @@ impl HttpSettings {
pub fn with_environment(self, env: &(dyn Fn(&str) -> Option<String> + Sync)) -> Self {
let enabled =
|name: &str| env(name).is_some_and(|value| value.trim().eq_ignore_ascii_case("true"));
let number = |name: &str| env(name).and_then(|value| value.trim().parse::<u32>().ok());
let seconds = |name: &str, default: u32| {
Duration::from_secs(u64::from(number(name).unwrap_or(default)))
};
Self {
ssl_verify: env("SSL_VERIFY")
.map(|value| SslVerify::parse(&value))
@ -81,6 +96,17 @@ impl HttpSettings {
user_agent: env("LITELLM_USER_AGENT").or(self.user_agent),
trust_proxy_env: self.trust_proxy_env || enabled("AIOHTTP_TRUST_ENV"),
ignore_proxy_env: self.ignore_proxy_env || enabled("DISABLE_AIOHTTP_TRUST_ENV"),
tcp_keepalive: enabled("AIOHTTP_SO_KEEPALIVE")
.then(|| TcpKeepalive {
idle: seconds("AIOHTTP_TCP_KEEPIDLE", 60),
interval: seconds("AIOHTTP_TCP_KEEPINTVL", 30),
retries: number("AIOHTTP_TCP_KEEPCNT").unwrap_or(5),
})
.or(self.tcp_keepalive),
pool_idle_timeout: number("AIOHTTP_KEEPALIVE_TIMEOUT")
.map_or(self.pool_idle_timeout, |timeout| {
Duration::from_secs(u64::from(timeout))
}),
..self
}
}
@ -188,6 +214,32 @@ mod tests {
assert_eq!(settings.ssl_ecdh_curve, None);
}
#[test]
fn socket_keepalive_follows_the_aiohttp_variables_with_python_defaults() {
let tuned = HttpSettings::default().with_environment(&env_of(&[
("AIOHTTP_SO_KEEPALIVE", "True"),
("AIOHTTP_TCP_KEEPIDLE", "45"),
("AIOHTTP_KEEPALIVE_TIMEOUT", "30"),
]));
assert_eq!(
tuned.tcp_keepalive,
Some(TcpKeepalive {
idle: Duration::from_secs(45),
interval: Duration::from_secs(30),
retries: 5,
})
);
assert_eq!(tuned.pool_idle_timeout, Duration::from_secs(30));
}
#[test]
fn socket_keepalive_stays_off_unless_enabled() {
let settings =
HttpSettings::default().with_environment(&env_of(&[("AIOHTTP_TCP_KEEPIDLE", "45")]));
assert_eq!(settings.tcp_keepalive, None);
assert_eq!(settings.pool_idle_timeout, Duration::from_secs(120));
}
#[test]
fn missing_files_fall_back_to_default_verification() {
let settings = HttpSettings {

View file

@ -0,0 +1,402 @@
use std::{fmt, path::Path, sync::Arc};
use rustls::{
CipherSuite, ClientConfig, DigitallySignedStruct, RootCertStore, SignatureScheme,
client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier},
crypto::{CryptoProvider, SupportedKxGroup, ring},
pki_types::{CertificateDer, PrivateKeyDer, ServerName, UnixTime, pem::PemObject},
};
use crate::{
config::{HttpClientConfig, Verify},
error::Error,
};
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum KeyExchangeGroup {
X25519,
Secp256r1,
Secp384r1,
}
impl KeyExchangeGroup {
pub(crate) fn from_openssl_name(name: &str) -> Result<Self, Unsupported> {
match name.trim().to_ascii_lowercase().as_str() {
"x25519" => Ok(Self::X25519),
"prime256v1" | "secp256r1" | "p-256" => Ok(Self::Secp256r1),
"secp384r1" | "p-384" => Ok(Self::Secp384r1),
_ => Err(Unsupported::EcdhCurve(name.to_owned())),
}
}
fn supported(self) -> &'static dyn SupportedKxGroup {
match self {
Self::X25519 => ring::kx_group::X25519,
Self::Secp256r1 => ring::kx_group::SECP256R1,
Self::Secp384r1 => ring::kx_group::SECP384R1,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub enum Tls12CipherSuite {
EcdheEcdsaAes128Gcm,
EcdheEcdsaAes256Gcm,
EcdheEcdsaChacha20,
EcdheRsaAes128Gcm,
EcdheRsaAes256Gcm,
EcdheRsaChacha20,
}
impl Tls12CipherSuite {
fn from_openssl_name(name: &str) -> Option<Self> {
match name {
"ECDHE-ECDSA-AES128-GCM-SHA256" => Some(Self::EcdheEcdsaAes128Gcm),
"ECDHE-ECDSA-AES256-GCM-SHA384" => Some(Self::EcdheEcdsaAes256Gcm),
"ECDHE-ECDSA-CHACHA20-POLY1305" => Some(Self::EcdheEcdsaChacha20),
"ECDHE-RSA-AES128-GCM-SHA256" => Some(Self::EcdheRsaAes128Gcm),
"ECDHE-RSA-AES256-GCM-SHA384" => Some(Self::EcdheRsaAes256Gcm),
"ECDHE-RSA-CHACHA20-POLY1305" => Some(Self::EcdheRsaChacha20),
_ => None,
}
}
fn suite(self) -> CipherSuite {
match self {
Self::EcdheEcdsaAes128Gcm => CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_128_GCM_SHA256,
Self::EcdheEcdsaAes256Gcm => CipherSuite::TLS_ECDHE_ECDSA_WITH_AES_256_GCM_SHA384,
Self::EcdheEcdsaChacha20 => CipherSuite::TLS_ECDHE_ECDSA_WITH_CHACHA20_POLY1305_SHA256,
Self::EcdheRsaAes128Gcm => CipherSuite::TLS_ECDHE_RSA_WITH_AES_128_GCM_SHA256,
Self::EcdheRsaAes256Gcm => CipherSuite::TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384,
Self::EcdheRsaChacha20 => CipherSuite::TLS_ECDHE_RSA_WITH_CHACHA20_POLY1305_SHA256,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq, Hash, thiserror::Error)]
pub enum Unsupported {
#[error(
"ssl_ecdh_curve {0:?} is not supported: rustls with ring only offers X25519, prime256v1 and secp384r1, so the default key exchange groups are used"
)]
EcdhCurve(String),
#[error(
"ssl_security_level {0:?} is not supported: rustls has one fixed security level, comparable to OpenSSL level 2, so legacy servers that need a lower level cannot be reached"
)]
SecurityLevel(String),
#[error(
"ssl_security_level entry {0:?} is not supported: rustls only offers ECDHE AEAD cipher suites, so the entry is ignored"
)]
CipherToken(String),
}
pub(crate) struct CipherSelection {
pub(crate) tls12_cipher_suites: Option<Vec<Tls12CipherSuite>>,
pub(crate) unsupported: Vec<Unsupported>,
}
enum CipherToken {
Suite(Tls12CipherSuite),
EverySuite,
Ordering,
Unsupported(Unsupported),
}
fn cipher_token(token: &str) -> CipherToken {
if let Some(suite) = Tls12CipherSuite::from_openssl_name(token) {
return CipherToken::Suite(suite);
}
match token {
"DEFAULT" | "ALL" | "HIGH" => CipherToken::EverySuite,
"@STRENGTH" | "@SECLEVEL=2" => CipherToken::Ordering,
level if level.starts_with("@SECLEVEL=") => {
CipherToken::Unsupported(Unsupported::SecurityLevel(level.to_owned()))
}
other => CipherToken::Unsupported(Unsupported::CipherToken(other.to_owned())),
}
}
pub(crate) fn parse_cipher_string(value: &str) -> CipherSelection {
let tokens: Vec<CipherToken> = tokenize(value)
.iter()
.map(|token| cipher_token(token))
.collect();
let every_suite = tokens
.iter()
.any(|token| matches!(token, CipherToken::EverySuite));
let mut suites: Vec<Tls12CipherSuite> = tokens
.iter()
.filter_map(|token| match token {
CipherToken::Suite(suite) => Some(*suite),
_ => None,
})
.collect();
suites.sort_unstable();
suites.dedup();
CipherSelection {
tls12_cipher_suites: (!every_suite && !suites.is_empty()).then_some(suites),
unsupported: tokens
.into_iter()
.filter_map(|token| match token {
CipherToken::Unsupported(unsupported) => Some(unsupported),
_ => None,
})
.collect(),
}
}
fn tokenize(value: &str) -> Vec<String> {
value
.split([':', ',', ' '])
.flat_map(|entry| match entry.split_once('@') {
Some((name, command)) => vec![name.to_owned(), format!("@{command}")],
None => vec![entry.to_owned()],
})
.filter(|token| !token.is_empty())
.collect()
}
pub fn client_config(config: &HttpClientConfig) -> Result<ClientConfig, Error> {
let base = ring::default_provider();
let provider = Arc::new(CryptoProvider {
kx_groups: config
.key_exchange_group
.map_or_else(|| base.kx_groups.clone(), |group| vec![group.supported()]),
cipher_suites: base
.cipher_suites
.iter()
.copied()
.filter(|suite| {
suite.tls13().is_some()
|| config
.tls12_cipher_suites
.as_ref()
.is_none_or(|allowed| allowed.iter().any(|a| a.suite() == suite.suite()))
})
.collect(),
..base
});
let builder = ClientConfig::builder_with_provider(Arc::clone(&provider))
.with_safe_default_protocol_versions()
.map_err(|error| Error::Client(error.to_string()))?;
let verified = match &config.verify {
Verify::Disabled => builder
.dangerous()
.with_custom_certificate_verifier(Arc::new(NoVerification(provider))),
Verify::BuiltInRoots => builder.with_root_certificates(built_in_roots()),
Verify::CaBundle(path) => builder.with_root_certificates(bundle_roots(path)?),
};
let mut tls = match &config.client_certificate {
None => verified.with_no_client_auth(),
Some(path) => {
let (chain, key) = identity(path)?;
verified
.with_client_auth_cert(chain, key)
.map_err(|error| invalid_pem(path, error))?
}
};
tls.alpn_protocols = if config.http2 {
vec![b"h2".to_vec(), b"http/1.1".to_vec()]
} else {
vec![b"http/1.1".to_vec()]
};
Ok(tls)
}
fn built_in_roots() -> RootCertStore {
let mut store = RootCertStore::empty();
store.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());
store
}
fn bundle_roots(path: &Path) -> Result<RootCertStore, Error> {
let certificates = certificates(path)?;
if certificates.is_empty() {
return Err(invalid_pem(path, "no certificates found"));
}
let mut store = RootCertStore::empty();
for certificate in certificates {
store
.add(certificate)
.map_err(|error| invalid_pem(path, error))?;
}
Ok(store)
}
fn identity(path: &Path) -> Result<(Vec<CertificateDer<'static>>, PrivateKeyDer<'static>), Error> {
let chain = certificates(path)?;
if chain.is_empty() {
return Err(invalid_pem(path, "no certificates found"));
}
let key =
PrivateKeyDer::from_pem_slice(&read(path)?).map_err(|error| invalid_pem(path, error))?;
Ok((chain, key))
}
fn certificates(path: &Path) -> Result<Vec<CertificateDer<'static>>, Error> {
CertificateDer::pem_slice_iter(&read(path)?)
.collect::<Result<_, _>>()
.map_err(|error| invalid_pem(path, error))
}
fn read(path: &Path) -> Result<Vec<u8>, Error> {
std::fs::read(path).map_err(|error| Error::Read {
path: path.to_path_buf(),
message: error.to_string(),
})
}
fn invalid_pem(path: &Path, message: impl fmt::Display) -> Error {
Error::InvalidPem {
path: path.to_path_buf(),
message: message.to_string(),
}
}
#[derive(Debug)]
struct NoVerification(Arc<CryptoProvider>);
impl ServerCertVerifier for NoVerification {
fn verify_server_cert(
&self,
_end_entity: &CertificateDer<'_>,
_intermediates: &[CertificateDer<'_>],
_server_name: &ServerName<'_>,
_ocsp_response: &[u8],
_now: UnixTime,
) -> Result<ServerCertVerified, rustls::Error> {
Ok(ServerCertVerified::assertion())
}
fn verify_tls12_signature(
&self,
_message: &[u8],
_cert: &CertificateDer<'_>,
_dss: &DigitallySignedStruct,
) -> Result<HandshakeSignatureValid, rustls::Error> {
Ok(HandshakeSignatureValid::assertion())
}
fn verify_tls13_signature(
&self,
_message: &[u8],
_cert: &CertificateDer<'_>,
_dss: &DigitallySignedStruct,
) -> Result<HandshakeSignatureValid, rustls::Error> {
Ok(HandshakeSignatureValid::assertion())
}
fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
self.0.signature_verification_algorithms.supported_schemes()
}
}
#[cfg(test)]
mod tests {
use rstest::rstest;
use rustls::NamedGroup;
use super::*;
use crate::HttpSettings;
fn config(settings: HttpSettings) -> HttpClientConfig {
HttpClientConfig::resolve(&settings).config
}
fn offered_groups(tls: &ClientConfig) -> Vec<NamedGroup> {
tls.crypto_provider()
.kx_groups
.iter()
.map(|group| group.name())
.collect()
}
fn offered_tls12_suites(tls: &ClientConfig) -> Vec<CipherSuite> {
tls.crypto_provider()
.cipher_suites
.iter()
.filter(|suite| suite.tls13().is_none())
.map(|suite| suite.suite())
.collect()
}
#[rstest]
#[case("X25519", NamedGroup::X25519)]
#[case("prime256v1", NamedGroup::secp256r1)]
#[case("secp384r1", NamedGroup::secp384r1)]
fn ecdh_curve_is_the_only_key_exchange_group_offered(
#[case] curve: &str,
#[case] expected: NamedGroup,
) {
let tls = client_config(&config(HttpSettings {
ssl_ecdh_curve: Some(curve.into()),
..HttpSettings::default()
}))
.unwrap();
assert_eq!(offered_groups(&tls), [expected]);
}
#[test]
fn default_settings_offer_every_group_and_suite_of_the_provider() {
let tls = client_config(&config(HttpSettings::default())).unwrap();
let provider = ring::default_provider();
assert_eq!(offered_groups(&tls).len(), provider.kx_groups.len());
assert_eq!(
tls.crypto_provider().cipher_suites.len(),
provider.cipher_suites.len()
);
}
#[test]
fn named_suites_are_the_only_tls12_suites_offered_and_tls13_stays() {
let tls = client_config(&config(HttpSettings {
ssl_security_level: Some("ECDHE-RSA-AES256-GCM-SHA384".into()),
..HttpSettings::default()
}))
.unwrap();
assert_eq!(
offered_tls12_suites(&tls),
[CipherSuite::TLS_ECDHE_RSA_WITH_AES_256_GCM_SHA384]
);
assert!(
tls.crypto_provider()
.cipher_suites
.iter()
.any(|suite| suite.tls13().is_some())
);
}
#[rstest]
#[case(true, &[b"h2".as_slice(), b"http/1.1".as_slice()])]
#[case(false, &[b"http/1.1".as_slice()])]
fn alpn_offers_h2_only_when_http2_is_on(#[case] http2: bool, #[case] expected: &[&[u8]]) {
let tls = client_config(&config(HttpSettings {
http2,
..HttpSettings::default()
}))
.unwrap();
assert_eq!(tls.alpn_protocols, expected);
}
#[test]
fn client_certificate_without_a_private_key_is_an_invalid_pem_error() {
let path = std::env::temp_dir().join(format!(
"litellm-http-cert-without-key-{}.pem",
std::process::id()
));
std::fs::write(
&path,
b"-----BEGIN CERTIFICATE-----\nAA==\n-----END CERTIFICATE-----\n",
)
.unwrap();
let result = client_config(&HttpClientConfig {
client_certificate: Some(path.clone()),
..config(HttpSettings::default())
})
.map(drop);
std::fs::remove_file(&path).unwrap();
assert!(matches!(
result,
Err(Error::InvalidPem { path: reported, .. }) if reported == path
));
}
}

View file

@ -16,7 +16,7 @@ use crate::{
},
custom_httpx::{
http_handler::{HeaderPolicy, execute_http_request, with_headers},
media::MediaFetcher,
media::{MediaFetcher, UrlPolicy},
transport,
},
};
@ -41,12 +41,13 @@ impl OcrClient {
pub fn new(
pool: &HttpClientPool,
config: &HttpClientConfig,
url_policy: UrlPolicy,
vertex_auth: VertexAuth,
) -> Result<Self, litellm_http::Error> {
Ok(Self {
provider_http: pool.client(config, ClientVariant::Provider)?,
polling_http: pool.client(config, ClientVariant::NoRedirect)?,
document_fetcher: MediaFetcher::new(pool, config)?,
document_fetcher: MediaFetcher::new(pool, config, url_policy)?,
vertex_auth,
})
}

View file

@ -7,7 +7,7 @@ use std::{
time::Duration,
};
use litellm_http::{ClientVariant, HttpClientConfig, HttpClientPool};
use litellm_http::{ClientVariant, EnvironmentProxies, HttpClientConfig, HttpClientPool};
use reqwest::{
Url,
dns::{Addrs, Name, Resolve, Resolving},
@ -35,10 +35,45 @@ pub enum Error {
Transport(#[from] crate::custom_httpx::transport::Error),
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct UrlPolicy {
pub validate: bool,
pub allowed_hosts: Vec<String>,
}
impl Default for UrlPolicy {
fn default() -> Self {
Self {
validate: true,
allowed_hosts: Vec::new(),
}
}
}
impl UrlPolicy {
fn allows(&self, host: &str, port: u16) -> bool {
let host = normalize_host(host);
let with_port = format!("{host}:{port}");
self.allowed_hosts
.iter()
.map(|entry| normalize_host(entry))
.any(|entry| entry == host || entry == with_port)
}
}
fn normalize_host(host: &str) -> String {
host.to_ascii_lowercase().trim_end_matches('.').to_owned()
}
type ProxyMatch = Arc<dyn Fn(&Url) -> bool + Send + Sync>;
#[derive(Clone)]
pub struct MediaFetcher {
client: reqwest::Client,
pinned: reqwest::Client,
unpinned: reqwest::Client,
uses_proxy: ProxyMatch,
address_resolver: Arc<dyn AddressResolver>,
url_policy: UrlPolicy,
allow_private_network: bool,
}
@ -65,19 +100,36 @@ impl MediaFetcher {
pub fn new(
pool: &HttpClientPool,
config: &HttpClientConfig,
url_policy: UrlPolicy,
) -> Result<Self, litellm_http::Error> {
Self::with_address_resolver(pool, config, Arc::new(SystemAddressResolver))
let uses_proxy: ProxyMatch = if config.trust_proxy_env {
let proxies = EnvironmentProxies::from_environment();
Arc::new(move |url| proxies.apply_to(url))
} else {
Arc::new(|_| false)
};
Self::with_resolution(
pool,
config,
url_policy,
Arc::new(SystemAddressResolver),
uses_proxy,
)
}
fn with_address_resolver(
fn with_resolution(
pool: &HttpClientPool,
config: &HttpClientConfig,
url_policy: UrlPolicy,
address_resolver: Arc<dyn AddressResolver>,
uses_proxy: ProxyMatch,
) -> Result<Self, litellm_http::Error> {
let client = pool.client(config, ClientVariant::Media)?;
Ok(Self {
client,
pinned: pool.client(config, ClientVariant::Media)?,
unpinned: pool.client(config, ClientVariant::UnpinnedMedia)?,
uses_proxy,
address_resolver,
url_policy,
allow_private_network: false,
})
}
@ -85,8 +137,11 @@ impl MediaFetcher {
#[cfg(any(test, feature = "test-support"))]
pub fn for_test(client: reqwest::Client) -> Self {
Self {
client,
pinned: client.clone(),
unpinned: client,
uses_proxy: Arc::new(|_| false),
address_resolver: Arc::new(AllowPrivateResolver),
url_policy: UrlPolicy::default(),
allow_private_network: true,
}
}
@ -107,9 +162,9 @@ impl MediaFetcher {
) -> Result<DownloadedMedia, Error> {
let mut redirects_followed = 0;
loop {
self.validate_url(&url).await?;
let mut response = self
.client
.client_for(&url)
.await?
.get(url.clone())
.send()
.await
@ -156,7 +211,10 @@ impl MediaFetcher {
}
}
async fn validate_url(&self, url: &Url) -> Result<(), Error> {
async fn client_for(&self, url: &Url) -> Result<&reqwest::Client, Error> {
if !self.url_policy.validate {
return Ok(&self.unpinned);
}
if !matches!(url.scheme(), "http" | "https")
|| !url.username().is_empty()
|| url.password().is_some()
@ -165,12 +223,28 @@ impl MediaFetcher {
}
let host = url.host_str().ok_or(Error::BlockedUrl)?;
if self.allow_private_network {
return Ok(());
}
if let Ok(ip) = host.parse::<IpAddr>() {
return (!is_blocked_ip(ip)).then_some(()).ok_or(Error::BlockedUrl);
return Ok(&self.pinned);
}
let port = url.port_or_known_default().ok_or(Error::BlockedUrl)?;
if self.url_policy.allows(host, port) {
return Ok(&self.unpinned);
}
self.validate_host(host, port).await?;
Ok(if (self.uses_proxy)(url) {
&self.unpinned
} else {
&self.pinned
})
}
async fn validate_host(&self, host: &str, port: u16) -> Result<(), Error> {
if let Ok(ip) = host
.trim_start_matches('[')
.trim_end_matches(']')
.parse::<IpAddr>()
{
return (!is_blocked_ip(ip)).then_some(()).ok_or(Error::BlockedUrl);
}
let addresses = self
.address_resolver
.resolve(host, port)
@ -360,14 +434,35 @@ mod tests {
address: SocketAddr,
blocked_hosts: HashSet<&'static str>,
) -> MediaFetcher {
MediaFetcher::with_address_resolver(
&HttpClientPool::new(Arc::new(LoopbackDnsResolver(address))),
&HttpClientConfig::resolve(&HttpSettings::default()).unwrap(),
fetcher(address, blocked_hosts, UrlPolicy::default(), false)
}
fn fetcher(
pinned_address: SocketAddr,
blocked_hosts: HashSet<&'static str>,
url_policy: UrlPolicy,
uses_proxy: bool,
) -> MediaFetcher {
let direct = HttpClientConfig {
trust_proxy_env: false,
..HttpClientConfig::resolve(&HttpSettings::default()).config
};
MediaFetcher::with_resolution(
&HttpClientPool::new(Arc::new(LoopbackDnsResolver(pinned_address))),
&direct,
url_policy,
Arc::new(TestAddressResolver { blocked_hosts }),
Arc::new(move |_| uses_proxy),
)
.expect("test fetcher builds")
}
const UNROUTABLE: SocketAddr =
SocketAddr::new(IpAddr::V4(std::net::Ipv4Addr::new(192, 0, 2, 1)), 9);
const OK_RESPONSE: &[u8] =
b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok";
fn policy(max_bytes: u64, max_redirects: usize) -> DownloadPolicy {
DownloadPolicy {
timeout: Duration::from_secs(1),
@ -541,14 +636,88 @@ mod tests {
async fn rejects_url_credentials_before_network_access() {
let fetcher = MediaFetcher::new(
&HttpClientPool::new(Arc::new(PublicDnsResolver)),
&HttpClientConfig::resolve(&HttpSettings::default()).expect("default settings resolve"),
&HttpClientConfig::resolve(&HttpSettings::default()).config,
UrlPolicy::default(),
)
.expect("media fetcher builds");
let url =
Url::parse("https://user:password@8.8.8.8/document").expect("credentialed URL parses");
assert!(matches!(
fetcher.validate_url(&url).await,
fetcher.fetch(url, policy(1, 0)).await,
Err(Error::BlockedUrl)
));
}
#[tokio::test]
async fn allowlisted_private_host_is_fetched_without_the_pinned_resolver() {
let (url, server, _) = serve_named("localhost", vec![OK_RESPONSE]).await;
let port = url.port().expect("test URL has a port");
let allowed = UrlPolicy {
validate: true,
allowed_hosts: vec![format!("LOCALHOST:{port}")],
};
let media = fetcher(UNROUTABLE, HashSet::from(["localhost"]), allowed, false)
.fetch(url, policy(2, 0))
.await
.expect("allowlisted host downloads");
server.await.expect("server completes");
assert_eq!(media.bytes, b"ok");
}
#[tokio::test]
async fn allowlist_entry_for_another_port_does_not_open_the_host() {
let (url, _server, _) = serve_named("localhost", vec![OK_RESPONSE]).await;
let other_port = UrlPolicy {
validate: true,
allowed_hosts: vec!["localhost:1".into()],
};
let result = fetcher(UNROUTABLE, HashSet::from(["localhost"]), other_port, false)
.fetch(url, policy(2, 0))
.await;
assert!(matches!(result, Err(Error::BlockedUrl)));
}
#[tokio::test]
async fn validation_off_fetches_private_hosts_and_follows_redirects() {
let (url, server, _) = serve_named(
"localhost",
vec![
b"HTTP/1.1 302 Found\r\nLocation: /moved\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
OK_RESPONSE,
],
)
.await;
let off = UrlPolicy {
validate: false,
allowed_hosts: Vec::new(),
};
let media = fetcher(UNROUTABLE, HashSet::from(["localhost"]), off, false)
.fetch(url, policy(2, 1))
.await
.expect("unvalidated download succeeds");
let requests = server.await.expect("server completes");
assert_eq!(media.bytes, b"ok");
assert!(requests[1].starts_with("GET /moved "));
}
#[tokio::test]
async fn proxied_urls_skip_the_pinned_resolver_but_keep_the_address_check() {
let (url, server, _) = serve_named("localhost", vec![OK_RESPONSE]).await;
let media = fetcher(UNROUTABLE, HashSet::new(), UrlPolicy::default(), true)
.fetch(url.clone(), policy(2, 0))
.await
.expect("public host behind a proxy downloads");
server.await.expect("server completes");
assert_eq!(media.bytes, b"ok");
let blocked = fetcher(
UNROUTABLE,
HashSet::from(["localhost"]),
UrlPolicy::default(),
true,
)
.fetch(url, policy(2, 0))
.await;
assert!(matches!(blocked, Err(Error::BlockedUrl)));
}
}

View file

@ -1,10 +1,11 @@
use std::{
collections::HashSet,
path::{Path, PathBuf},
sync::{Arc, LazyLock},
sync::{Arc, LazyLock, Mutex, PoisonError},
};
use litellm_http::{HttpClientConfig, HttpClientPool, HttpSettings, SslVerify};
use litellm_llms::custom_httpx::media::PublicDnsResolver;
use litellm_http::{HttpClientConfig, HttpClientPool, HttpSettings, SslVerify, Unsupported};
use litellm_llms::custom_httpx::media::{PublicDnsResolver, UrlPolicy};
use pyo3::{prelude::*, types::PyDict};
use crate::{errors::RustBridgeDeclined, python_settings::PythonSettings};
@ -12,6 +13,8 @@ use crate::{errors::RustBridgeDeclined, python_settings::PythonSettings};
static POOL: LazyLock<HttpClientPool> =
LazyLock::new(|| HttpClientPool::new(Arc::new(PublicDnsResolver)));
static REPORTED_UNSUPPORTED: LazyLock<Mutex<HashSet<Unsupported>>> = LazyLock::new(Mutex::default);
pub(crate) fn pool() -> &'static HttpClientPool {
&POOL
}
@ -21,22 +24,48 @@ pub(crate) fn call_config(
kwargs: &Bound<'_, PyDict>,
asynchronous: bool,
) -> PyResult<HttpClientConfig> {
decline_live_client(kwargs)?;
decline_custom_url_policy(&PythonSettings::UrlPolicy.read(py)?)?;
let configured = settings(&PythonSettings::Http.read(py)?)?
.with_environment(&|name| std::env::var(name).ok());
let settings = for_call(configured, call_ssl_verify(kwargs)?, asynchronous)
.without_missing_files(&|path: &Path| path.exists());
HttpClientConfig::resolve(&settings)
.map_err(|error| RustBridgeDeclined::new_err(error.to_string()))
let resolution = HttpClientConfig::resolve(&settings);
for unsupported in unreported(&REPORTED_UNSUPPORTED, resolution.unsupported) {
PythonSettings::warn(py, &unsupported.to_string())?;
}
Ok(resolution.config)
}
fn unreported(
reported: &Mutex<HashSet<Unsupported>>,
unsupported: Vec<Unsupported>,
) -> Vec<Unsupported> {
let mut reported = reported.lock().unwrap_or_else(PoisonError::into_inner);
unsupported
.into_iter()
.filter(|unsupported| reported.insert(unsupported.clone()))
.collect()
}
pub(crate) fn url_policy(py: Python<'_>) -> PyResult<UrlPolicy> {
let policy: PythonUrlPolicy =
PythonSettings::UrlPolicy
.read(py)?
.extract()
.map_err(|error: PyErr| {
RustBridgeDeclined::new_err(format!(
"litellm URL policy cannot be used by the Rust route: {error}"
))
})?;
Ok(UrlPolicy {
validate: policy.user_url_validation,
allowed_hosts: policy.user_url_allowed_hosts,
})
}
fn call_ssl_verify(kwargs: &Bound<'_, PyDict>) -> PyResult<Option<SslVerify>> {
kwargs
Ok(kwargs
.get_item("ssl_verify")?
.filter(|value| !value.is_none())
.map(|value| ssl_verify(&value, "the ssl_verify argument"))
.transpose()
.and_then(|value| ssl_verify(&value)))
}
fn for_call(
@ -51,35 +80,12 @@ fn for_call(
}
}
fn decline_live_client(kwargs: &Bound<'_, PyDict>) -> PyResult<()> {
if kwargs
.get_item("client")?
.is_some_and(|value| !value.is_none())
{
return Err(RustBridgeDeclined::new_err(
"client is a live Python HTTP client and cannot be used by the Rust route",
));
}
Ok(())
}
#[derive(FromPyObject)]
struct PythonUrlPolicy {
user_url_validation: bool,
user_url_allowed_hosts: Vec<String>,
}
fn decline_custom_url_policy(value: &Bound<'_, PyAny>) -> PyResult<()> {
match value.extract::<PythonUrlPolicy>() {
Ok(policy) if policy.user_url_validation && policy.user_url_allowed_hosts.is_empty() => {
Ok(())
}
Ok(_) | Err(_) => Err(RustBridgeDeclined::new_err(
"litellm.user_url_validation / user_url_allowed_hosts are applied by the Python route",
)),
}
}
#[derive(FromPyObject)]
struct PythonHttpSettings<'py> {
ssl_verify: Bound<'py, PyAny>,
@ -101,7 +107,7 @@ fn settings(value: &Bound<'_, PyAny>) -> PyResult<HttpSettings> {
))
})?;
Ok(HttpSettings {
ssl_verify: Some(ssl_verify(&python.ssl_verify, "litellm.ssl_verify")?),
ssl_verify: ssl_verify(&python.ssl_verify),
ssl_certificate: python.ssl_certificate.map(PathBuf::from),
ssl_security_level: python.ssl_security_level,
ssl_ecdh_curve: python.ssl_ecdh_curve,
@ -115,20 +121,18 @@ fn settings(value: &Bound<'_, PyAny>) -> PyResult<HttpSettings> {
})
}
fn ssl_verify(value: &Bound<'_, PyAny>, source: &str) -> PyResult<SslVerify> {
fn ssl_verify(value: &Bound<'_, PyAny>) -> Option<SslVerify> {
if let Ok(enabled) = value.extract::<bool>() {
return Ok(if enabled {
return Some(if enabled {
SslVerify::Enabled
} else {
SslVerify::Disabled
});
}
if let Ok(path) = value.extract::<String>() {
return Ok(SslVerify::parse(&path));
}
Err(RustBridgeDeclined::new_err(format!(
"{source} is a live Python object and cannot be used by the Rust route"
)))
value
.extract::<String>()
.ok()
.map(|path| SslVerify::parse(&path))
}
#[cfg(test)]
@ -247,50 +251,30 @@ user_agent='litellm/9.9.9',
Python::initialize();
Python::attach(|py| {
let settings = settings(&python_settings(py, overrides)).unwrap();
let config = HttpClientConfig::resolve(&settings).unwrap();
let config = HttpClientConfig::resolve(&settings).config;
assert_eq!(config.verify, expected);
});
}
#[test]
fn ssl_context_global_declines_instead_of_being_dropped() {
fn ssl_context_global_is_ignored_so_environment_and_defaults_apply() {
Python::initialize();
Python::attach(|py| {
let error = settings(&python_settings(py, "ssl_verify=object()")).unwrap_err();
assert!(error.is_instance_of::<RustBridgeDeclined>(py));
assert!(error.value(py).to_string().contains("litellm.ssl_verify"));
let settings = settings(&python_settings(py, "ssl_verify=object()")).unwrap();
assert_eq!(settings.ssl_verify, None);
});
}
fn url_policy<'py>(py: Python<'py>, fields: &str) -> Bound<'py, PyAny> {
let source = std::ffi::CString::new(format!(
"import types\npolicy = types.SimpleNamespace({fields})"
))
.unwrap();
let locals = PyDict::new(py);
py.run(&source, Some(&locals), Some(&locals)).unwrap();
locals.get_item("policy").unwrap().unwrap()
}
#[test]
fn default_url_policy_stays_on_the_rust_route() {
Python::initialize();
Python::attach(|py| {
let policy = url_policy(py, "user_url_validation=True, user_url_allowed_hosts=[]");
decline_custom_url_policy(&policy).unwrap();
});
}
#[rstest]
#[case::validation_off("user_url_validation=False, user_url_allowed_hosts=[]")]
#[case::allowlist("user_url_validation=True, user_url_allowed_hosts=['docs.internal']")]
#[case::mistyped("user_url_validation=True, user_url_allowed_hosts=None")]
fn custom_url_policy_declines_so_python_applies_it(#[case] fields: &str) {
Python::initialize();
Python::attach(|py| {
let error = decline_custom_url_policy(&url_policy(py, fields)).unwrap_err();
assert!(error.is_instance_of::<RustBridgeDeclined>(py));
});
fn unsupported_settings_are_reported_once_per_process() {
let reported = Mutex::default();
let curve = Unsupported::EcdhCurve("secp521r1".into());
let level = Unsupported::SecurityLevel("@SECLEVEL=1".into());
assert_eq!(
unreported(&reported, vec![curve.clone(), level.clone()]),
[curve.clone(), level]
);
assert_eq!(unreported(&reported, vec![curve]), []);
}
#[test]
@ -333,15 +317,19 @@ user_agent='litellm/9.9.9',
}
#[test]
fn live_ssl_context_argument_declines() {
fn live_ssl_context_argument_is_ignored_so_the_configured_value_applies() {
Python::initialize();
Python::attach(|py| {
let kwargs = PyDict::new(py);
kwargs
.set_item("ssl_verify", py.eval(c"object()", None, None).unwrap())
.unwrap();
let error = call_ssl_verify(&kwargs).unwrap_err();
assert!(error.is_instance_of::<RustBridgeDeclined>(py));
let configured = HttpSettings {
ssl_verify: Some(SslVerify::Disabled),
..HttpSettings::default()
};
let settings = for_call(configured.clone(), call_ssl_verify(&kwargs).unwrap(), true);
assert_eq!(settings, configured);
});
}
@ -357,33 +345,7 @@ user_agent='litellm/9.9.9',
..HttpSettings::default()
};
let settings = for_call(opted_out, None, asynchronous);
let config = HttpClientConfig::resolve(&settings).unwrap();
let config = HttpClientConfig::resolve(&settings).config;
assert_eq!(config.trust_proxy_env, expected);
}
#[test]
fn live_python_client_declines_before_dispatch() {
Python::initialize();
Python::attach(|py| {
let kwargs = PyDict::new(py);
kwargs
.set_item("client", py.eval(c"object()", None, None).unwrap())
.unwrap();
let error = decline_live_client(&kwargs).unwrap_err();
assert!(error.is_instance_of::<RustBridgeDeclined>(py));
});
}
#[rstest]
#[case::absent_client("{}")]
#[case::none_client("{'client': None}")]
#[case::proxy_shared_session("{'shared_session': object()}")]
fn calls_without_a_python_client_stay_on_the_rust_route(#[case] kwargs: &str) {
Python::initialize();
Python::attach(|py| {
let source = std::ffi::CString::new(kwargs).unwrap();
let kwargs = py.eval(&source, None, None).unwrap();
decline_live_client(kwargs.cast::<PyDict>().unwrap()).unwrap();
});
}
}

View file

@ -22,6 +22,11 @@ impl PythonSettings {
pub(crate) fn read(self, py: Python<'_>) -> PyResult<Bound<'_, PyAny>> {
py.import(MODULE)?.getattr(self.name())?.call0()
}
pub(crate) fn warn(py: Python<'_>, message: &str) -> PyResult<()> {
py.import(MODULE)?.getattr("warn")?.call1((message,))?;
Ok(())
}
}
#[cfg(test)]

View file

@ -38,8 +38,13 @@ fn run_ocr(
asynchronous: bool,
) -> PyResult<Py<PyAny>> {
let config = http::call_config(py, &kwargs, asynchronous)?;
let client = OcrClient::new(http::pool(), &config, VERTEX_AUTH.clone())
.map_err(|error| RustBridgeDeclined::new_err(error.to_string()))?;
let client = OcrClient::new(
http::pool(),
&config,
http::url_policy(py)?,
VERTEX_AUTH.clone(),
)
.map_err(|error| RustBridgeDeclined::new_err(error.to_string()))?;
run_legacy_call(
py,
if asynchronous { ASYNC_SURFACE } else { SURFACE },

View file

@ -24,6 +24,12 @@ class UrlPolicy:
user_url_allowed_hosts: Sequence[str]
def warn(message: str) -> None:
from litellm._logging import verbose_logger
verbose_logger.warning("%s", message)
def url_policy() -> UrlPolicy:
import litellm

View file

@ -1,4 +1,5 @@
import dataclasses
import logging
from pathlib import Path
from typing import Final
@ -65,3 +66,10 @@ def test_http_settings_ignores_environment_overrides(monkeypatch: pytest.MonkeyP
assert result.user_agent == default_user_agent()
assert result.ssl_verify is True
def test_warn_reaches_the_litellm_logger(caplog: pytest.LogCaptureFixture) -> None:
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
settings.warn("ssl_ecdh_curve 'secp521r1' is not supported")
assert [record.getMessage() for record in caplog.records] == ["ssl_ecdh_curve 'secp521r1' is not supported"]