s3: rootCAs/tlsHost

.github/workflows/flow.yml

@@ -77,8 +77,6 @@ jobs:
         run: |
           # No need to update man pages on package install
           sudo apt-get remove --purge man-db
-
-          sudo apt-get update
           sudo apt-get install libgeos-dev
 
       - run: go mod download
@@ -126,6 +124,15 @@ jobs:
         if: matrix.db-version.mysql == 'maria'
         run: docker run -d --rm --name mariadb -p 3306:3306 -e MARIADB_ROOT_PASSWORD=cipass mariadb:lts --log-bin=maria
 
+      - name: MinIO TLS
+        run: >
+          mkdir -p certs &&
+          openssl genrsa -out certs/cert.key 2048 &&
+          openssl req -new -key certs/cert.key -out certs/cert.csr -subj /CN=minio.local &&
+          openssl x509 -req -days 3650 -in certs/cert.csr -signkey certs/cert.key -out certs/cert.crt &&
+          chown -R 1001 certs &&
+          docker run -d --rm --name miniotls -p 9998:9998 -v "$PWD/certs:/certs" -e MINIO_SCHEME=https bitnami/minio:latest
+
       - name: create hstore extension, increase logical replication limits, and setup catalog database
         run: >
           docker exec "${{ job.services.catalog.id }}" psql -U postgres -c "CREATE EXTENSION hstore;"
@@ -184,6 +191,7 @@ jobs:
           AWS_ACCESS_KEY_ID: minio
           AWS_SECRET_ACCESS_KEY: miniosecret
           AWS_REGION: us-east-1
+          AWS_ENDPOINT_URL_S3_TLS: https://localhost:9998
           PEERDB_CLICKHOUSE_AWS_CREDENTIALS_AWS_ACCESS_KEY_ID: minio
           PEERDB_CLICKHOUSE_AWS_CREDENTIALS_AWS_SECRET_ACCESS_KEY: miniosecret
           PEERDB_CLICKHOUSE_AWS_CREDENTIALS_AWS_REGION: us-east-1

flow/connectors/clickhouse/clickhouse.go

@@ -52,19 +52,28 @@ func NewClickHouseConnector(
 		return nil, err
 	}
 
-	credentialsProvider, err := utils.GetAWSCredentialsProvider(ctx, "clickhouse", utils.PeerAWSCredentials{
-		Credentials: aws.Credentials{
-			AccessKeyID:     config.AccessKeyId,
-			SecretAccessKey: config.SecretAccessKey,
-		},
-		EndpointUrl: config.Endpoint,
-		Region:      config.Region,
-	})
+	var awsConfig utils.PeerAWSCredentials
+	var awsBucketPath string
+	if config.S3 != nil {
+		awsConfig = utils.NewPeerAWSCredentials(config.S3)
+		awsBucketPath = config.S3.Url
+	} else {
+		awsConfig = utils.PeerAWSCredentials{
+			Credentials: aws.Credentials{
+				AccessKeyID:     config.AccessKeyId,
+				SecretAccessKey: config.SecretAccessKey,
+			},
+			EndpointUrl: config.Endpoint,
+			Region:      config.Region,
+		}
+		awsBucketPath = config.S3Path
+	}
+
+	credentialsProvider, err := utils.GetAWSCredentialsProvider(ctx, "clickhouse", awsConfig)
 	if err != nil {
 		return nil, err
 	}
 
-	awsBucketPath := config.S3Path
 	if awsBucketPath == "" {
 		deploymentUID := internal.PeerDBDeploymentUID()
 		flowName, _ := ctx.Value(shared.FlowNameKey).(string)

flow/connectors/s3/s3.go

@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"strconv"
 
-	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"go.temporal.io/sdk/log"
 
@@ -30,15 +29,7 @@ func NewS3Connector(
 ) (*S3Connector, error) {
 	logger := internal.LoggerFromCtx(ctx)
 
-	provider, err := utils.GetAWSCredentialsProvider(ctx, "s3", utils.PeerAWSCredentials{
-		Credentials: aws.Credentials{
-			AccessKeyID:     config.GetAccessKeyId(),
-			SecretAccessKey: config.GetSecretAccessKey(),
-		},
-		RoleArn:     config.RoleArn,
-		EndpointUrl: config.Endpoint,
-		Region:      config.GetRegion(),
-	})
+	provider, err := utils.GetAWSCredentialsProvider(ctx, "s3", utils.NewPeerAWSCredentials(config))
 	if err != nil {
 		return nil, err
 	}

flow/connectors/snowflake/qrep.go

@@ -131,7 +131,6 @@ func (c *SnowflakeConnector) createExternalStage(ctx context.Context, stageName
 
 	cleanURL := fmt.Sprintf("s3://%s/%s/%s", s3o.Bucket, s3o.Prefix, config.FlowJobName)
 
-	s3Int := c.config.S3Integration
 	provider, err := utils.GetAWSCredentialsProvider(ctx, "snowflake", utils.PeerAWSCredentials{})
 	if err != nil {
 		return "", err
@@ -141,7 +140,7 @@ func (c *SnowflakeConnector) createExternalStage(ctx context.Context, stageName
 	if err != nil {
 		return "", err
 	}
-	if s3Int == "" {
+	if c.config.S3Integration == "" {
 		credsStr := fmt.Sprintf("CREDENTIALS=(AWS_KEY_ID='%s' AWS_SECRET_KEY='%s' AWS_TOKEN='%s')",
 			creds.AWS.AccessKeyID, creds.AWS.SecretAccessKey, creds.AWS.SessionToken)
 		stageStatement := `
@@ -156,7 +155,7 @@ func (c *SnowflakeConnector) createExternalStage(ctx context.Context, stageName
 		URL = '%s'
 		STORAGE_INTEGRATION = %s
 		FILE_FORMAT = (TYPE = AVRO);`
-		return fmt.Sprintf(stageStatement, stageName, cleanURL, s3Int), nil
+		return fmt.Sprintf(stageStatement, stageName, cleanURL, c.config.S3Integration), nil
 	}
 }
 

flow/connectors/utils/aws.go

@@ -2,6 +2,7 @@ package utils
 
 import (
 	"context"
+	"crypto/tls"
 	"fmt"
 	"net/http"
 	"net/url"
@@ -21,7 +22,9 @@ import (
 	"github.com/aws/smithy-go/ptr"
 	"github.com/google/uuid"
 
+	"github.com/PeerDB-io/peerdb/flow/generated/protos"
 	"github.com/PeerDB-io/peerdb/flow/internal"
+	"github.com/PeerDB-io/peerdb/flow/shared"
 )
 
 const (
@@ -45,15 +48,26 @@ type PeerAWSCredentials struct {
 	ChainedRoleArn *string
 	EndpointUrl    *string
 	Region         string
+	RootCAs        *string
+	TlsHost        string
 }
 
-type S3PeerCredentials struct {
-	AccessKeyID     string `json:"accessKeyId"`
-	SecretAccessKey string `json:"secretAccessKey"`
-	AwsRoleArn      string `json:"awsRoleArn"`
-	SessionToken    string `json:"sessionToken"`
-	Region          string `json:"region"`
-	Endpoint        string `json:"endpoint"`
+func NewPeerAWSCredentials(s3 *protos.S3Config) PeerAWSCredentials {
+	if s3 == nil {
+		return PeerAWSCredentials{}
+	}
+	return PeerAWSCredentials{
+		Credentials: aws.Credentials{
+			AccessKeyID:     s3.GetAccessKeyId(),
+			SecretAccessKey: s3.GetSecretAccessKey(),
+		},
+		RoleArn:        s3.RoleArn,
+		ChainedRoleArn: nil,
+		EndpointUrl:    s3.Endpoint,
+		Region:         s3.GetRegion(),
+		RootCAs:        s3.RootCa,
+		TlsHost:        s3.TlsHost,
+	}
 }
 
 type ClickHouseS3Credentials struct {
@@ -71,6 +85,7 @@ type AWSCredentialsProvider interface {
 	GetUnderlyingProvider() aws.CredentialsProvider
 	GetRegion() string
 	GetEndpointURL() string
+	GetTlsConfig() (*string, string)
 }
 
 type ConfigBasedAWSCredentialsProvider struct {
@@ -98,6 +113,10 @@ func (r *ConfigBasedAWSCredentialsProvider) GetEndpointURL() string {
 	return endpoint
 }
 
+func (r *ConfigBasedAWSCredentialsProvider) GetTlsConfig() (*string, string) {
+	return nil, ""
+}
+
 // Retrieve should be called as late as possible in order to have credentials with latest expiry
 func (r *ConfigBasedAWSCredentialsProvider) Retrieve(ctx context.Context) (AWSCredentials, error) {
 	retrieved, err := r.config.Credentials.Retrieve(ctx)
@@ -113,12 +132,16 @@ func (r *ConfigBasedAWSCredentialsProvider) Retrieve(ctx context.Context) (AWSCr
 type StaticAWSCredentialsProvider struct {
 	credentials AWSCredentials
 	region      string
+	rootCAs     *string
+	tlsHost     string
 }
 
-func NewStaticAWSCredentialsProvider(credentials AWSCredentials, region string) *StaticAWSCredentialsProvider {
+func NewStaticAWSCredentialsProvider(credentials AWSCredentials, region string, rootCAs *string, tlsHost string) *StaticAWSCredentialsProvider {
 	return &StaticAWSCredentialsProvider{
 		credentials: credentials,
 		region:      region,
+		rootCAs:     rootCAs,
+		tlsHost:     tlsHost,
 	}
 }
 
@@ -142,6 +165,10 @@ func (s *StaticAWSCredentialsProvider) GetEndpointURL() string {
 	return ""
 }
 
+func (s *StaticAWSCredentialsProvider) GetTlsConfig() (*string, string) {
+	return s.rootCAs, s.tlsHost
+}
+
 type AssumeRoleBasedAWSCredentialsProvider struct {
 	Provider aws.CredentialsProvider // New Credentials
 	config   aws.Config              // Initial Config
@@ -194,6 +221,10 @@ func (a *AssumeRoleBasedAWSCredentialsProvider) GetEndpointURL() string {
 	return endpoint
 }
 
+func (a *AssumeRoleBasedAWSCredentialsProvider) GetTlsConfig() (*string, string) {
+	return nil, ""
+}
+
 func getPeerDBAWSEnv(connectorName string, awsKey string) string {
 	return os.Getenv(fmt.Sprintf("PEERDB_%s_AWS_CREDENTIALS_%s", strings.ToUpper(connectorName), awsKey))
 }
@@ -203,6 +234,8 @@ func LoadPeerDBAWSEnvConfigProvider(connectorName string) *StaticAWSCredentialsP
 	secretAccessKey := getPeerDBAWSEnv(connectorName, "AWS_SECRET_ACCESS_KEY")
 	region := getPeerDBAWSEnv(connectorName, "AWS_REGION")
 	endpointUrl := getPeerDBAWSEnv(connectorName, "AWS_ENDPOINT_URL_S3")
+	rootCa := getPeerDBAWSEnv(connectorName, "ROOT_CA")
+	tlsHost := getPeerDBAWSEnv(connectorName, "TLS_HOST")
 	var endpointUrlPtr *string
 	if endpointUrl != "" {
 		endpointUrlPtr = &endpointUrl
@@ -212,13 +245,18 @@ func LoadPeerDBAWSEnvConfigProvider(connectorName string) *StaticAWSCredentialsP
 		return nil
 	}
 
+	var rootCAs *string
+	if rootCa != "" {
+		rootCAs = &rootCa
+	}
+
 	return NewStaticAWSCredentialsProvider(AWSCredentials{
 		AWS: aws.Credentials{
 			AccessKeyID:     accessKeyId,
 			SecretAccessKey: secretAccessKey,
 		},
 		EndpointUrl: endpointUrlPtr,
-	}, region)
+	}, region, rootCAs, tlsHost)
 }
 
 func GetAWSCredentialsProvider(ctx context.Context, connectorName string, peerCredentials PeerAWSCredentials) (AWSCredentialsProvider, error) {
@@ -230,7 +268,7 @@ func GetAWSCredentialsProvider(ctx context.Context, connectorName string, peerCr
 		staticProvider := NewStaticAWSCredentialsProvider(AWSCredentials{
 			AWS:         peerCredentials.Credentials,
 			EndpointUrl: peerCredentials.EndpointUrl,
-		}, peerCredentials.Region)
+		}, peerCredentials.Region, peerCredentials.RootCAs, peerCredentials.TlsHost)
 		if peerCredentials.RoleArn == nil || *peerCredentials.RoleArn == "" {
 			logger.Info("Received AWS credentials from peer for connector: " + connectorName)
 			return staticProvider, nil
@@ -364,6 +402,19 @@ func CreateS3Client(ctx context.Context, credsProvider AWSCredentialsProvider) (
 					region:      options.Region,
 				},
 			}
+		} else {
+			rootCAs, tlsHost := credsProvider.GetTlsConfig()
+			if rootCAs != nil || tlsHost != "" {
+				// start with a clone of DefaultTransport so we keep http2, idle-conns, etc.
+				tlsConfig, err := shared.CreateTlsConfig(tls.VersionTLS13, rootCAs, tlsHost, tlsHost, tlsHost == "")
+				if err != nil {
+					return nil, err
+				}
+
+				tr := http.DefaultTransport.(*http.Transport).Clone()
+				tr.TLSClientConfig = tlsConfig
+				options.HTTPClient = &http.Client{Transport: tr}
+			}
 		}
 	}
 

flow/e2e/clickhouse/clickhouse.go

@@ -64,25 +64,16 @@ func (s ClickHouseSuite) Peer() *protos.Peer {
 }
 
 func (s ClickHouseSuite) PeerForDatabase(dbname string) *protos.Peer {
-	region := ""
-	if s.s3Helper.S3Config.Region != nil {
-		region = *s.s3Helper.S3Config.Region
-	}
-
 	ret := &protos.Peer{
 		Name: e2e.AddSuffix(s, dbname),
 		Type: protos.DBType_CLICKHOUSE,
 		Config: &protos.Peer_ClickhouseConfig{
 			ClickhouseConfig: &protos.ClickhouseConfig{
-				Host:            "localhost",
-				Port:            9000,
-				Database:        dbname,
-				S3Path:          s.s3Helper.BucketName,
-				AccessKeyId:     *s.s3Helper.S3Config.AccessKeyId,
-				SecretAccessKey: *s.s3Helper.S3Config.SecretAccessKey,
-				Region:          region,
-				DisableTls:      true,
-				Endpoint:        s.s3Helper.S3Config.Endpoint,
+				Host:       "localhost",
+				Port:       9000,
+				Database:   dbname,
+				DisableTls: true,
+				S3:         s.s3Helper.S3Config,
 			},
 		},
 	}

flow/e2e/s3/cdc_s3_test.go

@@ -60,20 +60,22 @@ func (s PeerFlowE2ETestSuiteS3) Test_Complete_Simple_Flow_S3() {
 		ctx, cancel := context.WithTimeout(s.t.Context(), 25*time.Second)
 		defer cancel()
 		files, err := s.s3Helper.ListAllFiles(ctx, flowJobName)
-		s.t.Logf("Files in Test_Complete_Simple_Flow_S3 %s: %d", flowJobName, len(files))
 		e2e.EnvNoError(s.t, env, err)
+		s.t.Logf("Files in Test_Complete_Simple_Flow_S3 %s: %d", flowJobName, len(files))
 		return len(files) == 4
 	})
 
-	// s3 normalize is nop, so check peerdb_stats directly that batch finalized
 	pool, err := internal.GetCatalogConnectionPoolFromEnv(s.t.Context())
 	require.NoError(s.t, err)
-	var count int64
-	require.NoError(s.t, pool.QueryRow(s.t.Context(),
-		"select count(*) from peerdb_stats.cdc_batches where flow_name = $1 and end_time is not null",
-		flowJobName,
-	).Scan(&count))
-	require.Equal(s.t, int64(4), count)
+	e2e.EnvWaitFor(s.t, env, time.Minute, "waiting for cdc batch completion", func() bool {
+		// s3 normalize is nop, so check peerdb_stats directly that batch finalized
+		var count int64
+		require.NoError(s.t, pool.QueryRow(s.t.Context(),
+			"select count(*) from peerdb_stats.cdc_batches where flow_name = $1 and end_time is not null",
+			flowJobName,
+		).Scan(&count))
+		return count == 4
+	})
 
 	env.Cancel(s.t.Context())
 	e2e.RequireEnvCanceled(s.t, env)

flow/e2e/s3/qrep_flow_s3_test.go

@@ -116,6 +116,11 @@ func SetupSuiteMinIO(t *testing.T) PeerFlowE2ETestSuiteS3 {
 	return setupSuite(t, Minio)
 }
 
+func SetupSuiteMinIO_TLS(t *testing.T) PeerFlowE2ETestSuiteS3 {
+	t.Helper()
+	return setupSuite(t, MinioTls)
+}
+
 func (s PeerFlowE2ETestSuiteS3) Test_Complete_QRep_Flow_S3() {
 	if s.s3Helper == nil {
 		s.t.Skip("Skipping S3 test")