Add ability to configure TLS for Kafka and HTTP

.gitignore

@@ -12,3 +12,6 @@ vendor/
 build/
 dist/
 .localstack/
+
+#temporary directory created by tests
+tmp_replicator/

Makefile

@@ -11,8 +11,6 @@ go_dirs = `go list ./... | grep -v /build/ | grep -v /vendor/`
 build_dir       = build
 vendor_dir      = vendor
 integration_dir = integration
-cert_dir 		= $(integration_dir)/http
-abs_cert_dir	= $$(pwd)/$(cert_dir)
 ngrok_path		= ${NGROK_DIR}ngrok # Set NGROK_DIR to `/path/to/directory/` for local setup
 
 coverage_dir  = $(build_dir)/coverage
@@ -144,7 +142,6 @@ test: test-setup
 	GO111MODULE=on go tool cover -func=$(coverage_out)
 
 integration-test: test-setup
-	export CERT_DIR=$(abs_cert_dir); \
 	GO111MODULE=on go test $(go_dirs) -v -covermode=count -coverprofile=$(coverage_out)
 	GO111MODULE=on go tool cover -html=$(coverage_out) -o $(coverage_html)
 	GO111MODULE=on go tool cover -func=$(coverage_out)

cmd/cli/cli.go

@@ -22,6 +22,7 @@ import (
 	_ "net/http/pprof"
 
 	"github.com/snowplow-devops/stream-replicator/cmd"
+	"github.com/snowplow-devops/stream-replicator/pkg/common"
 	"github.com/snowplow-devops/stream-replicator/pkg/failure/failureiface"
 	"github.com/snowplow-devops/stream-replicator/pkg/models"
 	"github.com/snowplow-devops/stream-replicator/pkg/observer"
@@ -118,6 +119,11 @@ func RunCli(supportedSourceConfigPairs []sourceconfig.ConfigPair) {
 				stop <- struct{}{}
 			}()
 
+			err := common.DeleteTemporaryDir()
+			if err != nil {
+				log.Debugf(`error deleting tmp directory: %v`, err)
+			}
+
 			select {
 			case <-stop:
 				log.Debug("source.Stop() finished successfully!")

config/component_test.go

@@ -8,6 +8,7 @@ package config
 
 import (
 	"errors"
+	"os"
 	"path/filepath"
 	"reflect"
 	"testing"
@@ -19,6 +20,10 @@ import (
 	"github.com/snowplow-devops/stream-replicator/pkg/target"
 )
 
+func init() {
+	os.Clearenv()
+}
+
 func TestCreateTargetComponentHCL(t *testing.T) {
 	testCases := []struct {
 		File     string
@@ -73,9 +78,9 @@ func TestCreateTargetComponentHCL(t *testing.T) {
 				Headers:                 "",
 				BasicAuthUsername:       "",
 				BasicAuthPassword:       "",
-				CertFile:                "",
-				KeyFile:                 "",
-				CaFile:                  "",
+				TLSCert:                 "",
+				TLSKey:                  "",
+				TLSCa:                   "",
 				SkipVerifyTLS:           false,
 			},
 		},
@@ -90,9 +95,9 @@ func TestCreateTargetComponentHCL(t *testing.T) {
 				Headers:                 "{\"Accept-Language\":\"en-US\"}",
 				BasicAuthUsername:       "testUsername",
 				BasicAuthPassword:       "testPass",
-				CertFile:                "test.cert",
-				KeyFile:                 "test.key",
-				CaFile:                  "test.ca",
+				TLSCert:                 "dGVzdC5jZXJ0",
+				TLSKey:                  "dGVzdC5rZXkK",
+				TLSCa:                   "dGVzdC5jYQ==",
 				SkipVerifyTLS:           true,
 			},
 		},
@@ -112,9 +117,9 @@ func TestCreateTargetComponentHCL(t *testing.T) {
 				SASLUsername:   "",
 				SASLPassword:   "",
 				SASLAlgorithm:  "sha512",
-				CertFile:       "",
-				KeyFile:        "",
-				CaFile:         "",
+				TLSCert:        "",
+				TLSKey:         "",
+				TLSCa:          "",
 				SkipVerifyTLS:  false,
 				ForceSync:      false,
 				FlushFrequency: 0,
@@ -138,9 +143,9 @@ func TestCreateTargetComponentHCL(t *testing.T) {
 				SASLUsername:   "testUsername",
 				SASLPassword:   "testPass",
 				SASLAlgorithm:  "sha256",
-				CertFile:       "test.cert",
-				KeyFile:        "test.key",
-				CaFile:         "test.ca",
+				TLSCert:        "dGVzdC5jZXJ0",
+				TLSKey:         "dGVzdC5rZXkK",
+				TLSCa:          "dGVzdC5jYQ==",
 				SkipVerifyTLS:  true,
 				ForceSync:      true,
 				FlushFrequency: 2,
@@ -217,9 +222,9 @@ func TestCreateFailureTargetComponentENV(t *testing.T) {
 			SASLUsername:   "testUsername",
 			SASLPassword:   "testPass",
 			SASLAlgorithm:  "sha256",
-			CertFile:       "test.cert",
-			KeyFile:        "test.key",
-			CaFile:         "test.ca",
+			TLSCert:        "dGVzdA==",
+			TLSKey:         "dGVzdA==",
+			TLSCa:          "dGVzdA==",
 			SkipVerifyTLS:  true,
 			ForceSync:      true,
 			FlushFrequency: 2,
@@ -230,7 +235,7 @@ func TestCreateFailureTargetComponentENV(t *testing.T) {
 
 	t.Run(testCase.Name, func(t *testing.T) {
 		assert := assert.New(t)
-
+		t.Setenv("STREAM_REPLICATOR_CONFIG_FILE", "")
 		t.Setenv("FAILURE_TARGET_NAME", "kafka")
 		t.Setenv("FAILURE_TARGET_KAFKA_BROKERS", "testBrokers")
 		t.Setenv("FAILURE_TARGET_KAFKA_TOPIC_NAME", "testTopic")
@@ -244,9 +249,9 @@ func TestCreateFailureTargetComponentENV(t *testing.T) {
 		t.Setenv("FAILURE_TARGET_KAFKA_SASL_USERNAME", "testUsername")
 		t.Setenv("FAILURE_TARGET_KAFKA_SASL_PASSWORD", "testPass")
 		t.Setenv("FAILURE_TARGET_KAFKA_SASL_ALGORITHM", "sha256")
-		t.Setenv("FAILURE_TARGET_KAFKA_TLS_CERT_FILE", "test.cert")
-		t.Setenv("FAILURE_TARGET_KAFKA_TLS_KEY_FILE", "test.key")
-		t.Setenv("FAILURE_TARGET_KAFKA_TLS_CA_FILE", "test.ca")
+		t.Setenv("FAILURE_TARGET_KAFKA_TLS_CERT_B64", "dGVzdA==")
+		t.Setenv("FAILURE_TARGET_KAFKA_TLS_KEY_B64", "dGVzdA==")
+		t.Setenv("FAILURE_TARGET_KAFKA_TLS_CA_B64", "dGVzdA==")
 		t.Setenv("FAILURE_TARGET_KAFKA_TLS_SKIP_VERIFY_TLS", "true")
 		t.Setenv("FAILURE_TARGET_KAFKA_FORCE_SYNC_PRODUCER", "true")
 		t.Setenv("FAILURE_TARGET_KAFKA_FLUSH_FREQUENCY", "2")

config/config_test.go

@@ -44,6 +44,7 @@ func TestNewConfig(t *testing.T) {
 	observer, err := c.GetObserver(map[string]string{})
 	assert.NotNil(observer)
 	assert.Nil(err)
+	os.RemoveAll(`tmp_replicator`)
 }
 
 func TestNewConfig_FromEnv(t *testing.T) {

config/examples/failure-targets/http-extended.hcl

@@ -27,16 +27,16 @@ failure_target {
     basic_auth_password        = env.MY_AUTH_PASSWORD
 
     # The optional certificate file for client authentication
-    cert_file                  = "myLocalhost.crt"
+    tls_cert                  = "dGVzdCBzdHJpbmc="
 
     # The optional key file for client authentication
-    key_file                   = "MyLocalhost.key"
+    tls_key                   = "c29tZSBzdHJpbmc="
 
     # The optional certificate authority file for TLS client authentication
-    ca_file                    = "myRootCA.crt"
+    tls_ca                    = "b3RoZXIgc3RyaW5ncw=="
 
     # Whether to skip verifying ssl certificates chain (default: false)
-    # If cert_file and key_file are not provided, this setting is not applied.
+    # If tls_cert and tls_key are not provided, this setting is not applied.
     skip_verify_tls            = true
   }
 }

config/examples/failure-targets/kafka-extended.hcl

@@ -39,13 +39,13 @@ failure_target {
     sasl_algorithm      = "sha256"
 
     # The optional certificate file for client authentication
-    cert_file           = "myLocalhost.crt"
+    tls_cert           = "dGVzdCBzdHJpbmc="
 
     # The optional key file for client authentication
-    key_file            = "MyLocalhost.key"
+    tls_key            = "c29tZSBzdHJpbmc="
 
     # The optional certificate authority file for TLS client authentication
-    ca_file             = "myRootCA.crt"
+    tls_ca             = "b3RoZXIgc3RyaW5ncw=="
 
     # Whether to skip verifying ssl certificates chain (default: false)
     skip_verify_tls     = true

config/examples/targets/http-extended.hcl

@@ -27,16 +27,16 @@ target {
     basic_auth_password        = env.MY_AUTH_PASSWORD
 
     # The optional certificate file for client authentication
-    cert_file                  = "myLocalhost.crt"
+    tls_cert                  = "dGVzdCBzdHJpbmc="
 
     # The optional key file for client authentication
-    key_file                   = "MyLocalhost.key"
+    tls_key                   = "c29tZSBzdHJpbmc="
 
     # The optional certificate authority file for TLS client authentication
-    ca_file                    = "myRootCA.crt"
+    tls_ca                    = "b3RoZXIgc3RyaW5ncw=="
 
     # Whether to skip verifying ssl certificates chain (default: false)
-    # If cert_file and key_file are not provided, this setting is not applied.
+    # If tls_cert and tls_key are not provided, this setting is not applied.
     skip_verify_tls            = true
   }
 }

config/examples/targets/kafka-extended.hcl

@@ -39,13 +39,13 @@ target {
     sasl_algorithm      = "sha256"
 
     # The optional certificate file for client authentication
-    cert_file           = "myLocalhost.crt"
+    tls_cert           = "dGVzdCBzdHJpbmc="
 
     # The optional key file for client authentication
-    key_file            = "MyLocalhost.key"
+    tls_key            = "c29tZSBzdHJpbmc="
 
     # The optional certificate authority file for TLS client authentication
-    ca_file             = "myRootCA.crt"
+    tls_ca             = "b3RoZXIgc3RyaW5ncw=="
 
     # Whether to skip verifying ssl certificates chain (default: false)
     skip_verify_tls     = true

config/test-fixtures/target-http-extended.hcl

@@ -9,9 +9,9 @@ target {
     headers                    = "{\"Accept-Language\":\"en-US\"}"
     basic_auth_username        = "testUsername"
     basic_auth_password        = "testPass"
-    cert_file                  = "test.cert"
-    key_file                   = "test.key"
-    ca_file                    = "test.ca"
+    tls_cert                   = "dGVzdC5jZXJ0"
+    tls_key                    = "dGVzdC5rZXkK"
+    tls_ca                     = "dGVzdC5jYQ=="
     skip_verify_tls            = true
   }
 }

config/test-fixtures/target-kafka-extended.hcl

@@ -14,9 +14,9 @@ target {
     sasl_username       = "testUsername"
     sasl_password       = "testPass"
     sasl_algorithm      = "sha256"
-    cert_file           = "test.cert"
-    key_file            = "test.key"
-    ca_file             = "test.ca"
+    tls_cert            = "dGVzdC5jZXJ0"
+    tls_key             = "dGVzdC5rZXkK"
+    tls_ca              = "dGVzdC5jYQ=="
     skip_verify_tls     = true
     force_sync_producer = true
     flush_frequency     = 2

integration/http/localhost.key

@@ -24,4 +24,4 @@ itxdXvoomlhwDKZv0Y+vPm4V9SBx/36ubf6bM6vKoZTSuv2+ktA/uInFW+y/1mLH
 KD1JlQKBgQDNZry00fN3iJ9stUNEYVaAtXQ1a0/LY/r2NuC04IwemwOyFUvzY7G9
 sNXeIxTYjQ9OCp9+EE1n6Q3yg63MmTrNuD51f0h2tftokYBaoYBny34HuQf0N7qF
 laOI6yiORZ4eGdYrpCq+q+J0fAkRca0M4Nq/lDEw4bric38WpPxV3Q==
------END RSA PRIVATE KEY-----
+-----END RSA PRIVATE KEY-----

integration/http/rootCA.crt

@@ -27,4 +27,4 @@ Ggg1Qo5z0+XT2l+2KhOC02ydgHV1/tT6cVVX3ZkBvvb/WPHmVp9bT8zqeJzrMQkM
 9DaKEyZKw+LYy7sZp4p4giE/JAzBLidsfIdznhYguPjKgboPMfiJvapzyZPEJsDu
 ShYb5uIlytHwAVlGiUgjx+z/YXBQN1vWsCm5pVL4RGdXdcq5HZzZRaJxAUBrfmiU
 uCJPEnUJ1emIqakgSy3yA+9WtQ==
------END CERTIFICATE-----
+-----END CERTIFICATE-----