[AUTHN-1982] Implement OAuthBearer mechanism for Kafka client to fetch JWT token by communicating with SPIRE agent #1015

chang-you · 2023-06-23T18:52:41Z

What is the purpose of the change

The goal of these changes is to write a test that integrates the Kafka client with a SPIRE server that runs in your local machine to fetch JWT token via the workload API.

Create handleJWTTokenRefreshEvent function to handle token refresh events
Create retrieveJWTToken function to fetch JWT from SPIRE agent
Import SPIFFE Workload API
Integrate OAuthBearer mechanism into Kafka Producer to enable secure communication

References

JIRA: https://confluentinc.atlassian.net/browse/AUTHN-1982

What will this work accomplish/help with?

This example test will serve as a guide for integrating Confluent’s Kafka Golang client with SPIRE to mint SPIFFE SVID’s in the form of JWT’s, this will be the client-side changes needed, we will have a separate ticket to implement the server-side changes where Kafka will be able to validate these credentials with SPIRE via the SPIRE agent/the SPIFFE workload API.

Verify the change

The test should run successfully against an actual Kafka cluster in that the followings are verified:
1- Authentication with a Kafka cluster works successfully.
2- Token refresh events are handled properly by fetching a new token every time the refresh event is fired.

Producer + Consumer:

Documentation

To run the SPIRE server on kubernetes, either remote or on minikube, follow this guide

…er and identityPollId as extentions to oauthBearerToken

milindl

Thanks for this PR @chang-you . I gave it a first-pass review, it's a good example to add.

Apart from the comments on the code, there are a few other things:

Is it possible to add a README into the folder (like inside examples/spire_consumer_example/README). It can talk about:
a. how custom OAuth handlers are to be implemented,
b. how to run this example locally.
c. make it more clear that the client doesn't provide support for SPIRE/SPIFFE, but rather, that this is supposed to be an example implementation of a custom OAuth handler.
I can help with writing this README if you prefer, especially part a. and c.
Is there an open guide on how to set up a SPIRE server locally? As the linked guide is inside Confluent's internal page, and external users of the library can't access it.
Add the binary to the examples/.gitignore and the example link to examples/README. Similar to other examples. And add an entry into the CHANGELOG.md, too.

milindl · 2023-08-25T06:26:32Z

examples/spire_consumer_example/spire_consumer_example.go

+				// But in this example we choose to terminate
+				// the application if all brokers are down.
+				fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n", e.Code(), e)
+				if e.Code() == kafka.ErrAllBrokersDown {


This if block isn't needed, you can remove the comment too, and replace with just

// Errors should generally be considered // informational, the client will try to // automatically recover. fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n", e.Code(), e)

milindl · 2023-08-25T06:27:11Z

examples/spire_consumer_example/spire_consumer_example.go

+		"group.id":                 "myGroup",
+		"session.timeout.ms":       6000,
+		"auto.offset.reset":        "earliest",
+		"enable.auto.offset.store": false,


Remove this config, as it's not relevant to this example and will complicate things

milindl · 2023-08-25T06:28:10Z

examples/spire_consumer_example/spire_consumer_example.go

+				if e.Headers != nil {
+					fmt.Printf("%% Headers: %v\n", e.Headers)
+				}
+				_, err := c.StoreMessage(e)


Remove the StoreMessage and the subsequent error check, as it's not needed after removing "enable.auto.offset.store"

milindl · 2023-08-25T06:34:00Z

examples/spire_producer.example/spire_producer.example.go

regarding this file: Would it make sense to keep just one file (either the consumer or the producer, but not both)?
Since this is an example to demonstrate the usage of a custom oauth handler, what we're actually doing with the client shouldn't be the main focus of the example, what do you think?

Thanks for the comments! That makes sense too.
@arvindth what do you think? Thank you!

I agree, a single producer or consumer example is probably good enough.

@milindl @arvindth but I noticed that under the examples folder, most of the previous examples: protobuf/oauthbearer/json/avro include both producer example and consumer example, should we keep both?

milindl · 2023-08-25T06:48:39Z

examples/spire_consumer_example/spire_consumer_example.go

+	}
+}
+
+func retrieveJWTToken(ctx context.Context, principal, socketPath string, audience []string) (kafka.OAuthBearerToken, func() error, error) {


Add a comment for this method.

We don't need to describe the exact logic that the method uses (in fact, it's probably a fair idea to say that we can use any arbitrary logic, like here we are using the go-spiffe library as long as we construct a kafka.OAuthBearerToken token correctly).

Thanks for the comments. To clarify:

We don't need to describe the exact logic that the method use

Here do you suggest that we don't need to include a description of this method in the readme, or is there something specific I should modify within this method?

arvindth · 2023-08-31T22:03:37Z

examples/spire_consumer_example/spire_consumer_example.go

+	"time"
+)
+
+// handleJWTTokenRefreshEvent retrieves JWT from the SPIRE workload API and


Suggested change

// handleJWTTokenRefreshEvent retrieves JWT from the SPIRE workload API and

// handleJWTTokenRefreshEvent retrieves JWT from the SPIFFE workload API and

for correct terminology.

arvindth · 2023-08-31T22:04:22Z

examples/spire_producer.example/spire_producer.example.go

+	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+)
+
+// handleJWTTokenRefreshEvent retrieves JWT from the SPIRE workload API and


Suggested change

// handleJWTTokenRefreshEvent retrieves JWT from the SPIRE workload API and

// handleJWTTokenRefreshEvent retrieves JWT from the SPIFFE workload API and

arvindth · 2023-09-13T18:45:50Z

@chang-you what's the status of this PR?

Changes made.

arvindth · 2023-11-28T02:33:53Z

examples/spire_consumer_example/spire_consumer_example.go

+// It must be invoked whenever kafka.OAuthBearerTokenRefresh appears on the client's event channel,
+// which will occur whenever the client requires a token (i.e. when it first starts and when the
+// previously-received token is 80% of the way to its expiration time).
+func handleJWTTokenRefreshEvent(ctx context.Context, client kafka.Handle, principal, socketPath string, audience []string) {


@chang-you I don't think we're using the principal here correctly. In the oauth producer case, the principal needs to be passed in because it's the principal for which we get a token, and the token would then contain the principal in the sub claim.

For spire though, the principal value should be a spiffe id, and we should be using this spiffe id to look through all the svids returned by the spire agent to see if one of them matches the given principal, and use that one.

@arvindth Do you suggest we should use FetchJWTSVIDs instead of FetchJWTSVID and use the JWTSVID with its spiffeid.ID that match the sub?

arvindth · 2023-11-28T02:34:16Z

examples/spire_consumer_example/spire_consumer_example.go

+		"bootstrap.servers":       bootstrapServers,
+		"security.protocol":       "SASL_SSL",
+		"sasl.mechanisms":         "OAUTHBEARER",
+		"sasl.oauthbearer.config": principal,


I'm not sure if this parameter is even required. Could we remove it and see if it works as expected?

jshahc · 2023-12-19T04:03:41Z

examples/spire_consumer_example/spire_consumer_example.go

+		Extensions: extensions,
+	}
+
+	return oauthBearerToken, jwtSource.Close, nil


Doesn't seem necessary to return jwtSource.Close, as defer jwtSource.Close() is called above

chang-you requested a review from moe-omar June 23, 2023 18:55

chang-you marked this pull request as draft June 24, 2023 06:40

chang-you marked this pull request as ready for review June 24, 2023 06:41

chang-you changed the title ~~Implement OAuthBearer mechanism for creating Kafka Producer by communicating with SPIRE agent~~ Implement OAuthBearer mechanism for Kafka client to fetch JWT token by communicating with SPIRE agent Jun 30, 2023

milindl self-requested a review July 3, 2023 10:26

chang-you added 4 commits July 7, 2023 14:46

Created handleJWTTokenRefreshEvent and retrieveJWTToken functions

f3f7a55

Updated closer error and changed oauthConf to principal

1fdc6b0

Modified config to use the defualt unsecure method, pass logicalClust…

d1f1fbd

…er and identityPollId as extentions to oauthBearerToken

Deleted imported workload API, clean the code

87ba39f

chang-you force-pushed the spire_producer_example_0623 branch from 49fe984 to 87ba39f Compare July 7, 2023 21:59

chang-you added 4 commits July 7, 2023 15:04

Deleted Spire workload API

1f69a9e

Updated go.mod

9e3286f

Updated Context, added spire_consumer_example.go

ff33ee2

Updated spire_consumer_example.go and comment description.

728a155

chang-you changed the title ~~Implement OAuthBearer mechanism for Kafka client to fetch JWT token by communicating with SPIRE agent~~ [AUTHN-1982] Implement OAuthBearer mechanism for Kafka client to fetch JWT token by communicating with SPIRE agent Jul 20, 2023

milindl reviewed Aug 25, 2023

View reviewed changes

arvindth previously requested changes Aug 31, 2023

View reviewed changes

chang-you added 3 commits September 13, 2023 13:01

remove some unnecessary blocks

92056a9

Empty-Commit

decdb12

update CHANGELOG.md and examples/.gitignore

89573cd

arvindth reviewed Nov 28, 2023

View reviewed changes

jshahc reviewed Dec 19, 2023

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[AUTHN-1982] Implement OAuthBearer mechanism for Kafka client to fetch JWT token by communicating with SPIRE agent #1015

[AUTHN-1982] Implement OAuthBearer mechanism for Kafka client to fetch JWT token by communicating with SPIRE agent #1015

chang-you commented Jun 23, 2023 •

edited by arvindth

milindl left a comment

milindl Aug 25, 2023

milindl Aug 25, 2023

milindl Aug 25, 2023

milindl Aug 25, 2023

chang-you Sep 13, 2023

arvindth Sep 13, 2023

chang-you Oct 10, 2023 •

edited

milindl Aug 25, 2023

chang-you Sep 13, 2023

arvindth Aug 31, 2023

arvindth Aug 31, 2023

arvindth commented Sep 13, 2023

arvindth Nov 28, 2023

chang-you Dec 1, 2023 •

edited

arvindth Nov 28, 2023

jshahc Dec 19, 2023

chang-you Dec 19, 2023

	// handleJWTTokenRefreshEvent retrieves JWT from the SPIRE workload API and
	// handleJWTTokenRefreshEvent retrieves JWT from the SPIFFE workload API and

[AUTHN-1982] Implement OAuthBearer mechanism for Kafka client to fetch JWT token by communicating with SPIRE agent #1015

Are you sure you want to change the base?

[AUTHN-1982] Implement OAuthBearer mechanism for Kafka client to fetch JWT token by communicating with SPIRE agent #1015

Conversation

chang-you commented Jun 23, 2023 • edited by arvindth

What is the purpose of the change

References

What will this work accomplish/help with?

Verify the change

Documentation

milindl left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chang-you Oct 10, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arvindth commented Sep 13, 2023

Choose a reason for hiding this comment

chang-you Dec 1, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chang-you commented Jun 23, 2023 •

edited by arvindth

chang-you Oct 10, 2023 •

edited

chang-you Dec 1, 2023 •

edited