tests: add a test for no-revisions volume

This tests if volume works, but more importantly if all "unsafe"
operations are blocked when its VM is running:
- cloning to another volume
- using as a source volume for snapshot

The test has also commented out part about a backup - but that currently
is blocked in the backup code directly.

QubesOS/qubes-issues#8767

Author: marmarek

qubes/tests/integ/storage.py

@@ -379,6 +379,112 @@ async def _test_005_size_after_clone(self):
         await qubes.utils.coro_maybe(testvol2.import_volume(testvol))
         self.assertEqual(testvol2.size, size)
 
+    def test_006_no_revisions(self):
+        """Test if no-revisions volume persists data, and blocks parallel
+        access"""
+        return self.loop.run_until_complete(self._test_006_no_revisions())
+
+    async def _test_006_no_revisions(self):
+        self.vm1.storage.set_revisions_to_keep("private", -1)
+        dir_path = "/var/tmp/test-pool2"
+        pool2 = await self.app.add_pool(
+            dir_path=dir_path, name="test-pool2", driver="file"
+        )
+        self.addCleanup(shutil.rmtree, dir_path)
+
+        size = 32 * 1024 * 1024
+        volume_config = {
+            "pool": self.pool.name,
+            "size": size,
+            "save_on_stop": True,
+            "rw": True,
+            "revisions_to_keep": -1,
+        }
+        testvol = self.vm1.storage.init_volume("testvol", volume_config)
+        await qubes.utils.coro_maybe(testvol.create())
+        volume2_config = {
+            "pool": self.pool.name,
+            "size": size,
+            "snap_on_start": True,
+            "source": testvol.vid,
+            "rw": True,
+        }
+        testvol2 = self.vm2.storage.init_volume("testvol2", volume2_config)
+        await qubes.utils.coro_maybe(testvol2.create())
+        del testvol
+        del testvol2
+
+        volume3_config = {
+            "pool": pool2.name,
+            "size": size,
+            "save_on_stop": True,
+            "rw": True,
+        }
+        testvol3 = self.vm2.storage.init_volume("testvol3", volume3_config)
+        await qubes.utils.coro_maybe(testvol3.create())
+
+        self.app.save()
+        await self.vm1.start()
+        await self.wait_for_session(self.vm1)
+        # non-volatile image not clean
+        await self.vm1.run_for_stdio(
+            "head -c {} /dev/zero 2>&1 | diff -q /dev/xvde - 2>&1".format(size),
+            user="root",
+        )
+
+        await self.vm1.run_for_stdio("echo test123 > /dev/xvde", user="root")
+        await self.vm1.shutdown(wait=True)
+        await self.vm1.start()
+        # non-volatile image volatile
+        with self.assertRaises(subprocess.CalledProcessError):
+            await self.vm1.run_for_stdio(
+                "head -c {} /dev/zero 2>&1 | diff -q /dev/xvde - 2>&1".format(
+                    size
+                ),
+                user="root",
+            )
+
+        # should not work
+        with self.assertRaises(qubes.exc.QubesException):
+            await self.vm2.start()
+        # clone within the same pool
+        with self.assertRaises(qubes.exc.QubesException):
+            await self.vm2.storage.import_volume(
+                self.vm2.storage.get_volume("private"),
+                self.vm1.storage.get_volume("private"),
+            )
+        # clone to a different pool
+        with self.assertRaises(qubes.exc.QubesException):
+            await self.vm2.storage.import_volume(
+                self.vm2.storage.get_volume("testvol3"),
+                self.vm1.storage.get_volume("testvol"),
+            )
+        # export is not blocked explicitly now, but its uses are (cross-pool
+        # clone, backup)
+        # with self.assertRaises(qubes.exc.QubesException):
+        #    path = await self.vm1.storage.export("testvol")
+        #    # just in case it was allowed...
+        #    await self.vm1.storage.export_end("testvol", path)
+
+        # now shutdown vm1 and the above operations should work
+        await self.vm1.shutdown(wait=True)
+        await self.vm2.storage.import_volume(
+            self.vm2.storage.get_volume("private"),
+            self.vm1.storage.get_volume("private"),
+        )
+        await self.vm2.storage.import_volume(
+            self.vm2.storage.get_volume("testvol3"),
+            self.vm1.storage.get_volume("testvol"),
+        )
+        path = await self.vm1.storage.export("testvol")
+        await self.vm1.storage.export_end("testvol", path)
+
+        await self.vm2.start()
+        stdout, stderr = await self.vm2.run_for_stdio(
+            "head -c 7 /dev/xvde", user="root"
+        )
+        self.assertEqual(stdout, b"test123")
+
 
 class StorageFile(StorageTestMixin, qubes.tests.SystemTestCase):
     def init_pool(self):