+ def scan_encrypted(self, directory=None):
+ device = self.encryption_metadata['device']
+ lockbox = self.encryption_metadata['lockbox']
+ encryption_type = self.encryption_metadata['type']
+ osd_metadata = {}
+ # Get the PARTUUID of the device to make sure have the right one and
+ # that maps to the data device
+ device_uuid = disk.get_partuuid(device)
+ dm_path = '/dev/mapper/%s' % device_uuid
+ # check if this partition is already mapped
+ device_status = encryption.status(device_uuid)
+
+ # capture all the information from the lockbox first, reusing the
+ # directory scan method
+ if self.device_mounts.get(lockbox):
+ lockbox_path = self.device_mounts.get(lockbox)[0]
+ lockbox_metadata = self.scan_directory(lockbox_path)
+ # ceph-disk stores the fsid as osd-uuid in the lockbox, thanks ceph-disk
+ dmcrypt_secret = encryption.get_dmcrypt_key(
+ None, # There is no ID stored in the lockbox
+ lockbox_metadata['osd-uuid'],
+ os.path.join(lockbox_path, 'keyring')
+ )
+ else:
+ with system.tmp_mount(lockbox) as lockbox_path:
+ lockbox_metadata = self.scan_directory(lockbox_path)
+ # ceph-disk stores the fsid as osd-uuid in the lockbox, thanks ceph-disk
+ dmcrypt_secret = encryption.get_dmcrypt_key(
+ None, # There is no ID stored in the lockbox
+ lockbox_metadata['osd-uuid'],
+ os.path.join(lockbox_path, 'keyring')
+ )
+
+ if not device_status:
+ # Note how both these calls need b64decode. For some reason, the
+ # way ceph-disk creates these keys, it stores them in the monitor
+ # *undecoded*, requiring this decode call again. The lvm side of
+ # encryption doesn't need it, so we are assuming here that anything
+ # that `simple` scans, will come from ceph-disk and will need this
+ # extra decode call here
+ dmcrypt_secret = base64.b64decode(dmcrypt_secret)
+ if encryption_type == 'luks':
+ encryption.luks_open(dmcrypt_secret, device, device_uuid)
+ else:
+ encryption.plain_open(dmcrypt_secret, device, device_uuid)
+
+ # If we have a directory, use that instead of checking for mounts
+ if directory:
+ osd_metadata = self.scan_directory(directory)
+ else:
+ # Now check if that mapper is mounted already, to avoid remounting and
+ # decrypting the device
+ dm_path_mount = self.device_mounts.get(dm_path)
+ if dm_path_mount:
+ osd_metadata = self.scan_directory(dm_path_mount[0])
+ else:
+ with system.tmp_mount(dm_path, encrypted=True) as device_path:
+ osd_metadata = self.scan_directory(device_path)
+
+ osd_metadata['encrypted'] = True
+ osd_metadata['encryption_type'] = encryption_type
+ osd_metadata['lockbox.keyring'] = parse_keyring(lockbox_metadata['keyring'])
+ return osd_metadata
+