Create KafkaCredentialStore

docs/src/main/mdoc/certificates.md

@@ -0,0 +1,34 @@
+---
+id: security
+title: Security & Certificates
+---
+
+## Security: certificates, trust stores, and passwords 
+
+The `KafkaCredentialStore` can be used to create the necessary.
+
+The parameters passed in are string representations of the client private key, client certificate
+and service certificate. the `properties` field in `KafkaCredentialStore` can then be applied to
+any of the `*Settings` classes by using the `withProperties(kafkaCredentialStore.properties)`.
+
+```scala mdoc
+import cats.effect._
+import cats.syntax.all._
+import fs2.kafka.security._
+
+def loadKafkaSetup[F[_]: Async: ContextShift](
+    clientPrivateKey: String,
+    clientCertificate: String,
+    serviceCertificate: String,
+): Resource[F, KafkaCredentialStore] =
+  Blocker[F].evalMap { blocker =>
+    (
+      ClientPrivateKey.fromString(clientPrivateKey).liftTo[F],
+      ClientCertificate.fromString(clientCertificate).liftTo[F],
+      ServiceCertificate.fromString(serviceCertificate).liftTo[F],
+    ).tupled.flatMap {
+      case (clientPrivateKey, clientCertificate, serviceCertificate) =>
+        KafkaCredentialStore[F](clientPrivateKey, clientCertificate, serviceCertificate, blocker)
+    }
+  }
+```

modules/core/src/main/scala/fs2/kafka/security/ClientCertificate.scala

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018-2021 OVO Energy Limited
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package fs2.kafka.security
+
+import java.security.cert.{Certificate, CertificateException}
+
+sealed abstract class ClientCertificate {
+  def value: Certificate
+}
+
+object ClientCertificate {
+  def fromString(clientCertificate: String): Either[CertificateException, ClientCertificate] =
+    internal.CertificateOps.loadFromString(clientCertificate).map { certificate =>
+      new ClientCertificate {
+        override final val value: Certificate =
+          certificate
+
+        override final def toString: String =
+          s"ClientCertificate(${clientCertificate.valueShortHash})"
+      }
+    }
+
+  def fromCertificate(certificate: Certificate): ClientCertificate =
+    new ClientCertificate {
+      override def value: Certificate = certificate
+    }
+}

modules/core/src/main/scala/fs2/kafka/security/ClientPrivateKey.scala

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2018-2021 OVO Energy Limited
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package fs2.kafka.security
+
+import cats.syntax.all._
+
+import java.nio.charset.StandardCharsets
+import java.security.{GeneralSecurityException, KeyFactory, PrivateKey}
+import java.security.spec.PKCS8EncodedKeySpec
+import java.util.Base64
+
+sealed abstract class ClientPrivateKey {
+  def value: PrivateKey
+}
+
+object ClientPrivateKey {
+  def fromString(clientPrivateKey: String): Either[GeneralSecurityException, ClientPrivateKey] =
+    Either.catchOnly[GeneralSecurityException] {
+      new ClientPrivateKey {
+        override final val value: PrivateKey =
+          KeyFactory
+            .getInstance("RSA")
+            .generatePrivate {
+              new PKCS8EncodedKeySpec(
+                Base64.getDecoder.decode {
+                  clientPrivateKey
+                    .replace("-----BEGIN PRIVATE KEY-----", "")
+                    .replace("-----END PRIVATE KEY-----", "")
+                    .filterNot(_.isWhitespace)
+                    .getBytes(StandardCharsets.UTF_8)
+                }
+              )
+            }
+
+        override final def toString: String =
+          s"ClientPrivateKey(${clientPrivateKey.valueShortHash})"
+      }
+    }
+
+  def fromPrivateKey(privateKey: PrivateKey): ClientPrivateKey =
+    new ClientPrivateKey {
+      override def value: PrivateKey = privateKey
+
+      override final def toString: String =
+        s"ClientPrivateKey(${privateKey.toString.valueShortHash})"
+    }
+}

modules/core/src/main/scala/fs2/kafka/security/KafkaCredentialStore.scala

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2018-2021 OVO Energy Limited
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package fs2.kafka.security
+
+import cats.effect._
+import cats.syntax.all._
+
+import java.nio.file.{Files, Path}
+import java.security.KeyStore
+
+sealed abstract class KafkaCredentialStore {
+  def keyStoreFile: KeyStoreFile
+
+  def keyStorePassword: KeyStorePassword
+
+  def trustStoreFile: TrustStoreFile
+
+  def trustStorePassword: TrustStorePassword
+
+  def properties: Map[String, String]
+}
+
+object KafkaCredentialStore {
+  final def apply[F[_]: ContextShift](
+    clientPrivateKey: ClientPrivateKey,
+    clientCertificate: ClientCertificate,
+    serviceCertificate: ServiceCertificate,
+    blocker: Blocker
+  )(implicit F: Sync[F]): F[KafkaCredentialStore] =
+    for {
+      setupDetails <- KafkaCredentialStore.createTemporary[F]
+      _ <- setupKeyStore(
+        clientPrivateKey = clientPrivateKey,
+        clientCertificate = clientCertificate,
+        keyStoreFile = setupDetails.keyStoreFile,
+        keyStorePassword = setupDetails.keyStorePassword,
+        blocker = blocker
+      )
+      _ <- setupTrustStore(
+        serviceCertificate = serviceCertificate,
+        trustStoreFile = setupDetails.trustStoreFile,
+        trustStorePassword = setupDetails.trustStorePassword,
+        blocker = blocker
+      )
+    } yield setupDetails
+
+  private final def setupStore[F[_]: ContextShift](
+    storeType: String,
+    storePath: Path,
+    storePasswordChars: Array[Char],
+    setupStore: KeyStore => Unit,
+    blocker: Blocker
+  )(implicit F: Sync[F]): F[Unit] =
+    blocker.delay {
+      val keyStore = KeyStore.getInstance(storeType)
+      keyStore.load(null, storePasswordChars)
+      setupStore(keyStore)
+
+      val outputStream = Files.newOutputStream(storePath)
+
+      try {
+        keyStore.store(outputStream, storePasswordChars)
+      } finally {
+        outputStream.close()
+      }
+    }
+
+  private final def setupKeyStore[F[_]: ContextShift](
+    clientPrivateKey: ClientPrivateKey,
+    clientCertificate: ClientCertificate,
+    keyStoreFile: KeyStoreFile,
+    keyStorePassword: KeyStorePassword,
+    blocker: Blocker
+  )(implicit F: Sync[F]): F[Unit] = {
+    val keyStorePasswordChars =
+      keyStorePassword.value.toCharArray
+
+    setupStore(
+      storeType = "PKCS12",
+      storePath = keyStoreFile.path,
+      storePasswordChars = keyStorePasswordChars,
+      setupStore = _.setEntry(
+        "service_key",
+        new KeyStore.PrivateKeyEntry(
+          clientPrivateKey.value,
+          Array(clientCertificate.value)
+        ),
+        new KeyStore.PasswordProtection(keyStorePasswordChars)
+      ),
+      blocker = blocker
+    )
+  }
+
+  private final def setupTrustStore[F[_]: ContextShift](
+    serviceCertificate: ServiceCertificate,
+    trustStoreFile: TrustStoreFile,
+    trustStorePassword: TrustStorePassword,
+    blocker: Blocker
+  )(implicit F: Sync[F]): F[Unit] =
+    setupStore(
+      storeType = "JKS",
+      storePath = trustStoreFile.path,
+      storePasswordChars = trustStorePassword.value.toCharArray,
+      setupStore = _.setCertificateEntry("CA", serviceCertificate.value),
+      blocker = blocker
+    )
+
+  private final def createTemporary[F[_]](implicit F: Sync[F]): F[KafkaCredentialStore] =
+    for {
+      _keyStoreFile <- KeyStoreFile.createTemporary[F]
+      _keyStorePassword <- KeyStorePassword.createTemporary[F]
+      _trustStoreFile <- TrustStoreFile.createTemporary[F]
+      _trustStorePassword <- TrustStorePassword.createTemporary[F]
+    } yield {
+      new KafkaCredentialStore {
+        override final val keyStoreFile: KeyStoreFile =
+          _keyStoreFile
+
+        override final val keyStorePassword: KeyStorePassword =
+          _keyStorePassword
+
+        override final val trustStoreFile: TrustStoreFile =
+          _trustStoreFile
+
+        override final val trustStorePassword: TrustStorePassword =
+          _trustStorePassword
+
+        override final val properties: Map[String, String] =
+          Map(
+            "security.protocol" -> "SSL",
+            "ssl.truststore.location" -> trustStoreFile.pathAsString,
+            "ssl.truststore.password" -> trustStorePassword.value,
+            "ssl.keystore.type" -> "PKCS12",
+            "ssl.keystore.location" -> keyStoreFile.pathAsString,
+            "ssl.keystore.password" -> keyStorePassword.value,
+            "ssl.key.password" -> keyStorePassword.value
+          )
+
+        override final def toString: String =
+          s"KafkaCredentialStore($keyStoreFile, $keyStorePassword, $trustStoreFile, $trustStorePassword)"
+      }
+    }
+}

modules/core/src/main/scala/fs2/kafka/security/KeyStoreFile.scala

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2018-2021 OVO Energy Limited
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package fs2.kafka.security
+
+import cats.effect.Sync
+import cats.syntax.all._
+
+import java.nio.file.Path
+
+sealed abstract class KeyStoreFile {
+  def path: Path
+
+  def pathAsString: String
+}
+
+private[security] object KeyStoreFile {
+  final def createTemporary[F[_]](implicit F: Sync[F]): F[KeyStoreFile] =
+    internal.FileOps.createTemp("client.keystore-", ".p12").map { _path =>
+      new KeyStoreFile {
+        override final val path: Path =
+          _path
+
+        override final def pathAsString: String =
+          path.toString
+
+        override final def toString: String =
+          s"KeyStoreFile($pathAsString)"
+      }
+    }
+}

modules/core/src/main/scala/fs2/kafka/security/KeyStorePassword.scala

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2018-2021 OVO Energy Limited
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package fs2.kafka.security
+
+import cats.effect.Sync
+
+import java.util.UUID
+
+sealed abstract class KeyStorePassword {
+  def value: String
+}
+
+private[security] object KeyStorePassword {
+  def createTemporary[F[_]](implicit F: Sync[F]): F[KeyStorePassword] =
+    F.delay {
+      val _value = UUID.randomUUID().toString
+
+      new KeyStorePassword {
+        override final val value: String =
+          _value
+
+        override final def toString: String =
+          s"KeyStorePassword(${value.valueShortHash})"
+      }
+    }
+}

modules/core/src/main/scala/fs2/kafka/security/ServiceCertificate.scala

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2018-2021 OVO Energy Limited
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package fs2.kafka.security
+
+import java.security.cert.{Certificate, CertificateException}
+
+sealed abstract class ServiceCertificate {
+  def value: Certificate
+}
+
+object ServiceCertificate {
+  def fromString(serviceCertificate: String): Either[CertificateException, ServiceCertificate] =
+    internal.CertificateOps.loadFromString(serviceCertificate).map { certificate =>
+      new ServiceCertificate {
+        override final val value: Certificate =
+          certificate
+
+        override final def toString: String =
+          s"ServiceCertificate(${serviceCertificate.valueShortHash})"
+      }
+    }
+
+  def fromCertificate(certificate: Certificate): ServiceCertificate =
+    new ServiceCertificate {
+      override def value: Certificate = certificate
+    }
+}