Add OAUTHBEARER/OIDC support

[lpsinger]

Add OAUTHBEARER / OpenID Connect (OIDC) support as described in KIP-768. This is the authentication mechanism used by the GCN/TACH Kafka broker.

• The SASLMethod enum gets a new member, OAUTHBEARER.
• The SASLAuth constructor gets a new optional keyword argument, token_endpoint.
• The presence of the token_endpoint keyword argument causes the default SASL method to change to OAUTHBEARER and the oauthbearer method to change to oidc, because this option is required for OpenID Connect and ignored for all other auth methods.
• In OIDC mode, the username and password positional arguments are interpreted as the client ID and client secret.

[cnweaver]

I'm having a bit of a tough time testing this. I've been able to set up a local broker which allows unsecured OAUTHBEARER mechanism, but this patch and the new OIDC method appears not to actually be compatible with that, and I'm not sure what it will take to run a broker with full OIDC.

Aside from that, I had a lot of trouble with getting a new enough version of librdkafka to support this. As far as I can tell, this is not yet in the latest release (1.8.2). I got it working (to the point that my client can try to use it) by building the current trunk of librdkafka, but we certainly can't expect hopskotch users (or other consumers of adc-streaming if there are any) to do that typically. Am I missing something here?

[lpsinger]

Correct. These features are in Apache Kafka 3.1.0 (recently released) but they are still just in master for librdkafka and should debut in version 1.9.0 which will be released any day now. I agree with you, and I don't advocate pushing a release of adc-streaming with this in it until it's fully supported upstream in releases of librdkafka and confluent-kaka-python. But hopefully you're OK with getting most of the code review done in advance?

For the very near term, you can test using development wheels that I have built:

pip install --extra-index-url https://asd.gsfc.nasa.gov/Leo.Singer/pypi confluent-kafka==1.8.3+bleeding.edge

[cnweaver]

Okay, I agree that we can definitely do the review before that release happens. In, fact, if we can check librdkafka's builtin.features, we could detect at runtime whether this is supported and give a nice error message (as you're likely aware, the current failure mode is a bit obscure), which I think could allow us to merge early.

I have to build the dependencies from source anyway, since the wheels are broken on older Mac OS versions (the widely observed Symbol not found: ____chkstk_darwin issue).

I will try setting up full OIDC auth in Kafka, since I would really like to do the due-diligence of running this successfully myself, but if push comes to shove if you've been able to run it against some broker, that shouldn't hold up accepting the PR.

[lpsinger]

I will try setting up full OIDC auth in Kafka, since I would really like to do the due-diligence of running this successfully myself, but if push comes to shove if you've been able to run it against some broker, that shouldn't hold up accepting the PR.

librdkafka uses Magnus Edenhill's own trivup for bringing up test Kafka brokers. I think it already has OIDC support.

I certainly have been able to run this against our broker. To test it, you are going to need not just a Kafka broker but also an OIDC provider.

[cnweaver]

SCiMMA already has an OIDC provider (COmanage) through which I can provision clients as needed, so that's not an obstacle. However, it does look like trivup has support, so if nothing else I may be able to learn from it how to do the configuration correctly.

[lpsinger]

SCiMMA already has an OIDC provider (COmanage) through which I can provision clients as needed, so that's not an obstacle. However, it does look like trivup has support, so if nothing else I may be able to learn from it how to do the configuration correctly.

Here are the relevant parts of our Kafka broker configuration. (Note that we are using SASL_PLAINTEXT rather than SASL_SSL because we have a load balancer external to Kafka doing the TLS termination.)

# Server Basics
broker.id=${MyId}

# Socket Server Settings
listeners=INTERNAL://:${InternalBrokerPort},EXTERNAL://:${ExternalBrokerPort0}
listener.security.protocol.map=INTERNAL:PLAINTEXT,EXTERNAL:SASL_PLAINTEXT
inter.broker.listener.name=INTERNAL
advertised.listeners=INTERNAL://:${InternalBrokerPort},EXTERNAL://${LoadBalancer}:${ExternalBrokerPort}

# Authentication
sasl.enabled.mechanisms=OAUTHBEARER
listener.name.external.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.secured.OAuthBearerValidatorCallbackHandler
listener.name.external.oauthbearer.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;
sasl.oauthbearer.jwks.endpoint.url=https://${IdentityProviderDomain}/.well-known/jwks.json

[cnweaver]

I've been able to test that this (in combination with the hop-client patch) works for me. If librdkafka's builtin.features is exposed to pyton in any way, I can't find it, which is a shame, but we can live without the feature check.

[lpsinger]

I've been able to test that this (in combination with the hop-client patch) works for me. If librdkafka's builtin.features is exposed to pyton in any way, I can't find it, which is a shame, but we can live without the feature check.

There isn't, but we can check confluent_kafka.libversion(). I think the check should occur right before connection, not at configuration. Do you agree?

[cnweaver]

Indeed, there's no reason to complain if the user doesn't happen to touch on this feature by actually requesting to make a connection using it.

[lpsinger]

Indeed, there's no reason to complain if the user doesn't happen to touch on this feature by actually requesting to make a connection using it.

OK, done. Please review.

[cnweaver]

The new error message works well for me, and I like that we can extend the code which produces it to cover other features in the future if we need to.

[lpsinger]

@cnweaver, since upstream is taking longer than I expected to do a release of librdkafka, I wrote a pure Python implementation of the OIDC auth that will work with the current stable release of confluent-kafka-python. Please take a look.

[cnweaver]

I haven't had a chance to try running this, but it looks simple enough and makes sense to me. Let me check with @mjuric , but it seems like we should be able to merge this without wiating for librdkafka.

[lpsinger]

Thank you! When do you anticipate doing a release?

[cnweaver]

This morning was my plan.

[lpsinger]

Hold off a bit please. This was working yesterday but I am having some problems with it.

[cnweaver]

I already made https://github.com/astronomy-commons/adc-streaming/releases/tag/v2.1.0 , but I do not have the capability to publish to PyPI or Conda, so it isn't yet on those. If necessary we can make a 2.1.1 patch release. I did have to add a fix yesterday; if what you're seeing is a KeyError when not using SASL.OAUTHBEARER that would be taken care of.

[lpsinger]

No, I'm hitting some obscure bug in librdkafka: confluentinc/librdkafka#3263 (comment).

I think that the reason that I missed this is that I tested the callback itself but not the Consumer.consume() method, assuming foolishly that oauth_cb would just work.

[lpsinger]

Code:

#!/usr/bin/env python
import logging
logging.basicConfig(level=logging.DEBUG)

from confluent_kafka import Consumer, Producer
from uuid import uuid4
from pprint import pprint


def set_oauth_cb(config):
    """Implement client support for KIP-768 OpenID Connect.

    Apache Kafka 3.1.0 supports authentication using OpenID Client Credentials.
    Native support for Python is coming in the next release of librdkafka
    (version 1.9.0). Meanwhile, this is a pure Python implementation of the
    refresh token callback.
    """
    if config.pop('sasl.oauthbearer.method', None) != 'oidc':
        return

    client_id = config.pop('sasl.oauthbearer.client.id')
    client_secret = config.pop('sasl.oauthbearer.client.secret')
    scope = config.pop('sasl.oauthbearer.scope', None)
    token_endpoint = config.pop('sasl.oauthbearer.token.endpoint.url')

    from authlib.integrations.requests_client import OAuth2Session
    session = OAuth2Session(client_id, client_secret, scope=scope)
    
    def oauth_cb(*_, **__):
        token = session.fetch_token(
            token_endpoint, grant_type='client_credentials')
        return token['access_token'], token['expires_at']

    config['oauth_cb'] = oauth_cb


# Fill in client credentials here
config = {
    'bootstrap.servers': '...',
    'security.protocol': 'sasl_ssl',
    'sasl.mechanisms': 'OAUTHBEARER',
    'sasl.oauthbearer.method': 'oidc',
    'sasl.oauthbearer.client.id': '...',
    'sasl.oauthbearer.client.secret': '...',
    'sasl.oauthbearer.token.endpoint.url': '...',
    'group.id': str(uuid4()),
    'log_level': 7
}

set_oauth_cb(config)

consumer = Consumer(config)
consumer.subscribe(['foobar'])
for message in consumer.consume():
    print(message.value())

Output:

*** rdkafka.c:3861:rd_kafka_poll_cb: assert: !*"cant handle op type" ***
rd_kafka_t 0x7ff613055c00: rdkafka#consumer-1
 producer.msg_cnt 0 (0 bytes)
 rk_rep reply queue: 0 ops
 brokers:
 rd_kafka_broker_t 0x7ff612098c00: :0/internal NodeId -1 in state INIT (for 0.000s)
  refcnt 3
  outbuf_cnt: 0 waitresp_cnt: 0
  0 messages sent, 0 bytes, 0 errors, 0 timeouts
  0 messages received, 0 bytes, 0 errors
  0 messageset transmissions were retried
  0 toppars:
 rd_kafka_broker_t 0x7ff613052c00: sasl_ssl://...:9092/bootstrap NodeId -1 in state INIT (for 0.000s)
  refcnt 3
  outbuf_cnt: 0 waitresp_cnt: 0
  0 messages sent, 0 bytes, 0 errors, 0 timeouts
  0 messages received, 0 bytes, 0 errors
  0 messageset transmissions were retried
  0 toppars:
 rd_kafka_broker_t 0x7ff612097e00: GroupCoordinator NodeId -1 in state INIT (for 0.000s)
  refcnt 4
  outbuf_cnt: 0 waitresp_cnt: 0
  0 messages sent, 0 bytes, 0 errors, 0 timeouts
  0 messages received, 0 bytes, 0 errors
  0 messageset transmissions were retried
  0 toppars:
 cgrp:
  fe467669-59b1-4667-af31-43801f7e1b0f in state query-coord, flags 0x0
   coord_id -1, broker (none)
  toppars:
 topics:

Metadata cache with 0 entries:
Abort trap: 6

[lpsinger]

FYI, this is reportedly fixed in librdkafka master, to be included in the next release. confluentinc/librdkafka#3263 (comment)