Create KafkaCredentialStore

[agustafson]

Fix for #521

Moving the code in https://github.com/ovotech/ciris-aiven-kafka for credential loading.

Example code using ciris would now be:

def loadKafkaSetup[F[_]: Async: ContextShift]: Resource[F, KafkaCredentialStore] = {
  implicit val stringClientPrivateKeyConfigDecoder: ConfigDecoder[String, ClientPrivateKey] =
    ConfigDecoder.identity[String].mapOption("ClientPrivateKey") {
      ClientPrivateKey(_).toOption
    }
  implicit val stringClientCertificateConfigDecoder: ConfigDecoder[String, ClientCertificate] =
    ConfigDecoder.identity[String].mapOption("ClientCertificate") {
      ClientCertificate(_).toOption
    }
  implicit val stringServiceCertificateConfigDecoder: ConfigDecoder[String, ServiceCertificate] =
    ConfigDecoder.identity[String].mapOption("ServiceCertificate") {
      ServiceCertificate(_).toOption
    }

  Blocker[F].evalMap { blocker =>
    (
      env("KAFKA_CLIENT_PRIVATE_KEY").as[ClientPrivateKey],
      env("KAFKA_CLIENT_CERT").as[ClientCertificate],
      env("KAFKA_SERVER_CERT").as[ServiceCertificate]
    ).parTupled.load[F].flatMap {
      case (clientPrivateKey, clientCertificate, serviceCertificate) =>
        KafkaCredentialStore[F](clientPrivateKey, clientCertificate, serviceCertificate, blocker)
    }
  }
}

[LMnet]

Maybe we could consider moving it to a separate module?

[agustafson]

Thanks @LMnet, I did wonder about that. It just seemed like it wasn't that separate from the core. Happy to move it out if that's what people want 👍

[agustafson]

Actually, this code is blatantly copied from Secret in ciris. Perhaps it should be its own library, because I'm sure I've seen code very similar to this in a few different places.

[bplommer]

I initially agreed with @LMnet, but I think it's better to keep it in the same module so we can have the withCredentials helpers on the settings classes.

[bplommer]

Thanks for this! I'll take a look soon

[bplommer]

I assume you're happy for this code to be copied over @vlovgr?

[vlovgr]

I assume you're happy for this code to be copied over @vlovgr?

Absolutely. 👍

[bplommer]

Kafka 2.7.0 adds support for textual SSL credentials, so we may not need this after all: apache/kafka#9345

[agustafson]

Kafka 2.7.0 adds support for textual SSL credentials, so we may not need this after all: apache/kafka#9345

Happy to adapt this PR so that it can handle textual SSL creds or the code in this PR, if that would help

[bplommer]

Happy to adapt this PR so that it can handle textual SSL creds or the code in this PR, if that would help

Thanks, that would be great! I guess you saw the snippet I posted on Slack, but if not here it is:

consumerSettings
  .withProperties(Map(
  "security.protocol" -> "SSL",
  "ssl.truststore.type" -> "PEM",
  "ssl.truststore.certificates" ->
    """
      -----BEGIN CERTIFICATE-----
      // CA certificate
      -----END CERTIFICATE-----
      |""".replace('\n', ' '),
  "ssl.keystore.type" -> "PEM",
  "ssl.keystore.key" -> 
    """
      -----BEGIN PRIVATE KEY-----
      // access key
      -----END PRIVATE KEY-----""".replace('\n', ' ')
    ,
  "ssl.keystore.certificate.chain" -> 
    """
      -----BEGIN CERTIFICATE-----
      // access certificate
      -----END CERTIFICATE-----
    """.replace('\n', ' ')
))

[agustafson]

Happy to adapt this PR so that it can handle textual SSL creds or the code in this PR, if that would help

Thanks, that would be great! I guess you saw the snippet I posted on Slack, but if not here it is:

consumerSettings
  .withProperties(Map(
  "security.protocol" -> "SSL",
  "ssl.truststore.type" -> "PEM",
  "ssl.truststore.certificates" ->
    """
      -----BEGIN CERTIFICATE-----
      // CA certificate
      -----END CERTIFICATE-----
      |""".replace('\n', ' '),
  "ssl.keystore.type" -> "PEM",
  "ssl.keystore.key" -> 
    """
      -----BEGIN PRIVATE KEY-----
      // access key
      -----END PRIVATE KEY-----""".replace('\n', ' ')
    ,
  "ssl.keystore.certificate.chain" -> 
    """
      -----BEGIN CERTIFICATE-----
      // access certificate
      -----END CERTIFICATE-----
    """.replace('\n', ' ')
))

Cool, I'll add a few String-based methods and see how it looks

[bplommer]

I haven’t forgotten about this! I’ll get to it this weekend.

[bplommer]

I don't think Pkcs12KafkaCredentialStore is needed now that we have PemKafkaCredentialStore - I'm going to edit the PR to remove it assuming that's ok with you.