diff --git a/ironic_python_agent/agent.py b/ironic_python_agent/agent.py
index a9ed441e5..e5244a912 100644
--- a/ironic_python_agent/agent.py
+++ b/ironic_python_agent/agent.py
@@ -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)
diff --git a/ironic_python_agent/config.py b/ironic_python_agent/config.py
index f311bf187..75527b39c 100644
--- a/ironic_python_agent/config.py
+++ b/ironic_python_agent/config.py
@@ -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()
diff --git a/ironic_python_agent/disk_utils.py b/ironic_python_agent/disk_utils.py
index 4195b3ca8..dfa7b27c6 100644
--- a/ironic_python_agent/disk_utils.py
+++ b/ironic_python_agent/disk_utils.py
@@ -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)
diff --git a/ironic_python_agent/errors.py b/ironic_python_agent/errors.py
index d004e90b3..0a678c29f 100644
--- a/ironic_python_agent/errors.py
+++ b/ironic_python_agent/errors.py
@@ -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)
diff --git a/ironic_python_agent/extensions/standby.py b/ironic_python_agent/extensions/standby.py
index 7a365aa0f..d3bf04af2 100644
--- a/ironic_python_agent/extensions/standby.py
+++ b/ironic_python_agent/extensions/standby.py
@@ -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:
-        qemu_img.convert_image(image, device, out_format='host_device',
-                               cache='directsync', out_of_order=True,
-                               sparse_size='0')
+        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',
+                                   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)
diff --git a/ironic_python_agent/format_inspector.py b/ironic_python_agent/format_inspector.py
new file mode 100644
index 000000000..44cdd4ae7
--- /dev/null
+++ b/ironic_python_agent/format_inspector.py
@@ -0,0 +1,1044 @@
+# 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.
+
+"""
+This is a python implementation of virtual disk format inspection routines
+gathered from various public specification documents, as well as qemu disk
+driver code. It attempts to store and parse the minimum amount of data
+required, and in a streaming-friendly manner to collect metadata about
+complex-format images.
+
+This was imported from the Ironic fix. A copy of this inspector
+exists in multiple projects, including Ironic, Nova, and Cinder. Do not
+modify this version without modifying all versions.
+
+TODO(JayF): Remove this module, replace with oslo_utils version once released
+"""
+
+import struct
+
+from oslo_log import log as logging
+from oslo_utils import units
+
+LOG = logging.getLogger(__name__)
+
+
+def chunked_reader(fileobj, chunk_size=512):
+    while True:
+        chunk = fileobj.read(chunk_size)
+        if not chunk:
+            break
+        yield chunk
+
+
+class CaptureRegion(object):
+    """Represents a region of a file we want to capture.
+
+    A region of a file we want to capture requires a byte offset into
+    the file and a length. This is expected to be used by a data
+    processing loop, calling capture() with the most recently-read
+    chunk. This class handles the task of grabbing the desired region
+    of data across potentially multiple fractional and unaligned reads.
+
+    :param offset: Byte offset into the file starting the region
+    :param length: The length of the region
+    """
+
+    def __init__(self, offset, length):
+        self.offset = offset
+        self.length = length
+        self.data = b''
+
+    @property
+    def complete(self):
+        """Returns True when we have captured the desired data."""
+        return self.length == len(self.data)
+
+    def capture(self, chunk, current_position):
+        """Process a chunk of data.
+
+        This should be called for each chunk in the read loop, at least
+        until complete returns True.
+
+        :param chunk: A chunk of bytes in the file
+        :param current_position: The position of the file processed by the
+                                 read loop so far. Note that this will be
+                                 the position in the file *after* the chunk
+                                 being presented.
+        """
+        read_start = current_position - len(chunk)
+        if (read_start <= self.offset <= current_position
+                or self.offset <= read_start <= (self.offset + self.length)):
+            if read_start < self.offset:
+                lead_gap = self.offset - read_start
+            else:
+                lead_gap = 0
+            self.data += chunk[lead_gap:]
+            self.data = self.data[:self.length]
+
+
+class ImageFormatError(Exception):
+    """An unrecoverable image format error that aborts the process."""
+    pass
+
+
+class TraceDisabled(object):
+    """A logger-like thing that swallows tracing when we do not want it."""
+
+    def debug(self, *a, **k):
+        pass
+
+    info = debug
+    warning = debug
+    error = debug
+
+
+class FileInspector(object):
+    """A stream-based disk image inspector.
+
+    This base class works on raw images and is subclassed for more
+    complex types. It is to be presented with the file to be examined
+    one chunk at a time, during read processing and will only store
+    as much data as necessary to determine required attributes of
+    the file.
+    """
+
+    def __init__(self, tracing=False):
+        self._total_count = 0
+
+        # NOTE(danms): The logging in here is extremely verbose for a reason,
+        # but should never really be enabled at that level at runtime. To
+        # retain all that work and assist in future debug, we have a separate
+        # debug flag that can be passed from a manual tool to turn it on.
+        if tracing:
+            self._log = logging.getLogger(str(self))
+        else:
+            self._log = TraceDisabled()
+        self._capture_regions = {}
+
+    def _capture(self, chunk, only=None):
+        for name, region in self._capture_regions.items():
+            if only and name not in only:
+                continue
+            if not region.complete:
+                region.capture(chunk, self._total_count)
+
+    def eat_chunk(self, chunk):
+        """Call this to present chunks of the file to the inspector."""
+        pre_regions = set(self._capture_regions.keys())
+
+        # Increment our position-in-file counter
+        self._total_count += len(chunk)
+
+        # Run through the regions we know of to see if they want this
+        # data
+        self._capture(chunk)
+
+        # Let the format do some post-read processing of the stream
+        self.post_process()
+
+        # Check to see if the post-read processing added new regions
+        # which may require the current chunk.
+        new_regions = set(self._capture_regions.keys()) - pre_regions
+        if new_regions:
+            self._capture(chunk, only=new_regions)
+
+    def post_process(self):
+        """Post-read hook to process what has been read so far.
+
+        This will be called after each chunk is read and potentially captured
+        by the defined regions. If any regions are defined by this call,
+        those regions will be presented with the current chunk in case it
+        is within one of the new regions.
+        """
+        pass
+
+    def region(self, name):
+        """Get a CaptureRegion by name."""
+        return self._capture_regions[name]
+
+    def new_region(self, name, region):
+        """Add a new CaptureRegion by name."""
+        if self.has_region(name):
+            # This is a bug, we tried to add the same region twice
+            raise ImageFormatError('Inspector re-added region %s' % name)
+        self._capture_regions[name] = region
+
+    def has_region(self, name):
+        """Returns True if named region has been defined."""
+        return name in self._capture_regions
+
+    @property
+    def format_match(self):
+        """Returns True if the file appears to be the expected format."""
+        return True
+
+    @property
+    def virtual_size(self):
+        """Returns the virtual size of the disk image, or zero if unknown."""
+        return self._total_count
+
+    @property
+    def actual_size(self):
+        """Returns the total size of the file, usually smaller than virtual_size.
+
+        NOTE: this will only be accurate if the entire file is read and processed.
+        """  # noqa
+        return self._total_count
+
+    @property
+    def complete(self):
+        """Returns True if we have all the information needed."""
+        return all(r.complete for r in self._capture_regions.values())
+
+    def __str__(self):
+        """The string name of this file format."""
+        return 'raw'
+
+    @property
+    def context_info(self):
+        """Return info on amount of data held in memory for auditing.
+
+        This is a dict of region:sizeinbytes items that the inspector
+        uses to examine the file.
+        """
+        return {name: len(region.data) for name, region in
+                self._capture_regions.items()}
+
+    @classmethod
+    def from_file(cls, filename):
+        """Read as much of a file as necessary to complete inspection.
+
+        NOTE: Because we only read as much of the file as necessary, the
+        actual_size property will not reflect the size of the file, but the
+        amount of data we read before we satisfied the inspector.
+
+        Raises ImageFormatError if we cannot parse the file.
+        """
+        inspector = cls()
+        with open(filename, 'rb') as f:
+            for chunk in chunked_reader(f):
+                inspector.eat_chunk(chunk)
+                if inspector.complete:
+                    # No need to eat any more data
+                    break
+        if not inspector.complete or not inspector.format_match:
+            raise ImageFormatError('File is not in requested format')
+        return inspector
+
+    def safety_check(self):
+        """Perform some checks to determine if this file is safe.
+
+        Returns True if safe, False otherwise. It may raise ImageFormatError
+        if safety cannot be guaranteed because of parsing or other errors.
+        """
+        return True
+
+
+# The qcow2 format consists of a big-endian 72-byte header, of which
+# only a small portion has information we care about:
+#
+# Dec   Hex   Name
+#   0  0x00   Magic 4-bytes 'QFI\xfb'
+#   4  0x04   Version (uint32_t, should always be 2 for modern files)
+#  . . .
+#   8  0x08   Backing file offset (uint64_t)
+#  24  0x18   Size in bytes (unint64_t)
+#  . . .
+#  72  0x48   Incompatible features bitfield (6 bytes)
+#
+# https://gitlab.com/qemu-project/qemu/-/blob/master/docs/interop/qcow2.txt
+class QcowInspector(FileInspector):
+    """QEMU QCOW2 Format
+
+    This should only require about 32 bytes of the beginning of the file
+    to determine the virtual size, and 104 bytes to perform the safety check.
+    """
+
+    BF_OFFSET = 0x08
+    BF_OFFSET_LEN = 8
+    I_FEATURES = 0x48
+    I_FEATURES_LEN = 8
+    I_FEATURES_DATAFILE_BIT = 3
+    I_FEATURES_MAX_BIT = 4
+
+    def __init__(self, *a, **k):
+        super(QcowInspector, self).__init__(*a, **k)
+        self.new_region('header', CaptureRegion(0, 512))
+
+    def _qcow_header_data(self):
+        magic, version, bf_offset, bf_sz, cluster_bits, size = (
+            struct.unpack('>4sIQIIQ', self.region('header').data[:32]))
+        return magic, size
+
+    @property
+    def has_header(self):
+        return self.region('header').complete
+
+    @property
+    def virtual_size(self):
+        if not self.region('header').complete:
+            return 0
+        if not self.format_match:
+            return 0
+        magic, size = self._qcow_header_data()
+        return size
+
+    @property
+    def format_match(self):
+        if not self.region('header').complete:
+            return False
+        magic, size = self._qcow_header_data()
+        return magic == b'QFI\xFB'
+
+    @property
+    def has_backing_file(self):
+        if not self.region('header').complete:
+            return None
+        if not self.format_match:
+            return False
+        bf_offset_bytes = self.region('header').data[
+            self.BF_OFFSET:self.BF_OFFSET + self.BF_OFFSET_LEN]
+        # nonzero means "has a backing file"
+        bf_offset, = struct.unpack('>Q', bf_offset_bytes)
+        return bf_offset != 0
+
+    @property
+    def has_unknown_features(self):
+        if not self.region('header').complete:
+            return None
+        if not self.format_match:
+            return False
+        i_features = self.region('header').data[
+            self.I_FEATURES:self.I_FEATURES + self.I_FEATURES_LEN]
+
+        # This is the maximum byte number we should expect any bits to be set
+        max_byte = self.I_FEATURES_MAX_BIT // 8
+
+        # The flag bytes are in big-endian ordering, so if we process
+        # them in index-order, they're reversed
+        for i, byte_num in enumerate(reversed(range(self.I_FEATURES_LEN))):
+            if byte_num == max_byte:
+                # If we're in the max-allowed byte, allow any bits less than
+                # the maximum-known feature flag bit to be set
+                allow_mask = ((1 << self.I_FEATURES_MAX_BIT) - 1)
+            elif byte_num > max_byte:
+                # If we're above the byte with the maximum known feature flag
+                # bit, then we expect all zeroes
+                allow_mask = 0x0
+            else:
+                # Any earlier-than-the-maximum byte can have any of the flag
+                # bits set
+                allow_mask = 0xFF
+
+            if i_features[i] & ~allow_mask:
+                LOG.warning('Found unknown feature bit in byte %i: %s/%s',
+                            byte_num, bin(i_features[byte_num] & ~allow_mask),
+                            bin(allow_mask))
+                return True
+
+        return False
+
+    @property
+    def has_data_file(self):
+        if not self.region('header').complete:
+            return None
+        if not self.format_match:
+            return False
+        i_features = self.region('header').data[
+            self.I_FEATURES:self.I_FEATURES + self.I_FEATURES_LEN]
+
+        # First byte of bitfield, which is i_features[7]
+        byte = self.I_FEATURES_LEN - 1 - self.I_FEATURES_DATAFILE_BIT // 8
+        # Third bit of bitfield, which is 0x04
+        bit = 1 << (self.I_FEATURES_DATAFILE_BIT - 1 % 8)
+        return bool(i_features[byte] & bit)
+
+    def __str__(self):
+        return 'qcow2'
+
+    def safety_check(self):
+        return (not self.has_backing_file
+                and not self.has_data_file
+                and not self.has_unknown_features)
+
+
+class QEDInspector(FileInspector):
+    def __init__(self, tracing=False):
+        super().__init__(tracing)
+        self.new_region('header', CaptureRegion(0, 512))
+
+    @property
+    def format_match(self):
+        if not self.region('header').complete:
+            return False
+        return self.region('header').data.startswith(b'QED\x00')
+
+    def safety_check(self):
+        # QED format is not supported by anyone, but we want to detect it
+        # and mark it as just always unsafe.
+        return False
+
+
+# The VHD (or VPC as QEMU calls it) format consists of a big-endian
+# 512-byte "footer" at the beginning of the file with various
+# information, most of which does not matter to us:
+#
+# Dec   Hex   Name
+#   0  0x00   Magic string (8-bytes, always 'conectix')
+#  40  0x28   Disk size (uint64_t)
+#
+# https://github.com/qemu/qemu/blob/master/block/vpc.c
+class VHDInspector(FileInspector):
+    """Connectix/MS VPC VHD Format
+
+    This should only require about 512 bytes of the beginning of the file
+    to determine the virtual size.
+    """
+
+    def __init__(self, *a, **k):
+        super(VHDInspector, self).__init__(*a, **k)
+        self.new_region('header', CaptureRegion(0, 512))
+
+    @property
+    def format_match(self):
+        return self.region('header').data.startswith(b'conectix')
+
+    @property
+    def virtual_size(self):
+        if not self.region('header').complete:
+            return 0
+
+        if not self.format_match:
+            return 0
+
+        return struct.unpack('>Q', self.region('header').data[40:48])[0]
+
+    def __str__(self):
+        return 'vhd'
+
+
+# The VHDX format consists of a complex dynamic little-endian
+# structure with multiple regions of metadata and data, linked by
+# offsets with in the file (and within regions), identified by MSFT
+# GUID strings. The header is a 320KiB structure, only a few pieces of
+# which we actually need to capture and interpret:
+#
+#     Dec    Hex  Name
+#      0 0x00000  Identity (Technically 9-bytes, padded to 64KiB, the first
+#                 8 bytes of which are 'vhdxfile')
+# 196608 0x30000  The Region table (64KiB of a 32-byte header, followed
+#                 by up to 2047 36-byte region table entry structures)
+#
+# The region table header includes two items we need to read and parse,
+# which are:
+#
+# 196608 0x30000  4-byte signature ('regi')
+# 196616 0x30008  Entry count (uint32-t)
+#
+# The region table entries follow the region table header immediately
+# and are identified by a 16-byte GUID, and provide an offset of the
+# start of that region. We care about the "metadata region", identified
+# by the METAREGION class variable. The region table entry is (offsets
+# from the beginning of the entry, since it could be in multiple places):
+#
+#      0 0x00000 16-byte MSFT GUID
+#     16 0x00010 Offset of the actual metadata region (uint64_t)
+#
+# When we find the METAREGION table entry, we need to grab that offset
+# and start examining the region structure at that point. That
+# consists of a metadata table of structures, which point to places in
+# the data in an unstructured space that follows. The header is
+# (offsets relative to the region start):
+#
+#      0 0x00000 8-byte signature ('metadata')
+#      . . .
+#     16 0x00010 2-byte entry count (up to 2047 entries max)
+#
+# This header is followed by the specified number of metadata entry
+# structures, identified by GUID:
+#
+#      0 0x00000 16-byte MSFT GUID
+#     16 0x00010 4-byte offset (uint32_t, relative to the beginning of
+#                the metadata region)
+#
+# We need to find the "Virtual Disk Size" metadata item, identified by
+# the GUID in the VIRTUAL_DISK_SIZE class variable, grab the offset,
+# add it to the offset of the metadata region, and examine that 8-byte
+# chunk of data that follows.
+#
+# The "Virtual Disk Size" is a naked uint64_t which contains the size
+# of the virtual disk, and is our ultimate target here.
+#
+# https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-vhdx/83e061f8-f6e2-4de1-91bd-5d518a43d477
+class VHDXInspector(FileInspector):
+    """MS VHDX Format
+
+    This requires some complex parsing of the stream. The first 256KiB
+    of the image is stored to get the header and region information,
+    and then we capture the first metadata region to read those
+    records, find the location of the virtual size data and parse
+    it. This needs to store the metadata table entries up until the
+    VDS record, which may consist of up to 2047 32-byte entries at
+    max.  Finally, it must store a chunk of data at the offset of the
+    actual VDS uint64.
+
+    """
+    METAREGION = '8B7CA206-4790-4B9A-B8FE-575F050F886E'
+    VIRTUAL_DISK_SIZE = '2FA54224-CD1B-4876-B211-5DBED83BF4B8'
+    VHDX_METADATA_TABLE_MAX_SIZE = 32 * 2048  # From qemu
+
+    def __init__(self, *a, **k):
+        super(VHDXInspector, self).__init__(*a, **k)
+        self.new_region('ident', CaptureRegion(0, 32))
+        self.new_region('header', CaptureRegion(192 * 1024, 64 * 1024))
+
+    def post_process(self):
+        # After reading a chunk, we may have the following conditions:
+        #
+        # 1. We may have just completed the header region, and if so,
+        #    we need to immediately read and calculate the location of
+        #    the metadata region, as it may be starting in the same
+        #    read we just did.
+        # 2. We may have just completed the metadata region, and if so,
+        #    we need to immediately calculate the location of the
+        #    "virtual disk size" record, as it may be starting in the
+        #    same read we just did.
+        if self.region('header').complete and not self.has_region('metadata'):
+            region = self._find_meta_region()
+            if region:
+                self.new_region('metadata', region)
+        elif self.has_region('metadata') and not self.has_region('vds'):
+            region = self._find_meta_entry(self.VIRTUAL_DISK_SIZE)
+            if region:
+                self.new_region('vds', region)
+
+    @property
+    def format_match(self):
+        return self.region('ident').data.startswith(b'vhdxfile')
+
+    @staticmethod
+    def _guid(buf):
+        """Format a MSFT GUID from the 16-byte input buffer."""
+        guid_format = '<IHHBBBBBBBB'
+        return '%08X-%04X-%04X-%02X%02X-%02X%02X%02X%02X%02X%02X' % (
+            struct.unpack(guid_format, buf))
+
+    def _find_meta_region(self):
+        # The region table entries start after a 16-byte table header
+        region_entry_first = 16
+
+        # Parse the region table header to find the number of regions
+        regi, cksum, count, reserved = struct.unpack(
+            '<IIII', self.region('header').data[:16])
+        if regi != 0x69676572:
+            raise ImageFormatError('Region signature not found at %x' % (
+                self.region('header').offset))
+
+        if count >= 2048:
+            raise ImageFormatError('Region count is %i (limit 2047)' % count)
+
+        # Process the regions until we find the metadata one; grab the
+        # offset and return
+        self._log.debug('Region entry first is %x', region_entry_first)
+        self._log.debug('Region entries %i', count)
+        meta_offset = 0
+        for i in range(0, count):
+            entry_start = region_entry_first + (i * 32)
+            entry_end = entry_start + 32
+            entry = self.region('header').data[entry_start:entry_end]
+            self._log.debug('Entry offset is %x', entry_start)
+
+            # GUID is the first 16 bytes
+            guid = self._guid(entry[:16])
+            if guid == self.METAREGION:
+                # This entry is the metadata region entry
+                meta_offset, meta_len, meta_req = struct.unpack(
+                    '<QII', entry[16:])
+                self._log.debug('Meta entry %i specifies offset: %x',
+                                i, meta_offset)
+                # NOTE(danms): The meta_len in the region descriptor is the
+                # entire size of the metadata table and data. This can be
+                # very large, so we should only capture the size required
+                # for the maximum length of the table, which is one 32-byte
+                # table header, plus up to 2047 32-byte entries.
+                meta_len = 2048 * 32
+                return CaptureRegion(meta_offset, meta_len)
+
+        self._log.warning('Did not find metadata region')
+        return None
+
+    def _find_meta_entry(self, desired_guid):
+        meta_buffer = self.region('metadata').data
+        if len(meta_buffer) < 32:
+            # Not enough data yet for full header
+            return None
+
+        # Make sure we found the metadata region by checking the signature
+        sig, reserved, count = struct.unpack('<8sHH', meta_buffer[:12])
+        if sig != b'metadata':
+            raise ImageFormatError(
+                'Invalid signature for metadata region: %r' % sig)
+
+        entries_size = 32 + (count * 32)
+        if len(meta_buffer) < entries_size:
+            # Not enough data yet for all metadata entries. This is not
+            # strictly necessary as we could process whatever we have until
+            # we find the V-D-S one, but there are only 2047 32-byte
+            # entries max (~64k).
+            return None
+
+        if count >= 2048:
+            raise ImageFormatError(
+                'Metadata item count is %i (limit 2047)' % count)
+
+        for i in range(0, count):
+            entry_offset = 32 + (i * 32)
+            guid = self._guid(meta_buffer[entry_offset:entry_offset + 16])
+            if guid == desired_guid:
+                # Found the item we are looking for by id.
+                # Stop our region from capturing
+                item_offset, item_length, _reserved = struct.unpack(
+                    '<III',
+                    meta_buffer[entry_offset + 16:entry_offset + 28])
+                item_length = min(item_length,
+                                  self.VHDX_METADATA_TABLE_MAX_SIZE)
+                self.region('metadata').length = len(meta_buffer)
+                self._log.debug('Found entry at offset %x', item_offset)
+                # Metadata item offset is from the beginning of the metadata
+                # region, not the file.
+                return CaptureRegion(
+                    self.region('metadata').offset + item_offset,
+                    item_length)
+
+        self._log.warning('Did not find guid %s', desired_guid)
+        return None
+
+    @property
+    def virtual_size(self):
+        # Until we have found the offset and have enough metadata buffered
+        # to read it, return "unknown"
+        if not self.has_region('vds') or not self.region('vds').complete:
+            return 0
+
+        size, = struct.unpack('<Q', self.region('vds').data)
+        return size
+
+    def __str__(self):
+        return 'vhdx'
+
+
+# The VMDK format comes in a large number of variations, but the
+# single-file 'monolithicSparse' version 4 one is mostly what we care
+# about. It contains a 512-byte little-endian header, followed by a
+# variable-length "descriptor" region of text. The header looks like:
+#
+#   Dec  Hex  Name
+#     0 0x00  4-byte magic string 'KDMV'
+#     4 0x04  Version (uint32_t)
+#     8 0x08  Flags (uint32_t, unused by us)
+#    16 0x10  Number of 512 byte sectors in the disk (uint64_t)
+#    24 0x18  Granularity (uint64_t, unused by us)
+#    32 0x20  Descriptor offset in 512-byte sectors (uint64_t)
+#    40 0x28  Descriptor size in 512-byte sectors (uint64_t)
+#
+# After we have the header, we need to find the descriptor region,
+# which starts at the sector identified in the "descriptor offset"
+# field, and is "descriptor size" 512-byte sectors long. Once we have
+# that region, we need to parse it as text, looking for the
+# createType=XXX line that specifies the mechanism by which the data
+# extents are stored in this file. We only support the
+# "monolithicSparse" format, so we just need to confirm that this file
+# contains that specifier.
+#
+# https://www.vmware.com/app/vmdk/?src=vmdk
+class VMDKInspector(FileInspector):
+    """vmware VMDK format (monolithicSparse and streamOptimized variants only)
+
+    This needs to store the 512 byte header and the descriptor region
+    which should be just after that. The descriptor region is some
+    variable number of 512 byte sectors, but is just text defining the
+    layout of the disk.
+    """
+
+    # The beginning and max size of the descriptor is also hardcoded in Qemu
+    # at 0x200 and 1MB - 1
+    DESC_OFFSET = 0x200
+    DESC_MAX_SIZE = (1 << 20) - 1
+    GD_AT_END = 0xffffffffffffffff
+
+    def __init__(self, *a, **k):
+        super(VMDKInspector, self).__init__(*a, **k)
+        self.new_region('header', CaptureRegion(0, 512))
+
+    def post_process(self):
+        # If we have just completed the header region, we need to calculate
+        # the location and length of the descriptor, which should immediately
+        # follow and may have been partially-read in this read.
+        if not self.region('header').complete:
+            return
+
+        (sig, ver, _flags, _sectors, _grain, desc_sec, desc_num,
+         _numGTEsperGT, _rgdOffset, gdOffset) = struct.unpack(
+            '<4sIIQQQQIQQ', self.region('header').data[:64])
+
+        if sig != b'KDMV':
+            raise ImageFormatError('Signature KDMV not found: %r' % sig)
+
+        if ver not in (1, 2, 3):
+            raise ImageFormatError('Unsupported format version %i' % ver)
+
+        if gdOffset == self.GD_AT_END:
+            # This means we have a footer, which takes precedence over the
+            # header, which we cannot support since we stream.
+            raise ImageFormatError('Unsupported VMDK footer')
+
+        # Since we parse both desc_sec and desc_num (the location of the
+        # VMDK's descriptor, expressed in 512 bytes sectors) we enforce a
+        # check on the bounds to create a reasonable CaptureRegion. This
+        # is similar to how it's done in qemu.
+        desc_offset = desc_sec * 512
+        desc_size = min(desc_num * 512, self.DESC_MAX_SIZE)
+        if desc_offset != self.DESC_OFFSET:
+            raise ImageFormatError("Wrong descriptor location")
+
+        if not self.has_region('descriptor'):
+            self.new_region('descriptor', CaptureRegion(
+                desc_offset, desc_size))
+
+    @property
+    def format_match(self):
+        return self.region('header').data.startswith(b'KDMV')
+
+    @property
+    def virtual_size(self):
+        if not self.has_region('descriptor'):
+            # Not enough data yet
+            return 0
+
+        descriptor_rgn = self.region('descriptor')
+        if not descriptor_rgn.complete:
+            # Not enough data yet
+            return 0
+
+        descriptor = descriptor_rgn.data
+        type_idx = descriptor.index(b'createType="') + len(b'createType="')
+        type_end = descriptor.find(b'"', type_idx)
+        # Make sure we don't grab and log a huge chunk of data in a
+        # maliciously-formatted descriptor region
+        if type_end - type_idx < 64:
+            vmdktype = descriptor[type_idx:type_end]
+        else:
+            vmdktype = b'formatnotfound'
+        if vmdktype not in (b'monolithicSparse', b'streamOptimized'):
+            LOG.warning('Unsupported VMDK format %s', vmdktype)
+            return 0
+
+        # If we have the descriptor, we definitely have the header
+        _sig, _ver, _flags, sectors, _grain, _desc_sec, _desc_num = (
+            struct.unpack('<IIIQQQQ', self.region('header').data[:44]))
+
+        return sectors * 512
+
+    def safety_check(self):
+        if (not self.has_region('descriptor')
+                or not self.region('descriptor').complete):
+            return False
+
+        try:
+            # Descriptor is padded to 512 bytes
+            desc_data = self.region('descriptor').data.rstrip(b'\x00')
+            # Descriptor is actually case-insensitive ASCII text
+            desc_text = desc_data.decode('ascii').lower()
+        except UnicodeDecodeError:
+            LOG.error('VMDK descriptor failed to decode as ASCII')
+            raise ImageFormatError('Invalid VMDK descriptor data')
+
+        extent_access = ('rw', 'rdonly', 'noaccess')
+        header_fields = []
+        extents = []
+        ddb = []
+
+        # NOTE(danms): Cautiously parse the VMDK descriptor. Each line must
+        # be something we understand, otherwise we refuse it.
+        for line in [x.strip() for x in desc_text.split('\n')]:
+            if line.startswith('#') or not line:
+                # Blank or comment lines are ignored
+                continue
+            elif line.startswith('ddb'):
+                # DDB lines are allowed (but not used by us)
+                ddb.append(line)
+            elif '=' in line and ' ' not in line.split('=')[0]:
+                # Header fields are a single word followed by an '=' and some
+                # value
+                header_fields.append(line)
+            elif line.split(' ')[0] in extent_access:
+                # Extent lines start with one of the three access modes
+                extents.append(line)
+            else:
+                # Anything else results in a rejection
+                LOG.error('Unsupported line %r in VMDK descriptor', line)
+                raise ImageFormatError('Invalid VMDK descriptor data')
+
+        # Check all the extent lines for concerning content
+        for extent_line in extents:
+            if '/' in extent_line:
+                LOG.error('Extent line %r contains unsafe characters',
+                          extent_line)
+                return False
+
+        if not extents:
+            LOG.error('VMDK file specified no extents')
+            return False
+
+        return True
+
+    def __str__(self):
+        return 'vmdk'
+
+
+# The VirtualBox VDI format consists of a 512-byte little-endian
+# header, some of which we care about:
+#
+#  Dec   Hex  Name
+#   64  0x40  4-byte Magic (0xbeda107f)
+#   . . .
+#  368 0x170  Size in bytes (uint64_t)
+#
+# https://github.com/qemu/qemu/blob/master/block/vdi.c
+class VDIInspector(FileInspector):
+    """VirtualBox VDI format
+
+    This only needs to store the first 512 bytes of the image.
+    """
+
+    def __init__(self, *a, **k):
+        super(VDIInspector, self).__init__(*a, **k)
+        self.new_region('header', CaptureRegion(0, 512))
+
+    @property
+    def format_match(self):
+        if not self.region('header').complete:
+            return False
+
+        signature, = struct.unpack('<I', self.region('header').data[0x40:0x44])
+        return signature == 0xbeda107f
+
+    @property
+    def virtual_size(self):
+        if not self.region('header').complete:
+            return 0
+        if not self.format_match:
+            return 0
+
+        size, = struct.unpack('<Q', self.region('header').data[0x170:0x178])
+        return size
+
+    def __str__(self):
+        return 'vdi'
+
+
+class ISOInspector(FileInspector):
+    """ISO 9660 and UDF format
+
+    we need to check the first 32KB + descriptor size
+    to look for the ISO 9660 or UDF signature.
+
+    http://wiki.osdev.org/ISO_9660
+    http://wiki.osdev.org/UDF
+    mkisofs --help  | grep udf
+
+    The Universal Disc Format or UDF is the filesystem used on DVDs and
+    Blu-Ray discs.UDF is an extension of ISO 9660 and shares the same
+    header structure and initial layout.
+
+    Like the CDFS(ISO 9660) file system,
+    the UDF file system uses a 2048 byte sector size,
+    and it designates that the first 16 sectors can be used by the OS
+    to store proprietary data or boot logic.
+
+    That means we need to check the first 32KB + descriptor size
+    to look for the ISO 9660 or UDF signature.
+    both formats have an extent based layout, so we can't determine
+    ahead of time where the descriptor will be located.
+
+    fortunately, the ISO 9660 and UDF formats have a Primary Volume Descriptor
+    located at the beginning of the image, which contains the volume size.
+
+    """
+
+    def __init__(self, *a, **k):
+        super(ISOInspector, self).__init__(*a, **k)
+        self.new_region('system_area', CaptureRegion(0, 32 * units.Ki))
+        self.new_region('header', CaptureRegion(32 * units.Ki, 2 * units.Ki))
+
+    @property
+    def format_match(self):
+        if not self.complete:
+            return False
+        signature = self.region('header').data[1:6]
+        assert len(signature) == 5
+        return signature in (b'CD001', b'NSR02', b'NSR03')
+
+    @property
+    def virtual_size(self):
+        if not self.complete:
+            return 0
+        if not self.format_match:
+            return 0
+
+        # the header size is 2KB or 1 sector
+        # the first header field is the descriptor type which is 1 byte
+        # the second field is the standard identifier which is 5 bytes
+        # the third field is the version which is 1 byte
+        # the rest of the header contains type specific data is 2041 bytes
+        # see http://wiki.osdev.org/ISO_9660#The_Primary_Volume_Descriptor
+
+        # we need to check that the descriptor type is 1
+        # to ensure that this is a primary volume descriptor
+        descriptor_type = self.region('header').data[0]
+        if descriptor_type != 1:
+            return 0
+        # The size in bytes of a logical block is stored at offset 128
+        # and is 2 bytes long encoded in both little and big endian
+        # int16_LSB-MSB so the field is 4 bytes long
+        logical_block_size_data = self.region('header').data[128:132]
+        assert len(logical_block_size_data) == 4
+        # given the encoding we only need to read half the field so we
+        # can use the first 2 bytes which are the little endian part
+        # this is normally 2048 or 2KB but we need to check as it can be
+        # different according to the ISO 9660 standard.
+        logical_block_size, = struct.unpack('<H', logical_block_size_data[:2])
+        # The volume space size is the total number of logical blocks
+        # and is stored at offset 80 and is 8 bytes long
+        # as with the logical block size the field is encoded in both
+        # little and big endian as an int32_LSB-MSB
+        volume_space_size_data = self.region('header').data[80:88]
+        assert len(volume_space_size_data) == 8
+        # given the encoding we only need to read half the field so we
+        # can use the first 4 bytes which are the little endian part
+        volume_space_size, = struct.unpack('<L', volume_space_size_data[:4])
+        # the virtual size is the volume space size * logical block size
+        return volume_space_size * logical_block_size
+
+    def __str__(self):
+        return 'iso'
+
+
+class InfoWrapper(object):
+    """A file-like object that wraps another and updates a format inspector.
+
+    This passes chunks to the format inspector while reading. If the inspector
+    fails, it logs the error and stops calling it, but continues proxying data
+    from the source to its user.
+    """
+
+    def __init__(self, source, fmt):
+        self._source = source
+        self._format = fmt
+        self._error = False
+
+    def __iter__(self):
+        return self
+
+    def _process_chunk(self, chunk):
+        if not self._error:
+            try:
+                self._format.eat_chunk(chunk)
+            except Exception as e:
+                # Absolutely do not allow the format inspector to break
+                # our streaming of the image. If we failed, just stop
+                # trying, log and keep going.
+                LOG.error('Format inspector failed, aborting: %s', e)
+                self._error = True
+
+    def __next__(self):
+        try:
+            chunk = next(self._source)
+        except StopIteration:
+            raise
+        self._process_chunk(chunk)
+        return chunk
+
+    def read(self, size):
+        chunk = self._source.read(size)
+        self._process_chunk(chunk)
+        return chunk
+
+    def close(self):
+        if hasattr(self._source, 'close'):
+            self._source.close()
+
+
+ALL_FORMATS = {
+    'raw': FileInspector,
+    'qcow2': QcowInspector,
+    'vhd': VHDInspector,
+    'vhdx': VHDXInspector,
+    'vmdk': VMDKInspector,
+    'vdi': VDIInspector,
+    'qed': QEDInspector,
+    'iso': ISOInspector,
+}
+
+
+def get_inspector(format_name):
+    """Returns a FormatInspector class based on the given name.
+
+    :param format_name: The name of the disk_format (raw, qcow2, etc).
+    :returns: A FormatInspector or None if unsupported.
+    """
+
+    return ALL_FORMATS.get(format_name)
+
+
+def detect_file_format(filename):
+    """Attempts to detect the format of a file.
+
+    This runs through a file one time, running all the known inspectors in
+    parallel. It stops reading the file once all of them matches or all of
+    them are sure they don't match.
+
+    :param filename: The path to the file to inspect.
+    :returns: A FormatInspector instance matching the file.
+    :raises: ImageFormatError if multiple formats are detected.
+    """
+    inspectors = {k: v() for k, v in ALL_FORMATS.items()}
+    detections = []
+    with open(filename, 'rb') as f:
+        for chunk in chunked_reader(f):
+            for format, inspector in list(inspectors.items()):
+                try:
+                    inspector.eat_chunk(chunk)
+                except ImageFormatError:
+                    # No match, so stop considering this format
+                    inspectors.pop(format)
+                    continue
+                if (inspector.format_match and inspector.complete
+                        and format != 'raw'):
+                    # record all match (other than raw)
+                    detections.append(inspector)
+                    inspectors.pop(format)
+            if all(i.complete for i in inspectors.values()):
+                # If all the inspectors are sure they are not a match, avoid
+                # reading to the end of the file to settle on 'raw'.
+                break
+
+    if len(detections) > 1:
+        all_formats = [str(inspector) for inspector in detections]
+        raise ImageFormatError(
+            'Multiple formats detected: %s' % ', '.join(all_formats))
+
+    return inspectors['raw'] if not detections else detections[0]
diff --git a/ironic_python_agent/partition_utils.py b/ironic_python_agent/partition_utils.py
index f410ef62f..555f6a271 100644
--- a/ironic_python_agent/partition_utils.py
+++ b/ironic_python_agent/partition_utils.py
@@ -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:
diff --git a/ironic_python_agent/qemu_img.py b/ironic_python_agent/qemu_img.py
new file mode 100644
index 000000000..7ce38a09a
--- /dev/null
+++ b/ironic_python_agent/qemu_img.py
@@ -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
diff --git a/ironic_python_agent/tests/unit/base.py b/ironic_python_agent/tests/unit/base.py
index 59d78ce8a..fce6007ae 100644
--- a/ironic_python_agent/tests/unit/base.py
+++ b/ironic_python_agent/tests/unit/base.py
@@ -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
diff --git a/ironic_python_agent/tests/unit/extensions/test_standby.py b/ironic_python_agent/tests/unit/extensions/test_standby.py
index a0b161036..c5d467fc1 100644
--- a/ironic_python_agent/tests/unit/extensions/test_standby.py
+++ b/ironic_python_agent/tests/unit/extensions/test_standby.py
@@ -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)
diff --git a/ironic_python_agent/tests/unit/test_disk_utils.py b/ironic_python_agent/tests/unit/test_disk_utils.py
index fe7772360..18ea8fd8d 100644
--- a/ironic_python_agent/tests/unit/test_disk_utils.py
+++ b/ironic_python_agent/tests/unit/test_disk_utils.py
@@ -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')
diff --git a/ironic_python_agent/tests/unit/test_format_inspector.py b/ironic_python_agent/tests/unit/test_format_inspector.py
new file mode 100644
index 000000000..405e4492d
--- /dev/null
+++ b/ironic_python_agent/tests/unit/test_format_inspector.py
@@ -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)
diff --git a/ironic_python_agent/tests/unit/test_partition_utils.py b/ironic_python_agent/tests/unit/test_partition_utils.py
index e99ca4584..a78a9fcc3 100644
--- a/ironic_python_agent/tests/unit/test_partition_utils.py
+++ b/ironic_python_agent/tests/unit/test_partition_utils.py
@@ -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):
diff --git a/ironic_python_agent/tests/unit/test_qemu_img.py b/ironic_python_agent/tests/unit/test_qemu_img.py
new file mode 100644
index 000000000..8645eb8c6
--- /dev/null
+++ b/ironic_python_agent/tests/unit/test_qemu_img.py
@@ -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,
+        ])
diff --git a/releasenotes/notes/image-security-5c23b890409101c9.yaml b/releasenotes/notes/image-security-5c23b890409101c9.yaml
new file mode 100644
index 000000000..3bb9a66eb
--- /dev/null
+++ b/releasenotes/notes/image-security-5c23b890409101c9.yaml
@@ -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.