diff --git a/CHANGELOG.md b/CHANGELOG.md index d28f2b5..607588f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,25 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- Client API: protocol version negotiation. `Config::protocol_version` + selects the version to advertise in the startup message, defaulting to 3.0. + Following libpq, a test version `3.9999` is available to request the newest + minor version a server supports: the client handles the server's + `NegotiateProtocolVersion` response (both the full 32-bit version form used + by PostgreSQL 18+ and the minor-only form used by older servers) and adopts + the negotiated version for the rest of the connection. + +### Fixed + +- Client API: backend messages are now decoded with the rules of the protocol + version the client actually advertised, instead of always 3.2. Previously a + 4-byte protocol 3.0 cancel key was decoded as `SecretKey::Bytes` instead of + `SecretKey::I32`. + ## [0.40.7] - 2026-08-12 ### Fixed diff --git a/src/api/client/auth.rs b/src/api/client/auth.rs index bcf9121..fbfe21f 100644 --- a/src/api/client/auth.rs +++ b/src/api/client/auth.rs @@ -9,10 +9,10 @@ use crate::api::auth::sasl::scram::ScramClientAuth; use crate::error::{ErrorInfo, PgWireClientError, PgWireClientResult, PgWireResult}; use crate::messages::response::ReadyForQuery; use crate::messages::startup::{ - Authentication, BackendKeyData, ParameterStatus, Password, PasswordMessageFamily, - SASLInitialResponse, SASLResponse, SecretKey, Startup, + Authentication, BackendKeyData, NegotiateProtocolVersion, ParameterStatus, Password, + PasswordMessageFamily, SASLInitialResponse, SASLResponse, SecretKey, Startup, }; -use crate::messages::{PgWireBackendMessage, PgWireFrontendMessage}; +use crate::messages::{PgWireBackendMessage, PgWireFrontendMessage, ProtocolVersion}; use super::{ClientInfo, ReadyState, ServerInformation}; @@ -43,6 +43,10 @@ pub trait StartupHandler: Send { PgWireBackendMessage::Authentication(authentication) => { self.on_authentication(client, authentication).await?; } + PgWireBackendMessage::NegotiateProtocolVersion(negotiation) => { + self.on_negotiate_protocol_version(client, negotiation) + .await?; + } PgWireBackendMessage::ParameterStatus(parameter_status) => { self.on_parameter_status(client, parameter_status).await?; } @@ -78,6 +82,34 @@ pub trait StartupHandler: Send { + Send, PgWireClientError: From<>::Error>; + /// Handle a `NegotiateProtocolVersion` message from the server. + /// + /// The default implementation adopts the negotiated version for the rest + /// of the connection. Both the full 32-bit version number used by + /// PostgreSQL 18+ and pgwire, and the minor-only form used by older + /// servers, are understood. Startup parameters reported as unrecognized + /// by the server are ignored, matching libpq's behavior for non-`_pq_` + /// options. + async fn on_negotiate_protocol_version( + &mut self, + client: &mut C, + message: NegotiateProtocolVersion, + ) -> PgWireClientResult<()> + where + C: ClientInfo + Sink + Unpin + Send, + PgWireClientError: From<>::Error>, + { + match negotiated_version(&message) { + Some(version) => { + client.set_protocol_version(version); + Ok(()) + } + None => Err(PgWireClientError::UnexpectedMessage(Box::new( + PgWireBackendMessage::NegotiateProtocolVersion(message), + ))), + } + } + /// Handle a parameter status message from the server. async fn on_parameter_status( &mut self, @@ -127,11 +159,17 @@ impl StartupHandler for DefaultStartupHandler { C: ClientInfo + Sink + Unpin + Send, PgWireClientError: From<>::Error>, { - // TODO: customize protocol version let mut startup = Startup::new(); let config = client.config(); + // Advertise the configured protocol version. The connection decodes + // with the rules of the advertised version until the server lowers it + // via NegotiateProtocolVersion. + let (major, minor) = config.get_protocol_version().version_number(); + startup.protocol_number_major = major; + startup.protocol_number_minor = minor; + if let Some(application_name) = &config.application_name { startup .parameters @@ -321,3 +359,107 @@ where }; auth_client.verify_server_final(&message) } + +/// Interpret the version number reported by a `NegotiateProtocolVersion` +/// message and return the protocol version to use for the rest of the +/// connection. +/// +/// This is the client-side counterpart of the server's +/// `api::auth::protocol_negotiation`. Two wire dialects exist: +/// +/// - PostgreSQL 18+ and pgwire report the **full 32-bit protocol version +/// number** (e.g. `196610` for 3.2), which is always greater than `65535`, +/// - older servers report only the **minor version** (e.g. `2`), leaving the +/// major version unchanged from the client's request. +/// +/// Returns `None` if the reported version cannot be mapped to a version this +/// crate supports (e.g. a newer major version). +fn negotiated_version(message: &NegotiateProtocolVersion) -> Option { + let value = message.newest_minor_protocol; + if value < 0 { + return None; + } + + let (major, minor) = if value > i32::from(u16::MAX) { + // full 32-bit protocol version number + (((value >> 16) & 0xFFFF) as u16, (value & 0xFFFF) as u16) + } else { + // minor-only form, the major version is unchanged + (3, value as u16) + }; + + match ProtocolVersion::from_version_number(major, minor) { + Some(version) => Some(version), + // A minor version we don't have a variant for: protocol 3.2 is what + // changed the wire formats we care about (secret keys), so an unknown + // minor below 2 behaves like 3.0, and one at or above 2 falls back to + // our newest 3.x. + None if major == 3 && minor >= 2 => Some(ProtocolVersion::PROTOCOL3_2), + None if major == 3 => Some(ProtocolVersion::PROTOCOL3_0), + None => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_negotiated_version() { + // The same version can arrive in two wire dialects, and both forms + // must negotiate to the same version: + // + // - the full 32-bit protocol version number (`i32::from(version)`), + // used by PostgreSQL 18+ and current pgwire, + // - the minor version alone, used by older servers, which leaves the + // major version unchanged from the client's request. + // + // Realistic on the wire: full 3.2 (PostgreSQL 18+), bare 0 (older + // PostgreSQL, which only supports 3.0) and bare 2 (pgwire before the + // #439 fix sent the minor alone). Full-form 3.0 never occurs, but is + // unambiguous. + for (full_form, minor_form, expected) in [ + ( + i32::from(ProtocolVersion::PROTOCOL3_2), + 2, + ProtocolVersion::PROTOCOL3_2, + ), + ( + i32::from(ProtocolVersion::PROTOCOL3_0), + 0, + ProtocolVersion::PROTOCOL3_0, + ), + ] { + assert_eq!( + negotiated_version(&NegotiateProtocolVersion::new(full_form, vec![])), + Some(expected) + ); + assert_eq!( + negotiated_version(&NegotiateProtocolVersion::new(minor_form, vec![])), + Some(expected) + ); + } + + // Unknown minors: below 2 behaves like 3.0, at or above 2 falls back + // to our newest 3.x + assert_eq!( + negotiated_version(&NegotiateProtocolVersion::new(1, vec![])), + Some(ProtocolVersion::PROTOCOL3_0) + ); + assert_eq!( + negotiated_version(&NegotiateProtocolVersion::new(5, vec![])), + Some(ProtocolVersion::PROTOCOL3_2) + ); + + // An unknown major version cannot be mapped + assert_eq!( + negotiated_version(&NegotiateProtocolVersion::new((4 << 16) | 2, vec![])), + None + ); + // A negative value is invalid + assert_eq!( + negotiated_version(&NegotiateProtocolVersion::new(-1, vec![])), + None + ); + } +} diff --git a/src/api/client/config.rs b/src/api/client/config.rs index cda7883..2ed6bf7 100644 --- a/src/api/client/config.rs +++ b/src/api/client/config.rs @@ -17,6 +17,7 @@ use std::time::Duration; use std::{fmt, iter, mem, str}; use crate::error::PgWireClientError; +use crate::messages::ProtocolVersion; /// Properties required of a session. #[derive(Debug, Copy, Clone, PartialEq, Eq)] @@ -222,6 +223,7 @@ pub struct Config { pub(crate) target_session_attrs: TargetSessionAttrs, pub(crate) channel_binding: ChannelBinding, pub(crate) load_balance_hosts: LoadBalanceHosts, + pub(crate) protocol_version: ProtocolVersion, } impl Default for Config { @@ -256,6 +258,7 @@ impl Config { target_session_attrs: TargetSessionAttrs::Any, channel_binding: ChannelBinding::Prefer, load_balance_hosts: LoadBalanceHosts::Disable, + protocol_version: ProtocolVersion::PROTOCOL3_0, } } @@ -551,6 +554,25 @@ impl Config { self.load_balance_hosts } + /// Sets the protocol version to advertise in the startup message. + /// + /// Defaults to `PROTOCOL3_0`. + /// + /// Following libpq, you can set [`ProtocolVersion::PROTOCOL3_9999`] to + /// request the newest minor version the server supports: the server + /// replies with `NegotiateProtocolVersion` and the connection proceeds at + /// the negotiated version. This is useful for exercising the negotiation + /// path against any server. + pub fn protocol_version(&mut self, protocol_version: ProtocolVersion) -> &mut Config { + self.protocol_version = protocol_version; + self + } + + /// Gets the protocol version to advertise in the startup message. + pub fn get_protocol_version(&self) -> ProtocolVersion { + self.protocol_version + } + fn param(&mut self, key: &str, value: &str) -> Result<(), PgWireClientError> { match key { "user" => { @@ -749,6 +771,7 @@ impl fmt::Debug for Config { config_dbg .field("target_session_attrs", &self.target_session_attrs) .field("channel_binding", &self.channel_binding) + .field("protocol_version", &self.protocol_version) .finish() } } diff --git a/src/api/client/mod.rs b/src/api/client/mod.rs index d40f414..b4a434c 100644 --- a/src/api/client/mod.rs +++ b/src/api/client/mod.rs @@ -31,6 +31,15 @@ pub trait ClientInfo { /// Returns client protocol version fn protocol_version(&self) -> ProtocolVersion; + /// Sets the protocol version in effect for this connection. + /// + /// Custom [`StartupHandler`](auth::StartupHandler) implementations should + /// call this with the version they advertise in the `Startup` message, so + /// that subsequent backend messages are decoded with the rules of that + /// version. The default startup handler and the negotiation flow handle + /// this automatically. + fn set_protocol_version(&mut self, version: ProtocolVersion); + // TODO: transaction state } diff --git a/src/api/mod.rs b/src/api/mod.rs index 8985c29..781e9af 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -347,7 +347,7 @@ impl PidSecretKeyGenerator for RandomPidSecretKeyGenerator { let pid = self.next_pid.fetch_add(1, Ordering::Relaxed); let secret_key = match client.protocol_version() { ProtocolVersion::PROTOCOL3_0 => SecretKey::I32(rand::random::()), - ProtocolVersion::PROTOCOL3_2 => { + ProtocolVersion::PROTOCOL3_2 | ProtocolVersion::PROTOCOL3_9999 => { let mut bytes = vec![0u8; 32]; rand::fill(&mut bytes); SecretKey::Bytes(bytes.into()) diff --git a/src/messages/mod.rs b/src/messages/mod.rs index c23a0f4..458b19f 100644 --- a/src/messages/mod.rs +++ b/src/messages/mod.rs @@ -17,6 +17,19 @@ pub enum ProtocolVersion { #[default] /// Protocol version 3.2 PROTOCOL3_2, + /// Protocol version 3.9999 + /// + /// This is not a real protocol version. Following libpq, a client can + /// request this version in its startup message to ask for the newest + /// minor version the server supports: a compliant server replies with + /// `NegotiateProtocolVersion` and the connection proceeds at the + /// negotiated version. This exercises the negotiation path against any + /// server. + /// + /// Note that [`ProtocolVersion::from_version_number`] deliberately does + /// not map to this variant: a server must never accept 3.9999 as-is, it + /// always negotiates it down to a supported version. + PROTOCOL3_9999, } impl ProtocolVersion { @@ -25,6 +38,7 @@ impl ProtocolVersion { match &self { Self::PROTOCOL3_0 => (3, 0), Self::PROTOCOL3_2 => (3, 2), + Self::PROTOCOL3_9999 => (3, 9999), } } @@ -1028,6 +1042,16 @@ mod test { fn test_protocol_version() { assert_eq!(196608i32, i32::from(ProtocolVersion::PROTOCOL3_0)); assert_eq!(196610i32, i32::from(ProtocolVersion::PROTOCOL3_2)); + assert_eq!(206607i32, i32::from(ProtocolVersion::PROTOCOL3_9999)); + // 3.9999 is encodable (a client requests it to trigger negotiation) + // but is never accepted as a supported version by a server. + assert_eq!( + None, + ProtocolVersion::from_version_number( + ProtocolVersion::PROTOCOL3_9999.version_number().0, + ProtocolVersion::PROTOCOL3_9999.version_number().1 + ) + ); } // A count read as signed (0xffff -> -1 -> usize::MAX) must become a decode diff --git a/src/messages/startup.rs b/src/messages/startup.rs index 5de357a..8eaf11a 100644 --- a/src/messages/startup.rs +++ b/src/messages/startup.rs @@ -505,7 +505,9 @@ impl SecretKey { Self::validate_bytes_len(data_len)?; match ctx.protocol_version { - ProtocolVersion::PROTOCOL3_2 => Ok(SecretKey::Bytes(buf.split_to(data_len).freeze())), + ProtocolVersion::PROTOCOL3_2 | ProtocolVersion::PROTOCOL3_9999 => { + Ok(SecretKey::Bytes(buf.split_to(data_len).freeze())) + } ProtocolVersion::PROTOCOL3_0 => Ok(SecretKey::I32(buf.get_i32())), } } @@ -776,6 +778,13 @@ impl Message for SASLResponse { #[non_exhaustive] #[derive(PartialEq, Eq, Debug, new)] pub struct NegotiateProtocolVersion { + /// Version number reported by the server. + /// + /// PostgreSQL 18+ (protocol 3.2) and pgwire send the **full 32-bit + /// protocol version number** here (e.g. `196610` for 3.2), while older + /// servers send only the **minor version** (e.g. `2`). The client-side + /// negotiation logic in `pgwire::api::client::auth` understands both + /// forms. pub newest_minor_protocol: i32, pub unsupported_options: Vec, } diff --git a/src/tokio/client.rs b/src/tokio/client.rs index f1793c2..73cc3fa 100644 --- a/src/tokio/client.rs +++ b/src/tokio/client.rs @@ -106,6 +106,10 @@ impl ClientInfo for PgWireClient { fn protocol_version(&self) -> ProtocolVersion { self.socket.codec().decode_context.protocol_version } + + fn set_protocol_version(&mut self, version: ProtocolVersion) { + self.socket.codec_mut().decode_context.protocol_version = version; + } } impl Sink for PgWireClient { @@ -155,6 +159,10 @@ impl PgWireClient { tls_connector: tls_connector_for_cancel, }; + // Decode backend messages with the rules of the protocol version we + // are about to advertise, until the server negotiates it down. + client.set_protocol_version(config.get_protocol_version()); + startup_handler.startup(&mut client).await?; // loop until finished while let Some(message_result) = client.socket.next().await { @@ -453,3 +461,77 @@ fn get_addr(config: &Config) -> Result { Err(PgWireClientError::InvalidConfig("host".to_string())) } + +#[cfg(all(test, feature = "server-api"))] +mod tests { + use std::sync::Arc; + + use tokio::net::TcpListener; + + use super::PgWireClient; + use crate::api::PgWireServerHandlers; + use crate::api::client::ClientInfo; + use crate::api::client::auth::DefaultStartupHandler; + use crate::api::client::config::Config; + use crate::messages::ProtocolVersion; + use crate::messages::startup::SecretKey; + use crate::tokio::server::process_socket; + + struct TestHandlers; + + impl PgWireServerHandlers for TestHandlers {} + + async fn spawn_test_server() -> u16 { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + tokio::spawn(async move { + loop { + let (socket, _) = listener.accept().await.unwrap(); + let handlers = Arc::new(TestHandlers); + tokio::spawn(async move { + let _ = process_socket(socket, None, handlers).await; + }); + } + }); + port + } + + #[tokio::test] + async fn client_negotiates_3_9999_down_to_3_2() { + let port = spawn_test_server().await; + + let mut config = Config::new(); + config.host("127.0.0.1"); + config.port(port); + config.user("pgwire"); + config.protocol_version(ProtocolVersion::PROTOCOL3_9999); + + let client = PgWireClient::connect(Arc::new(config), DefaultStartupHandler::new(), None) + .await + .unwrap(); + + // The server did not accept 3.9999 as-is; it negotiated down to its + // newest supported version. + assert_eq!(client.protocol_version(), ProtocolVersion::PROTOCOL3_2); + // Protocol 3.2 backend keys are 32 bytes long. + assert!(matches!(client.secret_key(), SecretKey::Bytes(key) if key.len() == 32)); + } + + #[tokio::test] + async fn client_default_3_0_keeps_i32_secret_key() { + let port = spawn_test_server().await; + + let mut config = Config::new(); + config.host("127.0.0.1"); + config.port(port); + config.user("pgwire"); + + let client = PgWireClient::connect(Arc::new(config), DefaultStartupHandler::new(), None) + .await + .unwrap(); + + assert_eq!(client.protocol_version(), ProtocolVersion::PROTOCOL3_0); + // A protocol 3.0 cancel key is decoded as a 4-byte i32, not as bytes. + assert!(matches!(client.secret_key(), SecretKey::I32(_))); + } +}