refactor: remove support for client provided ca cert

Makefile

@@ -172,18 +172,15 @@ compose_down: ## Removes integration tests Docker resources
 
 integration_test: ## Runs integration tests
 	echo -e "\033[0;36mNo providerTLS\033[0m"
-	cd tests/integration ; go test -count=1 -v -timeout 5m -cover -coverprofile=int_coverage.out -tags=integration ./
+	cd tests/integration ; go test -shuffle=on -count=1 -v -timeout 10m -cover -coverprofile=int_coverage.out -tags=integration ./...
 
 integration_test_tls: ## Runs integration tests with TLS enabled
 	echo -e "\033[0;31mProvider TLS enabled\033[0m"
-	cd tests/integration ; ARKE_PROVIDER_TLS=true ARKE_BROKER_PORT=5671 go test -timeout 5m -count=1 -v -cover -coverprofile=int_coverage.out -tags=integration ./
+	cd tests/integration ; ARKE_PROVIDER_TLS=true ARKE_BROKER_PORT=5671 go test -shuffle=on -timeout 10m -count=1 -v -cover -coverprofile=int_coverage.out -tags=integration ./...
 
-integration_test_tls_send_ca: ## Runs integraiton tests with TLS enabled by sending TLS certs
-	echo "\033[0;31mProvider TLS enabled (sending CA cert)\033[0m"
-	cd tests/integration ; ARKE_PROVIDER_TLS=sendCA ARKE_BROKER_PORT=5671 go test -count=1 -v -cover -coverprofile=int_coverage.out -tags=integration ./
-integration: compose integration_test ## Runs compose and integration_test
+integration_tls_off: compose integration_test ## Runs compose and integration_test
 
-integration_all: integration integration_test_tls integration_test_tls_send_ca ## Runs all integration_test* targets.
+integration_tls_on: compose integration_test_tls ## Runs compose and integration_test_tls
 
 failover_test: ## Runs connection-resilience and failover integration tests (requires a running arke + RabbitMQ with management API)
 	cd tests/integration ; go test -count=1 -v -timeout 3m -tags=failover ./

README.md

@@ -182,10 +182,7 @@ port.
 
 ### Back-end (Broker connection)
 
-Pass `tls: true` (and optionally `ca_certificate` bytes) in
-`ConnectionConfiguration` when calling `Connect`. Arke will use the
-provided CA certificate for verification, or fall back to the system
-trust store if none is provided.
+Pass `tls: true` in `ConnectionConfiguration` when calling `Connect`.
 Set `ARKE_TRUSTED_CA_CERTIFICATES_PEM_FILE` to supply additional
 trusted CA certificates globally at the process level.
 

api/protobuf-spec/arke.proto

@@ -25,13 +25,12 @@ message ConnectionConfiguration {
     string provider = 3; // Provider type, currently only amqp091.
     string tenant = 4; // Tenant name for this connection. Tenant is not required
     Credentials credentials = 5; // Authentication credentials.
-    bytes ca_certificate = 7; // TLS Certificate authority for broker. Implies tls.
-    bool tls = 8; // Should this provider connection use TLS. If used in conjunction with CaCertificate, the certificate will be used for verification. If no CaCertificate is provided then the providers certificate must be trusted by the system certificates.
+    bool tls = 8; // Should this provider connection use TLS.
     string client_name = 9; // The name of the client connecting.
     int32 admin_port = 10; // The administrative port for the provider (eg. RabbitMQ management port) for any actions needing to be performed by the provider (eg. modifying bindings for RabbitMQ)
 
-    reserved 6, 11;
-    reserved "prefetch_count", "publisher_name";
+    reserved 6, 7, 11;
+    reserved "prefetch_count", "ca_certificate", "publisher_name";
 }
 
 /**

doc/arke_protocol.md

@@ -103,8 +103,7 @@ RabbitMQ and Kafka.
 | provider | [string](#string) |  | Provider type, currently only amqp091. |
 | tenant | [string](#string) |  | Tenant name for this connection. Tenant is not required |
 | credentials | [Credentials](#arke-Credentials) |  | Authentication credentials. |
-| ca_certificate | [bytes](#bytes) |  | TLS Certificate authority for broker. Implies tls. |
-| tls | [bool](#bool) |  | Should this provider connection use TLS. If used in conjunction with CaCertificate, the certificate will be used for verification. If no CaCertificate is provided then the providers certificate must be trusted by the system certificates. |
+| tls | [bool](#bool) |  | Should this provider connection use TLS. |
 | client_name | [string](#string) |  | The name of the client connecting. |
 | admin_port | [int32](#int32) |  | The administrative port for the provider (eg. RabbitMQ management port) for any actions needing to be performed by the provider (eg. modifying bindings for RabbitMQ) |
 

doc/design/deployment-operations-runbook.md

@@ -301,9 +301,10 @@ not rejected — check logs for rate-limit warnings.
 1. Verify the backend broker is reachable from the Arke pod:
   `kubectl exec <pod> -- nc -zv <broker-host> <broker-port>`.
 2. Check TLS configuration: if the broker requires TLS, ensure
-`ConnectionConfiguration.Tls = true` and that
-`SAS_ARKE_TRUSTED_CA_CERTIFICATES_PEM_FILE` points to the correct CA bundle.
+`ConnectionConfiguration.Tls = true`.
 3. Check broker credentials in `ConnectionConfiguration.Credentials`.
+4. Check that `ARKE_TRUSTED_CA_CERTIFICATES_PEM_FILE` points to the correct CA
+    bundle.
 
 ### Connections not cleaned up (connectionMap growing)
 

internal/provider/connectors/amqp091/amqp091.go

@@ -1920,8 +1920,7 @@ func (bd *BrokerDetails) connect() (bool, error) {
 
 	// Use TLS in these scenarios:
 	// * ConnectionConfiguration.TLS = true
-	// * ConnectionConfiguration.CaCertificate is not empty
-	if cf.GetTls() || len(cf.GetCaCertificate()) > 0 {
+	if cf.GetTls() {
 		bd.tlsEnabled = true
 		scheme = "amqps"
 	}

internal/provider/connectors/amqp091/amqp091_test.go

@@ -296,42 +296,44 @@ func TestConnect_TLS_SkipVerify(t *testing.T) {
 	amock.AssertExpectations(t)
 }
 
-func TestConnect_TLS_WithCert(t *testing.T) {
-	prov := NewAMQP091Provider()
-	assert.NotNil(t, prov)
+func TestConnect_TLS(t *testing.T) {
+	for _, verify := range []bool{true, false} {
+		t.Run(fmt.Sprintf("verify=%v", verify), func(t *testing.T) {
+			prov := NewAMQP091Provider()
+			assert.NotNil(t, prov)
 
-	oldGetClientIdentifier := GetClientIdentifier
-	GetClientIdentifier = func(context.Context) (string, error) {
-		return "1234", nil
-	}
+			oldGetClientIdentifier := GetClientIdentifier
+			GetClientIdentifier = func(context.Context) (string, error) {
+				return "1234", nil
+			}
 
-	amock := &amqpConnectionMock{}
-	amock.On("Connect").Return(nil)
-	errs := make(chan amqp091Error)
-	amock.On("NotifyClose").Return(errs)
-	oldNewAmqpConn091 := NewAmqpConn091
-	NewAmqpConn091 = func(string, string, *tls.Config) amqp091ConnectionShim {
-		return amock
-	}
+			amock := &amqpConnectionMock{}
+			amock.On("Connect").Return(nil)
+			errs := make(chan amqp091Error)
+			amock.On("NotifyClose").Return(errs)
+			oldNewAmqpConn091 := NewAmqpConn091
+			NewAmqpConn091 = func(string, string, *tls.Config) amqp091ConnectionShim {
+				return amock
+			}
 
-	defer func() {
-		GetClientIdentifier = oldGetClientIdentifier
-		NewAmqpConn091 = oldNewAmqpConn091
-	}()
+			defer func() {
+				GetClientIdentifier = oldGetClientIdentifier
+				NewAmqpConn091 = oldNewAmqpConn091
+			}()
 
-	ctx := context.Background()
-	cc := &pb.ConnectionConfiguration{}
-	cc.Tls = true
-	cc.CaCertificate = []byte("asdf")
-	err := prov.Connect(ctx, cc, false)
-	defer stopWatcher(ctx, prov)
+			ctx := context.Background()
+			cc := &pb.ConnectionConfiguration{}
+			cc.Tls = true
+			err := prov.Connect(ctx, cc, verify)
+			defer stopWatcher(ctx, prov)
 
-	assert.Nil(t, err)
+			assert.Nil(t, err)
 
-	amock.AssertExpectations(t)
-	// TODO: Figure out a good way to get tlsConfig and see if the cert is set
+			amock.AssertExpectations(t)
+			// TODO: Figure out a good way to get tlsConfig and see if the cert is set
+		})
+	}
 }
-
 func TestConnect_Stats(t *testing.T) {
 	prov := NewAMQP091Provider()
 

test/config/env.go

@@ -41,21 +41,18 @@ func ConnectionConfigurationFromEnv() pb.ConnectionConfiguration {
 	brokerType := getenv("ARKE_BROKER_TYPE", "amqp091")
 	tenant := getenv("ARKE_BROKER_TENANT", "/")
 
-	caCertificate := []byte(getenv("ARKE_BROKER_CA_CERTIFICATE", ""))
-
 	creds := &pb.Credentials{
 		Username: username,
 		Password: password,
 	}
 
 	connConf := pb.ConnectionConfiguration{
-		Host:          hostname,
-		Port:          int32(port),
-		Credentials:   creds,
-		Provider:      brokerType,
-		Tenant:        tenant,
-		CaCertificate: caCertificate,
-		AdminPort:     int32(adminPort),
+		Host:        hostname,
+		Port:        int32(port),
+		Credentials: creds,
+		Provider:    brokerType,
+		Tenant:      tenant,
+		AdminPort:   int32(adminPort),
 	}
 	return connConf //nolint
 }

test/config/env_test.go

@@ -86,7 +86,6 @@ func TestConnectionConfigurationFromEnv_Defaults(t *testing.T) {
 		"ARKE_BROKER_PASSWORD",
 		"ARKE_BROKER_TYPE",
 		"ARKE_BROKER_TENANT",
-		"ARKE_BROKER_CA_CERTIFICATE",
 	}
 
 	for _, v := range envVars {
@@ -107,7 +106,6 @@ func TestConnectionConfigurationFromEnv_Defaults(t *testing.T) {
 	assert.Equal(t, "guest", config.Credentials.Password)
 	assert.Equal(t, "amqp091", config.Provider)
 	assert.Equal(t, "/", config.Tenant)
-	assert.Empty(t, config.CaCertificate)
 }
 
 func TestConnectionConfigurationFromEnv_CustomValues(t *testing.T) {
@@ -119,7 +117,6 @@ func TestConnectionConfigurationFromEnv_CustomValues(t *testing.T) {
 	os.Setenv("ARKE_BROKER_PASSWORD", "secret123")
 	os.Setenv("ARKE_BROKER_TYPE", "amqp10")
 	os.Setenv("ARKE_BROKER_TENANT", "/custom-tenant")
-	os.Setenv("ARKE_BROKER_CA_CERTIFICATE", "-----BEGIN CERTIFICATE-----")
 
 	defer func() {
 		os.Unsetenv("ARKE_BROKER_HOSTNAME")
@@ -129,7 +126,6 @@ func TestConnectionConfigurationFromEnv_CustomValues(t *testing.T) {
 		os.Unsetenv("ARKE_BROKER_PASSWORD")
 		os.Unsetenv("ARKE_BROKER_TYPE")
 		os.Unsetenv("ARKE_BROKER_TENANT")
-		os.Unsetenv("ARKE_BROKER_CA_CERTIFICATE")
 	}()
 
 	config := ConnectionConfigurationFromEnv()
@@ -141,7 +137,6 @@ func TestConnectionConfigurationFromEnv_CustomValues(t *testing.T) {
 	assert.Equal(t, "secret123", config.Credentials.Password)
 	assert.Equal(t, "amqp10", config.Provider)
 	assert.Equal(t, "/custom-tenant", config.Tenant)
-	assert.Equal(t, []byte("-----BEGIN CERTIFICATE-----"), config.CaCertificate)
 }
 
 func TestConnectionConfigurationFromEnv_PartialCustomValues(t *testing.T) {
@@ -164,8 +159,7 @@ func TestConnectionConfigurationFromEnv_PartialCustomValues(t *testing.T) {
 	assert.Equal(t, "myuser", config.Credentials.Username)
 	assert.Equal(t, "guest", config.Credentials.Password) // default
 	assert.Equal(t, "rabbitmq", config.Provider)
-	assert.Equal(t, "/", config.Tenant)   // default
-	assert.Empty(t, config.CaCertificate) // default
+	assert.Equal(t, "/", config.Tenant) // default
 }
 
 func TestConnectionConfigurationFromEnv_PortParsing(t *testing.T) {
@@ -319,45 +313,6 @@ func TestConnectionConfigurationFromEnv_TenantValues(t *testing.T) {
 	}
 }
 
-func TestConnectionConfigurationFromEnv_CACertificate(t *testing.T) {
-	tests := []struct {
-		name        string
-		certificate string
-		expected    []byte
-	}{
-		{
-			name:        "no certificate",
-			certificate: "",
-			expected:    []byte(""),
-		},
-		{
-			name:        "simple certificate string",
-			certificate: "-----BEGIN CERTIFICATE-----\nMIIC...\n-----END CERTIFICATE-----",
-			expected:    []byte("-----BEGIN CERTIFICATE-----\nMIIC...\n-----END CERTIFICATE-----"),
-		},
-		{
-			name:        "single line certificate",
-			certificate: "cert-data",
-			expected:    []byte("cert-data"),
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			if tt.certificate != "" {
-				os.Setenv("ARKE_BROKER_CA_CERTIFICATE", tt.certificate)
-				defer os.Unsetenv("ARKE_BROKER_CA_CERTIFICATE")
-			} else {
-				os.Unsetenv("ARKE_BROKER_CA_CERTIFICATE")
-			}
-
-			config := ConnectionConfigurationFromEnv()
-
-			assert.Equal(t, tt.expected, config.CaCertificate)
-		})
-	}
-}
-
 func TestConnectionConfigurationFromEnv_EmptyStringValues(t *testing.T) {
 	// Set all vars to empty strings
 	os.Setenv("ARKE_BROKER_HOSTNAME", "")

test/messagefunctions/messagefunctions.go

@@ -26,15 +26,7 @@ func connectConfig(clientName string) *pb.ConnectionConfiguration {
 
 	providerTLS := strings.ToLower(os.Getenv("ARKE_PROVIDER_TLS"))
 
-	switch providerTLS {
-	case "sendca":
-		cacert, err := os.ReadFile("certs/testca/ca_certificate.pem")
-		if err != nil {
-			log.Fatalf("Error reading provider CA cert: %v", err)
-		}
-		connConfig.Tls = true
-		connConfig.CaCertificate = cacert
-	case "true":
+	if providerTLS == "true" {
 		connConfig.Tls = true
 	}
 

tests/integration/connection_helpers_test.go

@@ -44,14 +44,7 @@ func connectConfig(clientName string) *pb.ConnectionConfiguration {
 
 	providerTLS := strings.ToLower(os.Getenv("ARKE_PROVIDER_TLS"))
 
-	if providerTLS == "sendca" {
-		cacert, err := os.ReadFile("certs/testca/ca_certificate.pem")
-		if err != nil {
-			log.Fatalf("Error reading provider CA cert: %v", err)
-		}
-		connConfig.Tls = true
-		connConfig.CaCertificate = cacert
-	} else if providerTLS == "true" {
+	if providerTLS == "true" {
 		connConfig.Tls = true
 	}