Use utility function for size validation

This allows it to be used by other code in the future.

qubes/api/__init__.py

@@ -210,6 +210,15 @@ def enforce(predicate):
         if not predicate:
             raise PermissionDenied()
 
+    def validate_size(self, untrusted_size: bytes) -> int:
+        self.enforce(isinstance(untrusted_size, bytes))
+        if not untrusted_size.isdigit():
+            raise qubes.api.ProtocolError('Size must be ASCII digits (only)')
+        if len(untrusted_size) >= 20:
+            raise qubes.api.ProtocolError('Sizes limited to 19 decimal digits')
+        if untrusted_size[0] == 48 and untrusted_size != b'0':
+            raise qubes.api.ProtocolError('Spurious leading zeros not allowed')
+        return int(untrusted_size)
 
 class QubesDaemonProtocol(asyncio.Protocol):
     buffer_size = 65536

qubes/api/admin.py

@@ -450,15 +450,8 @@ async def vm_volume_clone_to(self, untrusted_payload):
         scope='local', write=True)
     async def vm_volume_resize(self, untrusted_payload):
         self.enforce(self.arg in self.dest.volumes.keys())
-        untrusted_size = untrusted_payload.decode('ascii').strip()
-        del untrusted_payload
-        self.enforce(untrusted_size.isdigit())   # only digits, forbid '-' too
-        self.enforce(len(untrusted_size) <= 20)  # limit to about 2^64
-
-        size = int(untrusted_size)
-
+        size = self.validate_size(untrusted_payload.strip())
         self.fire_event_for_permission(size=size)
-
         try:
             await self.dest.storage.resize(self.arg, size)
         finally:  # even if calling qubes.ResizeDisk inside the VM failed
@@ -493,17 +486,7 @@ async def vm_volume_clear(self):
         scope='local', write=True)
     async def vm_volume_set_revisions_to_keep(self, untrusted_payload):
         self.enforce(self.arg in self.dest.volumes.keys())
-        untrusted_payload = untrusted_payload.strip()
-        if not untrusted_payload.isdigit():
-            raise qubes.api.ProtocolError('Invalid value')
-        try:
-            untrusted_value = int(untrusted_payload.decode('ascii', 'strict'))
-        except (UnicodeDecodeError, ValueError):
-            raise qubes.api.ProtocolError('Invalid value')
-        del untrusted_payload
-        self.enforce(untrusted_value >= 0)
-        newvalue = untrusted_value
-        del untrusted_value
+        newvalue = self.validate_size(untrusted_payload)
 
         self.fire_event_for_permission(newvalue=newvalue)
 
@@ -749,14 +732,7 @@ async def pool_set_revisions_to_keep(self, untrusted_payload):
         self.enforce(self.dest.name == 'dom0')
         self.enforce(self.arg in self.app.pools.keys())
         pool = self.app.pools[self.arg]
-        untrusted_payload = untrusted_payload.strip()
-        if not untrusted_payload.isdigit() or len(untrusted_payload) > 20:
-            raise qubes.api.ProtocolError('Invalid value')
-        untrusted_value = int(untrusted_payload)
-        del untrusted_payload
-        self.enforce(untrusted_value >= 0)
-        newvalue = untrusted_value
-        del untrusted_value
+        newvalue = self.validate_size(untrusted_payload)
 
         self.fire_event_for_permission(newvalue=newvalue)
 

qubes/api/internal.py

@@ -91,14 +91,8 @@ async def vm_volume_import(self, untrusted_payload):
         if not self.dest.is_halted():
             raise qubes.exc.QubesVMNotHaltedError(self.dest)
 
-        requested_size = None
-        if untrusted_payload:
-            if not untrusted_payload.isdigit() or len(untrusted_payload) >= 21:
-                raise qubes.api.ProtocolError('Invalid value')
-            untrusted_value = int(untrusted_payload)
-            self.enforce(0 < untrusted_value < (1 << 64))
-            requested_size = untrusted_value
-            del untrusted_value
+        requested_size = (self.validate_size(untrusted_payload)
+                          if untrusted_payload else None)
         del untrusted_payload
 
         path = await self.dest.storage.import_data(

qubes/tests/api_admin.py

@@ -606,6 +606,22 @@ def test_120_vm_volume_resize(self):
         self.assertEqual(self.vm.storage.mock_calls,
             [unittest.mock.call.resize('private', 1024000000)])
 
+    def test_120_vm_volume_resize_zero(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()
+        self.vm.storage.resize.side_effect = self.dummy_coro
+        value = self.call_mgmt_func(b'admin.vm.volume.Resize',
+            b'test-vm1', b'private', b'0')
+        self.assertIsNone(value)
+        self.assertEqual(self.vm.volumes.mock_calls,
+            [unittest.mock.call.keys()])
+        self.assertEqual(self.vm.storage.mock_calls,
+            [unittest.mock.call.resize('private', 0)])
+
     def test_120_vm_volume_resize_invalid_size1(self):
         self.vm.volumes = unittest.mock.MagicMock()
         volumes_conf = {
@@ -614,7 +630,7 @@ def test_120_vm_volume_resize_invalid_size1(self):
         self.vm.volumes.configure_mock(**volumes_conf)
         self.vm.storage = unittest.mock.Mock()
         self.vm.storage.resize.side_effect = self.dummy_coro
-        with self.assertRaises(qubes.api.PermissionDenied):
+        with self.assertRaises(qubes.api.ProtocolError):
             self.call_mgmt_func(b'admin.vm.volume.Resize',
                 b'test-vm1', b'private', b'no-int-size')
         self.assertEqual(self.vm.volumes.mock_calls,
@@ -629,13 +645,43 @@ def test_120_vm_volume_resize_invalid_size2(self):
         self.vm.volumes.configure_mock(**volumes_conf)
         self.vm.storage = unittest.mock.Mock()
         self.vm.storage.resize.side_effect = self.dummy_coro
-        with self.assertRaises(qubes.api.PermissionDenied):
+        with self.assertRaises(qubes.api.ProtocolError):
             self.call_mgmt_func(b'admin.vm.volume.Resize',
                 b'test-vm1', b'private', b'-1')
         self.assertEqual(self.vm.volumes.mock_calls,
             [unittest.mock.call.keys()])
         self.assertFalse(self.vm.storage.called)
 
+    def test_120_vm_volume_resize_invalid_size3(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()
+        self.vm.storage.resize.side_effect = self.dummy_coro
+        with self.assertRaises(qubes.api.ProtocolError):
+            self.call_mgmt_func(b'admin.vm.volume.Resize',
+                b'test-vm1', b'private', b'10000000000000000000')
+        self.assertEqual(self.vm.volumes.mock_calls,
+            [unittest.mock.call.keys()])
+        self.assertFalse(self.vm.storage.called)
+
+    def test_120_vm_volume_resize_invalid_size4(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()
+        self.vm.storage.resize.side_effect = self.dummy_coro
+        with self.assertRaises(qubes.api.ProtocolError):
+            self.call_mgmt_func(b'admin.vm.volume.Resize',
+                b'test-vm1', b'private', b'01')
+        self.assertEqual(self.vm.volumes.mock_calls,
+            [unittest.mock.call.keys()])
+        self.assertFalse(self.vm.storage.called)
+
     def test_130_pool_list(self):
         self.app.pools = ['file', 'lvm']
         value = self.call_mgmt_func(b'admin.pool.List', b'dom0')