feat: Support HTTP basic authentication for schema registry connections

cmd/kaf/consume.go

@@ -162,6 +162,8 @@ func withConsumerGroup(ctx context.Context, client sarama.Client, topic, group s
 		errorExit("Failed to create consumer group: %v", err)
 	}
 
+	schemaCache = getSchemaCache()
+
 	err = cg.Consume(ctx, []string{topic}, &g{})
 	if err != nil {
 		errorExit("Error on consume: %v", err)

cmd/kaf/kaf.go

@@ -207,6 +207,7 @@ func onInit() {
 	// Any set flags override the configuration
 	if schemaRegistryURL != "" {
 		currentCluster.SchemaRegistryURL = schemaRegistryURL
+		currentCluster.SchemaRegistryCredentials = nil
 	}
 
 	if brokersFlag != nil {
@@ -247,7 +248,12 @@ func getSchemaCache() (cache *avro.SchemaCache) {
 	if currentCluster.SchemaRegistryURL == "" {
 		return nil
 	}
-	cache, err := avro.NewSchemaCache(currentCluster.SchemaRegistryURL)
+	var username, password string
+	if creds := currentCluster.SchemaRegistryCredentials; creds != nil {
+		username = creds.Username
+		password = creds.Password
+	}
+	cache, err := avro.NewSchemaCache(currentCluster.SchemaRegistryURL, username, password)
 	if err != nil {
 		errorExit("Unable to get schema cache :%v\n", err)
 	}

examples/schema_registry_basic_auth.yaml

@@ -0,0 +1,12 @@
+clusters:
+- name: local
+  brokers:
+  - localhost:9092
+  SASL: null
+  TLS: null 
+  security-protocol: ""
+  version: "1.0.0"
+  schema-registry-url: https://schema.registry.url
+  schema-registry-credentials:
+    username: httpbasicauthuser
+    password: mypasswordisnotsobasic

pkg/avro/schema.go

@@ -1,7 +1,9 @@
 package avro
 
 import (
+	"encoding/base64"
 	"encoding/binary"
+	"net/http"
 	"sync"
 
 	schemaregistry "github.com/Landoop/schema-registry"
@@ -23,9 +25,32 @@ type SchemaCache struct {
 	codecsBySchemaID map[int]*cachedCodec
 }
 
+type transport struct {
+	underlyingTransport http.RoundTripper
+	encodedCredentials  string
+}
+
+// RoundTrip wraps the underlying transport's RoundTripper and injects a
+// HTTP Basic authentication header if credentials are provided.
+func (t *transport) RoundTrip(req *http.Request) (*http.Response, error) {
+	if t.encodedCredentials != "" {
+		req.Header.Add("Authorization", "Basic "+t.encodedCredentials)
+	}
+	return t.underlyingTransport.RoundTrip(req)
+}
+
 // NewSchemaCache returns a new Cache instance
-func NewSchemaCache(url string) (*SchemaCache, error) {
-	client, err := schemaregistry.NewClient(url)
+func NewSchemaCache(url string, username string, password string) (*SchemaCache, error) {
+	var encodedCredentials string
+	if username != "" {
+		encodedCredentials = base64.StdEncoding.EncodeToString([]byte(username + ":" + password))
+	}
+	httpClient := &http.Client{Transport: &transport{
+		underlyingTransport: http.DefaultTransport,
+		encodedCredentials:  encodedCredentials,
+	}}
+
+	client, err := schemaregistry.NewClient(url, schemaregistry.UsingClient(httpClient))
 	if err != nil {
 		return nil, err
 	}

pkg/config/config.go

@@ -26,14 +26,20 @@ type TLS struct {
 	Insecure      bool
 }
 
+type SchemaRegistryCredentials struct {
+	Username string `yaml:"username"`
+	Password string `yaml:"password"`
+}
+
 type Cluster struct {
-	Name              string
-	Version           string   `yaml:"version"`
-	Brokers           []string `yaml:"brokers"`
-	SASL              *SASL    `yaml:"SASL"`
-	TLS               *TLS     `yaml:"TLS"`
-	SecurityProtocol  string   `yaml:"security-protocol"`
-	SchemaRegistryURL string   `yaml:"schema-registry-url"`
+	Name                      string
+	Version                   string                     `yaml:"version"`
+	Brokers                   []string                   `yaml:"brokers"`
+	SASL                      *SASL                      `yaml:"SASL"`
+	TLS                       *TLS                       `yaml:"TLS"`
+	SecurityProtocol          string                     `yaml:"security-protocol"`
+	SchemaRegistryURL         string                     `yaml:"schema-registry-url"`
+	SchemaRegistryCredentials *SchemaRegistryCredentials `yaml:"schema-registry-credentials"`
 }
 
 type Config struct {