Is it possible connect to GCP Managed Service for Apache Kafka (MSK) cluster with SASL/OAUTHBEARER?

[mhan38]

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?audien...

• 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!