diff --git a/rust/Cargo.lock b/rust/Cargo.lock index 7820a7b67674..da782e41e68b 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -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", diff --git a/rust/Cargo.toml b/rust/Cargo.toml index a1f963b9b0f8..60ba138e8b5c 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -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" diff --git a/rust/src/cmd/src/cli.rs b/rust/src/cmd/src/cli.rs index a3fa9b05500b..d45baadb0155 100644 --- a/rust/src/cmd/src/cli.rs +++ b/rust/src/cmd/src/cli.rs @@ -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, /// 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, + + /// The file path to the SSL cert file. Enables TLS when set. + #[arg(long)] + #[serde(default)] + pub ssl_certfile: Option, + + /// The CA certificates file used to verify client certificates (mTLS). + #[arg(long)] + #[serde(default)] + pub ssl_ca_certs: Option, + + /// 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, + /// 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 { + 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 { diff --git a/rust/src/cmd/src/cli/tests.rs b/rust/src/cmd/src/cli/tests.rs index a9b11ca18f7f..551603fce0d7 100644 --- a/rust/src/cmd/src/cli/tests.rs +++ b/rust/src/cmd/src/cli/tests.rs @@ -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 ': argument is not implemented in Rust frontend yet + error: invalid value '/prefix' for '--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 ': + error: invalid value '{"model_tag":"Qwen/Qwen3-0.6B","root_path":"/prefix"}' for '--args-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 ': + error: invalid value '{"model_tag":"Qwen/Qwen3-0.6B","response_role":"assistant","root_path":"/prefix"}' for '--args-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); diff --git a/rust/src/cmd/src/cli/unsupported.rs b/rust/src/cmd/src/cli/unsupported.rs index e7fb4bc0ba7f..521a188d6ac3 100644 --- a/rust/src/cmd/src/cli/unsupported.rs +++ b/rust/src/cmd/src/cli/unsupported.rs @@ -526,18 +526,6 @@ pub struct ServerUnsupportedArgs { #[arg(long)] pub disable_access_log_for_endpoints: Option, - /// The file path to the SSL key file. - #[arg(long)] - pub ssl_keyfile: Option, - - /// The file path to the SSL cert file. - #[arg(long)] - pub ssl_certfile: Option, - - /// The CA certificates file. - #[arg(long)] - pub ssl_ca_certs: Option, - /// Refresh SSL Context when SSL certificate files change #[arg( long, @@ -547,15 +535,6 @@ pub struct ServerUnsupportedArgs { )] pub enable_ssl_refresh: Option, - /// Whether client certificate is required (see stdlib ssl module's). - #[arg(long)] - pub ssl_cert_reqs: Option, - - /// 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, - /// FastAPI root_path when app is behind a path based routing proxy. #[arg(long)] pub root_path: Option, diff --git a/rust/src/engine-core-client/src/tests/client.rs b/rust/src/engine-core-client/src/tests/client.rs index a7ccd5981644..11c403c56377 100644 --- a/rust/src/engine-core-client/src/tests/client.rs +++ b/rust/src/engine-core-client/src/tests/client.rs @@ -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 = 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 = 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() ); diff --git a/rust/src/server/Cargo.toml b/rust/src/server/Cargo.toml index c73da7a0a94c..59c7f8d17441 100644 --- a/rust/src/server/Cargo.toml +++ b/rust/src/server/Cargo.toml @@ -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 diff --git a/rust/src/server/examples/external_engine_openai_qwen.rs b/rust/src/server/examples/external_engine_openai_qwen.rs index 6eea1afe703b..03e474ad4c47 100644 --- a/rust/src/server/examples/external_engine_openai_qwen.rs +++ b/rust/src/server/examples/external_engine_openai_qwen.rs @@ -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}"); diff --git a/rust/src/server/src/config.rs b/rust/src/server/src/config.rs index c601bbfb6344..ae8466b32163 100644 --- a/rust/src/server/src/config.rs +++ b/rust/src/server/src/config.rs @@ -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, + /// PEM private key file. When `None`, the key is read from `cert_file` + /// (combined PEM). + pub key_file: Option, + /// PEM CA bundle used to verify client certificates (mTLS). Required when + /// `cert_reqs` is non-zero. + pub ca_certs: Option, + /// 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, +} + +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, /// 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, /// 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 { diff --git a/rust/src/server/src/grpc/mod.rs b/rust/src/server/src/grpc/mod.rs index 1fcb8674fee1..0a71f2edc122 100644 --- a/rust/src/server/src/grpc/mod.rs +++ b/rust/src/server/src/grpc/mod.rs @@ -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, +} + +impl GrpcTlsStream { + pub(crate) fn new(inner: SslStream) -> Self { + Self { inner } + } +} + +impl AsyncRead for GrpcTlsStream { + fn poll_read( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &mut ReadBuf<'_>, + ) -> Poll> { + 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> { + Pin::new(&mut self.get_mut().inner).poll_write(cx, buf) + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + 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> { + 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> { + 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, diff --git a/rust/src/server/src/grpc/tests.rs b/rust/src/server/src/grpc/tests.rs index 14156a410460..311b819e0c80 100644 --- a/rust/src/server/src/grpc/tests.rs +++ b/rust/src/server/src/grpc/tests.rs @@ -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, output_specs: Vec<(Vec, Option)>, -) -> ( - GenerateClient, - tokio::task::JoinHandle<()>, - MockEngineTask, -) { +) -> (GenerateServer, 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, ); 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, + output_specs: Vec<(Vec, Option)>, +) -> ( + GenerateClient, + 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, + output_specs: Vec<(Vec, Option)>, + 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, 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>>> { + 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, + keepalive: Option, +) -> (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(); +} diff --git a/rust/src/server/src/lib.rs b/rust/src/server/src/lib.rs index c22a06ddc328..55b4f5ffae30 100644 --- a/rust/src/server/src/lib.rs +++ b/rust/src/server/src/lib.rs @@ -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 { 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> { @@ -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, + app: Router, + shutdown: impl Future + 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( + mut listener: L, + app: Router, + shutdown: impl Future + 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| 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"); + } } diff --git a/rust/src/server/src/listener.rs b/rust/src/server/src/listener.rs index b7b715b0ebd7..75fe25aef515 100644 --- a/rust/src/server/src/listener.rs +++ b/rust/src/server/src/listener.rs @@ -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 { + 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; - type Io = Either; + 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 { - 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> { + 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.listener_addr() + } +} + #[cfg(test)] mod tests { use std::net::{Ipv4Addr, SocketAddrV4}; diff --git a/rust/src/server/src/tls.rs b/rust/src/server/src/tls.rs new file mode 100644 index 000000000000..aa77e8191ed4 --- /dev/null +++ b/rust/src/server/src/tls.rs @@ -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 { + 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 { + 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 { + 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(()) +} diff --git a/rust/src/server/src/tls_tests.rs b/rust/src/server/src/tls_tests.rs new file mode 100644 index 000000000000..7e0a6dccc30c --- /dev/null +++ b/rust/src/server/src/tls_tests.rs @@ -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 { + 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) -> 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) { + 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) -> (X509, PKey) { + 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::().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) { + 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) -> (X509, PKey) { + 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) -> (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, + 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 (`.pem` + `.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>>> { + 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 { + 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 { + 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(); +} diff --git a/vllm/v1/utils.py b/vllm/v1/utils.py index a083f309e0ec..2ddd151980f7 100644 --- a/vllm/v1/utils.py +++ b/vllm/v1/utils.py @@ -378,9 +378,11 @@ def __init__( # 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])