I’m trying to set up grepplabs/kafka-proxy on GKE with Workload Identity to connect to GCP Managed Service for Apache Kafka (MSK) via SASL/OAUTHBEARER. I got SASL/PLAIN authentication to work with Access Token, but we want to use OAUTHBEARER to avoid static keys. Been trying all kinds of token but I keep getting “Access token is not a Google OAuth token” error.
These are the tokens that I tried passin onto this function:
-
ID Token portion of serviceaccount token from
http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity?audience=https://kafka.googleapis.com -
Access Token from obtained via google.FindDefaultCredentials() from within the pod (this works as password for SASL/PLAIN).
-
Token obtained from kafka-auth-local-server running as a sidecar
-
Token generated with this go code (tried to mimic the payload for GcpLoginCallbackHandler)
func getJWT(creds *google.Credentials) (string, error) {
// Parse token expiry and email from the credentials
token, _ := creds.TokenSource.Token()
email := os.Getenv("GCP_SA_NAME")
payload := map[string]interface{}{
"exp": time.Now().Add(time.Until(token.Expiry)).Unix(),
"iat": time.Now().UTC().Unix(),
"iss": "Google",
"sub": email,
"scope": "kafka",
}
payloadJSON, err := json.Marshal(payload)
if err != nil {
return "", err
}
headerJSON, _ := json.Marshal(headerPayload)
return strings.Join([]string{
b64Encode(string(headerJSON)),
b64Encode(string(payloadJSON)),
b64Encode(token.AccessToken),
}, "."), nil
}
Would greatly appreciate if someone can point me in the right direction. Thank you!