org.junit.jupiter
junit-jupiter
diff --git a/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlWebSocketClient.java b/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlWebSocketClient.java
index 94a1acb..660d449 100644
--- a/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlWebSocketClient.java
+++ b/libsql-client/src/main/java/com/github/magnusp/libsql/client/LibsqlWebSocketClient.java
@@ -11,30 +11,27 @@
import com.github.magnusp.libsql.hrana.codec.WsResponse;
import com.github.magnusp.libsql.hrana.codec.WsServerMsg;
-import javax.net.ssl.SSLParameters;
-import javax.net.ssl.SSLSocket;
-import javax.net.ssl.SSLSocketFactory;
+import org.apache.hc.client5.http.websocket.api.WebSocket;
+import org.apache.hc.client5.http.websocket.api.WebSocketClientConfig;
+import org.apache.hc.client5.http.websocket.api.WebSocketListener;
+import org.apache.hc.client5.http.websocket.client.CloseableWebSocketClient;
+import org.apache.hc.client5.http.websocket.client.WebSocketClients;
+import org.apache.hc.core5.io.CloseMode;
+import org.apache.hc.core5.util.Timeout;
+
import java.io.ByteArrayOutputStream;
import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.InetSocketAddress;
-import java.net.Socket;
import java.net.URI;
+import java.nio.ByteBuffer;
+import java.nio.CharBuffer;
import java.nio.charset.StandardCharsets;
-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-import java.security.SecureRandom;
import java.time.Duration;
-import java.util.Base64;
-import java.util.HashMap;
-import java.util.Locale;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
@@ -52,34 +49,32 @@
* All request methods block the calling thread until the matching response arrives. The caller
* (the JDBC connection) serializes calls, so a single interactive stream id is enough.
*
- *
Transport note. The WebSocket is spoken directly over a socket rather than
- * through {@link java.net.http.WebSocket}. sqld answers the upgrade with a {@code 101} that carries
- * {@code content-length: 31} and no body, and the JDK client completes its opening handshake only
- * after reading a full response body ({@code OpeningHandshake.send()} uses {@code
- * BodyHandlers.ofString()}), so it hangs against every sqld version. The framing below is the RFC
- * 6455 client subset Hrana needs: text/binary frames, fragmentation, ping/pong and close.
+ *
Transport. Apache HttpClient's {@code httpclient5-websocket} module speaks RFC
+ * 6455. The JDK's {@link java.net.http.WebSocket} cannot be used against sqld: sqld answers the
+ * upgrade with a {@code 101} that carries {@code content-length: 31} and no body, and the JDK client
+ * completes its opening handshake only after reading a full response body. HttpClient treats a
+ * {@code 101} as informational and upgrades without reading one.
*/
public class LibsqlWebSocketClient implements LibsqlClient {
private static final String SUBPROTOCOL_JSON = "hrana3";
private static final String SUBPROTOCOL_PROTOBUF = "hrana3-protobuf";
- private static final String WEBSOCKET_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11";
-
- private static final int OP_CONTINUATION = 0x0;
- private static final int OP_TEXT = 0x1;
- private static final int OP_BINARY = 0x2;
- private static final int OP_CLOSE = 0x8;
- private static final int OP_PING = 0x9;
- private static final int OP_PONG = 0xA;
private static final int CLOSE_NORMAL = 1000;
- private static final int MAX_FRAME_BYTES = 1024 * 1024 * 1024;
+ /**
+ * One message must be buffered whole before it can be decoded, so the bound matches the payload
+ * size a Hrana response may reach.
+ */
+ private static final int MAX_MESSAGE_BYTES = 1024 * 1024 * 1024;
+
+ private static final Duration CLOSE_TIMEOUT = Duration.ofSeconds(1);
private final LibsqlClientConfig config;
private final HranaWsCodec codec;
private final String subprotocol;
private final Duration timeout;
- private final Transport transport;
+ private final CloseableWebSocketClient client;
+ private WebSocket socket;
private final ConcurrentHashMap> pending =
new ConcurrentHashMap<>();
@@ -103,13 +98,16 @@ public LibsqlWebSocketClient(LibsqlClientConfig config, HranaWsCodec codec) {
? config.connectTimeout()
: LibsqlClientConfig.DEFAULT_CONNECT_TIMEOUT;
this.subprotocol = codec.isBinary() ? SUBPROTOCOL_PROTOBUF : SUBPROTOCOL_JSON;
- this.transport = new Transport(URI.create(config.url()), timeout, subprotocol);
+ CloseableWebSocketClient started = WebSocketClients.createDefault();
+ started.start();
+ this.client = started;
try {
+ this.socket = connect();
send(new WsClientMsg.Hello(authToken()));
awaitHello();
} catch (RuntimeException e) {
- transport.abort();
+ shutdown();
throw e;
}
}
@@ -119,6 +117,34 @@ private String authToken() {
return token == null || token.isBlank() ? null : token;
}
+ private WebSocket connect() {
+ WebSocketClientConfig webSocketConfig = WebSocketClientConfig.custom()
+ .setSubprotocols(List.of(subprotocol))
+ .setConnectTimeout(Timeout.of(timeout))
+ .setCloseWaitTimeout(Timeout.of(CLOSE_TIMEOUT))
+ .setMaxFrameSize(MAX_MESSAGE_BYTES)
+ .setMaxMessageSize(MAX_MESSAGE_BYTES)
+ // The transport this replaces negotiated no extension, so keep the wire bytes equal.
+ .enablePerMessageDeflate(false)
+ .build();
+ try {
+ return client.connect(
+ URI.create(config.url()), new HranaListener(), webSocketConfig)
+ .get(timeout.toMillis(), TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new LibsqlException("Interrupted while opening the Hrana WebSocket", e);
+ } catch (TimeoutException e) {
+ throw new LibsqlException(
+ "Timed out after " + timeout.toMillis() + " ms opening the Hrana WebSocket", e);
+ } catch (ExecutionException e) {
+ Throwable cause = e.getCause() != null ? e.getCause() : e;
+ throw new LibsqlException(
+ "Failed to open Hrana WebSocket to " + config.url() + ": " + cause.getMessage(),
+ cause);
+ }
+ }
+
@Override
public StmtResult executeOneShot(Stmt stmt) {
int streamId = nextStreamId.getAndIncrement();
@@ -207,7 +233,7 @@ public synchronized void close() {
// Closing a connection is best effort: the peer may already be gone and the server
// reaps the stream when it notices.
} finally {
- transport.close(CLOSE_NORMAL, "closing");
+ shutdown();
}
}
@@ -241,7 +267,24 @@ private void send(WsClientMsg message) {
} catch (IOException e) {
throw new LibsqlException("Failed to serialize Hrana message: " + e.getMessage(), e);
}
- transport.send(payload, codec.isBinary());
+ boolean queued = codec.isBinary()
+ ? socket.sendBinary(ByteBuffer.wrap(payload), true)
+ : socket.sendText(new String(payload, StandardCharsets.UTF_8), true);
+ if (!queued) {
+ throw new LibsqlException("Hrana WebSocket is closed", "WS_CLOSED", 0);
+ }
+ }
+
+ /**
+ * Stops the IO reactor, the executor, and the connection pool in one step. The socket close
+ * frame is queued first, but sqld never echoes it, so the teardown does not wait for the
+ * handshake.
+ */
+ private void shutdown() {
+ if (socket != null) {
+ socket.close(CLOSE_NORMAL, "closing");
+ }
+ client.close(CloseMode.IMMEDIATE);
}
private void awaitHello() {
@@ -302,7 +345,7 @@ private void dispatch(byte[] payload) {
} catch (IOException e) {
failAll(new LibsqlException(
"Failed to decode Hrana WebSocket message: " + e.getMessage(), e));
- transport.abort();
+ socket.close(CLOSE_NORMAL, "protocol violation");
return;
}
@@ -334,7 +377,7 @@ private void fail(int requestId, LibsqlException error) {
private void protocolViolation(String detail) {
failAll(new LibsqlException("Hrana WebSocket protocol violation: " + detail, "WS_PROTOCOL", 0));
- transport.abort();
+ socket.close(CLOSE_NORMAL, "protocol violation");
}
private void failAll(LibsqlException error) {
@@ -350,352 +393,56 @@ private void failAll(LibsqlException error) {
}
/**
- * A minimal RFC 6455 client connection: the opening handshake plus frame encoding and decoding.
+ * Reassembles frames into whole messages and routes them to {@link #dispatch(byte[])}.
*
- * A reader thread routes complete text/binary messages to {@link #dispatch(String)} and
- * fails the pending futures when the connection ends. It is a virtual thread, so an idle
- * connection never keeps the JVM alive.
+ *
The buffers the client passes in are only valid during the callback, so every chunk is
+ * copied before it is buffered. A message either arrives whole or in fragments ending with
+ * {@code last}, and the fragments of one message never interleave with another.
*/
- private final class Transport {
-
- private Socket socket;
- private InputStream in;
- private OutputStream out;
- private final Object writeLock = new Object();
-
- private volatile boolean open = true;
-
- private Transport(URI uri, Duration timeout, String subprotocol) {
- Socket connected = null;
- try {
- connected = connect(uri, timeout);
- this.socket = connected;
- this.in = connected.getInputStream();
- this.out = connected.getOutputStream();
- handshake(uri, in, out, timeout, subprotocol);
- } catch (IOException e) {
- closeQuietly(connected);
- throw new LibsqlException(
- "Failed to open Hrana WebSocket to " + uri + ": " + e.getMessage(), e);
- }
- Thread.ofVirtual().name("hrana-ws-reader").start(this::readLoop);
- }
-
- private void send(byte[] payload, boolean binary) {
- sendFrame(binary ? OP_BINARY : OP_TEXT, payload);
- }
+ private final class HranaListener implements WebSocketListener {
- private void close(int statusCode, String reason) {
- if (open) {
- try {
- sendFrame(OP_CLOSE, closePayload(statusCode, reason));
- } catch (LibsqlException e) {
- // The peer is already gone; closing the socket below is enough.
- }
- }
- open = false;
- failAll(new LibsqlException(
- "Hrana WebSocket closed with code " + statusCode, "WS_CLOSED", 0));
- closeQuietly(socket);
- }
-
- private void abort() {
- open = false;
- closeQuietly(socket);
- }
+ private final StringBuilder text = new StringBuilder();
+ private ByteArrayOutputStream binary;
- private void sendFrame(int opcode, byte[] payload) {
- synchronized (writeLock) {
- if (!open) {
- throw new LibsqlException("Hrana WebSocket is closed", "WS_CLOSED", 0);
- }
- try {
- out.write(0x80 | opcode);
- int length = payload.length;
- if (length <= 125) {
- out.write(0x80 | length);
- } else if (length <= 0xFFFF) {
- out.write(0x80 | 126);
- out.write(length >>> 8);
- out.write(length);
- } else {
- out.write(0x80 | 127);
- for (int shift = 56; shift >= 0; shift -= 8) {
- out.write((int) ((long) length >>> shift) & 0xFF);
- }
- }
- byte[] mask = new byte[4];
- ThreadLocalRandom.current().nextBytes(mask);
- out.write(mask);
- for (int i = 0; i < length; i++) {
- out.write(payload[i] ^ mask[i & 3]);
- }
- out.flush();
- } catch (IOException e) {
- open = false;
- throw new LibsqlException(
- "Failed to send Hrana message: " + e.getMessage(), e);
- }
+ @Override
+ public void onText(CharBuffer data, boolean last) {
+ text.append(data);
+ if (!last) {
+ return;
}
+ String message = text.toString();
+ text.setLength(0);
+ dispatch(message.getBytes(StandardCharsets.UTF_8));
}
- private void readLoop() {
- ByteArrayOutputStream fragments = new ByteArrayOutputStream();
- int fragmentedOpcode = -1;
- try {
- while (open) {
- int first = in.read();
- if (first < 0) {
- break;
- }
- int second = readByte(in);
- boolean fin = (first & 0x80) != 0;
- int opcode = first & 0x0F;
- if ((second & 0x80) != 0) {
- protocolViolation("server sent a masked frame");
- return;
- }
- long length = second & 0x7F;
- if (length == 126) {
- length = (readByte(in) << 8) | readByte(in);
- } else if (length == 127) {
- length = 0;
- for (int i = 0; i < 8; i++) {
- length = (length << 8) | readByte(in);
- }
- }
- if (length < 0 || length > MAX_FRAME_BYTES) {
- protocolViolation("frame is too large: " + length + " bytes");
- return;
- }
- byte[] payload = in.readNBytes((int) length);
- if (payload.length != length) {
- break;
- }
-
- switch (opcode) {
- case OP_TEXT, OP_BINARY -> {
- if (fragmentedOpcode != -1) {
- protocolViolation("new message before the fragmented one finished");
- return;
- }
- if (fin) {
- deliver(opcode, payload);
- } else {
- fragmentedOpcode = opcode;
- fragments.write(payload);
- }
- }
- case OP_CONTINUATION -> {
- if (fragmentedOpcode == -1) {
- protocolViolation("continuation without a started message");
- return;
- }
- fragments.write(payload);
- if (fin) {
- byte[] complete = fragments.toByteArray();
- int messageOpcode = fragmentedOpcode;
- fragments.reset();
- fragmentedOpcode = -1;
- deliver(messageOpcode, complete);
- }
- }
- case OP_PING -> sendFrame(OP_PONG, payload);
- case OP_PONG -> {
- // Unsolicited pongs are allowed and need no answer.
- }
- case OP_CLOSE -> {
- handlePeerClose(payload);
- return;
- }
- default -> {
- protocolViolation("unknown frame opcode 0x" + Integer.toHexString(opcode));
- return;
- }
- }
- }
- if (open) {
- connectionLost("connection closed by the server without a close frame");
- }
- } catch (IOException e) {
- if (open) {
- connectionLost("Hrana WebSocket connection failed: " + e.getMessage());
- }
- } finally {
- open = false;
- closeQuietly(socket);
+ @Override
+ public void onBinary(ByteBuffer data, boolean last) {
+ if (binary == null) {
+ binary = new ByteArrayOutputStream();
}
- }
-
- private void deliver(int opcode, byte[] payload) {
- boolean expectsBinary = codec.isBinary();
- if ((opcode == OP_BINARY) != expectsBinary) {
- protocolViolation("frame type does not match the negotiated encoding");
+ byte[] chunk = new byte[data.remaining()];
+ data.get(chunk);
+ binary.write(chunk, 0, chunk.length);
+ if (!last) {
return;
}
- dispatch(payload);
+ byte[] message = binary.toByteArray();
+ binary = null;
+ dispatch(message);
}
- private void handlePeerClose(byte[] payload) {
- int statusCode = payload.length >= 2
- ? ((payload[0] & 0xFF) << 8) | (payload[1] & 0xFF)
- : CLOSE_NORMAL;
- String reason = payload.length > 2
- ? new String(payload, 2, payload.length - 2, StandardCharsets.UTF_8)
- : "";
- try {
- // Echo the close frame while the connection is still writable.
- sendFrame(OP_CLOSE, closePayload(statusCode, ""));
- } catch (LibsqlException e) {
- // The peer closed; the socket is closed by the reader's finally block.
- }
- open = false;
- connectionLost(reason.isBlank()
- ? "Hrana WebSocket closed with code " + statusCode
- : "Hrana WebSocket closed with code " + statusCode + ": " + reason);
+ @Override
+ public void onError(Throwable cause) {
+ failAll(new LibsqlException(
+ "Hrana WebSocket connection failed: " + cause.getMessage(), cause));
}
- private void connectionLost(String detail) {
+ @Override
+ public void onClose(int statusCode, String reason) {
+ String detail = reason == null || reason.isBlank()
+ ? "Hrana WebSocket closed with code " + statusCode
+ : "Hrana WebSocket closed with code " + statusCode + ": " + reason;
failAll(new LibsqlException(detail, "WS_CLOSED", 0));
}
}
-
- private static Socket connect(URI uri, Duration timeout) throws IOException {
- boolean tls = "wss".equalsIgnoreCase(uri.getScheme());
- String host = uri.getHost();
- if (host == null) {
- throw new IOException("WebSocket URL has no host: " + uri);
- }
- int port = uri.getPort() > 0 ? uri.getPort() : (tls ? 443 : 80);
-
- Socket socket = new Socket();
- try {
- int connectTimeout = (int) Math.min(Integer.MAX_VALUE, Math.max(0, timeout.toMillis()));
- socket.connect(new InetSocketAddress(host, port), connectTimeout);
- socket.setTcpNoDelay(true);
- if (!tls) {
- return socket;
- }
- SSLSocketFactory factory = (SSLSocketFactory) SSLSocketFactory.getDefault();
- SSLSocket ssl = (SSLSocket) factory.createSocket(socket, host, port, true);
- SSLParameters parameters = ssl.getSSLParameters();
- parameters.setEndpointIdentificationAlgorithm("HTTPS");
- ssl.setSSLParameters(parameters);
- ssl.startHandshake();
- return ssl;
- } catch (IOException e) {
- closeQuietly(socket);
- throw e;
- }
- }
-
- private static void handshake(URI uri, InputStream in, OutputStream out, Duration timeout,
- String subprotocol) throws IOException {
- String key = createKey();
- String expectedAccept = acceptKey(key);
- String path = uri.getRawPath() == null || uri.getRawPath().isEmpty() ? "/" : uri.getRawPath();
- if (uri.getRawQuery() != null && !uri.getRawQuery().isEmpty()) {
- path = path + "?" + uri.getRawQuery();
- }
- String host = uri.getHost();
- int port = uri.getPort();
- String hostHeader = port > 0 ? host + ":" + port : host;
-
- StringBuilder request = new StringBuilder(256);
- request.append("GET ").append(path).append(" HTTP/1.1\r\n")
- .append("Host: ").append(hostHeader).append("\r\n")
- .append("Upgrade: websocket\r\n")
- .append("Connection: Upgrade\r\n")
- .append("Sec-WebSocket-Key: ").append(key).append("\r\n")
- .append("Sec-WebSocket-Version: 13\r\n")
- .append("Sec-WebSocket-Protocol: ").append(subprotocol).append("\r\n")
- .append("\r\n");
- out.write(request.toString().getBytes(StandardCharsets.US_ASCII));
- out.flush();
-
- String statusLine = readLine(in);
- if (statusLine == null || !statusLine.contains(" 101 ")) {
- throw new IOException("WebSocket upgrade rejected: " + statusLine);
- }
- Map headers = new HashMap<>();
- String line;
- while ((line = readLine(in)) != null && !line.isEmpty()) {
- int colon = line.indexOf(':');
- if (colon > 0) {
- headers.put(line.substring(0, colon).trim().toLowerCase(Locale.ROOT),
- line.substring(colon + 1).trim());
- }
- }
- if (!"websocket".equalsIgnoreCase(headers.get("upgrade"))) {
- throw new IOException("WebSocket upgrade response missing Upgrade: websocket");
- }
- if (!expectedAccept.equals(headers.get("sec-websocket-accept"))) {
- throw new IOException("WebSocket upgrade response has an invalid Sec-WebSocket-Accept");
- }
- String negotiated = headers.get("sec-websocket-protocol");
- if (!subprotocol.equals(negotiated)) {
- throw new IOException("Server did not negotiate subprotocol " + subprotocol
- + " (got " + negotiated + ")");
- }
- }
-
- private static String createKey() {
- byte[] nonce = new byte[16];
- new SecureRandom().nextBytes(nonce);
- return Base64.getEncoder().encodeToString(nonce);
- }
-
- private static String acceptKey(String key) {
- try {
- MessageDigest sha1 = MessageDigest.getInstance("SHA-1");
- byte[] digest = sha1.digest((key + WEBSOCKET_GUID).getBytes(StandardCharsets.US_ASCII));
- return Base64.getEncoder().encodeToString(digest);
- } catch (NoSuchAlgorithmException e) {
- throw new IllegalStateException("SHA-1 is required by the WebSocket protocol", e);
- }
- }
-
- private static byte[] closePayload(int statusCode, String reason) {
- byte[] reasonBytes = reason == null ? new byte[0] : reason.getBytes(StandardCharsets.UTF_8);
- byte[] payload = new byte[2 + reasonBytes.length];
- payload[0] = (byte) (statusCode >>> 8);
- payload[1] = (byte) statusCode;
- System.arraycopy(reasonBytes, 0, payload, 2, reasonBytes.length);
- return payload;
- }
-
- private static String readLine(InputStream in) throws IOException {
- StringBuilder line = new StringBuilder();
- int b;
- while ((b = in.read()) >= 0) {
- if (b == '\n') {
- int length = line.length();
- if (length > 0 && line.charAt(length - 1) == '\r') {
- line.setLength(length - 1);
- }
- return line.toString();
- }
- line.append((char) b);
- }
- return line.isEmpty() ? null : line.toString();
- }
-
- private static int readByte(InputStream in) throws IOException {
- int b = in.read();
- if (b < 0) {
- throw new IOException("Unexpected end of stream");
- }
- return b;
- }
-
- private static void closeQuietly(Socket socket) {
- if (socket == null) {
- return;
- }
- try {
- socket.close();
- } catch (IOException ignored) {
- // Nothing useful can be done with a failure to close.
- }
- }
}
diff --git a/libsql-itests/pom.xml b/libsql-itests/pom.xml
index 737baad..6391406 100644
--- a/libsql-itests/pom.xml
+++ b/libsql-itests/pom.xml
@@ -60,6 +60,15 @@
org.apache.maven.plugins
maven-failsafe-plugin
+
+
+
+
+ ${project.build.testOutputDirectory}/tls/truststore.p12
+
+ changeit
+
+
diff --git a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java
index 6025e8e..6042057 100644
--- a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java
+++ b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/SqldContainerBase.java
@@ -1,10 +1,12 @@
package com.github.magnusp.libsql.itest;
-import org.junit.jupiter.api.BeforeAll;
import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.Network;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.MountableFile;
import java.time.Duration;
@@ -12,7 +14,11 @@
public abstract class SqldContainerBase {
static final String LIBSQL_IMAGE = requireLibsqlImage();
+ private static final String NGINX_IMAGE = "nginx:1.27-alpine";
private static final int PORT = 8080;
+ private static final int TLS_PORT = 8443;
+
+ private static final Network NETWORK = Network.newNetwork();
private static String requireLibsqlImage() {
String image = System.getProperty("libsql.image");
@@ -26,6 +32,8 @@ private static String requireLibsqlImage() {
@Container
public static final GenericContainer LIBSQL_CONTAINER = new GenericContainer<>(LIBSQL_IMAGE)
+ .withNetwork(NETWORK)
+ .withNetworkAliases("sqld")
.withExposedPorts(PORT)
.withEnv("SQLD_NODE", "primary")
.withEnv("SQLD_DISABLE_INTELLIGENT_THROTTLING", "true")
@@ -35,18 +43,35 @@ private static String requireLibsqlImage() {
.forStatusCode(405)
.withStartupTimeout(Duration.ofSeconds(60)));
- @BeforeAll
- public static void startContainer() {
+ /**
+ * A TLS terminator in front of sqld, so the {@code wss://} tests exercise a real handshake.
+ * Started on demand: only the WebSocket tests need it.
+ */
+ private static final GenericContainer TLS_PROXY = new GenericContainer<>(DockerImageName.parse(NGINX_IMAGE))
+ .withNetwork(NETWORK)
+ .withExposedPorts(TLS_PORT)
+ .withCopyFileToContainer(MountableFile.forClasspathResource("tls/nginx.conf"), "/etc/nginx/nginx.conf")
+ .withCopyFileToContainer(
+ MountableFile.forClasspathResource("tls/server.crt.pem"), "/etc/nginx/tls/server.crt.pem")
+ .withCopyFileToContainer(
+ MountableFile.forClasspathResource("tls/server.key.pem"), "/etc/nginx/tls/server.key.pem")
+ .waitingFor(Wait.forListeningPort().withStartupTimeout(Duration.ofSeconds(60)));
+
+ static String getWebSocketUrl() {
+ return "ws://" + LIBSQL_CONTAINER.getHost() + ":" + LIBSQL_CONTAINER.getMappedPort(PORT);
+ }
+
+ static synchronized String getSecureWebSocketUrl() {
if (!LIBSQL_CONTAINER.isRunning()) {
LIBSQL_CONTAINER.start();
}
+ if (!TLS_PROXY.isRunning()) {
+ TLS_PROXY.start();
+ }
+ return "wss://" + TLS_PROXY.getHost() + ":" + TLS_PROXY.getMappedPort(TLS_PORT);
}
- public static String getHttpUrl() {
+ static String getHttpUrl() {
return "http://" + LIBSQL_CONTAINER.getHost() + ":" + LIBSQL_CONTAINER.getMappedPort(PORT);
}
-
- public static String getWebSocketUrl() {
- return "ws://" + LIBSQL_CONTAINER.getHost() + ":" + LIBSQL_CONTAINER.getMappedPort(PORT);
- }
}
diff --git a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketProtobufTransportIT.java b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketProtobufTransportIT.java
index 1d82ff3..2689554 100644
--- a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketProtobufTransportIT.java
+++ b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketProtobufTransportIT.java
@@ -1,30 +1,38 @@
package com.github.magnusp.libsql.itest;
-import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* Drives the JDBC driver over Hrana 3 Protobuf on a WebSocket ({@code hrana3-protobuf}
- * subprotocol). The observable JDBC behavior must match the JSON transport.
+ * subprotocol). The observable JDBC behavior must match the JSON transport, over {@code ws://} and
+ * {@code wss://} alike.
*/
public class WebSocketProtobufTransportIT extends SqldContainerBase {
- private static Connection connect() throws SQLException {
- return DriverManager.getConnection(
- "jdbc:libsql:" + getWebSocketUrl() + "?encoding=protobuf");
+ static Stream schemes() {
+ return Stream.of("ws", "wss");
}
- @Test
- void selectReturnsTypedValuesIncluding64BitIntegerAndBlob() throws SQLException {
- try (Connection connection = connect();
+ private static Connection connect(String scheme) throws SQLException {
+ String url = "wss".equals(scheme) ? getSecureWebSocketUrl() : getWebSocketUrl();
+ return DriverManager.getConnection("jdbc:libsql:" + url + "?encoding=protobuf");
+ }
+
+ @ParameterizedTest
+ @MethodSource("schemes")
+ void selectReturnsTypedValuesIncluding64BitIntegerAndBlob(String scheme) throws SQLException {
+ try (Connection connection = connect(scheme);
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(
"SELECT 9007199254740993 AS big, -9223372036854775808 AS min,"
@@ -40,28 +48,30 @@ void selectReturnsTypedValuesIncluding64BitIntegerAndBlob() throws SQLException
}
}
- @Test
- void transactionCommitsAndRollsBackOnOneStream() throws SQLException {
- try (Connection connection = connect()) {
+ @ParameterizedTest
+ @MethodSource("schemes")
+ void transactionCommitsAndRollsBackOnOneStream(String scheme) throws SQLException {
+ try (Connection connection = connect(scheme)) {
try (Statement statement = connection.createStatement()) {
- statement.execute("CREATE TABLE ws_proto_tx (id INTEGER PRIMARY KEY, name TEXT)");
+ statement.execute(
+ "CREATE TABLE ws_proto_tx_" + scheme + " (id INTEGER PRIMARY KEY, name TEXT)");
}
connection.setAutoCommit(false);
try (Statement statement = connection.createStatement()) {
- statement.execute("INSERT INTO ws_proto_tx VALUES (1, 'committed')");
+ statement.execute("INSERT INTO ws_proto_tx_" + scheme + " VALUES (1, 'committed')");
}
connection.commit();
try (Statement statement = connection.createStatement()) {
- statement.execute("INSERT INTO ws_proto_tx VALUES (2, 'rolled back')");
+ statement.execute("INSERT INTO ws_proto_tx_" + scheme + " VALUES (2, 'rolled back')");
}
connection.rollback();
connection.setAutoCommit(true);
try (Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(
- "SELECT id, name FROM ws_proto_tx ORDER BY id")) {
+ "SELECT id, name FROM ws_proto_tx_" + scheme + " ORDER BY id")) {
assertThat(resultSet.next()).isTrue();
assertThat(resultSet.getInt(1)).isEqualTo(1);
assertThat(resultSet.getString(2)).isEqualTo("committed");
@@ -70,9 +80,10 @@ void transactionCommitsAndRollsBackOnOneStream() throws SQLException {
}
}
- @Test
- void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable() throws SQLException {
- try (Connection connection = connect(); Statement statement = connection.createStatement()) {
+ @ParameterizedTest
+ @MethodSource("schemes")
+ void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable(String scheme) throws SQLException {
+ try (Connection connection = connect(scheme); Statement statement = connection.createStatement()) {
assertThatThrownBy(() -> statement.executeQuery("SELECT * FROM ws_proto_missing_table"))
.isInstanceOf(SQLException.class)
.hasMessageContaining("no such table");
diff --git a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketTransportIT.java b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketTransportIT.java
index bd60e25..d4bedb9 100644
--- a/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketTransportIT.java
+++ b/libsql-itests/src/test/java/com/github/magnusp/libsql/itest/WebSocketTransportIT.java
@@ -4,6 +4,8 @@
import com.github.magnusp.libsql.jdbc.LibsqlUrlParser;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.MethodSource;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
@@ -12,6 +14,7 @@
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;
+import java.util.stream.Stream;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -19,16 +22,26 @@
/**
* Drives the JDBC driver over Hrana 3 JSON on a WebSocket ({@code hrana3} subprotocol), which is
* the same server and the same protocol structures as the HTTP transport but a different envelope.
+ * Every case runs over plain {@code ws://} and over {@code wss://} through a TLS terminator.
*/
public class WebSocketTransportIT extends SqldContainerBase {
- private static Connection connect() throws SQLException {
- return DriverManager.getConnection("jdbc:libsql:" + getWebSocketUrl());
+ static Stream schemes() {
+ return Stream.of("ws", "wss");
}
- @Test
- void selectReturnsTypedValues() throws SQLException {
- try (Connection connection = connect();
+ private static String url(String scheme) {
+ return "wss".equals(scheme) ? getSecureWebSocketUrl() : getWebSocketUrl();
+ }
+
+ private static Connection connect(String scheme) throws SQLException {
+ return DriverManager.getConnection("jdbc:libsql:" + url(scheme));
+ }
+
+ @ParameterizedTest
+ @MethodSource("schemes")
+ void selectReturnsTypedValues(String scheme) throws SQLException {
+ try (Connection connection = connect(scheme);
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery(
"SELECT 1 AS i, 1.5 AS f, 'hello' AS t, x'010203' AS b, NULL AS n")) {
@@ -42,27 +55,30 @@ void selectReturnsTypedValues() throws SQLException {
}
}
- @Test
- void transactionCommitsAndRollsBackOnOneStream() throws SQLException {
- try (Connection connection = connect()) {
+ @ParameterizedTest
+ @MethodSource("schemes")
+ void transactionCommitsAndRollsBackOnOneStream(String scheme) throws SQLException {
+ try (Connection connection = connect(scheme)) {
try (Statement statement = connection.createStatement()) {
- statement.execute("CREATE TABLE ws_tx (id INTEGER PRIMARY KEY, name TEXT)");
+ statement.execute(
+ "CREATE TABLE ws_tx_" + scheme + " (id INTEGER PRIMARY KEY, name TEXT)");
}
connection.setAutoCommit(false);
try (Statement statement = connection.createStatement()) {
- statement.execute("INSERT INTO ws_tx VALUES (1, 'committed')");
+ statement.execute("INSERT INTO ws_tx_" + scheme + " VALUES (1, 'committed')");
}
connection.commit();
try (Statement statement = connection.createStatement()) {
- statement.execute("INSERT INTO ws_tx VALUES (2, 'rolled back')");
+ statement.execute("INSERT INTO ws_tx_" + scheme + " VALUES (2, 'rolled back')");
}
connection.rollback();
connection.setAutoCommit(true);
try (Statement statement = connection.createStatement();
- ResultSet resultSet = statement.executeQuery("SELECT id, name FROM ws_tx ORDER BY id")) {
+ ResultSet resultSet = statement.executeQuery(
+ "SELECT id, name FROM ws_tx_" + scheme + " ORDER BY id")) {
assertThat(resultSet.next()).isTrue();
assertThat(resultSet.getInt(1)).isEqualTo(1);
assertThat(resultSet.getString(2)).isEqualTo("committed");
@@ -71,9 +87,10 @@ void transactionCommitsAndRollsBackOnOneStream() throws SQLException {
}
}
- @Test
- void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable() throws SQLException {
- try (Connection connection = connect(); Statement statement = connection.createStatement()) {
+ @ParameterizedTest
+ @MethodSource("schemes")
+ void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable(String scheme) throws SQLException {
+ try (Connection connection = connect(scheme); Statement statement = connection.createStatement()) {
assertThatThrownBy(() -> statement.executeQuery("SELECT * FROM ws_missing_table"))
.isInstanceOf(SQLException.class)
.hasMessageContaining("no such table");
@@ -85,22 +102,25 @@ void failingStatementMapsToSqlExceptionAndLeavesTheConnectionUsable() throws SQL
}
}
- @Test
- void metadataAndBatchWorkOverWebSocket() throws SQLException {
- try (Connection connection = connect()) {
+ @ParameterizedTest
+ @MethodSource("schemes")
+ void metadataAndBatchWorkOverWebSocket(String scheme) throws SQLException {
+ try (Connection connection = connect(scheme)) {
DatabaseMetaData metaData = connection.getMetaData();
assertThat(metaData.getDatabaseProductName()).isEqualTo("SQLite (libSQL)");
- assertThat(metaData.getURL()).isEqualTo(getWebSocketUrl());
+ assertThat(metaData.getURL()).isEqualTo(url(scheme));
try (Statement statement = connection.createStatement()) {
- statement.execute("CREATE TABLE ws_batch (id INTEGER PRIMARY KEY, name TEXT)");
- statement.addBatch("INSERT INTO ws_batch VALUES (1, 'a')");
- statement.addBatch("INSERT INTO ws_batch VALUES (2, 'b')");
+ statement.execute(
+ "CREATE TABLE ws_batch_" + scheme + " (id INTEGER PRIMARY KEY, name TEXT)");
+ statement.addBatch("INSERT INTO ws_batch_" + scheme + " VALUES (1, 'a')");
+ statement.addBatch("INSERT INTO ws_batch_" + scheme + " VALUES (2, 'b')");
assertThat(statement.executeBatch()).containsExactly(1, 1);
}
try (Statement statement = connection.createStatement();
- ResultSet resultSet = statement.executeQuery("SELECT COUNT(*) FROM ws_batch")) {
+ ResultSet resultSet = statement.executeQuery(
+ "SELECT COUNT(*) FROM ws_batch_" + scheme)) {
assertThat(resultSet.next()).isTrue();
assertThat(resultSet.getInt(1)).isEqualTo(2);
}
diff --git a/libsql-itests/src/test/resources/tls/README.md b/libsql-itests/src/test/resources/tls/README.md
new file mode 100644
index 0000000..f3cd231
--- /dev/null
+++ b/libsql-itests/src/test/resources/tls/README.md
@@ -0,0 +1,19 @@
+# TLS fixtures for the `wss://` integration tests
+
+`server.crt.pem`, `server.key.pem`, and `truststore.p12` belong to a self-signed
+certificate for `localhost` and `127.0.0.1` that exists only so the test suite can
+put a TLS terminator in front of sqld. The key is public and worthless.
+
+Regenerate them when they expire (2036-10-21) with:
+
+```bash
+openssl req -x509 -newkey rsa:2048 -sha256 -days 3650 -nodes \
+ -keyout server.key.pem -out server.crt.pem \
+ -subj "/CN=localhost" -addext "subjectAltName=DNS:localhost,IP:127.0.0.1"
+openssl pkcs12 -export -in server.crt.pem -inkey server.key.pem \
+ -name libsql-test-tls -out truststore.p12 -passout pass:changeit
+```
+
+`libsql-itests/pom.xml` points the failsafe JVM at `truststore.p12` through
+`javax.net.ssl.trustStore`, so the driver trusts this certificate and nothing else
+during the integration tests.
diff --git a/libsql-itests/src/test/resources/tls/nginx.conf b/libsql-itests/src/test/resources/tls/nginx.conf
new file mode 100644
index 0000000..4f2b4e1
--- /dev/null
+++ b/libsql-itests/src/test/resources/tls/nginx.conf
@@ -0,0 +1,28 @@
+worker_processes 1;
+
+events {
+ worker_connections 128;
+}
+
+http {
+ map $http_upgrade $connection_upgrade {
+ default upgrade;
+ '' close;
+ }
+
+ server {
+ listen 8443 ssl;
+ ssl_certificate /etc/nginx/tls/server.crt.pem;
+ ssl_certificate_key /etc/nginx/tls/server.key.pem;
+
+ location / {
+ proxy_pass http://sqld:8080;
+ proxy_http_version 1.1;
+ proxy_set_header Upgrade $http_upgrade;
+ proxy_set_header Connection $connection_upgrade;
+ proxy_set_header Host $host;
+ proxy_read_timeout 3600s;
+ proxy_send_timeout 3600s;
+ }
+ }
+}
diff --git a/libsql-itests/src/test/resources/tls/server.crt.pem b/libsql-itests/src/test/resources/tls/server.crt.pem
new file mode 100644
index 0000000..c183ecc
--- /dev/null
+++ b/libsql-itests/src/test/resources/tls/server.crt.pem
@@ -0,0 +1,19 @@
+-----BEGIN CERTIFICATE-----
+MIIDJTCCAg2gAwIBAgIUAWaPySVWG8RjRqRpm4F4ksenO8cwDQYJKoZIhvcNAQEL
+BQAwFDESMBAGA1UEAwwJbG9jYWxob3N0MB4XDTI2MTAyNDA3MzYzOVoXDTM2MTAy
+MTA3MzYzOVowFDESMBAGA1UEAwwJbG9jYWxob3N0MIIBIjANBgkqhkiG9w0BAQEF
+AAOCAQ8AMIIBCgKCAQEA06oL0ySDw5sHes8jRkiFF17/MeLMC96wrmm2p524z8uq
+XOAAIQyVvboobvID6JvLnGUMFnrGJf9aTjsPYora4vvWNyz7ShcgsLVNrv1yMugv
++As+gr8UcKFR19Kf4AViMt8u+MsxwQl+QACy1+MqctxBzlFJfNa3hUTgp+5aowJC
+DDtplhBICbBiBNtgPxEvQGp+yZEGxLrETA6JspsBgzszyuFMtwa9hiTmHdFsCHIl
+JFo1qim2Y1kXWYQ4iQ+BBQf7cOMXazm9xz17HcXBWXRR5Azzmul1+tCMFiUZXAua
+xLn8nqzUDiScYIuViKxY+kRo7rRKEJ6/cbdKy8IAhQIDAQABo28wbTAdBgNVHQ4E
+FgQUYhIs3fdvgMr2UkvOB9jKCAgP2LEwHwYDVR0jBBgwFoAUYhIs3fdvgMr2UkvO
+B9jKCAgP2LEwDwYDVR0TAQH/BAUwAwEB/zAaBgNVHREEEzARgglsb2NhbGhvc3SH
+BH8AAAEwDQYJKoZIhvcNAQELBQADggEBAKROksqonPMSnQ1HL3jLnxZmpgjdagZ1
+E2O+kpPillH84+HU1zrV8bhdif59GRC58cxVyep6V8NUS49pScyaB1B8mv+MLDGq
+dukuX6qTNIOxXyFmvKDH6xqQEBAjrws/VeK94tjpmoHLUtC4/+LACwrxK6js9RnE
+S40m5Qc17f05mE2Rhi8NpHJc6w9g/vd54BQsvMpxqT+oTaayFjos/Vc+VqcGZAzI
+nCp8cpazDbaUQSm/RKSuwZFPEi5Yj3xte0Jv9V1MhOrrqXX42jt18oME3Ql2xPlE
+RhsfmolCB3DmLQB100X6UZLJlT1pd6q17qsUzzDNTNfHCduI59s0eXE=
+-----END CERTIFICATE-----
diff --git a/libsql-itests/src/test/resources/tls/server.key.pem b/libsql-itests/src/test/resources/tls/server.key.pem
new file mode 100644
index 0000000..1c404b6
--- /dev/null
+++ b/libsql-itests/src/test/resources/tls/server.key.pem
@@ -0,0 +1,28 @@
+-----BEGIN PRIVATE KEY-----
+MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQDTqgvTJIPDmwd6
+zyNGSIUXXv8x4swL3rCuabannbjPy6pc4AAhDJW9uihu8gPom8ucZQwWesYl/1pO
+Ow9iitri+9Y3LPtKFyCwtU2u/XIy6C/4Cz6CvxRwoVHX0p/gBWIy3y74yzHBCX5A
+ALLX4ypy3EHOUUl81reFROCn7lqjAkIMO2mWEEgJsGIE22A/ES9Aan7JkQbEusRM
+DomymwGDOzPK4Uy3Br2GJOYd0WwIciUkWjWqKbZjWRdZhDiJD4EFB/tw4xdrOb3H
+PXsdxcFZdFHkDPOa6XX60IwWJRlcC5rEufyerNQOJJxgi5WIrFj6RGjutEoQnr9x
+t0rLwgCFAgMBAAECggEAPkKnLsjvX0LDXPsAD8T2LS062KQzTAyJoQwkd8J7odfx
+OVeEgjQu+bt/+D6FKs9lAZBxkU006ullWVyjBmP8xca5OMqo8h7CdWgnPSF9DX9h
+2M8J6LCnJQg2yErCz0UWIQuoR/ORI4yUz+PEt58xFc83Jiaf/LI2EmPzx0Hj9E2F
+EjgFIKthb8KU86B2zMMjgoajT/QY7CQ8YTMkPvsiqaAAUWk/6TFUVD+4piKnQQ9W
+Lct8iZI7CuKn7Pd51g0BvTscTSojfjeMMRwZKeMx/JRoleLFT6K+2Vk4QWZOCUMD
+XFzGKxs7Sol4piM6THAFQ/08cbgNB6zsjxw4aPlRswKBgQD2Mfwh/tuMlllbymIx
+8XBp6wL0qfTjDK3UaNs/B9lOFFdXkxhqGQUl3TyghIpY6BqAz2YX5h9UIsY1o+Zx
+tR1DvuzpF/+VcKzffdDvBYaCtXPjPjLszLuib/5G7yjQQhBjceJSQ73j8soizexA
+tSSJvHerlTy7OcGw4br5eh/lhwKBgQDcGAKPZEIYzG1AdQ7Z3iHYoHq8z50wtcH3
+T80811hMO6DOE3834WeAMzpyvQrmKjwLIOUURtKKgGZv2qKG2+3wg4vh7qvsXwYv
+vho+E5TC3P44DQzkWIGub/0fV+Ml2y/89fm6QKyMGauIKmy0BseF4lDQaA+/O/nZ
+0lBsr5IskwKBgQDrgQch8exMwKm8D439HmsdNsz+A/IYm1FRRfisg7KQaQ7f2/rX
+yxVMkiRjnEXHJEft/fc4V06Sqi0WtLFFBi53VpTDqMw5RK5pn64UZyGzuSWhNCd8
+s5+tlBoV+sL87KVGq3MGpHxqKB7KSld70ougIiQLg1ez9Vzd+iBkczk90wKBgCZD
+vl4zrhIlqHDmaRiTuAqzi+M7rTaEwx4+L45+WgmW7FQnq5yJgbcWMPk1yKlPenTs
+2Sb3US9nN2gG+O69WdbQ6Ab1uco0kxzFzkYwnVJrnX8dNbqVcKZlLfq/CuzP+L3f
+/fTVvan15I6d4ZxC+XlbaXVyQhqbUMr/95+NvrolAoGAQxz92susMxl8ImTgABxJ
+lkctHT9g75RRBN8m1egwrjrSyoTHU/KGvHhjtXJKH3btFZDxICmCgykdnFN0PD8Q
+QrRvDqO6wFlIM1R83c8acr3ws4Ff/2c1eEgOBakpD/UrWyrufXRDe0p1aKPTy+zb
+ClDi/Q5BnPXM+FkDXttytK4=
+-----END PRIVATE KEY-----
diff --git a/libsql-itests/src/test/resources/tls/truststore.p12 b/libsql-itests/src/test/resources/tls/truststore.p12
new file mode 100644
index 0000000..b84eb31
Binary files /dev/null and b/libsql-itests/src/test/resources/tls/truststore.p12 differ
diff --git a/pom.xml b/pom.xml
index 00219a1..1a50741 100644
--- a/pom.xml
+++ b/pom.xml
@@ -46,6 +46,8 @@
2.0.5
7.1.0
4.36.2
+
+ 5.7-beta1
ghcr.io/tursodatabase/libsql-server:v0.24.33
@@ -76,6 +78,13 @@
${protobuf.version}
+
+