redpanda_oauth_test: Improve test example

- Create a user under the demo realm, with realm-admin priv
- Update client's service account to include an email address
  - This will appear in the access token as the grant includes
    the 'email' scope.
- Adjust usage to account for keycloak-python based impl

tests/rptest/clients/oauth_producer.py

@@ -16,11 +16,9 @@
 
 from confluent_kafka import Producer
 
+
 class OAuthProducer:
-    def __init__(self,
-                 redpanda,
-                 *,
-                 oauth_config=None):
+    def __init__(self, redpanda, *, oauth_config=None):
         self._redpanda = redpanda
         self._oauth_config = oauth_config
 
@@ -47,19 +45,22 @@ def _get_token(self, conf, _):
             payload = {
                 'client_id': conf.client_id,
                 'client_secret': conf.client_secret,
-                'audience': 'localhost',
+                'audience': 'redpanda',
                 'grant_type': 'client_credentials',
+                'scope': ' '.join(conf.scopes),
             }
-            self._redpanda.logger.debug(f"GETTING TOKEN: {conf.token_endpoint}, payload: {payload}")
+            self._redpanda.logger.debug(
+                f"GETTING TOKEN: {conf.token_endpoint}, payload: {payload}")
 
-            resp = requests.post(conf.token_endpoint,
-                                 headers={'content-type': 'application/x-www-form-urlencoded'},
-                                 auth=(conf.client_id, conf.client_secret),
-                                 data=payload)
-            self._redpanda.logger.info(f"response status: {resp.status_code}, body: {resp.content}")
+            resp = requests.post(
+                conf.token_endpoint,
+                headers={'content-type': 'application/x-www-form-urlencoded'},
+                auth=(conf.client_id, conf.client_secret),
+                data=payload)
+            self._redpanda.logger.info(
+                f"response status: {resp.status_code}, body: {resp.content}")
             token = resp.json()
-            return token['access_token'], time.time() + float(token['expires_in'])
+            return token['access_token'], time.time() + float(
+                token['expires_in'])
         except Exception as e:
             self._redpanda.logger.error(f"Exception: {e}")
-
-

tests/rptest/tests/redpanda_oauth_test.py

@@ -9,6 +9,7 @@
 
 import time
 import functools
+import json
 
 from rptest.clients.oauth_producer import OAuthProducer
 from rptest.clients.python_librdkafka import PythonLibrdkafka
@@ -24,14 +25,15 @@
 
 CLIENT_ID = 'myapp'
 
+
 class RedpandaOIDCTestBase(Test):
     """
     Base class for tests that use the Redpanda service with OIDC
     """
     def __init__(self,
                  test_context,
                  num_nodes=5,
-                 sasl_mechanisms=[ 'SCRAM', 'OAUTHBEARER'],
+                 sasl_mechanisms=['SCRAM', 'OAUTHBEARER'],
                  **kwargs):
         super(RedpandaOIDCTestBase, self).__init__(test_context, **kwargs)
         self.produce_messages = []
@@ -64,10 +66,13 @@ def delivery_report(self, err, msg):
 
         """
         if err is not None:
-            self.produce_errors.append('Delivery failed for User record {}: {}'.format(msg.key(), err))
+            self.produce_errors.append(
+                'Delivery failed for User record {}: {}'.format(
+                    msg.key(), err))
             return
-        self.produce_messages.append('User record {} successfully produced to {} [{}] at offset {}'.format(
-                msg.key(), msg.topic(), msg.partition(), msg.offset()))
+        self.produce_messages.append(
+            'User record {} successfully produced to {} [{}] at offset {}'.
+            format(msg.key(), msg.topic(), msg.partition(), msg.offset()))
 
     def setUp(self):
         self.produce_messages.clear()
@@ -77,7 +82,6 @@ def setUp(self):
 
 
 class RedpandaOIDCTest(RedpandaOIDCTestBase):
-
     @cluster(num_nodes=5)
     def test_init(self):
         kc_node = self.keycloak.nodes[0]
@@ -90,31 +94,42 @@ def test_init(self):
 
         self.rpk.create_topic('foo')
 
-        # TODO: Improve understanding of client config space and hopefully bake some
-        #       of this into KeycloakService.
-        self.keycloak.create_client(kc_node, {
-            'clientId': CLIENT_ID,
-            'enabled': True,
-            'serviceAccountsEnabled': True,
-            'standardFlowEnabled': True,
-            'directAccessGrantsEnabled': True,
-            'implicitFlowEnabled': True,
-        })
-
-        cfg = self.keycloak.get_oauth_config(kc_node, CLIENT_ID)
+        self.keycloak.admin.create_user('norma',
+                                        'desmond',
+                                        realm_admin=True,
+                                        email='[email protected]')
+        self.keycloak.login_admin_user(kc_node, 'norma', 'desmond')
+        self.keycloak.admin.create_client(CLIENT_ID)
+
+        # add an email address to myapp client's service user. this should
+        # appear alongside the access token.
+        self.keycloak.admin.update_user(f'service-account-{CLIENT_ID}',
+                                        email='[email protected]')
+
+        cfg = self.keycloak.generate_oauth_config(kc_node, CLIENT_ID)
         assert cfg.client_secret is not None
         assert cfg.token_endpoint is not None
         p_client = OAuthProducer(self.redpanda, oauth_config=cfg)
         producer = p_client.get_producer()
 
-        producer.produce(topic='foo', key='bar', value='23', on_delivery=self.delivery_report)
-        producer.produce(topic='foo', key='baz', value='23', on_delivery=self.delivery_report)
-        producer.produce(topic='foo', key='qux', value='23', on_delivery=self.delivery_report)
+        producer.produce(topic='foo',
+                         key='bar',
+                         value='23',
+                         on_delivery=self.delivery_report)
+        producer.produce(topic='foo',
+                         key='baz',
+                         value='23',
+                         on_delivery=self.delivery_report)
+        producer.produce(topic='foo',
+                         key='qux',
+                         value='23',
+                         on_delivery=self.delivery_report)
 
         self.logger.info('Flushing {} records...'.format(len(producer)))
 
         # Without the OIDC PoC, Producer.flush raises an AttributeError for some
-        # reason. With OIDC support in place, this works as expected.
+        # reason (rather than just failing). With OIDC support in place, this works
+        # as expected.
         # TODO: Remove Me
         with expect_exception(AttributeError, lambda _: True):
             producer.flush()