Introduce integration test for TLS session tickets

Adds isolated SessionTicketsIT which checks if session tickets mechanism
behaves as expected. It does that by watching Java's SSL debug logs and
ensuring that specific substrings appear in them.

Since the server supports tickets with TLSv1.3, only that version is tested.
In that version the session resumption without server-side state is done through
pre-shared keys. The details are described in rfc8446 (see section 2.2).

Author: Bouncheck

integration-tests/src/test/java/com/datastax/oss/driver/core/ssl/SessionTicketsIT.java

@@ -0,0 +1,248 @@
+package com.datastax.oss.driver.core.ssl;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import com.datastax.oss.driver.api.core.CqlSession;
+import com.datastax.oss.driver.api.testinfra.ccm.CcmBridge;
+import com.datastax.oss.driver.api.testinfra.ccm.CustomCcmRule;
+import com.datastax.oss.driver.api.testinfra.requirement.BackendRequirement;
+import com.datastax.oss.driver.api.testinfra.requirement.BackendType;
+import com.datastax.oss.driver.api.testinfra.session.SessionUtils;
+import com.datastax.oss.driver.categories.IsolatedTests;
+import com.google.common.collect.ImmutableList;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.security.KeyStore;
+import java.security.SecureRandom;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.logging.Handler;
+import java.util.logging.Level;
+import java.util.logging.LogRecord;
+import java.util.logging.Logger;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+@Category(IsolatedTests.class)
+@BackendRequirement(
+    type = BackendType.SCYLLA,
+    minInclusive = "2025.2.0",
+    description = "Requires server side support for session tickets")
+public class SessionTicketsIT {
+  @ClassRule
+  public static final CustomCcmRule CCM_RULE =
+      CustomCcmRule.builder()
+          .withSsl()
+          .withCassandraConfiguration("client_encryption_options.enable_session_tickets", "true")
+          .build();
+
+  @Test
+  public void should_use_session_tickets_when_TLSv13() throws Exception {
+    // This test relies on parsing SSL debug logs which may not be identical across different
+    // Java versions. Additionally, the expected number of specific messages depends on default
+    // driver behavior.
+    // Currently the driver negotiates the protocol version starting from DSE_V2 down to V4, and
+    // does not
+    // try to use V6. If that ever changes or Scylla starts supporting V5 or starts to send
+    // different
+    // number of tickets after hello, the expected counts will need to be updated.
+
+    // Set up SSL debug logging
+    System.setProperty("javax.net.debug", "");
+    Logger sslLogger = Logger.getLogger("javax.net.ssl");
+    Level originalLevel = sslLogger.getLevel();
+    sslLogger.setLevel(Level.ALL);
+
+    final int expectedNumOfClientHellos =
+        6; // 4 come from ControlConnection (DSE_V2, DSE_V1, V5, V4 protocols) + 2 from dedicated
+    // Channels to 2 shards
+
+    List<OccurrenceCounter> counters =
+        ImmutableList.of(
+            new OccurrenceCounter(
+                "Consuming CertificateVerify handshake message", count -> count == 1),
+            // It looks like server currently sends two tickets after each handshake
+            new OccurrenceCounter(
+                "Consuming NewSessionTicket", count -> count == expectedNumOfClientHellos * 2),
+            new OccurrenceCounter(
+                "Produced ClientHello handshake message",
+                count -> count == expectedNumOfClientHellos),
+            new OccurrenceCounter(
+                "Consuming ServerHello handshake message",
+                count -> count == expectedNumOfClientHellos),
+            new OccurrenceCounter(
+                "Try resuming session", count -> count == expectedNumOfClientHellos - 1),
+            new OccurrenceCounter(
+                "Consumed extension: pre_shared_key",
+                count -> count == expectedNumOfClientHellos - 1),
+            new OccurrenceCounter(
+                "Resuming session:", count -> count == expectedNumOfClientHellos - 1),
+            new OccurrenceCounter(
+                "Using PSK to derive early secret",
+                count -> count == expectedNumOfClientHellos - 1),
+            new OccurrenceCounter(
+                "Negotiated protocol version: TLSv1.3",
+                count -> count == expectedNumOfClientHellos));
+
+    // Custom handler to capture log messages
+    ByteArrayOutputStream logCapture = new ByteArrayOutputStream();
+    TlsDebugLogHandler handler = new TlsDebugLogHandler(logCapture, counters);
+    sslLogger.setUseParentHandlers(false);
+    sslLogger.addHandler(handler);
+
+    try {
+      SSLContext context = createSslContext("TLSv1.3");
+      try (CqlSession session =
+          (CqlSession)
+              SessionUtils.baseBuilder()
+                  .addContactEndPoints(CCM_RULE.getContactPoints())
+                  .withSslContext(context)
+                  .build()) {
+        session.execute("select * from system.local where key='local'");
+      }
+      List<String> errors = new ArrayList<>();
+      for (OccurrenceCounter counter : counters) {
+        if (!counter.isWithinNorm()) {
+          errors.add(
+              "Counter not within norm: "
+                  + counter
+                  + ", inspect the function passed to constructor at "
+                  + counter.getCreationLocation());
+        }
+      }
+      if (!errors.isEmpty()) {
+        StringBuilder sb = new StringBuilder();
+        sb.append("One or more assertions failed:\n");
+        for (String error : errors) {
+          sb.append(error).append("\n");
+        }
+        throw new AssertionError(sb.toString());
+      }
+    } finally {
+      sslLogger.removeHandler(handler);
+      sslLogger.setLevel(originalLevel);
+    }
+  }
+
+  private SSLContext createSslContext(String protocol) throws Exception {
+    SSLContext context = SSLContext.getInstance(protocol);
+
+    TrustManagerFactory tmf =
+        TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
+    try (InputStream tsf =
+        Files.newInputStream(
+            Paths.get(CcmBridge.DEFAULT_CLIENT_TRUSTSTORE_FILE.getAbsolutePath()))) {
+      KeyStore ts = KeyStore.getInstance("JKS");
+      char[] password = CcmBridge.DEFAULT_CLIENT_TRUSTSTORE_PASSWORD.toCharArray();
+      ts.load(tsf, password);
+      tmf.init(ts);
+    }
+
+    KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+    try (InputStream ksf =
+        Files.newInputStream(Paths.get(CcmBridge.DEFAULT_CLIENT_KEYSTORE_FILE.getAbsolutePath()))) {
+      KeyStore ks = KeyStore.getInstance("JKS");
+      char[] password = CcmBridge.DEFAULT_CLIENT_KEYSTORE_PASSWORD.toCharArray();
+      ks.load(ksf, password);
+      kmf.init(ks, password);
+    }
+
+    context.init(kmf.getKeyManagers(), tmf.getTrustManagers(), new SecureRandom());
+    return context;
+  }
+
+  static class TlsDebugLogHandler extends Handler {
+    private final ByteArrayOutputStream outputStream;
+    private final List<OccurrenceCounter> counters;
+
+    TlsDebugLogHandler(ByteArrayOutputStream outputStream, List<OccurrenceCounter> counters) {
+      this.outputStream = outputStream;
+      this.counters = counters;
+    }
+
+    @Override
+    public void publish(LogRecord record) {
+      try {
+        for (OccurrenceCounter counter : counters) {
+          counter.incrementIfFound(record.getMessage());
+        }
+        outputStream.write((record.getMessage() + "\n").getBytes(UTF_8));
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
+    @Override
+    public void flush() {
+      try {
+        outputStream.flush();
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+
+    @Override
+    public void close() throws SecurityException {
+      try {
+        outputStream.close();
+      } catch (IOException e) {
+        throw new RuntimeException(e);
+      }
+    }
+  }
+
+  static class OccurrenceCounter {
+    private final AtomicInteger count = new AtomicInteger(0);
+    private final String substring; // Exact substring to look for
+    private final Function<Integer, Boolean> isWithinNorm;
+    private final StackTraceElement creationLocation;
+
+    public OccurrenceCounter(String substring, Function<Integer, Boolean> isWithinNorm) {
+      this.substring = substring;
+      this.isWithinNorm = isWithinNorm;
+      this.creationLocation = new Throwable().getStackTrace()[1]; // [1] gets the caller's location
+    }
+
+    /**
+     * Increment the counter if the substring is found in the log line. Multiple occurrences count
+     * as one.
+     *
+     * @param logLine log line to check
+     */
+    public void incrementIfFound(String logLine) {
+      if (logLine.contains(substring)) {
+        count.incrementAndGet();
+      }
+    }
+
+    public int get() {
+      return count.get();
+    }
+
+    public boolean isWithinNorm() {
+      return isWithinNorm.apply(count.get());
+    }
+
+    public String getSubstring() {
+      return substring;
+    }
+
+    public String getCreationLocation() {
+      return creationLocation.toString();
+    }
+
+    @Override
+    public String toString() {
+      return "OccurrenceCounter{substring='" + substring + "', count=" + count.get() + "}";
+    }
+  }
+}