Merge branch 'dev-v2.8' into 2.8

CHANGELOG.rst

@@ -7,6 +7,14 @@
 Changelog
 ---------
 
+v.2.8.12 2022-10-26
+===================
+
+Bugfixes
+--------
+
+* CVE-2022-43685: fix potential user account takeover via user create
+
 v.2.8.11 2022-09-28
 ===================
 

ckan/__init__.py

@@ -1,6 +1,6 @@
 # encoding: utf-8
 
-__version__ = '2.8.11'
+__version__ = '2.8.12'
 
 __description__ = 'CKAN Software'
 __long_description__ = \

ckan/logic/action/create.py

@@ -953,6 +953,13 @@ def user_create(context, data_dict):
 
     _check_access('user_create', context, data_dict)
 
+    author_obj = model.User.get(context.get('user'))
+    if data_dict.get("id"):
+        is_sysadmin = (author_obj and author_obj.sysadmin)
+        if not is_sysadmin or model.User.get(data_dict["id"]):
+            data_dict.pop("id", None)
+    context.pop("user_obj", None)
+
     data, errors = _validate(data_dict, schema, context)
 
     if errors:

ckan/tests/logic/action/test_create.py

@@ -1201,3 +1201,112 @@ def test_user_create_password_hash_not_for_normal_users(self):
 
         user_obj = model.User.get(user['id'])
         assert user_obj.password != 'pretend-this-is-a-valid-hash'
+
+    def test_anon_user_create_does_not_update(self):
+
+        user1 = factories.User(about="This is user 1")
+        user_dict = {
+            "id": user1["id"],
+            "name": "some_name",
+            "email": "some_email@example.com",
+            "password": "test1234",
+        }
+
+        context = {
+            "user": None,
+            "ignore_auth": False,
+        }
+
+        user2 = helpers.call_action("user_create", context=context, **user_dict)
+        assert user2["id"] != user1["id"]
+        assert user2["about"] != "This is user 1"
+
+    def test_normal_user_create_does_not_update(self):
+
+        user1 = factories.User(about="This is user 1")
+        user_dict = {
+            "id": user1["id"],
+            "name": "some_name",
+            "email": "some_email@example.com",
+            "password": "test1234",
+        }
+
+        context = {
+            "user": factories.User()["name"],
+            "ignore_auth": False,
+        }
+
+        user2 = helpers.call_action("user_create", context=context, **user_dict)
+        assert user2["id"] != user1["id"]
+        assert user2["about"] != "This is user 1"
+
+    def test_sysadmin_user_create_does_not_update(self):
+
+        user1 = factories.User(about="This is user 1")
+        user_dict = {
+            "id": user1["id"],
+            "name": "some_name",
+            "email": "some_email@example.com",
+            "password": "test1234",
+        }
+
+        context = {
+            "user": factories.Sysadmin()["name"],
+            "ignore_auth": False,
+        }
+
+        user2 = helpers.call_action("user_create", context=context, **user_dict)
+        assert user2["id"] != user1["id"]
+        assert user2["about"] != "This is user 1"
+
+    def test_anon_user_can_not_provide_id(self):
+
+        user_dict = {
+            "id": "custom_id",
+            "name": "some_name",
+            "email": "some_email@example.com",
+            "password": "test1234",
+        }
+
+        context = {
+            "user": None,
+            "ignore_auth": False,
+        }
+
+        user = helpers.call_action("user_create", context=context, **user_dict)
+        assert user["id"] != "custom_id"
+
+    def test_normal_user_can_not_provide_id(self):
+
+        user_dict = {
+            "id": "custom_id",
+            "name": "some_name",
+            "email": "some_email@example.com",
+            "password": "test1234",
+        }
+
+        context = {
+            "user": None,
+            "ignore_auth": False,
+        }
+
+        user = helpers.call_action("user_create", context=context, **user_dict)
+        assert user["id"] != "custom_id"
+
+    def test_sysadmin_can_provide_custom_id(self):
+
+        sysadmin = factories.Sysadmin()
+
+        user_dict = {
+            "id": "custom_id",
+            "name": "some_name",
+            "email": "some_email@example.com",
+            "password": "test1234",
+        }
+        context = {
+            "user": sysadmin["name"],
+            "ignore_auth": False,
+        }
+
+        user = helpers.call_action("user_create", context=context, **user_dict)
+        assert user["id"] == "custom_id"