Watch on Kubernetes Secrets and update user passwords if needed

Watch on Kubernetes Secrets that have the
`operator.cloud.crate.io/user-password` label assigned and update the
users of all CrateDB resources in the same namespace if the password
changed.

CHANGES.rst

@@ -5,6 +5,10 @@ Changelog
 Unreleased
 ----------
 
+* Watch on Kubernetes Secrets that have the
+  ``operator.cloud.crate.io/user-password`` label assigned and update the users
+  of all CrateDB resources in the same namespace if the password changed.
+
 * Fixed an inconsistent behavior where the configuration option
   :envvar:`CLOUD_PROVIDER` would override an explicitly defined
   ``node.attr.zone`` in either ``.spec.cluster.settings``,

crate/operator/cratedb.py

@@ -86,6 +86,22 @@ async def create_user(cursor: Cursor, username: str, password: str) -> None:
     await cursor.execute(f"GRANT ALL PRIVILEGES TO {username_ident}")
 
 
+async def update_user(cursor: Cursor, username: str, password: str) -> None:
+    """
+    Update the users password.
+
+    :param cursor: A database cursor object to the CrateDB cluster where the
+        user should be updated.
+    :param username: The username of the user that should be updated.
+    :param password: The new password.
+    """
+
+    username_ident = quote_ident(username, cursor._impl)
+    await cursor.execute(
+        f"ALTER USER {username_ident} SET (password = %s)", (password,)
+    )
+
+
 async def get_number_of_nodes(cursor: Cursor) -> int:
     """
     Return the number of nodes in the cluster from ``sys.nodes``, which is the

crate/operator/main.py

@@ -53,9 +53,10 @@
 from crate.operator.kube_auth import login_via_kubernetes_asyncio
 from crate.operator.operations import get_total_nodes_count, restart_cluster
 from crate.operator.scale import scale_cluster
+from crate.operator.update_user_password import update_user_password
 from crate.operator.upgrade import upgrade_cluster
 from crate.operator.utils.kopf import subhandler_partial
-from crate.operator.utils.kubeapi import ensure_user_password_label
+from crate.operator.utils.kubeapi import ensure_user_password_label, get_host
 from crate.operator.webhooks import (
     WebhookScaleNodePayload,
     WebhookScalePayload,
@@ -592,3 +593,51 @@ async def update_cratedb_resource(
                     await ensure_user_password_label(
                         core, namespace, user_spec["password"]["secretKeyRef"]["name"]
                     )
+
+
+@kopf.on.update("", "v1", "secrets", labels={LABEL_USER_PASSWORD: "true"})
+async def secret_update(
+    namespace: str,
+    name: str,
+    diff: kopf.Diff,
+    logger: logging.Logger,
+    **kwargs,
+):
+    async with ApiClient() as api_client:
+        coapi = CustomObjectsApi(api_client)
+        core = CoreV1Api(api_client)
+
+        for operation, field_path, old_value, new_value in diff:
+            custom_objects = await coapi.list_namespaced_custom_object(
+                namespace=namespace,
+                group=API_GROUP,
+                version="v1",
+                plural=RESOURCE_CRATEDB,
+            )
+
+            for crate_custom_object in custom_objects["items"]:
+                host = await get_host(
+                    core, namespace, crate_custom_object["metadata"]["name"]
+                )
+
+                for user_spec in crate_custom_object["spec"]["users"]:
+                    expected_field_path = (
+                        "data",
+                        user_spec["password"]["secretKeyRef"]["key"],
+                    )
+                    if (
+                        user_spec["password"]["secretKeyRef"]["name"] == name
+                        and field_path == expected_field_path
+                    ):
+                        kopf.register(
+                            fn=subhandler_partial(
+                                update_user_password,
+                                host,
+                                user_spec["name"],
+                                old_value,
+                                new_value,
+                                logger,
+                            ),
+                            id=f"update-{crate_custom_object['metadata']['name']}-{user_spec['name']}",  # noqa
+                            timeout=config.BOOTSTRAP_TIMEOUT,
+                        )

crate/operator/update_user_password.py

@@ -0,0 +1,46 @@
+# CrateDB Kubernetes Operator
+# Copyright (C) 2020 Crate.IO GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import logging
+
+from crate.operator.constants import BACKOFF_TIME
+from crate.operator.cratedb import get_connection, update_user
+from crate.operator.utils.formatting import b64decode
+
+
+# update_user_password(host, username, old_password, new_password)
+async def update_user_password(
+    host: str,
+    username: str,
+    old_password: str,
+    new_password: str,
+    logger: logging.Logger,
+):
+    """
+    Update the password of a given ``user_spec`` in a CrateDB cluster.
+
+    :param host: The host of the CrateDB resource that should be updated.
+    :param username: The username of the user of the CrateDB resource that
+        should be updated.
+    :param old_password: The old password of the user that should be updated.
+    :param new_password: The new password of the user that should be updated.
+    """
+    async with get_connection(
+        host, b64decode(old_password), username, timeout=BACKOFF_TIME / 4.0
+    ) as conn:
+        async with conn.cursor() as cursor:
+            logger.info("Updating password for user '%s'", username)
+            await update_user(cursor, username, b64decode(new_password))

docs/source/concepts.rst

@@ -33,6 +33,27 @@ The entire bootstrapping process may not take longer than (by default) 1800
 seconds before it is considered failed. The timeout can be configured with the
 :envvar:`BOOTSTRAP_TIMEOUT` environment variable.
 
+User Passwords
+~~~~~~~~~~~~~~
+
+When creating the CrateDB users specified under ``.spec.users``, the operator
+will add the ``operator.cloud.crate.io/user-password`` label to each of the
+Kubernetes Secrets assigned to one of the users. To keep backward
+compatibility, it also adds the label to Kubernetes Secrets referenced in
+existing CrateDB resources :func:`on resume <kopf:kopf.on.resume>`.
+
+The ``operator.cloud.crate.io/user-password`` label is used to filter the
+events when watching for changes on one of the Kubernetes Secrets. If one of
+the Kubernetes Secrets is updated, the operator will update all CrateDB users
+that use that secret by iterating over all CrateDB resources. The operator
+updates the password in a CrateDB cluster by logging in to one of the CrateDB
+nodes with the corresponding username and old password. It will then use the
+:ref:`cratedb:ref-alter-user` query to update the password.
+
+.. note::
+
+   If one changes the CrateDB user's password directly in CrateDB, the operator
+   won't be able to update that user anymore.
 
 Cluster Restart
 ---------------

tests/test_update_password.py

@@ -0,0 +1,148 @@
+# CrateDB Kubernetes Operator
+# Copyright (C) 2020 Crate.IO GmbH
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+
+import asyncio
+
+import pytest
+from kubernetes_asyncio.client import (
+    CoreV1Api,
+    CustomObjectsApi,
+    V1ObjectMeta,
+    V1Secret,
+)
+from psycopg2 import DatabaseError, OperationalError
+
+from crate.operator.constants import (
+    API_GROUP,
+    BACKOFF_TIME,
+    LABEL_USER_PASSWORD,
+    RESOURCE_CRATEDB,
+)
+from crate.operator.cratedb import get_connection
+from crate.operator.utils.formatting import b64encode
+from crate.operator.utils.kubeapi import get_public_host
+
+from .utils import assert_wait_for
+
+pytestmark = [pytest.mark.k8s, pytest.mark.asyncio]
+
+
+async def is_password_set(host: str, system_password: str, user: str) -> bool:
+    try:
+        async with get_connection(host, system_password, user, timeout=5.0) as conn:
+            async with conn.cursor() as cursor:
+                await cursor.execute("SELECT 1")
+                row = await cursor.fetchone()
+                return bool(row and row[0] == 1)
+    except (DatabaseError, OperationalError):
+        return False
+
+
+async def test_update_cluster_password(
+    faker, namespace, cleanup_handler, kopf_runner, api_client
+):
+    coapi = CustomObjectsApi(api_client)
+    core = CoreV1Api(api_client)
+    name = faker.domain_word()
+    password = faker.password(length=40)
+    new_password = faker.password(length=40)
+    username = faker.user_name()
+
+    cleanup_handler.append(
+        core.delete_persistent_volume(name=f"temp-pv-{namespace.metadata.name}-{name}")
+    )
+    await asyncio.gather(
+        core.create_namespaced_secret(
+            namespace=namespace.metadata.name,
+            body=V1Secret(
+                data={"password": b64encode(password)},
+                metadata=V1ObjectMeta(
+                    name=f"user-{name}", labels={LABEL_USER_PASSWORD: "true"}
+                ),
+                type="Opaque",
+            ),
+        ),
+    )
+
+    await coapi.create_namespaced_custom_object(
+        group=API_GROUP,
+        version="v1",
+        plural=RESOURCE_CRATEDB,
+        namespace=namespace.metadata.name,
+        body={
+            "apiVersion": "cloud.crate.io/v1",
+            "kind": "CrateDB",
+            "metadata": {"name": name},
+            "spec": {
+                "cluster": {
+                    "imageRegistry": "crate",
+                    "name": "my-crate-cluster",
+                    "version": "4.1.5",
+                },
+                "nodes": {
+                    "data": [
+                        {
+                            "name": "data",
+                            "replicas": 1,
+                            "resources": {
+                                "cpus": 0.5,
+                                "memory": "1Gi",
+                                "heapRatio": 0.25,
+                                "disk": {
+                                    "storageClass": "default",
+                                    "size": "16GiB",
+                                    "count": 1,
+                                },
+                            },
+                        }
+                    ]
+                },
+                "users": [
+                    {
+                        "name": username,
+                        "password": {
+                            "secretKeyRef": {
+                                "key": "password",
+                                "name": f"user-{name}",
+                            }
+                        },
+                    },
+                ],
+            },
+        },
+    )
+
+    host = await asyncio.wait_for(
+        get_public_host(core, namespace.metadata.name, name),
+        timeout=BACKOFF_TIME * 5,  # It takes a while to retrieve an external IP on AKS.
+    )
+
+    await core.patch_namespaced_secret(
+        namespace=namespace.metadata.name,
+        name=f"user-{name}",
+        body=V1Secret(
+            data={"password": b64encode(new_password)},
+        ),
+    )
+
+    await assert_wait_for(
+        True,
+        is_password_set,
+        host,
+        new_password,
+        username,
+        timeout=BACKOFF_TIME,
+    )