diff --git a/ironic_python_agent/config.py b/ironic_python_agent/config.py
index 75527b39c..f29277c64 100644
--- a/ironic_python_agent/config.py
+++ b/ironic_python_agent/config.py
@@ -378,7 +378,7 @@ cli_opts = [
                      'security, but this option allows disabling it if there '
                      'is a compatibility concern.'),
     cfg.ListOpt('permitted_image_formats',
-                default='raw,qcow2',
+                default='raw,gpt,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 '
diff --git a/ironic_python_agent/disk_utils.py b/ironic_python_agent/disk_utils.py
index dfa7b27c6..6f00b36f2 100644
--- a/ironic_python_agent/disk_utils.py
+++ b/ironic_python_agent/disk_utils.py
@@ -32,11 +32,11 @@ from ironic_lib import utils
 from oslo_concurrency import processutils
 from oslo_config import cfg
 from oslo_utils import excutils
+from oslo_utils.imageutils import format_inspector
 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
@@ -54,6 +54,9 @@ MAX_CONFIG_DRIVE_SIZE_MB = 64
 # Maximum disk size supported by MBR is 2TB (2 * 1024 * 1024 MB)
 MAX_DISK_SIZE_MB_SUPPORTED_BY_MBR = 2097152
 
+# NOTE(JayF): Image types we write bit-perfect to disk, no conversion
+RAW_LIKE_IMAGETYPES = ['gpt', 'raw']
+
 
 def list_partitions(device):
     """Get partitions information from given device.
@@ -394,20 +397,20 @@ def dd(src, dst, conv_flags=None):
 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)
+        if not inspector_cls:
+            msg = "Security: Unable to safety check image"
+            LOG.error(msg)
+            raise errors.InvalidImage(details=msg)
+        inspector_cls.safety_check()
 
-    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)
+    except format_inspector.ImageFormatError:
+        msg = "Security: Image matched multiple potential formats"
+        LOG.exception(msg)
+        raise errors.InvalidImage(details=msg)
+
+    except format_inspector.SafetyCheckFailed:
+        msg = "Security: Image failed safety check"
+        LOG.exception(msg)
         raise errors.InvalidImage(details=msg)
 
     return inspector_cls
@@ -418,10 +421,12 @@ def get_and_validate_image_format(filename, ironic_disk_format):
 
     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
+    specified format matches the provided one and that the format is in the
     allowed list of formats.
 
+    If the image format provided by Ironic is a type which doesn't need
+    conversion, we avoid all introspection of the image and use of qemu-img.
+
     It also performs a basic safety check on the image.
 
     This entire process can be bypassed, and the older code path used,
@@ -439,10 +444,10 @@ def get_and_validate_image_format(filename, ironic_disk_format):
         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.
+        if ironic_disk_format in RAW_LIKE_IMAGETYPES:
+            # NOTE(JayF): IPA avoids converting images which claim to be a
+            #             type that doesn't need conversion. This negates any
+            #             security risk the image file could pose.
             img_format = ironic_disk_format
             size = os.path.getsize(filename)
         else:
diff --git a/ironic_python_agent/format_inspector.py b/ironic_python_agent/format_inspector.py
deleted file mode 100644
index 44cdd4ae7..000000000
--- a/ironic_python_agent/format_inspector.py
+++ /dev/null
@@ -1,1044 +0,0 @@
-# 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/tests/unit/test_disk_utils.py b/ironic_python_agent/tests/unit/test_disk_utils.py
index 18ea8fd8d..450b302ce 100644
--- a/ironic_python_agent/tests/unit/test_disk_utils.py
+++ b/ironic_python_agent/tests/unit/test_disk_utils.py
@@ -19,17 +19,17 @@ import stat
 from unittest import mock
 
 from ironic_lib import exception
-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 format_inspector
 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
+from ironic_python_agent.tests.unit import base
 
 CONF = cfg.CONF
 
@@ -53,7 +53,12 @@ class MockFormatInspectorCls(object):
             return (self.virtual_size_mb * units.Mi) + 1 - units.Mi
 
     def safety_check(self):
-        return self.safe
+        if self.safe:
+            return True
+        else:
+            raise format_inspector.SafetyCheckFailed(
+                failures={'mockfail': format_inspector.SafetyViolation()}
+            )
 
 
 def _get_fake_qemu_image_info(file_format='qcow2', virtual_size=0):
@@ -62,7 +67,7 @@ def _get_fake_qemu_image_info(file_format='qcow2', virtual_size=0):
 
 
 @mock.patch.object(utils, 'execute', autospec=True)
-class ListPartitionsTestCase(base.IronicLibTestCase):
+class ListPartitionsTestCase(base.IronicAgentTest):
 
     def test_correct(self, execute_mock):
         output = """
@@ -137,7 +142,7 @@ BYT;
 
 
 @mock.patch.object(utils, 'execute', autospec=True)
-class MakePartitionsTestCase(base.IronicLibTestCase):
+class MakePartitionsTestCase(base.IronicAgentTest):
 
     def setUp(self):
         super(MakePartitionsTestCase, self).setUp()
@@ -337,7 +342,7 @@ class MakePartitionsTestCase(base.IronicLibTestCase):
 
 
 @mock.patch.object(utils, 'execute', autospec=True)
-class DestroyMetaDataTestCase(base.IronicLibTestCase):
+class DestroyMetaDataTestCase(base.IronicAgentTest):
 
     def setUp(self):
         super(DestroyMetaDataTestCase, self).setUp()
@@ -501,7 +506,7 @@ class DestroyMetaDataTestCase(base.IronicLibTestCase):
 
 
 @mock.patch.object(utils, 'execute', autospec=True)
-class GetDeviceByteSizeTestCase(base.IronicLibTestCase):
+class GetDeviceByteSizeTestCase(base.IronicAgentTest):
 
     def setUp(self):
         super(GetDeviceByteSizeTestCase, self).setUp()
@@ -517,7 +522,7 @@ class GetDeviceByteSizeTestCase(base.IronicLibTestCase):
 
 @mock.patch.object(disk_utils, 'dd', autospec=True)
 @mock.patch.object(qemu_img, 'convert_image', autospec=True)
-class PopulateImageTestCase(base.IronicLibTestCase):
+class PopulateImageTestCase(base.IronicAgentTest):
 
     def test_populate_raw_image(self, mock_cg, mock_dd):
         source_format = 'raw'
@@ -538,7 +543,7 @@ class PopulateImageTestCase(base.IronicLibTestCase):
 
 
 @mock.patch('time.sleep', lambda sec: None)
-class OtherFunctionTestCase(base.IronicLibTestCase):
+class OtherFunctionTestCase(base.IronicAgentTest):
 
     @mock.patch.object(os, 'stat', autospec=True)
     @mock.patch.object(stat, 'S_ISBLK', autospec=True)
@@ -616,7 +621,7 @@ class OtherFunctionTestCase(base.IronicLibTestCase):
 
 
 @mock.patch.object(utils, 'execute', autospec=True)
-class FixGptStructsTestCases(base.IronicLibTestCase):
+class FixGptStructsTestCases(base.IronicAgentTest):
 
     def setUp(self):
         super(FixGptStructsTestCases, self).setUp()
@@ -659,7 +664,7 @@ Identified 1 problems!
 
 
 @mock.patch.object(utils, 'execute', autospec=True)
-class TriggerDeviceRescanTestCase(base.IronicLibTestCase):
+class TriggerDeviceRescanTestCase(base.IronicAgentTest):
     def test_trigger(self, mock_execute):
         self.assertTrue(disk_utils.trigger_device_rescan('/dev/fake'))
         mock_execute.assert_has_calls([
@@ -707,7 +712,7 @@ LSBLK_NORMAL = (
 
 
 @mock.patch.object(utils, 'execute', autospec=True)
-class GetDeviceInformationTestCase(base.IronicLibTestCase):
+class GetDeviceInformationTestCase(base.IronicAgentTest):
 
     def test_normal(self, mock_execute):
         mock_execute.return_value = LSBLK_NORMAL, ""
@@ -746,7 +751,7 @@ class GetDeviceInformationTestCase(base.IronicLibTestCase):
 
 
 @mock.patch.object(utils, 'execute', autospec=True)
-class GetPartitionTableTypeTestCase(base.IronicLibTestCase):
+class GetPartitionTableTypeTestCase(base.IronicAgentTest):
     def test_gpt(self, mocked_execute):
         self._test_by_type(mocked_execute, 'gpt', 'gpt')
 
@@ -786,7 +791,7 @@ Number  Start   End     Size    File system  Name  Flags
 
 @mock.patch.object(disk_utils, 'list_partitions', autospec=True)
 @mock.patch.object(disk_utils, 'get_partition_table_type', autospec=True)
-class FindEfiPartitionTestCase(base.IronicLibTestCase):
+class FindEfiPartitionTestCase(base.IronicAgentTest):
 
     def test_find_efi_partition(self, mocked_type, mocked_parts):
         mocked_parts.return_value = [
@@ -826,7 +831,7 @@ class FindEfiPartitionTestCase(base.IronicLibTestCase):
         self.assertIsNone(disk_utils.find_efi_partition('/dev/sda'))
 
 
-class WaitForDisk(base.IronicLibTestCase):
+class WaitForDisk(base.IronicAgentTest):
 
     def setUp(self):
         super(WaitForDisk, self).setUp()
@@ -961,7 +966,7 @@ class WaitForDisk(base.IronicLibTestCase):
         mock_exc.assert_has_calls([fuser_call, fuser_call])
 
 
-class GetAndValidateImageFormat(base.IronicLibTestCase):
+class GetAndValidateImageFormat(base.IronicAgentTest):
     @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):
@@ -1041,7 +1046,7 @@ class GetAndValidateImageFormat(base.IronicLibTestCase):
         mock_info.assert_called_once()
 
 
-class ImageInspectionTest(base.IronicLibTestCase):
+class ImageInspectionTest(base.IronicAgentTest):
     @mock.patch.object(format_inspector, 'detect_file_format', autospec=True)
     def test_image_inspection_pass(self, mock_fi):
         inspector = MockFormatInspectorCls('qcow2', 0, True)
diff --git a/ironic_python_agent/tests/unit/test_format_inspector.py b/ironic_python_agent/tests/unit/test_format_inspector.py
deleted file mode 100644
index 405e4492d..000000000
--- a/ironic_python_agent/tests/unit/test_format_inspector.py
+++ /dev/null
@@ -1,664 +0,0 @@
-# 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/requirements.txt b/requirements.txt
index b943d1f16..a1ec91634 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -4,7 +4,7 @@ oslo.config>=5.2.0 # Apache-2.0
 oslo.concurrency>=3.26.0 # Apache-2.0
 oslo.log>=4.6.1 # Apache-2.0
 oslo.service>=1.24.0 # Apache-2.0
-oslo.utils>=3.34.0 # Apache-2.0
+oslo.utils>=7.3.0 # Apache-2.0
 Pint>=0.5 # BSD
 psutil>=3.2.2 # BSD
 pyudev>=0.18 # LGPLv2.1+