Make PostgreSQL ClickPipe TLS failures actionable

README.md

@@ -777,14 +777,25 @@ clickhousectl cloud clickpipe create kinesis <service-id> \
   --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 <service-id> \
   --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 <service-id> \
+  --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 <service-id> \
   --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 <PATH>`; the CLI
+reads that file and sends its contents in the create request. Certificate
+hostname verification defaults to `--host`. Use `--tls-host <HOSTNAME>` 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

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 <PATH>.
+  Certificate hostname verification uses --host unless --tls-host <HOSTNAME>
+  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<String>,
 
-    /// TLS hostname
-    #[arg(long)]
+    /// Certificate hostname override (defaults to --host)
+    #[arg(long, value_name = "HOSTNAME")]
     pub tls_host: Option<String>,
 
-    /// 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<String>,
 
     /// 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 <PATH>`.",
+        );
+    }
+
+    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 <HOSTNAME>`.",
+        );
+    }
+
+    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 <PATH>"), "{help}");
+        assert!(help.contains("--tls-host <HOSTNAME>"), "{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 <PATH>"));
+
+        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 <HOSTNAME>"));
+        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 <HOSTNAME>"));
+        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]

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 <PATH>"), "{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 <HOSTNAME>"), "{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;