[CSS-6503] Add OAuth support for non-charmed external clients (#168)

metadata.yaml

@@ -42,6 +42,10 @@ requires:
   trusted-certificate:
     interface: tls-certificates
     optional: true
+  oauth:
+    interface: oauth
+    limit: 1
+    optional: true
 
 provides:
   kafka-client:

src/charm.py

@@ -28,6 +28,7 @@
 from core.cluster import ClusterState
 from core.models import Substrates
 from core.structured_config import CharmConfig
+from events.oauth import OAuthHandler
 from events.password_actions import PasswordActionEvents
 from events.provider import KafkaProvider
 from events.tls import TLSHandler
@@ -76,6 +77,7 @@ def __init__(self, *args):
         self.password_action_events = PasswordActionEvents(self)
         self.zookeeper = ZooKeeperHandler(self)
         self.tls = TLSHandler(self)
+        self.oauth = OAuthHandler(self)
         self.provider = KafkaProvider(self)
         self.upgrade = KafkaUpgrade(
             self,

src/core/cluster.py

@@ -17,14 +17,16 @@
 from ops import Framework, Object, Relation
 from ops.model import Unit
 
-from core.models import KafkaBroker, KafkaClient, KafkaCluster, ZooKeeper
+from core.models import KafkaBroker, KafkaClient, KafkaCluster, OAuth, ZooKeeper
 from literals import (
     INTERNAL_USERS,
+    OAUTH_REL_NAME,
     PEER,
     REL_NAME,
     SECRETS_UNIT,
     SECURITY_PROTOCOL_PORTS,
     ZK,
+    AuthMechanism,
     Status,
     Substrates,
 )
@@ -63,6 +65,11 @@ def client_relations(self) -> set[Relation]:
         """The relations of all client applications."""
         return set(self.model.relations[REL_NAME])
 
+    @property
+    def oauth_relation(self) -> Relation | None:
+        """The OAuth relation."""
+        return self.model.get_relation(OAUTH_REL_NAME)
+
     # --- CORE COMPONENTS ---
 
     @property
@@ -127,6 +134,13 @@ def zookeeper(self) -> ZooKeeper:
             local_app=self.cluster.app,
         )
 
+    @property
+    def oauth(self) -> OAuth:
+        """The oauth relation state."""
+        return OAuth(
+            relation=self.oauth_relation,
+        )
+
     @property
     def clients(self) -> set[KafkaClient]:
         """The state for all related client Applications."""
@@ -180,10 +194,11 @@ def super_users(self) -> str:
     @property
     def port(self) -> int:
         """Return the port to be used internally."""
+        mechanism: AuthMechanism = "SCRAM-SHA-512"
         return (
-            SECURITY_PROTOCOL_PORTS["SASL_SSL"].client
+            SECURITY_PROTOCOL_PORTS["SASL_SSL", mechanism].client
             if (self.cluster.tls_enabled and self.unit_broker.certificate)
-            else SECURITY_PROTOCOL_PORTS["SASL_PLAINTEXT"].client
+            else SECURITY_PROTOCOL_PORTS["SASL_PLAINTEXT", mechanism].client
         )
 
     @property

src/core/models.py

@@ -5,7 +5,9 @@
 """Collection of state objects for the Kafka relations, apps and units."""
 
 import logging
+from typing import MutableMapping
 
+import requests
 from charms.data_platform_libs.v0.data_interfaces import Data, DataPeerData, DataPeerUnitData
 from charms.zookeeper.v0.client import QuorumLeaderNotFoundError, ZooKeeperManager
 from kazoo.client import AuthFailedError, NoNodeError
@@ -451,3 +453,49 @@ def extra_user_roles(self) -> str:
         When `admin` is set, the Kafka charm interprets this as a new super.user.
         """
         return self.relation_data.get("extra-user-roles", "")
+
+
+class OAuth:
+    """State collection metadata for the oauth relation."""
+
+    def __init__(self, relation: Relation | None):
+        self.relation = relation
+
+    @property
+    def relation_data(self) -> MutableMapping[str, str]:
+        """Oauth relation data object."""
+        if not self.relation or not self.relation.app:
+            return {}
+
+        return self.relation.data[self.relation.app]
+
+    @property
+    def issuer_url(self) -> str:
+        """The issuer URL to identify the IDP."""
+        return self.relation_data.get("issuer_url", "")
+
+    @property
+    def jwks_endpoint(self) -> str:
+        """The JWKS endpoint needed to validate JWT tokens."""
+        return self.relation_data.get("jwks_endpoint", "")
+
+    @property
+    def introspection_endpoint(self) -> str:
+        """The introspection endpoint needed to validate non-JWT tokens."""
+        return self.relation_data.get("introspection_endpoint", "")
+
+    @property
+    def jwt_access_token(self) -> bool:
+        """A flag indicating if the access token is JWT or not."""
+        return self.relation_data.get("jwt_access_token", "false").lower() == "true"
+
+    @property
+    def uses_trusted_ca(self) -> bool:
+        """A flag indicating if the IDP uses certificates signed by a trusted CA."""
+        try:
+            requests.get(self.issuer_url, timeout=10)
+            return True
+        except requests.exceptions.SSLError:
+            return False
+        except requests.exceptions.RequestException:
+            return True

src/events/oauth.py

@@ -0,0 +1,40 @@
+# Copyright 2023 Canonical Ltd.
+# See LICENSE file for licensing details.
+
+"""Manager for handling Kafka OAuth configuration."""
+
+import logging
+from typing import TYPE_CHECKING
+
+from charms.hydra.v0.oauth import ClientConfig, OAuthRequirer
+from ops.framework import EventBase, Object
+
+from literals import OAUTH_REL_NAME
+
+if TYPE_CHECKING:
+    from charm import KafkaCharm
+
+logger = logging.getLogger(__name__)
+
+
+class OAuthHandler(Object):
+    """Handler for managing oauth relations."""
+
+    def __init__(self, charm):
+        super().__init__(charm, "oauth")
+        self.charm: "KafkaCharm" = charm
+
+        client_config = ClientConfig("https://kafka.local", "openid email", ["client_credentials"])
+        self.oauth = OAuthRequirer(charm, client_config, relation_name=OAUTH_REL_NAME)
+        self.framework.observe(
+            self.charm.on[OAUTH_REL_NAME].relation_changed, self._on_oauth_relation_changed
+        )
+        self.framework.observe(
+            self.charm.on[OAUTH_REL_NAME].relation_broken, self._on_oauth_relation_changed
+        )
+
+    def _on_oauth_relation_changed(self, event: EventBase) -> None:
+        """Handler for `_on_oauth_relation_changed` event."""
+        if not self.charm.unit.is_leader() or not self.charm.state.brokers:
+            return
+        self.charm._on_config_changed(event)

src/literals.py

@@ -23,6 +23,7 @@
 PEER = "cluster"
 ZK = "zookeeper"
 REL_NAME = "kafka-client"
+OAUTH_REL_NAME = "oauth"
 TLS_RELATION = "certificates"
 TRUSTED_CERTIFICATE_RELATION = "trusted-certificate"
 TRUSTED_CA_RELATION = "trusted-ca"
@@ -53,7 +54,8 @@
 USER = 584788
 GROUP = "root"
 
-AuthMechanism = Literal["SASL_PLAINTEXT", "SASL_SSL", "SSL"]
+AuthProtocol = Literal["SASL_PLAINTEXT", "SASL_SSL", "SSL"]
+AuthMechanism = Literal["SCRAM-SHA-512", "OAUTHBEARER", "SSL"]
 Scope = Literal["INTERNAL", "CLIENT"]
 DebugLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR"]
 DatabagScope = Literal["unit", "app"]
@@ -84,10 +86,12 @@ class Ports:
     internal: int
 
 
-SECURITY_PROTOCOL_PORTS: dict[AuthMechanism, Ports] = {
-    "SASL_PLAINTEXT": Ports(9092, 19092),
-    "SASL_SSL": Ports(9093, 19093),
-    "SSL": Ports(9094, 19094),
+SECURITY_PROTOCOL_PORTS: dict[tuple[AuthProtocol, AuthMechanism], Ports] = {
+    ("SASL_PLAINTEXT", "SCRAM-SHA-512"): Ports(9092, 19092),
+    ("SASL_PLAINTEXT", "OAUTHBEARER"): Ports(9095, 19095),
+    ("SASL_SSL", "SCRAM-SHA-512"): Ports(9093, 19093),
+    ("SASL_SSL", "OAUTHBEARER"): Ports(9096, 19096),
+    ("SSL", "SSL"): Ports(9094, 19094),
 }