Feat: Support client JWT auth method for kafka oidc #4057
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Signed-off-by: Albert Callarisa <[email protected]>
acroca
force-pushed
the
client-assertion-oauth
branch
from
c23e275 to
9cd61ec
Compare
acroca
force-pushed
the
client-assertion-oauth
branch
2 times, most recently
from
2e0fe18 to
6cb1166
Compare
Signed-off-by: Albert Callarisa <[email protected]>
acroca
force-pushed
the
client-assertion-oauth
branch
from
6cb1166 to
742b291
Compare
acroca
changed the title
JoshVanL
requested changes
Oct 10, 2025
Comment on lines
+243
to
251
| if m.OidcClientAuthMethod == "client_secret" && m.OidcClientSecret == "" { | ||
| return nil, errors.New("kafka error: missing OIDC Client Secret for authType 'oidc' (client_secret)") | ||
| } | ||
| if m.OidcClientAuthMethod == "client_jwt" && m.OidcClientAssertionCert == "" { | ||
| return nil, errors.New("kafka error: missing OIDC Client Assertion Cert for authType 'oidc' (client_jwt)") | ||
| } | ||
| if m.OidcClientAuthMethod == "client_jwt" && m.OidcClientAssertionKey == "" { | ||
| return nil, errors.New("kafka error: missing OIDC Client Assertion Key for authType 'oidc' (client_jwt)") | ||
| } |
There was a problem hiding this comment.
Please can we use pointers to signal presence.
Comment on lines
+180
to
+196
| block, _ := pem.Decode([]byte(ts.ClientAssertionKey)) | ||
| if block == nil { | ||
| return nil, errors.New("invalid PEM private key for client assertion") | ||
| } | ||
| pk, err := x509.ParsePKCS8PrivateKey(block.Bytes) | ||
| if err != nil { | ||
| // try PKCS1 | ||
| if rsaKey, err2 := x509.ParsePKCS1PrivateKey(block.Bytes); err2 == nil { | ||
| pk = rsaKey | ||
| } else { | ||
| return nil, fmt.Errorf("unable to parse private key: %w", err) | ||
| } | ||
| } | ||
| rsaKey, ok := pk.(*rsa.PrivateKey) | ||
| if !ok { | ||
| return nil, errors.New("client_jwt requires RSA private key") | ||
| } |
| Subject(ts.ClientID). | ||
| Audience([]string{audClaim}). | ||
| IssuedAt(now). | ||
| Expiration(now.Add(1 * time.Minute)). |
| urlValues.Set("scope", strings.Join(ts.Scopes, " ")) | ||
| } | ||
|
|
||
| timeoutCtx, cancel := ctx.WithTimeout(ctx.TODO(), tokenRequestTimeout) |
Comment on lines
+240
to
+263
| defer cancel() | ||
| ts.configureClient() | ||
| req, err := http.NewRequestWithContext(timeoutCtx, http.MethodPost, ts.TokenEndpoint.TokenURL, strings.NewReader(urlValues.Encode())) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| req.Header.Set("Content-Type", "application/x-www-form-urlencoded") | ||
| resp, err := ts.httpClient.Do(req) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| defer resp.Body.Close() | ||
| if resp.StatusCode < 200 || resp.StatusCode >= 300 { | ||
| return nil, fmt.Errorf("token endpoint returned %d", resp.StatusCode) | ||
| } | ||
| var tr tokenResponse | ||
| if err := json.NewDecoder(resp.Body).Decode(&tr); err != nil { | ||
| return nil, err | ||
| } | ||
| if tr.AccessToken == "" { | ||
| return nil, errors.New("no access_token in response") | ||
| } | ||
| ts.CachedToken = oauth2.Token{AccessToken: tr.AccessToken, Expiry: time.Now().Add(time.Duration(tr.ExpiresIn) * time.Second)} | ||
| return ts.asSaramaToken(), nil |
There was a problem hiding this comment.
Is there not something that does all this for us in a library?
Comment on lines
+21
to
+34
| -----BEGIN CERTIFICATE----- | ||
| MIIEJTCCAg2gAwIBAgIUS0RIkoWlMVd8UfE/I7IEzKD4kmwwDQYJKoZIhvcNAQEL | ||
| BQAwGDEWMBQGA1UEAwwNbG9jYWwtb2lkYy1jYTAeFw0yNTEwMDMxMDI3NDlaFw0y | ||
| ODAxMDYxMDI3NDlaMBsxGTAXBgNVBAMMEGRhcHItb2lkYy1jbGllbnQwggEiMA0G | ||
| CSqGSIb3DQEBAQUAA4IBDwAwggEKAoIBAQDQIqGyGGlQRqXNDlSuHRDGmnJ4dQDb | ||
| jK/KxeC+wXQiDCSfHgP7NOjZOOCiQYKAUs9MRg+tKepxFRORZLAbRlU9kuU4Z7bg | ||
| E6wDMztCqTASUZtwZeN36iJDcQ++xnmAi8j0bvs7w00wrgPMXPnT09Sf9sQyEkR2 | ||
| t26OLcQnZHwrUWHIbZeXqvRF7uYG00GKjFmqI+FjrmDx0hoENew+Jzpk7S/7m5t2 | ||
| h/weVAr5REtGTIQxbX8nNmoO2JFbyILBcfP9M9R9zxiTEFVgP5sl2QfLFerPpULM | ||
| 1tAiCO+oPk8CAFiD9TKe+HR/Us4uIxgRmxjgBnzmFSJjDflvK2DslpzPAgMBAAGj | ||
| ZDBiMAsGA1UdDwQEAwIFoDATBgNVHSUEDDAKBggrBgEFBQcDAjAdBgNVHQ4EFgQU | ||
| oCgedGjEdPHOBRq6OwKKGUmpUXIwHwYDVR0jBBgwFoAUImUtm30c0E0Z1Xsep9o/ | ||
| nXsqyGYwDQYJKoZIhvcNAQELBQADggIBAKafBCz1EUny2pKOI8xTHVkoZjhTvtm2 | ||
| Ly2DicAEwtESXlBta64bOug4mTq8RLdKxLa8xSBSZxzygTfb+q8IcuuYL0/DAttZ |
There was a problem hiding this comment.
Please do not use hard coded certificates and keys in tests. They get copied and used in the real world, and they also expire.
Always generate them in the test, and use secret refs
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Ref: #4003
At the moment the
oidcauth type only supports authentication usingclientIDandclientSecret.This PR adds authentication using certificates instead. The component will use those certificates to build the client assertion and call the IdP to get a token.
This adds the following new fields in the kafka components:
oidcClientAuthMethodto select between the oldclient_secretmechanism, or the newclient_jwtmechanism.oidcClientAssertionCertoidcClientAssertionKeyoidcResourceoidcAudienceIn this PR I also added tests for both authentication mechanisms against a local kafka, using keycloak as an IdP.