api/admin: implement admin.vm.volume.Set.rw method

Allow setting 'rw' volume property.

Makefile

@@ -30,6 +30,7 @@ ADMIN_API_METHODS_SIMPLE = \
 	admin.pool.volume.Resize \
 	admin.pool.volume.Revert \
 	admin.pool.volume.Set.revisions_to_keep \
+	admin.pool.volume.Set.rw \
 	admin.pool.volume.Snapshot \
 	admin.property.Get \
 	admin.property.GetDefault \
@@ -100,6 +101,7 @@ ADMIN_API_METHODS_SIMPLE = \
 	admin.vm.volume.Resize \
 	admin.vm.volume.Revert \
 	admin.vm.volume.Set.revisions_to_keep \
+	admin.vm.volume.Set.rw \
 	admin.vm.Stats \
 	$(null)
 

qubes/api/admin.py

@@ -503,6 +503,26 @@ class QubesAdminAPI(qubes.api.AbstractQubesAPI):
         self.dest.volumes[self.arg].revisions_to_keep = newvalue
         self.app.save()
 
+    @qubes.api.method('admin.vm.volume.Set.rw',
+        scope='local', write=True)
+    @asyncio.coroutine
+    def vm_volume_set_rw(self, untrusted_payload):
+        assert self.arg in self.dest.volumes.keys()
+        try:
+            newvalue = qubes.property.bool(None, None,
+                untrusted_payload.decode('ascii'))
+        except (UnicodeDecodeError, ValueError):
+            raise qubes.api.ProtocolError('Invalid value')
+        del untrusted_payload
+
+        self.fire_event_for_permission(newvalue=newvalue)
+
+        if not self.dest.is_halted():
+            raise qubes.exc.QubesVMNotHaltedError(self.dest)
+
+        self.dest.volumes[self.arg].rw = newvalue
+        self.app.save()
+
     @qubes.api.method('admin.vm.tag.List', no_payload=True,
         scope='local', read=True)
     @asyncio.coroutine

qubes/tests/api_admin.py

@@ -2371,6 +2371,34 @@ class TC_00_VMs(AdminAPITestCase):
             self.call_mgmt_func(b'admin.vm.volume.Set.revisions_to_keep',
                 b'test-vm1', b'private', b'abc')
 
+    def test_680_vm_volume_set_rw(self):
+        self.vm.volumes = unittest.mock.MagicMock()
+        volumes_conf = {
+            'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
+        }
+        self.vm.volumes.configure_mock(**volumes_conf)
+        self.vm.storage = unittest.mock.Mock()
+        value = self.call_mgmt_func(b'admin.vm.volume.Set.rw',
+            b'test-vm1', b'private', b'True')
+        self.assertIsNone(value)
+        self.assertEqual(self.vm.volumes.mock_calls,
+            [unittest.mock.call.keys(),
+            ('__getitem__', ('private',), {})])
+        self.assertEqual(self.vm.volumes['private'].rw, True)
+        self.app.save.assert_called_once_with()
+
+    def test_681_vm_volume_set_rw_invalid(self):
+        self.vm.volumes = unittest.mock.MagicMock()
+        volumes_conf = {
+            'keys.return_value': ['root', 'private', 'volatile', 'kernel'],
+        }
+        self.vm.volumes.configure_mock(**volumes_conf)
+        self.vm.storage = unittest.mock.Mock()
+        with self.assertRaises(AssertionError):
+            self.call_mgmt_func(b'admin.vm.volume.Set.revisions_to_keep',
+                b'test-vm1', b'private', b'abc')
+        self.assertFalse(self.app.save.called)
+
     def test_990_vm_unexpected_payload(self):
         methods_with_no_payload = [
             b'admin.vm.List',