fix: terminate connection for invalid messages (#154)

* test: add tests for invalid messages * fix: terminate connection for invalid messages The backend should protect itself against a malicious client by dropping the connection if it receives an invalid message, or a message that claims to be longer than is possible. * fix: terminate connection for invalid messages The backend should protect itself against a malicious client by dropping the connection if it receives an invalid message, or a message that claims to be longer than is possible. * fix: increase max message length * fix: remove max message length check

Author: olavloite

src/main/java/com/google/cloud/spanner/pgadapter/ConnectionHandler.java

@@ -187,6 +187,9 @@ public void run() {
  try {
  message.nextHandler();
  message.send();
+ } catch (IllegalArgumentException | IllegalStateException | EOFException fatalException) {
+ this.handleError(output, fatalException);
+ this.status = ConnectionStatus.TERMINATED;
  } catch (Exception e) {
  this.handleError(output, e);
  }
@@ -226,7 +229,9 @@ public void run() {
  /** Called when a Terminate message is received. This closes this {@link ConnectionHandler}. */
  public void handleTerminate() {
  closeAllPortals();
- this.spannerConnection.close();
+ if (this.spannerConnection != null) {
+ this.spannerConnection.close();
+ }
  this.status = ConnectionStatus.TERMINATED;
  }
 
@@ -237,6 +242,8 @@ public void handleTerminate() {
  void terminate() throws IOException {
  if (this.status != ConnectionStatus.TERMINATED) {
  handleTerminate();
+ }
+ if (!socket.isClosed()) {
  socket.close();
  }
  }

src/main/java/com/google/cloud/spanner/pgadapter/ProxyServer.java

@@ -174,7 +174,16 @@ void runServer() throws IOException {
  getLocalPort(), e));
  } finally {
  for (ConnectionHandler handler : this.handlers) {
- handler.terminate();
+ try {
+ handler.terminate();
+ } catch (Exception exception) {
+ logger.log(
+ Level.WARNING,
+ exception,
+ () ->
+ String.format(
+ "Connection handler %s could not be terminated: %s", handler, exception));
+ }
  }
  logger.log(Level.INFO, () -> String.format("Socket on port %d stopped", getLocalPort()));
  notifyStopped();

src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/BootstrapMessage.java

@@ -33,6 +33,7 @@
  */
 @InternalApi
 public abstract class BootstrapMessage extends WireMessage {
+ private static final int MAX_BOOTSTRAP_MESSAGE_LENGTH = 1 << 8;
 
  public BootstrapMessage(ConnectionHandler connection, int length) {
  super(connection, length);
@@ -48,6 +49,9 @@ public BootstrapMessage(ConnectionHandler connection, int length) {
  */
  public static BootstrapMessage create(ConnectionHandler connection) throws Exception {
  int length = connection.getConnectionMetadata().getInputStream().readInt();
+ if (length > MAX_BOOTSTRAP_MESSAGE_LENGTH) {
+ throw new IllegalArgumentException("Invalid bootstrap message length: " + length);
+ }
  int protocol = connection.getConnectionMetadata().getInputStream().readInt();
  switch (protocol) {
  case SSLMessage.IDENTIFIER:

src/main/java/com/google/cloud/spanner/pgadapter/wireprotocol/WireMessage.java

@@ -16,6 +16,7 @@
 
 import com.google.api.core.InternalApi;
 import com.google.cloud.spanner.pgadapter.ConnectionHandler;
+import com.google.common.base.Preconditions;
 import java.io.DataInputStream;
 import java.io.DataOutputStream;
 import java.io.IOException;
@@ -37,6 +38,7 @@ public abstract class WireMessage {
  protected ConnectionHandler connection;
 
  public WireMessage(ConnectionHandler connection, int length) {
+ Preconditions.checkArgument(length >= 4);
  this.connection = connection;
  this.inputStream = connection.getConnectionMetadata().getInputStream();
  this.outputStream = connection.getConnectionMetadata().getOutputStream();

src/test/java/com/google/cloud/spanner/pgadapter/InvalidMessagesTest.java

@@ -0,0 +1,126 @@
+// Copyright 2022 Google LLC
+//
+// 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
+//
+// http://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 com.google.cloud.spanner.pgadapter;
+
+import static org.junit.Assert.assertEquals;
+
+import com.google.cloud.spanner.pgadapter.wireprotocol.SSLMessage;
+import com.google.cloud.spanner.pgadapter.wireprotocol.StartupMessage;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class InvalidMessagesTest extends AbstractMockServerTest {
+
+ @Test
+ public void testConnectionWithoutMessages() throws IOException {
+ try (Socket ignored = new Socket("localhost", pgServer.getLocalPort())) {
+ // Do nothing, just close the socket again.
+ }
+ }
+
+ @Test
+ public void testGarbledStartupMessage() throws IOException {
+ try (Socket socket = new Socket("localhost", pgServer.getLocalPort())) {
+ socket.getOutputStream().write("foo".getBytes(StandardCharsets.UTF_8));
+ }
+ }
+
+ @Test
+ public void testDropConnectionAfterStartup() throws IOException {
+ try (Socket socket = new Socket("localhost", pgServer.getLocalPort())) {
+ try (DataOutputStream outputStream = new DataOutputStream(socket.getOutputStream())) {
+ // Send a startup message and then quit.
+ outputStream.writeInt(8); // length == 8
+ outputStream.writeInt(StartupMessage.IDENTIFIER);
+ outputStream.flush();
+ }
+ }
+ }
+
+ @Test
+ public void testDropConnectionAfterRefusedSSL() throws IOException {
+ try (Socket socket = new Socket("localhost", pgServer.getLocalPort())) {
+ try (DataInputStream inputStream = new DataInputStream(socket.getInputStream());
+ DataOutputStream outputStream = new DataOutputStream(socket.getOutputStream())) {
+ // Request SSL.
+ outputStream.writeInt(8); // length == 8
+ outputStream.writeInt(SSLMessage.IDENTIFIER);
+ outputStream.flush();
+
+ // Verify that it is refused by the server.
+ byte response = inputStream.readByte();
+ assertEquals('N', response);
+ }
+ }
+ }
+
+ @Test
+ public void testDropConnectionAfterStartupMessage() throws IOException {
+ try (Socket socket = new Socket("localhost", pgServer.getLocalPort())) {
+ try (DataInputStream inputStream = new DataInputStream(socket.getInputStream());
+ DataOutputStream outputStream = new DataOutputStream(socket.getOutputStream())) {
+ // Request startup.
+ outputStream.writeInt(17);
+ outputStream.writeInt(StartupMessage.IDENTIFIER);
+ outputStream.writeBytes("user");
+ outputStream.writeByte(0);
+ outputStream.writeBytes("foo");
+ outputStream.writeByte(0);
+ outputStream.flush();
+
+ // Verify that the server responds with auth OK.
+ assertEquals('R', inputStream.readByte());
+ assertEquals(8, inputStream.readInt());
+ assertEquals(0, inputStream.readInt()); // 0 == success
+ }
+ }
+ }
+
+ @Test
+ public void testSendGarbageAfterStartupMessage() throws IOException {
+ try (Socket socket = new Socket("localhost", pgServer.getLocalPort())) {
+ try (DataInputStream inputStream = new DataInputStream(socket.getInputStream());
+ DataOutputStream outputStream = new DataOutputStream(socket.getOutputStream())) {
+ // Request startup.
+ outputStream.writeInt(17);
+ outputStream.writeInt(StartupMessage.IDENTIFIER);
+ outputStream.writeBytes("user");
+ outputStream.writeByte(0);
+ outputStream.writeBytes("foo");
+ outputStream.writeByte(0);
+ outputStream.flush();
+
+ // Then send a random message with no meaning and drop the connection.
+ outputStream.writeInt(20);
+ outputStream.writeChar(' ');
+ outputStream.flush();
+
+ // Read until the end of the stream. The stream should be closed by the backend.
+ int bytesRead = 0;
+ while (inputStream.read() > -1 && bytesRead < 1 << 16) {
+ bytesRead++;
+ }
+ assertEquals(-1, inputStream.read());
+ }
+ }
+ }
+}