diff --git a/README.md b/README.md index 87d1cc2fd..9acac06fb 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ 1. The [**Java Nats Examples**](https://github.com/nats-io/java-nats-examples) GitHub repo, a collection of simple use case examples. ## Table of Contents +* [Important](#important) * [Client and Orbit](#client-and-orbit) * [Simplification](#simplification) * [Consumer Info Calls](#consumer-info-calls) @@ -40,6 +41,10 @@ * [Examples](#examples) * [License](#license) +## Important + +Version 2.26.5 turns TLS hostname verification on by default, so a connection to a server whose certificate does not name the host in the url now fails. See [Version 2.26.5 Hostname Verification On By Default](#version-2265-hostname-verification-on-by-default) for who is affected and how to opt out. + ## Client and Orbit NATS client functionality is split across two layers: the **core client** @@ -204,6 +209,32 @@ rather than shrinking the general `executor`. ## Recent Version Notes +### Version 2.26.5 Hostname Verification On By Default + +Starting with 2.26.5, the client checks that the certificate a server presents was issued for the host in the url you connect with. Before 2.26.5 it did not: any certificate your trust store accepted was accepted for any host, and `opentls` accepted any certificate at all. The check applies to every TLS connection, including `opentls`, and to servers the client discovers from the cluster, not only the ones in your options. + +When the certificate does not name the host, the connection fails and the error names the host it was looking for, for example `No subject alternative names matching IP address 10.0.0.5 found`. + +You are not affected if you do not use TLS, or if you connect by a name, such as `tls://nats.example.com:4222`, and the certificate was issued for that name. This is the normal case with a certificate from a public or private CA. + +You are affected if: + +1. You connect by ip address, such as `tls://10.0.0.5:4222`, and the certificate does not list that ip address as a subject alternative name. Certificates issued for a hostname usually do not. +2. You connect by a name the certificate does not carry: a load balancer or proxy address, an internal alias, or a Kubernetes service name, in front of servers whose certificate names something else. +3. You use `opentls` or a trust-all context against a certificate that does not name the host you connect with. The common case is development against a self-signed certificate for `localhost` while connecting by `127.0.0.1`, or the other way round. +4. Your cluster advertises its servers by ip address and each server has its own certificate for its own name. Reconnects to the advertised ips are checked against the name you connected with. +5. You supply your own `SSLContext`. The check applies to it too, unless your trust manager is an `X509ExtendedTrustManager`, which is left to do its own identity check. + +To check a certificate, list its subject alternative names and compare them with the host in your url: + +``` +openssl x509 -in server.pem -noout -ext subjectAltName +``` + +If you are affected, connect by a name the certificate carries, or reissue the certificate with the names and ip addresses you connect with. For a cluster that advertises ips, set `client_advertise` on the servers so they advertise hostnames instead. + +To keep the old behavior, call `tlsVerifyHostname(false)` on the options builder, or set the property `io.nats.client.tls.verify.hostname=false`. This removes the proof that the server is the one named in the url, so treat it as a development setting, not a fix. The full rules are in [TLS configuration](docs/TLS_CONFIGURATION.md). + ### Version 2.26.0 Advanced Request Behavior This release adds an opt-in connection option, `advancedRequestBehavior`, that surfaces *why* a request @@ -705,6 +736,8 @@ When options are built, the SSLContext will be accepted or created in the follow 3. If `opentls` is true or any of the bootstrap servers has `opentls` as their scheme, a generic SSLContext will be created that **"trusts all certs"**. Asking for both, by setting `secure` and `opentls`, or with a bootstrap list that mixes `opentls` with `tls` or `wss`, is rejected; see [TLS configuration](docs/TLS_CONFIGURATION.md). 4. If `secure` is true or any of the bootstrap servers has `tls` or `wss` as their scheme, the `javax.net.ssl.SSLContext.getDefault()` will be used. +When the options end up with an SSLContext by any of these routes, a bootstrap server with the `ws` scheme is rejected when the options are built, because a `ws` connection is never upgraded to TLS. Use `wss` for that server. A `ws` server the cluster advertises in `connect_urls` is skipped when the pool hands it out, reported through the error listener's `serverRejected`, and dropped from the pool; see [TLS configuration](docs/TLS_CONFIGURATION.md). + ### Publishing Once connected, publishing is accomplished via one of three methods: diff --git a/docs/TLS_CONFIGURATION.md b/docs/TLS_CONFIGURATION.md index d81d14f39..c091a786b 100644 --- a/docs/TLS_CONFIGURATION.md +++ b/docs/TLS_CONFIGURATION.md @@ -17,6 +17,8 @@ The context is accepted or created once, when `Options.Builder.build()` runs, in Rules 4 and 5 look at the bootstrap servers only when neither `opentls` nor `secure` was set explicitly. Asking for both is an error, section 5. +When the options end up with a context, by any of rules 1 to 5, a `ws://` server in the list is an error at `build()`: `ws://` is a plain websocket and is never upgraded to TLS, so that server would be reached without it. Use `wss://` for it. A `nats://` server is allowed, because it is upgraded to TLS like every other server of the connection. A `ws://` server that reaches the connection another way, advertised by the cluster in `connect_urls` or supplied by a custom `ServerPool`, is never dialed: when the pool hands it out, the connection skips it and goes on to the next server, reports it through `ErrorListener.serverRejected`, and tells the pool through `ServerPool.serverRejected`, so the default pool drops it and does not take it back from a later `connect_urls`. + ## 2. The default context: `tls://`, `wss://`, `secure()` `SSLContext.getDefault()` is the JVM's default: it verifies the server's certificate chain against the JVM trust store, `cacerts` in the JDK's `lib/security`, or whatever the `javax.net.ssl.trustStore` system properties point at. It presents the client certificate from `javax.net.ssl.keyStore` when a server asks for one. It does not read the operating system's certificate store; a CA installed in Windows or macOS is not trusted by the JVM unless it is also in the JVM trust store or the JVM is configured to use the system store. diff --git a/src/main/java/io/nats/client/ErrorListener.java b/src/main/java/io/nats/client/ErrorListener.java index 9974af54d..e70140bd0 100644 --- a/src/main/java/io/nats/client/ErrorListener.java +++ b/src/main/java/io/nats/client/ErrorListener.java @@ -1,4 +1,4 @@ -// Copyright 2015-2018 The NATS Authors +// Copyright 2015-2026 The NATS 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: @@ -150,6 +150,18 @@ default void flowControlProcessed(Connection conn, JetStreamSubscription sub, St */ default void socketWriteTimeout(Connection conn) {} + /** + * Called by the connection when the server pool supplied a server that the connection will never attempt, + * because it cannot be used with the options: for example a plain ws:// server when the options carry + * an SSL context, since a plain websocket is never upgraded to TLS. The connection goes on to the next + * server, and the pool is told through ServerPool.serverRejected so it stops supplying the server. + * + * @param conn The connection that rejected the server + * @param server The server that was rejected, as a url + * @param reason Why it was rejected + */ + default void serverRejected(Connection conn, String server, String reason) {} + /** * General message producing function which understands the possible parameters to listener calls. * @param label the label for the message diff --git a/src/main/java/io/nats/client/Options.java b/src/main/java/io/nats/client/Options.java index 4e249a524..c9d3f59d4 100644 --- a/src/main/java/io/nats/client/Options.java +++ b/src/main/java/io/nats/client/Options.java @@ -2507,6 +2507,16 @@ else if (useDefaultTls) { } } + // A connection with an SSL context requires TLS to every server, but a ws:// server is a plain + // websocket and is never upgraded, so it would be reached without TLS. nats:// servers are upgraded. + if (sslContext != null) { + for (NatsUri natsUri : natsServerUris) { + if (WEBSOCKET_PROTOCOL.equals(natsUri.getScheme())) { + throw new IllegalStateException("Options require TLS (secure, opentls, an SSL context, keystore or truststore, an SSL context factory, or a tls, wss or opentls server url), but the server list has a ws:// server, which is never upgraded to TLS. Use wss:// for that server."); + } + } + } + if (tlsFirst && sslContext == null) { throw new IllegalStateException("SSL context required for tls handshake first"); } diff --git a/src/main/java/io/nats/client/ServerPool.java b/src/main/java/io/nats/client/ServerPool.java index bfecd998e..88a482a3b 100644 --- a/src/main/java/io/nats/client/ServerPool.java +++ b/src/main/java/io/nats/client/ServerPool.java @@ -101,6 +101,18 @@ default List resolveHostToIps(@NonNull String host, boolean maxOneResult */ void connectFailed(@NonNull NatsUri nuri); + /** + * The connection will never attempt the server, because it cannot be used with the options: + * for example a plain ws:// server when the options carry an SSL context, since a plain websocket + * is never upgraded to TLS. The connection has already taken the server from the pool and goes on + * to the next one. The pool should stop supplying the server, including when it is discovered again. + * The default implementation does nothing. A pool that keeps supplying it does not make the connection + * attempt it: when the pool supplies a rejected server again and has supplied nothing usable since, + * the connection ends that connect or reconnect. + * @param nuri the server that was rejected + */ + default void serverRejected(@NonNull NatsUri nuri) {} + /** * Get the list of servers known to the pool. Purely informational * @return the list of servers diff --git a/src/main/java/io/nats/client/impl/ErrorListenerConsoleImpl.java b/src/main/java/io/nats/client/impl/ErrorListenerConsoleImpl.java index 7742db889..e83e7d69e 100644 --- a/src/main/java/io/nats/client/impl/ErrorListenerConsoleImpl.java +++ b/src/main/java/io/nats/client/impl/ErrorListenerConsoleImpl.java @@ -1,4 +1,4 @@ -// Copyright 2024 The NATS Authors +// Copyright 2024-2026 The NATS 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: @@ -98,4 +98,12 @@ public void flowControlProcessed(Connection conn, JetStreamSubscription sub, Str public void socketWriteTimeout(Connection conn) { System.out.println(supplyMessage("[SEVERE] socketWriteTimeout", conn, null, null)); } + + /** + * {@inheritDoc} + */ + @Override + public void serverRejected(Connection conn, String server, String reason) { + System.out.println(supplyMessage("[WARNING] serverRejected", conn, null, null, "Server: ", server, "Reason: ", reason)); + } } diff --git a/src/main/java/io/nats/client/impl/ErrorListenerLoggerImpl.java b/src/main/java/io/nats/client/impl/ErrorListenerLoggerImpl.java index 298a3161b..53f235c7d 100644 --- a/src/main/java/io/nats/client/impl/ErrorListenerLoggerImpl.java +++ b/src/main/java/io/nats/client/impl/ErrorListenerLoggerImpl.java @@ -1,4 +1,4 @@ -// Copyright 2021 The NATS Authors +// Copyright 2021-2026 The NATS 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: @@ -102,4 +102,12 @@ public void flowControlProcessed(Connection conn, JetStreamSubscription sub, Str public void socketWriteTimeout(Connection conn) { LOGGER.severe(() -> supplyMessage("socketWriteTimeout", conn, null, null)); } + + /** + * {@inheritDoc} + */ + @Override + public void serverRejected(Connection conn, String server, String reason) { + LOGGER.warning(() -> supplyMessage("serverRejected", conn, null, null, "Server: ", server, "Reason: ", reason)); + } } diff --git a/src/main/java/io/nats/client/impl/NatsConnection.java b/src/main/java/io/nats/client/impl/NatsConnection.java index 15c91db12..f42673545 100644 --- a/src/main/java/io/nats/client/impl/NatsConnection.java +++ b/src/main/java/io/nats/client/impl/NatsConnection.java @@ -249,7 +249,25 @@ protected void connectImpl(boolean reconnectOnConnect) throws InterruptedExcepti boolean keepGoing = true; NatsUri first = null; NatsUri cur; + // A rejected server, mapped to the count of usable servers handed out when it was last supplied. + // The pool is told to drop a rejected server; if it supplies one again with nothing usable in between, it has nothing else. + Map rejected = new HashMap<>(); + long usableCount = 0; while (keepGoing && (cur = serverPool.peekNextServer()) != null) { + String rejection = rejectionReason(cur); + if (rejection != null) { + serverPool.nextServer(); // b/c we only peeked, and the pool must have handed it out before it is told + Long usableAtLastRejection = rejected.put(cur, usableCount); + if (usableAtLastRejection == null) { + rejectServer(cur, rejection); + failList.add(cur); + } + else if (usableAtLastRejection == usableCount) { + break; // the pool kept a rejected server and handed out nothing usable since it last supplied it + } + continue; // before first is set: a rejected server must not end the pass + } + usableCount++; if (first == null) { first = cur; } @@ -472,7 +490,22 @@ protected void reconnectImplConnect() throws InterruptedException { int totalRounds = 0; NatsUri first = null; NatsUri cur; + // see connect: a rejected server, mapped to the count of usable servers handed out when it was last supplied + Map rejected = new HashMap<>(); + long usableCount = 0; while ((cur = serverPool.nextServer()) != null) { + String rejection = rejectionReason(cur); + if (rejection != null) { + Long usableAtLastRejection = rejected.put(cur, usableCount); + if (usableAtLastRejection == null) { + rejectServer(cur, rejection); + } + else if (usableAtLastRejection == usableCount) { + return; // the pool kept a rejected server and handed out nothing usable since it last supplied it + } + continue; // before first is set and before the delay: a rejected server is not an attempt + } + usableCount++; if (first == null) { first = cur; if (options.reconnectDelayBehavior() == Options.ReconnectDelayBehavior.BeforeAllRounds) { @@ -562,6 +595,23 @@ else if (remainingNanos < 1_000_000_000) { // is called from reconnect and connect // will wait for any previous attempt to complete, using the reader.stop and // writer.stop + // A server the connection will never attempt because it cannot be used with the options: a ws:// server is + // a plain websocket that is never upgraded to TLS, so with an SSL context it would be reached without TLS. + // Both connect loops ask this when the pool hands a server out, before anything is sent to it. The bootstrap + // list cannot hold such a server, so one comes from connect_urls or from a custom ServerPool. + protected String rejectionReason(NatsUri nuri) { + if (options.isTLSRequired() && WEBSOCKET_PROTOCOL.equals(nuri.getScheme())) { + return "the options require TLS and a ws:// server is never upgraded to TLS"; + } + return null; + } + + // The pool has already handed the server out. Tell it so it stops supplying the server, and report it. + protected void rejectServer(NatsUri nuri, String reason) { + serverPool.serverRejected(nuri); + makeCallback(() -> options.getErrorListener().serverRejected(this, nuri.toString(), reason)); + } + protected void tryToConnect(NatsUri cur, NatsUri resolved, long now) { clearCurrentServer(); connectingServer = cur; // the origin of discovered servers in the initial INFO, before currentServer is set diff --git a/src/main/java/io/nats/client/impl/NatsServerPool.java b/src/main/java/io/nats/client/impl/NatsServerPool.java index 6d02e3267..ed71f3a5b 100644 --- a/src/main/java/io/nats/client/impl/NatsServerPool.java +++ b/src/main/java/io/nats/client/impl/NatsServerPool.java @@ -36,10 +36,12 @@ public class NatsServerPool implements ServerPool { protected boolean hasSecureServer; protected NatsUri lastConnected; protected String defaultScheme; + protected final List rejected; public NatsServerPool() { listLock = new ReentrantLock(); entryList = new ArrayList<>(); // this gets updated occasionally + rejected = new ArrayList<>(); options = Options.builder().build(); // this will get updated when initialize is called } @@ -113,10 +115,16 @@ public boolean acceptDiscoveredUrls(@NonNull List<@NonNull String> discoveredSer // 2. Build a list for discovered // - since we'll need the NatsUris later // - and to have a list to use to prune removed gossiped servers + // - a server the connection rejected is never supplied again, so it is left out here. + // The match is the exact uri: the scheme is what made it unusable, so the same + // host and port under another scheme is still a candidate List discovered = new ArrayList<>(); for (String d : discoveredServers) { try { - discovered.add(new NatsUri(d, defaultScheme)); + NatsUri nuri = new NatsUri(d, defaultScheme); + if (!rejected.contains(nuri)) { + discovered.add(nuri); + } } catch (URISyntaxException ignore) { // should never actually happen } @@ -285,6 +293,30 @@ public void connectFailed(@NonNull NatsUri nuri) { } } + /** + * {@inheritDoc} + */ + @Override + public void serverRejected(@NonNull NatsUri nuri) { + // 1. Remember the exact uri so a later discovery does not bring it back + // 2. Remove it from the list. Work from the end because nextServer moved the one being tried to the end + listLock.lock(); + try { + if (!rejected.contains(nuri)) { + rejected.add(nuri); + } + for (int x = entryList.size() - 1; x >= 0; x--) { + if (entryList.get(x).nuri.equals(nuri)) { + entryList.remove(x); + return; + } + } + } + finally { + listLock.unlock(); + } + } + @Override @NonNull public List getServerList() { diff --git a/src/test/java/io/nats/client/OptionsTests.java b/src/test/java/io/nats/client/OptionsTests.java index 7dde2b7fd..a200f4bb7 100644 --- a/src/test/java/io/nats/client/OptionsTests.java +++ b/src/test/java/io/nats/client/OptionsTests.java @@ -1574,6 +1574,17 @@ public void testSslContextIsProvided() throws Exception { assertNotNull(new Options.Builder().server("tls://localhost,opentls://localhost").opentls().build().getSslContext()); SSLContext ctx = SslTestingHelper.createTestSSLContext(); assertEquals(ctx, new Options.Builder().server("tls://localhost,opentls://localhost").sslContext(ctx).build().getSslContext()); + + // with TLS required, a ws:// server would be reached without TLS: a builder error. nats:// servers are upgraded. + assertThrows(IllegalStateException.class, () -> new Options.Builder().server("wss://a,ws://b").build()); + assertThrows(IllegalStateException.class, () -> new Options.Builder().server("tls://a,ws://b").build()); + assertThrows(IllegalStateException.class, () -> new Options.Builder().server("opentls://a,ws://b").build()); + assertThrows(IllegalStateException.class, () -> new Options.Builder().server("ws://a").secure().build()); + assertThrows(IllegalStateException.class, () -> new Options.Builder().server("ws://a").opentls().build()); + assertThrows(IllegalStateException.class, () -> new Options.Builder().server("ws://a").sslContext(ctx).build()); + assertNotNull(new Options.Builder().server("nats://a,tls://b").build().getSslContext()); + assertNotNull(new Options.Builder().server("nats://a").secure().build().getSslContext()); + assertNull(new Options.Builder().server("ws://a,nats://b").build().getSslContext()); } @SuppressWarnings("deprecation") diff --git a/src/test/java/io/nats/client/impl/ListenerForTesting.java b/src/test/java/io/nats/client/impl/ListenerForTesting.java index 3c85925d8..ae94fb797 100644 --- a/src/test/java/io/nats/client/impl/ListenerForTesting.java +++ b/src/test/java/io/nats/client/impl/ListenerForTesting.java @@ -1,4 +1,4 @@ -// Copyright 2015-2018 The NATS Authors +// Copyright 2015-2026 The NATS 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: @@ -60,6 +60,7 @@ public class ListenerForTesting implements ErrorListener, ConnectionListener { private final List pullStatusErrors = new ArrayList<>(); private final List heartbeatAlarms = new ArrayList<>(); private final List flowControlProcessedEvents = new ArrayList<>(); + private final List serverRejectedEvents = new ArrayList<>(); public final AtomicInteger socketWriteTimeoutCount = new AtomicInteger(); private final boolean printExceptions; @@ -100,6 +101,7 @@ public void reset() { pullStatusErrors.clear(); heartbeatAlarms.clear(); flowControlProcessedEvents.clear(); + serverRejectedEvents.clear(); socketWriteTimeoutCount.set(0); } @@ -378,6 +380,10 @@ public List getFlowControlProcessedEvents() { return flowControlProcessedEvents; } + public List getServerRejectedEvents() { + return serverRejectedEvents; + } + public int getSocketWriteTimeoutCount() { return socketWriteTimeoutCount.get(); } @@ -556,6 +562,11 @@ public void socketWriteTimeout(Connection conn) { socketWriteTimeoutCount.incrementAndGet(); } + @Override + public void serverRejected(Connection conn, String server, String reason) { + serverRejectedEvents.add(new ServerRejectedEvent(server, reason)); + } + public static class StatusEvent { String sid; Status status; @@ -616,6 +627,24 @@ public String toString() { } } + public static class ServerRejectedEvent { + final String server; + final String reason; + + public ServerRejectedEvent(String server, String reason) { + this.server = server; + this.reason = reason; + } + + @Override + public String toString() { + return "ServerRejectedEvent{" + + "server='" + server + '\'' + + ", reason='" + reason + '\'' + + '}'; + } + } + private static String extractSid(JetStreamSubscription sub) { return ((NatsJetStreamSubscription)sub).getSID(); } diff --git a/src/test/java/io/nats/client/impl/ServerPoolTests.java b/src/test/java/io/nats/client/impl/ServerPoolTests.java index 8c42cdb91..ab194bf97 100644 --- a/src/test/java/io/nats/client/impl/ServerPoolTests.java +++ b/src/test/java/io/nats/client/impl/ServerPoolTests.java @@ -1,4 +1,4 @@ -// Copyright 2022 The NATS Authors +// Copyright 2022-2026 The NATS 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: @@ -126,6 +126,34 @@ public void testPruning() throws URISyntaxException { validateNslp(nsp, null, false, BOOT_ONE, BOOT_TWO, DISC_ONE, DISC_TWO); } + @Test + public void testServerRejected() throws URISyntaxException { + // a rejected server leaves the pool and does not come back when it is discovered again + Options o = new Options.Builder().noRandomize().servers(bootstrap).build(); + NatsServerPool nsp = newNatsServerPool(o, null, discoveredServers); + validateNslp(nsp, null, true, combined); + + nsp.serverRejected(new NatsUri(DISC_ONE)); + validateNslp(nsp, null, true, BOOT_ONE, BOOT_TWO, DISC_TWO, DISC_THREE); + + // discovered again: not taken back, and not counted as an unknown server + assertFalse(nsp.acceptDiscoveredUrls(discoveredServers)); + validateNslp(nsp, null, true, BOOT_ONE, BOOT_TWO, DISC_TWO, DISC_THREE); + + // a server the pool does not hold is remembered all the same; the empty result prunes the gossiped servers as usual + nsp.serverRejected(new NatsUri("nats://d4")); + assertFalse(nsp.acceptDiscoveredUrls(Collections.singletonList("nats://d4"))); + validateNslp(nsp, null, true, BOOT_ONE, BOOT_TWO); + + // the memory is the exact uri: the scheme is what made the server unusable, so the same host and port + // under another scheme is still accepted while the rejected uri itself is not + nsp.serverRejected(new NatsUri("ws://d9")); + assertTrue(nsp.acceptDiscoveredUrls(Collections.singletonList("nats://d9"))); + validateNslp(nsp, null, true, BOOT_ONE, BOOT_TWO, "nats://d9"); + assertFalse(nsp.acceptDiscoveredUrls(Arrays.asList("ws://d9", "nats://d9"))); + validateNslp(nsp, null, true, BOOT_ONE, BOOT_TWO, "nats://d9"); + } + private static NatsServerPool newNatsServerPool(Options o, NatsUri last, List discoveredServers) { NatsServerPool nsp = new NatsServerPool(); nsp.initialize(o); diff --git a/src/test/java/io/nats/client/impl/TLSConnectTests.java b/src/test/java/io/nats/client/impl/TLSConnectTests.java index 7011e3a48..a9799923f 100644 --- a/src/test/java/io/nats/client/impl/TLSConnectTests.java +++ b/src/test/java/io/nats/client/impl/TLSConnectTests.java @@ -15,23 +15,33 @@ import io.nats.client.*; import io.nats.client.ConnectionListener.Events; +import io.nats.client.support.NatsUri; import io.nats.client.support.ssl.ExpiringClientCertUtil; import io.nats.client.support.ssl.ExpiringComponents; +import io.nats.client.support.ssl.SniTestServer; import io.nats.client.support.ssl.SslTestingHelper; import io.nats.client.utils.CloseOnUpgradeAttempt; import io.nats.client.utils.ResourceUtils; import io.nats.client.utils.TestBase; +import org.jspecify.annotations.NonNull; import org.junit.jupiter.api.Test; import javax.net.ssl.SSLContext; import javax.net.ssl.SSLException; import java.io.IOException; +import java.net.InetAddress; +import java.net.ServerSocket; import java.net.SocketException; +import java.net.SocketTimeoutException; import java.nio.file.Path; import java.security.NoSuchAlgorithmException; import java.security.cert.CertificateException; import java.time.Duration; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; import java.util.Date; +import java.util.List; import java.util.Properties; import java.util.concurrent.CountDownLatch; import java.util.concurrent.Future; @@ -701,4 +711,181 @@ boolean hasSslOrSocketCauseInChain(Throwable t) { return false; } } -} \ No newline at end of file + + // ---- a ws:// server when the options require TLS ---- + + private static Options.Builder tlsRequiring(int port, SniTestServer.Certificates certificates, ListenerForTesting listener) throws Exception { + // the certificate carries 127.0.0.1 as an ip subject alternative name, so the default hostname verification passes + return new Options.Builder() + .server("tls://127.0.0.1:" + port) + .sslContext(certificates.clientContext()) + .connectionListener(listener) + .errorListener(listener) + .noRandomize() + .connectionTimeout(Duration.ofSeconds(2)); + } + + // a pool that hands out the given server first; it can ignore serverRejected and keep the server, + // and it can hold only that server + private static ServerPool poolWithFirst(NatsUri first, boolean keepsRejected, boolean only) { + return new NatsServerPool() { + @Override + public void initialize(@NonNull Options opts) { + super.initialize(opts); + listLock.lock(); + try { + if (only) { + entryList.clear(); + } + entryList.add(0, new ServerPoolEntry(first, false)); + } + finally { + listLock.unlock(); + } + } + + @Override + public void serverRejected(@NonNull NatsUri nuri) { + if (!keepsRejected) { + super.serverRejected(nuri); + } + } + }; + } + + private static boolean rejectedAsNeverUpgraded(ListenerForTesting.ServerRejectedEvent e, String server) { + return e.server.equals(server) && e.reason.contains("never upgraded to TLS"); + } + + private static void assertNotReportedAsException(ListenerForTesting listener, String server) { + for (Exception e : listener.getExceptions()) { + assertFalse(e.getMessage() != null && e.getMessage().contains(server), e.toString()); + } + } + + @Test + public void testDiscoveredPlainWebsocketServerIsRejectedOnce() throws Exception { + // a ws:// entry in connect_urls enters the pool like any other entry. When the pool first hands it out, + // a connection that requires TLS skips it without dialing it, goes on to the next server, reports it + // through serverRejected, and tells the pool, which drops it and does not take it back from a later INFO + SniTestServer.Certificates certificates = new SniTestServer.Certificates(SniTestServer.HOST, true); + ListenerForTesting listener = new ListenerForTesting(); + try (ServerSocket plainWs = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) { + String wsUrl = "ws://127.0.0.1:" + plainWs.getLocalPort(); + try (SniTestServer second = new SniTestServer(certificates, false, Collections.singletonList(wsUrl), true); + SniTestServer first = new SniTestServer(certificates, false, Arrays.asList(wsUrl, "127.0.0.1:" + second.port()), true)) { + Options options = tlsRequiring(first.port(), certificates, listener) + .reconnectWait(Duration.ofMillis(50)) + .maxReconnects(3) + .build(); + Connection nc = Nats.connect(options); + try { + nc.flush(Duration.ofSeconds(2)); + assertTrue(listener._eventually(2000, listener::getConnectionEvents, e -> e == Events.DISCOVERED_SERVERS)); + Collection servers = nc.getServers(); + assertTrue(servers.contains(wsUrl), servers.toString()); + assertTrue(servers.contains("tls://127.0.0.1:" + second.port()), servers.toString()); + assertTrue(listener.getServerRejectedEvents().isEmpty()); + + // the first server goes away: the pool hands out the ws entry, then the second server + listener.prepForStatusChange(Events.RECONNECTED); + first.close(); + assertTrue(listener.waitForStatusChange(5, TimeUnit.SECONDS)); + assertEquals("tls://127.0.0.1:" + second.port(), nc.getConnectedUrl()); + + assertTrue(listener._eventually(2000, listener::getServerRejectedEvents, e -> rejectedAsNeverUpgraded(e, wsUrl))); + assertEquals(1, listener.getServerRejectedEvents().size(), listener.getServerRejectedEvents().toString()); + assertNotReportedAsException(listener, wsUrl); + // the second server's INFO advertised the ws entry again and the pool did not take it back + servers = nc.getServers(); + assertFalse(servers.contains(wsUrl), servers.toString()); + plainWs.setSoTimeout(500); + assertThrows(SocketTimeoutException.class, plainWs::accept); + } + finally { + nc.close(); + } + } + } + } + + @Test + public void testPlainWebsocketServerFromThePoolIsSkipped() throws Exception { + // a custom ServerPool can hand the connection a ws:// server. With TLS required it is skipped without + // being dialed, reported through serverRejected, and handed back to the pool, which drops it; + // the connection goes on to the next server + SniTestServer.Certificates certificates = new SniTestServer.Certificates(SniTestServer.HOST, true); + ListenerForTesting listener = new ListenerForTesting(); + try (ServerSocket plainWs = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1")); + SniTestServer server = new SniTestServer(certificates, false, 0, true)) { + NatsUri wsUri = new NatsUri("ws://127.0.0.1:" + plainWs.getLocalPort()); + Options options = tlsRequiring(server.port(), certificates, listener).serverPool(poolWithFirst(wsUri, false, false)).build(); + try (Connection nc = Nats.connect(options)) { + nc.flush(Duration.ofSeconds(2)); + assertEquals("tls://127.0.0.1:" + server.port(), nc.getConnectedUrl()); + assertTrue(listener._eventually(2000, listener::getServerRejectedEvents, e -> rejectedAsNeverUpgraded(e, wsUri.toString()))); + assertEquals(1, listener.getServerRejectedEvents().size(), listener.getServerRejectedEvents().toString()); + assertNotReportedAsException(listener, wsUri.toString()); + assertFalse(nc.getServers().contains(wsUri.toString()), nc.getServers().toString()); + plainWs.setSoTimeout(500); + assertThrows(SocketTimeoutException.class, plainWs::accept); + } + } + } + + @Test + public void testConnectEndsWhenThePoolKeepsOnlyARejectedServer() throws Exception { + // a pool that ignores serverRejected keeps handing the ws:// server out. With nothing usable in between, + // the connect ends and fails naming it, instead of spinning on the pool + SniTestServer.Certificates certificates = new SniTestServer.Certificates(SniTestServer.HOST, true); + ListenerForTesting listener = new ListenerForTesting(); + try (ServerSocket plainWs = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) { + NatsUri wsUri = new NatsUri("ws://127.0.0.1:" + plainWs.getLocalPort()); + Options options = tlsRequiring(NatsTestServer.nextPort(), certificates, listener) + .serverPool(poolWithFirst(wsUri, true, true)) + .build(); + IOException e = assertThrows(IOException.class, () -> Nats.connect(options)); + assertTrue(e.getMessage().contains("Unable to connect"), e.getMessage()); + assertTrue(e.getMessage().contains(wsUri.toString()), e.getMessage()); + assertTrue(listener._eventually(2000, listener::getServerRejectedEvents, ev -> rejectedAsNeverUpgraded(ev, wsUri.toString()))); + assertEquals(1, listener.getServerRejectedEvents().size(), listener.getServerRejectedEvents().toString()); + plainWs.setSoTimeout(500); + assertThrows(SocketTimeoutException.class, plainWs::accept); + } + } + + @Test + public void testReconnectEndsWhenThePoolKeepsOnlyARejectedServer() throws Exception { + // the pool keeps the ws:// server next to the TLS server. The connect lands on the TLS server with one + // report. When the TLS server goes away, the reconnect reports the ws:// server once more, uses up the + // TLS server's attempts, and ends when the pool hands the ws:// server back with nothing usable since + SniTestServer.Certificates certificates = new SniTestServer.Certificates(SniTestServer.HOST, true); + ListenerForTesting listener = new ListenerForTesting(); + try (ServerSocket plainWs = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1")); + SniTestServer server = new SniTestServer(certificates, false, 0, true)) { + NatsUri wsUri = new NatsUri("ws://127.0.0.1:" + plainWs.getLocalPort()); + Options options = tlsRequiring(server.port(), certificates, listener) + .serverPool(poolWithFirst(wsUri, true, false)) + .reconnectWait(Duration.ofMillis(50)) + .maxReconnects(2) + .build(); + Connection nc = Nats.connect(options); + try { + nc.flush(Duration.ofSeconds(2)); + assertEquals("tls://127.0.0.1:" + server.port(), nc.getConnectedUrl()); + assertTrue(listener._eventually(2000, listener::getServerRejectedEvents, ev -> rejectedAsNeverUpgraded(ev, wsUri.toString()))); + assertEquals(1, listener.getServerRejectedEvents().size(), listener.getServerRejectedEvents().toString()); + + listener.prepForStatusChange(Events.CLOSED); + server.close(); + assertTrue(listener.waitForStatusChange(10, TimeUnit.SECONDS)); + assertEquals(2, listener.getServerRejectedEvents().size(), listener.getServerRejectedEvents().toString()); + plainWs.setSoTimeout(500); + assertThrows(SocketTimeoutException.class, plainWs::accept); + } + finally { + nc.close(); + } + } + } +} diff --git a/src/test/java/io/nats/client/impl/TlsHostnameTests.java b/src/test/java/io/nats/client/impl/TlsHostnameTests.java index 9b8aa5a26..a0da492f6 100644 --- a/src/test/java/io/nats/client/impl/TlsHostnameTests.java +++ b/src/test/java/io/nats/client/impl/TlsHostnameTests.java @@ -133,7 +133,7 @@ private static Options.Builder options(SniTestServer server, String host, Diagno .server("tls://" + host + ":" + server.port()) .serverPool(new NatsServerPool() { @Override - public List resolveHostToIps(String hostToResolve, boolean maxOneResult, boolean includeIPV6) { + public List resolveHostToIps(@NonNull String hostToResolve, boolean maxOneResult, boolean includeIPV6) { poolAsked.set(true); assertEquals(host, hostToResolve); return Collections.singletonList("127.0.0.1"); @@ -514,7 +514,7 @@ private static Options.Builder verifying(SniTestServer server, String host, SSLC .server("tls://" + host + ":" + server.port()) .serverPool(new NatsServerPool() { @Override - public List resolveHostToIps(String hostToResolve, boolean maxOneResult, boolean includeIPV6) { + public List resolveHostToIps(@NonNull String hostToResolve, boolean maxOneResult, boolean includeIPV6) { return Collections.singletonList("127.0.0.1"); } }) diff --git a/src/test/java/io/nats/client/support/ssl/SniTestServer.java b/src/test/java/io/nats/client/support/ssl/SniTestServer.java index f00224a61..042bb3338 100644 --- a/src/test/java/io/nats/client/support/ssl/SniTestServer.java +++ b/src/test/java/io/nats/client/support/ssl/SniTestServer.java @@ -39,6 +39,7 @@ import java.security.Principal; import java.security.PrivateKey; import java.security.cert.X509Certificate; +import java.util.Collections; import java.util.Date; import java.util.List; import java.util.concurrent.CompletableFuture; @@ -127,6 +128,13 @@ public SniTestServer(Certificates certificates, boolean tlsFirst, int discovered * for tests about what the client does with a certificate rather than how the server selects it */ public SniTestServer(Certificates certificates, boolean tlsFirst, int discoveredPort, boolean alwaysServeValid) throws Exception { + this(certificates, tlsFirst, discoveredPort == 0 ? Collections.emptyList() : Collections.singletonList("127.0.0.1:" + discoveredPort), alwaysServeValid); + } + + /** + * @param discovered the connect_urls entries to send in INFO, exactly as given; empty sends no connect_urls + */ + public SniTestServer(Certificates certificates, boolean tlsFirst, List discovered, boolean alwaysServeValid) throws Exception { listener = new ServerSocket(0, 10, InetAddress.getByName("127.0.0.1")); SSLContext context = SSLContext.getInstance("TLSv1.2"); context.init(new KeyManager[]{new X509ExtendedKeyManager() { @@ -167,7 +175,7 @@ public X509Certificate[] getCertificateChain(String alias) { while (!closed) { Socket raw = listener.accept(); sockets.add(raw); - executor.submit(() -> serve(raw, context, tlsFirst, discoveredPort)); + executor.submit(() -> serve(raw, context, tlsFirst, discovered)); } } catch (IOException e) { @@ -180,8 +188,8 @@ public X509Certificate[] getCertificateChain(String alias) { public int port() { return listener.getLocalPort(); } - private void serve(Socket raw, SSLContext context, boolean tlsFirst, int discoveredPort) { - String urls = discoveredPort == 0 ? "" : ",\"connect_urls\":[\"127.0.0.1:" + discoveredPort + "\"]"; + private void serve(Socket raw, SSLContext context, boolean tlsFirst, List discovered) { + String urls = discovered.isEmpty() ? "" : ",\"connect_urls\":[\"" + String.join("\",\"", discovered) + "\"]"; String info = "INFO {\"server_id\":\"sni-test\",\"version\":\"2.11.0\",\"proto\":1," + "\"max_payload\":1048576,\"tls_required\":true" + urls + "}\r\n"; try (Socket ignored = raw) {