Limit Sysadmin Updating (no self modify, no system user modify)

changes/8155.feature

@@ -0,0 +1 @@
+Sysadmins can no longer demote themselves or the system user.

ckan/logic/schema.py

@@ -422,7 +422,7 @@ def default_user_schema(
         not_empty: Validator, strip_value: Validator,
         email_validator: Validator, user_about_validator: Validator,
         ignore: Validator, boolean_validator: Validator,
-        json_object: Validator) -> Schema:
+        json_object: Validator, limit_sysadmin_update: Validator) -> Schema:
     return {
         'id': [ignore_missing, unicode_safe],
         'name': [
@@ -435,7 +435,8 @@ def default_user_schema(
                   unicode_safe],
         'about': [ignore_missing, user_about_validator, unicode_safe],
         'created': [ignore],
-        'sysadmin': [ignore_missing, ignore_not_sysadmin],
+        'sysadmin': [ignore_missing, ignore_not_sysadmin,
+                     limit_sysadmin_update],
         'reset_key': [ignore],
         'activity_streams_email_notifications': [ignore_missing,
                                                  boolean_validator],

ckan/logic/validators.py

@@ -25,7 +25,7 @@
 import ckan.authz as authz
 from ckan.model.core import State
 
-from ckan.common import _
+from ckan.common import _, config
 from ckan.types import (
     FlattenDataDict, FlattenKey, Validator, Context, FlattenErrorDict)
 
@@ -562,6 +562,55 @@ def ignore_not_sysadmin(key: FlattenKey, data: FlattenDataDict,
     data.pop(key)
 
 
+def limit_sysadmin_update(key: FlattenKey, data: FlattenDataDict,
+                          errors: FlattenErrorDict, context: Context) -> Any:
+    """
+    Should not be able to modify your own sysadmin privs, or the system user's
+    """
+    value = data.get(key)
+
+    if value is None:
+        # sysadmin key not in data, return here
+        return
+
+    contextual_user = context.get('auth_user_obj')
+    site_id = config.get('ckan.site_id')
+    contextual_user_name = context.get('user')
+
+    if not contextual_user and contextual_user_name:
+        contextual_user = context['model'].User.get(contextual_user_name)
+
+    if not contextual_user:
+        # no auth user supplied, fail here
+        errors[key].append(_('Unauthorized to modify sysadmin privileges'))
+        return
+
+    # system user should be able to do anything still
+    if contextual_user_name == site_id:
+        return
+
+    user = context.get('user_obj')
+
+    if not user:
+        # user_obj not set, fail here
+        errors[key].append(_('Cannot modify sysadmin privileges for non-existent user'))
+        return
+
+    # sysadmin not being updated, return here
+    if value == user.sysadmin:
+        return
+
+    # cannot change your own sysadmin value
+    if user.name == contextual_user_name:
+        errors[key].append(_('Cannot modify your own sysadmin privileges'))
+
+    # cannot change site user sysadmin value
+    if user.name == site_id:
+        errors[key].append(_('Cannot modify sysadmin privileges for system user'))
+
+    return
+
+
 def ignore_not_group_admin(key: FlattenKey, data: FlattenDataDict,
                            errors: FlattenErrorDict, context: Context) -> Any:
     '''Ignore if the user is not allowed to administer for the group specified.'''

ckan/tests/helpers.py

@@ -123,7 +123,7 @@ def call_action(action_name: str, context=None, **kwargs):
     """
     if context is None:
         context = {}
-    context.setdefault("user", "127.0.0.1")
+    context.setdefault("user", config.get('ckan.site_id'))
     context.setdefault("ignore_auth", True)
     return logic.get_action(action_name)(context, kwargs)
 

ckan/tests/logic/action/test_patch.py

@@ -7,6 +7,8 @@
 from ckan.tests import helpers, factories
 from ckan.logic.action.get import package_show as core_package_show
 from ckan.logic import ValidationError
+from ckan.plugins.toolkit import config
+from ckan import model
 
 
 @pytest.mark.usefixtures("non_clean_db")
@@ -234,3 +236,44 @@ def test_resource_patch_for_update(self):
         with mock.patch.dict('ckan.logic._actions', {'package_show': mock_package_show}):
             helpers.call_action('resource_patch', id=resource['id'], description='hey')
             assert mock_package_show.call_args_list[0][0][0].get('for_update') is True
+
+    def test_user_patch_sysadmin(self):
+
+        sysadmin = factories.Sysadmin()
+        user = factories.User()
+
+        # cannot change your own sysadmin privs
+        with pytest.raises(ValidationError):
+            helpers.call_action(
+                "user_patch",
+                id=sysadmin["id"],
+                sysadmin=False,
+                context={"user": sysadmin["name"]},
+            )
+
+        # cannot change system user privs
+        site_id = config.get('ckan.site_id')
+        with pytest.raises(ValidationError):
+            helpers.call_action(
+                "user_patch",
+                id=site_id,
+                sysadmin=False,
+                context={"user": sysadmin["name"]},
+            )
+
+        # user dicts do not have sysadmin key, get from db
+        user = model.User.get(user["id"])
+
+        assert user.sysadmin is False
+
+        helpers.call_action(
+                "user_patch",
+                id=user["id"],
+                sysadmin=True,
+                context={"user": sysadmin["name"]},
+            )
+
+        # user dicts do not have sysadmin key, get from db
+        new_sysadmin = model.User.get(user["id"])
+
+        assert new_sysadmin.sysadmin is True