osd_fsid,
lockbox_secret)
dmcrypt_secret = encryption_utils.get_dmcrypt_key(osd_id, osd_fsid)
- encryption_utils.luks_open(dmcrypt_secret,
- osd_block_lv.__dict__['lv_path'],
- osd_block_lv.__dict__['lv_uuid'],
- with_tpm=self.with_tpm)
+ lv_path: str = osd_block_lv.__dict__['lv_path']
+ if disk.has_holders(lv_path):
+ real_path_device = os.path.realpath(lv_path)
+ holders = disk.get_block_device_holders()
+
+ if real_path_device in holders.keys() and real_path_device in holders.values():
+ osd_lv_path = disk.get_lvm_mapper_path_from_dm(next(k for k, v in holders.items() if v == real_path_device))
+ else:
+ encryption_utils.luks_open(dmcrypt_secret,
+ osd_block_lv.__dict__['lv_path'],
+ osd_block_lv.__dict__['lv_uuid'],
+ with_tpm=self.with_tpm)
else:
osd_lv_path = osd_block_lv.__dict__['lv_path']
from ceph_volume.util import system, disk
from ceph_volume.util import prepare as prepare_utils
from ceph_volume.util import encryption as encryption_utils
+from ceph_volume.util.device import Device
from ceph_volume.devices.lvm.common import rollback_osd
from ceph_volume.devices.raw.list import direct_report
from typing import Any, Dict, List, Optional, TYPE_CHECKING
self.pre_activate_tpm2(device)
found = direct_report(self.devices)
+ holders = disk.get_block_device_holders()
for osd_uuid, meta in found.items():
+ realpath_device = os.path.realpath(meta['device'])
+ parent_device = holders.get(realpath_device)
+ if parent_device and any('ceph.cluster_fsid' in lv.lv_tags for lv in Device(parent_device).lvs):
+ continue
osd_id = meta['osd_id']
if self.osd_id is not None and str(osd_id) != str(self.osd_id):
continue
self.with_tpm = 1
self.temp_mapper: str = f'activating-{os.path.basename(device)}'
self.temp_mapper_path: str = f'/dev/mapper/{self.temp_mapper}'
- encryption_utils.luks_open(
- '',
- device,
- self.temp_mapper,
- self.with_tpm
- )
- bluestore_header: Dict[str, Any] = disk.get_bluestore_header(self.temp_mapper_path)
- if not bluestore_header:
- raise RuntimeError(f"{device} doesn't have BlueStore signature.")
-
- kname: str = disk.get_parent_device_from_mapper(self.temp_mapper_path, abspath=False)
- device_type = bs_mapping_type[bluestore_header[self.temp_mapper_path]['description']]
- new_mapper: str = f'ceph-{self.osd_fsid}-{kname}-{device_type}-dmcrypt'
- self.block_device_path = f'/dev/mapper/{new_mapper}'
- self.devices.append(self.block_device_path)
- encryption_utils.rename_mapper(self.temp_mapper, new_mapper)
+ if not disk.BlockSysFs(device).has_active_dmcrypt_mapper:
+ encryption_utils.luks_open(
+ '',
+ device,
+ self.temp_mapper,
+ self.with_tpm
+ )
+ bluestore_header: Dict[str, Any] = disk.get_bluestore_header(self.temp_mapper_path)
+ if not bluestore_header:
+ raise RuntimeError(f"{device} doesn't have BlueStore signature.")
+
+ kname: str = disk.get_parent_device_from_mapper(self.temp_mapper_path, abspath=False)
+ device_type = bs_mapping_type[bluestore_header[self.temp_mapper_path]['description']]
+ new_mapper: str = f'ceph-{self.osd_fsid}-{kname}-{device_type}-dmcrypt'
+ self.block_device_path = f'/dev/mapper/{new_mapper}'
+ self.devices.append(self.block_device_path)
+ # An option could be to simply rename the mapper but the uuid remains unchanged in sysfs
+ encryption_utils.luks_close(self.temp_mapper)
+ encryption_utils.luks_open('', device, new_mapper, self.with_tpm)
lv_tags=f'ceph.type=db,ceph.db_uuid=fake-db-uuid,ceph.block_uuid=fake-block-uuid,ceph.wal_uuid=fake-wal-uuid,ceph.osd_id=0,ceph.osd_fsid=abcd,ceph.cluster_name=ceph,{encrypted},ceph.cephx_lockbox_secret=abcd',
lv_uuid='fake-db-uuid'),
Volume(lv_name='lv_foo-db',
- lv_path='/fake-db-path',
- vg_name='vg_foo_db',
+ lv_path='/fake-wal-path',
+ vg_name='vg_foo_wal',
lv_tags=f'ceph.type=wal,ceph.block_uuid=fake-block-uuid,ceph.wal_uuid=fake-wal-uuid,ceph.db_uuid=fake-db-uuid,ceph.osd_id=0,ceph.osd_fsid=abcd,ceph.cluster_name=ceph,{encrypted},ceph.cephx_lockbox_secret=abcd',
lv_uuid='fake-wal-uuid')]
self.lvm_bs._activate(lvs)
{'args': (['ln', '-snf', '/fake-db-path',
'/var/lib/ceph/osd/ceph-0/block.db'],),
'kwargs': {}},
- {'args': (['ln', '-snf', '/fake-db-path',
+ {'args': (['ln', '-snf', '/fake-wal-path',
'/var/lib/ceph/osd/ceph-0/block.wal'],),
'kwargs': {}},
{'args': (['systemctl', 'enable',
import pytest
from ceph_volume.util import disk
from mock.mock import patch, Mock, MagicMock, mock_open
+from pyfakefs.fake_filesystem_unittest import TestCase
class TestFunctions:
with patch('builtins.open', mock_open(read_data='test--foo--vg-test--foo--lv')):
assert disk.get_lvm_mapper_path_from_dm('/dev/dm-123') == '/dev/mapper/test--foo--vg-test--foo--lv'
+ @patch('ceph_volume.util.disk.get_block_device_holders', MagicMock(return_value={'/dev/dmcrypt-mapper-123': '/dev/sda'}))
+ @patch('os.path.realpath', MagicMock(return_value='/dev/sda'))
+ def test_has_holders_true(self):
+ assert disk.has_holders('/dev/sda')
+
+ @patch('ceph_volume.util.disk.get_block_device_holders', MagicMock(return_value={'/dev/dmcrypt-mapper-123': '/dev/sda'}))
+ @patch('os.path.realpath', MagicMock(return_value='/dev/sdb'))
+ def test_has_holders_false(self):
+ assert not disk.has_holders('/dev/sda')
+
+ @patch('ceph_volume.util.disk.get_block_device_holders', MagicMock(return_value={'/dev/dmcrypt-mapper-123': '/dev/sda'}))
+ @patch('os.path.realpath', MagicMock(return_value='/dev/foobar'))
+ def test_has_holders_device_does_not_exist(self):
+ assert not disk.has_holders('/dev/foobar')
+
class TestLsblkParser(object):
def test_parses_whitespace_values(self):
def test_device_path_is_a_path(self, fake_filesystem):
device_path = '/var/lib/ceph/osd/ceph-0'
fake_filesystem.create_dir(device_path)
- assert not disk.has_bluestore_label(device_path)
\ No newline at end of file
+ assert not disk.has_bluestore_label(device_path)
+
+
+class TestBlockSysFs(TestCase):
+ def setUp(self) -> None:
+ self.setUpPyfakefs()
+ self.fs.create_dir('/fake-area/foo/holders')
+ self.fs.create_dir('/fake-area/bar2/holders')
+ self.fs.create_file('/fake-area/bar2/holders/dm-0')
+ self.fs.create_file('/fake-area/foo/holders/dm-1')
+ self.fs.create_file('/fake-area/bar2/partition', contents='2')
+ self.fs.create_dir('/sys/dev/block')
+ self.fs.create_dir('/sys/block/foo')
+ self.fs.create_symlink('/sys/dev/block/8:0', '/fake-area/foo')
+ self.fs.create_symlink('/sys/dev/block/252:2', '/fake-area/bar2')
+ self.fs.create_file('/sys/block/dm-0/dm/uuid', contents='CRYPT-LUKS2-1234-abcdef')
+ self.fs.create_file('/sys/block/dm-1/dm/uuid', contents='LVM-abcdef')
+
+ def test_init(self) -> None:
+ b = disk.BlockSysFs('/dev/foo')
+ assert b.path == '/dev/foo'
+ assert b.sys_dev_block == '/sys/dev/block'
+ assert b.sys_block == '/sys/block'
+
+ def test_get_sys_dev_block_path(self) -> None:
+ b = disk.BlockSysFs('/dev/foo')
+ assert b.get_sys_dev_block_path == '/sys/dev/block/8:0'
+
+ def test_is_partition_true(self) -> None:
+ b = disk.BlockSysFs('/dev/bar2')
+ assert b.is_partition
+
+ def test_is_partition_false(self) -> None:
+ b = disk.BlockSysFs('/dev/foo')
+ assert not b.is_partition
+
+ def test_holders(self) -> None:
+ b1 = disk.BlockSysFs('/dev/bar2')
+ b2 = disk.BlockSysFs('/dev/foo')
+ assert b1.holders == ['dm-0']
+ assert b2.holders == ['dm-1']
+
+ def test_has_active_dmcrypt_mapper(self) -> None:
+ b = disk.BlockSysFs('/dev/bar2')
+ assert b.has_active_dmcrypt_mapper
+
+ def test_has_active_mappers(self) -> None:
+ b = disk.BlockSysFs('/dev/foo')
+ assert b.has_active_mappers
+
+ def test_active_mappers_dmcrypt(self) -> None:
+ b = disk.BlockSysFs('/dev/bar2')
+ assert b.active_mappers()
+ assert b.active_mappers()['dm-0']
+ assert b.active_mappers()['dm-0']['type'] == 'CRYPT'
+ assert b.active_mappers()['dm-0']['dmcrypt_mapping'] == 'abcdef'
+ assert b.active_mappers()['dm-0']['dmcrypt_type'] == 'LUKS2'
+ assert b.active_mappers()['dm-0']['dmcrypt_uuid'] == '1234'
+
+ def test_active_mappers_lvm(self) -> None:
+ b = disk.BlockSysFs('/dev/foo')
+ assert b.active_mappers()
+ assert b.active_mappers()['dm-1']
+ assert b.active_mappers()['dm-1']['type'] == 'LVM'
+ assert b.active_mappers()['dm-1']['uuid'] == 'abcdef'
encryption.luks_open('abcd', '/dev/foo', '/dev/bar')
assert m_call.call_args[0][0] == expected
+ @patch('ceph_volume.util.encryption.bypass_workqueue', return_value=False)
+ @patch('ceph_volume.util.encryption.process.call')
+ def test_luks_open_command_with_tpm(self, m_call, m_bypass_workqueue, conf_ceph_stub):
+ fake_mapping: str = 'fake-mapping'
+ fake_device: str = 'fake-device'
+ expected = [
+ '/usr/lib/systemd/systemd-cryptsetup',
+ 'attach',
+ fake_mapping,
+ fake_device,
+ '-',
+ 'tpm2-device=auto,discard,headless=true,nofail',
+ ]
+ encryption.luks_open('', fake_device, fake_mapping, 1)
+ assert m_call.call_args[0][0] == expected
+
+ @patch('ceph_volume.util.encryption.bypass_workqueue', return_value=True)
+ @patch('ceph_volume.util.encryption.process.call')
+ def test_luks_open_command_with_tpm_bypass_workqueue(self, m_call, m_bypass_workqueue, conf_ceph_stub):
+ fake_mapping: str = 'fake-mapping'
+ fake_device: str = 'fake-device'
+ expected = [
+ '/usr/lib/systemd/systemd-cryptsetup',
+ 'attach',
+ fake_mapping,
+ fake_device,
+ '-',
+ 'tpm2-device=auto,discard,headless=true,nofail,no-read-workqueue,no-write-workqueue',
+ ]
+ encryption.luks_open('', fake_device, fake_mapping, 1)
+ assert m_call.call_args[0][0] == expected
+
class TestCephLuks2:
@patch.object(encryption.CephLuks2, 'get_osd_fsid', Mock(return_value='abcd-1234'))
return result
+def has_holders(device: str) -> bool:
+ """Check if a given device has any associated holders.
+
+ This function determines whether the specified device has associated holders
+ (e.g., other devices that depend on it) by checking if the device's real path
+ appears in the values of the dictionary returned by `get_block_device_holders`.
+
+ Args:
+ device (str): The path to the device (e.g., '/dev/sdX') to check.
+
+ Returns:
+ bool: True if the device has holders, False otherwise.
+ """
+ return os.path.realpath(device) in get_block_device_holders().values()
+
def get_parent_device_from_mapper(mapper: str, abspath: bool = True) -> str:
"""Get the parent device corresponding to a given device mapper.
with open(sys_block_path, 'r') as f:
content: str = f.read()
result = f'/dev/mapper/{content}'
- return result
+ return result.strip()
+
+
+class BlockSysFs:
+ def __init__(self,
+ path: str,
+ sys_dev_block: str = '/sys/dev/block',
+ sys_block: str = '/sys/block') -> None:
+ """
+ Initializes a BlockSysFs object.
+
+ Args:
+ path (str): The path to the block device.
+ sys_dev_block (str, optional): Path to the sysfs directory containing block devices.
+ Defaults to '/sys/dev/block'.
+ sys_block (str, optional): Path to the sysfs directory containing block information.
+ Defaults to '/sys/block'.
+ """
+ self.path: str = path
+ self.name: str = os.path.basename(os.path.realpath(self.path))
+ self.sys_dev_block: str = sys_dev_block
+ self.sys_block: str = sys_block
+
+ @property
+ def is_partition(self) -> bool:
+ """
+ Checks if the current block device is a partition.
+
+ Returns:
+ bool: True if it is a partition, False otherwise.
+ """
+ path: str = os.path.join(self.get_sys_dev_block_path, 'partition')
+ return os.path.exists(path)
+
+ @property
+ def holders(self) -> List[str]:
+ """
+ Retrieves the holders of the current block device.
+
+ Returns:
+ List[str]: A list of holders (other devices) associated with this block device.
+ """
+ result: List[str] = []
+ path: str = os.path.join(self.get_sys_dev_block_path, 'holders')
+ if os.path.exists(path):
+ result = os.listdir(path)
+ return result
+
+ @property
+ def get_sys_dev_block_path(self) -> str:
+ """
+ Gets the sysfs path for the current block device.
+
+ Returns:
+ str: The sysfs path corresponding to this block device.
+ """
+ sys_dev_block_path: str = ''
+ devices: List[str] = os.listdir(self.sys_dev_block)
+ for device in devices:
+ path = os.path.join(self.sys_dev_block, device)
+ if os.path.realpath(path).split('/')[-1:][0] == self.name:
+ sys_dev_block_path = path
+ return sys_dev_block_path
+
+ @property
+ def has_active_mappers(self) -> bool:
+ """
+ Checks if there are any active device mappers for the current block device.
+
+ Returns:
+ bool: True if active mappers exist, False otherwise.
+ """
+ return len(self.active_mappers()) > 0
+
+ @property
+ def has_active_dmcrypt_mapper(self) -> bool:
+ """
+ Checks if there is an active dm-crypt (disk encryption) mapper for the current block device.
+
+ Returns:
+ bool: True if an active dm-crypt mapper exists, False otherwise.
+ """
+ return any(value.get('type') == 'CRYPT' for value in self.active_mappers().values())
+
+ def active_mappers(self) -> Dict[str, Any]:
+ """
+ Retrieves information about active device mappers for the current block device.
+
+ Returns:
+ Dict[str, Any]: A dictionary containing details about active device mappers.
+ Keys are the holders, and values provide details like type,
+ dm-crypt metadata, and LVM UUIDs.
+ """
+ result: Dict[str, Any] = {}
+ for holder in self.holders:
+ path: str = os.path.join(self.sys_block, holder, 'dm/uuid')
+ if os.path.exists(path):
+ result[holder] = {}
+ with open(path, 'r') as f:
+ content: str = f.read().strip()
+ content_split: List[str] = content.split('-', maxsplit=3)
+ mapper_type: str = content_split[0]
+ result[holder]['type'] = mapper_type
+ if mapper_type == 'CRYPT':
+ result[holder]['dmcrypt_type'] = content_split[1]
+ result[holder]['dmcrypt_uuid'] = content_split[2]
+ result[holder]['dmcrypt_mapping'] = content_split[3]
+ if mapper_type == 'LVM':
+ result[holder]['uuid'] = content_split[1]
+ return result
\ No newline at end of file
:param key: dmcrypt secret key
:param device: absolute path to device
:param mapping: mapping name used to correlate device. Usually a UUID
+ :param with_tpm: whether to use tpm2 token enrollment.
"""
command: List[str] = []
if with_tpm:
mapping,
device,
'-',
- 'tpm2-device=auto,discard']
+ 'tpm2-device=auto,discard,headless=true,nofail']
if bypass_workqueue(device):
command[-1] += ',no-read-workqueue,no-write-workqueue'
else:
projects / ceph.git / commitdiff