Add SSL support..

SSL is disabled by default to avoid POLA violations. It is possible to enable and control SSL behavior via url parameters: - `sslmode=<mode>` enable ssl (prefer/require/verify-ca/verify-full [recommended]) - `sslrootcert=<path.pem>` specifies trusted certificates (JDK cacert if missing) Client certificate authentication is not implemented, due to lack of time and interest, but it should be easy to add.

Author: alexdupre

db-async-common/src/main/scala/com/github/mauricio/async/db/Configuration.scala

@@ -37,6 +37,7 @@ object Configuration {
  * @param port database port, defaults to 5432
  * @param password password, defaults to no password
  * @param database database name, defaults to no database
+ * @param ssl ssl configuration
  * @param charset charset for the connection, defaults to UTF-8, make sure you know what you are doing if you
  * change this
  * @param maximumMessageSize the maximum size a message from the server could possibly have, this limits possible
@@ -55,6 +56,7 @@ case class Configuration(username: String,
  port: Int = 5432,
  password: Option[String] = None,
  database: Option[String] = None,
+ ssl: SSLConfiguration = SSLConfiguration(),
  charset: Charset = Configuration.DefaultCharset,
  maximumMessageSize: Int = 16777216,
  allocator: ByteBufAllocator = PooledByteBufAllocator.DEFAULT,

db-async-common/src/main/scala/com/github/mauricio/async/db/SSLConfiguration.scala

@@ -0,0 +1,31 @@
+package com.github.mauricio.async.db
+
+import java.io.File
+
+import SSLConfiguration.Mode
+
+/**
+ *
+ * Contains the SSL configuration necessary to connect to a database.
+ *
+ * @param mode whether and with what priority a SSL connection will be negotiated, default disabled
+ * @param rootCert path to PEM encoded trusted root certificates, None to use internal JDK cacerts, defaults to None
+ *
+ */
+case class SSLConfiguration(mode: Mode.Value = Mode.Disable, rootCert: Option[java.io.File] = None)
+
+object SSLConfiguration {
+
+ object Mode extends Enumeration {
+ val Disable = Value("disable") // only try a non-SSL connection
+ val Prefer = Value("prefer") // first try an SSL connection; if that fails, try a non-SSL connection
+ val Require = Value("require") // only try an SSL connection, but don't verify Certificate Authority
+ val VerifyCA = Value("verify-ca") // only try an SSL connection, and verify that the server certificate is issued by a trusted certificate authority (CA)
+ val VerifyFull = Value("verify-full") // only try an SSL connection, verify that the server certificate is issued by a trusted CA and that the server host name matches that in the certificate
+ }
+
+ def apply(properties: Map[String, String]): SSLConfiguration = SSLConfiguration(
+ mode = Mode.withName(properties.get("sslmode").getOrElse("disable")),
+ rootCert = properties.get("sslrootcert").map(new File(_))
+ )
+}

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/codec/MessageDecoder.scala

@@ -17,7 +17,7 @@
 package com.github.mauricio.async.db.postgresql.codec
 
 import com.github.mauricio.async.db.postgresql.exceptions.{MessageTooLongException}
-import com.github.mauricio.async.db.postgresql.messages.backend.ServerMessage
+import com.github.mauricio.async.db.postgresql.messages.backend.{ServerMessage, SSLResponseMessage}
 import com.github.mauricio.async.db.postgresql.parsers.{AuthenticationStartupParser, MessageParsersRegistry}
 import com.github.mauricio.async.db.util.{BufferDumper, Log}
 import java.nio.charset.Charset
@@ -31,15 +31,21 @@ object MessageDecoder {
  val DefaultMaximumSize = 16777216
 }
 
-class MessageDecoder(charset: Charset, maximumMessageSize : Int = MessageDecoder.DefaultMaximumSize) extends ByteToMessageDecoder {
+class MessageDecoder(sslEnabled: Boolean, charset: Charset, maximumMessageSize : Int = MessageDecoder.DefaultMaximumSize) extends ByteToMessageDecoder {
 
  import MessageDecoder.log
 
  private val parser = new MessageParsersRegistry(charset)
 
+ private var sslChecked = false
+
  override def decode(ctx: ChannelHandlerContext, b: ByteBuf, out: java.util.List[Object]): Unit = {
 
- if (b.readableBytes() >= 5) {
+ if (sslEnabled & !sslChecked) {
+ val code = b.readByte()
+ sslChecked = true
+ out.add(new SSLResponseMessage(code == 'S'))
+ } else if (b.readableBytes() >= 5) {
 
  b.markReaderIndex()
 

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/codec/MessageEncoder.scala

@@ -44,12 +44,13 @@ class MessageEncoder(charset: Charset, encoderRegistry: ColumnEncoderRegistry) e
  override def encode(ctx: ChannelHandlerContext, msg: AnyRef, out: java.util.List[Object]) = {
 
  val buffer = msg match {
+ case SSLRequestMessage => SSLMessageEncoder.encode()
+ case message: StartupMessage => startupEncoder.encode(message)
  case message: ClientMessage => {
  val encoder = (message.kind: @switch) match {
  case ServerMessage.Close => CloseMessageEncoder
  case ServerMessage.Execute => this.executeEncoder
  case ServerMessage.Parse => this.openEncoder
- case ServerMessage.Startup => this.startupEncoder
  case ServerMessage.Query => this.queryEncoder
  case ServerMessage.PasswordMessage => this.credentialEncoder
  case _ => throw new EncoderNotAvailableException(message)

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/codec/PostgreSQLConnectionHandler.scala

@@ -17,6 +17,7 @@
 package com.github.mauricio.async.db.postgresql.codec
 
 import com.github.mauricio.async.db.Configuration
+import com.github.mauricio.async.db.SSLConfiguration.Mode
 import com.github.mauricio.async.db.column.{ColumnDecoderRegistry, ColumnEncoderRegistry}
 import com.github.mauricio.async.db.postgresql.exceptions._
 import com.github.mauricio.async.db.postgresql.messages.backend._
@@ -38,6 +39,12 @@ import com.github.mauricio.async.db.postgresql.messages.backend.RowDescriptionMe
 import com.github.mauricio.async.db.postgresql.messages.backend.ParameterStatusMessage
 import io.netty.channel.socket.nio.NioSocketChannel
 import io.netty.handler.codec.CodecException
+import io.netty.handler.ssl.{SslContextBuilder, SslHandler}
+import io.netty.handler.ssl.util.InsecureTrustManagerFactory
+import io.netty.util.concurrent.FutureListener
+import javax.net.ssl.{SSLParameters, TrustManagerFactory}
+import java.security.KeyStore
+import java.io.FileInputStream
 
 object PostgreSQLConnectionHandler {
  final val log = Log.get[PostgreSQLConnectionHandler]
@@ -79,7 +86,7 @@ class PostgreSQLConnectionHandler
 
  override def initChannel(ch: channel.Channel): Unit = {
  ch.pipeline.addLast(
- new MessageDecoder(configuration.charset, configuration.maximumMessageSize),
+ new MessageDecoder(configuration.ssl.mode != Mode.Disable, configuration.charset, configuration.maximumMessageSize),
  new MessageEncoder(configuration.charset, encoderRegistry),
  PostgreSQLConnectionHandler.this)
  }
@@ -120,13 +127,61 @@ class PostgreSQLConnectionHandler
  }
 
  override def channelActive(ctx: ChannelHandlerContext): Unit = {
- ctx.writeAndFlush(new StartupMessage(this.properties))
+ if (configuration.ssl.mode == Mode.Disable)
+ ctx.writeAndFlush(new StartupMessage(this.properties))
+ else
+ ctx.writeAndFlush(SSLRequestMessage)
  }
 
  override def channelRead0(ctx: ChannelHandlerContext, msg: Object): Unit = {
 
  msg match {
 
+ case SSLResponseMessage(supported) =>
+ if (supported) {
+ val ctxBuilder = SslContextBuilder.forClient()
+ if (configuration.ssl.mode >= Mode.VerifyCA) {
+ configuration.ssl.rootCert.fold {
+ val tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm())
+ val ks = KeyStore.getInstance(KeyStore.getDefaultType())
+ val cacerts = new FileInputStream(System.getProperty("java.home") + "/lib/security/cacerts")
+ try {
+ ks.load(cacerts, "changeit".toCharArray)
+ } finally {
+ cacerts.close()
+ }
+ tmf.init(ks)
+ ctxBuilder.trustManager(tmf)
+ } { path =>
+ ctxBuilder.trustManager(path)
+ }
+ } else {
+ ctxBuilder.trustManager(InsecureTrustManagerFactory.INSTANCE)
+ }
+ val sslContext = ctxBuilder.build()
+ val sslEngine = sslContext.newEngine(ctx.alloc(), configuration.host, configuration.port)
+ if (configuration.ssl.mode >= Mode.VerifyFull) {
+ val sslParams = sslEngine.getSSLParameters()
+ sslParams.setEndpointIdentificationAlgorithm("HTTPS")
+ sslEngine.setSSLParameters(sslParams)
+ }
+ val handler = new SslHandler(sslEngine)
+ ctx.pipeline().addFirst(handler)
+ handler.handshakeFuture.addListener(new FutureListener[channel.Channel]() {
+ def operationComplete(future: io.netty.util.concurrent.Future[channel.Channel]) {
+ if (future.isSuccess()) {
+ ctx.writeAndFlush(new StartupMessage(properties))
+ } else {
+ connectionDelegate.onError(future.cause())
+ }
+ }
+ })
+ } else if (configuration.ssl.mode < Mode.Require) {
+ ctx.writeAndFlush(new StartupMessage(properties))
+ } else {
+ connectionDelegate.onError(new IllegalArgumentException("SSL is not supported on server"))
+ }
+
  case m: ServerMessage => {
 
  (m.kind : @switch) match {

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/encoders/SSLMessageEncoder.scala

@@ -0,0 +1,16 @@
+package com.github.mauricio.async.db.postgresql.encoders
+
+import io.netty.buffer.ByteBuf
+import io.netty.buffer.Unpooled
+
+object SSLMessageEncoder {
+
+ def encode(): ByteBuf = {
+ val buffer = Unpooled.buffer()
+ buffer.writeInt(8)
+ buffer.writeShort(1234)
+ buffer.writeShort(5679)
+ buffer
+ }
+
+}

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/encoders/StartupMessageEncoder.scala

@@ -21,13 +21,11 @@ import com.github.mauricio.async.db.util.ByteBufferUtils
 import java.nio.charset.Charset
 import io.netty.buffer.{Unpooled, ByteBuf}
 
-class StartupMessageEncoder(charset: Charset) extends Encoder {
+class StartupMessageEncoder(charset: Charset) {
 
  //private val log = Log.getByName("StartupMessageEncoder")
 
- override def encode(message: ClientMessage): ByteBuf = {
-
- val startup = message.asInstanceOf[StartupMessage]
+ def encode(startup: StartupMessage): ByteBuf = {
 
  val buffer = Unpooled.buffer()
  buffer.writeInt(0)

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/messages/backend/SSLResponseMessage.scala

@@ -0,0 +1,3 @@
+package com.github.mauricio.async.db.postgresql.messages.backend
+
+case class SSLResponseMessage(supported: Boolean)

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/messages/backend/ServerMessage.scala

@@ -43,7 +43,6 @@ object ServerMessage {
  final val Query = 'Q'
  final val RowDescription = 'T'
  final val ReadyForQuery = 'Z'
- final val Startup = '0'
  final val Sync = 'S'
 }
 

postgresql-async/src/main/scala/com/github/mauricio/async/db/postgresql/messages/frontend/InitialClientMessage.scala

@@ -0,0 +1,3 @@
+package com.github.mauricio.async.db.postgresql.messages.frontend
+
+trait InitialClientMessage