flowctl: automatically encrypt endpoint configs

Updates flowctl to always and automatically encrypt any unencrypted endpoint
configs.  There's no reason to ever use plain text credentials, so there's no
option to disable encryption. The encryption happens whenever we upsert to
`draft_specs` for any reason, which means that endpoint configs will be
encrypted whenever users run any of the following commands:
- `draft author`
- `catalog test`
- `catalog publish`

Encryption will be skipped for any tasks that don't have their connector as
part of `connector_tags`.  This is because we currently need to pull the
`endpoint_spec_schema` from the `connector_tags` table.  A future improvement
can allow flowctl to run the `spec` RPC to get the endpoint spec schemas of
connectors that don't have a `connector_tags` row.

The local endpoint configs are never updated as part of this. If users desire
to replace their local plain text configs with the encrypted ones, then it's
recommended that they run `draft author` followed by `draft develop` with
`--overwrite`.

Author: psFried

crates/dekaf/src/main.rs

@@ -13,6 +13,7 @@ use flow_client::{
     LOCAL_PG_URL,
 };
 use futures::TryStreamExt;
+use proto_flow::flow;
 use rustls::pki_types::CertificateDer;
 use std::{
     fs::File,
@@ -245,7 +246,13 @@ async fn main() -> anyhow::Result<()> {
         (cli.api_endpoint, cli.api_key)
     };
 
-    let client_base = flow_client::Client::new(cli.agent_endpoint, api_key, api_endpoint, None);
+    let client_base = flow_client::Client::new(
+        cli.agent_endpoint,
+        api_key,
+        api_endpoint,
+        None,
+        ::flow_client::DEFAULT_CONFIG_ENCRYPTION_URL.clone(),
+    );
     let signing_token = jsonwebtoken::EncodingKey::from_base64_secret(&cli.data_plane_access_key)?;
 
     let task_manager = Arc::new(TaskManager::new(

crates/flow-client/src/client.rs

@@ -21,6 +21,8 @@ pub struct Client {
     // a single connection pool. The clones can have different headers while
     // still re-using the same connection pool, so this will work across refreshes.
     pg_parent: postgrest::Postgrest,
+    // URL of the config encryption service.
+    config_encryption_url: url::Url,
 }
 
 impl Client {
@@ -30,6 +32,7 @@ impl Client {
         pg_api_token: String,
         pg_url: Url,
         user_access_token: Option<String>,
+        config_encryption_url: Url,
     ) -> Self {
         // Build journal and shard clients with an empty default service address.
         // We'll use their with_endpoint_and_metadata() routines to cheaply clone
@@ -56,6 +59,7 @@ impl Client {
             journal_client,
             shard_client,
             user_access_token,
+            config_encryption_url,
         }
     }
 
@@ -167,6 +171,49 @@ impl Client {
             anyhow::bail!("POST {path}: {status}: {body}");
         }
     }
+
+    /// Calls the endpoint configuration service to encrypt the given endpoint
+    /// configuration, using the provided `endpoint_spec_schema`.
+    pub async fn encrypt_endpoint_config(
+        &self,
+        plaintext_config: &models::RawValue,
+        endpoint_spec_schema: &models::RawValue,
+    ) -> anyhow::Result<models::RawValue> {
+        #[derive(serde::Serialize)]
+        struct EncryptRequest {
+            config: models::RawValue,
+            schema: models::RawValue,
+        }
+
+        let encrypt_endpoint = format!("{}v1/encrypt-config", self.config_encryption_url);
+
+        let request = EncryptRequest {
+            config: plaintext_config.clone(),
+            schema: endpoint_spec_schema.clone(),
+        };
+
+        // The encryption service does not currently require any sort of
+        // authentication, so there's no auth header added here. We'll of course
+        // need to update this if we ever add authentiation to that endpoint.
+        let mut req = self
+            .http_client
+            .post(&encrypt_endpoint)
+            .header("Content-Type", "application/json")
+            .json(&request);
+
+        let response = req.send().await?;
+
+        if !response.status().is_success() {
+            anyhow::bail!(
+                "Config encryption failed: {} {}",
+                response.status(),
+                response.text().await?
+            )
+        }
+
+        let encrypt_response: models::RawValue = response.json().await?;
+        Ok(encrypt_response)
+    }
 }
 
 #[derive(Debug, Clone, serde::Deserialize, serde::Serialize)]

crates/flow-client/src/lib.rs

@@ -57,11 +57,13 @@ lazy_static::lazy_static! {
     pub static ref DEFAULT_AGENT_URL:  url::Url = url::Url::parse("https://agent-api-1084703453822.us-central1.run.app").unwrap();
     pub static ref DEFAULT_DASHBOARD_URL: url::Url = url::Url::parse("https://dashboard.estuary.dev/").unwrap();
     pub static ref DEFAULT_PG_URL: url::Url = url::Url::parse("https://eyrcnmuzzyriypdajwdk.supabase.co/rest/v1").unwrap();
+    pub static ref DEFAULT_CONFIG_ENCRYPTION_URL: url::Url = url::Url::parse("https://config-encryptions.estuary.dev/").unwrap();
 
     // Used only when profile is "local".
     pub static ref LOCAL_AGENT_URL: url::Url = url::Url::parse("http://localhost:8675/").unwrap();
     pub static ref LOCAL_DASHBOARD_URL: url::Url = url::Url::parse("http://localhost:3000/").unwrap();
     pub static ref LOCAL_PG_URL: url::Url = url::Url::parse("http://localhost:5431/rest/v1").unwrap();
+    pub static ref LOCAL_CONFIG_ENCRYPTION_URL: url::Url = url::Url::parse("http://localhost:8765/").unwrap();
 }
 
 pub const DEFAULT_PG_PUBLIC_TOKEN: &str = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJzdXBhYmFzZSIsInJlZiI6ImV5cmNubXV6enlyaXlwZGFqd2RrIiwicm9sZSI6ImFub24iLCJpYXQiOjE2NDg3NTA1NzksImV4cCI6MTk2NDMyNjU3OX0.y1OyXD3-DYMz10eGxzo1eeamVMMUwIIeOoMryTRAoco";

crates/flowctl/src/catalog/publish.rs

@@ -26,13 +26,13 @@ pub async fn do_publish(ctx: &mut CliContext, args: &Publish) -> anyhow::Result<
     // since that can fail due to missing/expired credentials.
     anyhow::ensure!(args.auto_approve || std::io::stdin().is_tty(), "The publish command must be run interactively unless the `--auto-approve` flag is provided");
 
-    let (draft_catalog, _validations) =
+    let (mut draft_catalog, _validations) =
         local_specs::load_and_validate(&ctx.client, &args.source).await?;
 
     let draft = draft::create_draft(&ctx.client).await?;
     println!("Created draft: {}", &draft.id);
     tracing::info!(draft_id = %draft.id, "created draft");
-    draft::upsert_draft_specs(&ctx.client, draft.id, &draft_catalog).await?;
+    draft::author(&ctx.client, draft.id, &mut draft_catalog).await?;
 
     let removed = draft::remove_unchanged(&ctx.client, draft.id).await?;
     if !removed.is_empty() {

crates/flowctl/src/catalog/test.rs

@@ -16,13 +16,13 @@ pub struct TestArgs {
 /// and discoverable to users. There's also no need for any confirmation steps, since we're not
 /// actually modifying the published specs.
 pub async fn do_test(ctx: &mut CliContext, args: &TestArgs) -> anyhow::Result<()> {
-    let (draft_catalog, _validations) =
+    let (mut draft_catalog, _validations) =
         local_specs::load_and_validate(&ctx.client, &args.source).await?;
 
     let draft = draft::create_draft(&ctx.client).await?;
     println!("Created draft: {}", &draft.id);
     tracing::info!(draft_id = %draft.id, "created draft");
-    let spec_rows = draft::upsert_draft_specs(&ctx.client, draft.id, &draft_catalog).await?;
+    let spec_rows = draft::author(&ctx.client, draft.id, &mut draft_catalog).await?;
     println!("Running tests for catalog items:");
     ctx.write_all(spec_rows, ())?;
     println!("Starting tests...");

crates/flowctl/src/config.rs

@@ -2,8 +2,9 @@ use anyhow::Context;
 use std::path::PathBuf;
 
 use flow_client::{
-    client::RefreshToken, DEFAULT_AGENT_URL, DEFAULT_DASHBOARD_URL, DEFAULT_PG_PUBLIC_TOKEN,
-    DEFAULT_PG_URL, LOCAL_AGENT_URL, LOCAL_DASHBOARD_URL, LOCAL_PG_PUBLIC_TOKEN, LOCAL_PG_URL,
+    client::RefreshToken, DEFAULT_AGENT_URL, DEFAULT_CONFIG_ENCRYPTION_URL, DEFAULT_DASHBOARD_URL,
+    DEFAULT_PG_PUBLIC_TOKEN, DEFAULT_PG_URL, LOCAL_AGENT_URL, LOCAL_CONFIG_ENCRYPTION_URL,
+    LOCAL_DASHBOARD_URL, LOCAL_PG_PUBLIC_TOKEN, LOCAL_PG_URL,
 };
 
 /// Configuration of `flowctl`.
@@ -36,6 +37,9 @@ pub struct Config {
     /// used to generate access_token when it's unset or expires.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub user_refresh_token: Option<RefreshToken>,
+    /// URL endpoint for the config encryption service.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub config_encryption_url: Option<url::Url>,
 
     #[serde(skip)]
     is_local: bool,
@@ -101,6 +105,16 @@ impl Config {
         }
     }
 
+    pub fn get_config_encryption_url(&self) -> &url::Url {
+        if let Some(config_encryption_url) = &self.config_encryption_url {
+            config_encryption_url
+        } else if self.is_local {
+            &LOCAL_CONFIG_ENCRYPTION_URL
+        } else {
+            &DEFAULT_CONFIG_ENCRYPTION_URL
+        }
+    }
+
     /// Loads the config corresponding to the given named `profile`.
     /// This loads from:
     /// - $HOME/.config/flowctl/${profile}.json on linux
@@ -181,12 +195,14 @@ impl Config {
 
         Ok(())
     }
+
     pub fn build_anon_client(&self) -> flow_client::Client {
         flow_client::Client::new(
             self.get_agent_url().clone(),
             self.get_pg_public_token().to_string(),
             self.get_pg_url().clone(),
             None,
+            self.get_config_encryption_url().clone(),
         )
     }
 

crates/flowctl/src/draft/author.rs

@@ -1,4 +1,4 @@
-use crate::{api_exec, catalog::SpecSummaryItem, local_specs};
+use crate::{api_exec, catalog::SpecSummaryItem, draft::encrypt, local_specs};
 use anyhow::Context;
 use futures::{stream::FuturesOrdered, StreamExt};
 use serde::Serialize;
@@ -24,7 +24,18 @@ pub async fn clear_draft(client: &crate::Client, draft_id: models::Id) -> anyhow
     Ok(())
 }
 
-pub async fn upsert_draft_specs(
+/// Encrypts any unencrypted endpoint configurations in the draft catalog,
+/// and then upserts the draft specs to the given draft ID.
+pub async fn author(
+    client: &crate::Client,
+    draft_id: models::Id,
+    draft: &mut tables::DraftCatalog,
+) -> anyhow::Result<Vec<SpecSummaryItem>> {
+    encrypt::encrypt_endpoint_configs(draft, client).await?;
+    upsert_draft_specs(client, draft_id, &*draft).await
+}
+
+async fn upsert_draft_specs(
     client: &crate::Client,
     draft_id: models::Id,
     draft: &tables::DraftCatalog,
@@ -131,10 +142,10 @@ pub async fn do_author(
     Author { source }: &Author,
 ) -> anyhow::Result<()> {
     let draft_id = ctx.config.selected_draft()?;
-    let (draft, _) = local_specs::load_and_validate(&ctx.client, &source).await?;
+    let (mut draft, _) = local_specs::load_and_validate(&ctx.client, &source).await?;
 
     clear_draft(&ctx.client, draft_id).await?;
-    let rows = upsert_draft_specs(&ctx.client, draft_id, &draft).await?;
+    let rows = author(&ctx.client, draft_id, &mut draft).await?;
 
     ctx.write_all(rows, ())
 }