redpanda: allow using SASL and TLS together (#2140)

The Admin HTTP client used in the user setup would before fail when trying to
talk to the admin API via HTTP. With TLS enabled, it'll have to use https.

Signed-off-by: Stephan Renatus <[email protected]>
Co-authored-by: Manuel de la Peña <[email protected]>

modules/redpanda/admin_api.go

@@ -12,10 +12,19 @@ import (
 
 type AdminAPIClient struct {
 	BaseURL string
+	client  *http.Client
 }
 
 func NewAdminAPIClient(baseURL string) *AdminAPIClient {
-	return &AdminAPIClient{BaseURL: baseURL}
+	return &AdminAPIClient{
+		BaseURL: baseURL,
+		client:  http.DefaultClient,
+	}
+}
+
+func (cl *AdminAPIClient) WithHTTPClient(c *http.Client) *AdminAPIClient {
+	cl.client = c
+	return cl
 }
 
 type createUserRequest struct {
@@ -46,7 +55,7 @@ func (cl *AdminAPIClient) CreateUser(ctx context.Context, username, password str
 	}
 	req.Header.Set("Content-Type", "application/json")
 
-	resp, err := http.DefaultClient.Do(req)
+	resp, err := cl.client.Do(req)
 	if err != nil {
 		return fmt.Errorf("request failed: %w", err)
 	}

modules/redpanda/redpanda.go

@@ -3,9 +3,12 @@ package redpanda
 import (
 	"bytes"
 	"context"
+	"crypto/tls"
+	"crypto/x509"
 	_ "embed"
 	"fmt"
 	"math"
+	"net/http"
 	"os"
 	"path/filepath"
 	"text/template"
@@ -193,15 +196,39 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize
 		return nil, fmt.Errorf("failed to wait for Redpanda readiness: %w", err)
 	}
 
+	scheme := "http"
+	if settings.EnableTLS {
+		scheme += "s"
+	}
+
 	// 9. Create Redpanda Service Accounts if configured to do so.
 	if len(settings.ServiceAccounts) > 0 {
 		adminAPIPort, err := container.MappedPort(ctx, nat.Port(defaultAdminAPIPort))
 		if err != nil {
 			return nil, fmt.Errorf("failed to get mapped Admin API port: %w", err)
 		}
 
-		adminAPIUrl := fmt.Sprintf("http://%v:%d", hostIP, adminAPIPort.Int())
+		adminAPIUrl := fmt.Sprintf("%s://%v:%d", scheme, hostIP, adminAPIPort.Int())
 		adminCl := NewAdminAPIClient(adminAPIUrl)
+		if settings.EnableTLS {
+			cert, err := tls.X509KeyPair(settings.cert, settings.key)
+			if err != nil {
+				return nil, fmt.Errorf("failed to create admin client with cert: %w", err)
+			}
+			caCertPool := x509.NewCertPool()
+			caCertPool.AppendCertsFromPEM(settings.cert)
+			adminCl = adminCl.WithHTTPClient(&http.Client{
+				Timeout: 5 * time.Second,
+				Transport: &http.Transport{
+					ForceAttemptHTTP2:   true,
+					TLSHandshakeTimeout: 10 * time.Second,
+					TLSClientConfig: &tls.Config{
+						Certificates: []tls.Certificate{cert},
+						RootCAs:      caCertPool,
+					},
+				},
+			})
+		}
 
 		for username, password := range settings.ServiceAccounts {
 			if err := adminCl.CreateUser(ctx, username, password); err != nil {
@@ -210,11 +237,6 @@ func RunContainer(ctx context.Context, opts ...testcontainers.ContainerCustomize
 		}
 	}
 
-	scheme := "http"
-	if settings.EnableTLS {
-		scheme += "s"
-	}
-
 	return &Container{Container: container, urlScheme: scheme}, nil
 }
 

modules/redpanda/redpanda_test.go

@@ -300,6 +300,50 @@ func TestRedpandaWithTLS(t *testing.T) {
 	require.Error(t, results.FirstErr(), kerr.UnknownTopicOrPartition)
 }
 
+func TestRedpandaWithTLSAndSASL(t *testing.T) {
+	cert, err := tls.X509KeyPair(localhostCert, localhostKey)
+	require.NoError(t, err, "failed to load key pair")
+
+	ctx := context.Background()
+
+	container, err := RunContainer(ctx,
+		WithTLS(localhostCert, localhostKey),
+		WithEnableSASL(),
+		WithEnableKafkaAuthorization(),
+		WithNewServiceAccount("superuser-1", "test"),
+		WithSuperusers("superuser-1"),
+	)
+	require.NoError(t, err)
+
+	t.Cleanup(func() {
+		if err := container.Terminate(ctx); err != nil {
+			t.Fatalf("failed to terminate container: %s", err)
+		}
+	})
+
+	caCertPool := x509.NewCertPool()
+	caCertPool.AppendCertsFromPEM(localhostCert)
+
+	tlsConfig := &tls.Config{
+		Certificates: []tls.Certificate{cert},
+		RootCAs:      caCertPool,
+	}
+
+	broker, err := container.KafkaSeedBroker(ctx)
+	require.NoError(t, err)
+
+	kafkaCl, err := kgo.NewClient(
+		kgo.SeedBrokers(broker),
+		kgo.DialTLSConfig(tlsConfig),
+		kgo.SASL(scram.Auth{
+			User: "superuser-1",
+			Pass: "test",
+		}.AsSha256Mechanism()),
+	)
+	require.NoError(t, err)
+	defer kafkaCl.Close()
+}
+
 func TestRedpandaListener_Simple(t *testing.T) {
 	ctx := context.Background()