openstack/ironic-python-agent
Code Issues Proposed changes

Inspect non-raw images for safety

Browse Source 
When IPA gets a non-raw image, it performs an on-the-fly conversion
using qemu-img convert, as well as running qemu-img frequently to get
basic information about the image before validating it.

Now, we ensure that before any qemu-img calls are made, that we have
inspected the image for safety and pass through the detected format.

If given a disk_format=raw image and image streaming is enabled
(default), we retain the existing behavior of not inspecting it in
any way and streaming it bit-perfect to the device. In this case, we
never use qemu-based tools on the image at all.

If given a disk_format=raw image and image streaming is disabled, this
change fixes a bug where the image may have been converted if it was not
actually raw in the first place. We now stream these bit-perfect to the
device.

Adds two config options:
- [DEFAULT]/disable_deep_image_inspection, which can be set to "True" in
  order to disable all security features. Do not do this.
- [DEFAULT]/permitted_image_formats, default raw,qcow2, for image types
  IPA should accept.

Both of these configuration options are wired up to be set by the lookup
data returned by Ironic at lookup time.

This uses a image format inspection module imported from Nova; this
inspector will eventually live in oslo.utils, at which point we'll
migrate our usage of the inspector to it.

Closes-Bug: #2071740
Change-Id: I5254b80717cb5a7f9084e3eff32a00b968f987b7
This commit is contained in:
Jay Faulkner 2024-07-30 11:18:14 -07:00
parent ba5c1bfe2a
commit e303a369dc
15 changed files with 2703 additions and 133 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
ironic_python_agent/agent.py
View File

@ -467,6 +467,12 @@ class IronicPythonAgent(base.ExecuteCommandMixin):
if config.get('metrics_statsd'):
for opt, val in config.items():
setattr(cfg.CONF.metrics_statsd, opt, val)
if config.get('disable_deep_image_inspection') is not None:
cfg.CONF.set_override('disable_deep_image_inspection',
config['disable_deep_image_inspection'])
if config.get('permitted_image_formats') is not None:
cfg.CONF.set_override('permitted_image_formats',
config['permitted_image_formats'])
md5_allowed = config.get('agent_md5_checksum_enable')
if md5_allowed is not None:
cfg.CONF.set_override('md5_enabled', md5_allowed)

36
ironic_python_agent/config.py
View File

@ -370,6 +370,21 @@ cli_opts = [
help='If the agent should rebuild the configuration drive '
'using a local filesystem, instead of letting Ironic '
'determine if this action is necessary.'),
cfg.BoolOpt('disable_deep_image_inspection',
default=False,
help='This disables the additional deep image inspection '
'the agent does before converting and writing an image. '
'Generally, this should remain enabled for maximum '
'security, but this option allows disabling it if there '
'is a compatibility concern.'),
cfg.ListOpt('permitted_image_formats',
default='raw,qcow2',
help='The supported list of image formats which are '
'permitted for deployment with Ironic Python Agent. If '
'an image format outside of this list is detected, the '
'image validation logic will fail the deployment '
'process. This check is skipped if deep image '
'inspection is disabled.'),
]
disk_utils_opts = [
@ -395,6 +410,13 @@ disk_utils_opts = [
default=10,
help='Maximum number of attempts to try to read the '
'partition.'),
cfg.IntOpt('image_convert_memory_limit',
default=2048,
help='Memory limit for "qemu-img convert" in MiB. Implemented '
'via the address space resource limit.'),
cfg.IntOpt('image_convert_attempts',
default=3,
help='Number of attempts to convert an image.'),
]
disk_part_opts = [
@ -412,10 +434,6 @@ disk_part_opts = [
' having failed.')
]
CONF.register_cli_opts(cli_opts)
CONF.register_opts(disk_utils_opts, group='disk_utils')
CONF.register_opts(disk_part_opts, group='disk_partitioner')
def list_opts():
return [('DEFAULT', cli_opts),
@ -423,6 +441,13 @@ def list_opts():
('disk_partitioner', disk_part_opts)]
def populate_config():
"""Populate configuration. In a method so tests can easily utilize it."""
CONF.register_cli_opts(cli_opts)
CONF.register_opts(disk_utils_opts, group='disk_utils')
CONF.register_opts(disk_part_opts, group='disk_partitioner')
def override(params):
"""Override configuration with values from a dictionary.
@ -447,3 +472,6 @@ def override(params):
LOG.warning('Unable to override configuration option %(key)s '
'with %(value)r: %(exc)s',
{'key': key, 'value': value, 'exc': exc})
populate_config()

111
ironic_python_agent/disk_utils.py
View File

@ -28,7 +28,6 @@ import time
from ironic_lib.common.i18n import _
from ironic_lib import exception
from ironic_lib import qemu_img
from ironic_lib import utils
from oslo_concurrency import processutils
from oslo_config import cfg
@ -36,6 +35,9 @@ from oslo_utils import excutils
import tenacity
from ironic_python_agent import disk_partitioner
from ironic_python_agent import errors
from ironic_python_agent import format_inspector
from ironic_python_agent import qemu_img
CONF = cfg.CONF
@ -389,12 +391,97 @@ def dd(src, dst, conv_flags=None):
*extra_args)
def populate_image(src, dst, conv_flags=None):
data = qemu_img.image_info(src)
if data.file_format == 'raw':
def _image_inspection(filename):
try:
inspector_cls = format_inspector.detect_file_format(filename)
if (not inspector_cls
or not hasattr(inspector_cls, 'safety_check')
or not inspector_cls.safety_check()):
err = "Security: Image failed safety check"
LOG.error(err)
raise errors.InvalidImage(details=err)
except (format_inspector.ImageFormatError, AttributeError):
# NOTE(JayF): Because we already validated the format is OK and matches
# expectation, it should be impossible for us to get an
# ImageFormatError or AttributeError. We handle it anyway
# for completeness.
msg = "Security: Unable to safety check image"
LOG.error(msg)
raise errors.InvalidImage(details=msg)
return inspector_cls
def get_and_validate_image_format(filename, ironic_disk_format):
"""Get the format of a given image file and ensure it's allowed.
This method uses the format inspector originally written for glance to
safely detect the image format. It also sanity checks to ensure any
specified format matches the provided one (except raw; which in some
cases is a request to convert to raw) and that the format is in the
allowed list of formats.
It also performs a basic safety check on the image.
This entire process can be bypassed, and the older code path used,
by setting CONF.disable_deep_image_inspection to True.
See https://bugs.launchpad.net/ironic/+bug/2071740 for full details on
why this must always happen.
:param filename: The name of the image file to validate.
:param ironic_disk_format: The ironic-provided expected format of the image
:returns: tuple of validated img_format (str) and size (int)
"""
if CONF.disable_deep_image_inspection:
data = qemu_img.image_info(filename)
img_format = data.file_format
size = data.virtual_size
else:
if ironic_disk_format == 'raw':
# NOTE(JayF): IPA unconditionally writes raw images to disk without
# conversion with dd or raw python, not qemu-img, it's
# not required to safety check raw images.
img_format = ironic_disk_format
size = os.path.getsize(filename)
else:
img_format_cls = _image_inspection(filename)
img_format = str(img_format_cls)
size = img_format_cls.virtual_size
if img_format not in CONF.permitted_image_formats:
msg = ("Security: Detected image format was %s, but only %s "
"are allowed")
fmts = ', '.join(CONF.permitted_image_formats)
LOG.error(msg, img_format, fmts)
raise errors.InvalidImage(
details=msg % (img_format, fmts)
)
elif ironic_disk_format and ironic_disk_format != img_format:
msg = ("Security: Expected format was %s, but image was "
"actually %s" % (ironic_disk_format, img_format))
LOG.error(msg)
raise errors.InvalidImage(details=msg)
return img_format, size
def populate_image(src, dst, conv_flags=None,
source_format=None, is_raw=False):
"""Populate a provided destination device with the image
:param src: An image already security checked in format disk_format
:param dst: A location, usually a partition or block device,
to write the image
:param conv_flags: Conversion flags to pass to dd if provided
:param source_format: format of the image
:param is_raw: Ironic indicates image is raw; do not convert!
"""
if is_raw:
dd(src, dst, conv_flags=conv_flags)
else:
qemu_img.convert_image(src, dst, 'raw', True, sparse_size='0')
qemu_img.convert_image(src, dst, 'raw', True,
sparse_size='0', source_format=source_format)
def block_uuid(dev):
@ -412,20 +499,6 @@ def block_uuid(dev):
return info.get('PARTUUID', '')
def get_image_mb(image_path, virtual_size=True):
"""Get size of an image in Megabyte."""
mb = 1024 * 1024
if not virtual_size:
image_byte = os.path.getsize(image_path)
else:
data = qemu_img.image_info(image_path)
image_byte = data.virtual_size
# round up size to MB
image_mb = int((image_byte + mb - 1) / mb)
return image_mb
def get_dev_byte_size(dev):
"""Get the device size in bytes."""
byte_sz, cmderr = utils.execute('blockdev', '--getsize64', dev)

9
ironic_python_agent/errors.py
View File

@ -376,3 +376,12 @@ class ProtectedDeviceError(CleaningError):
self.message = details
super(CleaningError, self).__init__(details)
class InvalidImage(DeploymentError):
"""Error raised when an image fails validation for any reason."""
message = 'The provided image is not valid for use'
def __init__(self, details=None):
super(InvalidImage, self).__init__(details)

81
ironic_python_agent/extensions/standby.py
View File

@ -20,10 +20,10 @@ import time
from urllib import parse as urlparse
from ironic_lib import exception
from ironic_lib import qemu_img
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_log import log
from oslo_utils import units
import requests
from ironic_python_agent import disk_utils
@ -31,6 +31,7 @@ from ironic_python_agent import errors
from ironic_python_agent.extensions import base
from ironic_python_agent import hardware
from ironic_python_agent import partition_utils
from ironic_python_agent import qemu_img
from ironic_python_agent import utils
CONF = cfg.CONF
@ -277,7 +278,8 @@ def _fetch_checksum(checksum, image_info):
checksum, "Checksum file does not contain name %s" % expected_fname)
def _write_partition_image(image, image_info, device, configdrive=None):
def _write_partition_image(image, image_info, device, configdrive=None,
source_format=None, is_raw=False, size=0):
"""Call disk_util to create partition and write the partition image.
:param image: Local path to image file to be written to the partition.
@ -288,6 +290,10 @@ def _write_partition_image(image, image_info, device, configdrive=None):
:param configdrive: A string containing the location of the config
drive as a URL OR the contents (as gzip/base64)
of the configdrive. Optional, defaults to None.
:param source_format: The actual format of the partition image.
Must be provided if deep image inspection is enabled.
:param is_raw: Ironic indicates the image is raw; do not convert it
:param size: Virtual size, in MB, of provided image.
:raises: InvalidCommandParamsError if the partition is too small for the
provided image.
@ -307,10 +313,9 @@ def _write_partition_image(image, image_info, device, configdrive=None):
cpu_arch = hardware.dispatch_to_managers('get_cpus').architecture
if image is not None:
image_mb = disk_utils.get_image_mb(image)
if image_mb > int(root_mb):
if size > int(root_mb):
msg = ('Root partition is too small for requested image. Image '
'virtual size: {} MB, Root size: {} MB').format(image_mb,
'virtual size: {} MB, Root size: {} MB').format(size,
root_mb)
raise errors.InvalidCommandParamsError(msg)
@ -324,12 +329,15 @@ def _write_partition_image(image, image_info, device, configdrive=None):
configdrive=configdrive,
boot_mode=boot_mode,
disk_label=disk_label,
cpu_arch=cpu_arch)
cpu_arch=cpu_arch,
source_format=source_format,
is_raw=is_raw)
except processutils.ProcessExecutionError as e:
raise errors.ImageWriteError(device, e.exit_code, e.stdout, e.stderr)
def _write_whole_disk_image(image, image_info, device):
def _write_whole_disk_image(image, image_info, device, source_format=None,
is_raw=False):
"""Writes a whole disk image to the specified device.
:param image: Local path to image file to be written to the disk.
@ -337,22 +345,40 @@ def _write_whole_disk_image(image, image_info, device):
This parameter is currently unused by the function.
:param device: The device name, as a string, on which to store the image.
Example: '/dev/sda'
:param source_format: The format of the whole disk image to be written.
:param is_raw: Ironic indicates the image is raw; do not convert it
:raises: ImageWriteError if the command to write the image encounters an
error.
:raises: InvalidImage if asked to write an image without a format when
not permitted
"""
# FIXME(dtantsur): pass the real node UUID for logging
disk_utils.destroy_disk_metadata(device, '')
disk_utils.udev_settle()
command = ['qemu-img', 'convert',
'-t', 'directsync', '-S', '0', '-O', 'host_device', '-W',
image, device]
LOG.info('Writing image with command: %s', ' '.join(command))
try:
if is_raw:
# TODO(JayF): We should unify all these dd/convert_image calls
# into disk_utils.populate_image().
# NOTE(JayF): Since we do not safety check raw images, we must use
# dd to write them to ensure maximum security. This may cause
# failures in situations where images are configured as raw but
# are actually in need of conversion. Those cases can no longer
# be transparently handled safely.
LOG.info('Writing raw image %s to device %s', image, device)
disk_utils.dd(image, device)
else:
command = ['qemu-img', 'convert',
'-t', 'directsync', '-S', '0', '-O', 'host_device',
'-W']
if source_format:
command += ['-f', source_format]
command += [image, device]
LOG.info('Writing image with command: %s', ' '.join(command))
qemu_img.convert_image(image, device, out_format='host_device',
cache='directsync', out_of_order=True,
sparse_size='0')
sparse_size='0',
source_format=source_format)
except processutils.ProcessExecutionError as e:
raise errors.ImageWriteError(device, e.exit_code, e.stdout, e.stderr)
@ -370,14 +396,28 @@ def _write_image(image_info, device, configdrive=None):
of the configdrive. Optional, defaults to None.
:raises: ImageWriteError if the command to write the image encounters an
error.
:raises: InvalidImage if the image does not pass security inspection
"""
starttime = time.time()
image = _image_location(image_info)
ironic_disk_format = image_info.get('disk_format')
is_raw = ironic_disk_format == 'raw'
# NOTE(JayF): The below method call performs a required security check
# and must remain in place. See bug #2071740
source_format, size = disk_utils.get_and_validate_image_format(
image, ironic_disk_format)
size_mb = int((size + units.Mi - 1) / units.Mi)
uuids = {}
if image_info.get('image_type') == 'partition':
uuids = _write_partition_image(image, image_info, device, configdrive)
uuids = _write_partition_image(image, image_info, device,
configdrive,
source_format=source_format,
is_raw=is_raw, size=size_mb)
else:
_write_whole_disk_image(image, image_info, device)
_write_whole_disk_image(image, image_info, device,
source_format=source_format,
is_raw=is_raw)
totaltime = time.time() - starttime
LOG.info('Image %(image)s written to device %(device)s in %(totaltime)s '
'seconds', {'image': image, 'device': device,
@ -907,16 +947,20 @@ class StandbyExtension(base.BaseAgentExtension):
device = hardware.dispatch_to_managers('get_os_install_device',
permit_refresh=True)
disk_format = image_info.get('disk_format')
requested_disk_format = image_info.get('disk_format')
stream_raw_images = image_info.get('stream_raw_images', False)
# don't write image again if already cached
if self.cached_image_id != image_info['id']:
if self.cached_image_id is not None:
LOG.debug('Already had %s cached, overwriting',
self.cached_image_id)
if stream_raw_images and disk_format == 'raw':
if stream_raw_images and requested_disk_format == 'raw':
if image_info.get('image_type') == 'partition':
# NOTE(JayF): This only creates partitions due to image
# being None
self.partition_uuids = _write_partition_image(None,
image_info,
device,
@ -926,6 +970,9 @@ class StandbyExtension(base.BaseAgentExtension):
self.partition_uuids = {}
stream_to = device
# NOTE(JayF): Images that claim to be raw are not inspected at
# all, as they never interact with qemu-img and are
# streamed directly to disk unmodified.
self._stream_raw_image_onto_device(image_info, stream_to)
else:
self._cache_and_write_image(image_info, device, configdrive)

1044
ironic_python_agent/format_inspector.py Normal file
View File

File diff suppressed because it is too large Load Diff

9
ironic_python_agent/partition_utils.py
View File

@ -187,7 +187,8 @@ def get_labelled_partition(device_path, label, node_uuid):
def work_on_disk(dev, root_mb, swap_mb, ephemeral_mb, ephemeral_format,
image_path, node_uuid, preserve_ephemeral=False,
configdrive=None, boot_mode="bios",
tempdir=None, disk_label=None, cpu_arch="", conv_flags=None):
tempdir=None, disk_label=None, cpu_arch="", conv_flags=None,
source_format=None, is_raw=False):
"""Create partitions and copy an image to the root partition.
:param dev: Path for the device to work on.
@ -218,6 +219,9 @@ def work_on_disk(dev, root_mb, swap_mb, ephemeral_mb, ephemeral_format,
:param conv_flags: Flags that need to be sent to the dd command, to control
the conversion of the original file when copying to the host. It can
contain several options separated by commas.
:param source_format: The format of the disk image to be written.
If set, must be "raw" or the actual disk format of the image.
:param is_raw: Ironic indicator image is raw; not to be converted
:returns: a dictionary containing the following keys:
'root uuid': UUID of root partition
'efi system partition uuid': UUID of the uefi system partition
@ -295,7 +299,8 @@ def work_on_disk(dev, root_mb, swap_mb, ephemeral_mb, ephemeral_format,
utils.unlink_without_raise(configdrive_file)
if image_path is not None:
disk_utils.populate_image(image_path, root_part, conv_flags=conv_flags)
disk_utils.populate_image(image_path, root_part, conv_flags=conv_flags,
source_format=source_format, is_raw=is_raw)
LOG.info("Image for %(node)s successfully populated",
{'node': node_uuid})
else:

153
ironic_python_agent/qemu_img.py Normal file
View File

@ -0,0 +1,153 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import logging
import os
from ironic_lib import utils
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils import imageutils
from oslo_utils import units
import tenacity
from ironic_python_agent import errors
"""
Imported from ironic_lib/qemu-img.py from commit
c3d59dfffc9804273b49c0556ee09419a35917c1
See https://bugs.launchpad.net/ironic/+bug/2071740 for more details as to why
it moved.
This module also exists in the Ironic repo. Do not modify this module
without also modifying that module.
"""
CONF = cfg.CONF
LOG = logging.getLogger(__name__)
# Limit the memory address space to 1 GiB when running qemu-img
QEMU_IMG_LIMITS = None
def _qemu_img_limits():
global QEMU_IMG_LIMITS
if QEMU_IMG_LIMITS is None:
QEMU_IMG_LIMITS = processutils.ProcessLimits(
address_space=CONF.disk_utils.image_convert_memory_limit
* units.Mi)
return QEMU_IMG_LIMITS
def _retry_on_res_temp_unavailable(exc):
if (isinstance(exc, processutils.ProcessExecutionError)
and ('Resource temporarily unavailable' in exc.stderr
or 'Cannot allocate memory' in exc.stderr)):
return True
return False
def image_info(path, source_format=None):
"""Return an object containing the parsed output from qemu-img info.
This must only be called on images already validated as safe by the
format inspector.
:param path: The path to an image you need information on
:param source_format: The format of the source image. If this is omitted
when deep inspection is enabled, this will raise
InvalidImage.
"""
# NOTE(JayF): This serves as a final exit hatch: if we have deep
# image inspection enabled, but someone calls this method without an
# explicit disk_format, there's no way for us to do the call securely.
if not source_format and not CONF.disable_deep_image_inspection:
msg = ("Security: qemu_img.image_info called unsafely while deep "
"image inspection is enabled. This should not be possible, "
"please contact Ironic developers.")
raise errors.InvalidImage(details=msg)
if not os.path.exists(path):
raise FileNotFoundError("File %s does not exist" % path)
cmd = [
'env', 'LC_ALL=C', 'LANG=C',
'qemu-img', 'info', path,
'--output=json'
]
if source_format:
cmd += ['-f', source_format]
out, err = utils.execute(cmd, prlimit=_qemu_img_limits())
return imageutils.QemuImgInfo(out, format='json')
@tenacity.retry(
retry=tenacity.retry_if_exception(_retry_on_res_temp_unavailable),
stop=tenacity.stop_after_attempt(CONF.disk_utils.image_convert_attempts),
reraise=True)
def convert_image(source, dest, out_format, run_as_root=False, cache=None,
out_of_order=False, sparse_size=None, source_format=None):
"""Convert image to other format.
This method is only to be run against images who have passed
format_inspector's safety check, and with the format reported by it
passed in. Any other usage is a major security risk.
"""
cmd = ['qemu-img', 'convert', '-O', out_format]
if cache is not None:
cmd += ['-t', cache]
if sparse_size is not None:
cmd += ['-S', sparse_size]
if source_format is not None:
cmd += ['-f', source_format]
elif not CONF.disable_deep_image_inspection:
# NOTE(JayF): This serves as a final exit hatch: if we have deep
# image inspection enabled, but someone calls this method without an
# explicit disk_format, there's no way for us to do the conversion
# securely.
msg = ("Security: qemu_img.convert_image called unsafely while deep "
"image inspection is enabled. This should not be possible, "
"please notify Ironic developers.")
LOG.error(msg)
raise errors.InvalidImage(details=msg)
if out_of_order:
cmd.append('-W')
cmd += [source, dest]
# NOTE(TheJulia): Statically set the MALLOC_ARENA_MAX to prevent leaking
# and the creation of new malloc arenas which will consume the system
# memory. If limited to 1, qemu-img consumes ~250 MB of RAM, but when
# another thread tries to access a locked section of memory in use with
# another thread, then by default a new malloc arena is created,
# which essentially balloons the memory requirement of the machine.
# Default for qemu-img is 8 * nCPU * ~250MB (based on defaults +
# thread/code/process/library overhead. In other words, 64 GB. Limiting
# this to 3 keeps the memory utilization in happy cases below the overall
# threshold which is in place in case a malicious image is attempted to
# be passed through qemu-img.
env_vars = {'MALLOC_ARENA_MAX': '3'}
try:
utils.execute(*cmd, run_as_root=run_as_root,
prlimit=_qemu_img_limits(),
use_standard_locale=True,
env_variables=env_vars)
except processutils.ProcessExecutionError as e:
if ('Resource temporarily unavailable' in e.stderr
or 'Cannot allocate memory' in e.stderr):
LOG.debug('Failed to convert image, retrying. Error: %s', e)
# Sync disk caches before the next attempt
utils.execute('sync')
raise

2
ironic_python_agent/tests/unit/base.py
View File

@ -25,6 +25,7 @@ from oslo_log import log
from oslo_service import sslutils
from oslotest import base as test_base
from ironic_python_agent import config
from ironic_python_agent.extensions import base as ext_base
from ironic_python_agent import hardware
@ -40,6 +41,7 @@ class IronicAgentTest(test_base.BaseTestCase):
def setUp(self):
super(IronicAgentTest, self).setUp()
config.populate_config()
self._set_config()
# Ban running external processes via 'execute' like functions. If the

119
ironic_python_agent/tests/unit/extensions/test_standby.py
View File

@ -20,6 +20,7 @@ from unittest import mock
from ironic_lib import exception
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils import units
import requests
from ironic_python_agent import errors
@ -33,6 +34,11 @@ from ironic_python_agent import utils
CONF = cfg.CONF
def _virtual_size(size=1):
"""Convert a virtual size in mb to bytes"""
return (size * units.Mi) + 1 - units.Mi
def _build_fake_image_info(url='http://example.org'):
return {
'id': 'fake_id',
@ -41,6 +47,7 @@ def _build_fake_image_info(url='http://example.org'):
'image_type': 'whole-disk-image',
'os_hash_algo': 'sha256',
'os_hash_value': 'fake-checksum',
'disk_format': 'qcow2'
}
@ -60,7 +67,9 @@ def _build_fake_partition_image_info():
'disk_label': 'msdos',
'deploy_boot_mode': 'bios',
'os_hash_algo': 'sha256',
'os_hash_value': 'fake-checksum'}
'os_hash_value': 'fake-checksum',
'disk_format': 'qcow2'
}
class TestStandbyExtension(base.IronicAgentTest):
@ -279,18 +288,23 @@ class TestStandbyExtension(base.IronicAgentTest):
None,
image_info['id'])
@mock.patch(
'ironic_python_agent.disk_utils.get_and_validate_image_format',
autospec=True)
@mock.patch('ironic_python_agent.disk_utils.fix_gpt_partition',
autospec=True)
@mock.patch('ironic_python_agent.disk_utils.trigger_device_rescan',
autospec=True)
@mock.patch('ironic_lib.qemu_img.convert_image', autospec=True)
@mock.patch('ironic_python_agent.qemu_img.convert_image', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.udev_settle', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.destroy_disk_metadata',
autospec=True)
def test_write_image(self, wipe_mock, udev_mock, convert_mock,
rescan_mock, fix_gpt_mock):
rescan_mock, fix_gpt_mock, validate_mock):
image_info = _build_fake_image_info()
device = '/dev/sda'
source_format = image_info['disk_format']
validate_mock.return_value = (source_format, 0)
location = standby._image_location(image_info)
standby._write_image(image_info, device)
@ -299,7 +313,9 @@ class TestStandbyExtension(base.IronicAgentTest):
out_format='host_device',
cache='directsync',
out_of_order=True,
sparse_size='0')
sparse_size='0',
source_format=source_format)
validate_mock.assert_called_once_with(location, source_format)
wipe_mock.assert_called_once_with(device, '')
udev_mock.assert_called_once_with()
rescan_mock.assert_called_once_with(device)
@ -309,24 +325,33 @@ class TestStandbyExtension(base.IronicAgentTest):
autospec=True)
@mock.patch('ironic_python_agent.disk_utils.trigger_device_rescan',
autospec=True)
@mock.patch('ironic_lib.qemu_img.convert_image', autospec=True)
@mock.patch('ironic_python_agent.qemu_img.convert_image', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.udev_settle', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.destroy_disk_metadata',
autospec=True)
def test_write_image_gpt_fails(self, wipe_mock, udev_mock, convert_mock,
rescan_mock, fix_gpt_mock):
image_info = _build_fake_image_info()
@mock.patch(
'ironic_python_agent.disk_utils.get_and_validate_image_format',
autospec=True)
def test_write_image_gpt_fails(self, validate_mock, wipe_mock, udev_mock,
convert_mock, rescan_mock, fix_gpt_mock):
device = '/dev/sda'
image_info = _build_fake_image_info()
validate_mock.return_value = (image_info['disk_format'], 0)
fix_gpt_mock.side_effect = exception.InstanceDeployFailure
standby._write_image(image_info, device)
@mock.patch('ironic_lib.qemu_img.convert_image', autospec=True)
@mock.patch('ironic_python_agent.qemu_img.convert_image', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.udev_settle', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.destroy_disk_metadata',
autospec=True)
def test_write_image_fails(self, wipe_mock, udev_mock, convert_mock):
@mock.patch(
'ironic_python_agent.disk_utils.get_and_validate_image_format',
autospec=True)
def test_write_image_fails(self, validate_mock, wipe_mock, udev_mock,
convert_mock):
image_info = _build_fake_image_info()
validate_mock.return_value = (image_info['disk_format'], 0)
device = '/dev/sda'
convert_mock.side_effect = processutils.ProcessExecutionError
@ -339,10 +364,12 @@ class TestStandbyExtension(base.IronicAgentTest):
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.get_image_mb', autospec=True)
@mock.patch(
'ironic_python_agent.disk_utils.get_and_validate_image_format',
autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
def test_write_partition_image_exception(self, work_on_disk_mock,
image_mb_mock,
validate_mock,
execute_mock, open_mock,
dispatch_mock):
image_info = _build_fake_partition_image_info()
@ -355,11 +382,13 @@ class TestStandbyExtension(base.IronicAgentTest):
pr_ep = image_info['preserve_ephemeral']
boot_mode = image_info['deploy_boot_mode']
disk_label = image_info['disk_label']
source_format = image_info['disk_format']
cpu_arch = self.fake_cpu.architecture
image_path = standby._image_location(image_info)
image_mb_mock.return_value = 1
validate_mock.return_value = (image_info['disk_format'],
_virtual_size(1))
dispatch_mock.return_value = self.fake_cpu
exc = errors.ImageWriteError
Exception_returned = processutils.ProcessExecutionError
@ -367,7 +396,7 @@ class TestStandbyExtension(base.IronicAgentTest):
self.assertRaises(exc, standby._write_image, image_info,
device, 'configdrive')
image_mb_mock.assert_called_once_with(image_path)
validate_mock.assert_called_once_with(image_path, source_format)
work_on_disk_mock.assert_called_once_with(device, root_mb, swap_mb,
ephemeral_mb,
ephemeral_format,
@ -377,16 +406,20 @@ class TestStandbyExtension(base.IronicAgentTest):
preserve_ephemeral=pr_ep,
boot_mode=boot_mode,
disk_label=disk_label,
cpu_arch=cpu_arch)
cpu_arch=cpu_arch,
source_format=source_format,
is_raw=False)
@mock.patch.object(utils, 'get_node_boot_mode', lambda self: 'bios')
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.get_image_mb', autospec=True)
@mock.patch(
'ironic_python_agent.disk_utils.get_and_validate_image_format',
autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
def test_write_partition_image_no_node_uuid(self, work_on_disk_mock,
image_mb_mock,
validate_mock,
execute_mock, open_mock,
dispatch_mock):
image_info = _build_fake_partition_image_info()
@ -400,19 +433,19 @@ class TestStandbyExtension(base.IronicAgentTest):
pr_ep = image_info['preserve_ephemeral']
boot_mode = image_info['deploy_boot_mode']
disk_label = image_info['disk_label']
source_format = image_info['disk_format']
cpu_arch = self.fake_cpu.architecture
image_path = standby._image_location(image_info)
image_mb_mock.return_value = 1
validate_mock.return_value = (source_format, _virtual_size(1))
dispatch_mock.return_value = self.fake_cpu
uuids = {'root uuid': 'root_uuid'}
expected_uuid = {'root uuid': 'root_uuid'}
image_mb_mock.return_value = 1
work_on_disk_mock.return_value = uuids
standby._write_image(image_info, device, 'configdrive')
image_mb_mock.assert_called_once_with(image_path)
validate_mock.assert_called_once_with(image_path, source_format)
work_on_disk_mock.assert_called_once_with(device, root_mb, swap_mb,
ephemeral_mb,
ephemeral_format,
@ -422,7 +455,9 @@ class TestStandbyExtension(base.IronicAgentTest):
preserve_ephemeral=pr_ep,
boot_mode=boot_mode,
disk_label=disk_label,
cpu_arch=cpu_arch)
cpu_arch=cpu_arch,
source_format=source_format,
is_raw=False)
self.assertEqual(expected_uuid, work_on_disk_mock.return_value)
self.assertIsNone(node_uuid)
@ -430,26 +465,29 @@ class TestStandbyExtension(base.IronicAgentTest):
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.get_image_mb', autospec=True)
@mock.patch(
'ironic_python_agent.disk_utils.get_and_validate_image_format',
autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
def test_write_partition_image_exception_image_mb(self,
work_on_disk_mock,
image_mb_mock,
validate_mock,
execute_mock,
open_mock,
dispatch_mock):
dispatch_mock.return_value = self.fake_cpu
image_info = _build_fake_partition_image_info()
device = '/dev/sda'
source_format = image_info['disk_format']
image_path = standby._image_location(image_info)
image_mb_mock.return_value = 20
validate_mock.return_value = (source_format, _virtual_size(20))
exc = errors.InvalidCommandParamsError
self.assertRaises(exc, standby._write_image, image_info,
device)
image_mb_mock.assert_called_once_with(image_path)
validate_mock.assert_called_once_with(image_path, source_format)
self.assertFalse(work_on_disk_mock.called)
@mock.patch.object(utils, 'get_node_boot_mode', lambda self: 'bios')
@ -457,8 +495,10 @@ class TestStandbyExtension(base.IronicAgentTest):
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.get_image_mb', autospec=True)
def test_write_partition_image(self, image_mb_mock, work_on_disk_mock,
@mock.patch(
'ironic_python_agent.disk_utils.get_and_validate_image_format',
autospec=True)
def test_write_partition_image(self, validate_mock, work_on_disk_mock,
execute_mock, open_mock, dispatch_mock):
image_info = _build_fake_partition_image_info()
device = '/dev/sda'
@ -470,17 +510,18 @@ class TestStandbyExtension(base.IronicAgentTest):
pr_ep = image_info['preserve_ephemeral']
boot_mode = image_info['deploy_boot_mode']
disk_label = image_info['disk_label']
source_format = image_info['disk_format']
cpu_arch = self.fake_cpu.architecture
image_path = standby._image_location(image_info)
uuids = {'root uuid': 'root_uuid'}
expected_uuid = {'root uuid': 'root_uuid'}
image_mb_mock.return_value = 1
validate_mock.return_value = (source_format, _virtual_size(1))
dispatch_mock.return_value = self.fake_cpu
work_on_disk_mock.return_value = uuids
standby._write_image(image_info, device, 'configdrive')
image_mb_mock.assert_called_once_with(image_path)
validate_mock.assert_called_once_with(image_path, source_format)
work_on_disk_mock.assert_called_once_with(device, root_mb, swap_mb,
ephemeral_mb,
ephemeral_format,
@ -490,7 +531,9 @@ class TestStandbyExtension(base.IronicAgentTest):
preserve_ephemeral=pr_ep,
boot_mode=boot_mode,
disk_label=disk_label,
cpu_arch=cpu_arch)
cpu_arch=cpu_arch,
source_format=source_format,
is_raw=False)
self.assertEqual(expected_uuid, work_on_disk_mock.return_value)
@ -1578,11 +1621,13 @@ class TestStandbyExtension(base.IronicAgentTest):
@mock.patch.object(hardware, 'dispatch_to_managers', autospec=True)
@mock.patch('builtins.open', autospec=True)
@mock.patch('ironic_python_agent.utils.execute', autospec=True)
@mock.patch('ironic_python_agent.disk_utils.get_image_mb', autospec=True)
@mock.patch(
'ironic_python_agent.disk_utils.get_and_validate_image_format',
autospec=True)
@mock.patch.object(partition_utils, 'work_on_disk', autospec=True)
def test_write_partition_image_no_node_uuid_uefi(
self, work_on_disk_mock,
image_mb_mock,
validate_mock,
execute_mock, open_mock,
dispatch_mock):
image_info = _build_fake_partition_image_info()
@ -1594,19 +1639,19 @@ class TestStandbyExtension(base.IronicAgentTest):
ephemeral_format = image_info['ephemeral_format']
node_uuid = image_info['node_uuid']
pr_ep = image_info['preserve_ephemeral']
source_format = image_info['disk_format']
validate_mock.return_value = (source_format, _virtual_size(1))
cpu_arch = self.fake_cpu.architecture
image_path = standby._image_location(image_info)
image_mb_mock.return_value = 1
dispatch_mock.return_value = self.fake_cpu
uuids = {'root uuid': 'root_uuid'}
expected_uuid = {'root uuid': 'root_uuid'}
image_mb_mock.return_value = 1
work_on_disk_mock.return_value = uuids
standby._write_image(image_info, device, 'configdrive')
image_mb_mock.assert_called_once_with(image_path)
validate_mock.assert_called_once_with(image_path, source_format)
work_on_disk_mock.assert_called_once_with(device, root_mb, swap_mb,
ephemeral_mb,
ephemeral_format,
@ -1616,7 +1661,9 @@ class TestStandbyExtension(base.IronicAgentTest):
preserve_ephemeral=pr_ep,
boot_mode='uefi',
disk_label='gpt',
cpu_arch=cpu_arch)
cpu_arch=cpu_arch,
source_format=source_format,
is_raw=False)
self.assertEqual(expected_uuid, work_on_disk_mock.return_value)
self.assertIsNone(node_uuid)

190
ironic_python_agent/tests/unit/test_disk_utils.py
View File

@ -13,22 +13,54 @@
# License for the specific language governing permissions and limitations
# under the License.
import json
import os
import stat
from unittest import mock
from ironic_lib import exception
from ironic_lib import qemu_img
from ironic_lib.tests import base
from ironic_lib import utils
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils.imageutils import QemuImgInfo
from oslo_utils import units
from ironic_python_agent import disk_utils
from ironic_python_agent.errors import InvalidImage
from ironic_python_agent import format_inspector
from ironic_python_agent import qemu_img
CONF = cfg.CONF
class MockFormatInspectorCls(object):
def __init__(self, img_format='qcow2', virtual_size_mb=0, safe=False):
self.img_format = img_format
self.virtual_size_mb = virtual_size_mb
self.safe = safe
def __str__(self):
return self.img_format
@property
def virtual_size(self):
# NOTE(JayF): Allow the mock-user to input MBs but
# backwards-calculate so code in _write_image can still work
if self.virtual_size_mb == 0:
return 0
else:
return (self.virtual_size_mb * units.Mi) + 1 - units.Mi
def safety_check(self):
return self.safe
def _get_fake_qemu_image_info(file_format='qcow2', virtual_size=0):
fake_data = {'format': file_format, 'virtual-size': virtual_size, }
return QemuImgInfo(cmd_output=json.dumps(fake_data), format='json')
@mock.patch.object(utils, 'execute', autospec=True)
class ListPartitionsTestCase(base.IronicLibTestCase):
@ -484,31 +516,24 @@ class GetDeviceByteSizeTestCase(base.IronicLibTestCase):
@mock.patch.object(disk_utils, 'dd', autospec=True)
@mock.patch.object(qemu_img, 'image_info', autospec=True)
@mock.patch.object(qemu_img, 'convert_image', autospec=True)
class PopulateImageTestCase(base.IronicLibTestCase):
def test_populate_raw_image(self, mock_cg, mock_qinfo, mock_dd):
type(mock_qinfo.return_value).file_format = mock.PropertyMock(
return_value='raw')
disk_utils.populate_image('src', 'dst')
def test_populate_raw_image(self, mock_cg, mock_dd):
source_format = 'raw'
disk_utils.populate_image('src', 'dst',
source_format=source_format,
is_raw=True)
mock_dd.assert_called_once_with('src', 'dst', conv_flags=None)
self.assertFalse(mock_cg.called)
def test_populate_raw_image_with_convert(self, mock_cg, mock_qinfo,
mock_dd):
type(mock_qinfo.return_value).file_format = mock.PropertyMock(
return_value='raw')
disk_utils.populate_image('src', 'dst', conv_flags='sparse')
mock_dd.assert_called_once_with('src', 'dst', conv_flags='sparse')
self.assertFalse(mock_cg.called)
def test_populate_qcow2_image(self, mock_cg, mock_qinfo, mock_dd):
type(mock_qinfo.return_value).file_format = mock.PropertyMock(
return_value='qcow2')
disk_utils.populate_image('src', 'dst')
def test_populate_qcow2_image(self, mock_cg, mock_dd):
source_format = 'qcow2'
disk_utils.populate_image('src', 'dst',
source_format=source_format, is_raw=False)
mock_cg.assert_called_once_with('src', 'dst', 'raw', True,
sparse_size='0')
sparse_size='0',
source_format=source_format)
self.assertFalse(mock_dd.called)
@ -542,32 +567,6 @@ class OtherFunctionTestCase(base.IronicLibTestCase):
disk_utils.is_block_device, device)
mock_os.assert_has_calls([mock.call(device)] * 2)
@mock.patch.object(os.path, 'getsize', autospec=True)
@mock.patch.object(qemu_img, 'image_info', autospec=True)
def test_get_image_mb(self, mock_qinfo, mock_getsize):
mb = 1024 * 1024
mock_getsize.return_value = 0
type(mock_qinfo.return_value).virtual_size = mock.PropertyMock(
return_value=0)
self.assertEqual(0, disk_utils.get_image_mb('x', False))
self.assertEqual(0, disk_utils.get_image_mb('x', True))
mock_getsize.return_value = 1
type(mock_qinfo.return_value).virtual_size = mock.PropertyMock(
return_value=1)
self.assertEqual(1, disk_utils.get_image_mb('x', False))
self.assertEqual(1, disk_utils.get_image_mb('x', True))
mock_getsize.return_value = mb
type(mock_qinfo.return_value).virtual_size = mock.PropertyMock(
return_value=mb)
self.assertEqual(1, disk_utils.get_image_mb('x', False))
self.assertEqual(1, disk_utils.get_image_mb('x', True))
mock_getsize.return_value = mb + 1
type(mock_qinfo.return_value).virtual_size = mock.PropertyMock(
return_value=mb + 1)
self.assertEqual(2, disk_utils.get_image_mb('x', False))
self.assertEqual(2, disk_utils.get_image_mb('x', True))
def _test_count_mbr_partitions(self, output, mock_execute):
mock_execute.return_value = (output, '')
out = disk_utils.count_mbr_partitions('/dev/fake')
@ -960,3 +959,104 @@ class WaitForDisk(base.IronicLibTestCase):
fuser_call = mock.call(*fuser_cmd, check_exit_code=[0, 1])
self.assertEqual(2, mock_exc.call_count)
mock_exc.assert_has_calls([fuser_call, fuser_call])
class GetAndValidateImageFormat(base.IronicLibTestCase):
@mock.patch.object(disk_utils, '_image_inspection', autospec=True)
@mock.patch('os.path.getsize', autospec=True)
def test_happy_raw(self, mock_size, mock_ii):
"""Valid raw image"""
CONF.set_override('disable_deep_image_inspection', False)
mock_size.return_value = 13
fmt = 'raw'
self.assertEqual(
(fmt, 13),
disk_utils.get_and_validate_image_format('/fake/path', fmt))
mock_ii.assert_not_called()
mock_size.assert_called_once_with('/fake/path')
@mock.patch.object(disk_utils, '_image_inspection', autospec=True)
def test_happy_qcow2(self, mock_ii):
"""Valid qcow2 image"""
CONF.set_override('disable_deep_image_inspection', False)
fmt = 'qcow2'
mock_ii.return_value = MockFormatInspectorCls(fmt, 0, True)
self.assertEqual(
(fmt, 0),
disk_utils.get_and_validate_image_format('/fake/path', fmt)
)
mock_ii.assert_called_once_with('/fake/path')
@mock.patch.object(disk_utils, '_image_inspection', autospec=True)
def test_format_type_disallowed(self, mock_ii):
"""qcow3 images are not allowed in default config"""
CONF.set_override('disable_deep_image_inspection', False)
fmt = 'qcow3'
mock_ii.return_value = MockFormatInspectorCls(fmt, 0, True)
self.assertRaises(InvalidImage,
disk_utils.get_and_validate_image_format,
'/fake/path', fmt)
mock_ii.assert_called_once_with('/fake/path')
@mock.patch.object(disk_utils, '_image_inspection', autospec=True)
def test_format_mismatch(self, mock_ii):
"""ironic_disk_format=qcow2, but we detect it as a qcow3"""
CONF.set_override('disable_deep_image_inspection', False)
fmt = 'qcow2'
mock_ii.return_value = MockFormatInspectorCls('qcow3', 0, True)
self.assertRaises(InvalidImage,
disk_utils.get_and_validate_image_format,
'/fake/path', fmt)
@mock.patch.object(disk_utils, '_image_inspection', autospec=True)
@mock.patch.object(qemu_img, 'image_info', autospec=True)
def test_format_mismatch_but_disabled(self, mock_info, mock_ii):
"""qcow3 ironic_disk_format ignored because deep inspection disabled"""
CONF.set_override('disable_deep_image_inspection', True)
fmt = 'qcow2'
fake_info = _get_fake_qemu_image_info(file_format=fmt, virtual_size=0)
qemu_img.image_info.return_value = fake_info
# note the input is qcow3, the output is qcow2: this mismatch is
# forbidden if CONF.disable_deep_image_inspection is False
self.assertEqual(
(fmt, 0),
disk_utils.get_and_validate_image_format('/fake/path', 'qcow3'))
mock_ii.assert_not_called()
mock_info.assert_called_once()
@mock.patch.object(disk_utils, '_image_inspection', autospec=True)
@mock.patch.object(qemu_img, 'image_info', autospec=True)
def test_safety_check_fail_but_disabled(self, mock_info, mock_ii):
"""unsafe image ignored because inspection is disabled"""
CONF.set_override('disable_deep_image_inspection', True)
fmt = 'qcow2'
fake_info = _get_fake_qemu_image_info(file_format=fmt, virtual_size=0)
qemu_img.image_info.return_value = fake_info
# note the input is qcow3, the output is qcow2: this mismatch is
# forbidden if CONF.disable_deep_image_inspection is False
self.assertEqual(
(fmt, 0),
disk_utils.get_and_validate_image_format('/fake/path', 'qcow3'))
mock_ii.assert_not_called()
mock_info.assert_called_once()
class ImageInspectionTest(base.IronicLibTestCase):
@mock.patch.object(format_inspector, 'detect_file_format', autospec=True)
def test_image_inspection_pass(self, mock_fi):
inspector = MockFormatInspectorCls('qcow2', 0, True)
mock_fi.return_value = inspector
self.assertEqual(inspector, disk_utils._image_inspection('/fake/path'))
@mock.patch.object(format_inspector, 'detect_file_format', autospec=True)
def test_image_inspection_fail_safety_check(self, mock_fi):
inspector = MockFormatInspectorCls('qcow2', 0, False)
mock_fi.return_value = inspector
self.assertRaises(InvalidImage, disk_utils._image_inspection,
'/fake/path')
@mock.patch.object(format_inspector, 'detect_file_format', autospec=True)
def test_image_inspection_fail_format_error(self, mock_fi):
mock_fi.side_effect = format_inspector.ImageFormatError
self.assertRaises(InvalidImage, disk_utils._image_inspection,
'/fake/path')

664
ironic_python_agent/tests/unit/test_format_inspector.py Normal file
View File

@ -0,0 +1,664 @@
# Copyright 2020 Red Hat, Inc
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import io
import os
import re
import struct
import subprocess
import tempfile
from unittest import mock
from oslo_utils import units
from ironic_python_agent import format_inspector
from ironic_python_agent.tests.unit.base import IronicAgentTest
TEST_IMAGE_PREFIX = 'ironic-unittest-formatinspector-'
def get_size_from_qemu_img(filename):
output = subprocess.check_output('qemu-img info "%s"' % filename,
shell=True)
for line in output.split(b'\n'):
m = re.search(b'^virtual size: .* .([0-9]+) bytes', line.strip())
if m:
return int(m.group(1))
raise Exception('Could not find virtual size with qemu-img')
class TestFormatInspectors(IronicAgentTest):
block_execute = False
def setUp(self):
super(TestFormatInspectors, self).setUp()
self._created_files = []
def tearDown(self):
super(TestFormatInspectors, self).tearDown()
for fn in self._created_files:
try:
os.remove(fn)
except Exception:
pass
def _create_iso(self, image_size, subformat='9660'):
"""Create an ISO file of the given size.
:param image_size: The size of the image to create in bytes
:param subformat: The subformat to use, if any
"""
# these tests depend on mkisofs
# being installed and in the path,
# if it is not installed, skip
try:
subprocess.check_output('mkisofs --version', shell=True)
except Exception:
self.skipTest('mkisofs not installed')
size = image_size // units.Mi
base_cmd = "mkisofs"
if subformat == 'udf':
# depending on the distribution mkisofs may not support udf
# and may be provided by genisoimage instead. As a result we
# need to check if the command supports udf via help
# instead of checking the installed version.
# mkisofs --help outputs to stderr so we need to
# redirect it to stdout to use grep.
try:
subprocess.check_output(
'mkisofs --help 2>&1 | grep udf', shell=True)
except Exception:
self.skipTest('mkisofs does not support udf format')
base_cmd += " -udf"
prefix = TEST_IMAGE_PREFIX
prefix += '-%s-' % subformat
fn = tempfile.mktemp(prefix=prefix, suffix='.iso')
self._created_files.append(fn)
subprocess.check_output(
'dd if=/dev/zero of=%s bs=1M count=%i' % (fn, size),
shell=True)
# We need to use different file as input and output as the behavior
# of mkisofs is version dependent if both the input and the output
# are the same and can cause test failures
out_fn = "%s.iso" % fn
subprocess.check_output(
'%s -V "TEST" -o %s %s' % (base_cmd, out_fn, fn),
shell=True)
self._created_files.append(out_fn)
return out_fn
def _create_img(
self, fmt, size, subformat=None, options=None,
backing_file=None):
"""Create an image file of the given format and size.
:param fmt: The format to create
:param size: The size of the image to create in bytes
:param subformat: The subformat to use, if any
:param options: A dictionary of options to pass to the format
:param backing_file: The backing file to use, if any
"""
if fmt == 'iso':
return self._create_iso(size, subformat)
if fmt == 'vhd':
# QEMU calls the vhd format vpc
fmt = 'vpc'
# these tests depend on qemu-img being installed and in the path,
# if it is not installed, skip. we also need to ensure that the
# format is supported by qemu-img, this can vary depending on the
# distribution so we need to check if the format is supported via
# the help output.
try:
subprocess.check_output(
'qemu-img --help | grep %s' % fmt, shell=True)
except Exception:
self.skipTest(
'qemu-img not installed or does not support %s format' % fmt)
if options is None:
options = {}
opt = ''
prefix = TEST_IMAGE_PREFIX
if subformat:
options['subformat'] = subformat
prefix += subformat + '-'
if options:
opt += '-o ' + ','.join('%s=%s' % (k, v)
for k, v in options.items())
if backing_file is not None:
opt += ' -b %s -F raw' % backing_file
fn = tempfile.mktemp(prefix=prefix,
suffix='.%s' % fmt)
self._created_files.append(fn)
subprocess.check_output(
'qemu-img create -f %s %s %s %i' % (fmt, opt, fn, size),
shell=True)
return fn
def _create_allocated_vmdk(self, size_mb, subformat=None):
# We need a "big" VMDK file to exercise some parts of the code of the
# format_inspector. A way to create one is to first create an empty
# file, and then to convert it with the -S 0 option.
if subformat is None:
# Matches qemu-img default, see `qemu-img convert -O vmdk -o help`
subformat = 'monolithicSparse'
prefix = TEST_IMAGE_PREFIX
prefix += '-%s-' % subformat
fn = tempfile.mktemp(prefix=prefix, suffix='.vmdk')
self._created_files.append(fn)
raw = tempfile.mktemp(prefix=prefix, suffix='.raw')
self._created_files.append(raw)
# Create a file with pseudo-random data, otherwise it will get
# compressed in the streamOptimized format
subprocess.check_output(
'dd if=/dev/urandom of=%s bs=1M count=%i' % (raw, size_mb),
shell=True)
# Convert it to VMDK
subprocess.check_output(
'qemu-img convert -f raw -O vmdk -o subformat=%s -S 0 %s %s' % (
subformat, raw, fn),
shell=True)
return fn
def _test_format_at_block_size(self, format_name, img, block_size):
fmt = format_inspector.get_inspector(format_name)()
self.assertIsNotNone(fmt,
'Did not get format inspector for %s' % (
format_name))
wrapper = format_inspector.InfoWrapper(open(img, 'rb'), fmt)
while True:
chunk = wrapper.read(block_size)
if not chunk:
break
wrapper.close()
return fmt
def _test_format_at_image_size(self, format_name, image_size,
subformat=None):
"""Test the format inspector for the given format at the given image size.
:param format_name: The format to test
:param image_size: The size of the image to create in bytes
:param subformat: The subformat to use, if any
""" # noqa
img = self._create_img(format_name, image_size, subformat=subformat)
# Some formats have internal alignment restrictions making this not
# always exactly like image_size, so get the real value for comparison
virtual_size = get_size_from_qemu_img(img)
# Read the format in various sizes, some of which will read whole
# sections in a single read, others will be completely unaligned, etc.
block_sizes = [64 * units.Ki, 1 * units.Mi]
# ISO images have a 32KB system area at the beginning of the image
# as a result reading that in 17 or 512 byte blocks takes too long,
# causing the test to fail. The 64KiB block size is enough to read
# the system area and header in a single read. the 1MiB block size
# adds very little time to the test so we include it.
if format_name != 'iso':
block_sizes.extend([17, 512])
for block_size in block_sizes:
fmt = self._test_format_at_block_size(format_name, img, block_size)
self.assertTrue(fmt.format_match,
'Failed to match %s at size %i block %i' % (
format_name, image_size, block_size))
self.assertEqual(virtual_size, fmt.virtual_size,
('Failed to calculate size for %s at size %i '
'block %i') % (format_name, image_size,
block_size))
memory = sum(fmt.context_info.values())
self.assertLess(memory, 512 * units.Ki,
'Format used more than 512KiB of memory: %s' % (
fmt.context_info))
def _test_format(self, format_name, subformat=None):
# Try a few different image sizes, including some odd and very small
# sizes
for image_size in (512, 513, 2057, 7):
self._test_format_at_image_size(format_name, image_size * units.Mi,
subformat=subformat)
def test_qcow2(self):
self._test_format('qcow2')
def test_iso_9660(self):
self._test_format('iso', subformat='9660')
def test_iso_udf(self):
self._test_format('iso', subformat='udf')
def _generate_bad_iso(self):
# we want to emulate a malicious user who uploads a an
# ISO file has a qcow2 header in the system area
# of the ISO file
# we will create a qcow2 image and an ISO file
# and then copy the qcow2 header to the ISO file
# e.g.
# mkisofs -o orig.iso /etc/resolv.conf
# qemu-img create orig.qcow2 -f qcow2 64M
# dd if=orig.qcow2 of=outcome bs=32K count=1
# dd if=orig.iso of=outcome bs=32K skip=1 seek=1
qcow = self._create_img('qcow2', 10 * units.Mi)
iso = self._create_iso(64 * units.Mi, subformat='9660')
# first ensure the files are valid
iso_fmt = self._test_format_at_block_size('iso', iso, 4 * units.Ki)
self.assertTrue(iso_fmt.format_match)
qcow_fmt = self._test_format_at_block_size('qcow2', qcow, 4 * units.Ki)
self.assertTrue(qcow_fmt.format_match)
# now copy the qcow2 header to an ISO file
prefix = TEST_IMAGE_PREFIX
prefix += '-bad-'
fn = tempfile.mktemp(prefix=prefix, suffix='.iso')
self._created_files.append(fn)
subprocess.check_output(
'dd if=%s of=%s bs=32K count=1' % (qcow, fn),
shell=True)
subprocess.check_output(
'dd if=%s of=%s bs=32K skip=1 seek=1' % (iso, fn),
shell=True)
return qcow, iso, fn
def test_bad_iso_qcow2(self):
_, _, fn = self._generate_bad_iso()
iso_check = self._test_format_at_block_size('iso', fn, 4 * units.Ki)
qcow_check = self._test_format_at_block_size('qcow2', fn, 4 * units.Ki)
# this system area of the ISO file is not considered part of the format
# the qcow2 header is in the system area of the ISO file
# so the ISO file is still valid
self.assertTrue(iso_check.format_match)
# the qcow2 header is in the system area of the ISO file
# but that will be parsed by the qcow2 format inspector
# and it will match
self.assertTrue(qcow_check.format_match)
# if we call format_inspector.detect_file_format it should detect
# and raise an exception because both match internally.
e = self.assertRaises(
format_inspector.ImageFormatError,
format_inspector.detect_file_format, fn)
self.assertIn('Multiple formats detected', str(e))
def test_vhd(self):
self._test_format('vhd')
def test_vhdx(self):
self._test_format('vhdx')
def test_vmdk(self):
self._test_format('vmdk')
def test_vmdk_stream_optimized(self):
self._test_format('vmdk', 'streamOptimized')
def test_from_file_reads_minimum(self):
img = self._create_img('qcow2', 10 * units.Mi)
file_size = os.stat(img).st_size
fmt = format_inspector.QcowInspector.from_file(img)
# We know everything we need from the first 512 bytes of a QCOW image,
# so make sure that we did not read the whole thing when we inspect
# a local file.
self.assertLess(fmt.actual_size, file_size)
def test_qed_always_unsafe(self):
img = self._create_img('qed', 10 * units.Mi)
fmt = format_inspector.get_inspector('qed').from_file(img)
self.assertTrue(fmt.format_match)
self.assertFalse(fmt.safety_check())
def _test_vmdk_bad_descriptor_offset(self, subformat=None):
format_name = 'vmdk'
image_size = 10 * units.Mi
descriptorOffsetAddr = 0x1c
BAD_ADDRESS = 0x400
img = self._create_img(format_name, image_size, subformat=subformat)
# Corrupt the header
fd = open(img, 'r+b')
fd.seek(descriptorOffsetAddr)
fd.write(struct.pack('<Q', BAD_ADDRESS // 512))
fd.close()
# Read the format in various sizes, some of which will read whole
# sections in a single read, others will be completely unaligned, etc.
for block_size in (64 * units.Ki, 512, 17, 1 * units.Mi):
fmt = self._test_format_at_block_size(format_name, img, block_size)
self.assertTrue(fmt.format_match,
'Failed to match %s at size %i block %i' % (
format_name, image_size, block_size))
self.assertEqual(0, fmt.virtual_size,
('Calculated a virtual size for a corrupt %s at '
'size %i block %i') % (format_name, image_size,
block_size))
def test_vmdk_bad_descriptor_offset(self):
self._test_vmdk_bad_descriptor_offset()
def test_vmdk_bad_descriptor_offset_stream_optimized(self):
self._test_vmdk_bad_descriptor_offset(subformat='streamOptimized')
def _test_vmdk_bad_descriptor_mem_limit(self, subformat=None):
format_name = 'vmdk'
image_size = 5 * units.Mi
virtual_size = 5 * units.Mi
descriptorOffsetAddr = 0x1c
descriptorSizeAddr = descriptorOffsetAddr + 8
twoMBInSectors = (2 << 20) // 512
# We need a big VMDK because otherwise we will not have enough data to
# fill-up the CaptureRegion.
img = self._create_allocated_vmdk(image_size // units.Mi,
subformat=subformat)
# Corrupt the end of descriptor address so it "ends" at 2MB
fd = open(img, 'r+b')
fd.seek(descriptorSizeAddr)
fd.write(struct.pack('<Q', twoMBInSectors))
fd.close()
# Read the format in various sizes, some of which will read whole
# sections in a single read, others will be completely unaligned, etc.
for block_size in (64 * units.Ki, 512, 17, 1 * units.Mi):
fmt = self._test_format_at_block_size(format_name, img, block_size)
self.assertTrue(fmt.format_match,
'Failed to match %s at size %i block %i' % (
format_name, image_size, block_size))
self.assertEqual(virtual_size, fmt.virtual_size,
('Failed to calculate size for %s at size %i '
'block %i') % (format_name, image_size,
block_size))
memory = sum(fmt.context_info.values())
self.assertLess(memory, 1.5 * units.Mi,
'Format used more than 1.5MiB of memory: %s' % (
fmt.context_info))
def test_vmdk_bad_descriptor_mem_limit(self):
self._test_vmdk_bad_descriptor_mem_limit()
def test_vmdk_bad_descriptor_mem_limit_stream_optimized(self):
self._test_vmdk_bad_descriptor_mem_limit(subformat='streamOptimized')
def test_qcow2_safety_checks(self):
# Create backing and data-file names (and initialize the backing file)
backing_fn = tempfile.mktemp(prefix='backing')
self._created_files.append(backing_fn)
with open(backing_fn, 'w') as f:
f.write('foobar')
data_fn = tempfile.mktemp(prefix='data')
self._created_files.append(data_fn)
# A qcow with no backing or data file is safe
fn = self._create_img('qcow2', 5 * units.Mi, None)
inspector = format_inspector.QcowInspector.from_file(fn)
self.assertTrue(inspector.safety_check())
# A backing file makes it unsafe
fn = self._create_img('qcow2', 5 * units.Mi, None,
backing_file=backing_fn)
inspector = format_inspector.QcowInspector.from_file(fn)
self.assertFalse(inspector.safety_check())
# A data-file makes it unsafe
fn = self._create_img('qcow2', 5 * units.Mi,
options={'data_file': data_fn,
'data_file_raw': 'on'})
inspector = format_inspector.QcowInspector.from_file(fn)
self.assertFalse(inspector.safety_check())
# Trying to load a non-QCOW file is an error
self.assertRaises(format_inspector.ImageFormatError,
format_inspector.QcowInspector.from_file,
backing_fn)
def test_qcow2_feature_flag_checks(self):
data = bytearray(512)
data[0:4] = b'QFI\xFB'
inspector = format_inspector.QcowInspector()
inspector.region('header').data = data
# All zeros, no feature flags - all good
self.assertFalse(inspector.has_unknown_features)
# A feature flag set in the first byte (highest-order) is not
# something we know about, so fail.
data[0x48] = 0x01
self.assertTrue(inspector.has_unknown_features)
# The first bit in the last byte (lowest-order) is known (the dirty
# bit) so that should pass
data[0x48] = 0x00
data[0x4F] = 0x01
self.assertFalse(inspector.has_unknown_features)
# Currently (as of 2024), the high-order feature flag bit in the low-
# order byte is not assigned, so make sure we reject it.
data[0x4F] = 0x80
self.assertTrue(inspector.has_unknown_features)
def test_vdi(self):
self._test_format('vdi')
def _test_format_with_invalid_data(self, format_name):
fmt = format_inspector.get_inspector(format_name)()
wrapper = format_inspector.InfoWrapper(open(__file__, 'rb'), fmt)
while True:
chunk = wrapper.read(32)
if not chunk:
break
wrapper.close()
self.assertFalse(fmt.format_match)
self.assertEqual(0, fmt.virtual_size)
memory = sum(fmt.context_info.values())
self.assertLess(memory, 512 * units.Ki,
'Format used more than 512KiB of memory: %s' % (
fmt.context_info))
def test_qcow2_invalid(self):
self._test_format_with_invalid_data('qcow2')
def test_vhd_invalid(self):
self._test_format_with_invalid_data('vhd')
def test_vhdx_invalid(self):
self._test_format_with_invalid_data('vhdx')
def test_vmdk_invalid(self):
self._test_format_with_invalid_data('vmdk')
def test_vdi_invalid(self):
self._test_format_with_invalid_data('vdi')
def test_vmdk_invalid_type(self):
fmt = format_inspector.get_inspector('vmdk')()
wrapper = format_inspector.InfoWrapper(open(__file__, 'rb'), fmt)
while True:
chunk = wrapper.read(32)
if not chunk:
break
wrapper.close()
fake_rgn = mock.MagicMock()
fake_rgn.complete = True
fake_rgn.data = b'foocreateType="someunknownformat"bar'
with mock.patch.object(fmt, 'has_region', return_value=True,
autospec=True):
with mock.patch.object(fmt, 'region', return_value=fake_rgn,
autospec=True):
self.assertEqual(0, fmt.virtual_size)
class TestFormatInspectorInfra(IronicAgentTest):
def _test_capture_region_bs(self, bs):
data = b''.join(chr(x).encode() for x in range(ord('A'), ord('z')))
regions = [
format_inspector.CaptureRegion(3, 9),
format_inspector.CaptureRegion(0, 256),
format_inspector.CaptureRegion(32, 8),
]
for region in regions:
# None of them should be complete yet
self.assertFalse(region.complete)
pos = 0
for i in range(0, len(data), bs):
chunk = data[i:i + bs]
pos += len(chunk)
for region in regions:
region.capture(chunk, pos)
self.assertEqual(data[3:12], regions[0].data)
self.assertEqual(data[0:256], regions[1].data)
self.assertEqual(data[32:40], regions[2].data)
# The small regions should be complete
self.assertTrue(regions[0].complete)
self.assertTrue(regions[2].complete)
# This region extended past the available data, so not complete
self.assertFalse(regions[1].complete)
def test_capture_region(self):
for block_size in (1, 3, 7, 13, 32, 64):
self._test_capture_region_bs(block_size)
def _get_wrapper(self, data):
source = io.BytesIO(data)
fake_fmt = mock.create_autospec(format_inspector.get_inspector('raw'))
return format_inspector.InfoWrapper(source, fake_fmt)
def test_info_wrapper_file_like(self):
data = b''.join(chr(x).encode() for x in range(ord('A'), ord('z')))
wrapper = self._get_wrapper(data)
read_data = b''
while True:
chunk = wrapper.read(8)
if not chunk:
break
read_data += chunk
self.assertEqual(data, read_data)
def test_info_wrapper_iter_like(self):
data = b''.join(chr(x).encode() for x in range(ord('A'), ord('z')))
wrapper = self._get_wrapper(data)
read_data = b''
for chunk in wrapper:
read_data += chunk
self.assertEqual(data, read_data)
def test_info_wrapper_file_like_eats_error(self):
wrapper = self._get_wrapper(b'123456')
wrapper._format.eat_chunk.side_effect = Exception('fail')
data = b''
while True:
chunk = wrapper.read(3)
if not chunk:
break
data += chunk
# Make sure we got all the data despite the error
self.assertEqual(b'123456', data)
# Make sure we only called this once and never again after
# the error was raised
wrapper._format.eat_chunk.assert_called_once_with(b'123')
def test_info_wrapper_iter_like_eats_error(self):
fake_fmt = mock.create_autospec(format_inspector.get_inspector('raw'))
wrapper = format_inspector.InfoWrapper(iter([b'123', b'456']),
fake_fmt)
fake_fmt.eat_chunk.side_effect = Exception('fail')
data = b''
for chunk in wrapper:
data += chunk
# Make sure we got all the data despite the error
self.assertEqual(b'123456', data)
# Make sure we only called this once and never again after
# the error was raised
fake_fmt.eat_chunk.assert_called_once_with(b'123')
def test_get_inspector(self):
self.assertEqual(format_inspector.QcowInspector,
format_inspector.get_inspector('qcow2'))
self.assertIsNone(format_inspector.get_inspector('foo'))
class TestFormatInspectorsTargeted(IronicAgentTest):
def _make_vhd_meta(self, guid_raw, item_length):
# Meta region header, padded to 32 bytes
data = struct.pack('<8sHH', b'metadata', 0, 1)
data += b'0' * 20
# Metadata table entry, 16-byte GUID, 12-byte information,
# padded to 32-bytes
data += guid_raw
data += struct.pack('<III', 256, item_length, 0)
data += b'0' * 6
return data
def test_vhd_table_over_limit(self):
ins = format_inspector.VHDXInspector()
meta = format_inspector.CaptureRegion(0, 0)
desired = b'012345678ABCDEF0'
# This is a poorly-crafted image that specifies a larger table size
# than is allowed
meta.data = self._make_vhd_meta(desired, 33 * 2048)
ins.new_region('metadata', meta)
new_region = ins._find_meta_entry(ins._guid(desired))
# Make sure we clamp to our limit of 32 * 2048
self.assertEqual(
format_inspector.VHDXInspector.VHDX_METADATA_TABLE_MAX_SIZE,
new_region.length)
def test_vhd_table_under_limit(self):
ins = format_inspector.VHDXInspector()
meta = format_inspector.CaptureRegion(0, 0)
desired = b'012345678ABCDEF0'
meta.data = self._make_vhd_meta(desired, 16 * 2048)
ins.new_region('metadata', meta)
new_region = ins._find_meta_entry(ins._guid(desired))
# Table size was under the limit, make sure we get it back
self.assertEqual(16 * 2048, new_region.length)

28
ironic_python_agent/tests/unit/test_partition_utils.py
View File

@ -16,7 +16,6 @@ import tempfile
from unittest import mock
from ironic_lib import exception
from ironic_lib import qemu_img
from ironic_lib import utils
from oslo_concurrency import processutils
from oslo_config import cfg
@ -27,6 +26,7 @@ from ironic_python_agent import disk_utils
from ironic_python_agent import errors
from ironic_python_agent import hardware
from ironic_python_agent import partition_utils
from ironic_python_agent import qemu_img
from ironic_python_agent.tests.unit import base
@ -453,13 +453,15 @@ class WorkOnDiskTestCase(base.IronicAgentTest):
@mock.patch.object(utils, 'mkfs', lambda fs, path, label=None: None)
@mock.patch.object(disk_utils, 'block_uuid', lambda p: 'uuid')
@mock.patch.object(disk_utils, 'populate_image', lambda image_path,
root_path, conv_flags=None: None)
root_path, conv_flags=None, source_format=None,
is_raw=False: None)
def test_gpt_disk_label(self):
ephemeral_part = '/dev/fake-part1'
swap_part = '/dev/fake-part2'
root_part = '/dev/fake-part3'
ephemeral_mb = 256
ephemeral_format = 'exttest'
source_format = 'raw'
self.mock_mp.return_value = {'ephemeral': ephemeral_part,
'swap': swap_part,
@ -472,7 +474,8 @@ class WorkOnDiskTestCase(base.IronicAgentTest):
self.swap_mb, ephemeral_mb,
ephemeral_format,
self.image_path, self.node_uuid,
disk_label='gpt', conv_flags=None)
disk_label='gpt', conv_flags=None,
source_format=source_format, is_raw=True)
self.assertEqual(self.mock_ibd.call_args_list, calls)
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
self.swap_mb, ephemeral_mb,
@ -492,6 +495,8 @@ class WorkOnDiskTestCase(base.IronicAgentTest):
"""Test that we create a fat filesystem with UEFI localboot."""
root_part = '/dev/fake-part1'
efi_part = '/dev/fake-part2'
source_format = 'format'
self.mock_mp.return_value = {'root': root_part,
'efi system partition': efi_part}
self.mock_ibd.return_value = True
@ -502,7 +507,8 @@ class WorkOnDiskTestCase(base.IronicAgentTest):
self.swap_mb, self.ephemeral_mb,
self.ephemeral_format,
self.image_path, self.node_uuid,
boot_mode="uefi")
boot_mode="uefi",
source_format=source_format, is_raw=False)
self.mock_mp.assert_called_once_with(self.dev, self.root_mb,
self.swap_mb, self.ephemeral_mb,
@ -515,8 +521,9 @@ class WorkOnDiskTestCase(base.IronicAgentTest):
self.assertEqual(self.mock_ibd.call_args_list, mock_ibd_calls)
mock_mkfs.assert_called_once_with(fs='vfat', path=efi_part,
label='efi-part')
mock_populate_image.assert_called_once_with(self.image_path,
root_part, conv_flags=None)
mock_populate_image.assert_called_once_with(
self.image_path, root_part, conv_flags=None,
source_format=source_format, is_raw=False)
mock_block_uuid.assert_any_call(root_part)
mock_block_uuid.assert_any_call(efi_part)
mock_trigger_device_rescan.assert_called_once_with(self.dev)
@ -595,6 +602,7 @@ class WorkOnDiskTestCase(base.IronicAgentTest):
root_part = '/dev/fake-part3'
ephemeral_mb = 256
ephemeral_format = 'exttest'
fmt = 'format'
self.mock_mp.return_value = {'ephemeral': ephemeral_part,
'swap': swap_part,
@ -604,11 +612,15 @@ class WorkOnDiskTestCase(base.IronicAgentTest):
self.swap_mb, ephemeral_mb,
ephemeral_format,
self.image_path, self.node_uuid,
disk_label='gpt', conv_flags='sparse')
disk_label='gpt', conv_flags='sparse',
source_format=fmt,
is_raw=False)
mock_populate_image.assert_called_once_with(self.image_path,
root_part,
conv_flags='sparse')
conv_flags='sparse',
source_format=fmt,
is_raw=False)
class CreateConfigDriveTestCases(base.IronicAgentTest):

332
ironic_python_agent/tests/unit/test_qemu_img.py Normal file
View File

@ -0,0 +1,332 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
from unittest import mock
from ironic_lib.tests import base
from ironic_lib import utils
from oslo_concurrency import processutils
from oslo_config import cfg
from oslo_utils import imageutils
from ironic_python_agent import errors
from ironic_python_agent import qemu_img
CONF = cfg.CONF
class ImageInfoTestCase(base.IronicLibTestCase):
@mock.patch.object(os.path, 'exists', return_value=False, autospec=True)
def test_image_info_path_doesnt_exist_disabled(self, path_exists_mock):
CONF.set_override('disable_deep_image_inspection', True)
self.assertRaises(FileNotFoundError, qemu_img.image_info, 'noimg')
path_exists_mock.assert_called_once_with('noimg')
@mock.patch.object(utils, 'execute', return_value=('out', 'err'),
autospec=True)
@mock.patch.object(imageutils, 'QemuImgInfo', autospec=True)
@mock.patch.object(os.path, 'exists', return_value=True, autospec=True)
def test_image_info_path_exists_disabled(self, path_exists_mock,
image_info_mock, execute_mock):
CONF.set_override('disable_deep_image_inspection', True)
qemu_img.image_info('img')
path_exists_mock.assert_called_once_with('img')
execute_mock.assert_called_once_with(
['env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info', 'img',
'--output=json'], prlimit=mock.ANY)
image_info_mock.assert_called_once_with('out', format='json')
@mock.patch.object(utils, 'execute', return_value=('out', 'err'),
autospec=True)
@mock.patch.object(imageutils, 'QemuImgInfo', autospec=True)
@mock.patch.object(os.path, 'exists', return_value=True, autospec=True)
def test_image_info_path_exists_safe(
self, path_exists_mock, image_info_mock, execute_mock):
qemu_img.image_info('img', source_format='qcow2')
path_exists_mock.assert_called_once_with('img')
execute_mock.assert_called_once_with(
['env', 'LC_ALL=C', 'LANG=C', 'qemu-img', 'info', 'img',
'--output=json', '-f', 'qcow2'],
prlimit=mock.ANY
)
image_info_mock.assert_called_once_with('out', format='json')
@mock.patch.object(utils, 'execute', return_value=('out', 'err'),
autospec=True)
@mock.patch.object(imageutils, 'QemuImgInfo', autospec=True)
@mock.patch.object(os.path, 'exists', return_value=True, autospec=True)
def test_image_info_path_exists_unsafe(
self, path_exists_mock, image_info_mock, execute_mock):
# Call without source_format raises
self.assertRaises(errors.InvalidImage,
qemu_img.image_info, 'img')
# safety valve! Don't run **anything** against the image without
# source_format unless specifically permitted
path_exists_mock.assert_not_called()
execute_mock.assert_not_called()
image_info_mock.assert_not_called()
class ConvertImageTestCase(base.IronicLibTestCase):
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_disabled(self, execute_mock):
CONF.set_override('disable_deep_image_inspection', True)
qemu_img.convert_image('source', 'dest', 'out_format')
execute_mock.assert_called_once_with(
'qemu-img', 'convert', '-O',
'out_format', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_flags_disabled(self, execute_mock):
CONF.set_override('disable_deep_image_inspection', True)
qemu_img.convert_image('source', 'dest', 'out_format',
cache='directsync', out_of_order=True,
sparse_size='0')
execute_mock.assert_called_once_with(
'qemu-img', 'convert', '-O',
'out_format', '-t', 'directsync',
'-S', '0', '-W', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_retries_disabled(self, execute_mock):
CONF.set_override('disable_deep_image_inspection', True)
ret_err = 'qemu: qemu_thread_create: Resource temporarily unavailable'
execute_mock.side_effect = [
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
('', ''),
]
qemu_img.convert_image('source', 'dest', 'out_format')
convert_call = mock.call('qemu-img', 'convert', '-O',
'out_format', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
execute_mock.assert_has_calls([
convert_call,
mock.call('sync'),
convert_call,
mock.call('sync'),
convert_call,
])
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_retries_alternate_error_disabled(self, exe_mock):
CONF.set_override('disable_deep_image_inspection', True)
ret_err = 'Failed to allocate memory: Cannot allocate memory\n'
exe_mock.side_effect = [
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
('', ''),
]
qemu_img.convert_image('source', 'dest', 'out_format')
convert_call = mock.call('qemu-img', 'convert', '-O',
'out_format', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
exe_mock.assert_has_calls([
convert_call,
mock.call('sync'),
convert_call,
mock.call('sync'),
convert_call,
])
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_retries_and_fails_disabled(self, execute_mock):
CONF.set_override('disable_deep_image_inspection', True)
ret_err = 'qemu: qemu_thread_create: Resource temporarily unavailable'
execute_mock.side_effect = [
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err),
]
self.assertRaises(processutils.ProcessExecutionError,
qemu_img.convert_image,
'source', 'dest', 'out_format')
convert_call = mock.call('qemu-img', 'convert', '-O',
'out_format', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
execute_mock.assert_has_calls([
convert_call,
mock.call('sync'),
convert_call,
mock.call('sync'),
convert_call,
])
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_just_fails_disabled(self, execute_mock):
CONF.set_override('disable_deep_image_inspection', True)
ret_err = 'Aliens'
execute_mock.side_effect = [
processutils.ProcessExecutionError(stderr=ret_err),
]
self.assertRaises(processutils.ProcessExecutionError,
qemu_img.convert_image,
'source', 'dest', 'out_format')
convert_call = mock.call('qemu-img', 'convert', '-O',
'out_format', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
execute_mock.assert_has_calls([
convert_call,
])
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image(self, execute_mock):
qemu_img.convert_image('source', 'dest', 'out_format',
source_format='fmt')
execute_mock.assert_called_once_with(
'qemu-img', 'convert', '-O',
'out_format', '-f', 'fmt',
'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_flags(self, execute_mock):
qemu_img.convert_image('source', 'dest', 'out_format',
cache='directsync', out_of_order=True,
sparse_size='0', source_format='fmt')
execute_mock.assert_called_once_with(
'qemu-img', 'convert', '-O',
'out_format', '-t', 'directsync',
'-S', '0', '-f', 'fmt', '-W', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_retries(self, execute_mock):
ret_err = 'qemu: qemu_thread_create: Resource temporarily unavailable'
execute_mock.side_effect = [
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
('', ''),
]
qemu_img.convert_image('source', 'dest', 'out_format',
source_format='fmt')
convert_call = mock.call('qemu-img', 'convert', '-O',
'out_format', '-f', 'fmt', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
execute_mock.assert_has_calls([
convert_call,
mock.call('sync'),
convert_call,
mock.call('sync'),
convert_call,
])
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_retries_alternate_error(self, execute_mock):
ret_err = 'Failed to allocate memory: Cannot allocate memory\n'
execute_mock.side_effect = [
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
('', ''),
]
qemu_img.convert_image('source', 'dest', 'out_format',
source_format='fmt')
convert_call = mock.call('qemu-img', 'convert', '-O',
'out_format', '-f', 'fmt', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
execute_mock.assert_has_calls([
convert_call,
mock.call('sync'),
convert_call,
mock.call('sync'),
convert_call,
])
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_retries_and_fails(self, execute_mock):
ret_err = 'qemu: qemu_thread_create: Resource temporarily unavailable'
execute_mock.side_effect = [
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err), ('', ''),
processutils.ProcessExecutionError(stderr=ret_err),
]
self.assertRaises(processutils.ProcessExecutionError,
qemu_img.convert_image,
'source', 'dest', 'out_format', source_format='fmt')
convert_call = mock.call('qemu-img', 'convert', '-O',
'out_format', '-f', 'fmt', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
execute_mock.assert_has_calls([
convert_call,
mock.call('sync'),
convert_call,
mock.call('sync'),
convert_call,
])
@mock.patch.object(utils, 'execute', autospec=True)
def test_convert_image_just_fails(self, execute_mock):
ret_err = 'Aliens'
execute_mock.side_effect = [
processutils.ProcessExecutionError(stderr=ret_err),
]
self.assertRaises(processutils.ProcessExecutionError,
qemu_img.convert_image,
'source', 'dest', 'out_format', source_format='fmt')
convert_call = mock.call('qemu-img', 'convert', '-O',
'out_format', '-f', 'fmt', 'source', 'dest',
run_as_root=False,
prlimit=mock.ANY,
use_standard_locale=True,
env_variables={'MALLOC_ARENA_MAX': '3'})
execute_mock.assert_has_calls([
convert_call,
])

48
releasenotes/notes/image-security-5c23b890409101c9.yaml Normal file
View File

@ -0,0 +1,48 @@
---
security:
- |
Ironic-Python-Agent now checks any supplied image format value against
the detected format of the image file and will prevent deployments should
the values mismatch.
- |
Images previously misconfigured as raw despite being in another format,
in some non-default configurations, may have been mistakenly converted if
needed. Ironic-Python-Agent will no longer perform conversion in any case
for images with metadata indicating in raw format.
- |
Ironic-Python-Agent *always* inspects any non-raw user image content for
safety before running any qemu-based utilities on the image. This is
utilized to identify the format of the image and to verify the overall
safety of the image. Any images with unknown or unsafe feature uses are
explicitly rejected. This can be disabled in both IPA and Ironic by setting
``[conductor]disable_deep_image_inspection`` to ``True`` for the Ironic
deployment. Image inspection is the primary mitigation for CVE-2024-44082
being tracked in
`bug 2071740 <https://bugs.launchpad.net/ironic-python-agent/+bug/2071740>`_.
Operators may desire to set
``[conductor]conductor_always_validates_images`` on Ironic conductors to
mitigate the issue before they have upgraded their Ironic-Python-Agent.
- |
Ironic-Python-Agent now explicitly enforces a list of permitted image
types for deployment, defaulting to "raw" and "qcow2". Other image types
may work, but are not explicitly supported and must be enabled. This can
be modified by setting ``[conductor]permitted_image_formats`` for all
Ironic services.
fixes:
- |
Fixes multiple issues in the handling of images as it related to
execution of the ``qemu-img`` utility. When using this utility to convert
an unsafe image, a malicious user can extract information from a node
while Ironic-Python-Agent is deploying or converting an image.
Ironic-Python-Agent now inspects all non-raw images for safety, and never
runs qemu-based utilities on raw images. This fix is tracked as
CVE-2024-44082 and `bug 2071740 <https://bugs.launchpad
.net/ironic-python-agent/+bug/2071740>`_.
- |
Images with metadata indicating a "raw" disk format may have been
transparently converted from another format. Now, these images will have
their exact contents imaged to disk without modification.
upgrade:
- |
Deployers implementing their own ``HardwareManagers`` must to audit
their code for unsafe uses of `qemu-img` and related methods.