Throw BackupAlreadyRunningError when backup is already running

Instead of generic PermissionDenied.

requires https://github.com/QubesOS/qubes-core-admin-client/pull/115
references QubesOS/qubes-issues#5432

qubes/api/admin.py

@@ -1507,7 +1507,8 @@ class QubesAdminAPI(qubes.api.AbstractQubesAPI):
             self._backup_progress_callback, self.arg)
 
         # forbid running the same backup operation twice at the time
-        self.enforce(self.arg not in self.app.api_admin_running_backups)
+        if self.arg in self.app.api_admin_running_backups:
+            raise qubes.exc.BackupAlreadyRunningError()
 
         backup_task = asyncio.ensure_future(backup.backup_do())
         self.app.api_admin_running_backups[self.arg] = backup_task

qubes/exc.py

@@ -164,6 +164,14 @@ class BackupCancelledError(QubesException):
             msg or 'Backup cancelled')
 
 
+class BackupAlreadyRunningError(QubesException):
+    '''Thrown at user when they try to run the same backup twice at
+    the same time'''
+    def __init__(self, msg=None):
+        super(BackupAlreadyRunningError, self).__init__(
+            msg or 'Backup already running')
+
+
 class QubesMemoryError(QubesVMError, MemoryError):
     '''Cannot start domain, because not enough memory is available'''
     def __init__(self, vm, msg=None):

qubes/tests/api_admin.py

@@ -2072,6 +2072,31 @@ class TC_00_VMs(AdminAPITestCase):
             self.call_mgmt_func(b'admin.backup.Cancel', b'dom0',
                 b'testprofile')
 
+    def test_611_backup_already_running(self):
+        if not hasattr(self.app, 'api_admin_running_backups'):
+            self.app.api_admin_running_backups = {}
+
+        self.app.api_admin_running_backups['testprofile'] = 'test'
+        self.addCleanup(self.app.api_admin_running_backups.pop, 'testprofile')
+
+        backup_profile = (
+            'include:\n'
+            ' - test-vm1\n'
+            'destination_vm: test-vm1\n'
+            'destination_path: /home/user\n'
+            'passphrase_text: test\n'
+        )
+
+        with tempfile.TemporaryDirectory() as profile_dir:
+            with open(os.path.join(profile_dir, 'testprofile.conf'), 'w') as \
+                    profile_file:
+                profile_file.write(backup_profile)
+            with unittest.mock.patch('qubes.config.backup_profile_dir',
+                                     profile_dir):
+                with self.assertRaises(qubes.exc.BackupAlreadyRunningError):
+                    self.call_mgmt_func(b'admin.backup.Execute', b'dom0',
+                                        b'testprofile')
+
     @unittest.mock.patch('qubes.backup.Backup')
     def test_620_backup_execute(self, mock_backup):
         backup_profile = (