Merge pull request #1470 from mwhudson/tpm-partitioning

do partitioning based on information from gadget

scripts/runtests.sh

@@ -174,11 +174,7 @@ for answers in examples/answers*.yaml; do
             --bootloader uefi \
             --snaps-from-examples \
             --source-catalog $catalog
-        if [ "$answers" = examples/answers-tpm.yaml ]; then
+        validate install
-            validate skip
-        else
-            validate install
-        fi
         grep -q 'finish: subiquity/Install/install/postinstall/run_unattended_upgrades: SUCCESS: downloading and installing security updates' $tmpdir/subiquity-server-debug.log
     else
         # The OOBE doesn't exist in WSL < 20.04

subiquity/client/controllers/filesystem.py

@@ -26,7 +26,6 @@ from subiquity.common.filesystem.manipulator import FilesystemManipulator
 from subiquity.common.types import (
     ProbeStatus,
     StorageEncryption,
-    StorageEncryptionSupport,
     )
 from subiquity.models.filesystem import (
     Bootloader,
@@ -38,7 +37,7 @@ from subiquity.ui.views import (
     GuidedDiskSelectionView,
     )
 from subiquity.ui.views.filesystem.probing import (
-    DefectiveEncryptionError,
+    CoreBootClassicError,
     SlowProbing,
     ProbingFailed,
     )
@@ -92,12 +91,12 @@ class FilesystemController(SubiquityTuiController, FilesystemManipulator):
                       self.ui.body)
 
     def make_guided_ui(self, status):
-        se = self.storage_encryption = status.storage_encryption
+        if status.core_boot_classic_error != '':
-        if se is not None and se.support == StorageEncryptionSupport.DEFECTIVE:
+            return CoreBootClassicError(self, status.core_boot_classic_error)
-            return DefectiveEncryptionError(self)
         if status.status == ProbeStatus.FAILED:
             self.app.show_error_report(status.error_report)
             return ProbingFailed(self, status.error_report)
+        self.storage_encryption = status.storage_encryption
         if status.error_report:
             self.app.show_error_report(status.error_report)
         return GuidedDiskSelectionView(self, status.disks)

subiquity/common/types.py

@@ -358,6 +358,7 @@ class GuidedStorageResponse:
     status: ProbeStatus
     error_report: Optional[ErrorReportRef] = None
     disks: Optional[List[Disk]] = None
+    core_boot_classic_error: str = ''
     storage_encryption: Optional[StorageEncryption] = None
 
 

subiquity/server/controllers/filesystem.py

@@ -23,6 +23,8 @@ import platform
 import select
 from typing import List
 
+from curtin.storage_config import ptable_uuid_to_flag_entry
+
 import pyudev
 
 from subiquitycore.async_helpers import (
@@ -66,6 +68,7 @@ from subiquity.common.types import (
     ModifyPartitionV2,
     ProbeStatus,
     ReformatDisk,
+    StorageEncryptionSupport,
     StorageResponse,
     StorageResponseV2,
     )
@@ -79,6 +82,7 @@ from subiquity.models.filesystem import (
 from subiquity.server.controller import (
     SubiquityController,
     )
+from subiquity.server import snapdapi
 from subiquity.server.types import InstallerChannels
 
 
@@ -86,6 +90,22 @@ log = logging.getLogger("subiquity.server.controllers.filesystem")
 block_discover_log = logging.getLogger('block-discover')
 
 
+system_defective_encryption_text = _("""
+The model being installed requires TPM-backed encryption but this
+system does not support it.
+""")
+
+system_multiple_volumes_text = _("""
+The model being installed defines multiple volumes, which is not currently
+supported.
+""")
+
+system_non_gpt_text = _("""
+The model being installed defines a volume with a partition table type other
+than GPT, which is not currently supported.
+""")
+
+
 class FilesystemController(SubiquityController, FilesystemManipulator):
 
     endpoint = API.storage
@@ -116,6 +136,7 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
             (InstallerChannels.CONFIGURED, 'source'),
             self._get_system_task.start_sync)
         self._system = None
+        self._core_boot_classic_error = ''
 
     def load_autoinstall_data(self, data):
         log.debug("load_autoinstall_data %s", data)
@@ -141,8 +162,21 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
         if label is None:
             self._system = None
             return
-        self._system = await self.app.snapdapi.v2.systems[label].GET()
+        system = await self.app.snapdapi.v2.systems[label].GET()
-        log.debug("got system %s", self._system)
+        log.debug("got system %s", system)
+        if len(system.volumes) == 0:
+            # This means the system does not define a gadget or kernel
+            # so isn't a core boot classic system.
+            return
+        self._system = system
+        if len(system.volumes) > 1:
+            self._core_boot_classic_error = system_multiple_volumes_text
+        [volume] = system.volumes.values()
+        if volume.schema != 'gpt':
+            self._core_boot_classic_error = system_non_gpt_text
+        if system.storage_encryption.support == \
+           StorageEncryptionSupport.DEFECTIVE:
+            self._core_boot_classic_error = system_defective_encryption_text
 
     @with_context()
     async def apply_autoinstall_config(self, context=None):
@@ -371,17 +405,65 @@ class FilesystemController(SubiquityController, FilesystemManipulator):
         se = None
         if self._system is not None:
             se = self._system.storage_encryption
+            offsets_and_sizes = list(self._offsets_and_sizes_for_system())
+            _structure, last_offset, last_size = offsets_and_sizes[-1]
+            min_size = last_offset + last_size
         return GuidedStorageResponse(
             status=ProbeStatus.DONE,
             error_report=self.full_probe_error(),
             disks=[labels.for_client(d, min_size=min_size) for d in disks],
+            core_boot_classic_error=self._core_boot_classic_error,
             storage_encryption=se)
 
+    def _offsets_and_sizes_for_system(self):
+        offset = self.model._partition_alignment_data['gpt'].min_start_offset
+        [volume] = self._system.volumes.values()
+        for structure in volume.structure:
+            if structure.role == snapdapi.Role.MBR:
+                continue
+            if ',' not in structure.type:
+                continue
+            if structure.offset is not None:
+                offset = structure.offset
+            yield (structure, offset, structure.size)
+            offset = offset + structure.size
+
+    def _add_structure(self, *, disk: Disk, offset: int, size: int,
+                       is_last: bool, structure: snapdapi.VolumeStructure):
+        print(size)
+        if structure.role == snapdapi.Role.SYSTEM_DATA and is_last:
+            size = gaps.largest_gap(disk).size
+        flag = ptable_uuid_to_flag_entry(structure.gpt_part_uuid())[0]
+        part = self.model.add_partition(
+            disk, offset=offset, size=size, flag=flag,
+            partition_name=structure.name)
+        if structure.filesystem:
+            part.wipe = 'superblock'
+            fs = self.model.add_filesystem(
+                part, structure.filesystem, label=structure.label)
+            if structure.role == snapdapi.Role.SYSTEM_DATA:
+                self.model.add_mount(fs, '/')
+            elif structure.role == snapdapi.Role.SYSTEM_BOOT:
+                self.model.add_mount(fs, '/boot')
+            elif flag == 'boot':
+                self.model.add_mount(fs, '/boot/efi')
+
+    def apply_system(self, disk_id):
+        disk = self.model._one(id=disk_id)
+        self.reformat(disk)
+
+        [volume] = self._system.volumes.values()
+
+        for structure, offset, size in self._offsets_and_sizes_for_system():
+            self._add_structure(
+                disk=disk, offset=offset, size=size, structure=structure,
+                is_last=structure == volume.structure[-1])
+        disk._partitions.sort(key=lambda p: p.number)
+
     async def guided_POST(self, data: GuidedChoice) -> StorageResponse:
         log.debug(data)
         if self._system is not None:
-            # Here is where we will apply the gadget info from the
+            self.apply_system(data.disk_id)
-            # system to the disk!
             await self.configured()
         else:
             self.guided(GuidedChoiceV2.from_guided_choice(data))

subiquity/server/controllers/tests/test_filesystem.py

@@ -14,6 +14,7 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import copy
+import json
 from unittest import mock, TestCase, IsolatedAsyncioTestCase
 
 from parameterized import parameterized
@@ -35,7 +36,7 @@ from subiquity.models.tests.test_filesystem import (
     make_model,
     make_partition,
     )
-
+from subiquity.server import snapdapi
 
 bootloaders = [(bl, ) for bl in list(Bootloader)]
 bootloaders_and_ptables = [(bl, pt)
@@ -457,3 +458,73 @@ class TestGuidedV2(IsolatedAsyncioTestCase):
         self.assertEqual(
                 disk_size - (1 << 20), parts[-1].offset + parts[-1].size,
                 disk_size)
+
+
+class TestApplySystem(TestCase):
+
+    def setUp(self):
+        self.app = make_app()
+        self.app.opts.bootloader = 'UEFI'
+        self.app.report_start_event = mock.Mock()
+        self.app.report_finish_event = mock.Mock()
+        self.app.prober = mock.Mock()
+        self.fsc = FilesystemController(app=self.app)
+        self.fsc._configured = True
+        self.fsc.model = make_model(Bootloader.UEFI)
+
+    def test_add_structure(self):
+        disk = make_disk(self.fsc.model)
+        self.fsc._add_structure(
+            disk=disk,
+            offset=100,
+            size=2 << 20,
+            is_last=False,
+            structure=snapdapi.VolumeStructure(
+                name='ptname',
+                type="83,0FC63DAF-8483-4772-8E79-3D69D8477DE4",
+                label='label',
+                size=1 << 20,
+                role=snapdapi.Role.SYSTEM_DATA,
+                filesystem='ext4'))
+        [part] = disk.partitions()
+        self.assertEqual(part.offset, 100)
+        self.assertEqual(part.partition_name, 'ptname')
+        self.assertEqual(part.flag, 'linux')
+        self.assertEqual(part.size, 2 << 20)
+        self.assertEqual(part.fs().fstype, 'ext4')
+        self.assertEqual(part.fs().mount().path, '/')
+        self.assertEqual(part.wipe, 'superblock')
+
+    def test_add_structure_no_fs(self):
+        disk = make_disk(self.fsc.model)
+        self.fsc._add_structure(
+            disk=disk,
+            offset=100,
+            size=2 << 20,
+            is_last=False,
+            structure=snapdapi.VolumeStructure(
+                type="83,0FC63DAF-8483-4772-8E79-3D69D8477DE4",
+                size=1 << 20,
+                filesystem=None))
+        [part] = disk.partitions()
+        self.assertEqual(part.size, 2 << 20)
+        self.assertEqual(part.fs(), None)
+        self.assertEqual(part.wipe, None)
+
+    def test_from_sample_data(self):
+        self.fsc.model = model = make_model(Bootloader.UEFI)
+        disk = make_disk(model)
+        with open('examples/snaps/v2-systems-unavailable.json') as fp:
+            self.fsc._system = snapdapi.snapd_serializer.deserialize(
+                snapdapi.SystemDetails, json.load(fp)['result'])
+        self.fsc.apply_system(disk.id)
+        partition_count = len([
+            structure
+            for structure in self.fsc._system.volumes['pc'].structure
+            if structure.role != snapdapi.Role.MBR
+            ])
+        self.assertEqual(
+            partition_count,
+            len(disk.partitions()))
+        mounts = {m.path for m in model._all(type='mount')}
+        self.assertEqual(mounts, {'/', '/boot', '/boot/efi'})

subiquity/server/snapdapi.py

@@ -141,6 +141,9 @@ class VolumeStructure:
     content: Optional[List[VolumeContent]] = None
     update: VolumeUpdate = attr.Factory(VolumeUpdate)
 
+    def gpt_part_uuid(self):
+        return self.type.split(',', 1)[1].upper()
+
 
 @attr.s(auto_attribs=True)
 class Volume:
@@ -252,7 +255,7 @@ def make_api_client(async_snapd):
             content = await async_snapd.get(path[1:], **params)
         else:
             content = await async_snapd.post(path[1:], json, **params)
-        response = serializer.deserialize(Response, content)
+        response = snapd_serializer.deserialize(Response, content)
         if response.type == ResponseType.SYNC:
             content = content['result']
         elif response.type == ResponseType.ASYNC:
@@ -261,10 +264,11 @@ def make_api_client(async_snapd):
             yield _FakeError()
         yield _FakeResponse(content)
 
-    serializer = Serializer(
+    return make_client(SnapdAPI, make_request, serializer=snapd_serializer)
-        ignore_unknown_fields=True, serialize_enums_by='value')
 
-    return make_client(SnapdAPI, make_request, serializer=serializer)
+
+snapd_serializer = Serializer(
+    ignore_unknown_fields=True, serialize_enums_by='value')
 
 
 async def post_and_wait(client, meth, *args, **kw):

subiquity/ui/views/filesystem/probing.py

@@ -86,20 +86,14 @@ class ProbingFailed(BaseView):
         self.controller.app.show_error_report(self.error_ref)
 
 
-defective_text = _("""
+class CoreBootClassicError(BaseView):
-The model being installed requires TPM-backed encryption but this
-system does not support it.
-""")
 
+    title = _("Cannot install core boot classic system")
 
-class DefectiveEncryptionError(BaseView):
+    def __init__(self, controller, msg):
-
-    title = _("Encryption requirements not met")
-
-    def __init__(self, controller):
         self.controller = controller
         super().__init__(screen([
-            Text(rewrap(_(defective_text))),
+            Text(rewrap(_(msg))),
             Text(""),
             ],
             [other_btn(_("Back"), on_press=self.cancel)]))