From 7d7ed4c93b109a8040bde498c2fe04e1ecef1c60 Mon Sep 17 00:00:00 2001 From: Christian Bianchi Date: Thu, 23 Apr 2026 11:37:43 +0200 Subject: [PATCH 1/6] feat(catalog): add GCP Cloud KMS support for secret decryption Add a PEERDB_KMS_PROVIDER env var ("aws" default, "gcp") to select the KMS backend for catalog credential decryption. When set to "gcp", uses the GCP metadata server for Workload Identity access tokens and calls the Cloud KMS REST API to decrypt. The key ID is the raw GCP resource name (projects/.../cryptoKeys/...). AWS behavior is unchanged and remains the default. Co-Authored-By: Claude Opus 4.6 (1M context) --- nexus/catalog/Cargo.toml | 4 +- nexus/catalog/src/lib.rs | 82 ++++++++++++++++++++++++++++++++++++++-- nexus/server/src/main.rs | 14 +++++-- 3 files changed, 92 insertions(+), 8 deletions(-) diff --git a/nexus/catalog/Cargo.toml b/nexus/catalog/Cargo.toml index d21492dee..eeba71593 100644 --- a/nexus/catalog/Cargo.toml +++ b/nexus/catalog/Cargo.toml @@ -6,13 +6,15 @@ edition = "2024" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html [features] -default = ["aws"] +default = ["aws", "gcp"] aws = ["dep:aws-config", "dep:aws-sdk-kms"] +gcp = ["dep:reqwest"] [dependencies] anyhow = "1" aws-config = { version = "1", optional = true } aws-sdk-kms = { version = "1", optional = true } +reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"], optional = true } async-trait = "0.1" base64 = "0.22" chacha20poly1305 = { version = "0.11.0-rc.1", default-features = false, features = ["alloc"] } diff --git a/nexus/catalog/src/lib.rs b/nexus/catalog/src/lib.rs index 01cb4c912..76893eb30 100644 --- a/nexus/catalog/src/lib.rs +++ b/nexus/catalog/src/lib.rs @@ -32,10 +32,23 @@ mod embedded { pub struct Catalog { pg: Client, kms_key_id: Option>, + kms_provider: String, +} + +pub async fn kms_decrypt( + encrypted_payload: &str, + kms_key_id: &str, + kms_provider: &str, +) -> anyhow::Result { + match kms_provider { + "gcp" => gcp_kms_decrypt(encrypted_payload, kms_key_id).await, + "aws" | "" => aws_kms_decrypt(encrypted_payload, kms_key_id).await, + other => Err(anyhow!("unsupported KMS provider: {}", other)), + } } #[cfg(feature = "aws")] -pub async fn kms_decrypt(encrypted_payload: &str, kms_key_id: &str) -> anyhow::Result { +async fn aws_kms_decrypt(encrypted_payload: &str, kms_key_id: &str) -> anyhow::Result { let region_provider = RegionProviderChain::default_provider().or_else("us-east-1"); let config = aws_config::defaults(BehaviorVersion::v2026_01_12()) .region(region_provider) @@ -61,10 +74,71 @@ pub async fn kms_decrypt(encrypted_payload: &str, kms_key_id: &str) -> anyhow::R } #[cfg(not(feature = "aws"))] -pub async fn kms_decrypt(_encrypted_payload: &str, _kms_key_id: &str) -> anyhow::Result { +async fn aws_kms_decrypt(_encrypted_payload: &str, _kms_key_id: &str) -> anyhow::Result { Err(anyhow::anyhow!("AWS KMS support not compiled in")) } +#[cfg(feature = "gcp")] +async fn gcp_kms_decrypt(encrypted_payload: &str, kms_key_id: &str) -> anyhow::Result { + // kms_key_id is the full GCP resource name: + // projects/PROJECT/locations/LOCATION/keyRings/RING/cryptoKeys/KEY + + // Get access token from GKE Workload Identity metadata server + let token = reqwest::Client::new() + .get("http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token") + .header("Metadata-Flavor", "Google") + .send() + .await + .context("failed to get GCP access token from metadata server")? + .json::() + .await + .context("failed to parse GCP token response")?; + + let access_token = token["access_token"] + .as_str() + .ok_or_else(|| anyhow!("no access_token in metadata response"))?; + + // Call Cloud KMS decrypt API + let url = format!( + "https://cloudkms.googleapis.com/v1/{}:decrypt", + kms_key_id + ); + + let resp = reqwest::Client::new() + .post(&url) + .bearer_auth(access_token) + .json(&serde_json::json!({ "ciphertext": encrypted_payload })) + .send() + .await + .context("failed to call GCP KMS decrypt API")?; + + if !resp.status().is_success() { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + return Err(anyhow!("GCP KMS decrypt failed ({}): {}", status, body)); + } + + let body: serde_json::Value = resp + .json() + .await + .context("failed to parse GCP KMS decrypt response")?; + + let plaintext_b64 = body["plaintext"] + .as_str() + .ok_or_else(|| anyhow!("no plaintext in GCP KMS response"))?; + + let plaintext = BASE64_STANDARD + .decode(plaintext_b64) + .context("failed to decode GCP KMS plaintext")?; + + Ok(String::from_utf8(plaintext).context("GCP KMS plaintext is not valid UTF-8")?) +} + +#[cfg(not(feature = "gcp"))] +async fn gcp_kms_decrypt(_encrypted_payload: &str, _kms_key_id: &str) -> anyhow::Result { + Err(anyhow::anyhow!("GCP KMS support not compiled in")) +} + async fn run_migrations(client: &mut Client) -> anyhow::Result<()> { let migration_report = embedded::migrations::runner() .run_async(client) @@ -131,11 +205,13 @@ impl Catalog { pub async fn new( pt_config: pt::peerdb_peers::PostgresConfig, kms_key_id: &Option>, + kms_provider: &str, ) -> anyhow::Result { let (pg, _) = connect_postgres(&pt_config).await?; Ok(Self { pg, kms_key_id: kms_key_id.clone(), + kms_provider: kms_provider.to_string(), }) } @@ -147,7 +223,7 @@ impl Catalog { let mut enc_keys = env::var("PEERDB_ENC_KEYS")?; if let Some(kms_key_id) = &self.kms_key_id { - enc_keys = kms_decrypt(&enc_keys, kms_key_id.as_ref()).await?; + enc_keys = kms_decrypt(&enc_keys, kms_key_id.as_ref(), &self.kms_provider).await?; } // TODO use serde derive diff --git a/nexus/server/src/main.rs b/nexus/server/src/main.rs index 0869718ef..fdaf254d0 100644 --- a/nexus/server/src/main.rs +++ b/nexus/server/src/main.rs @@ -988,12 +988,16 @@ struct Args { /// KMS Key ID for decrypting the catalog password #[clap(long, env = "PEERDB_KMS_KEY_ID")] kms_key_id: Option>, + + /// KMS provider: "aws" (default) or "gcp" + #[clap(long, default_value = "aws", env = "PEERDB_KMS_PROVIDER")] + kms_provider: String, } // Get catalog config from args async fn get_catalog_config(args: &Args) -> anyhow::Result> { let password = if let Some(kms_key_id) = &args.kms_key_id { - kms_decrypt(&args.catalog_password, kms_key_id).await? + kms_decrypt(&args.catalog_password, kms_key_id, &args.kms_provider).await? } else { args.catalog_password.clone() }; @@ -1059,11 +1063,12 @@ fn setup_tracing(log_dir: Option<&str>) -> TracerGuards { async fn run_migrations( config: &CatalogConfig<'_>, kms_key_id: &Option>, + kms_provider: &str, ) -> anyhow::Result<()> { // retry connecting to the catalog 3 times with 30 seconds delay // if it fails, return an error for _ in 0..3 { - match Catalog::new(config.to_postgres_config(), kms_key_id).await { + match Catalog::new(config.to_postgres_config(), kms_key_id, kms_provider).await { Ok(mut catalog) => { catalog.run_migrations().await?; return Ok(()); @@ -1158,7 +1163,7 @@ pub async fn main() -> anyhow::Result<()> { } if !args.migrations_disabled { - run_migrations(&catalog_config, &args.kms_key_id).await?; + run_migrations(&catalog_config, &args.kms_key_id, &args.kms_provider).await?; } if args.migrations_only { return Ok(()); @@ -1194,10 +1199,11 @@ pub async fn main() -> anyhow::Result<()> { let authenticator = authenticator.clone(); let pg_config = catalog_config.to_postgres_config(); let kms_key_id = args.kms_key_id.clone(); + let kms_provider = args.kms_provider.clone(); let tls_acceptor = tls_acceptor.clone(); tokio::task::spawn(async move { - match Catalog::new(pg_config, &kms_key_id).await { + match Catalog::new(pg_config, &kms_key_id, &kms_provider).await { Ok(catalog) => { let nexus = Arc::new(NexusBackend::new( Arc::new(catalog), From 42ecc812b46c0fdc3c19fe834e0eba9aaa5009f6 Mon Sep 17 00:00:00 2001 From: Christian Bianchi Date: Thu, 23 Apr 2026 15:19:05 +0200 Subject: [PATCH 2/6] feat(flow): add GCP Cloud KMS support for Go flow components The Go flow worker, maintenance worker, and other flow components also decrypt catalog credentials via KMS. Add the same PEERDB_KMS_PROVIDER=gcp dispatch to the Go side, using the GCP metadata server for Workload Identity tokens and the Cloud KMS REST API for decryption. Co-Authored-By: Claude Opus 4.6 (1M context) --- flow/internal/env.go | 71 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 70 insertions(+), 1 deletion(-) diff --git a/flow/internal/env.go b/flow/internal/env.go index fc8e03814..1f3d91c1a 100644 --- a/flow/internal/env.go +++ b/flow/internal/env.go @@ -3,7 +3,10 @@ package internal import ( "context" "encoding/base64" + "encoding/json" "fmt" + "io" + "net/http" "os" "reflect" "strconv" @@ -17,7 +20,8 @@ import ( ) const ( - KmsKeyIDEnvVar = "PEERDB_KMS_KEY_ID" + KmsKeyIDEnvVar = "PEERDB_KMS_KEY_ID" + KmsProviderEnvVar = "PEERDB_KMS_PROVIDER" ) // getEnvUint returns the value of the environment variable with the given name @@ -71,6 +75,16 @@ func decryptWithKms(ctx context.Context, data []byte) ([]byte, error) { return data, nil } + provider := GetEnvString(KmsProviderEnvVar, "aws") + switch provider { + case "gcp": + return decryptWithGcpKms(ctx, data, keyID) + default: + return decryptWithAwsKms(ctx, data, keyID) + } +} + +func decryptWithAwsKms(ctx context.Context, data []byte, keyID string) ([]byte, error) { cfg, err := config.LoadDefaultConfig(ctx) if err != nil { return nil, fmt.Errorf("failed to load AWS config: %w", err) @@ -88,6 +102,61 @@ func decryptWithKms(ctx context.Context, data []byte) ([]byte, error) { return decrypted.Plaintext, nil } +func decryptWithGcpKms(ctx context.Context, data []byte, keyID string) ([]byte, error) { + // Get access token from GKE Workload Identity metadata server + req, err := http.NewRequestWithContext(ctx, http.MethodGet, + "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token", nil) + if err != nil { + return nil, fmt.Errorf("failed to create metadata request: %w", err) + } + req.Header.Set("Metadata-Flavor", "Google") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to get GCP access token: %w", err) + } + defer resp.Body.Close() + + var tokenResp struct { + AccessToken string `json:"access_token"` + } + if err := json.NewDecoder(resp.Body).Decode(&tokenResp); err != nil { + return nil, fmt.Errorf("failed to parse GCP token response: %w", err) + } + + // Call Cloud KMS decrypt API + ciphertext := base64.StdEncoding.EncodeToString(data) + body := fmt.Sprintf(`{"ciphertext":"%s"}`, ciphertext) + + kmsURL := fmt.Sprintf("https://cloudkms.googleapis.com/v1/%s:decrypt", keyID) + kmsReq, err := http.NewRequestWithContext(ctx, http.MethodPost, kmsURL, strings.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("failed to create KMS request: %w", err) + } + kmsReq.Header.Set("Authorization", "Bearer "+tokenResp.AccessToken) + kmsReq.Header.Set("Content-Type", "application/json") + + kmsResp, err := http.DefaultClient.Do(kmsReq) + if err != nil { + return nil, fmt.Errorf("failed to call GCP KMS: %w", err) + } + defer kmsResp.Body.Close() + + if kmsResp.StatusCode != http.StatusOK { + respBody, _ := io.ReadAll(kmsResp.Body) + return nil, fmt.Errorf("GCP KMS decrypt failed (%d): %s", kmsResp.StatusCode, string(respBody)) + } + + var kmsResult struct { + Plaintext string `json:"plaintext"` + } + if err := json.NewDecoder(kmsResp.Body).Decode(&kmsResult); err != nil { + return nil, fmt.Errorf("failed to parse GCP KMS response: %w", err) + } + + return base64.StdEncoding.DecodeString(kmsResult.Plaintext) +} + var kmsCache sync.Map func GetKmsDecryptedEnvBase64EncodedBytes(ctx context.Context, name string, defaultValue []byte) ([]byte, error) { From 6bf16ae5b006f7fe03ab40f01ed33d620f8322f4 Mon Sep 17 00:00:00 2001 From: Christian Bianchi Date: Mon, 27 Apr 2026 17:09:59 +0200 Subject: [PATCH 3/6] fix(catalog,flow): address clippy and golangci-lint warnings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Rust: remove needless Ok(x?) → x (clippy::needless_question_mark) - Go: use http.NoBody instead of nil request body (gocritic::httpNoBody) Co-Authored-By: Claude Opus 4.6 (1M context) --- flow/internal/env.go | 2 +- nexus/catalog/src/lib.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/flow/internal/env.go b/flow/internal/env.go index 1f3d91c1a..c8c484882 100644 --- a/flow/internal/env.go +++ b/flow/internal/env.go @@ -105,7 +105,7 @@ func decryptWithAwsKms(ctx context.Context, data []byte, keyID string) ([]byte, func decryptWithGcpKms(ctx context.Context, data []byte, keyID string) ([]byte, error) { // Get access token from GKE Workload Identity metadata server req, err := http.NewRequestWithContext(ctx, http.MethodGet, - "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token", nil) + "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token", http.NoBody) if err != nil { return nil, fmt.Errorf("failed to create metadata request: %w", err) } diff --git a/nexus/catalog/src/lib.rs b/nexus/catalog/src/lib.rs index 76893eb30..234dde387 100644 --- a/nexus/catalog/src/lib.rs +++ b/nexus/catalog/src/lib.rs @@ -131,7 +131,7 @@ async fn gcp_kms_decrypt(encrypted_payload: &str, kms_key_id: &str) -> anyhow::R .decode(plaintext_b64) .context("failed to decode GCP KMS plaintext")?; - Ok(String::from_utf8(plaintext).context("GCP KMS plaintext is not valid UTF-8")?) + String::from_utf8(plaintext).context("GCP KMS plaintext is not valid UTF-8") } #[cfg(not(feature = "gcp"))] From acf7c23aa111fe70a93cb7df6f0427599c706ddc Mon Sep 17 00:00:00 2001 From: Christian Bianchi Date: Mon, 27 Apr 2026 17:17:21 +0200 Subject: [PATCH 4/6] refactor(flow): use GCP KMS client library instead of raw HTTP Replace manual metadata token fetch + raw REST API calls with the official cloud.google.com/go/kms client library. The client handles authentication via Application Default Credentials (Workload Identity on GKE) automatically. This reduces ~50 lines of HTTP plumbing to ~15 lines of typed SDK calls and eliminates manual token management, JSON serialization, and error response parsing. Pinned cloud.google.com/go/kms at v1.24.0 to stay within the google.golang.org/api v0.257.0 constraint. Co-Authored-By: Claude Opus 4.6 (1M context) --- flow/go.mod | 5 ++++ flow/go.sum | 2 ++ flow/internal/env.go | 61 ++++++++------------------------------------ 3 files changed, 18 insertions(+), 50 deletions(-) diff --git a/flow/go.mod b/flow/go.mod index e4e57a7e6..b0b046c80 100644 --- a/flow/go.mod +++ b/flow/go.mod @@ -96,6 +96,11 @@ require ( require github.com/quasilyte/go-ruleguard/dsl v0.3.23 +require ( + cloud.google.com/go/kms v1.24.0 // PINNED: v1.26.0+ requires google.golang.org/api v0.265.0 + cloud.google.com/go/longrunning v0.7.0 // indirect +) + require ( cel.dev/expr v0.25.1 // indirect cloud.google.com/go/auth v0.18.1 diff --git a/flow/go.sum b/flow/go.sum index 17bcbd50a..5249bb675 100644 --- a/flow/go.sum +++ b/flow/go.sum @@ -15,6 +15,8 @@ cloud.google.com/go/datacatalog v1.26.1 h1:bCRKA8uSQN8wGW3Tw0gwko4E9a64GRmbW1nCb cloud.google.com/go/datacatalog v1.26.1/go.mod h1:2Qcq8vsHNxMDgjgadRFmFG47Y+uuIVsyEGUrlrKEdrg= cloud.google.com/go/iam v1.5.3 h1:+vMINPiDF2ognBJ97ABAYYwRgsaqxPbQDlMnbHMjolc= cloud.google.com/go/iam v1.5.3/go.mod h1:MR3v9oLkZCTlaqljW6Eb2d3HGDGK5/bDv93jhfISFvU= +cloud.google.com/go/kms v1.24.0 h1:SWltUuoPhTdv9q/P0YEAWQfoYT32O5HdfPgTiWMvrH8= +cloud.google.com/go/kms v1.24.0/go.mod h1:QDH3z2SJ50lfNOE8EokKC1G40i7I0f8xTMCoiptcb5g= cloud.google.com/go/logging v1.13.1 h1:O7LvmO0kGLaHY/gq8cV7T0dyp6zJhYAOtZPX4TF3QtY= cloud.google.com/go/logging v1.13.1/go.mod h1:XAQkfkMBxQRjQek96WLPNze7vsOmay9H5PqfsNYDqvw= cloud.google.com/go/longrunning v0.7.0 h1:FV0+SYF1RIj59gyoWDRi45GiYUMM3K1qO51qoboQT1E= diff --git a/flow/internal/env.go b/flow/internal/env.go index c8c484882..272fb44e7 100644 --- a/flow/internal/env.go +++ b/flow/internal/env.go @@ -3,16 +3,14 @@ package internal import ( "context" "encoding/base64" - "encoding/json" "fmt" - "io" - "net/http" "os" "reflect" "strconv" - "strings" "sync" + gcpkms "cloud.google.com/go/kms/apiv1" + gcpkmspb "cloud.google.com/go/kms/apiv1/kmspb" "github.com/aws/aws-sdk-go-v2/aws" "github.com/aws/aws-sdk-go-v2/config" "github.com/aws/aws-sdk-go-v2/service/kms" @@ -103,58 +101,21 @@ func decryptWithAwsKms(ctx context.Context, data []byte, keyID string) ([]byte, } func decryptWithGcpKms(ctx context.Context, data []byte, keyID string) ([]byte, error) { - // Get access token from GKE Workload Identity metadata server - req, err := http.NewRequestWithContext(ctx, http.MethodGet, - "http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token", http.NoBody) + client, err := gcpkms.NewKeyManagementClient(ctx) if err != nil { - return nil, fmt.Errorf("failed to create metadata request: %w", err) + return nil, fmt.Errorf("failed to create GCP KMS client: %w", err) } - req.Header.Set("Metadata-Flavor", "Google") + defer client.Close() - resp, err := http.DefaultClient.Do(req) - if err != nil { - return nil, fmt.Errorf("failed to get GCP access token: %w", err) - } - defer resp.Body.Close() - - var tokenResp struct { - AccessToken string `json:"access_token"` - } - if err := json.NewDecoder(resp.Body).Decode(&tokenResp); err != nil { - return nil, fmt.Errorf("failed to parse GCP token response: %w", err) - } - - // Call Cloud KMS decrypt API - ciphertext := base64.StdEncoding.EncodeToString(data) - body := fmt.Sprintf(`{"ciphertext":"%s"}`, ciphertext) - - kmsURL := fmt.Sprintf("https://cloudkms.googleapis.com/v1/%s:decrypt", keyID) - kmsReq, err := http.NewRequestWithContext(ctx, http.MethodPost, kmsURL, strings.NewReader(body)) - if err != nil { - return nil, fmt.Errorf("failed to create KMS request: %w", err) - } - kmsReq.Header.Set("Authorization", "Bearer "+tokenResp.AccessToken) - kmsReq.Header.Set("Content-Type", "application/json") - - kmsResp, err := http.DefaultClient.Do(kmsReq) + resp, err := client.Decrypt(ctx, &gcpkmspb.DecryptRequest{ + Name: keyID, + Ciphertext: data, + }) if err != nil { - return nil, fmt.Errorf("failed to call GCP KMS: %w", err) - } - defer kmsResp.Body.Close() - - if kmsResp.StatusCode != http.StatusOK { - respBody, _ := io.ReadAll(kmsResp.Body) - return nil, fmt.Errorf("GCP KMS decrypt failed (%d): %s", kmsResp.StatusCode, string(respBody)) - } - - var kmsResult struct { - Plaintext string `json:"plaintext"` - } - if err := json.NewDecoder(kmsResp.Body).Decode(&kmsResult); err != nil { - return nil, fmt.Errorf("failed to parse GCP KMS response: %w", err) + return nil, fmt.Errorf("failed to decrypt with GCP KMS: %w", err) } - return base64.StdEncoding.DecodeString(kmsResult.Plaintext) + return resp.Plaintext, nil } var kmsCache sync.Map From 018b7e6bac55d28d7612a537193479b8ec822115 Mon Sep 17 00:00:00 2001 From: Christian Bianchi Date: Mon, 27 Apr 2026 17:21:50 +0200 Subject: [PATCH 5/6] fix(flow): restore strings import and register KMS pin - Add back strings import (still used by strings.TrimSpace) - Register cloud.google.com/go/kms v1.24.0 in check_pinned_versions Co-Authored-By: Claude Opus 4.6 (1M context) --- flow/cmd/check_pinned_versions/main.go | 1 + flow/internal/env.go | 1 + 2 files changed, 2 insertions(+) diff --git a/flow/cmd/check_pinned_versions/main.go b/flow/cmd/check_pinned_versions/main.go index cf81912a3..3bfa57923 100644 --- a/flow/cmd/check_pinned_versions/main.go +++ b/flow/cmd/check_pinned_versions/main.go @@ -18,6 +18,7 @@ import ( // the Renovate package rules in renovate.json that skips automated updates for that dependency. var pinnedVersions = map[string]string{ "cloud.google.com/go/bigquery": "v1.72.0", + "cloud.google.com/go/kms": "v1.24.0", "cloud.google.com/go/pubsub/v2": "v2.3.0", "github.com/aws/aws-sdk-go-v2/feature/s3/manager": "v1.21.0", "google.golang.org/api": "v0.257.0", diff --git a/flow/internal/env.go b/flow/internal/env.go index 272fb44e7..0725b3515 100644 --- a/flow/internal/env.go +++ b/flow/internal/env.go @@ -7,6 +7,7 @@ import ( "os" "reflect" "strconv" + "strings" "sync" gcpkms "cloud.google.com/go/kms/apiv1" From 09d796d5c7e6de749d2126db8657d9d9826a1aa2 Mon Sep 17 00:00:00 2001 From: Christian Bianchi Date: Thu, 7 May 2026 13:49:00 +0200 Subject: [PATCH 6/6] address review: enumerate all KMS providers in switch, error on unknown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Per review on https://github.com/PeerDB-io/peerdb/pull/4243#discussion_r3201100678 — define KmsProviderAWS/KmsProviderGCP constants, list all cases (aws + empty for back-compat, gcp), default branch returns an error matching the Rust side in nexus/catalog/src/lib.rs:43-47. --- flow/internal/env.go | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/flow/internal/env.go b/flow/internal/env.go index 0725b3515..e81a76a6d 100644 --- a/flow/internal/env.go +++ b/flow/internal/env.go @@ -21,6 +21,9 @@ import ( const ( KmsKeyIDEnvVar = "PEERDB_KMS_KEY_ID" KmsProviderEnvVar = "PEERDB_KMS_PROVIDER" + + KmsProviderAWS = "aws" + KmsProviderGCP = "gcp" ) // getEnvUint returns the value of the environment variable with the given name @@ -74,12 +77,14 @@ func decryptWithKms(ctx context.Context, data []byte) ([]byte, error) { return data, nil } - provider := GetEnvString(KmsProviderEnvVar, "aws") + provider := GetEnvString(KmsProviderEnvVar, KmsProviderAWS) switch provider { - case "gcp": + case KmsProviderAWS, "": + return decryptWithAwsKms(ctx, data, keyID) + case KmsProviderGCP: return decryptWithGcpKms(ctx, data, keyID) default: - return decryptWithAwsKms(ctx, data, keyID) + return nil, fmt.Errorf("unsupported KMS provider %q (expected %q or %q)", provider, KmsProviderAWS, KmsProviderGCP) } }