glance/glance/tests/functional/v2/test_member_api_policy.py

# Copyright 2021 Red Hat, Inc.
# All Rights Reserved.
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.

from unittest import mock

import oslo_policy.policy
from oslo_utils.fixture import uuidsentinel as uuids

from glance.api import policy
from glance.tests import functional


class TestImageMembersPolicy(functional.SynchronousAPIBase):
    def setUp(self):
        super(TestImageMembersPolicy, self).setUp()
        self.policy = policy.Enforcer(suppress_deprecation_warnings=True)

    def load_data(self, share_image=False):
        output = {}
        path = "/v2/images"
        data = {
            'name': 'shared-image',
            'visibility': 'shared',
        }
        response = self.api_post(path, json=data)
        self.assertEqual(201, response.status_code)
        image_id = response.json['id']
        output['image_id'] = image_id

        if share_image:
            path = '/v2/images/%s/members' % image_id
            data = {
                'member': uuids.random_member
            }
            response = self.api_post(path, json=data)
            member = response.json
            self.assertEqual(200, response.status_code)
            self.assertEqual(image_id, member['image_id'])
            self.assertEqual('pending', member['status'])
            output['member_id'] = member['member_id']

        return output

    def set_policy_rules(self, rules):
        self.policy.set_rules(
            oslo_policy.policy.Rules.from_dict(rules),
            overwrite=True)

    def start_server(self):
        with mock.patch.object(policy, 'Enforcer') as mock_enf:
            mock_enf.return_value = self.policy
            super(TestImageMembersPolicy, self).start_server()

    def test_member_add_basic(self):
        self.start_server()
        output = self.load_data()
        path = '/v2/images/%s/members' % output['image_id']

        # Make sure we can add member to image (can share image)
        data = {
            'member': uuids.random_member
        }
        response = self.api_post(path, json=data)
        self.assertEqual(200, response.status_code)
        member = response.json
        self.assertEqual(output['image_id'], member['image_id'])
        self.assertEqual('pending', member['status'])

        # Now disable add permissions
        self.set_policy_rules({
            'add_member': '!',
            'get_image': '@'
        })

        # Make sure add returns 403
        response = self.api_post(path, json=data)
        self.assertEqual(403, response.status_code)

        # Now disable both permissions and make sure you will get
        # 404 Not Found
        self.set_policy_rules({
            'add_member': '!',
            'get_image': '!'
        })

        # Make sure add returns 404
        response = self.api_post(path, json=data)
        self.assertEqual(404, response.status_code)

    def test_member_update_basic(self):
        self.start_server()
        output = self.load_data(share_image=True)
        path = '/v2/images/%s/members/%s' % (output['image_id'],
                                             output['member_id'])
        # Make sure we can update image membership
        data = {
            'status': 'accepted'
        }
        response = self.api_put(path, json=data)
        self.assertEqual(200, response.status_code)
        member = response.json
        self.assertEqual(output['image_id'], member['image_id'])
        self.assertEqual('accepted', member['status'])

        # Now disable modify permissions
        self.set_policy_rules({
            'modify_member': '!',
            'get_image': '@'
        })

        # Make sure update returns 403
        response = self.api_put(path, json=data)
        self.assertEqual(403, response.status_code)

        # Now disable both permissions and make sure you will get
        # 404 Not Found
        self.set_policy_rules({
            'modify_member': '!',
            'get_image': '!',
            'get_member': '@'
        })
        # image owner is not allowed to update image membership so
        # passing different project in headers
        headers = self._headers({
            'X-Tenant-Id': 'fake-tenant-id',
        })
        response = self.api_put(path, headers=headers, json=data)
        self.assertEqual(404, response.status_code)

    def test_member_list_basic(self):
        self.start_server()
        output = self.load_data(share_image=True)
        path = '/v2/images/%s/members' % (output['image_id'])
        # # Make sure we can list image members
        response = self.api_get(path)
        self.assertEqual(200, response.status_code)
        self.assertEqual(1, len(response.json['members']))

        # Now disable list permissions
        self.set_policy_rules({
            'get_members': '!',
            'get_image': '@',
        })

        # Make sure get returns 403
        response = self.api_get(path)
        self.assertEqual(403, response.status_code)

        # Now disable both permissions and make sure you will get
        # 404 Not Found
        self.set_policy_rules({
            'get_members': '!',
            'get_image': '!',
        })

        # Make sure get returns 404
        response = self.api_get(path)
        self.assertEqual(404, response.status_code)

        # Now enable get_members and disable get_member
        self.set_policy_rules({
            'get_members': '@',
            'get_member': '!',
            'get_image': '@',
        })

        # Make sure we get empty list as get_member is disabled
        response = self.api_get(path)
        self.assertEqual(200, response.status_code)
        self.assertEqual(0, len(response.json['members']))

    def test_member_get_basic(self):
        self.start_server()
        output = self.load_data(share_image=True)
        path = '/v2/images/%s/members/%s' % (
            output['image_id'],
            output['member_id'])
        # Make sure we can get member
        response = self.api_get(path)
        self.assertEqual(200, response.status_code)
        member = response.json
        self.assertEqual(output['image_id'], member['image_id'])
        self.assertEqual('pending', member['status'])

        # Now disable get permissions
        self.set_policy_rules({'get_member': '!'})

        # Make sure get returns 404 as we are not exposing it
        response = self.api_get(path)
        self.assertEqual(404, response.status_code)

    def test_member_delete_basic(self):
        self.start_server()
        output = self.load_data(share_image=True)
        path = '/v2/images/%s/members/%s' % (output['image_id'],
                                             output['member_id'])
        # Make sure we can delete image member
        response = self.api_delete(path)
        self.assertEqual(204, response.status_code)
        # Verifiy it is deleted
        response = self.api_get(path)
        self.assertEqual(404, response.status_code)

        # Now disable delete permissions and create image
        # membership again
        self.set_policy_rules({
            'delete_member': '!',
            'add_member': '@',
            'get_image': '@'
        })
        add_path = '/v2/images/%s/members' % output['image_id']
        data = {
            'member': uuids.random_member
        }
        response = self.api_post(add_path, json=data)
        self.assertEqual(200, response.status_code)

        # Make sure delete returns 403
        response = self.api_delete(path)
        self.assertEqual(403, response.status_code)

        # Now disable delete_member and get_image permission and make
        # sure you will get 404 Not Found
        self.set_policy_rules({
            'delete_member': '!',
            'get_image': '!',
            'get_member': '@'
        })
        response = self.api_delete(path)
        self.assertEqual(404, response.status_code)

    def test_image_sharing_not_allowed(self):
        # This test verifies that image having visibility other than
        # shared is not allowed for sharing
        self.start_server()
        path = "/v2/images"
        for visibility in ('community', 'private', 'public'):
            data = {
                'name': '%s-image' % visibility,
                'visibility': visibility,
            }
            # create image
            response = self.api_post(path, json=data)
            image = response.json
            self.assertEqual(201, response.status_code)
            self.assertEqual(visibility, image['visibility'])

            # Sharing image should return 403 response
            member_path = '/v2/images/%s/members' % image['id']
            data = {
                'member': uuids.random_member
            }
            response = self.api_post(member_path, json=data)
            self.assertEqual(403, response.status_code)