Add support for sslnegotation=direct.

[resolves #651]

Author: mp911de

README.md

@@ -26,6 +26,7 @@
 import io.r2dbc.postgresql.client.MultiHostConfiguration;
 import io.r2dbc.postgresql.client.SSLConfig;
 import io.r2dbc.postgresql.client.SSLMode;
+import io.r2dbc.postgresql.client.SSLNegotiation;
 import io.r2dbc.postgresql.client.SingleHostConfiguration;
 import io.r2dbc.postgresql.codec.Codec;
 import io.r2dbc.postgresql.codec.Codecs;
@@ -406,6 +407,8 @@ public static final class Builder {
 
         private SSLMode sslMode = SSLMode.DISABLE;
 
+        private SSLNegotiation sslNegotiation = SSLNegotiation.POSTGRES;
+
         @Nullable
         private CharSequence sslPassword = null;
 
@@ -971,6 +974,18 @@ public Builder sslMode(SSLMode sslMode) {
             return this;
         }
 
+        /**
+         * Configure ssl negotiation. Useful if the server is known to support SSL directly (e.g. Postgres 17+ or a SSL tunnel).
+         *
+         * @param sslNegotiation the SSL negotiation mechanism to use.
+         * @return this {@link Builder}
+         * @since 1.1
+         */
+        public Builder sslNegotiation(SSLNegotiation sslNegotiation) {
+            this.sslNegotiation = Assert.requireNonNull(sslNegotiation, "sslNegotiation must be not be null");
+            return this;
+        }
+
         /**
          * Configure ssl password.
          *
@@ -1168,19 +1183,23 @@ private SSLConfig createSslConfig(boolean sslSni) {
                 return SSLConfig.disabled();
             }
 
-            Function<SocketAddress, SSLParameters> sslParametersFunctionToUse = getSslParametersFactory(sslSni, this.sslParametersFactory);
-            return new SSLConfig(this.sslMode, createSslProvider(), this.sslEngineCustomizer, sslParametersFunctionToUse, this.sslHostnameVerifier);
+            Function<SocketAddress, SSLParameters> sslParametersFunctionToUse = getSslParametersFactory(sslSni, this.sslNegotiation, this.sslParametersFactory);
+            return new SSLConfig(this.sslNegotiation, this.sslMode, createSslProvider(), this.sslEngineCustomizer, sslParametersFunctionToUse, this.sslHostnameVerifier);
         }
 
-        private static Function<SocketAddress, SSLParameters> getSslParametersFactory(boolean sslSni, Function<SocketAddress, SSLParameters> sslParametersFunction) {
-            if (!sslSni) {
+        private static Function<SocketAddress, SSLParameters> getSslParametersFactory(boolean sslSni, SSLNegotiation sslNegotiation, Function<SocketAddress, SSLParameters> sslParametersFunction) {
+            if (!sslSni && sslNegotiation != SSLNegotiation.DIRECT) {
                 return sslParametersFunction;
             }
 
             return socket -> {
 
                 SSLParameters sslParameters = sslParametersFunction.apply(socket);
 
+                if (sslNegotiation == SSLNegotiation.DIRECT) {
+                    sslParameters.setApplicationProtocols(new String[]{"postgresql"});
+                }
+
                 if (socket instanceof InetSocketAddress) {
 
                     InetSocketAddress inetSocketAddress = (InetSocketAddress) socket;

src/main/java/io/r2dbc/postgresql/PostgresqlConnectionConfiguration.java

@@ -19,6 +19,7 @@
 import io.netty.handler.ssl.SslContextBuilder;
 import io.r2dbc.postgresql.client.DefaultHostnameVerifier;
 import io.r2dbc.postgresql.client.SSLMode;
+import io.r2dbc.postgresql.client.SSLNegotiation;
 import io.r2dbc.postgresql.codec.Codecs;
 import io.r2dbc.postgresql.codec.Json;
 import io.r2dbc.postgresql.extension.Extension;
@@ -211,6 +212,20 @@ public final class PostgresqlConnectionFactoryProvider implements ConnectionFact
      */
     public static final Option<SSLMode> SSL_MODE_ALIAS = Option.valueOf("sslmode");
 
+    /**
+     * Ssl negotiation mechanism. Default: Postgres
+     *
+     * @since 1.1
+     */
+    public static final Option<SSLNegotiation> SSL_NEGOTIATION = Option.valueOf("sslNegotiation");
+
+    /**
+     * Ssl negotiation mechanism alias. Default: Postgres
+     *
+     * @since 1.1
+     */
+    public static final Option<SSLNegotiation> SSL_NEGOTIATION_ALIAS = Option.valueOf("sslnegotiation");
+
     /**
      * SSL key password
      */
@@ -408,6 +423,10 @@ private static void setupSsl(PostgresqlConnectionConfiguration.Builder builder,
             mapper.from(SSL_MODE_ALIAS).map(PostgresqlConnectionFactoryProvider::toSSLMode).to(builder::sslMode);
         });
 
+        mapper.from(SSL_NEGOTIATION).map(PostgresqlConnectionFactoryProvider::toSSLNegotiation).to(builder::sslNegotiation).otherwise(() -> {
+            mapper.from(SSL_NEGOTIATION_ALIAS).map(PostgresqlConnectionFactoryProvider::toSSLNegotiation).to(builder::sslNegotiation);
+        });
+
         mapper.fromTyped(SSL_CERT).to(builder::sslCert);
         mapper.fromTyped(SSL_CONTEXT_BUILDER_CUSTOMIZER).to(builder::sslContextBuilderCustomizer);
         mapper.fromTyped(SSL_KEY).to(builder::sslKey);
@@ -441,6 +460,14 @@ private static SSLMode toSSLMode(Object it) {
         return (SSLMode) it;
     }
 
+    private static SSLNegotiation toSSLNegotiation(Object it) {
+        if (it instanceof String) {
+            return SSLNegotiation.fromValue(it.toString());
+        }
+
+        return (SSLNegotiation) it;
+    }
+
     @SuppressWarnings("unchecked")
     private static Map<String, String> convertToMap(Object options) {
         if (options instanceof Map) {

src/main/java/io/r2dbc/postgresql/PostgresqlConnectionFactoryProvider.java

@@ -17,17 +17,14 @@
 package io.r2dbc.postgresql;
 
 import io.r2dbc.postgresql.authentication.AuthenticationHandler;
-import io.r2dbc.postgresql.authentication.PasswordAuthenticationHandler;
-import io.r2dbc.postgresql.authentication.SASLAuthenticationHandler;
+import io.r2dbc.postgresql.authentication.UsernameAndPassword;
 import io.r2dbc.postgresql.client.Client;
 import io.r2dbc.postgresql.client.ConnectionContext;
 import io.r2dbc.postgresql.client.ConnectionSettings;
 import io.r2dbc.postgresql.client.PostgresStartupParameterProvider;
 import io.r2dbc.postgresql.client.StartupMessageFlow;
 import io.r2dbc.postgresql.message.backend.AuthenticationMessage;
-import io.r2dbc.postgresql.util.Assert;
 import reactor.core.publisher.Mono;
-import reactor.util.annotation.Nullable;
 
 import java.net.SocketAddress;
 
@@ -57,15 +54,7 @@ private static PostgresStartupParameterProvider getParameterProvider(PostgresqlC
     }
 
     protected AuthenticationHandler getAuthenticationHandler(AuthenticationMessage message, UsernameAndPassword usernameAndPassword, ConnectionContext context) {
-        if (PasswordAuthenticationHandler.supports(message)) {
-            CharSequence password = Assert.requireNonNull(usernameAndPassword.getPassword(), "Password must not be null");
-            return new PasswordAuthenticationHandler(password, usernameAndPassword.getUsername());
-        } else if (SASLAuthenticationHandler.supports(message)) {
-            CharSequence password = Assert.requireNonNull(usernameAndPassword.getPassword(), "Password must not be null");
-            return new SASLAuthenticationHandler(password, usernameAndPassword.getUsername(), context);
-        } else {
-            throw new IllegalStateException(String.format("Unable to provide AuthenticationHandler capable of handling %s", message));
-        }
+        return AuthenticationHandler.getAuthenticationHandler(message, usernameAndPassword, context);
     }
 
     Mono<UsernameAndPassword> getCredentials() {
@@ -75,25 +64,4 @@ Mono<UsernameAndPassword> getCredentials() {
         });
     }
 
-    static class UsernameAndPassword {
-
-        final String username;
-
-        final @Nullable CharSequence password;
-
-        public UsernameAndPassword(String username, @Nullable CharSequence password) {
-            this.username = username;
-            this.password = password;
-        }
-
-        public String getUsername() {
-            return this.username;
-        }
-
-        @Nullable
-        public CharSequence getPassword() {
-            return this.password;
-        }
-    }
-
 }

src/main/java/io/r2dbc/postgresql/SingleHostConnectionFunction.java

@@ -16,8 +16,10 @@
 
 package io.r2dbc.postgresql.authentication;
 
+import io.r2dbc.postgresql.client.ConnectionContext;
 import io.r2dbc.postgresql.message.backend.AuthenticationMessage;
 import io.r2dbc.postgresql.message.frontend.FrontendMessage;
+import io.r2dbc.postgresql.util.Assert;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -36,4 +38,24 @@ public interface AuthenticationHandler {
     @Nullable
     FrontendMessage handle(AuthenticationMessage message);
 
+    /**
+     * Return a suitable {@link AuthenticationHandler} for the given {@link AuthenticationMessage} and {@link UsernameAndPassword}.
+     *
+     * @param message             the message to handle
+     * @param usernameAndPassword authentication credentials
+     * @param context             connection context
+     * @return the authentication handler
+     * @since 1.1
+     */
+    static AuthenticationHandler getAuthenticationHandler(AuthenticationMessage message, UsernameAndPassword usernameAndPassword, ConnectionContext context) {
+        if (PasswordAuthenticationHandler.supports(message)) {
+            CharSequence password = Assert.requireNonNull(usernameAndPassword.getPassword(), "Password must not be null");
+            return new PasswordAuthenticationHandler(password, usernameAndPassword.getUsername());
+        } else if (SASLAuthenticationHandler.supports(message)) {
+            CharSequence password = Assert.requireNonNull(usernameAndPassword.getPassword(), "Password must not be null");
+            return new SASLAuthenticationHandler(password, usernameAndPassword.getUsername(), context);
+        } else {
+            throw new IllegalStateException(String.format("Unable to provide AuthenticationHandler capable of handling %s", message));
+        }
+    }
 }

src/main/java/io/r2dbc/postgresql/authentication/AuthenticationHandler.java

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.r2dbc.postgresql.authentication;
+
+import reactor.util.annotation.Nullable;
+
+public class UsernameAndPassword {
+
+    private final String username;
+
+    private final @Nullable CharSequence password;
+
+    public UsernameAndPassword(String username, @Nullable CharSequence password) {
+        this.username = username;
+        this.password = password;
+    }
+
+    public String getUsername() {
+        return this.username;
+    }
+
+    @Nullable
+    public CharSequence getPassword() {
+        return this.password;
+    }
+
+}

src/main/java/io/r2dbc/postgresql/authentication/UsernameAndPassword.java

@@ -22,13 +22,13 @@
 import java.net.SocketAddress;
 
 /**
- * SSL handler assuming the endpoint is a SSL tunnel and not a Postgres endpoint.
+ * SSL handler assuming the endpoint uses SSL directly (Postgres 17+ endpoint or an SSL tunnel).
  */
-final class SSLTunnelHandlerAdapter extends AbstractPostgresSSLHandlerAdapter {
+final class DirectSSLHandlerAdapter extends AbstractPostgresSSLHandlerAdapter {
 
     private final SSLConfig sslConfig;
 
-    SSLTunnelHandlerAdapter(ByteBufAllocator alloc, SocketAddress socketAddress, SSLConfig sslConfig) {
+    DirectSSLHandlerAdapter(ByteBufAllocator alloc, SocketAddress socketAddress, SSLConfig sslConfig) {
         super(alloc, socketAddress, sslConfig);
         this.sslConfig = sslConfig;
     }

src/main/java/io/r2dbc/postgresql/client/SSLTunnelHandlerAdapter.java renamed to src/main/java/io/r2dbc/postgresql/client/DirectSSLHandlerAdapter.java

@@ -434,8 +434,8 @@ private static void registerSslHandler(SocketAddress socketAddress, SSLConfig ss
             if (sslConfig.getSslMode().startSsl()) {
 
                 AbstractPostgresSSLHandlerAdapter sslAdapter;
-                if (sslConfig.getSslMode() == SSLMode.TUNNEL) {
-                    sslAdapter = new SSLTunnelHandlerAdapter(channel.alloc(), socketAddress, sslConfig);
+                if (sslConfig.isDirectSsl()) {
+                    sslAdapter = new DirectSSLHandlerAdapter(channel.alloc(), socketAddress, sslConfig);
                 } else {
                     sslAdapter = new SSLSessionHandlerAdapter(channel.alloc(), socketAddress, sslConfig);
                 }

src/main/java/io/r2dbc/postgresql/client/ReactorNettyClient.java

@@ -32,6 +32,8 @@ public final class SSLConfig {
     @Nullable
     private final HostnameVerifier hostnameVerifier;
 
+    private final SSLNegotiation sslNegotiation;
+
     private final SSLMode sslMode;
 
     @Nullable
@@ -42,11 +44,13 @@ public final class SSLConfig {
     private final Function<SocketAddress, SSLParameters> sslParametersFactory;
 
     public SSLConfig(SSLMode sslMode, @Nullable Supplier<SslContext> sslProvider, @Nullable HostnameVerifier hostnameVerifier) {
-        this(sslMode, sslProvider, Function.identity(), it -> new SSLParameters(), hostnameVerifier);
+        this(SSLNegotiation.POSTGRES, sslMode, sslProvider, Function.identity(), it -> new SSLParameters(), hostnameVerifier);
     }
 
-    public SSLConfig(SSLMode sslMode, @Nullable Supplier<SslContext> sslProvider, Function<SSLEngine, SSLEngine> sslEngineCustomizer, Function<SocketAddress, SSLParameters> sslParametersFactory,
+    public SSLConfig(SSLNegotiation sslNegotiation, SSLMode sslMode, @Nullable Supplier<SslContext> sslProvider, Function<SSLEngine, SSLEngine> sslEngineCustomizer, Function<SocketAddress,
+                         SSLParameters> sslParametersFactory,
                      @Nullable HostnameVerifier hostnameVerifier) {
+        this.sslNegotiation = sslNegotiation;
         if (sslMode != SSLMode.DISABLE) {
             Assert.requireNonNull(sslProvider, "SslContext provider is required for ssl mode " + sslMode);
         }
@@ -68,6 +72,15 @@ HostnameVerifier getHostnameVerifier() {
         return this.hostnameVerifier;
     }
 
+    @SuppressWarnings("deprecation")
+    public boolean isDirectSsl() {
+        return getSslMode() == SSLMode.TUNNEL || getSslNegotiation() == SSLNegotiation.DIRECT || getSslNegotiation() == SSLNegotiation.TUNNEL;
+    }
+
+    public SSLNegotiation getSslNegotiation() {
+        return this.sslNegotiation;
+    }
+
     public SSLMode getSslMode() {
         return this.sslMode;
     }
@@ -83,13 +96,13 @@ public Function<SSLEngine, SSLEngine> getSslEngineCustomizer() {
         return this.sslEngineCustomizer;
     }
 
-
     public Function<SocketAddress, SSLParameters> getSslParametersFactory() {
         return this.sslParametersFactory;
     }
 
     public SSLConfig mutateMode(SSLMode newMode) {
         return new SSLConfig(
+            this.sslNegotiation,
             newMode,
             this.sslProvider,
             this.sslEngineCustomizer,

src/main/java/io/r2dbc/postgresql/client/SSLConfig.java

@@ -16,6 +16,8 @@
 
 package io.r2dbc.postgresql.client;
 
+import java.util.Arrays;
+
 public enum SSLMode {
     /**
      * I don't care about security and don't want to pay the overhead for encryption
@@ -48,8 +50,10 @@ public enum SSLMode {
     VERIFY_FULL("verify-full"),
 
     /**
-     * I want to use a SSL tunnel instead of following Postgres SSL handshake protocol.
+     * I want to use an SSL tunnel instead of following Postgres SSL handshake protocol.
+     * @deprecated since 1.1, use {@link SSLNegotiation#TUNNEL} and configure your own SSL tunnel.
      */
+    @Deprecated
     TUNNEL("tunnel");
 
     private final String value;
@@ -68,7 +72,7 @@ public boolean startSsl() {
 
     @Override
     public String toString() {
-        return value;
+        return this.value;
     }
 
     public boolean verifyCertificate() {
@@ -85,6 +89,6 @@ public static SSLMode fromValue(String sslModeString) {
                 return sslMode;
             }
         }
-        throw new IllegalArgumentException("Invalid ssl mode value: " + sslModeString);
+        throw new IllegalArgumentException("Invalid ssl mode value: " + sslModeString + ". Supported values are: " + Arrays.toString(values()));
     }
 }