Parse the $O file.

decode_objfile.py

# Parse $O File
# Copyright Matthew Seyer 2018
# Apache License Version 2
#
# decode_objfile.py FILE [OUTPUT_TEMPLATE]
#
# Examples:
# Output JSON lines:
# python .\decode_objfile.py '$O'
#
# Output via an output template
# python .\decode_objfile.py '$O' "{mft_reference[entry]},{object_id[uuid]},{object_id[timestamp]}"
#

import sys
import json
import struct
import codecs
import logging
import binascii
import datetime

logging.basicConfig(
    level=logging.ERROR
)


class InvalidIndxPageHeader(Exception):
    def __init__(self, message):
        super(InvalidIndxPageHeader, self).__init__(message)


class FileTime(datetime.datetime):
    """datetime.datetime object is immutable, so we will create a class to inherit
    datetime.datetime so we can set a custom nanosecond.
    """
    def __new__(cls, *args, **kwargs):
        return datetime.datetime.__new__(cls, *args, **kwargs)

    @staticmethod
    def from_dt_object(dt_object, nanoseconds=0):
        ft = FileTime(
            dt_object.year,
            dt_object.month,
            dt_object.day,
            dt_object.hour,
            dt_object.minute,
            dt_object.second,
            dt_object.microsecond
        )
        ft.nanoseconds = nanoseconds
        return ft

    def __str__(self):
        return "{0.year}-{0.month:02}-{0.day:02} {0.hour:02}:{0.minute:02}:{0.second:02}.{0.nanoseconds}".format(self)


class NtfsReference(object):
    def __init__(self, buf):
        self._buffer = buf

    @property
    def reference(self):
        return struct.unpack("<Q", self._buffer[0:8])[0]

    @property
    def entry(self):
        return struct.unpack("<IH", self._buffer[0:6])[0]

    @property
    def sequence(self):
        return struct.unpack("<H", self._buffer[6:8])[0]

    def as_dict(self):
        return {
            "reference": self.reference,
            "entry": self.entry,
            "sequence": self.sequence
        }


class ObjectId(object):
    def __init__(self, buf):
        self._buffer = buf

    def raw(self):
        return bytes(self._buffer)

    @property
    def timestamp(self):
        # http://computerforensics.parsonage.co.uk/downloads/TheMeaningofLIFE.pdf
        # The file ObjectID is a time based version which means it is created using a system time.
        # The time is a 60 bit time value, a count of 100 nanosecond intervals of UTC since midnight
        # at the start of 15th October 1582.

        # Get le uint64
        le_timestamp = struct.unpack("<Q", self._buffer[0:8])[0]

        # remove first 4 bits used for version
        le_timestamp = le_timestamp - (le_timestamp & 0xf000000000000000)

        # see http://computerforensics.parsonage.co.uk/downloads/TheMeaningofLIFE.pdf
        le_timestamp = le_timestamp - 5748192000000000

        dt_object = datetime.datetime(1601, 1, 1) + datetime.timedelta(
            microseconds=le_timestamp / 10
        )

        # filetime is 100 nanosecond resolution
        nanoseconds = str(le_timestamp % 10000000).zfill(7) + '00'

        filetime = FileTime.from_dt_object(
            dt_object, nanoseconds=nanoseconds
        )

        return filetime

    @property
    def timestamp_uint64(self):
        le_timestamp = struct.unpack("<Q", self._buffer[0:8])[0]
        le_timestamp = le_timestamp - (le_timestamp & 0xf000000000000000)
        return le_timestamp

    @property
    def version(self):
        high_order = struct.unpack(">H", self._buffer[6:8])[0]
        return high_order & 0x000f

    @property
    def variant(self):
        field = struct.unpack(">H", self._buffer[8:10])[0]
        return field >> 14

    @property
    def sequence(self):
        field = struct.unpack(">H", self._buffer[8:10])[0]
        return field & 0x3FFF

    @property
    def mac(self):
        return bytes(self._buffer[10:16])

    def as_dict(self):
        return {
            "uuid": str(self),
            "hex": self.raw(),
            "timestamp": str(self.timestamp),
            "timestamp_uint64": self.timestamp_uint64,
            "version": self.version,
            "variant": self.variant,
            "sequence": self.sequence,
            "mac": self.mac
        }

    def __str__(self):
        return "{:08x}-{:04x}-{:04x}-{:02x}{:02x}-{:02x}{:02x}{:02x}{:02x}{:02x}{:02x}".format(
            struct.unpack("<L", self._buffer[0:4])[0],
            struct.unpack("<H", self._buffer[4:6])[0],
            struct.unpack("<H", self._buffer[6:8])[0],
            struct.unpack("<B", self._buffer[8:9])[0],
            struct.unpack("<B", self._buffer[9:10])[0],
            struct.unpack("<B", self._buffer[10:11])[0],
            struct.unpack("<B", self._buffer[11:12])[0],
            struct.unpack("<B", self._buffer[12:13])[0],
            struct.unpack("<B", self._buffer[13:14])[0],
            struct.unpack("<B", self._buffer[14:15])[0],
            struct.unpack("<B", self._buffer[15:16])[0]
        )


class IndexHeader(object):
    def __init__(self, buf):
        update_seq_off = struct.unpack("<H", buf[4:6])[0]
        update_seq_size = struct.unpack("<H", buf[6:8])[0]
        self._buffer = bytearray(buf[0:update_seq_off+update_seq_size*2])

    def block_size(self):
        """The block size of the index will be the update sequence size - 1 * 512.
        The update sequence array gets applied every 512 bytes (the first 2 bytes is the value)
        """
        return (self.update_sequence_size - 1) * 512

    def get_fixup_array(self):
        """Return the update sequence array as a list of 2 bytes each.
        """
        so = self.update_sequence_offset
        eo = self.update_sequence_offset+(self.update_sequence_size*2)
        raw_buf = self._buffer[so:eo]
        return [raw_buf[i:i + 2] for i in range(0, len(raw_buf), 2)]

    @property
    def signature(self):
        return bytes(self._buffer[0:4])

    @property
    def update_sequence_offset(self):
        return struct.unpack("<H", self._buffer[4:6])[0]

    @property
    def update_sequence_size(self):
        return struct.unpack("<H", self._buffer[6:8])[0]

    @property
    def logfile_sequence_number(self):
        return struct.unpack("<Q", self._buffer[8:16])[0]

    @property
    def vcn(self):
        return struct.unpack("<Q", self._buffer[16:24])[0]

    @property
    def index_entry_offset(self):
        return struct.unpack("<I", self._buffer[24:28])[0]

    @property
    def index_entry_size(self):
        return struct.unpack("<I", self._buffer[28:32])[0]

    @property
    def allocated_index_entry_size(self):
        return struct.unpack("<I", self._buffer[32:36])[0]

    @property
    def leaf_node(self):
        return struct.unpack("<B", self._buffer[36:37])[0]

    @property
    def update_sequence(self):
        return binascii.b2a_hex(
            self._buffer[40:42]
        )


class IndexOEntry(object):
    def __init__(self, buf, offset=None):
        self._offset = offset
        logging.debug("Index Entry at Offset: {}".format(self._offset))
        offset = struct.unpack("<H", buf[0:2])[0]
        size = struct.unpack("<H", buf[2:4])[0]
        self._buffer = buf[0:offset+size]

    def get_offset(self):
        return self._offset

    @property
    def data_offset(self):
        """This should be 32"""
        return struct.unpack("<H", self._buffer[0:2])[0]

    @property
    def data_size(self):
        """This should be 56"""
        return struct.unpack("<H", self._buffer[2:4])[0]

    @property
    def entry_size(self):
        """This should be 88"""
        return struct.unpack("<H", self._buffer[8:10])[0]

    @property
    def key_size(self):
        """This should be 16"""
        return struct.unpack("<H", self._buffer[10:12])[0]

    @property
    def flags(self):
        """1 = Entry has subnodes; 2 = Last Entry"""
        return struct.unpack("<H", self._buffer[12:14])[0]

    @property
    def object_id(self):
        return ObjectId(self._buffer[16:32])

    @property
    def mft_reference(self):
        return NtfsReference(
            self._buffer[32:40]
        )

    @property
    def birth_volume(self):
        return ObjectId(self._buffer[40:56])

    @property
    def birth_object(self):
        return ObjectId(self._buffer[56:72])

    @property
    def birth_domain(self):
        return ObjectId(self._buffer[72:88])

    def as_dict(self):
        return {
            "offset": self._offset,
            "flags": self.flags,
            "object_id": self.object_id.as_dict(),
            "mft_reference": self.mft_reference.as_dict(),
            "birth_volume": self.birth_volume.as_dict(),
            "birth_object": self.birth_object.as_dict(),
            "birth_domain": self.birth_domain.as_dict()
        }


class IndexPage(object):
    def __init__(self, file_handle, offset):
        self._offset = offset
        logging.debug("Parsing Index Page at offset: {}".format(self._offset))
        raw_buffer = file_handle.read(64)

        if not bytes(raw_buffer[0:4]) == b"INDX":
            raise(
                InvalidIndxPageHeader(
                    "Invalid Page Header Signature [{}] at offset: {}".format(
                        bytes(raw_buffer[0:4]),
                        self._offset
                    )
                )
            )

        self.header = IndexHeader(
            raw_buffer
        )
        block_size = self.header.block_size()
        self._index_block_buf = bytearray(
            raw_buffer + file_handle.read(
                block_size - 64
            )
        )
        self._fix_raw_block()

    def get_page_size(self):
        return self.header.block_size()

    def _fix_raw_block(self):
        """Apply the update sequence array to their respected offsets.
        """
        fix_up_array = self.header.get_fixup_array()

        # first item in array is the update sequence value
        for i in range(self.header.update_sequence_size-1):
            v1 = fix_up_array[i+1][0]
            v2 = fix_up_array[i + 1][1]
            self._index_block_buf[(i*512)+510] = v1
            self._index_block_buf[(i*512)+511] = v2

    def iter_entries(self):
        pointer = self.header.index_entry_offset + 24
        entry = IndexOEntry(
            self._index_block_buf[pointer:],
            offset=self._offset+pointer
        )
        pointer += entry.entry_size

        while True:
            yield entry

            if pointer >= self.header.index_entry_size:
                break

            entry = IndexOEntry(
                self._index_block_buf[pointer:],
                offset=self._offset+pointer
            )
            pointer += entry.entry_size


class ObjectIdFile(object):
    def __init__(self, file_handle):
        self._file_handle = file_handle
        self._offset = 0
        self._file_handle.seek(0, 2)
        self._file_size = self._file_handle.tell()
        self._file_handle.seek(0, 0)

    def iter_index_pages(self):
        index = IndexPage(
            self._file_handle,
            offset=self._offset
        )
        self._offset += index.get_page_size()

        while True:
            yield index

            if self._offset == self._file_size:
                break

            self._file_handle.seek(
                self._offset
            )

            try:
                index = IndexPage(
                    self._file_handle,
                    offset=self._offset
                )
            except InvalidIndxPageHeader as error:
                logging.error(error)
                break

            self._offset += index.get_page_size()


class ComplexEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, bytes):
            return obj.hex()
        return json.JSONEncoder.default(self, obj)


def main():
    filename = sys.argv[1]

    out_template = None
    if len(sys.argv) > 2:
        out_template = sys.argv[2]

    with open(filename, 'rb') as fh:
        obj_id_file = ObjectIdFile(
            fh
        )

        for index_page in obj_id_file.iter_index_pages():
            for entry in index_page.iter_entries():
                if out_template:
                    print(
                        out_template.format(
                            **entry.as_dict()
                        )
                    )
                else:
                    print(
                        json.dumps(entry.as_dict(), cls=ComplexEncoder)
                    )


if __name__ == "__main__":
    main()