Add permission check when creating restore

Currently, general users have rights to restore other users'
checkpoints. We should add permission check to avoid the security
risk.
Closes-Bug: #1805004

Change-Id: If0f957a3aa8f25778833d7611342fab6b8efa388

karbor/api/v1/restores.py

@@ -232,6 +232,8 @@ def create(self, req, body):
         # call restore rpc API of protection service
         try:
             self.protection_api.restore(context, restoreobj, restore_auth)
+        except exception.AccessCheckpointNotAllowed as error:
+            raise exc.HTTPForbidden(explanation=error.msg)
         except Exception:
             # update the status of restore
             update_dict = {

karbor/services/protection/manager.py

@@ -182,7 +182,8 @@ def copy(self, context, plan):
                                    exception.CheckpointNotFound,
                                    exception.CheckpointNotAvailable,
                                    exception.FlowError,
-                                   exception.InvalidInput)
+                                   exception.InvalidInput,
+                                   exception.AccessCheckpointNotAllowed)
     def restore(self, context, restore, restore_auth):
         LOG.info("Starting restore service:restore action")
 
@@ -197,6 +198,11 @@ def restore(self, context, restore, restore_auth):
         checkpoint_collection = provider.get_checkpoint_collection()
         checkpoint = checkpoint_collection.get(checkpoint_id)
 
+        if not context.is_admin and (
+                checkpoint.project_id != context.project_id):
+            raise exception.AccessCheckpointNotAllowed(
+                checkpoint_id=checkpoint_id)
+
         if checkpoint.status != constants.CHECKPOINT_STATUS_AVAILABLE:
             raise exception.CheckpointNotAvailable(
                 checkpoint_id=checkpoint_id)

karbor/tests/unit/api/v1/test_restores.py

@@ -75,6 +75,16 @@ def test_restore_create_Invalidcheckpoint_id(self):
         self.assertRaises(exception.ValidationError, self.controller.create,
                           req, body=body)
 
+    @mock.patch('karbor.services.protection.api.API.restore')
+    def test_restore_create_with_checkpoint_not_allowed_exception(
+            self, mock_restore):
+        mock_restore.side_effect = exception.AccessCheckpointNotAllowed
+        restore = self._restore_in_request_body()
+        body = {"restore": restore}
+        req = fakes.HTTPRequest.blank('/v1/restores')
+        self.assertRaises(exc.HTTPForbidden, self.controller.create,
+                          req, body=body)
+
     @mock.patch(
         'karbor.api.v1.restores.RestoresController._get_all')
     def test_restore_list_detail(self, moak_get_all):

karbor/tests/unit/protection/test_manager.py

@@ -146,6 +146,20 @@ def test_protect_in_error(self, mock_flow):
                           None,
                           fakes.fake_protection_plan())
 
+    @mock.patch.object(provider.ProviderRegistry, 'show_provider')
+    def test_restore_with_project_id_not_same(self, mock_provider):
+        mock_provider.return_value = fakes.FakeProvider()
+        context = mock.MagicMock(project_id='fake_project_id_1',
+                                 is_admin=False)
+        fake_restore = {
+            'checkpoint_id': 'fake_checkpoint',
+            'provider_id': 'fake_provider_id',
+            'parameters': None
+        }
+        self.assertRaises(
+            oslo_messaging.ExpectedException, self.pro_manager.restore,
+            context, fake_restore, None)
+
     @mock.patch.object(provider.ProviderRegistry, 'show_provider')
     def test_list_checkpoints(self, mock_provider):
         fake_provider = fakes.FakeProvider()