connect the schema-registry with a Go client using TLS

[djedjethai]

Hi,

I have successfully deployed a Kafka container configured to use TLS, using Docker Compose. I use the confluentinc/confluent-kafka-go client to produce and consume messages with the Kafka broker, it works as expected.

p, err := kafka.NewProducer(&kafka.ConfigMap{
		"bootstrap.servers":        kafkaURL,
		"security.protocol":        "ssl",
		"ssl.key.location":         pathKeyProducer,
		"ssl.key.password":         "xxx",
		"ssl.certificate.location": pathCertProducer,
		"ssl.ca.location":                     "./secrets/broker.cer.pem",
		"enable.ssl.certificate.verification": false, // use self-signed certs
		"debug":                               "security,broker",
	})

If I add a Schema Registry container, the deployment of the containers remains stable. It appears that the communication between the Kafka broker and the Schema Registry through TLS is working as expected.

    # image: confluentinc/cp-server:7.0.1
    image: confluentinc/cp-kafka:7.1.10
    container_name: broker
    hostname: broker
    ports:
      - "9093:9093"
      - "29093:29093"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
        # that for the kfk controller to connect as there is only one brocker 
        # it become the controller by default, so should connect to himeself, so localhost...
        # KAFKA_ADVERTISED_LISTENERS: SSL://broker:9093 # then app on localhost can not connect
        # works for app running on localhost but then SR can not connect.....
        # KAFKA_ADVERTISED_LISTENERS: SSL://localhost:9093
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: |
        SSL:SSL,SSL_PLAINTEXT:SSL
      KAFKA_ADVERTISED_LISTENERS: |
        SSL://broker:9093,SSL_PLAINTEXT://localhost:29093
      KAFKA_SSL_KEYSTORE_FILENAME: broker.keystore.jks
      KAFKA_SSL_KEYSTORE_CREDENTIALS: cert_creds
      KAFKA_SSL_KEY_CREDENTIALS: cert_creds
      KAFKA_SSL_TRUSTSTORE_FILENAME: broker.truststore.jks
      KAFKA_SSL_TRUSTSTORE_CREDENTIALS: cert_creds
      KAFKA_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM: " "
      KAFKA_SSL_CLIENT_AUTH: requested
      KAFKA_SECURITY_PROTOCOL: SSL
      KAFKA_SECURITY_INTER_BROKER_PROTOCOL: SSL
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_OPTS: "-Djavax.net.debug=ssl"

  schemaregistry:
    image: confluentinc/cp-schema-registry:7.1.0 
    container_name: schema-registry
    hostname: schema-registry
    ports:
      - "8081:8081"
      - "8181:8181"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: 'localhost'
        # SCHEMA_REGISTRY_LISTENERS: 'http://localhost:8081'
      SCHEMA_REGISTRY_LISTENERS: 'https://localhost:8181'
      SCHEMA_REGISTRY_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM: " "
        # SCHEMA_REGISTRY_LISTENERS: 'http://localhost:8081,https://localhost:8181'
      SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: 'zookeeper:2181'
        # works if the broker advertize on port SSL://broker:9093
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'SSL://broker:9093' 
      SCHEMA_REGISTRY_KAFKASTORE_SECURITY_PROTOCOL: SSL
      SCHEMA_REGISTRY_KAFKASTORE_SSL_KEYSTORE_LOCATION: /etc/schema-registry/secrets/schema-registry.keystore.jks
      SCHEMA_REGISTRY_SSL_KEYSTORE_LOCATION: /etc/schema-registry/secrets/schema-registry.keystore.jks
      SCHEMA_REGISTRY_KAFKASTORE_SSL_KEYSTORE_PASSWORD: xxx
      SCHEMA_REGISTRY_SSL_KEYSTORE_PASSWORD: xxx
      SCHEMA_REGISTRY_KAFKASTORE_SSL_KEY_PASSWORD: xxx
      SCHEMA_REGISTRY_SSL_KEY_PASSWORD: xxx
      SCHEMA_REGISTRY_KAFKASTORE_SSL_TRUSTSTORE_LOCATION: /etc/schema-registry/secrets/schema-registry.truststore.jks
      SCHEMA_REGISTRY_SSL_TRUSTSTORE_LOCATION: /etc/schema-registry/secrets/schema-registry.truststore.jks
      SCHEMA_REGISTRY_KAFKASTORE_SSL_TRUSTSTORE_PASSWORD: xxx
      SCHEMA_REGISTRY_SSL_TRUSTSTORE_PASSWORD: xxx
      SCHEMA_REGISTRY_SCHEMA_REGISTRY_INTER_INSTANCE_PROTOCOL: https
        # SCHEMA_REGISTRY_SCHEMA_REGISTRY_INTER_INSTANCE_PROTOCOL: http
      SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
      SCHEMA_REGISTRY_SSL_CLIENT_AUTH: 'true'

I use the Schema Registry client confluentinc/confluent-kafka-go/schemaregistry, which does not implement TLS. So I am working on it. In my certificate setup (I use self-signed certificates), since there is no difference between the broker.cer.pem and the schemaregistry.cer.pem (and it works when producing to the broker), I am using the producer's certificates for the Schema Registry client to authenticate with the Schema Registry. I load the certificates like so.

func configureTLS(conf *Config, tlsConf *tls.Config) error {

	certFile := conf.SslCertificateLocation
	keyFile := conf.SslKeyLocation
	caFile := conf.SslCaLocation
	unsafe := conf.SslDisableEndpointVerification

	var err error
	if certFile != "" {
		if keyFile == "" {
			return errors.New(
				"SslKeyLocation needs to be provided if using SslCertificateLocation")
		}

		// Read the certificate file content
		certPEM, err := ioutil.ReadFile(certFile)
		if err != nil {
			return err
		}

		// Provide a callback function to decrypt the private key
		keyPEM, err := ioutil.ReadFile(keyFile)
		if err != nil {
			return err
		}

		var cert tls.Certificate
		cert, err = tls.X509KeyPair(certPEM, keyPEM)
		if err != nil {
			fmt.Println("err loading keyPair: ", err)
			return err
		}

		tlsConf.Certificates = []tls.Certificate{cert}
	}

	if caFile != "" {
		if unsafe {
			log.Println("WARN: endpoint verification is currently disabled. " +
				"This feature should be configured for development purposes only")
		}
		var caCert []byte
		caCert, err := ioutil.ReadFile(caFile)
		if err != nil {
			return err
		}

		tlsConf.RootCAs = x509.NewCertPool()
		if !tlsConf.RootCAs.AppendCertsFromPEM(caCert) {
			return fmt.Errorf("could not parse certificate from %s", caFile)
		}
	}

	tlsConf.BuildNameToCertificate()

	return err
}

But the connection/handcheck with the schema-registry fail, got this error:

2024/01/27 16:03:47 Error producing Message:  Post "https://localhost:8181/subjects/my-topic-value/versions?normalize=false": EOF
2024/01/27 16:03:47 Message produced, offset is:  -1

Any hint ??
thanks