def create_editors_reviewers_groups()

in kinto-remote-settings/src/kinto_remote_settings/signer/listeners.py [0:0]

• 50 lines of code
• 9 McCabe index (conditional complexity)

def create_editors_reviewers_groups(event, resources, editors_group, reviewers_group):
    if event.request.prefixed_userid == PLUGIN_USERID:
        return

    bid = event.payload["bucket_id"]
    bucket_uri = instance_uri(event.request, "bucket", id=bid)

    current_user_id = event.request.prefixed_userid
    principals = event.request.prefixed_principals

    authz = event.request.registry.getUtility(IAuthorizationPolicy)

    for impacted in event.impacted_objects:
        new_collection = impacted["new"]

        # Skip if collection is not configured for review.
        resource, _ = pick_resource_and_signer(
            event.request,
            resources,
            bucket_id=event.payload["bucket_id"],
            collection_id=new_collection["id"],
        )
        if resource is None:
            continue

        source_collection = resource["source"]["collection"]
        _editors_group = editors_group.format(collection_id=source_collection)
        _reviewers_group = reviewers_group.format(collection_id=source_collection)

        required_perms = authz.get_bound_permissions(bucket_uri, "group:create")
        permission = event.request.registry.permission
        if not permission.check_permission(principals, required_perms):
            return

        group_perms = {"write": [current_user_id]}
        for group, members in (
            (_editors_group, [current_user_id]),
            (_reviewers_group, []),
        ):
            ensure_resource_exists(
                request=event.request,
                resource_name="group",
                parent_id=bucket_uri,
                obj={"id": group, "members": members},
                permissions=group_perms,
                matchdict={"bucket_id": bid, "id": group},
            )

        # Allow those groups to write to the source collection.
        permission = event.request.registry.permission
        collection_uri = instance_uri(
            event.request,
            "collection",
            bucket_id=bid,
            id=resource["source"]["collection"],
        )
        for group in (_editors_group, _reviewers_group):
            group_principal = instance_uri(
                event.request, "group", bucket_id=bid, id=group
            )
            permission.add_principal_to_ace(collection_uri, "write", group_principal)