Add an API for editing a group membership's role

h/presenters/__init__.py

@@ -5,6 +5,7 @@
 from h.presenters.document_json import DocumentJSONPresenter
 from h.presenters.document_searchindex import DocumentSearchIndexPresenter
 from h.presenters.group_json import GroupJSONPresenter, GroupsJSONPresenter
+from h.presenters.group_membership_json import GroupMembershipJSONPresenter
 from h.presenters.user_json import TrustedUserJSONPresenter, UserJSONPresenter
 
 __all__ = (
@@ -17,5 +18,6 @@
     "GroupJSONPresenter",
     "GroupsJSONPresenter",
     "UserJSONPresenter",
+    "GroupMembershipJSONPresenter",
     "TrustedUserJSONPresenter",
 )

h/presenters/group_membership_json.py

@@ -0,0 +1,12 @@
+class GroupMembershipJSONPresenter:
+    def __init__(self, membership):
+        self.membership = membership
+
+    def asdict(self):
+        return {
+            "authority": self.membership.group.authority,
+            "userid": self.membership.user.userid,
+            "username": self.membership.user.username,
+            "display_name": self.membership.user.display_name,
+            "roles": self.membership.roles,
+        }

h/schemas/api/group_membership.py

@@ -0,0 +1,18 @@
+from h.schemas.base import JSONSchema
+
+
+class EditGroupMembershipAPISchema(JSONSchema):
+    schema = {
+        "type": "object",
+        "properties": {
+            "roles": {
+                "type": "array",
+                "minItems": 1,
+                "maxItems": 1,
+                "items": {
+                    "enum": ["member", "moderator", "admin", "owner"],
+                },
+            }
+        },
+        "required": ["roles"],
+    }

h/security/permission_map.py

@@ -56,6 +56,7 @@
     ],
     Permission.Group.MEMBER_ADD: [[p.group_matches_authenticated_client_authority]],
     Permission.Group.MEMBER_REMOVE: [[p.group_member_remove]],
+    Permission.Group.MEMBER_EDIT: [[p.group_member_edit]],
     Permission.Group.MODERATE: [[p.group_created_by_user]],
     # --------------------------------------------------------------------- #
     # Annotations

h/security/permissions.py

@@ -43,6 +43,9 @@ class Group(Enum):
         MEMBER_REMOVE = "group:member:remove"
         """Remove a member from the group."""
 
+        MEMBER_EDIT = "group:member:edit"
+        """Change a member's role in a group."""
+
     class Annotation(Enum):
         """Permissions relating to annotations."""
 

h/security/predicates.py

@@ -13,7 +13,7 @@
 from itertools import chain
 
 from h.models.group import GroupMembershipRoles, JoinableBy, ReadableBy, WriteableBy
-from h.traversal import GroupMembershipContext
+from h.traversal import EditGroupMembershipContext, GroupMembershipContext
 
 
 def requires(*parent_predicates):
@@ -219,6 +219,62 @@ def get_authenticated_users_membership():
     )
 
 
+@requires(authenticated_user, group_found)
+def group_member_edit(
+    identity, context: EditGroupMembershipContext
+):  # pylint:disable=too-many-return-statements,too-complex
+    old_roles = context.membership.roles
+    new_roles = context.new_roles
+
+    def get_authenticated_users_roles():
+        """Return the authenticated users roles in the target group."""
+        for membership in identity.user.memberships:
+            if membership.group.id == context.group.id:
+                return membership.roles
+
+        return None
+
+    authenticated_users_roles = get_authenticated_users_roles()
+
+    if not authenticated_users_roles:
+        return False
+
+    if identity.user.userid == context.user.userid:
+        if GroupMembershipRoles.OWNER in authenticated_users_roles:
+            # Owners can change their own role to anything.
+            return True
+
+        if GroupMembershipRoles.ADMIN in authenticated_users_roles:
+            # Admins can change their own role to anything but admin.
+            return GroupMembershipRoles.OWNER not in new_roles
+
+        if GroupMembershipRoles.MODERATOR in authenticated_users_roles:
+            # Moderators can change their own role to anything but owner or admin.
+            return (
+                GroupMembershipRoles.OWNER not in new_roles
+                and GroupMembershipRoles.ADMIN not in new_roles
+            )
+
+        return False
+
+    if GroupMembershipRoles.OWNER in authenticated_users_roles:
+        # Owners can change any other member's role to any role.
+        return True
+
+    if GroupMembershipRoles.ADMIN in authenticated_users_roles:
+        # Admins can change the role of anyone but owners or admins to anything
+        # but owner or admin.
+        if (
+            GroupMembershipRoles.OWNER in old_roles + new_roles
+            or GroupMembershipRoles.ADMIN in old_roles + new_roles
+        ):
+            return False
+
+        return True
+
+    return False
+
+
 def resolve_predicates(mapping):
     """
     Expand predicates with requirements into concrete lists of predicates.

h/traversal/__init__.py

@@ -63,6 +63,7 @@
 from h.traversal.annotation import AnnotationContext, AnnotationRoot
 from h.traversal.group import GroupContext, GroupRequiredRoot, GroupRoot
 from h.traversal.group_membership import (
+    EditGroupMembershipContext,
     GroupMembershipContext,
     group_membership_api_factory,
 )
@@ -81,6 +82,7 @@
     "UserByIDRoot",
     "UserRoot",
     "GroupContext",
+    "EditGroupMembershipContext",
     "GroupMembershipContext",
     "group_membership_api_factory",
 )

h/traversal/group_membership.py

@@ -3,7 +3,7 @@
 from pyramid.httpexceptions import HTTPNotFound
 
 from h.exceptions import InvalidUserId
-from h.models import Group, GroupMembership, User
+from h.models import Group, GroupMembership, GroupMembershipRoles, User
 
 
 @dataclass
@@ -13,6 +13,14 @@ class GroupMembershipContext:
     membership: GroupMembership | None
 
 
+@dataclass
+class EditGroupMembershipContext:
+    group: Group
+    user: User
+    membership: GroupMembership
+    new_roles: list[GroupMembershipRoles]
+
+
 def group_membership_api_factory(request) -> GroupMembershipContext:
     user_service = request.find_service(name="user")
     group_service = request.find_service(name="group")

h/views/api/annotations.py

@@ -29,7 +29,7 @@
 from h.security import Permission
 from h.services import AnnotationWriteService
 from h.views.api.config import api_config
-from h.views.api.exceptions import PayloadError
+from h.views.api.helpers.json_payload import json_payload
 
 _ = i18n.TranslationStringFactory(__package__)
 
@@ -77,7 +77,7 @@ def search(request):
 def create(request):
     """Create an annotation from the POST payload."""
     schema = CreateAnnotationSchema(request)
-    appstruct = schema.validate(_json_payload(request))
+    appstruct = schema.validate(json_payload(request))
 
     annotation = request.find_service(AnnotationWriteService).create_annotation(
         data=appstruct
@@ -136,7 +136,7 @@ def update(context, request):
     schema = UpdateAnnotationSchema(
         request, context.annotation.target_uri, context.annotation.groupid
     )
-    appstruct = schema.validate(_json_payload(request))
+    appstruct = schema.validate(json_payload(request))
 
     annotation = request.find_service(AnnotationWriteService).update_annotation(
         context.annotation, data=appstruct
@@ -166,18 +166,6 @@ def delete(context, request):
     return {"id": context.annotation.id, "deleted": True}
 
 
-def _json_payload(request):
-    """
-    Return a parsed JSON payload for the request.
-
-    :raises PayloadError: if the body has no valid JSON body
-    """
-    try:
-        return request.json_body
-    except ValueError as err:
-        raise PayloadError() from err
-
-
 def _publish_annotation_event(request, annotation, action):
     """Publish an event to the annotations queue for this annotation action."""
     event = AnnotationEvent(request, annotation.id, action)

h/views/api/group_members.py

@@ -1,9 +1,15 @@
+import logging
+
 from pyramid.httpexceptions import HTTPNoContent, HTTPNotFound
 
-from h.presenters import UserJSONPresenter
+from h.presenters import GroupMembershipJSONPresenter
+from h.schemas.api.group_membership import EditGroupMembershipAPISchema
 from h.security import Permission
-from h.traversal import GroupContext, GroupMembershipContext
+from h.traversal import EditGroupMembershipContext, GroupContext, GroupMembershipContext
 from h.views.api.config import api_config
+from h.views.api.helpers.json_payload import json_payload
+
+log = logging.getLogger(__name__)
 
 
 @api_config(
@@ -15,8 +21,10 @@
     permission=Permission.Group.READ,
 )
 def list_members(context: GroupContext, _request):
-    """Return a list of a group's members."""
-    return [UserJSONPresenter(user).asdict() for user in context.group.members]
+    return [
+        GroupMembershipJSONPresenter(membership).asdict()
+        for membership in context.group.memberships
+    ]
 
 
 @api_config(
@@ -28,7 +36,6 @@ def list_members(context: GroupContext, _request):
     permission=Permission.Group.MEMBER_REMOVE,
 )
 def remove_member(context: GroupMembershipContext, request):
-    """Remove a member from a group."""
     group_members_service = request.find_service(name="group_members")
 
     group_members_service.member_leave(context.group, context.user.userid)
@@ -45,7 +52,6 @@ def remove_member(context: GroupMembershipContext, request):
     permission=Permission.Group.MEMBER_ADD,
 )
 def add_member(context: GroupMembershipContext, request):
-    """Add a member to a group."""
     group_members_service = request.find_service(name="group_members")
 
     if context.user.authority != context.group.authority:
@@ -54,3 +60,34 @@ def add_member(context: GroupMembershipContext, request):
     group_members_service.member_join(context.group, context.user.userid)
 
     return HTTPNoContent()
+
+
+@api_config(
+    versions=["v1", "v2"],
+    route_name="api.group_member",
+    request_method="PATCH",
+    link_name="group.member.edit",
+    description="Change a user's role in a group",
+)
+def edit_member(context: GroupMembershipContext, request):
+    appstruct = EditGroupMembershipAPISchema().validate(json_payload(request))
+    new_roles = appstruct["roles"]
+
+    if not request.has_permission(
+        Permission.Group.MEMBER_EDIT,
+        EditGroupMembershipContext(
+            context.group, context.user, context.membership, new_roles
+        ),
+    ):
+        raise HTTPNotFound()
+
+    if context.membership.roles != new_roles:
+        old_roles = context.membership.roles
+        context.membership.roles = new_roles
+        log.info(
+            "Changed group membership roles: %r (previous roles were: %r)",
+            context.membership,
+            old_roles,
+        )
+
+    return GroupMembershipJSONPresenter(context.membership).asdict()

h/views/api/groups.py

@@ -6,7 +6,7 @@
 from h.security import Permission
 from h.traversal import GroupContext
 from h.views.api.config import api_config
-from h.views.api.exceptions import PayloadError
+from h.views.api.helpers.json_payload import json_payload
 
 DEFAULT_GROUP_TYPE = "private"
 
@@ -47,7 +47,7 @@ def create(request):
     appstruct = CreateGroupAPISchema(
         default_authority=request.default_authority,
         group_authority=request.effective_authority,
-    ).validate(_json_payload(request))
+    ).validate(json_payload(request))
 
     group_service = request.find_service(name="group")
     group_create_service = request.find_service(name="group_create")
@@ -116,7 +116,7 @@ def update(context: GroupContext, request):
     appstruct = UpdateGroupAPISchema(
         default_authority=request.default_authority,
         group_authority=request.effective_authority,
-    ).validate(_json_payload(request))
+    ).validate(json_payload(request))
 
     group_update_service = request.find_service(name="group_update")
     group_service = request.find_service(name="group")
@@ -133,16 +133,3 @@ def update(context: GroupContext, request):
     group = group_update_service.update(context.group, **appstruct)
 
     return GroupJSONPresenter(group, request).asdict(expand=["organization", "scopes"])
-
-
-# @TODO This is a duplication of code in h.views.api — move to a util module
-def _json_payload(request):
-    """
-    Return a parsed JSON payload for the request.
-
-    :raises PayloadError: if the body has no valid JSON body
-    """
-    try:
-        return request.json_body
-    except ValueError as err:
-        raise PayloadError() from err

h/views/api/helpers/json_payload.py

@@ -0,0 +1,13 @@
+from h.views.api.exceptions import PayloadError
+
+
+def json_payload(request):
+    """
+    Return the parsed JSON payload from `request`.
+
+    :raise PayloadError: if the request has no valid JSON body
+    """
+    try:
+        return request.json_body
+    except ValueError as err:
+        raise PayloadError() from err

h/views/api/users.py

@@ -6,7 +6,7 @@
 from h.security import Permission
 from h.services.user_unique import DuplicateUserError
 from h.views.api.config import api_config
-from h.views.api.exceptions import PayloadError
+from h.views.api.helpers.json_payload import json_payload
 
 
 @api_config(
@@ -53,7 +53,7 @@ def create(request):
                              authority
     :raises HTTPConflict:    if user already exists
     """
-    appstruct = CreateUserAPISchema().validate(_json_payload(request))
+    appstruct = CreateUserAPISchema().validate(json_payload(request))
 
     # Enforce authority match
     client_authority = request.identity.auth_client.authority
@@ -90,15 +90,8 @@ def update(context, request):
     This API endpoint allows authorised clients (those able to provide a valid
     Client ID and Client Secret) to update users in their authority.
     """
-    appstruct = UpdateUserAPISchema().validate(_json_payload(request))
+    appstruct = UpdateUserAPISchema().validate(json_payload(request))
 
     user = request.find_service(name="user_update").update(context.user, **appstruct)
 
     return TrustedUserJSONPresenter(user).asdict()
-
-
-def _json_payload(request):
-    try:
-        return request.json_body
-    except ValueError as err:
-        raise PayloadError() from err