diff --git a/README.md b/README.md index 32473f46..6590452f 100644 --- a/README.md +++ b/README.md @@ -777,14 +777,25 @@ clickhousectl cloud clickpipe create kinesis \ --database default --table events \ --column "event_id:Int64" --column "name:String" -# From PostgreSQL (CDC) +# From PostgreSQL with a publicly trusted certificate (CDC) clickhousectl cloud clickpipe create postgres \ --name my-pg-pipe \ --host db.example.com --pg-database mydb \ --username "$POSTGRES_USERNAME" --password "$POSTGRES_PASSWORD" \ + --publication-name clickpipes \ --table-mapping "public.users:public_users" \ --table-mapping "public.orders:public_orders" +# From PostgreSQL with a private or self-signed CA (CDC) +clickhousectl cloud clickpipe create postgres \ + --name my-private-pg-pipe \ + --host 10.0.0.15 --pg-database mydb \ + --username "$POSTGRES_USERNAME" --password "$POSTGRES_PASSWORD" \ + --ca-certificate ./postgres-ca.pem \ + --tls-host postgres.internal.example.com \ + --publication-name clickpipes \ + --table-mapping "public.users:public_users" + # From MySQL (CDC) # --server-id sets the replication server ID (useful when multiple pipes read # from the same MySQL instance, or to avoid colliding with existing replicas) @@ -810,6 +821,36 @@ clickhousectl cloud clickpipe create bigquery \ --table-mapping "dataset.table:target_table" ``` +#### PostgreSQL ClickPipe prerequisites + +TLS and certificate verification are enabled by default. A source that serves a +complete, publicly trusted certificate chain needs neither `--ca-certificate` +nor `--tls-host`. If the source certificate uses a private or self-signed CA, +pass the CA certificate or bundle as PEM with `--ca-certificate `; the CLI +reads that file and sends its contents in the create request. Certificate +hostname verification defaults to `--host`. Use `--tls-host ` only +when the certificate is issued for a different hostname, such as when `--host` +is an IP address. These options preserve certificate verification; they do not +disable it. + +Before creating a PostgreSQL CDC ClickPipe: + +- Make the PostgreSQL host and port reachable from ClickHouse Cloud. Allow the + [ClickPipes static egress IPs](https://clickhouse.com/docs/integrations/clickpipes/networking/static-ips) + in the source firewall, security group, and `pg_hba.conf`, or configure + supported private connectivity. +- Enable logical replication (`wal_level=logical`) and provision sufficient WAL + senders and replication slots. +- Create a publication. The publication must contain every source table named + by `--table-mapping`; each table must have a primary key or an appropriate + replica identity. +- Give the source user permission to connect, `USAGE` on each mapped schema, + `SELECT` on each mapped table, and the PostgreSQL `REPLICATION` privilege. + +See the [PostgreSQL ClickPipes setup guide](https://clickhouse.com/docs/integrations/clickpipes/postgres), +the [generic PostgreSQL source setup guide](https://clickhouse.com/docs/integrations/clickpipes/postgres/source/generic), +and the [ClickPipes networking and static IP documentation](https://clickhouse.com/docs/integrations/clickpipes/networking/static-ips). + PostgreSQL ClickPipes require one or more complete `--table-mapping schema.table:target_table` values. Ports must be in `1..=65535`. `--auth IAM_ROLE` requires `--iam-role`; the CLI rejects diff --git a/crates/clickhousectl/src/cloud/clickpipes.rs b/crates/clickhousectl/src/cloud/clickpipes.rs index 6a9f8ead..348def78 100644 --- a/crates/clickhousectl/src/cloud/clickpipes.rs +++ b/crates/clickhousectl/src/cloud/clickpipes.rs @@ -346,7 +346,28 @@ POSTGRES INPUT RULES: At least one --table-mapping is required, in schema.table:target_table form. --auth IAM_ROLE requires --iam-role. With basic auth, --iam-role is rejected instead of being silently ignored. - --replication-slot-name is valid only with --replication-mode cdc_only.")] + --replication-slot-name is valid only with --replication-mode cdc_only. + +POSTGRES TLS: + TLS and certificate verification are enabled by default. A source whose + certificate chain is publicly trusted needs no CA file. For a private or + self-signed source CA, pass its PEM CA bundle with --ca-certificate . + Certificate hostname verification uses --host unless --tls-host + overrides it. + +PREREQUISITES: + ClickPipes must be able to reach the source; allow the ClickPipes static IPs + through its network controls. For CDC, enable logical replication, put every + mapped table in the publication, and grant the source user the required + schema, table, and replication privileges. + +DOCUMENTATION: + PostgreSQL ClickPipes setup: + https://clickhouse.com/docs/integrations/clickpipes/postgres + Generic PostgreSQL source setup: + https://clickhouse.com/docs/integrations/clickpipes/postgres/source/generic + ClickPipes networking and static IPs: + https://clickhouse.com/docs/integrations/clickpipes/networking/static-ips")] Postgres(PostgresCreateArgs), /// Create a ClickPipe from MySQL @@ -742,12 +763,12 @@ pub struct PostgresCreateArgs { #[arg(long, required_if_eq("auth", "IAM_ROLE"))] pub iam_role: Option, - /// TLS hostname - #[arg(long)] + /// Certificate hostname override (defaults to --host) + #[arg(long, value_name = "HOSTNAME")] pub tls_host: Option, - /// Path to CA certificate file - #[arg(long)] + /// Path to a PEM CA bundle for a private or self-signed source certificate + #[arg(long, value_name = "PATH")] pub ca_certificate: Option, /// Postgres publication name @@ -1958,6 +1979,38 @@ fn build_postgres_request( }) } +fn postgres_tls_error_hint(message: &str) -> Option<&'static str> { + let message = message.to_ascii_lowercase(); + + if message.contains("x509: certificate signed by unknown authority") { + return Some( + "The source certificate chain is not publicly trusted. For a private or \ + self-signed source CA, pass its PEM CA bundle with \ + `--ca-certificate `.", + ); + } + + if (message.contains("x509: certificate is valid for ") && message.contains(", not ")) + || (message.contains("x509: cannot validate certificate for ") + && message.contains("because it doesn't contain any ip sans")) + { + return Some( + "The source certificate does not match `--host`. Pass the certificate's \ + hostname with `--tls-host `.", + ); + } + + None +} + +fn add_postgres_tls_error_hint(mut error: CloudError) -> CloudError { + if let Some(hint) = postgres_tls_error_hint(&error.message) { + error.message.push_str("\n\nHint: "); + error.message.push_str(hint); + } + error +} + async fn clickpipe_create_postgres( client: &CloudClient, args: &PostgresCreateArgs, @@ -1968,7 +2021,8 @@ async fn clickpipe_create_postgres( let clickpipe = client .create_clickpipe(&org_id, &args.service_id, &request) - .await?; + .await + .map_err(add_postgres_tls_error_hint)?; print_created(&clickpipe, json)?; Ok(()) } @@ -3506,7 +3560,7 @@ mod tests { } #[test] - fn postgres_help_documents_conditional_input_rules() { + fn postgres_help_documents_input_tls_and_source_requirements() { let error = clickpipe_parse_error(&["create", "postgres", "--help"]); assert_eq!(error.kind(), clap::error::ErrorKind::DisplayHelp); let help = error.to_string(); @@ -3520,6 +3574,99 @@ mod tests { ); assert!(help.contains("silently ignored"), "{help}"); assert!(help.contains("--replication-mode cdc_only"), "{help}"); + assert!( + help.contains("TLS and certificate verification are enabled by default"), + "{help}" + ); + assert!(help.contains("publicly trusted needs no CA file"), "{help}"); + assert!(help.contains("PEM CA bundle"), "{help}"); + assert!(help.contains("--ca-certificate "), "{help}"); + assert!(help.contains("--tls-host "), "{help}"); + assert!(help.contains("uses --host unless"), "{help}"); + assert!(help.contains("enable logical replication"), "{help}"); + assert!(help.contains("mapped table in the publication"), "{help}"); + assert!( + help.contains("schema, table, and replication privileges"), + "{help}" + ); + assert!( + help.contains("https://clickhouse.com/docs/integrations/clickpipes/postgres"), + "{help}" + ); + assert!( + help.contains( + "https://clickhouse.com/docs/integrations/clickpipes/networking/static-ips" + ), + "{help}" + ); + } + + #[test] + fn readme_documents_postgres_tls_and_cdc_prerequisites() { + let readme = include_str!("../../../../README.md"); + let postgres = readme + .split_once("### ClickPipes") + .expect("ClickPipes section") + .1 + .split_once("#### Discovering a source schema") + .expect("next ClickPipes section") + .0; + + for expected in [ + "publicly trusted certificate", + "private or self-signed CA", + "--ca-certificate ./postgres-ca.pem", + "--tls-host postgres.internal.example.com", + "TLS and certificate verification are enabled by default", + "defaults to `--host`", + "ClickPipes static egress IPs", + "`wal_level=logical`", + "publication must contain every source table", + "`USAGE` on each mapped schema", + "https://clickhouse.com/docs/integrations/clickpipes/postgres/source/generic", + "https://clickhouse.com/docs/integrations/clickpipes/networking/static-ips", + ] { + assert!( + postgres.contains(expected), + "missing `{expected}`:\n{postgres}" + ); + } + + assert_eq!( + postgres.matches("--publication-name clickpipes").count(), + 2, + "both PostgreSQL examples must use the publication created in the prerequisites" + ); + } + + #[test] + fn postgres_tls_error_hints_are_narrow_and_preserve_api_detail() { + let unknown_authority = "BAD_REQUEST: failed to establish connection: tls: failed to verify certificate: \ + x509: certificate signed by unknown authority"; + let error = add_postgres_tls_error_hint(CloudError::new(unknown_authority)); + assert!(error.message.starts_with(unknown_authority)); + assert!(error.message.contains("--ca-certificate ")); + + let hostname = postgres_tls_error_hint( + "x509: certificate is valid for postgres.internal.example.com, not 10.0.0.8", + ) + .expect("hostname mismatch hint"); + assert!(hostname.contains("--tls-host ")); + let ip_sans = postgres_tls_error_hint( + "x509: cannot validate certificate for 10.0.0.8 because it doesn't contain any IP Sans", + ) + .expect("IP SAN hostname mismatch hint"); + assert!(ip_sans.contains("--tls-host ")); + assert!( + postgres_tls_error_hint( + "BAD_REQUEST: failed to establish connection: connection refused" + ) + .is_none() + ); + assert!( + postgres_tls_error_hint("tls: failed to verify certificate: certificate expired") + .is_none() + ); } #[test] diff --git a/crates/clickhousectl/tests/cli_request_shape_test.rs b/crates/clickhousectl/tests/cli_request_shape_test.rs index 4d4ba6de..cdf7bb5b 100644 --- a/crates/clickhousectl/tests/cli_request_shape_test.rs +++ b/crates/clickhousectl/tests/cli_request_shape_test.rs @@ -2430,6 +2430,62 @@ async fn postgres_ca_certificate_file_contents_flow_to_body() { ); } +#[tokio::test] +async fn postgres_unknown_authority_error_preserves_api_detail_and_adds_ca_hint() { + let mock = MockServer::start().await; + let api_error = "BAD_REQUEST: failed to establish connection: tls: failed to verify \ + certificate: x509: certificate signed by unknown authority"; + Mock::given(method("POST")) + .and(path_regex( + r"^/v1/organizations/[^/]+/services/[^/]+/clickpipes$", + )) + .respond_with(ResponseTemplate::new(400).set_body_json(serde_json::json!({ + "status": 400, + "error": api_error, + }))) + .mount(&mock) + .await; + + let args = postgres_args_minimal(); + let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect(); + let output = invoke_cli_with_cloud_credentials(&mock, &arg_refs); + assert_eq!(output.status.code(), Some(1)); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains(api_error), "{stderr}"); + assert!(stderr.contains("--ca-certificate "), "{stderr}"); + assert!( + stderr.contains("private or self-signed source CA"), + "{stderr}" + ); +} + +#[tokio::test] +async fn postgres_hostname_mismatch_error_preserves_api_detail_and_adds_tls_host_hint() { + let mock = MockServer::start().await; + let api_error = "BAD_REQUEST: failed to establish connection: tls: failed to verify \ + certificate: x509: certificate is valid for postgres.internal.example.com, \ + not 10.0.0.8"; + Mock::given(method("POST")) + .and(path_regex( + r"^/v1/organizations/[^/]+/services/[^/]+/clickpipes$", + )) + .respond_with(ResponseTemplate::new(400).set_body_json(serde_json::json!({ + "status": 400, + "error": api_error, + }))) + .mount(&mock) + .await; + + let args = postgres_args_minimal(); + let arg_refs: Vec<&str> = args.iter().map(String::as_str).collect(); + let output = invoke_cli_with_cloud_credentials(&mock, &arg_refs); + assert_eq!(output.status.code(), Some(1)); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains(api_error), "{stderr}"); + assert!(stderr.contains("--tls-host "), "{stderr}"); + assert!(stderr.contains("does not match `--host`"), "{stderr}"); +} + #[tokio::test] async fn postgres_replication_mode_snapshot_serializes() { let mock = start_mock_clickpipes_api().await;