From 81d1e61c3fd9303fbd56ffb080f0fc06fac8acca Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Philip=20Dub=C3=A9?= Date: Tue, 3 Jun 2025 05:41:40 +0000 Subject: [PATCH 01/12] s3: rootCAs/tlsHost consolidate on S3Config inside peer TODO: ui --- flow/connectors/clickhouse/clickhouse.go | 23 ++++--- flow/connectors/s3/s3.go | 11 +--- flow/connectors/snowflake/qrep.go | 7 +- flow/connectors/snowflake/qrep_avro_sync.go | 2 +- flow/connectors/utils/aws.go | 71 ++++++++++++++++++--- flow/e2e/s3/s3_helper.go | 13 +++- protos/peers.proto | 4 ++ 7 files changed, 96 insertions(+), 35 deletions(-) diff --git a/flow/connectors/clickhouse/clickhouse.go b/flow/connectors/clickhouse/clickhouse.go index 7ab49fef0..a5a4ac0f9 100644 --- a/flow/connectors/clickhouse/clickhouse.go +++ b/flow/connectors/clickhouse/clickhouse.go @@ -52,14 +52,21 @@ 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 + if config.S3 != nil { + awsConfig = utils.NewPeerAWSCredentials(config.S3) + } else { + awsConfig = utils.PeerAWSCredentials{ + Credentials: aws.Credentials{ + AccessKeyID: config.AccessKeyId, + SecretAccessKey: config.SecretAccessKey, + }, + EndpointUrl: config.Endpoint, + Region: config.Region, + } + } + + credentialsProvider, err := utils.GetAWSCredentialsProvider(ctx, "clickhouse", awsConfig) if err != nil { return nil, err } diff --git a/flow/connectors/s3/s3.go b/flow/connectors/s3/s3.go index 9a3aa9865..b362b17b1 100644 --- a/flow/connectors/s3/s3.go +++ b/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 } diff --git a/flow/connectors/snowflake/qrep.go b/flow/connectors/snowflake/qrep.go index 84f4027d0..2959fb0b9 100644 --- a/flow/connectors/snowflake/qrep.go +++ b/flow/connectors/snowflake/qrep.go @@ -131,8 +131,7 @@ 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{}) + provider, err := utils.GetAWSCredentialsProvider(ctx, "snowflake", utils.NewPeerAWSCredentials(c.config.S3)) 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 } } diff --git a/flow/connectors/snowflake/qrep_avro_sync.go b/flow/connectors/snowflake/qrep_avro_sync.go index 0380690e3..87f1eefc2 100644 --- a/flow/connectors/snowflake/qrep_avro_sync.go +++ b/flow/connectors/snowflake/qrep_avro_sync.go @@ -179,7 +179,7 @@ func (s *SnowflakeAvroSyncHandler) writeToAvroFile( s.logger.Info("OCF: Writing records to S3", slog.String(string(shared.PartitionIDKey), partitionID)) - provider, err := utils.GetAWSCredentialsProvider(ctx, "snowflake", utils.PeerAWSCredentials{}) + provider, err := utils.GetAWSCredentialsProvider(ctx, "snowflake", utils.NewPeerAWSCredentials(s.SnowflakeConnector.config.S3)) if err != nil { return nil, err } diff --git a/flow/connectors/utils/aws.go b/flow/connectors/utils/aws.go index f98697115..55c161779 100644 --- a/flow/connectors/utils/aws.go +++ b/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} + } } } diff --git a/flow/e2e/s3/s3_helper.go b/flow/e2e/s3/s3_helper.go index 4524d5eaf..cdee71f1d 100644 --- a/flow/e2e/s3/s3_helper.go +++ b/flow/e2e/s3/s3_helper.go @@ -26,6 +26,15 @@ type S3TestHelper struct { type S3Environment int +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"` +} + const ( Aws S3Environment = iota Gcs @@ -33,7 +42,7 @@ const ( ) func NewS3TestHelper(ctx context.Context, s3environment S3Environment) (*S3TestHelper, error) { - var config utils.S3PeerCredentials + var config S3PeerCredentials var endpoint string var credsPath string var bucketName string @@ -77,7 +86,7 @@ func NewS3TestHelper(ctx context.Context, s3environment S3Environment) (*S3TestH SessionToken: config.SessionToken, }, EndpointUrl: endpointUrlPtr, - }, config.Region) + }, config.Region, nil, "") client, err := utils.CreateS3Client(ctx, provider) if err != nil { return nil, err diff --git a/protos/peers.proto b/protos/peers.proto index c54ce35ff..727c14bd7 100644 --- a/protos/peers.proto +++ b/protos/peers.proto @@ -29,6 +29,7 @@ message SnowflakeConfig { optional string password = 10 [(peerdb_redacted) = true]; // defaults to _PEERDB_INTERNAL optional string metadata_schema = 11; + optional S3Config s3 = 12; } message GcpServiceAccount { @@ -141,6 +142,8 @@ message S3Config { optional string role_arn = 4; optional string region = 5; optional string endpoint = 6; + optional string root_ca = 7 [(peerdb_redacted) = true]; + string tls_host = 8; } message ClickhouseConfig{ @@ -159,6 +162,7 @@ message ClickhouseConfig{ optional string private_key = 13 [(peerdb_redacted) = true]; optional string root_ca = 14 [(peerdb_redacted) = true]; string tls_host = 15; + optional S3Config s3 = 16; } message SqlServerConfig { From 8429d547b10c87409d448202ff9f6a26b3d695f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Philip=20Dub=C3=A9?= Date: Tue, 3 Jun 2025 05:46:59 +0000 Subject: [PATCH 02/12] nexus --- nexus/analyzer/src/lib.rs | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/nexus/analyzer/src/lib.rs b/nexus/analyzer/src/lib.rs index 19601d3f5..4110ed6ed 100644 --- a/nexus/analyzer/src/lib.rs +++ b/nexus/analyzer/src/lib.rs @@ -603,11 +603,6 @@ fn parse_db_options(db_type: DbType, with_options: &[SqlOption]) -> anyhow::Resu Config::BigqueryConfig(bq_config) } DbType::Snowflake => { - let s3_int = opts - .get("s3_integration") - .map(|s| s.to_string()) - .unwrap_or_default(); - let snowflake_config = SnowflakeConfig { account_id: opts .get("account_id") @@ -637,7 +632,11 @@ fn parse_db_options(db_type: DbType, with_options: &[SqlOption]) -> anyhow::Resu .context("unable to parse query_timeout")?, password: opts.get("password").map(|s| s.to_string()), metadata_schema: opts.get("metadata_schema").map(|s| s.to_string()), - s3_integration: s3_int, + s3_integration: opts + .get("s3_integration") + .map(|s| s.to_string()) + .unwrap_or_default(), + s3: None, }; Config::SnowflakeConfig(snowflake_config) } @@ -728,6 +727,11 @@ fn parse_db_options(db_type: DbType, with_options: &[SqlOption]) -> anyhow::Resu region: opts.get("region").map(|s| s.to_string()), role_arn: opts.get("role_arn").map(|s| s.to_string()), endpoint: opts.get("endpoint").map(|s| s.to_string()), + root_ca: opts.get("root_ca").map(|s| s.to_string()), + tls_host: opts + .get("tls_host") + .map(|s| s.to_string()) + .unwrap_or_else(String::new), }; Config::S3Config(s3_config) } @@ -800,6 +804,7 @@ fn parse_db_options(db_type: DbType, with_options: &[SqlOption]) -> anyhow::Resu .get("tls_host") .map(|s| s.to_string()) .unwrap_or_default(), + s3: None, }; Config::ClickhouseConfig(clickhouse_config) } From 2fd1d257b85d5cada2659a8de3d724e98d05dfd2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Philip=20Dub=C3=A9?= Date: Tue, 3 Jun 2025 17:26:10 +0000 Subject: [PATCH 03/12] clippy --- nexus/analyzer/src/lib.rs | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/nexus/analyzer/src/lib.rs b/nexus/analyzer/src/lib.rs index 4110ed6ed..315e6cdc9 100644 --- a/nexus/analyzer/src/lib.rs +++ b/nexus/analyzer/src/lib.rs @@ -63,7 +63,7 @@ impl StatementAnalyzer for PeerExistanceAnalyzer<'_> { }; // Necessary as visit_relations fails to deeply visit some structures. - visit_statements(statement, |stmt| { + let _ = visit_statements(statement, |stmt| { match stmt { Statement::Drop { names, .. } => { for name in names { @@ -73,7 +73,7 @@ impl StatementAnalyzer for PeerExistanceAnalyzer<'_> { Statement::Declare { stmts } => { for stmt in stmts { if let Some(ref query) = stmt.for_query { - visit_relations(query, |relation| { + let _ = visit_relations(query, |relation| { analyze_name(&relation.0[0].value); ControlFlow::<()>::Continue(()) }); @@ -85,7 +85,7 @@ impl StatementAnalyzer for PeerExistanceAnalyzer<'_> { ControlFlow::<()>::Continue(()) }); - visit_relations(statement, |relation| { + let _ = visit_relations(statement, |relation| { analyze_name(&relation.0[0].value); ControlFlow::<()>::Continue(()) }); @@ -278,7 +278,7 @@ impl StatementAnalyzer for PeerDDLAnalyzer { let cdc_staging_path = match raw_options.remove("cdc_staging_path") { Some(Expr::Value(ast::Value::SingleQuotedString(s))) => Some(s.clone()), - _ => Some("".to_string()), + _ => None, }; let max_batch_size: Option = match raw_options.remove("max_batch_size") @@ -731,7 +731,7 @@ fn parse_db_options(db_type: DbType, with_options: &[SqlOption]) -> anyhow::Resu tls_host: opts .get("tls_host") .map(|s| s.to_string()) - .unwrap_or_else(String::new), + .unwrap_or_default(), }; Config::S3Config(s3_config) } From 3b56291b669b0f02ed9f162ecd01b1a42cf21af9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Philip=20Dub=C3=A9?= Date: Tue, 3 Jun 2025 17:29:48 +0000 Subject: [PATCH 04/12] fix ui build --- ui/app/peers/create/[peerType]/helpers/s3.ts | 2 ++ 1 file changed, 2 insertions(+) diff --git a/ui/app/peers/create/[peerType]/helpers/s3.ts b/ui/app/peers/create/[peerType]/helpers/s3.ts index 8e98b9164..3782ee443 100644 --- a/ui/app/peers/create/[peerType]/helpers/s3.ts +++ b/ui/app/peers/create/[peerType]/helpers/s3.ts @@ -59,4 +59,6 @@ export const blankS3Setting: S3Config = { roleArn: undefined, region: undefined, endpoint: '', + rootCa: undefined, + tlsHost: '', }; From da096bed92d6dba6071ce8d29d6c80ca1dece547 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Philip=20Dub=C3=A9?= Date: Tue, 3 Jun 2025 19:18:24 +0000 Subject: [PATCH 05/12] cleanup --- ui/app/peers/create/[peerType]/helpers/ch.ts | 67 +++++++++-------- .../peers/create/[peerType]/helpers/common.ts | 1 + ui/app/peers/create/[peerType]/helpers/s3.ts | 23 ++++++ ui/app/peers/create/[peerType]/page.tsx | 5 +- ui/components/PeerForms/BigqueryConfig.tsx | 6 +- ui/components/PeerForms/ClickhouseConfig.tsx | 45 ++--------- .../PeerForms/ElasticsearchConfigForm.tsx | 2 +- .../PeerForms/Eventhubs/EventhubConfig.tsx | 2 +- .../Eventhubs/EventhubGroupConfig.tsx | 2 +- ui/components/PeerForms/KafkaConfig.tsx | 4 +- ui/components/PeerForms/MySqlForm.tsx | 74 ++----------------- ui/components/PeerForms/PostgresForm.tsx | 71 ++---------------- ui/components/PeerForms/PubSubConfig.tsx | 2 +- ui/components/PeerForms/S3Form.tsx | 2 +- ui/components/PeerForms/SnowflakeForm.tsx | 31 +------- ui/components/PeerForms/common.ts | 65 ++++++++++++++++ 16 files changed, 161 insertions(+), 241 deletions(-) create mode 100644 ui/components/PeerForms/common.ts diff --git a/ui/app/peers/create/[peerType]/helpers/ch.ts b/ui/app/peers/create/[peerType]/helpers/ch.ts index dede1c2d8..f587be724 100644 --- a/ui/app/peers/create/[peerType]/helpers/ch.ts +++ b/ui/app/peers/create/[peerType]/helpers/ch.ts @@ -44,12 +44,43 @@ export const clickhouseSetting: PeerSetting[] = [ tips: 'If you are using a non-TLS connection for ClickHouse server, check this box.', optional: true, }, + { + label: 'Certificate', + stateHandler: (value, setter) => { + if (!value) { + // remove key from state if empty + setter((curr) => { + delete (curr as ClickhouseConfig)['certificate']; + return curr; + }); + } else setter((curr) => ({ ...curr, certificate: value as string })); + }, + type: 'file', + optional: true, + tips: 'This is only needed if the user is authenticated via certificate.', + }, + { + label: 'Private Key', + stateHandler: (value, setter) => { + if (!value) { + // remove key from state if empty + setter((curr) => { + delete (curr as ClickhouseConfig)['privateKey']; + return curr; + }); + } else setter((curr) => ({ ...curr, privateKey: value as string })); + }, + type: 'file', + optional: true, + tips: 'This is only needed if the user is authenticated via certificate.', + }, { label: 'S3 Path', stateHandler: (value, setter) => setter((curr) => ({ ...curr, s3Path: value as string })), tips: `This is an S3 bucket/object URL field. This bucket will be used as our intermediate stage for CDC`, placeholder: 's3://', + s3: true, }, { label: 'Access Key ID', @@ -58,6 +89,7 @@ export const clickhouseSetting: PeerSetting[] = [ tips: 'The AWS access key ID associated with your account.', helpfulLink: 'https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html', + s3: true, }, { label: 'Secret Access Key', @@ -66,12 +98,14 @@ export const clickhouseSetting: PeerSetting[] = [ tips: 'The AWS secret access key associated with the above bucket.', helpfulLink: 'https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_access-keys.html', + s3: true, }, { label: 'Region', stateHandler: (value, setter) => setter((curr) => ({ ...curr, region: value as string })), tips: 'The region where your bucket is located. For example, us-east-1.', + s3: true, }, { label: 'Endpoint', @@ -81,36 +115,7 @@ export const clickhouseSetting: PeerSetting[] = [ 'https://docs.aws.amazon.com/general/latest/gr/s3.html#s3_region', tips: 'An endpoint is the URL of the entry point for an AWS web service.', optional: true, - }, - { - label: 'Certificate', - stateHandler: (value, setter) => { - if (!value) { - // remove key from state if empty - setter((curr) => { - delete (curr as ClickhouseConfig)['certificate']; - return curr; - }); - } else setter((curr) => ({ ...curr, certificate: value as string })); - }, - type: 'file', - optional: true, - tips: 'This is only needed if the user is authenticated via certificate.', - }, - { - label: 'Private Key', - stateHandler: (value, setter) => { - if (!value) { - // remove key from state if empty - setter((curr) => { - delete (curr as ClickhouseConfig)['privateKey']; - return curr; - }); - } else setter((curr) => ({ ...curr, privateKey: value as string })); - }, - type: 'file', - optional: true, - tips: 'This is only needed if the user is authenticated via certificate.', + s3: true, }, { label: 'Root Certificate', @@ -126,6 +131,7 @@ export const clickhouseSetting: PeerSetting[] = [ type: 'file', optional: true, tips: 'If not provided, host CA roots will be used.', + s3: true, }, { label: 'TLS Hostname', @@ -134,6 +140,7 @@ export const clickhouseSetting: PeerSetting[] = [ setter((curr) => ({ ...curr, tlsHost: value as string })), tips: 'Overrides expected hostname during tls cert verification.', optional: true, + s3: true, }, ]; diff --git a/ui/app/peers/create/[peerType]/helpers/common.ts b/ui/app/peers/create/[peerType]/helpers/common.ts index 2a5b4ff2b..f53f6a5a8 100644 --- a/ui/app/peers/create/[peerType]/helpers/common.ts +++ b/ui/app/peers/create/[peerType]/helpers/common.ts @@ -21,6 +21,7 @@ export interface PeerSetting { default?: string | number; placeholder?: string; options?: { value: string; label: string }[]; + s3?: true | undefined; } export function getBlankSetting(dbType: string): PeerConfig { diff --git a/ui/app/peers/create/[peerType]/helpers/s3.ts b/ui/app/peers/create/[peerType]/helpers/s3.ts index 3782ee443..73cf12836 100644 --- a/ui/app/peers/create/[peerType]/helpers/s3.ts +++ b/ui/app/peers/create/[peerType]/helpers/s3.ts @@ -50,6 +50,29 @@ export const s3Setting: PeerSetting[] = [ 'https://docs.aws.amazon.com/IAM/latest/UserGuide/reference_identifiers.html#identifiers-arns', optional: true, }, + { + label: 'Root Certificate', + stateHandler: (value, setter) => { + if (!value) { + // remove key from state if empty + setter((curr) => { + delete (curr as S3Config)['rootCa']; + return curr; + }); + } else setter((curr) => ({ ...curr, rootCa: value as string })); + }, + type: 'file', + optional: true, + tips: 'If not provided, host CA roots will be used.', + }, + { + label: 'TLS Host', + field: 'tlsHost', + stateHandler: (value, setter) => + setter((curr) => ({ ...curr, tlsHost: value as string })), + tips: 'Overrides expected hostname during tls cert verification.', + optional: true, + }, ]; export const blankS3Setting: S3Config = { diff --git a/ui/app/peers/create/[peerType]/page.tsx b/ui/app/peers/create/[peerType]/page.tsx index a25fb70b6..b62ad770b 100644 --- a/ui/app/peers/create/[peerType]/page.tsx +++ b/ui/app/peers/create/[peerType]/page.tsx @@ -74,7 +74,7 @@ export default function CreateConfig({ return peerType; }; - const configComponentMap = (peerType: string) => { + const configComponentMap = () => { switch (getDBType()) { case 'POSTGRES': return ( @@ -82,7 +82,6 @@ export default function CreateConfig({ settings={postgresSetting} setter={setConfig} config={config as PostgresConfig} - type={peerType} /> ); case 'MYSQL': @@ -206,7 +205,7 @@ export default function CreateConfig({ Configuration -
{configComponentMap(peerType)}
+
{configComponentMap()}
- + ); } diff --git a/ui/components/PeerForms/KafkaConfig.tsx b/ui/components/PeerForms/KafkaConfig.tsx index 7fb40d6ae..4cb22af8a 100644 --- a/ui/components/PeerForms/KafkaConfig.tsx +++ b/ui/components/PeerForms/KafkaConfig.tsx @@ -26,7 +26,7 @@ export default function KafkaForm({ setter }: KafkaProps) { {!setting.optional && (