feat(catalog): add GCP Cloud KMS support for secret decryption (#4243)

## Summary
- Add `PEERDB_KMS_PROVIDER` env var to dispatch between AWS KMS and GCP
Cloud KMS for catalog secret decryption
- **Rust** (`nexus/catalog`): GCP Cloud KMS REST API via `reqwest`,
using a Workload Identity token fetched from the GKE metadata server
- **Go** (`flow/internal/env.go`): GCP Cloud KMS via the official
`cloud.google.com/go/kms/apiv1` client (uses Application Default
Credentials, which resolves to Workload Identity on GKE)
- Provider switch on both sides enumerates `aws` (and empty for
back-compat), `gcp`, and returns an explicit error for unknown values;
constants `KmsProviderAWS` / `KmsProviderGCP` introduced in Go to mirror
the Rust pattern
- Defaults to `aws` everywhere — no behavior change for existing AWS
deployments

## Test plan
- [ ] End-to-end test on GCP GKE cluster with Workload Identity,
decrypting a catalog secret encrypted with a GCP KMS key
- [ ] Verify `PEERDB_KMS_PROVIDER=` (empty), unset, and
`PEERDB_KMS_PROVIDER=aws` all still hit the AWS path
- [ ] Verify `PEERDB_KMS_PROVIDER=invalid` returns the new error rather
than silently falling through to AWS

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: whites11

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",

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

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=

flow/internal/env.go

@@ -10,14 +10,20 @@ import (
 	"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"
 	"golang.org/x/exp/constraints"
 )
 
 const (
-	KmsKeyIDEnvVar = "PEERDB_KMS_KEY_ID"
+	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
@@ -71,6 +77,18 @@ func decryptWithKms(ctx context.Context, data []byte) ([]byte, error) {
 		return data, nil
 	}
 
+	provider := GetEnvString(KmsProviderEnvVar, KmsProviderAWS)
+	switch provider {
+	case KmsProviderAWS, "":
+		return decryptWithAwsKms(ctx, data, keyID)
+	case KmsProviderGCP:
+		return decryptWithGcpKms(ctx, data, keyID)
+	default:
+		return nil, fmt.Errorf("unsupported KMS provider %q (expected %q or %q)", provider, KmsProviderAWS, KmsProviderGCP)
+	}
+}
+
+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 +106,24 @@ func decryptWithKms(ctx context.Context, data []byte) ([]byte, error) {
 	return decrypted.Plaintext, nil
 }
 
+func decryptWithGcpKms(ctx context.Context, data []byte, keyID string) ([]byte, error) {
+	client, err := gcpkms.NewKeyManagementClient(ctx)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create GCP KMS client: %w", err)
+	}
+	defer client.Close()
+
+	resp, err := client.Decrypt(ctx, &gcpkmspb.DecryptRequest{
+		Name:       keyID,
+		Ciphertext: data,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to decrypt with GCP KMS: %w", err)
+	}
+
+	return resp.Plaintext, nil
+}
+
 var kmsCache sync.Map
 
 func GetKmsDecryptedEnvBase64EncodedBytes(ctx context.Context, name string, defaultValue []byte) ([]byte, error) {

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"] }

nexus/catalog/src/lib.rs

@@ -32,10 +32,23 @@ mod embedded {
 pub struct Catalog {
     pg: Client,
     kms_key_id: Option<Arc<String>>,
+    kms_provider: String,
+}
+
+pub async fn kms_decrypt(
+    encrypted_payload: &str,
+    kms_key_id: &str,
+    kms_provider: &str,
+) -> anyhow::Result<String> {
+    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<String> {
+async fn aws_kms_decrypt(encrypted_payload: &str, kms_key_id: &str) -> anyhow::Result<String> {
     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<String> {
+async fn aws_kms_decrypt(_encrypted_payload: &str, _kms_key_id: &str) -> anyhow::Result<String> {
     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<String> {
+    // 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::<serde_json::Value>()
+        .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")?;
+
+    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<String> {
+    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<Arc<String>>,
+        kms_provider: &str,
     ) -> anyhow::Result<Self> {
         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

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<Arc<String>>,
+
+    /// 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<CatalogConfig<'_>> {
     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<Arc<String>>,
+    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),