diff options
| -rw-r--r-- | tools/__init__.py | 19 | ||||
| -rw-r--r-- | tools/libvmdk.py | 120 |
2 files changed, 132 insertions, 7 deletions
diff --git a/tools/__init__.py b/tools/__init__.py index 374823f..126d1eb 100644 --- a/tools/__init__.py +++ b/tools/__init__.py @@ -60,6 +60,25 @@ def unmount(path): @log +def rm(path): + """Remove a file. + + Args: + path (str): Path to file. + + Returns: + True if the command was successful, False otherwise. + """ + try: + os.remove(path) + except OSError as e: + L.error("failed to remove file %s: %r", path, e) + return False + + return True + + +@log def rmdir(path): """Remove a directory. diff --git a/tools/libvmdk.py b/tools/libvmdk.py index ca67161..c29d18a 100644 --- a/tools/libvmdk.py +++ b/tools/libvmdk.py @@ -1,9 +1,12 @@ +import io import logging import os +import struct import tempfile +from collections import namedtuple from subprocess import run -from . import log, rmdir +from . import log, rm, rmdir __all__ = ["mount"] @@ -36,6 +39,22 @@ def mount(path): Returns: Path to the directory containing a single virtual file named `vmdk1`. """ + # See also: https://github.com/libyal/libvmdk/issues/7 + # libvmdk currently can't mount VMDK files with type "monolithicSparse" + # if they've been renamed. There is already a created issue that has been + # marked as solved and closed. Since libvmdk still can't handle renamed + # single "monolithicSparse" images, extract name manually from descriptor, + # create a symlink in case of mismatch and try to mount image through + # this symlink. + name = _extract_name_from_descriptor(path) + if renamed := (name and name != os.path.basename(path)): + L.debug("%s has been renamed", path) + tempdir = tempfile.mkdtemp() + symlink = os.path.join(tempdir, name) + os.symlink(path, symlink) + L.debug("created symlink with original name: %s", symlink) + path = symlink + mp = tempfile.mkdtemp() cmd = ["vmdkmount", path, mp] try: @@ -43,12 +62,99 @@ def mount(path): except Exception as e: L.error("failed to execute command %s: %r", cmd, e) rmdir(mp) - return "" + if renamed: + rm(symlink) + rmdir(tempdir) + return None + + if renamed: + rm(symlink) + rmdir(tempdir) + + if os.path.ismount(mp): + return mp out, err = p.stdout.strip(), p.stderr.strip() - if p.returncode or not os.path.ismount(mp): - L.error("retcode: %d, stdout: %s, stderr: %s", p.returncode, out, err) - rmdir(mp) - return "" + L.error("retcode: %d, stdout: %s, stderr: %s", p.returncode, out, err) + rmdir(mp) + + return None + + +@log +def _extract_name_from_descriptor(path): + """ + VMware Virtual Disks Virtual Disk Format 1.1 + https://www.vmware.com/app/vmdk/?src=vmdk + + typedef uint64 SectorType; + typedef uint8 Bool; + + typedef struct SparseExtentHeader { + uint32 magicNumber; + uint32 version; + uint32 flags; + SectorType capacity; + SectorType grainSize; + SectorType descriptorOffset; + SectorType descriptorSize; + uint32 numGTEsPerGT; + SectorType rgdOffset; + SectorType gdOffset; + SectorType overHead; + Bool uncleanShutdown; + char singleEndLineChar; + char nonEndLineChar; + char doubleEndLineChar1; + char doubleEndLineChar2; + uint16 compressAlgorithm; + uint8 pad[433]; + } SparseExtentHeader; + + #define SPARSE_MAGICNUMBER 0x564d444b /* 'V' 'M' 'D' 'K' */ + """ + with open(path, "rb") as fh: + name = "" + if not (magic := fh.read(4)) or magic != b"KDMV": + return None + + fh.seek(-4, io.SEEK_CUR) + + fmt = "<IIIQQQQIQQQ?ccccHB" + size = struct.calcsize(fmt) + data = fh.read(size) + + Header = namedtuple( + "Header", + "magic version flags capacity grain_size desc_offset desc_size " + "num_gtes_per_gt rgd_offset gd_offset overhead is_dirty " + "single_end_line_char non_end_line_char first_dbl_end_line_char " + "second_dbl_end_line_char compress_algorithm pad" + ) + header = Header._make(struct.unpack_from(fmt, data)) + + # check if the descriptor is embedded + if not (header.desc_offset > 0): + return None + + # extract disc descriptor file + fh.seek(header.desc_offset * 512) + data = fh.read(header.desc_size * 512) + descriptor = data.split(b"\x00", 1)[0].decode() + + for line in descriptor.splitlines(): + if not (line := line.strip()) or line.startswith("#"): + continue + + if line.startswith("createType"): + if line[11:].strip('"') != "monolithicSparse": + break + else: + continue + + if line.startswith("RW "): + extent = line.split(" ", 3) + name = extent[3].strip('"') + break - return mp + return name |