karylab_agents/vllm
1
0
Fork 0
forked from Karylab-cklius/vllm
Code Pull Requests 2 Actions 1 Packages Activity

[Rust Frontend] Add static HTTPS and mTLS support for HTTP and gRPC (#45890)

Browse Source 
Co-authored-by: Bugen Zhao <i@bugenzhao.com>
Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
Signed-off-by: Tahsin Tunan <tahsintunan@gmail.com>
Signed-off-by: Bugen Zhao <i@bugenzhao.com>
This commit is contained in:
Tahsin Tunan
2026-06-30 01:45:59 +00:00
committed by GitHub
co-authored by Bugen Zhao mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>
parent 43916891b2
commit b8cb75b149
16 changed files with 1942 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/Cargo.lock
Generated
+69 -20
View File
@@ -272,6 +272,18 @@ version = "1.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
[[package]]
name = "auto_enums"
version = "0.8.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2e4487600931c9a89f8db7ffbdf3fbdd45bb7bd85e26861f659a463cd0dff966"
dependencies = [
"derive_utils",
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "auto_impl"
version = "1.3.0"
@@ -938,6 +950,17 @@ dependencies = [
"unicode-xid",
]
[[package]]
name = "derive_utils"
version = "0.15.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "362f47930db19fe7735f527e6595e4900316b893ebf6d48ad3d31be928d57dd6"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.117",
]
[[package]]
name = "digest"
version = "0.10.7"
@@ -1478,9 +1501,9 @@ dependencies = [
[[package]]
name = "h2"
version = "0.4.13"
version = "0.4.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54"
checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155"
dependencies = [
"atomic-waker",
"bytes",
@@ -1638,9 +1661,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
[[package]]
name = "hyper"
version = "1.8.1"
version = "1.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11"
checksum = "55281c53a1894c864990125767da440a4e630446785086f52523b20033b74498"
dependencies = [
"atomic-waker",
"bytes",
@@ -1653,7 +1676,6 @@ dependencies = [
"httpdate",
"itoa",
"pin-project-lite",
"pin-utils",
"smallvec",
"tokio",
"want",
@@ -2569,15 +2591,14 @@ dependencies = [
[[package]]
name = "openssl"
version = "0.10.76"
version = "0.10.81"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "951c002c75e16ea2c65b8c7e4d3d51d5530d8dfa7d060b4776828c88cfb18ecf"
checksum = "77823a27f0babb03091cb9ed9ef80af3b39dbc82f97e8fa530374b7dafd87a45"
dependencies = [
"bitflags",
"cfg-if",
"foreign-types",
"libc",
"once_cell",
"openssl-macros",
"openssl-sys",
]
@@ -2610,9 +2631,9 @@ dependencies = [
[[package]]
name = "openssl-sys"
version = "0.9.112"
version = "0.9.117"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "57d55af3b3e226502be1526dfdba67ab0e9c96fc293004e79576b2b9edb0dbdb"
checksum = "b47e7e6bb2c38cd930d25a23b40fa52e068c10e85f3e03a7f5ba5aaca5713695"
dependencies = [
"cc",
"libc",
@@ -2783,12 +2804,6 @@ version = "0.2.17"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
[[package]]
name = "pin-utils"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
[[package]]
name = "pkg-config"
version = "0.3.32"
@@ -2988,7 +3003,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7"
dependencies = [
"heck",
"itertools 0.10.5",
"itertools 0.14.0",
"log",
"multimap",
"petgraph",
@@ -3009,7 +3024,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "27c6023962132f4b30eb4c172c91ce92d933da334c59c23cddee82358ddafb0b"
dependencies = [
"anyhow",
"itertools 0.10.5",
"itertools 0.14.0",
"proc-macro2",
"quote",
"syn 2.0.117",
@@ -3503,9 +3518,9 @@ dependencies = [
[[package]]
name = "rustls-pki-types"
version = "1.14.0"
version = "1.14.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "be040f8b0a225e40375822a563fa9524378b9d63112f53e19ffff34df5d33fdd"
checksum = "30a7197ae7eb376e574fe940d068c30fe0462554a3ddbe4eca7838e049c937a9"
dependencies = [
"zeroize",
]
@@ -4385,6 +4400,22 @@ dependencies = [
"serde_json",
]
[[package]]
name = "tls-listener"
version = "0.11.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1461056cc1ef47003f7ee16e4cef3741068d4c7f6b627bfce49b7c00c120a530"
dependencies = [
"axum",
"futures-util",
"openssl",
"pin-project-lite",
"thiserror 2.0.18",
"tokio",
"tokio-openssl",
"tracing",
]
[[package]]
name = "tokenizers"
version = "0.22.2"
@@ -4457,6 +4488,17 @@ dependencies = [
"tokio",
]
[[package]]
name = "tokio-openssl"
version = "0.6.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "59df6849caa43bb7567f9a36f863c447d95a11d5903c9cc334ba32576a27eadd"
dependencies = [
"openssl",
"openssl-sys",
"tokio",
]
[[package]]
name = "tokio-rustls"
version = "0.26.4"
@@ -5220,6 +5262,7 @@ dependencies = [
"anyhow",
"async-openai",
"asynk-strim-attr",
"auto_enums",
"axum",
"bytes",
"clap",
@@ -5227,10 +5270,13 @@ dependencies = [
"expect-test",
"futures",
"http-body",
"hyper",
"hyper-util",
"indexmap 2.13.0",
"itertools 0.14.0",
"libc",
"llm-multimodal",
"openssl",
"prost",
"prost-types",
"rmp-serde",
@@ -5242,8 +5288,11 @@ dependencies = [
"sha2",
"socket2",
"subtle",
"tempfile",
"thiserror-ext",
"tls-listener",
"tokio",
"tokio-openssl",
"tokio-stream",
"tokio-util",
"tonic",
rust/Cargo.toml
+10
View File
@@ -26,6 +26,7 @@ arc-swap = "1.9.0"
async-openai = { version = "0.33.1", default-features = false, features = ["native-tls"] }
async-trait = "0.1.89"
asynk-strim-attr = "0.1.0"
auto_enums = { version = "0.8.9", features = ["tokio1"] }
axum = "0.8.8"
base64 = "0.22.1"
bytemuck = { version = "1.25.0", features = ["extern_crate_alloc"] }
@@ -43,6 +44,12 @@ half = { version = "2.7.1", features = ["bytemuck"] }
hex = "0.4.3"
hf-hub = { version = "0.5.0", default-features = false, features = ["tokio"] }
http-body = "1.0.1"
hyper = { version = "1.10.1", features = ["http1", "server"] }
hyper-util = { version = "0.1.20", features = [
"server-graceful",
"service",
"tokio",
] }
indexmap = "2.13.0"
itertools = "0.14.0"
libc = "0.2.177"
@@ -54,6 +61,7 @@ native-tls-vendored = { package = "native-tls", version = "0.2.18", features = [
ndarray = { version = "0.16.1", features = ["serde"] }
openai-harmony = { package = "oss-harmony", git = "https://github.com/oss-harmony/harmony", tag = "v0.0.11", default-features = false }
openai-protocol = "1.6.0"
openssl = "0.10"
parking_lot = "0.12.5"
paste = "1.0.15"
prometheus-client = "0.24.0"
@@ -89,6 +97,7 @@ thiserror = "2.0.16"
thiserror-ext = "0.3.0"
tiktoken-rs = "0.9.1"
time = { version = "0.3.47", features = ["formatting", "local-offset", "macros"] }
tls-listener = { version = "0.11.2", default-features = false, features = ["openssl", "tokio-net", "axum"] }
tokenizers = "0.22.0"
tokio = { version = "1.47.1", features = [
"macros",
@@ -97,6 +106,7 @@ tokio = { version = "1.47.1", features = [
"sync",
"time",
] }
tokio-openssl = "0.6"
tokio-stream = "0.1"
tokio-util = { version = "0.7.18", features = ["rt"] }
tonic = "0.14.5"
rust/src/cmd/src/cli.rs
+66 -1
View File
@@ -25,7 +25,7 @@ use vllm_managed_engine::ManagedEngineConfig;
use vllm_managed_engine::cli::{ManagedEngineArgs, repartition_managed_engine_args};
use vllm_server::{
ApiServerOptions, ChatTemplateContentFormatOption, Config, CoordinatorMode, CorsConfig,
HttpListenerMode, ParserSelection, RendererSelection,
DEFAULT_KEEP_ALIVE_TIMEOUT, HttpListenerMode, ParserSelection, RendererSelection, TlsConfig,
};
use crate::cli::unsupported::UnsupportedArgs;
@@ -154,6 +154,11 @@ pub struct SharedRuntimeArgs {
#[arg(long, default_value_t = 0)]
#[serde(default)]
pub shutdown_timeout: u64,
/// Maximum idle time (seconds) on a keep-alive HTTP connection before the
/// server closes it (default 5).
#[arg(long = "http-timeout-keep-alive", env = "VLLM_HTTP_TIMEOUT_KEEP_ALIVE")]
#[serde(default)]
pub http_timeout_keep_alive: Option<u64>,
/// The file path to the chat template, or the template in single-line form
/// for the specified model.
@@ -257,6 +262,34 @@ pub struct SharedRuntimeArgs {
#[serde(default)]
pub allow_credentials: bool,
/// The file path to the SSL key file. When omitted, the key is read from
/// `--ssl-certfile` (combined PEM).
#[arg(long)]
#[serde(default)]
pub ssl_keyfile: Option<String>,
/// The file path to the SSL cert file. Enables TLS when set.
#[arg(long)]
#[serde(default)]
pub ssl_certfile: Option<String>,
/// The CA certificates file used to verify client certificates (mTLS).
#[arg(long)]
#[serde(default)]
pub ssl_ca_certs: Option<String>,
/// Whether a client certificate is required: 0 = none, 1 = optional,
/// 2 = required (mirrors Python's `ssl.CERT_*`).
#[arg(long, default_value_t = 0, value_parser = clap::value_parser!(i32).range(0..=2))]
#[serde(default)]
pub ssl_cert_reqs: i32,
/// OpenSSL cipher string for HTTPS (TLS 1.2 and below).
/// When unset, the linked OpenSSL's default suites are used.
#[arg(long)]
#[serde(default)]
pub ssl_ciphers: Option<String>,
/// Unsupported Python vLLM frontend arguments recognized but not yet
/// implemented in Rust.
#[educe(Debug(ignore))]
@@ -277,6 +310,13 @@ impl SharedRuntimeArgs {
Duration::from_secs(self.shutdown_timeout)
}
/// Maximum idle time on a keep-alive HTTP connection before the server
/// closes it.
pub fn keep_alive_timeout(&self) -> Duration {
self.http_timeout_keep_alive
.map_or(DEFAULT_KEEP_ALIVE_TIMEOUT, Duration::from_secs)
}
/// Apply fallback logic for API key configuration from env variables.
fn apply_env_api_key_fallback(&mut self) {
if self.api_key.is_empty()
@@ -301,8 +341,10 @@ impl SharedRuntimeArgs {
) -> Config {
let ready_timeout = self.ready_timeout();
let shutdown_timeout = self.shutdown_timeout();
let keep_alive_timeout = self.keep_alive_timeout();
let api_server_options = self.api_server_options();
let cors = self.cors_config();
let tls = self.tls_config();
Config {
transport_mode: TransportMode::Bootstrapped {
@@ -329,10 +371,12 @@ impl SharedRuntimeArgs {
max_logprobs: self.max_logprobs,
api_server_options,
cors,
tls,
api_keys: self.api_key,
disable_log_stats: self.disable_log_stats,
grpc_port: self.grpc_port,
shutdown_timeout,
keep_alive_timeout,
}
}
@@ -349,8 +393,10 @@ impl SharedRuntimeArgs {
) -> Config {
let ready_timeout = self.ready_timeout();
let shutdown_timeout = self.shutdown_timeout();
let keep_alive_timeout = self.keep_alive_timeout();
let api_server_options = self.api_server_options();
let cors = self.cors_config();
let tls = self.tls_config();
Config {
transport_mode: TransportMode::HandshakeOwner {
@@ -375,10 +421,12 @@ impl SharedRuntimeArgs {
max_logprobs: self.max_logprobs,
api_server_options,
cors,
tls,
api_keys: self.api_key,
disable_log_stats: self.disable_log_stats,
grpc_port: self.grpc_port,
shutdown_timeout,
keep_alive_timeout,
}
}
@@ -398,6 +446,23 @@ impl SharedRuntimeArgs {
allow_credentials: self.allow_credentials,
}
}
/// Build the TLS config: `Some` when any `ssl_*` argument is set, else
/// `None` (plaintext). The combination is validated in [`Config::validate`].
fn tls_config(&self) -> Option<TlsConfig> {
let tls_requested = self.ssl_certfile.is_some()
|| self.ssl_keyfile.is_some()
|| self.ssl_ca_certs.is_some()
|| self.ssl_cert_reqs != 0
|| self.ssl_ciphers.is_some();
tls_requested.then(|| TlsConfig {
cert_file: self.ssl_certfile.clone(),
key_file: self.ssl_keyfile.clone(),
ca_certs: self.ssl_ca_certs.clone(),
cert_reqs: self.ssl_cert_reqs,
ciphers: self.ssl_ciphers.clone(),
})
}
}
fn default_engine_ready_timeout_secs() -> u64 {
rust/src/cmd/src/cli/tests.rs
+167 -9
View File
@@ -41,6 +41,7 @@ fn serve_args_forward_python_flags_with_separator() {
max_logprobs: None,
grpc_port: None,
shutdown_timeout: 0,
http_timeout_keep_alive: None,
chat_template: None,
default_chat_template_kwargs: None,
chat_template_content_format: Auto,
@@ -65,6 +66,11 @@ fn serve_args_forward_python_flags_with_separator() {
],
),
allow_credentials: false,
ssl_keyfile: None,
ssl_certfile: None,
ssl_ca_certs: None,
ssl_cert_reqs: 0,
ssl_ciphers: None,
},
managed_engine: ManagedEngineArgs {
python: "../vllm/.venv/bin/python",
@@ -363,6 +369,140 @@ fn serve_passes_enable_prompt_tokens_details_into_config() {
assert!(config.api_server_options.enable_prompt_tokens_details);
}
#[test]
fn serve_passes_tls_into_config() {
let cli = Cli::try_parse_from([
"vllm-rs",
"serve",
"Qwen/Qwen3-0.6B",
"--ssl-certfile",
"/tmp/cert.pem",
"--ssl-keyfile",
"/tmp/key.pem",
"--ssl-ca-certs",
"/tmp/ca.pem",
"--ssl-cert-reqs",
"2",
])
.unwrap();
let Command::Serve(args) = cli.command else {
panic!("expected serve args");
};
let config = args.to_frontend_config("tcp://127.0.0.1:62100".to_string());
let tls = config.tls.expect("tls configured");
assert_eq!(tls.cert_file.as_deref(), Some("/tmp/cert.pem"));
assert_eq!(tls.key_file.as_deref(), Some("/tmp/key.pem"));
assert_eq!(tls.ca_certs.as_deref(), Some("/tmp/ca.pem"));
assert_eq!(tls.cert_reqs, 2);
}
#[test]
fn serve_without_ssl_flags_has_no_tls() {
let cli = Cli::try_parse_from(["vllm-rs", "serve", "Qwen/Qwen3-0.6B"]).unwrap();
let Command::Serve(args) = cli.command else {
panic!("expected serve args");
};
let config = args.to_frontend_config("tcp://127.0.0.1:62100".to_string());
assert!(config.tls.is_none());
}
#[test]
fn serve_ssl_keyfile_without_certfile_fails_validation() {
let cli = Cli::try_parse_from([
"vllm-rs",
"serve",
"Qwen/Qwen3-0.6B",
"--ssl-keyfile",
"/tmp/key.pem",
])
.unwrap();
let Command::Serve(args) = cli.command else {
panic!("expected serve args");
};
let config = args.to_frontend_config("tcp://127.0.0.1:62100".to_string());
// TLS is requested (a key was given) but there is no certificate, so
// validation fails loud rather than silently serving plaintext.
assert_eq!(config.tls.as_ref().expect("tls requested").cert_file, None);
let err = config.validate().unwrap_err().to_string();
assert!(err.contains("--ssl-certfile is required"), "{err}");
}
#[test]
fn serve_mtls_without_ca_certs_fails_validation() {
let cli = Cli::try_parse_from([
"vllm-rs",
"serve",
"Qwen/Qwen3-0.6B",
"--ssl-certfile",
"/tmp/cert.pem",
"--ssl-cert-reqs",
"2",
])
.unwrap();
let Command::Serve(args) = cli.command else {
panic!("expected serve args");
};
let config = args.to_frontend_config("tcp://127.0.0.1:62100".to_string());
// Client-cert verification without a CA bundle has nothing to verify
// against, so it fails loud at startup.
let err = config.validate().unwrap_err().to_string();
assert!(err.contains("--ssl-ca-certs is required"), "{err}");
}
#[test]
fn frontend_args_json_passes_tls_into_config() {
let cli = Cli::try_parse_from([
"vllm-rs",
"frontend",
"--listen-fd",
"3",
"--input-address",
"ipc:///tmp/input.sock",
"--output-address",
"ipc:///tmp/output.sock",
"--args-json",
r#"{"model_tag":"Qwen/Qwen3-0.6B","ssl_certfile":"/tmp/cert.pem","ssl_keyfile":"/tmp/key.pem"}"#,
])
.unwrap();
let Command::Frontend(args) = cli.command else {
panic!("expected frontend args");
};
let config = args.into_config();
let tls = config.tls.expect("tls configured");
assert_eq!(tls.cert_file.as_deref(), Some("/tmp/cert.pem"));
assert_eq!(tls.key_file.as_deref(), Some("/tmp/key.pem"));
}
#[test]
fn frontend_args_json_rejects_out_of_range_cert_reqs() {
let cli = Cli::try_parse_from([
"vllm-rs",
"frontend",
"--listen-fd",
"3",
"--input-address",
"ipc:///tmp/input.sock",
"--output-address",
"ipc:///tmp/output.sock",
"--args-json",
r#"{"model_tag":"Qwen/Qwen3-0.6B","ssl_certfile":"/tmp/cert.pem","ssl_cert_reqs":5}"#,
])
.unwrap();
let Command::Frontend(args) = cli.command else {
panic!("expected frontend args");
};
// The JSON path bypasses clap's range check, so validate() is the only guard.
let config = args.into_config();
let err = config.validate().unwrap_err().to_string();
assert!(err.contains("--ssl-cert-reqs"), "{err}");
}
#[test]
fn frontend_args_json_passes_enable_request_id_headers_into_config() {
let cli = Cli::try_parse_from([
@@ -481,13 +621,13 @@ fn serve_args_reject_unsupported_flag_arg() {
"vllm-rs",
"serve",
"Qwen/Qwen3-0.6B",
"--ssl-keyfile",
"/tmp/key.pem",
"--root-path",
"/prefix",
])
.unwrap_err();
expect![[r#"
error: invalid value '/tmp/key.pem' for '--ssl-keyfile <SSL_KEYFILE>': argument is not implemented in Rust frontend yet
error: invalid value '/prefix' for '--root-path <ROOT_PATH>': argument is not implemented in Rust frontend yet
Remove this unsupported argument to continue.
@@ -562,6 +702,7 @@ fn frontend_args_accept_json() {
max_logprobs: None,
grpc_port: None,
shutdown_timeout: 0,
http_timeout_keep_alive: None,
chat_template: None,
default_chat_template_kwargs: None,
chat_template_content_format: Auto,
@@ -586,6 +727,11 @@ fn frontend_args_accept_json() {
],
),
allow_credentials: false,
ssl_keyfile: None,
ssl_certfile: None,
ssl_ca_certs: None,
ssl_cert_reqs: 0,
ssl_ciphers: None,
},
},
),
@@ -798,14 +944,14 @@ fn frontend_args_json_rejects_unsupported_fields() {
"--output-address",
"ipc:///tmp/output.sock",
"--args-json",
r#"{"model_tag":"Qwen/Qwen3-0.6B","ssl_keyfile":"/tmp/key.pem"}"#,
r#"{"model_tag":"Qwen/Qwen3-0.6B","root_path":"/prefix"}"#,
])
.unwrap_err();
expect![[r#"
error: invalid value '{"model_tag":"Qwen/Qwen3-0.6B","ssl_keyfile":"/tmp/key.pem"}' for '--args-json <JSON>':
error: invalid value '{"model_tag":"Qwen/Qwen3-0.6B","root_path":"/prefix"}' for '--args-json <JSON>':
The following arguments are not implemented in Rust frontend yet:
- ssl_keyfile
- root_path
Remove these arguments to continue.
@@ -825,16 +971,16 @@ fn frontend_args_json_aggregates_multiple_unsupported_fields() {
"--output-address",
"ipc:///tmp/output.sock",
"--args-json",
r#"{"model_tag":"Qwen/Qwen3-0.6B","response_role":"assistant","ssl_keyfile":"/tmp/key.pem"}"#,
r#"{"model_tag":"Qwen/Qwen3-0.6B","response_role":"assistant","root_path":"/prefix"}"#,
])
.unwrap_err();
let actual = error.to_string().replace(": \n", ":\n");
expect![[r#"
error: invalid value '{"model_tag":"Qwen/Qwen3-0.6B","response_role":"assistant","ssl_keyfile":"/tmp/key.pem"}' for '--args-json <JSON>':
error: invalid value '{"model_tag":"Qwen/Qwen3-0.6B","response_role":"assistant","root_path":"/prefix"}' for '--args-json <JSON>':
The following arguments are not implemented in Rust frontend yet:
- response_role
- ssl_keyfile
- root_path
Remove these arguments to continue.
@@ -1077,6 +1223,7 @@ fn serve_args_accept_handshake_aliases() {
max_logprobs: None,
grpc_port: None,
shutdown_timeout: 0,
http_timeout_keep_alive: None,
chat_template: None,
default_chat_template_kwargs: None,
chat_template_content_format: Auto,
@@ -1101,6 +1248,11 @@ fn serve_args_accept_handshake_aliases() {
],
),
allow_credentials: false,
ssl_keyfile: None,
ssl_certfile: None,
ssl_ca_certs: None,
ssl_cert_reqs: 0,
ssl_ciphers: None,
},
managed_engine: ManagedEngineArgs {
python: "python3",
@@ -1234,10 +1386,12 @@ fn serve_frontend_config_uses_dp_address_as_advertised_host() {
],
allow_credentials: false,
},
tls: None,
api_keys: [],
disable_log_stats: false,
grpc_port: None,
shutdown_timeout: 0ns,
keep_alive_timeout: 5s,
}
"#]]
.assert_debug_eq(&Config {
@@ -1315,10 +1469,12 @@ fn serve_frontend_config_keeps_tcp_transport_for_non_local_only_topology() {
],
allow_credentials: false,
},
tls: None,
api_keys: [],
disable_log_stats: false,
grpc_port: None,
shutdown_timeout: 0ns,
keep_alive_timeout: 5s,
}
"#]]
.assert_debug_eq(&config);
@@ -1414,10 +1570,12 @@ fn frontend_config_uses_external_coordinator_when_coordinator_address_is_present
],
allow_credentials: false,
},
tls: None,
api_keys: [],
disable_log_stats: false,
grpc_port: None,
shutdown_timeout: 0ns,
keep_alive_timeout: 5s,
}
"#]]
.assert_debug_eq(&config);
rust/src/cmd/src/cli/unsupported.rs
-21
View File
@@ -526,18 +526,6 @@ pub struct ServerUnsupportedArgs {
#[arg(long)]
pub disable_access_log_for_endpoints: Option<Noop>,
/// The file path to the SSL key file.
#[arg(long)]
pub ssl_keyfile: Option<Unsupported>,
/// The file path to the SSL cert file.
#[arg(long)]
pub ssl_certfile: Option<Unsupported>,
/// The CA certificates file.
#[arg(long)]
pub ssl_ca_certs: Option<Unsupported>,
/// Refresh SSL Context when SSL certificate files change
#[arg(
long,
@@ -547,15 +535,6 @@ pub struct ServerUnsupportedArgs {
)]
pub enable_ssl_refresh: Option<Unsupported>,
/// Whether client certificate is required (see stdlib ssl module's).
#[arg(long)]
pub ssl_cert_reqs: Option<Unsupported>,
/// SSL cipher suites for HTTPS (TLS 1.2 and below only).
/// Example: 'ECDHE-RSA-AES256-GCM-SHA384:ECDHE-RSA-CHACHA20-POLY1305'
#[arg(long)]
pub ssl_ciphers: Option<Unsupported>,
/// FastAPI root_path when app is behind a path based routing proxy.
#[arg(long)]
pub root_path: Option<Unsupported>,
rust/src/engine-core-client/src/tests/client.rs
+13 -7
View File
@@ -1285,18 +1285,24 @@ async fn dropping_multiple_live_streams_aborts_all_in_a_burst() {
)
.await;
let abort =
timeout(Duration::from_secs(1), recv_engine_message(dealer)).await.unwrap();
assert_eq!(abort[0].as_ref(), &[0x01]);
let ids: Vec<String> = rmp_serde::from_slice(&abort[1]).unwrap();
// Aborts may coalesce into one burst or split across several.
let mut aborted = BTreeSet::new();
while aborted.len() < 3 {
let abort =
timeout(Duration::from_secs(1), recv_engine_message(dealer)).await.unwrap();
assert_eq!(abort[0].as_ref(), &[0x01]);
let ids: Vec<String> = rmp_serde::from_slice(&abort[1]).unwrap();
aborted.extend(ids);
}
assert_eq!(
ids,
vec![
aborted,
BTreeSet::from([
"req-1".to_string(),
"req-2".to_string(),
"req-3".to_string()
]
])
);
// No spurious extra aborts.
assert!(
timeout(Duration::from_millis(100), recv_engine_message(dealer)).await.is_err()
);
rust/src/server/Cargo.toml
+7
View File
@@ -7,14 +7,18 @@ license.workspace = true
[dependencies]
anyhow.workspace = true
asynk-strim-attr.workspace = true
auto_enums.workspace = true
axum.workspace = true
educe.workspace = true
futures.workspace = true
http-body.workspace = true
hyper.workspace = true
hyper-util.workspace = true
indexmap.workspace = true
itertools.workspace = true
libc.workspace = true
llm-multimodal.workspace = true
openssl.workspace = true
prost.workspace = true
prost-types.workspace = true
rmpv.workspace = true
@@ -25,7 +29,9 @@ sha2.workspace = true
socket2.workspace = true
subtle.workspace = true
thiserror-ext.workspace = true
tls-listener.workspace = true
tokio.workspace = true
tokio-openssl.workspace = true
tokio-stream.workspace = true
tokio-util.workspace = true
tonic.workspace = true
@@ -54,6 +60,7 @@ clap.workspace = true
expect-test.workspace = true
rmp-serde.workspace = true
serial_test.workspace = true
tempfile.workspace = true
tower.workspace = true
vllm-engine-core-client = { workspace = true, features = ["test-util"] }
zeromq.workspace = true
rust/src/server/examples/external_engine_openai_qwen.rs
+2
View File
@@ -71,10 +71,12 @@ async fn main() -> Result<()> {
max_logprobs: None,
api_server_options: ApiServerOptions::default(),
cors: CorsConfig::default(),
tls: None,
api_keys: Vec::new(),
disable_log_stats: false,
grpc_port: None,
shutdown_timeout: Duration::ZERO,
keep_alive_timeout: Duration::from_secs(5),
};
let bind_address = format!("127.0.0.1:{port}");
rust/src/server/src/config.rs
+61
View File
@@ -10,6 +10,10 @@ use serde_json::Value;
use vllm_chat::{ChatTemplateContentFormatOption, ParserSelection, RendererSelection};
use vllm_engine_core_client::{CoordinatorMode as EngineCoreCoordinatorMode, TransportMode};
/// Default keep-alive idle timeout (seconds); also the head-read bound
/// when keep-alive is disabled (`0`).
pub const DEFAULT_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(5);
/// How the HTTP server obtains its listening socket.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub enum HttpListenerMode {
@@ -99,6 +103,54 @@ impl CorsConfig {
}
}
/// TLS settings mirroring Python's uvicorn `ssl_*` arguments.
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct TlsConfig {
/// PEM certificate chain file. Required when TLS is configured; may also
/// hold the private key (combined PEM) when `key_file` is unset.
pub cert_file: Option<String>,
/// PEM private key file. When `None`, the key is read from `cert_file`
/// (combined PEM).
pub key_file: Option<String>,
/// PEM CA bundle used to verify client certificates (mTLS). Required when
/// `cert_reqs` is non-zero.
pub ca_certs: Option<String>,
/// Client-certificate requirement, mirroring Python's `ssl.CERT_*`:
/// 0 = none, 1 = optional, 2 = required.
pub cert_reqs: i32,
/// OpenSSL cipher string for TLS 1.2 and below, mirroring Python's
/// `ssl.set_ciphers`. `None` keeps the forward-secret AEAD default.
pub ciphers: Option<String>,
}
impl TlsConfig {
/// Structurally validate the TLS arguments; the cert/key material is parsed
/// later, when the OpenSSL context is built.
pub fn validate(&self) -> Result<()> {
if self.cert_file.is_none() {
bail!(
"--ssl-certfile is required to enable TLS; \
--ssl-keyfile/--ssl-ca-certs/--ssl-cert-reqs/--ssl-ciphers \
cannot be used without it"
);
}
if !matches!(self.cert_reqs, 0..=2) {
bail!(
"--ssl-cert-reqs must be 0 (none), 1 (optional), or 2 (required), got {}",
self.cert_reqs
);
}
if self.cert_reqs != 0 && self.ca_certs.is_none() {
bail!(
"--ssl-ca-certs is required when --ssl-cert-reqs is {} \
(client certificate verification)",
self.cert_reqs
);
}
Ok(())
}
}
/// Normalized runtime configuration for the minimal OpenAI-compatible server.
#[derive(Educe, Clone, PartialEq, Eq, Serialize)]
#[educe(Debug)]
@@ -138,6 +190,9 @@ pub struct Config {
pub api_server_options: ApiServerOptions,
/// CORS settings applied to every HTTP response.
pub cors: CorsConfig,
/// TLS settings. `None` serves plaintext HTTP; `Some` terminates TLS at the
/// listener.
pub tls: Option<TlsConfig>,
/// API keys accepted as bearer tokens for guarded routes.
#[serde(skip_serializing)]
#[educe(Debug(method(fmt_redacted_api_keys)))]
@@ -150,6 +205,9 @@ pub struct Config {
pub grpc_port: Option<u16>,
/// Maximum time to wait for active HTTP/gRPC requests to drain on shutdown.
pub shutdown_timeout: Duration,
/// Maximum idle time on a keep-alive HTTP connection before the server
/// closes it (`VLLM_HTTP_TIMEOUT_KEEP_ALIVE`, default 5s).
pub keep_alive_timeout: Duration,
}
impl Config {
@@ -158,6 +216,9 @@ impl Config {
pub fn validate(&self) -> Result<()> {
vllm_chat::validate_parser_overrides(&self.tool_call_parser, &self.reasoning_parser)?;
self.cors.validate()?;
if let Some(tls) = &self.tls {
tls.validate()?;
}
if let Some(max_logprobs) = self.max_logprobs
&& max_logprobs < -1
{
rust/src/server/src/grpc/mod.rs
+78 -1
View File
@@ -4,16 +4,21 @@ mod convert;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use futures::{Stream, StreamExt as _};
use futures::{Stream, StreamExt as _, stream};
use thiserror_ext::AsReport as _;
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio::sync::mpsc;
use tokio_openssl::SslStream;
use tokio_stream::wrappers::ReceiverStream;
use tonic::transport::server::{Connected, TcpConnectInfo};
use tonic::{Request, Response, Status};
use tracing::info;
use vllm_text::{DecodedTextEvent, TextOutputStreamExt as _};
use self::convert::ResponseOpts;
use crate::listener::{Listener, ListenerIo};
use crate::state::AppState;
/// Generated protobuf/gRPC types for the `vllm` package.
@@ -26,6 +31,78 @@ pub use pb::generate_server::GenerateServer;
#[cfg(test)]
mod tests;
/// Newtype over `tokio-openssl`'s `SslStream` so we can implement tonic's
/// [`Connected`] on it (the orphan rule blocks doing so on the foreign type).
pub(crate) struct GrpcTlsStream {
inner: SslStream<ListenerIo>,
}
impl GrpcTlsStream {
pub(crate) fn new(inner: SslStream<ListenerIo>) -> Self {
Self { inner }
}
}
impl AsyncRead for GrpcTlsStream {
fn poll_read(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
Pin::new(&mut self.get_mut().inner).poll_read(cx, buf)
}
}
impl AsyncWrite for GrpcTlsStream {
fn poll_write(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &[u8],
) -> Poll<std::io::Result<usize>> {
Pin::new(&mut self.get_mut().inner).poll_write(cx, buf)
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Pin::new(&mut self.get_mut().inner).poll_flush(cx)
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Pin::new(&mut self.get_mut().inner).poll_shutdown(cx)
}
}
impl Connected for GrpcTlsStream {
type ConnectInfo = TcpConnectInfo;
fn connect_info(&self) -> TcpConnectInfo {
self.inner.get_ref().connect_info()
}
}
/// Adapt the shared server listener into tonic's incoming stream shape.
pub(crate) fn incoming(listener: Listener) -> impl Stream<Item = std::io::Result<ListenerIo>> {
stream::unfold(listener, |mut listener| async move {
let (io, _) = axum::serve::Listener::accept(&mut listener).await;
Some((Ok(io), listener))
})
}
/// Wrap the gRPC listener so each accepted connection completes a TLS handshake
/// before tonic serves it.
pub(crate) fn tls_incoming(
listener: Listener,
context: openssl::ssl::SslContext,
handshake_timeout: std::time::Duration,
) -> impl Stream<Item = std::io::Result<GrpcTlsStream>> {
tls_listener::builder(context)
.handshake_timeout(handshake_timeout)
.listen(listener)
.map(|res| {
res.map(|(inner, _addr)| GrpcTlsStream::new(inner))
.map_err(std::io::Error::other)
})
}
/// gRPC Generate service implementation backed by the shared application state.
pub struct GenerateServiceImpl {
state: Arc<AppState>,
rust/src/server/src/grpc/tests.rs
+357 -15
View File
@@ -1,11 +1,19 @@
use std::future::Future;
use std::io;
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use std::time::Duration;
use futures::StreamExt as _;
use hyper_util::rt::TokioIo;
use openssl::ssl::{SslConnector, SslFiletype, SslMethod};
use serial_test::serial;
use tonic::transport::Server as TonicServer;
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
use tokio::net::TcpStream;
use tokio_openssl::SslStream;
use tonic::transport::{Channel, Endpoint, Server as TonicServer, Uri};
use tower::service_fn;
use vllm_chat::{
ChatBackend, ChatLlm, ChatRenderer, ChatRequest, ChatTextBackend, DefaultChatOutputProcessor,
DynChatOutputProcessor, DynChatRenderer, NewChatOutputProcessorOptions, RenderedPrompt,
@@ -22,8 +30,11 @@ use zeromq::prelude::{SocketRecv, SocketSend};
use zeromq::{DealerSocket, PushSocket, ZmqMessage};
use super::pb::generate_client::GenerateClient;
use super::{GenerateServer, GenerateServiceImpl, pb};
use super::{GenerateServer, GenerateServiceImpl, incoming, pb, tls_incoming};
use crate::listener::Listener;
use crate::state::AppState;
use crate::tls;
use crate::tls_tests::{TestCerts, server_tls};
// ========================================================================================
// Helpers (mirrors the patterns in routes/tests.rs)
@@ -211,17 +222,12 @@ impl ChatRenderer for FakeTextBackend {
}
}
/// Spin up a gRPC server backed by a mock engine that serves a single request
/// with the given output specs. Returns the client, the gRPC server task, and
/// the mock engine task.
async fn grpc_test_server(
/// Build the gRPC service + mock engine that serves a single request with the
/// given output specs. Shared by the plaintext and TLS server fixtures.
async fn setup_grpc_service(
engine_id: impl Into<EngineId>,
output_specs: Vec<(Vec<u32>, Option<EngineCoreFinishReason>)>,
) -> (
GenerateClient<tonic::transport::Channel>,
tokio::task::JoinHandle<()>,
MockEngineTask,
) {
) -> (GenerateServer<GenerateServiceImpl>, MockEngineTask) {
let ipc = IpcNamespace::new().expect("create ipc namespace");
let handshake_address = ipc.handshake_endpoint();
let engine_id = engine_id.into();
@@ -259,14 +265,29 @@ async fn grpc_test_server(
Arc::new(FakeTextBackend) as Arc<dyn ChatTextBackend>,
);
let state = Arc::new(AppState::new(vec!["test-model".to_string()], chat));
let svc = GenerateServer::new(GenerateServiceImpl::new(state));
(
GenerateServer::new(GenerateServiceImpl::new(state)),
engine_task,
)
}
/// Spin up a plaintext gRPC server backed by a mock engine. Returns the client,
/// the gRPC server task, and the mock engine task.
async fn grpc_test_server(
engine_id: impl Into<EngineId>,
output_specs: Vec<(Vec<u32>, Option<EngineCoreFinishReason>)>,
) -> (
GenerateClient<tonic::transport::Channel>,
tokio::task::JoinHandle<()>,
MockEngineTask,
) {
let (svc, engine_task) = setup_grpc_service(engine_id, output_specs).await;
// Bind to an OS-assigned port.
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind grpc listener");
let addr = listener.local_addr().expect("local addr");
let server_task = tokio::spawn(async move {
let incoming = tokio_stream::wrappers::TcpListenerStream::new(listener);
let incoming = incoming(Listener::Tcp(listener));
TonicServer::builder()
.add_service(svc)
.serve_with_incoming(incoming)
@@ -274,7 +295,6 @@ async fn grpc_test_server(
.expect("grpc server");
});
// Connect the client.
let grpc_client = GenerateClient::connect(format!("http://{addr}"))
.await
.expect("connect grpc client");
@@ -282,6 +302,158 @@ async fn grpc_test_server(
(grpc_client, server_task, engine_task)
}
/// Spin up a TLS gRPC server (server cert from `certs`, `cert_reqs` mTLS mode).
/// Returns the address, the server task, and the mock engine task.
async fn grpc_tls_test_server(
engine_id: impl Into<EngineId>,
output_specs: Vec<(Vec<u32>, Option<EngineCoreFinishReason>)>,
certs: &TestCerts,
cert_reqs: i32,
) -> (String, tokio::task::JoinHandle<()>, MockEngineTask) {
let (svc, engine_task) = setup_grpc_service(engine_id, output_specs).await;
let context = tls::build_grpc_server_config(&server_tls(certs, cert_reqs))
.expect("build grpc tls config");
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind grpc listener");
let addr = listener.local_addr().expect("local addr").to_string();
let server_task = tokio::spawn(async move {
let incoming = tls_incoming(Listener::Tcp(listener), context, tls::TLS_HANDSHAKE_TIMEOUT);
TonicServer::builder()
.add_service(svc)
.serve_with_incoming(incoming)
.await
.expect("grpc tls server");
});
(addr, server_task, engine_task)
}
/// Build a tonic `Generate` client over a tokio-openssl connector, optionally
/// with a client identity for mTLS. Hand-rolled because tonic 0.14 ships no
/// OpenSSL transport.
async fn grpc_tls_client(
certs: &TestCerts,
addr: &str,
identity: Option<&str>,
) -> Result<GenerateClient<Channel>, tonic::transport::Error> {
let ca = certs.path("ca.pem");
let identity = identity.map(|name| {
(
certs.path(&format!("{name}.pem")),
certs.path(&format!("{name}.key")),
)
});
let target = addr.to_string();
let connector = service_fn(move |_: Uri| {
let ca = ca.clone();
let identity = identity.clone();
let target = target.clone();
async move {
let tcp = TcpStream::connect(&target).await?;
let mut builder =
SslConnector::builder(SslMethod::tls_client()).map_err(io::Error::other)?;
builder.set_ca_file(&ca).map_err(io::Error::other)?;
if let Some((cert, key)) = &identity {
builder.set_certificate_chain_file(cert).map_err(io::Error::other)?;
builder.set_private_key_file(key, SslFiletype::PEM).map_err(io::Error::other)?;
}
let mut config = builder.build().configure().map_err(io::Error::other)?;
config.set_verify_hostname(false);
config.set_alpn_protos(b"\x02h2").map_err(io::Error::other)?;
let ssl = config.into_ssl("127.0.0.1").map_err(io::Error::other)?;
let mut stream = SslStream::new(ssl, tcp).map_err(io::Error::other)?;
Pin::new(&mut stream).connect().await.map_err(io::Error::other)?;
Ok::<_, io::Error>(TokioIo::new(stream))
}
});
let channel = Endpoint::from_shared(format!("https://{addr}"))
.expect("grpc endpoint")
.connect_with_connector(connector)
.await?;
Ok(GenerateClient::new(channel))
}
/// Complete a raw TLS handshake against the gRPC port (offering ALPN `h2`) for
/// the ALPN-negotiation assertion.
async fn grpc_tls_handshake(
certs: &TestCerts,
addr: &str,
) -> io::Result<Pin<Box<SslStream<TcpStream>>>> {
let tcp = TcpStream::connect(addr).await?;
let mut builder = SslConnector::builder(SslMethod::tls_client()).map_err(io::Error::other)?;
builder.set_ca_file(certs.path("ca.pem")).map_err(io::Error::other)?;
let mut config = builder.build().configure().map_err(io::Error::other)?;
config.set_verify_hostname(false);
config.set_alpn_protos(b"\x02h2").map_err(io::Error::other)?;
let ssl = config.into_ssl("127.0.0.1").map_err(io::Error::other)?;
let mut stream = Box::pin(SslStream::new(ssl, tcp).map_err(io::Error::other)?);
stream.as_mut().connect().await.map_err(io::Error::other)?;
Ok(stream)
}
/// Spin up a plaintext gRPC server, optionally with HTTP/2 keepalive set to
/// `keepalive` for both the PING interval and the unanswered-PING timeout.
async fn grpc_server_with_keepalive(
engine_id: impl Into<EngineId>,
keepalive: Option<Duration>,
) -> (String, tokio::task::JoinHandle<()>, MockEngineTask) {
let (svc, engine_task) = setup_grpc_service(engine_id, default_stream_output_specs()).await;
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind grpc listener");
let addr = listener.local_addr().expect("local addr").to_string();
let mut builder = TonicServer::builder();
if let Some(interval) = keepalive {
builder = builder
.http2_keepalive_interval(Some(interval))
.http2_keepalive_timeout(Some(interval));
}
let server_task = tokio::spawn(async move {
let incoming = incoming(Listener::Tcp(listener));
builder
.add_service(svc)
.serve_with_incoming(incoming)
.await
.expect("grpc server");
});
(addr, server_task, engine_task)
}
/// Establish an HTTP/2 connection (preface + SETTINGS exchange) then go silent,
/// ACKing the server's SETTINGS but never its keepalive PINGs. Returns whether
/// the SERVER closes the connection within `wait`. A minimal hand-rolled h2 peer
/// because a real client auto-ACKs PINGs and so can never be kept-alive-evicted.
async fn h2_unresponsive_peer_closed_within(addr: &str, wait: Duration) -> bool {
let mut tcp = TcpStream::connect(addr).await.expect("connect");
tcp.write_all(b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n").await.expect("preface");
tcp.write_all(&[0, 0, 0, 0x4, 0, 0, 0, 0, 0]).await.expect("client settings");
let closed = tokio::time::timeout(wait, async {
let mut header = [0u8; 9];
while tcp.read_exact(&mut header).await.is_ok() {
let len = u32::from_be_bytes([0, header[0], header[1], header[2]]) as usize;
let frame_type = header[3];
let flags = header[4];
let mut payload = vec![0u8; len];
if tcp.read_exact(&mut payload).await.is_err() {
return;
}
// ACK the server's SETTINGS so the only thing left unanswered is PINGs.
if frame_type == 0x4 && flags & 0x1 == 0 {
let _ = tcp.write_all(&[0, 0, 0, 0x4, 0x1, 0, 0, 0, 0]).await;
}
}
})
.await;
closed.is_ok()
}
// ========================================================================================
// Tests
// ========================================================================================
@@ -720,3 +892,173 @@ async fn unary_generate_output_text_defaults_to_true() {
engine_task.await.expect("mock engine task");
server_task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn grpc_generate_succeeds_over_tls() {
let certs = TestCerts::generate();
let (addr, server_task, engine_task) = grpc_tls_test_server(
b"engine-grpc-tls-unary",
default_stream_output_specs(),
&certs,
0,
)
.await;
let mut client = grpc_tls_client(&certs, &addr, None).await.expect("tls client");
let response = client
.generate(pb::GenerateRequest {
request_id: "test-tls-unary".to_string(),
model: "test-model".to_string(),
prompt: Some(pb::generate_request::Prompt::Text("hello".to_string())),
stopping: Some(pb::StoppingCriteria {
max_new_tokens: 10,
..Default::default()
}),
response: Some(pb::ResponseOptions {
output_text: Some(true),
..Default::default()
}),
..Default::default()
})
.await
.expect("unary generate over tls")
.into_inner();
assert_eq!(response.outputs.expect("outputs present").text, "hi");
engine_task.await.expect("mock engine task");
server_task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn grpc_tls_negotiates_h2_alpn() {
let certs = TestCerts::generate();
let (addr, server_task, _engine_task) = grpc_tls_test_server(
b"engine-grpc-tls-alpn",
default_stream_output_specs(),
&certs,
0,
)
.await;
let stream = grpc_tls_handshake(&certs, &addr).await.expect("handshake");
assert_eq!(
stream.ssl().selected_alpn_protocol(),
Some(&b"h2"[..]),
"server must negotiate h2 ALPN"
);
server_task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn grpc_mtls_required_rejects_client_without_certificate() {
let certs = TestCerts::generate();
let (addr, server_task, _engine_task) = grpc_tls_test_server(
b"engine-grpc-tls-mtls-reject",
default_stream_output_specs(),
&certs,
2,
)
.await;
// With TLS 1.3 the missing-client-cert rejection surfaces on first use, not
// at the handshake, so drive an RPC and assert the call fails.
let outcome = match grpc_tls_client(&certs, &addr, None).await {
Err(_) => Err(()),
Ok(mut client) => client
.generate(pb::GenerateRequest {
request_id: "test-tls-mtls-reject".to_string(),
model: "test-model".to_string(),
prompt: Some(pb::generate_request::Prompt::Text("hello".to_string())),
stopping: Some(pb::StoppingCriteria {
max_new_tokens: 10,
..Default::default()
}),
..Default::default()
})
.await
.map(|_| ())
.map_err(|_| ()),
};
assert!(
outcome.is_err(),
"mTLS-required gRPC must reject a client without a certificate"
);
server_task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn grpc_mtls_required_accepts_valid_client_certificate() {
let certs = TestCerts::generate();
let (addr, server_task, engine_task) = grpc_tls_test_server(
b"engine-grpc-tls-mtls-accept",
default_stream_output_specs(),
&certs,
2,
)
.await;
let mut client = grpc_tls_client(&certs, &addr, Some("client")).await.expect("mtls client");
let response = client
.generate(pb::GenerateRequest {
request_id: "test-tls-mtls".to_string(),
model: "test-model".to_string(),
prompt: Some(pb::generate_request::Prompt::Text("hello".to_string())),
stopping: Some(pb::StoppingCriteria {
max_new_tokens: 10,
..Default::default()
}),
response: Some(pb::ResponseOptions {
output_text: Some(true),
..Default::default()
}),
..Default::default()
})
.await
.expect("mtls generate over tls")
.into_inner();
assert_eq!(response.outputs.expect("outputs present").text, "hi");
engine_task.await.expect("mock engine task");
server_task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn grpc_keepalive_closes_unresponsive_connection() {
let (addr, server_task, _engine_task) =
grpc_server_with_keepalive(b"engine-grpc-keepalive", Some(Duration::from_millis(150)))
.await;
let closed = h2_unresponsive_peer_closed_within(&addr, Duration::from_secs(5)).await;
assert!(
closed,
"keepalive must close a peer that stops answering PINGs"
);
server_task.abort();
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
#[serial]
async fn grpc_without_keepalive_keeps_unresponsive_connection_open() {
// Without keepalive the same unresponsive peer is NOT
// closed, proving the close above is attributable to keepalive.
let (addr, server_task, _engine_task) =
grpc_server_with_keepalive(b"engine-grpc-no-keepalive", None).await;
let closed = h2_unresponsive_peer_closed_within(&addr, Duration::from_secs(1)).await;
assert!(
!closed,
"without keepalive an idle h2 connection must stay open"
);
server_task.abort();
}
rust/src/server/src/lib.rs
+205 -35
View File
@@ -10,20 +10,34 @@ mod routes;
mod runtime;
mod server_info;
mod state;
mod tls;
#[cfg(test)]
mod tls_tests;
mod utils;
use std::future::Future;
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use anyhow::{Context as _, Result};
use axum::Router;
use axum::serve::ListenerExt as _;
pub use config::{ApiServerOptions, Config, CoordinatorMode, CorsConfig, HttpListenerMode};
use axum::body::Body;
use axum::http::Request;
pub use config::{
ApiServerOptions, Config, CoordinatorMode, CorsConfig, DEFAULT_KEEP_ALIVE_TIMEOUT,
HttpListenerMode, TlsConfig,
};
use futures::FutureExt as _;
use hyper::body::Incoming;
use hyper::server::conn::http1;
use hyper_util::rt::{TokioIo, TokioTimer};
use hyper_util::server::graceful::GracefulShutdown;
use hyper_util::service::TowerToHyperService;
use tokio::net::TcpListener;
use tokio::time::{Instant, sleep_until};
use tokio_stream::wrappers::TcpListenerStream;
use tokio_util::either::Either;
use tokio_util::sync::CancellationToken;
use tonic::transport::Server as TonicServer;
use tower::ServiceExt as _;
use tracing::{info, trace, warn};
use vllm_chat::{ChatLlm, LoadModelBackendsOptions, load_model_backends};
pub use vllm_chat::{ChatTemplateContentFormatOption, ParserSelection, RendererSelection};
@@ -36,6 +50,13 @@ use crate::routes::build_router;
use crate::server_info::ServerInfoSnapshot;
use crate::state::AppState;
/// How often the server PINGs an idle gRPC connection to reap a dead peer;
/// tonic enables no keepalive by default. 2h matches the gRPC-core default.
const GRPC_KEEPALIVE_INTERVAL: Duration = Duration::from_secs(7200);
/// How long the server waits for a keepalive PING reply before dropping the gRPC
/// connection. 20s matches the gRPC-core default.
const GRPC_KEEPALIVE_TIMEOUT: Duration = Duration::from_secs(20);
/// Resolve the public model names accepted by the frontend.
fn effective_served_model_names(model: &str, served_model_name: &[String]) -> Vec<String> {
if served_model_name.is_empty() {
@@ -45,6 +66,17 @@ fn effective_served_model_names(model: &str, served_model_name: &[String]) -> Ve
}
}
/// Choose the gRPC listener host. It follows the HTTP TCP host when there is
/// one; otherwise (unix socket or inherited fd) it defaults to IPv4 loopback
/// rather than all interfaces, so the side-car is never accidentally
/// network-exposed.
fn grpc_bind_host(listener_mode: &HttpListenerMode) -> &str {
match listener_mode {
HttpListenerMode::BindTcp { host, .. } => host.as_str(),
HttpListenerMode::BindUnix { .. } | HttpListenerMode::InheritedFd { .. } => "127.0.0.1",
}
}
/// Build the shared application state for one configured model and one engine
/// client.
async fn build_state(config: &Config) -> Result<Arc<AppState>> {
@@ -130,6 +162,15 @@ where
{
config.validate().context("invalid OpenAI frontend configuration")?;
// Build the TLS server config once, up front, so a bad cert/key fails fast
// before the (potentially long) engine handshake.
let tls_config = config
.tls
.as_ref()
.map(tls::build_server_config)
.transpose()
.context("invalid TLS configuration")?;
// Also check shutdown during the (potentially long) startup handshake.
let state = tokio::select! {
result = build_state(&config) => result?,
@@ -144,40 +185,39 @@ where
// Optionally bind the gRPC Generate server on a separate port. Bind
// synchronously here so bind errors (port in use, permission denied, ...)
// surface before we start serving, rather than being deferred until
// shutdown. The gRPC listener follows the same host as the HTTP listener so
// that enabling --grpc-port does not accidentally expose the service on all
// interfaces when HTTP is intentionally local-only.
// surface before serving rather than being deferred until shutdown.
let grpc_setup = if let Some(grpc_port) = config.grpc_port {
let grpc_host = match &config.listener_mode {
HttpListenerMode::BindTcp { host, .. } => host.as_str(),
HttpListenerMode::BindUnix { .. } | HttpListenerMode::InheritedFd { .. } => "0.0.0.0",
};
let grpc_host = grpc_bind_host(&config.listener_mode);
let grpc_listener = TcpListener::bind((grpc_host, grpc_port))
.await
.with_context(|| format!("failed to bind gRPC listener on {grpc_host}:{grpc_port}"))?;
let addr = grpc_listener.local_addr()?;
let grpc_listener = Listener::Tcp(grpc_listener);
// gRPC reuses the HTTP TLS config (same SslContext) plus ALPN h2.
let grpc_tls = config
.tls
.as_ref()
.map(tls::build_grpc_server_config)
.transpose()
.context("invalid gRPC TLS configuration")?;
let svc = grpc::GenerateServer::new(grpc::GenerateServiceImpl::new(state.clone()));
let svc = TonicServer::builder()
.http2_keepalive_interval(Some(GRPC_KEEPALIVE_INTERVAL))
.http2_keepalive_timeout(Some(GRPC_KEEPALIVE_TIMEOUT))
.layer(middleware::request_runtime_layer(state.clone()))
.add_service(svc);
info!(%addr, "starting gRPC server");
Some((grpc_listener, svc))
info!(%addr, tls = grpc_tls.is_some(), "starting gRPC server");
Some((grpc_listener, svc, grpc_tls))
} else {
None
};
info!(%bind_address, %model, "starting OpenAI server");
// Set TCP_NODELAY on accepted connections to reduce latency.
// By `tap_io` we will do this on every accepted connection.
let listener = listener.tap_io(|io| {
if let Either::Left(tcp_stream) = io
&& let Err(err) = tcp_stream.set_nodelay(true)
{
trace!(error = %err, "failed to enable TCP_NODELAY on accepted HTTP connection");
}
});
let scheme = if tls_config.is_some() {
"https"
} else {
"http"
};
info!(%bind_address, %scheme, %model, "starting OpenAI server");
// Run HTTP and gRPC concurrently under a child token of the caller's shutdown
// token. Caller cancellation propagates into both protocols; if either
@@ -208,17 +248,27 @@ where
}
});
// 0 disables keep-alive but still bounds the head read (default), so a
// silent client cannot hold the connection open.
let keep_alive_timeout = config.keep_alive_timeout;
let timeouts = ConnectionTimeouts {
handshake: tls::TLS_HANDSHAKE_TIMEOUT,
header_read: if keep_alive_timeout.is_zero() {
DEFAULT_KEEP_ALIVE_TIMEOUT
} else {
keep_alive_timeout
},
keep_alive_enabled: !keep_alive_timeout.is_zero(),
};
let http_fut = {
let shutdown = server_shutdown.child_token();
let server_shutdown = server_shutdown.clone();
let force_shutdown = force_shutdown.clone();
async move {
let server =
axum::serve(listener, app).with_graceful_shutdown(shutdown.cancelled_owned());
let result = tokio::select! {
result = server => {
result.context("HTTP server failed")
result = serve_listener(listener, tls_config, app, shutdown.cancelled_owned(), timeouts) => {
result
}
_ = force_shutdown.cancelled() => {
warn!("HTTP graceful shutdown deadline elapsed; aborting server");
@@ -236,16 +286,24 @@ where
let server_shutdown = server_shutdown.clone();
let force_shutdown = force_shutdown.clone();
async move {
let Some((grpc_listener, svc)) = grpc_setup else {
let Some((grpc_listener, svc, grpc_tls)) = grpc_setup else {
// No gRPC configured: just wait for shutdown so we do not race the
// join! by resolving early and tripping the cancellation token.
shutdown.cancelled().await;
return Ok(());
};
let server = svc.serve_with_incoming_shutdown(
TcpListenerStream::new(grpc_listener),
shutdown.cancelled_owned(),
);
// Box to unify the TLS and plaintext arms' different stream types.
let server = match grpc_tls {
Some(context) => {
let incoming =
grpc::tls_incoming(grpc_listener, context, tls::TLS_HANDSHAKE_TIMEOUT);
svc.serve_with_incoming_shutdown(incoming, shutdown.cancelled_owned()).boxed()
}
None => {
let incoming = grpc::incoming(grpc_listener);
svc.serve_with_incoming_shutdown(incoming, shutdown.cancelled_owned()).boxed()
}
};
let result = tokio::select! {
result = server => {
@@ -272,6 +330,99 @@ where
state.shutdown(shutdown_deadline).await
}
/// Per-connection timeouts applied while serving HTTP/HTTPS.
#[derive(Clone, Copy)]
pub(crate) struct ConnectionTimeouts {
/// Max time for a client to complete the TLS handshake (TLS path only).
pub(crate) handshake: Duration,
/// HTTP/1 header-read timeout (bounds idle keep-alive and the head read).
pub(crate) header_read: Duration,
/// Whether HTTP/1 keep-alive is enabled; `false` closes after each response.
pub(crate) keep_alive_enabled: bool,
}
/// Apply optional TLS termination and per-connection HTTP timeouts, then serve
/// `app`. Shared by [`serve_with_router_extension`] and the TLS tests.
async fn serve_listener(
listener: Listener,
tls: Option<openssl::ssl::SslContext>,
app: Router,
shutdown: impl Future<Output = ()> + Send + 'static,
timeouts: ConnectionTimeouts,
) -> Result<()> {
match tls {
Some(context) => {
// tls-listener terminates TLS (handshake + timeout); serve_connections
// owns the HTTP keep-alive/idle bound that axum::serve cannot express.
// Failed handshakes (incl. timeouts) log at ERROR via tls-listener.
let listener = tls_listener::builder(context)
.handshake_timeout(timeouts.handshake)
.listen(listener);
serve_connections(
listener,
app,
shutdown,
timeouts.header_read,
timeouts.keep_alive_enabled,
)
.await
.context("HTTPS server failed")
}
None => serve_connections(
listener,
app,
shutdown,
timeouts.header_read,
timeouts.keep_alive_enabled,
)
.await
.context("HTTP server failed"),
}
}
/// Serve `app` per connection (HTTP/1) with a keep-alive idle timeout and
/// graceful drain. Hand-rolled on hyper because [`axum::serve()`] takes no config.
async fn serve_connections<L>(
mut listener: L,
app: Router,
shutdown: impl Future<Output = ()> + Send,
header_read: Duration,
keep_alive_enabled: bool,
) -> Result<()>
where
L: axum::serve::Listener,
{
let graceful = GracefulShutdown::new();
let mut shutdown = std::pin::pin!(shutdown);
loop {
let (io, _addr) = tokio::select! {
conn = listener.accept() => conn,
() = &mut shutdown => break,
};
let service = TowerToHyperService::new(
app.clone().map_request(|req: Request<Incoming>| req.map(Body::new)),
);
let mut builder = http1::Builder::new();
builder.timer(TokioTimer::new()).header_read_timeout(header_read);
if !keep_alive_enabled {
builder.keep_alive(false);
}
let connection = builder.serve_connection(TokioIo::new(io), service);
let connection = graceful.watch(connection);
tokio::spawn(async move {
if let Err(err) = connection.await {
trace!(error = %err, "failed to serve connection");
}
});
}
drop(listener);
graceful.shutdown().await;
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
@@ -293,4 +444,23 @@ mod tests {
served_names
);
}
#[test]
fn grpc_bind_host_follows_http_tcp_host() {
let mode = HttpListenerMode::BindTcp {
host: "0.0.0.0".to_string(),
port: 8000,
};
assert_eq!(grpc_bind_host(&mode), "0.0.0.0");
}
#[test]
fn grpc_bind_host_defaults_to_loopback_without_tcp_host() {
let unix = HttpListenerMode::BindUnix {
path: "/tmp/vllm.sock".to_string(),
};
let inherited = HttpListenerMode::InheritedFd { fd: 3 };
assert_eq!(grpc_bind_host(&unix), "127.0.0.1");
assert_eq!(grpc_bind_host(&inherited), "127.0.0.1");
}
}
rust/src/server/src/listener.rs
+96 -14
View File
@@ -1,28 +1,49 @@
//! Unified HTTP listener wrapper for the Rust frontend.
//! Unified listener wrapper for the Rust frontend.
//!
//! This module hides the difference between TCP and Unix-domain listeners so
//! the rest of the server can bind or inherit one socket and pass it to
//! `axum::serve(...)` through a single type.
use std::io::Result;
use std::net::TcpListener as StdTcpListener;
use std::net::{SocketAddr, TcpListener as StdTcpListener};
use std::os::fd::{FromRawFd, IntoRawFd, OwnedFd};
use std::os::unix::net::UnixListener as StdUnixListener;
use std::pin::Pin;
use std::task::{Context, Poll, ready};
use auto_enums::enum_derive;
use socket2::Socket;
use tls_listener::{AsyncAccept, AsyncListener};
use tokio::net::{TcpListener, TcpStream, UnixListener, UnixStream};
use tokio_util::either::Either;
use tonic::transport::server::{Connected, TcpConnectInfo};
use tracing::trace;
use crate::HttpListenerMode;
/// Runtime listener type used by the OpenAI-compatible HTTP server, which is
/// either a TCP listener or a Unix-domain listener.
/// Runtime listener type used by the OpenAI-compatible HTTP or gRPC server,
/// which is either a TCP listener or a Unix-domain listener.
#[derive(Debug)]
pub enum Listener {
Tcp(TcpListener),
Unix(UnixListener),
}
/// Runtime listener I/O type which is either a TCP stream or a Unix-domain stream.
#[derive(Debug)]
#[enum_derive(tokio1::AsyncRead, tokio1::AsyncWrite)]
pub enum ListenerIo {
Tcp(TcpStream),
Unix(UnixStream),
}
/// Runtime listener address type which is either a TCP address or a Unix-domain address.
#[derive(Debug)]
#[allow(dead_code)]
pub enum ListenerAddr {
Tcp(SocketAddr),
Unix(tokio::net::unix::SocketAddr),
}
impl Listener {
/// Bind or adopt the listener described by the frontend configuration.
///
@@ -70,34 +91,95 @@ impl Listener {
Ok(Self::Tcp(TcpListener::from_std(std_listener)?))
}
}
fn listener_addr(&self) -> Result<ListenerAddr> {
match self {
Self::Tcp(listener) => listener.local_addr().map(ListenerAddr::Tcp),
Self::Unix(listener) => listener.local_addr().map(ListenerAddr::Unix),
}
}
}
impl Connected for ListenerIo {
type ConnectInfo = TcpConnectInfo;
fn connect_info(&self) -> TcpConnectInfo {
match self {
Self::Tcp(stream) => stream.connect_info(),
Self::Unix(_) => TcpConnectInfo {
local_addr: None,
remote_addr: None,
},
}
}
}
/// Attempt to set `TCP_NODELAY` on the accepted TCP stream.
fn enable_tcp_nodelay(stream: TcpStream) -> TcpStream {
if let Err(err) = stream.set_nodelay(true) {
trace!(error = %err, "failed to enable TCP_NODELAY on accepted TCP connection");
}
stream
}
/// Allow the unified listener to plug directly into `axum::serve(...)`.
impl axum::serve::Listener for Listener {
type Addr = Either<std::net::SocketAddr, tokio::net::unix::SocketAddr>;
type Io = Either<TcpStream, UnixStream>;
type Addr = ListenerAddr;
type Io = ListenerIo;
async fn accept(&mut self) -> (Self::Io, Self::Addr) {
match self {
Self::Tcp(listener) => {
let (io, addr) = listener.accept().await;
(Either::Left(io), Either::Left(addr))
let (io, addr) = axum::serve::Listener::accept(listener).await;
(
ListenerIo::Tcp(enable_tcp_nodelay(io)),
ListenerAddr::Tcp(addr),
)
}
Self::Unix(listener) => {
let (io, addr) = listener.accept().await;
(Either::Right(io), Either::Right(addr))
let (io, addr) = axum::serve::Listener::accept(listener).await;
(ListenerIo::Unix(io), ListenerAddr::Unix(addr))
}
}
}
fn local_addr(&self) -> Result<Self::Addr> {
match self {
Self::Tcp(listener) => listener.local_addr().map(Either::Left),
Self::Unix(listener) => listener.local_addr().map(Either::Right),
self.listener_addr()
}
}
/// Allow the unified listener to be adaptable to `tls_listener`.
impl AsyncAccept for Listener {
type Connection = ListenerIo;
type Address = ListenerAddr;
type Error = std::io::Error;
fn poll_accept(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<Result<(Self::Connection, Self::Address)>> {
match self.get_mut() {
Self::Tcp(listener) => {
let (io, addr) = ready!(listener.poll_accept(cx))?;
Poll::Ready(Ok((
ListenerIo::Tcp(enable_tcp_nodelay(io)),
ListenerAddr::Tcp(addr),
)))
}
Self::Unix(listener) => {
let (io, addr) = ready!(listener.poll_accept(cx))?;
Poll::Ready(Ok((ListenerIo::Unix(io), ListenerAddr::Unix(addr))))
}
}
}
}
impl AsyncListener for Listener {
fn local_addr(&self) -> Result<Self::Address> {
self.listener_addr()
}
}
#[cfg(test)]
mod tests {
use std::net::{Ipv4Addr, SocketAddrV4};
rust/src/server/src/tls.rs
+119
View File
@@ -0,0 +1,119 @@
//! OpenSSL server-config construction for TLS termination.
//!
//! Builds an OpenSSL [`SslContext`] from the uvicorn-style `ssl_*` arguments
//! (certificate chain, private key, mTLS client verifier, optional cipher list).
//! The `tls-listener` crate drives the handshake on each accepted connection.
//!
//! Crypto runs through whichever OpenSSL the binary links (system by default,
//! vendored when built with that feature).
use std::path::Path;
use std::time::Duration;
use anyhow::{Context as _, Result};
use openssl::ssl::{
AlpnError, SslAcceptor, SslAcceptorBuilder, SslContext, SslContextBuilder, SslFiletype,
SslMethod, SslOptions, SslVerifyMode, select_next_proto,
};
use crate::config::TlsConfig;
/// Time a client has to complete the TLS handshake before the connection is dropped.
pub(crate) const TLS_HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(60);
/// ALPN wire bytes for HTTP/2 (length-prefixed).
const ALPN_H2: &[u8] = b"\x02h2";
/// Build the shared OpenSSL acceptor from validated [`TlsConfig`]: the full
/// certificate chain, the private key (`key_file`, or the certificate file when
/// unset), the mTLS client verifier, and an optional cipher list.
///
/// Starts from the Mozilla intermediate baseline (forward-secret AEAD suites,
/// TLS 1.2 floor, server cipher preference, no compression), a slightly
/// stricter subset of the Python frontend's default suites; `--ssl-ciphers`
/// overrides it.
fn build_server_builder(tls: &TlsConfig) -> Result<SslAcceptorBuilder> {
let cert_file = tls.cert_file.as_deref().context("--ssl-certfile is required to enable TLS")?;
let mut builder = SslAcceptor::mozilla_intermediate_v5(SslMethod::tls_server())
.context("failed to initialize TLS")?;
builder.set_options(SslOptions::CIPHER_SERVER_PREFERENCE);
// Load the whole chain (leaf + intermediates), not just the leaf, so
// deployments behind an intermediate CA serve a complete chain.
ensure_exists(cert_file, "--ssl-certfile")?;
builder.set_certificate_chain_file(cert_file).with_context(|| {
format!("failed to parse certificate chain in --ssl-certfile {cert_file:?}")
})?;
// When `key_file` is unset the key is read from the certificate file
// (combined PEM).
let key_file = tls.key_file.as_deref().unwrap_or(cert_file);
ensure_exists(key_file, "private key file")?;
builder
.set_private_key_file(key_file, SslFiletype::PEM)
.with_context(|| format!("failed to parse private key in {key_file:?}"))?;
builder
.check_private_key()
.context("the certificate and private key do not match")?;
configure_client_auth(&mut builder, tls)?;
if let Some(ciphers) = tls.ciphers.as_deref().filter(|c| !c.is_empty()) {
builder
.set_cipher_list(ciphers)
.with_context(|| format!("invalid --ssl-ciphers {ciphers:?}"))?;
}
Ok(builder)
}
/// Build the HTTP [`SslContext`] (HTTP/1.1; no ALPN, matching uvicorn).
pub(crate) fn build_server_config(tls: &TlsConfig) -> Result<SslContext> {
Ok(build_server_builder(tls)?.build().into_context())
}
/// Build the gRPC [`SslContext`]: identical to [`build_server_config`] but
/// negotiates ALPN `h2`, which HTTP/2 over TLS requires.
pub(crate) fn build_grpc_server_config(tls: &TlsConfig) -> Result<SslContext> {
let mut builder = build_server_builder(tls)?;
builder.set_alpn_select_callback(|_ssl, client| {
select_next_proto(ALPN_H2, client).ok_or(AlpnError::NOACK)
});
Ok(builder.build().into_context())
}
/// Fail loudly with a flag-named message when a configured file is missing,
/// distinguishing it from a malformed-PEM error raised later by OpenSSL (whose
/// `ErrorStack` does not name the offending file).
fn ensure_exists(path: &str, what: &str) -> Result<()> {
std::fs::metadata(Path::new(path))
.map(drop)
.with_context(|| format!("failed to read {what} {path:?}"))
}
/// Apply the `cert_reqs` client-certificate policy: 0 = none, 1 = optional
/// (verify if presented, allow anonymous), 2 = required. `PEER` without a custom
/// verify callback still rejects a presented-but-untrusted certificate.
fn configure_client_auth(builder: &mut SslContextBuilder, tls: &TlsConfig) -> Result<()> {
if tls.cert_reqs == 0 {
builder.set_verify(SslVerifyMode::NONE);
return Ok(());
}
let ca_file = tls
.ca_certs
.as_deref()
.context("--ssl-ca-certs is required for client certificate verification")?;
ensure_exists(ca_file, "--ssl-ca-certs")?;
builder
.set_ca_file(ca_file)
.with_context(|| format!("failed to parse --ssl-ca-certs {ca_file:?}"))?;
let mut mode = SslVerifyMode::PEER;
if tls.cert_reqs == 2 {
mode |= SslVerifyMode::FAIL_IF_NO_PEER_CERT;
}
builder.set_verify(mode);
Ok(())
}
rust/src/server/src/tls_tests.rs
+688
View File
@@ -0,0 +1,688 @@
//! TLS tests: `build_server_config` unit checks plus end-to-end OpenSSL handshakes
//! through the production `serve_listener` path, with a trivial router since TLS
//! terminates below the app.
use std::pin::Pin;
use std::time::Duration;
use axum::Router;
use axum::routing::get;
use openssl::asn1::Asn1Time;
use openssl::bn::{BigNum, MsbOption};
use openssl::ec::{EcGroup, EcKey};
use openssl::hash::MessageDigest;
use openssl::nid::Nid;
use openssl::pkey::{PKey, Private};
use openssl::ssl::{SslConnector, SslFiletype, SslMethod, SslVersion};
use openssl::x509::extension::{BasicConstraints, KeyUsage, SubjectAlternativeName};
use openssl::x509::{X509, X509NameBuilder};
use tempfile::TempDir;
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
use tokio::net::TcpStream;
use tokio_openssl::SslStream;
use tokio_util::sync::CancellationToken;
use crate::config::{HttpListenerMode, TlsConfig};
use crate::listener::Listener;
use crate::{ConnectionTimeouts, serve_listener, tls};
// ============================================================================
// Test infrastructure
// ============================================================================
/// A throwaway CA + server/client/untrusted/chain cert set as PEM files in a
/// temp dir; dropping it deletes them.
pub(crate) struct TestCerts {
dir: TempDir,
}
impl TestCerts {
pub(crate) fn generate() -> Self {
let dir = tempfile::tempdir().expect("tempdir");
let (ca, ca_key) = build_ca();
let (server, server_key) = build_leaf("server", &["127.0.0.1", "localhost"], &ca, &ca_key);
let (client, client_key) = build_leaf("client", &[], &ca, &ca_key);
let (untrusted, untrusted_key) = build_self_signed("untrusted client");
// Leaf signed by an intermediate (itself signed by the root); the cert
// file holds leaf + intermediate, for the chain-serving test.
let (intermediate, intermediate_key) = build_intermediate(&ca, &ca_key);
let (chain_leaf, chain_leaf_key) = build_leaf(
"chain",
&["127.0.0.1", "localhost"],
&intermediate,
&intermediate_key,
);
let server_pem = pem(&server);
let server_key_pem = key_pem(&server_key);
let files = [
("ca.pem", pem(&ca)),
("server.pem", server_pem.clone()),
("server.key", server_key_pem.clone()),
("client.pem", pem(&client)),
("client.key", key_pem(&client_key)),
("untrusted_client.pem", pem(&untrusted)),
("untrusted_client.key", key_pem(&untrusted_key)),
(
"server_combined.pem",
format!("{server_pem}{server_key_pem}"),
),
(
"server_chain.pem",
format!("{}{}", pem(&chain_leaf), pem(&intermediate)),
),
("server_chain.key", key_pem(&chain_leaf_key)),
];
for (name, contents) in files {
std::fs::write(dir.path().join(name), contents).expect("write fixture");
}
Self { dir }
}
/// Absolute path to a fixture by name; the file need not exist.
pub(crate) fn path(&self, name: &str) -> String {
self.dir.path().join(name).to_str().expect("utf-8 path").to_string()
}
}
fn gen_key() -> PKey<Private> {
let group = EcGroup::from_curve_name(Nid::X9_62_PRIME256V1).expect("ec group");
let ec = EcKey::generate(&group).expect("ec key");
PKey::from_ec_key(ec).expect("pkey")
}
fn serial() -> openssl::asn1::Asn1Integer {
let mut bn = BigNum::new().expect("bignum");
bn.rand(159, MsbOption::MAYBE_ZERO, false).expect("rand serial");
bn.to_asn1_integer().expect("asn1 serial")
}
fn x509_name(cn: &str) -> openssl::x509::X509Name {
let mut builder = X509NameBuilder::new().expect("name builder");
builder.append_entry_by_text("CN", cn).expect("cn");
builder.build()
}
fn pem(cert: &X509) -> String {
String::from_utf8(cert.to_pem().expect("cert pem")).expect("utf-8 cert")
}
fn key_pem(key: &PKey<Private>) -> String {
String::from_utf8(key.private_key_to_pem_pkcs8().expect("key pem")).expect("utf-8 key")
}
/// A self-signed CA used to sign the server/client leaf certs.
fn build_ca() -> (X509, PKey<Private>) {
let key = gen_key();
let name = x509_name("vLLM Test CA");
let mut builder = X509::builder().expect("x509 builder");
builder.set_version(2).expect("version");
builder.set_serial_number(&serial()).expect("serial");
builder.set_subject_name(&name).expect("subject");
builder.set_issuer_name(&name).expect("issuer");
builder.set_pubkey(&key).expect("pubkey");
builder
.set_not_before(&Asn1Time::days_from_now(0).expect("nb"))
.expect("set nb");
builder
.set_not_after(&Asn1Time::days_from_now(3650).expect("na"))
.expect("set na");
builder
.append_extension(BasicConstraints::new().critical().ca().build().expect("bc"))
.expect("ext bc");
builder
.append_extension(
KeyUsage::new().critical().key_cert_sign().crl_sign().build().expect("ku"),
)
.expect("ext ku");
builder.sign(&key, MessageDigest::sha256()).expect("sign ca");
(builder.build(), key)
}
/// A CA-signed leaf cert with optional subject-alternative names (IP or DNS).
fn build_leaf(cn: &str, sans: &[&str], ca: &X509, ca_key: &PKey<Private>) -> (X509, PKey<Private>) {
let key = gen_key();
let mut builder = X509::builder().expect("x509 builder");
builder.set_version(2).expect("version");
builder.set_serial_number(&serial()).expect("serial");
builder.set_subject_name(&x509_name(cn)).expect("subject");
builder.set_issuer_name(ca.subject_name()).expect("issuer");
builder.set_pubkey(&key).expect("pubkey");
builder
.set_not_before(&Asn1Time::days_from_now(0).expect("nb"))
.expect("set nb");
builder
.set_not_after(&Asn1Time::days_from_now(3650).expect("na"))
.expect("set na");
builder
.append_extension(BasicConstraints::new().build().expect("bc"))
.expect("ext bc");
if !sans.is_empty() {
let mut san = SubjectAlternativeName::new();
for entry in sans {
if entry.parse::<std::net::IpAddr>().is_ok() {
san.ip(entry);
} else {
san.dns(entry);
}
}
let ext = san.build(&builder.x509v3_context(Some(ca), None)).expect("san");
builder.append_extension(ext).expect("ext san");
}
builder.sign(ca_key, MessageDigest::sha256()).expect("sign leaf");
(builder.build(), key)
}
/// A self-signed leaf not chained to the CA, for the untrusted-client test.
fn build_self_signed(cn: &str) -> (X509, PKey<Private>) {
let key = gen_key();
let name = x509_name(cn);
let mut builder = X509::builder().expect("x509 builder");
builder.set_version(2).expect("version");
builder.set_serial_number(&serial()).expect("serial");
builder.set_subject_name(&name).expect("subject");
builder.set_issuer_name(&name).expect("issuer");
builder.set_pubkey(&key).expect("pubkey");
builder
.set_not_before(&Asn1Time::days_from_now(0).expect("nb"))
.expect("set nb");
builder
.set_not_after(&Asn1Time::days_from_now(3650).expect("na"))
.expect("set na");
builder
.append_extension(BasicConstraints::new().build().expect("bc"))
.expect("ext bc");
builder.sign(&key, MessageDigest::sha256()).expect("sign self");
(builder.build(), key)
}
/// A CA-capable intermediate signed by the root, for the full-chain test.
fn build_intermediate(ca: &X509, ca_key: &PKey<Private>) -> (X509, PKey<Private>) {
let key = gen_key();
let mut builder = X509::builder().expect("x509 builder");
builder.set_version(2).expect("version");
builder.set_serial_number(&serial()).expect("serial");
builder
.set_subject_name(&x509_name("vLLM Test Intermediate CA"))
.expect("subject");
builder.set_issuer_name(ca.subject_name()).expect("issuer");
builder.set_pubkey(&key).expect("pubkey");
builder
.set_not_before(&Asn1Time::days_from_now(0).expect("nb"))
.expect("set nb");
builder
.set_not_after(&Asn1Time::days_from_now(3650).expect("na"))
.expect("set na");
builder
.append_extension(BasicConstraints::new().critical().ca().build().expect("bc"))
.expect("ext bc");
builder
.append_extension(
KeyUsage::new().critical().key_cert_sign().crl_sign().build().expect("ku"),
)
.expect("ext ku");
builder.sign(ca_key, MessageDigest::sha256()).expect("sign intermediate");
(builder.build(), key)
}
pub(crate) fn server_tls(certs: &TestCerts, cert_reqs: i32) -> TlsConfig {
TlsConfig {
cert_file: Some(certs.path("server.pem")),
key_file: Some(certs.path("server.key")),
ca_certs: (cert_reqs != 0).then(|| certs.path("ca.pem")),
cert_reqs,
ciphers: None,
}
}
/// A plaintext-listener TLS config for `build_server_config` checks (`cert_reqs`
/// 0, no client auth), with the cert/key files chosen by the caller.
fn build_tls(certs: &TestCerts, cert: &str, key: Option<&str>) -> TlsConfig {
TlsConfig {
cert_file: Some(certs.path(cert)),
key_file: key.map(|k| certs.path(k)),
ca_certs: None,
cert_reqs: 0,
ciphers: None,
}
}
/// Generous per-connection timeouts that never fire during the fast tests.
const TEST_TIMEOUTS: ConnectionTimeouts = ConnectionTimeouts {
handshake: Duration::from_secs(60),
header_read: Duration::from_secs(5),
keep_alive_enabled: true,
};
async fn spawn_server(tls_config: Option<TlsConfig>) -> (String, CancellationToken) {
spawn_server_with_timeouts(tls_config, TEST_TIMEOUTS).await
}
/// Bind an ephemeral listener and serve a trivial router via the production
/// `serve_listener`, optionally with TLS. The listener is bound (and thus
/// accepting into the backlog) before returning, so a client may connect
/// immediately without a sleep.
async fn spawn_server_with_timeouts(
tls_config: Option<TlsConfig>,
timeouts: ConnectionTimeouts,
) -> (String, CancellationToken) {
let listener = Listener::bind(&HttpListenerMode::BindTcp {
host: "127.0.0.1".to_string(),
port: 0,
})
.await
.expect("bind listener");
let addr = listener.local_addr().expect("local addr");
let server_config =
tls_config.map(|cfg| tls::build_server_config(&cfg).expect("build server config"));
let app = Router::new().route("/health", get(|| async { "ok" }));
let shutdown = CancellationToken::new();
let server_shutdown = shutdown.clone();
tokio::spawn(async move {
let _ = serve_listener(
listener,
server_config,
app,
server_shutdown.cancelled_owned(),
timeouts,
)
.await;
});
(addr, shutdown)
}
/// Open a TLS connection trusting the test CA and finish the handshake,
/// optionally presenting a client identity (`<name>.pem` + `<name>.key`) for
/// mTLS. Hostname verification is disabled (the IP-SAN match is not under test);
/// chain verification stays on, so an untrusted server cert is still rejected.
async fn connect_tls(
certs: &TestCerts,
addr: &str,
identity: Option<&str>,
) -> std::io::Result<Pin<Box<SslStream<TcpStream>>>> {
let tcp = TcpStream::connect(addr).await?;
let mut builder = SslConnector::builder(SslMethod::tls_client()).expect("connector builder");
builder.set_ca_file(certs.path("ca.pem")).expect("trust ca");
if let Some(name) = identity {
builder
.set_certificate_chain_file(certs.path(&format!("{name}.pem")))
.expect("client cert");
builder
.set_private_key_file(certs.path(&format!("{name}.key")), SslFiletype::PEM)
.expect("client key");
}
let connector = builder.build();
let mut config = connector.configure().expect("configure");
config.set_verify_hostname(false);
let ssl = config.into_ssl("127.0.0.1").expect("ssl");
let mut stream = Box::pin(SslStream::new(ssl, tcp).expect("client ssl stream"));
stream.as_mut().connect().await.map_err(std::io::Error::other)?;
Ok(stream)
}
/// Issue an HTTPS GET (with `Connection: close`), optionally with an mTLS identity.
async fn https_get(
certs: &TestCerts,
addr: &str,
identity: Option<&str>,
) -> std::io::Result<String> {
let mut stream = connect_tls(certs, addr, identity).await?;
stream
.write_all(b"GET /health HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n")
.await?;
let mut response = String::new();
stream.read_to_string(&mut response).await?;
Ok(response)
}
/// Attempt a handshake offering only a legacy CBC+SHA1 suite over TLS 1.2,
/// capping the version so TLS 1.3 cannot rescue the negotiation.
async fn legacy_suite_handshake(certs: &TestCerts, addr: &str) -> std::io::Result<()> {
let tcp = TcpStream::connect(addr).await?;
let mut builder = SslConnector::builder(SslMethod::tls_client()).expect("connector builder");
builder.set_ca_file(certs.path("ca.pem")).expect("trust ca");
builder.set_max_proto_version(Some(SslVersion::TLS1_2)).expect("cap tls1.2");
builder
.set_cipher_list("ECDHE-ECDSA-AES256-SHA:@SECLEVEL=0")
.expect("legacy cipher");
let connector = builder.build();
let mut config = connector.configure().expect("configure");
config.set_verify_hostname(false);
let ssl = config.into_ssl("127.0.0.1").expect("ssl");
let stream = SslStream::new(ssl, tcp).expect("client ssl stream");
tokio::pin!(stream);
stream.as_mut().connect().await.map_err(std::io::Error::other)
}
async fn plain_get(addr: &str) -> std::io::Result<String> {
let mut tcp = TcpStream::connect(addr).await?;
tcp.write_all(b"GET /health HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n")
.await?;
let mut response = String::new();
tcp.read_to_string(&mut response).await?;
Ok(response)
}
// ============================================================================
// Tests
// ============================================================================
#[test]
fn builds_from_combined_pem() {
// Key omitted: it is read from the combined cert+key file.
let certs = TestCerts::generate();
assert!(tls::build_server_config(&build_tls(&certs, "server_combined.pem", None)).is_ok());
}
#[test]
fn rejects_missing_cert_file() {
let certs = TestCerts::generate();
assert!(tls::build_server_config(&build_tls(&certs, "does_not_exist.pem", None)).is_err());
}
#[test]
fn accepts_valid_cipher_list() {
let certs = TestCerts::generate();
let mut cfg = build_tls(&certs, "server.pem", Some("server.key"));
cfg.ciphers = Some("ECDHE-ECDSA-AES256-GCM-SHA384".to_string());
assert!(tls::build_server_config(&cfg).is_ok());
}
#[test]
fn rejects_invalid_cipher_list() {
let certs = TestCerts::generate();
let mut cfg = build_tls(&certs, "server.pem", Some("server.key"));
cfg.ciphers = Some("THIS-IS-NOT-A-CIPHER".to_string());
assert!(tls::build_server_config(&cfg).is_err());
}
#[test]
fn rejects_mismatched_cert_and_key() {
// check_private_key must reject a key that does not match the certificate.
let certs = TestCerts::generate();
let tls = build_tls(&certs, "client.pem", Some("server.key"));
assert!(tls::build_server_config(&tls).is_err());
}
#[tokio::test]
async fn https_request_succeeds_over_tls() {
let certs = TestCerts::generate();
let (addr, shutdown) = spawn_server(Some(server_tls(&certs, 0))).await;
let response = https_get(&certs, &addr, None).await.expect("https request");
assert!(response.starts_with("HTTP/1.1 200"), "{response}");
shutdown.cancel();
}
#[tokio::test]
async fn serves_full_certificate_chain() {
// Cert file holds leaf + intermediate; a client trusting only the root can
// verify only if the server sends the intermediate, guarding against a
// leaf-only load.
let certs = TestCerts::generate();
let tls = TlsConfig {
cert_file: Some(certs.path("server_chain.pem")),
key_file: Some(certs.path("server_chain.key")),
ca_certs: None,
cert_reqs: 0,
ciphers: None,
};
let (addr, shutdown) = spawn_server(Some(tls)).await;
let response = https_get(&certs, &addr, None).await.expect("chained https request");
assert!(response.starts_with("HTTP/1.1 200"), "{response}");
shutdown.cancel();
}
#[tokio::test]
async fn rejects_legacy_cipher_only_client() {
let certs = TestCerts::generate();
let (addr, shutdown) = spawn_server(Some(server_tls(&certs, 0))).await;
let result = legacy_suite_handshake(&certs, &addr).await;
assert!(result.is_err(), "legacy-only client must be rejected");
shutdown.cancel();
}
#[tokio::test]
async fn ssl_ciphers_override_widens_past_preset() {
// Counterpart to rejects_legacy_cipher_only_client: --ssl-ciphers set to that
// same legacy suite lets the client through, proving the override beats the preset.
let certs = TestCerts::generate();
let mut tls = server_tls(&certs, 0);
tls.ciphers = Some("ECDHE-ECDSA-AES256-SHA:@SECLEVEL=0".to_string());
let (addr, shutdown) = spawn_server(Some(tls)).await;
let result = legacy_suite_handshake(&certs, &addr).await;
assert!(
result.is_ok(),
"override must allow the legacy suite: {result:?}"
);
shutdown.cancel();
}
#[tokio::test]
async fn mtls_required_rejects_client_without_certificate() {
let certs = TestCerts::generate();
let (addr, shutdown) = spawn_server(Some(server_tls(&certs, 2))).await;
let result = https_get(&certs, &addr, None).await;
assert!(
result.is_err(),
"handshake must fail without a client certificate"
);
shutdown.cancel();
}
#[tokio::test]
async fn mtls_required_accepts_valid_client_certificate() {
let certs = TestCerts::generate();
let (addr, shutdown) = spawn_server(Some(server_tls(&certs, 2))).await;
let response = https_get(&certs, &addr, Some("client")).await.expect("mtls request");
assert!(response.starts_with("HTTP/1.1 200"), "{response}");
shutdown.cancel();
}
#[tokio::test]
async fn mtls_optional_allows_anonymous_and_authenticated() {
let certs = TestCerts::generate();
let (addr, shutdown) = spawn_server(Some(server_tls(&certs, 1))).await;
let anonymous = https_get(&certs, &addr, None).await.expect("anonymous request");
assert!(anonymous.starts_with("HTTP/1.1 200"), "{anonymous}");
let authenticated =
https_get(&certs, &addr, Some("client")).await.expect("authenticated request");
assert!(authenticated.starts_with("HTTP/1.1 200"), "{authenticated}");
shutdown.cancel();
}
#[tokio::test]
async fn mtls_rejects_untrusted_client_certificate() {
// Optional (1) still verifies a presented cert, so a self-signed cert not
// chained to the CA is rejected in both modes, not just required (2).
let certs = TestCerts::generate();
for cert_reqs in [1, 2] {
let (addr, shutdown) = spawn_server(Some(server_tls(&certs, cert_reqs))).await;
let result = https_get(&certs, &addr, Some("untrusted_client")).await;
assert!(
result.is_err(),
"cert_reqs={cert_reqs}: untrusted client cert must be rejected"
);
shutdown.cancel();
}
}
#[tokio::test]
async fn plain_http_serves_when_tls_is_disabled() {
let (addr, shutdown) = spawn_server(None).await;
let response = plain_get(&addr).await.expect("http request");
assert!(response.starts_with("HTTP/1.1 200"), "{response}");
shutdown.cancel();
}
#[tokio::test]
async fn tls_handshake_timeout_drops_silent_client() {
// Silent client (no ClientHello) must be dropped at the handshake deadline.
let certs = TestCerts::generate();
let timeouts = ConnectionTimeouts {
handshake: Duration::from_millis(150),
header_read: Duration::from_secs(5),
keep_alive_enabled: true,
};
let (addr, shutdown) = spawn_server_with_timeouts(Some(server_tls(&certs, 0)), timeouts).await;
let mut tcp = TcpStream::connect(&addr).await.expect("connect");
let mut buf = [0u8; 1];
let read = tokio::time::timeout(Duration::from_secs(5), tcp.read(&mut buf)).await;
assert!(
matches!(read, Ok(Ok(0)) | Ok(Err(_))),
"server must drop a stalled TLS handshake (expected close, got {read:?})"
);
shutdown.cancel();
}
#[tokio::test]
async fn keep_alive_timeout_closes_idle_connection() {
// Idle keep-alive connection must be closed at the deadline.
let timeouts = ConnectionTimeouts {
handshake: Duration::from_secs(60),
header_read: Duration::from_millis(150),
keep_alive_enabled: true,
};
let (addr, shutdown) = spawn_server_with_timeouts(None, timeouts).await;
let mut tcp = TcpStream::connect(&addr).await.expect("connect");
// No `Connection: close`, so it stays alive until the idle deadline.
tcp.write_all(b"GET /health HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n")
.await
.expect("write request");
let drained = tokio::time::timeout(Duration::from_secs(5), async {
let mut buf = [0u8; 1024];
loop {
match tcp.read(&mut buf).await {
Ok(0) => return Ok(()),
Ok(_) => continue,
Err(err) => return Err(err),
}
}
})
.await;
assert!(
matches!(drained, Ok(Ok(()))),
"server must close an idle keep-alive connection (got {drained:?})"
);
shutdown.cancel();
}
#[tokio::test]
async fn keep_alive_timeout_closes_idle_tls_connection() {
// The keep-alive idle bound lives in serve_connections, below TLS; assert it
// still fires through tls-listener's post-handshake SslStream, not just plaintext.
let certs = TestCerts::generate();
let timeouts = ConnectionTimeouts {
handshake: Duration::from_secs(60),
header_read: Duration::from_millis(150),
keep_alive_enabled: true,
};
let (addr, shutdown) = spawn_server_with_timeouts(Some(server_tls(&certs, 0)), timeouts).await;
let mut stream = connect_tls(&certs, &addr, None).await.expect("handshake");
// No `Connection: close`, so the connection stays alive until the idle deadline.
stream
.write_all(b"GET /health HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n")
.await
.expect("write request");
let closed = tokio::time::timeout(Duration::from_secs(5), async {
let mut buf = [0u8; 1024];
loop {
// A clean close_notify (Ok(0)) or an abrupt TLS EOF both mean the
// server closed; only the outer timeout (still open) is a failure.
match stream.read(&mut buf).await {
Ok(0) | Err(_) => break,
Ok(_) => continue,
}
}
})
.await;
assert!(
closed.is_ok(),
"server must close an idle keep-alive TLS connection at the deadline"
);
shutdown.cancel();
}
#[tokio::test]
async fn idle_timeout_closes_silent_client() {
// Silent client closed by the header-read timeout (http1-only arms it from byte 0).
let timeouts = ConnectionTimeouts {
handshake: Duration::from_secs(60),
header_read: Duration::from_millis(150),
keep_alive_enabled: true,
};
let (addr, shutdown) = spawn_server_with_timeouts(None, timeouts).await;
let mut tcp = TcpStream::connect(&addr).await.expect("connect");
let mut buf = [0u8; 1];
let read = tokio::time::timeout(Duration::from_secs(5), tcp.read(&mut buf)).await;
assert!(
matches!(read, Ok(Ok(0)) | Ok(Err(_))),
"server must close a silent client (expected close, got {read:?})"
);
shutdown.cancel();
}
#[tokio::test]
async fn keep_alive_zero_disables_keep_alive() {
// 0 disables keep-alive (serve, then close), like uvicorn's timeout_keep_alive=0.
let timeouts = ConnectionTimeouts {
handshake: Duration::from_secs(60),
header_read: Duration::from_secs(5),
keep_alive_enabled: false,
};
let (addr, shutdown) = spawn_server_with_timeouts(None, timeouts).await;
let mut tcp = TcpStream::connect(&addr).await.expect("connect");
tcp.write_all(b"GET /health HTTP/1.1\r\nHost: 127.0.0.1\r\n\r\n")
.await
.expect("write request");
let mut response = String::new();
let read =
tokio::time::timeout(Duration::from_secs(5), tcp.read_to_string(&mut response)).await;
assert!(
read.is_ok(),
"server must close after one response, not hang"
);
assert!(response.starts_with("HTTP/1.1 200"), "{response}");
// Assert `Connection: close`, not just 200: a 0 header-read timeout would also
// serve an immediate request, so 200 alone wouldn't prove keep-alive is off.
assert!(
response.to_ascii_lowercase().contains("connection: close"),
"keep-alive must be disabled (expected Connection: close): {response}"
);
shutdown.cancel();
}
#[tokio::test]
async fn disabled_keep_alive_still_closes_silent_client() {
// Even with keep-alive off, the head read stays bounded, so a silent client
// is dropped rather than held open.
let timeouts = ConnectionTimeouts {
handshake: Duration::from_secs(60),
header_read: Duration::from_millis(150),
keep_alive_enabled: false,
};
let (addr, shutdown) = spawn_server_with_timeouts(None, timeouts).await;
let mut tcp = TcpStream::connect(&addr).await.expect("connect");
let mut buf = [0u8; 1];
let read = tokio::time::timeout(Duration::from_secs(5), tcp.read(&mut buf)).await;
assert!(
matches!(read, Ok(Ok(0)) | Ok(Err(_))),
"disabled keep-alive must still close a silent client (got {read:?})"
);
shutdown.cancel();
}
vllm/v1/utils.py
+4 -2
View File
@@ -378,9 +378,11 @@ class RustFrontendProcessManager:
# The Rust `frontend` subcommand parses --args-json via serde_json,
# which bypasses clap and therefore ignores any `#[arg(env = ...)]`
# declarations on SharedRuntimeArgs fields. Forward the env-driven
# ready timeout explicitly so VLLM_ENGINE_READY_TIMEOUT_S behaves the
# same on both Python and Rust frontends.
# values explicitly so VLLM_ENGINE_READY_TIMEOUT_S and
# VLLM_HTTP_TIMEOUT_KEEP_ALIVE behave the same on both Python and Rust
# frontends.
args_dict["engine_ready_timeout_secs"] = envs.VLLM_ENGINE_READY_TIMEOUT_S
args_dict["http_timeout_keep_alive"] = envs.VLLM_HTTP_TIMEOUT_KEEP_ALIVE
args_json = json.dumps(args_dict, sort_keys=True)
cmd.extend(["--args-json", args_json])