def enable_encryption_format()

in VMEncryption/main/handle.py [0:0]

• 76 lines of code
• 30 McCabe index (conditional complexity)

def enable_encryption_format(passphrase, disk_format_query, disk_util, force=False):
    logger.log('enable_encryption_format')
    logger.log('disk format query is {0}'.format(disk_format_query))

    json_parsed = json.loads(disk_format_query)

    if type(json_parsed) is dict:
        encryption_format_items = [json_parsed, ]
    elif type(json_parsed) is list:
        encryption_format_items = json_parsed
    else:
        raise Exception("JSON parse error. Input: {0}".format(disk_format_query))

    for encryption_item in encryption_format_items:
        dev_path_in_query = None

        if "scsi" in encryption_item and encryption_item["scsi"] != '':
            dev_path_in_query = disk_util.query_dev_sdx_path_by_scsi_id(encryption_item["scsi"])
        if "dev_path" in encryption_item and encryption_item["dev_path"] != '':
            dev_path_in_query = encryption_item["dev_path"]

        if not dev_path_in_query:
            raise Exception("Could not find a device path for Encryption Item: {0}".format(json.dumps(encryption_item)))

        devices = disk_util.get_device_items(dev_path_in_query)
        if len(devices) != 1:
            logger.log(msg=("the device with this path {0} have more than one sub device. so skip it.".format(dev_path_in_query)), level=CommonVariables.WarningLevel)
            continue
        else:
            device_item = devices[0]
            if device_item.file_system is None or device_item.file_system == "" or force:
                if device_item.mount_point:
                    disk_util.swapoff()
                    disk_util.umount(device_item.mount_point)
                mapper_name = str(uuid.uuid4())
                logger.log("encrypting " + str(device_item))
                encrypted_device_path = os.path.join(CommonVariables.dev_mapper_root, mapper_name)
                try:
                    se_linux_status = None
                    if DistroPatcher.distro_info[0].lower() == 'centos' and DistroPatcher.distro_info[1].startswith('7.0'):
                        se_linux_status = encryption_environment.get_se_linux()
                        if se_linux_status.lower() == 'enforcing':
                            encryption_environment.disable_se_linux()
                    encrypt_result = disk_util.encrypt_disk(dev_path=dev_path_in_query, passphrase_file=passphrase, mapper_name=mapper_name, header_file=None)
                finally:
                    if DistroPatcher.distro_info[0].lower() == 'centos' and DistroPatcher.distro_info[1].startswith('7.0'):
                        if se_linux_status is not None and se_linux_status.lower() == 'enforcing':
                            encryption_environment.enable_se_linux()

                if encrypt_result == CommonVariables.process_success:
                    # TODO: let customer specify the default file system in the
                    # parameter
                    file_system = None
                    if "file_system" in encryption_item and encryption_item["file_system"] != "":
                        file_system = encryption_item["file_system"]
                    else:
                        file_system = CommonVariables.default_file_system
                    format_disk_result = disk_util.format_disk(dev_path=encrypted_device_path, file_system=file_system)
                    if format_disk_result != CommonVariables.process_success:
                        logger.log(msg=("format of disk {0} failed with result: {1}".format(encrypted_device_path, format_disk_result)), level=CommonVariables.ErrorLevel)
                    crypt_item_to_update = CryptItem()
                    crypt_item_to_update.mapper_name = mapper_name
                    crypt_item_to_update.dev_path = dev_path_in_query
                    crypt_item_to_update.luks_header_path = None
                    crypt_item_to_update.file_system = file_system
                    crypt_item_to_update.uses_cleartext_key = False
                    crypt_item_to_update.current_luks_slot = 0

                    if "name" in encryption_item and encryption_item["name"] != "":
                        crypt_item_to_update.mount_point = os.path.join("/mnt/", str(encryption_item["name"]))
                    else:
                        crypt_item_to_update.mount_point = os.path.join("/mnt/", mapper_name)

                    # allow override through the new full_mount_point field
                    if "full_mount_point" in encryption_item and encryption_item["full_mount_point"] != "":
                        crypt_item_to_update.mount_point = os.path.join(str(encryption_item["full_mount_point"]))

                    logger.log(msg="modifying/removing the entry for unencrypted drive in fstab", level=CommonVariables.InfoLevel)
                    disk_util.modify_fstab_entry_encrypt(crypt_item_to_update.mount_point, os.path.join(CommonVariables.dev_mapper_root, mapper_name))

                    disk_util.make_sure_path_exists(crypt_item_to_update.mount_point)
                    update_crypt_item_result = disk_util.add_crypt_item(crypt_item_to_update, passphrase)
                    if not update_crypt_item_result:
                        logger.log(msg="update crypt item failed", level=CommonVariables.ErrorLevel)

                    mount_result = disk_util.mount_filesystem(dev_path=encrypted_device_path, mount_point=crypt_item_to_update.mount_point)
                    logger.log(msg=("mount result is {0}".format(mount_result)))
                else:
                    logger.log(msg="encryption failed with code {0}".format(encrypt_result), level=CommonVariables.ErrorLevel)
            else:
                logger.log(msg=("the item fstype is not empty {0}".format(device_item.file_system)))