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() 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')