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) <noreply@anthropic.com>

Author: whites11

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

@@ -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