remove cert chain and redact auth from log

crates/sparrow-main/src/prepare.rs

@@ -163,7 +163,7 @@ impl PrepareCommand {
                 .change_context(Error::MissingTableConfig)
                 .attach_printable("missing env var PULSAR_TOPIC")?;
 
-            let pulsar_config = Some(PulsarConfig {
+            let pulsar_config = PulsarConfig {
                 admin_service_url: config["webServiceUrl"].clone(),
                 broker_service_url: config["brokerServiceUrl"].clone(),
                 auth_plugin: config["authPlugin"].clone(),
@@ -172,13 +172,13 @@ impl PrepareCommand {
                 namespace: pulsar_namespace,
                 topic_name: pulsar_topic,
                 ..Default::default()
-            });
-            tracing::debug!("Pulsar config is {:?}", pulsar_config);
+            };
+            tracing::debug!("Pulsar config is {:?}", redact_auth_field(&pulsar_config));
 
             SourceData {
                 source: Some(source_data::Source::PulsarSubscription(
                     PulsarSubscription {
-                        config: pulsar_config,
+                        config: Some(pulsar_config),
                         subscription_id: pulsar_subscription,
                         last_publish_time: None,
                     },
@@ -229,3 +229,13 @@ impl PrepareCommand {
         Ok(())
     }
 }
+
+// Hack to remove auth fields from log output.
+//
+// Ideally, we use tonic or prost reflection to redact all
+// fields marked as sensitive.
+fn redact_auth_field(pulsar: &PulsarConfig) -> PulsarConfig {
+    let mut clone = pulsar.clone();
+    clone.auth_params = "...redacted...".to_owned();
+    clone
+}

crates/sparrow-runtime/src/execute/output/pulsar.rs

@@ -185,10 +185,7 @@ async fn build_client(
             token
         } else {
             error_stack::bail!(Error::PulsarAuth {
-                context: format!(
-                    "expected \"token:\" style prefix. Saw {}",
-                    pulsar.auth_params
-                )
+                context: format!("expected \"token:\" prefix",)
             })
         };
 
@@ -199,16 +196,6 @@ async fn build_client(
         client_builder = client_builder.with_auth(pulsar_auth);
     };
 
-    // Add TLS encryption
-    if !pulsar.certificate_chain.is_empty() {
-        // The default values for the other configs are explicitly show here for
-        // clarity. We can allow the user to configure these if requested.
-        client_builder = client_builder
-            .with_allow_insecure_connection(false)
-            .with_tls_hostname_verification_enabled(true)
-            .with_certificate_chain(pulsar.certificate_chain.as_bytes().to_vec());
-    };
-
     let client = client_builder
         .build()
         .await

crates/sparrow-runtime/src/execute/progress_reporter.rs

@@ -172,7 +172,6 @@ impl ProgressTracker {
                             tenant: config.tenant.clone(),
                             namespace: config.namespace.clone(),
                             topic_name: config.topic_name.clone(),
-                            certificate_chain: config.certificate_chain.clone(),
                             admin_service_url: config.admin_service_url.clone(),
                         }),
                     })),

proto/kaskada/kaskada/v1alpha/pulsar.proto

@@ -23,24 +23,21 @@ message PulsarConfig {
   // e.g. "token:xxx"
   string auth_params = 4 [(kaskada.v1alpha.sensitive) = true];
 
-  // A custom certificate chain to authenticate the server in TLS connections.
-  string certificate_chain = 5;
-
   // The topic tenant within the instance.
   //
   // Defaults to "public".
-  string tenant = 6;
+  string tenant = 5;
 
   // The administrative unit of topics, which acts as a grouping
   // mechanism for related topics.
   //
   // Defaults to "default".
-  string namespace = 7;
+  string namespace = 6;
 
   // The final part of the topic url.
   //
   // Defaults to a randomly generated uuid.
-  string topic_name = 8;
+  string topic_name = 7;
 }
 
 // Configuration for a single source of data from a pulsar topic.