import logging import subprocess import re # pylint: disable=locally-disabled, line-too-long, logging-fstring-interpolation, too-many-lines class DiskLibrary: def __init__(self): self.logger = logging.getLogger("OpengnsysDiskLibrary") self.logger.setLevel(logging.DEBUG) def split_device_partition(self, device): """ Parses a device file like /dev/sda3 into the root device (/dev/sda) and partition number (3) Args: device (str): Device in /dev Returns: [base_device, partno] """ r = re.compile("^(.*?)(\\d+)$") m = r.match(device) disk = m.group(1) partno = int(m.group(2)) self.logger.debug(f"{device} parsed into disk device {disk}, partition {partno}") return (disk, partno) def get_disk_json_data(self, device): """ Returns the partition JSON data dump for the entire disk, even if a partition is passed. This is specifically in the format used by sfdisk. Args: device (str): Block device, eg, /dev/sda3 Returns: str: JSON dump produced by sfdisk """ (disk, partno) = self.split_device_partition(device) result = subprocess.run(["/usr/sbin/sfdisk", "--json", disk], check=True, capture_output=True, encoding='utf-8') return result.stdout.strip() def get_disk_uuid(self, device): """ Returns the UUID of the disk itself, if there's a GPT partition table. Args: device (str): Block device, eg, /dev/sda3 Returns: str: UUID """ (disk, partno) = self.split_device_partition(device) result = subprocess.run(["/usr/sbin/sfdisk", "--disk-id", disk], check=True, capture_output=True, encoding='utf-8') return result.stdout.strip() def set_disk_uuid(self, device, uuid): (disk, partno) = self.split_device_partition(device) subprocess.run(["/usr/sbin/sfdisk", "--disk-id", disk, uuid], check=True, encoding='utf-8') def get_partition_uuid(self, device): """ Returns the UUID of the partition, if there's a GPT partition table. Args: device (str): Block device, eg, /dev/sda3 Returns: str: UUID """ (disk, partno) = self.split_device_partition(device) #result = subprocess.run(["/usr/sbin/sfdisk", "--part-uuid", disk, str(partno)], check=True, capture_output=True, encoding='utf-8') #return result.stdout.strip() guid = None result = subprocess.run(["/usr/sbin/sgdisk", "--info", str(partno), disk], check=True, capture_output=True, encoding='utf-8') for l in result.stdout.splitlines(): if 'Partition unique GUID' not in l: continue guid = l.replace ('Partition unique GUID: ', '') if guid is None: self.logger.error (f'failed to get UUID of disk "{disk}" part "{partno}"') return None return guid def set_partition_uuid(self, device, uuid): (disk, partno) = self.split_device_partition(device) subprocess.run(["/usr/sbin/sfdisk", "--part-uuid", disk, str(partno), uuid], check=True, encoding='utf-8') def get_partition_type(self, device): """ Returns the type UUID of the partition, if there's a GPT partition table. Args: device (str): Block device, eg, /dev/sda3 Returns: str: UUID """ (disk, partno) = self.split_device_partition(device) result = subprocess.run(["/usr/sbin/sfdisk", "--part-type", disk, str(partno)], check=True, capture_output=True, encoding='utf-8') return result.stdout.strip() def set_partition_type(self, device, uuid): (disk, partno) = self.split_device_partition(device) subprocess.run(["/usr/sbin/sfdisk", "--part-type", disk, str(partno), uuid], check=True, encoding='utf-8')