Add configuration for JWT authentication

CHANGES.rst

@@ -5,6 +5,8 @@ Changelog
 Unreleased
 ----------
 
+* Added configuration for JWT authentication from CrateDB version 5.7.2 onwards.
+
 2.40.2 (2024-08-01)
 -------------------
 

crate/operator/bootstrap.py

@@ -207,7 +207,7 @@ async def bootstrap_gc_admin_user(core: CoreV1Api, namespace: str, name: str):
             password = await resolve_secret_key_ref(
                 core, namespace, {"key": "password", "name": f"user-gc-{name}"}
             )
-            await create_user(cursor, GC_USERNAME, password)
+            await create_user(cursor, namespace, name, GC_USERNAME, password)
 
 
 async def bootstrap_users(
@@ -239,7 +239,7 @@ async def bootstrap_users(
                 await ensure_user_password_label(
                     core, namespace, secret_key_ref["name"]
                 )
-                await create_user(cursor, username, password)
+                await create_user(cursor, namespace, name, username, password)
 
 
 async def create_users(

crate/operator/config.py

@@ -150,6 +150,9 @@ class Config:
     #: and `recover_after_nodes`
     GATEWAY_SETTINGS_DATA_NODES_VERSION: str = "4.7.0"
 
+    #: From which version onwards CrateDB supports JWT authentication
+    CRATEDB_JWT_AUTH_VERSION: str = "5.7.2"
+
     #: Interval in seconds for which the operator will ping CrateDBs for their
     #: current health.
     CRATEDB_STATUS_CHECK_INTERVAL: Optional[int] = 60

crate/operator/cratedb.py

@@ -21,15 +21,17 @@
 
 import functools
 import logging
-from typing import Dict, Optional, Tuple, Union
+from typing import Dict, List, Optional, Tuple, Union
 
 import aiopg
 from aiopg import Cursor
 from psycopg2 import ProgrammingError
 from psycopg2.extensions import quote_ident
 
 from crate.operator.config import config
-from crate.operator.constants import SYSTEM_USERNAME
+from crate.operator.constants import GC_USERNAME, SYSTEM_USERNAME
+from crate.operator.utils.jwt import crate_version_supports_jwt
+from crate.operator.utils.kubeapi import get_cratedb_resource
 
 HEALTHINESS = {1: "GREEN", 2: "YELLOW", 3: "RED"}
 
@@ -69,15 +71,29 @@ def connection_factory(
     return functools.partial(get_connection, host, password, username, **kwargs)
 
 
-async def create_user(cursor: Cursor, username: str, password: str) -> None:
+async def create_user(
+    cursor: Cursor,
+    namespace: str,
+    name: str,
+    username: str,
+    password: str,
+    privileges: List[str] = None,
+) -> None:
     """
-    Create user ``username`` and grant it ``ALL PRIVILEGES``.
+    Create user ``username`` and grant it given privileges.
 
     :param cursor: A database cursor object to the CrateDB cluster where the
         user should be added.
+    :param namespace: The Kubernetes namespace of the CrateDB cluster.
+    :param name: The CrateDB custom resource name defining the CrateDB cluster.
     :param username: The username for the new user.
     :param password: The password for the new user.
+    :param privileges: An optional list of privileges granted to the user.
+        Defaults to ``ALL``
     """
+    privileges = privileges or ["ALL"]
+    cratedb = await get_cratedb_resource(namespace, name)
+    crate_version = cratedb["spec"]["cluster"]["version"]
     await cursor.execute(
         "SELECT count(*) = 1 FROM sys.users WHERE name = %s", (username,)
     )
@@ -86,10 +102,20 @@ async def create_user(cursor: Cursor, username: str, password: str) -> None:
 
     username_ident = quote_ident(username, cursor._impl)
     if not user_exists:
-        await cursor.execute(
-            f"CREATE USER {username_ident} WITH (password = %s)", (password,)
-        )
-    await cursor.execute(f"GRANT ALL PRIVILEGES TO {username_ident}")
+        query = f"CREATE USER {username_ident} WITH (password = %s"
+        params = [password]
+        if crate_version_supports_jwt(crate_version):
+            iss = cratedb["spec"].get("grandCentral", {}).get("jwkUrl")
+            query += ', jwt = {"iss" = %s, "username" = %s, "aud" = %s}'
+            params.extend([iss, username, name])
+
+        query += ")"
+        await cursor.execute(query, tuple(params))
+
+    await cursor.execute(f"GRANT {','.join(privileges)} TO {username_ident}")
+
+    if not username == GC_USERNAME:
+        await cursor.execute(f"DENY ALL ON SCHEMA gc TO {username_ident}")
 
 
 async def update_user(cursor: Cursor, username: str, password: str) -> None:

crate/operator/create.py

@@ -98,6 +98,7 @@
 )
 from crate.operator.utils import crate, quorum
 from crate.operator.utils.formatting import b64encode, format_bitmath
+from crate.operator.utils.jwt import crate_version_supports_jwt
 from crate.operator.utils.k8s_api_client import GlobalApiClient
 from crate.operator.utils.kopf import StateBasedSubHandler
 from crate.operator.utils.kubeapi import call_kubeapi
@@ -457,6 +458,15 @@ def get_statefulset_crate_command(
             }
         )
 
+    if crate_version_supports_jwt(crate_version):
+        settings.update(
+            {
+                "-Cauth.host_based.config.98.method": "jwt",
+                "-Cauth.host_based.config.98.protocol": "http",
+                "-Cauth.host_based.config.98.ssl": "on",
+            }
+        )
+
     # Availability zone retrieval at pod launch time
     if config.CLOUD_PROVIDER == CloudProvider.AWS:
         aws_cmd = (

crate/operator/grand_central.py

@@ -462,7 +462,7 @@ async def update_grand_central_deployment_image(
         # This also runs on creation events, so we need to double check that the
         # deployment exists before attempting to do anything.
         deployments = await apps.list_namespaced_deployment(namespace=namespace)
-        deployment = next(
+        deployment: V1Deployment = next(
             (
                 deploy
                 for deploy in deployments.items

crate/operator/handlers/handle_create_grand_central.py

@@ -28,8 +28,8 @@
     create_grand_central_backend,
     create_grand_central_user,
 )
-from crate.operator.operations import get_cratedb_resource
 from crate.operator.utils.kopf import subhandler_partial
+from crate.operator.utils.kubeapi import get_cratedb_resource
 
 logger = logging.getLogger(__name__)
 

crate/operator/handlers/handle_restore_backup.py

@@ -32,7 +32,7 @@
 from crate.operator.operations import (
     AfterClusterUpdateSubHandler,
     BeforeClusterUpdateSubHandler,
-    get_cratedb_resource,
+    RestoreUserJWTAuthSubHandler,
 )
 from crate.operator.restore_backup import (
     AfterRestoreBackupSubHandler,
@@ -46,6 +46,7 @@
     get_crash_pod_name,
     get_crash_scheme,
 )
+from crate.operator.utils.kubeapi import get_cratedb_resource
 from crate.operator.utils.notifications import FlushNotificationsSubHandler
 
 CLUSTER_RESTORE_FIELD_ID = "cluster_restore/spec.cluster.restoreSnapshot"
@@ -229,6 +230,19 @@ def register_restore_handlers(
     )
     depends_on.append(f"{CLUSTER_RESTORE_FIELD_ID}/restore_system_user_password")
 
+    kopf.register(
+        fn=RestoreUserJWTAuthSubHandler(
+            namespace,
+            name,
+            change_hash,
+            context,
+            depends_on=depends_on.copy(),
+        )(),
+        id="restore_user_jwt_auth",
+        backoff=get_backoff(),
+    )
+    depends_on.append(f"{CLUSTER_RESTORE_FIELD_ID}/restore_user_jwt_auth")
+
     kopf.register(
         fn=ValidateRestoreCompleteSubHandler(
             namespace,

crate/operator/handlers/handle_update_cratedb.py

@@ -37,6 +37,7 @@
     AfterClusterUpdateSubHandler,
     BeforeClusterUpdateSubHandler,
     RestartSubHandler,
+    RestoreUserJWTAuthSubHandler,
     StartClusterSubHandler,
     SuspendClusterSubHandler,
     set_cronjob_delay,
@@ -288,6 +289,18 @@ def register_restart_handlers(
     depends_on.append(f"{CLUSTER_UPDATE_ID}/restart")
     # Send a webhook success notification after upgrade and restart handlers
     if do_upgrade:
+        kopf.register(
+            fn=RestoreUserJWTAuthSubHandler(
+                namespace,
+                name,
+                change_hash,
+                context,
+                depends_on=depends_on.copy(),
+            )(),
+            id="restore_user_jwt_auth",
+            backoff=get_backoff(),
+        )
+        depends_on.append(f"{CLUSTER_UPDATE_ID}/restore_user_jwt_auth")
         kopf.register(
             fn=AfterUpgradeSubHandler(
                 namespace, name, change_hash, context, depends_on=depends_on.copy()