Module `refinery.units.formats.msi`

Expand source code Browse git

from __future__ import annotations

import codecs
import collections
import enum
import re
import struct

from functools import cached_property
from typing import NamedTuple

from refinery.lib import chunks, json
from refinery.lib.cab import Cabinet
from refinery.lib.id import buffer_offset, is_likely_msi, is_likely_pe
from refinery.lib.ole.file import DirectoryEntry, OleFile
from refinery.lib.structures import StructReader
from refinery.lib.types import Param, buf
from refinery.units import Arg
from refinery.units.formats.office.xtdoc import UnpackResult, msi_table_stream_name, xtdoc


class MsiType(enum.IntEnum):
    """
    Known data types for MSI table cell entries.
    """

    Long = 0x104
    Short = 0x502
    Binary = 0x900
    String = 0xD00
    StringLocalized = 0xF00
    Unknown = 0

    def __str__(self):
        return self.name


class MSITableColumnInfo(NamedTuple):
    """
    Represents information about an MSI table column. See also:
    https://doxygen.reactos.org/db/de4/msipriv_8h.html
    """
    number: int
    attributes: int

    @property
    def type(self) -> MsiType:
        try:
            if self.is_integer:
                return MsiType(self.attributes & 0xFFF)
            else:
                return MsiType(self.attributes & 0xF00)
        except Exception:
            return MsiType.Unknown

    @property
    def is_integer(self) -> bool:
        return self.attributes & 0x0F00 < 0x800

    @property
    def is_key(self) -> bool:
        return self.attributes & 0x2000 == 0x2000

    @property
    def is_nullable(self) -> bool:
        return self.attributes & 0x1000 == 0x1000

    @property
    def length(self) -> int:
        vt = self.type
        if vt is MsiType.Long:
            return 4
        if vt is MsiType.Short:
            return 2
        return self.attributes & 0xFF

    @property
    def struct_format(self) -> str:
        vt = self.type
        if vt is MsiType.Long:
            return 'I'
        elif vt is MsiType.Short:
            return 'H'
        else:
            return 'H'


_NEGATED_MSI_SIGNATURE = B'\x2F\x30\xEE\x1F\x5E\x4E\xE5\x1E'

_MSI_FOLDER_PROPERTIES = frozenset({
    'AdminToolsFolder',
    'AppDataFolder',
    'CommonAppDataFolder',
    'CommonFiles64Folder',
    'CommonFilesFolder',
    'DesktopFolder',
    'FavoritesFolder',
    'FontsFolder',
    'LocalAppDataFolder',
    'MyPicturesFolder',
    'NetHoodFolder',
    'PersonalFolder',
    'PrintHoodFolder',
    'ProgramFiles64Folder',
    'ProgramFilesFolder',
    'ProgramMenuFolder',
    'RecentFolder',
    'SendToFolder',
    'StartMenuFolder',
    'StartupFolder',
    'System16Folder',
    'System64Folder',
    'SystemFolder',
    'TempFolder',
    'TemplateFolder',
    'WindowsFolder',
    'WindowsVolume',
})
"""
The names of the properties that Windows Installer defines as the path of a system folder. A
directory whose key is one of them is located at that path, no matter what its `Directory_Parent`
entry says.
"""


class _ScriptItem(NamedTuple):
    row_index: int
    extension: str | None


def _formatted_name(
    name: str,
    properties: dict[str, str],
    files: dict[str, str],
    components: dict[str, str],
) -> str:
    """
    The value of a bracketed name in a string of the type Formatted.
    """
    if not name or name[0] in '~\\' or '{' in name or '}' in name:
        return F'[{name}]'
    prefix, key = name[0], name[1:]
    if prefix not in '%$!#' or not key:
        return properties.get(name, '')
    if prefix == '%':
        key = key.rstrip('%').upper()
        return F'%{key}%'
    if prefix == '$':
        return components.get(key, '')
    return files.get(key, '')


def _formatting_tables(
    tables: dict[str, list[dict[str, str]]],
) -> tuple[dict[str, str], dict[str, str], dict[str, str]]:
    """
    The values of the properties, files, and components that bracketed names in a string of the
    type Formatted refer to, from the Property, File, and Component tables. A file is given by its
    name, a component by a variable for its directory.
    """
    def columns(table: str):
        return (list(row.values()) for row in tables.get(table, ()))

    properties = {row[0]: row[1] for row in columns('Property')}
    files = {row[0]: row[2] for row in columns('File')}
    components = {row[0]: F'%{row[2]}%' for row in columns('Component')}
    return properties, files, components


def _format(
    string: str,
    properties: dict[str, str],
    files: dict[str, str],
    components: dict[str, str],
) -> str:
    """
    Formats a string of the type Formatted like Windows Installer: Each bracketed name is replaced
    by its value, the innermost first. A value that was substituted is not formatted again, so a
    value that names a property stays as it is.
    """
    stack: list[list[str]] = [[]]
    for token in re.finditer(r'\[\\(.)\]|\[~\]|\[|\]|[^\[\]]+', string, re.DOTALL):
        if (escaped := token[1]) is not None:
            stack[-1].append(escaped)
        elif (text := token[0]) == '[~]':
            stack[-1].append('\0')
        elif text == '[':
            stack.append([])
        elif text == ']' and len(stack) > 1:
            name = ''.join(stack.pop())
            stack[-1].append(_formatted_name(name, properties, files, components))
        else:
            stack[-1].append(text)
    while len(stack) > 1:
        rest = ''.join(stack.pop())
        stack[-1].append(F'[{rest}')
    return ''.join(stack[0])


def _default_dir_target(default_dir: str) -> str:
    """
    The long target name of a value of the `DefaultDir` column of the `Directory` table: the
    target is the part before the colon, or the whole value when there is none, and the long
    name is the part after the pipe.
    """
    target = default_dir.partition(':')[0]
    short, _, long = target.partition('|')
    return long or short


def _property_segments(
    value: str,
    properties: dict[str, str],
    visiting: frozenset[str] = frozenset(),
) -> tuple[str, ...]:
    """
    The path segments of the value of a property that anchors a directory: a backslash separates
    segments like a slash does, an empty segment and a drive prefix add nothing, and a bracketed
    name contributes the segments of the property it names, or the name itself when that property
    is not defined.
    """
    segments: list[str] = []
    for part in value.replace('\\', '/').split('/'):
        if not part or (len(part) == 2 and part[1] == ':'):
            continue
        for index, piece in enumerate(re.split(r'\[([^\[\]]*)\]', part)):
            if not piece:
                continue
            if not index % 2:
                segments.append(piece)
            elif piece not in visiting and (inner := properties.get(piece)) is not None:
                segments.extend(_property_segments(inner, properties, visiting | {piece}))
            else:
                segments.append(piece)
    return tuple(segments)


def _install_directories(tables: dict[str, list[dict[str, str]]]) -> dict[str, tuple[str, ...]]:
    """
    The path segments of the install location of every directory in the `Directory` table,
    resolved the way the CostFinalize action of Windows Installer resolves them: a directory
    whose key is a defined property, in the `Property` table or among the system folder
    properties, is located at the property's value no matter what its `Directory_Parent` entry
    says, and any other directory is located in its parent under the target part of its
    `DefaultDir` entry. A directory whose parent chain neither terminates nor reaches a defined
    property has no segments of its own.
    """
    properties = {
        row['Property']: row['Value']
        for row in tables.get('Property', ())
        if row.get('Property') is not None
    }
    parent_of: dict[str, str] = {}
    target_of: dict[str, str] = {}
    for row in tables.get('Directory', ()):
        key = row.get('Directory')
        if not key:
            continue
        parent = row.get('Directory_Parent') or ''
        parent_of[key] = '' if parent == key else parent
        target_of[key] = _default_dir_target(row.get('DefaultDir') or '')

    def anchored(key: str) -> tuple[str, ...] | None:
        if key in _MSI_FOLDER_PROPERTIES:
            return (key,)
        if (value := properties.get(key)) is not None:
            return _property_segments(value, properties)

    resolved: dict[str, tuple[str, ...]] = {}
    for key in parent_of:
        chain: list[str] = []
        current = key
        while current not in resolved:
            if (segments := anchored(current)) is not None:
                resolved[current] = segments
                break
            parent = parent_of[current]
            if parent and parent not in parent_of:
                xtmsi.log_warn(F'directory {current} has an unknown parent directory {parent}')
                parent = ''
            if not parent or current in chain:
                if current in chain:
                    xtmsi.log_warn(F'the parent directories of {current} form a cycle')
                resolved[current] = ()
                break
            chain.append(current)
            current = parent
        segments = resolved[current]
        for member in reversed(chain):
            segments = tuple(
                segment
                for segment in (*segments, target_of[member])
                if segment and segment != '.'
            )
            resolved[member] = segments
    return resolved


class MSIStringData:
    def __init__(self, string_data: buf, string_pool: buf):
        data = StructReader(string_data)
        pool = StructReader(string_pool)
        self.strings: list[bytes] = []
        self.provided_ref_count: list[int] = []
        self.computed_ref_count: list[int] = []
        self.codepage = pool.u16()
        self._unknown = pool.u16()
        while not pool.eof:
            size = pool.u16()
            rc = pool.u16()
            if size == 0 and rc != 0:
                size = pool.u32()
            string = data.read_bytes(size)
            self.strings.append(string)
            self.provided_ref_count.append(rc)
            self.computed_ref_count.append(0)

    @cached_property
    def codec(self):
        try:
            return codecs.lookup(F'cp{self.codepage}').name
        except Exception:
            xtmsi.log_info('failed looking up codec', self.codepage)
            return 'latin1'

    def __len__(self):
        return len(self.strings)

    def __iter__(self):
        yield from range(1, len(self) + 1)

    def __contains__(self, index):
        return 0 < index <= len(self)

    def ref(self, index: int, increment=True) -> str:
        assert index > 0
        index -= 1
        if increment:
            self.computed_ref_count[index] += 1
        string = self.strings[index]
        try:
            return string.decode(self.codec)
        except UnicodeDecodeError:
            return string.decode('latin1')


class xtmsi(xtdoc, docs=(
    'Extract files and metadata from Microsoft Installer (MSI) archives.'
    '{p}{PathExtractorUnit}{p}{0}'
)):
    """
    The synthetic file %s contains parsed MSI table information, similar to the output of the Orca
    tool. Binary streams are placed in a virtual folder called "Binary", and extracted scripts from
    custom actions are separately extracted in a virtual folder named "Action". Files from embedded
    cabinets are extracted at their install path, which the File, Component, and Directory tables
    give for each of them.
    """

    _SYNTHETIC_STREAMS_FILENAME = 'MsiTables.json'
    _SYNTHETIC_STREAMS_TOPLEVEL = 'MsiTables'

    # https://learn.microsoft.com/en-us/windows/win32/msi/summary-list-of-all-custom-action-types
    _CUSTOM_ACTION_TYPES = {
        0x01: 'DLL file stored in a Binary table stream.',
        0x02: 'EXE file stored in a Binary table stream.',
        0x05: 'JScript file stored in a Binary table stream.',
        0x06: 'VBScript file stored in a Binary table stream.',
        0x11: 'DLL file that is installed with a product.',
        0x12: 'EXE file that is installed with a product.',
        0x13: 'Displays a specified error message and returns failure, terminating the installation.',
        0x15: 'JScript file that is installed with a product.',
        0x16: 'VBScript file that is installed with a product.',
        0x22: 'EXE file having a path referencing a directory.',
        0x23: 'Directory set with formatted text.',
        0x25: 'JScript text stored in this sequence table.',
        0x26: 'VBScript text stored in this sequence table.',
        0x32: 'EXE file having a path specified by a property value.',
        0x33: 'Property set with formatted text.',
        0x35: 'JScript text specified by a property value.',
        0x36: 'VBScript text specified by a property value.',
    }

    def __init__(
        self, *paths,
        nocab: Param[bool, Arg.Switch('-N', help='Do not list and extract embedded CAB archives.')] = False,
        **keywords,
    ):
        super().__init__(*paths, nocab=nocab, **keywords)

    def unpack(self, data: buf):
        document = OleFile(self._get_msi_from_overlay(data))
        streams: list[UnpackResult] = []
        streams_by_entry: dict[DirectoryEntry, UnpackResult] = {}
        read: set[UnpackResult] = set()

        for entry, result in self._streams(document):
            streams.append(result)
            streams_by_entry[entry] = result

        def table_stream(table: str) -> UnpackResult | None:
            if (entry := document.find(msi_table_stream_name(table))) is None:
                return None
            if (result := streams_by_entry.get(entry)) is None or result in read:
                return None
            return result

        def stream(table: str):
            if (result := table_stream(table)) is None:
                raise KeyError(F'table stream not found: {table}')
            read.add(result)
            content = result.get_data()
            result.release()
            return content

        def column_formats(table: dict[str, MSITableColumnInfo]) -> str:
            return ''.join(v.struct_format for v in table.values())

        def stream_to_rows(data: buf, row_format: str):
            row_size = struct.calcsize(F'<{row_format}')
            row_count = int(len(data) / row_size)
            reader = StructReader(data)
            columns = [reader.read_struct(F'<{sc * row_count}') for sc in row_format]
            for i in range(row_count):
                yield [int(c[i]) for c in columns]

        tables: dict[str, dict[str, MSITableColumnInfo]] = collections.defaultdict(collections.OrderedDict)
        strings = MSIStringData(stream('_StringData'), stream('_StringPool'))

        for tbl_name_id, col_number, col_name_id, col_attributes in stream_to_rows(stream('_Columns'), 'HHHH'):
            tbl_name = strings.ref(tbl_name_id)
            col_name = strings.ref(col_name_id)
            tables[tbl_name][col_name] = MSITableColumnInfo(col_number, col_attributes)

        table_names_given = {strings.ref(k) for k in chunks.unpack(stream('_Tables'), 2, False)}
        table_names_known = set(tables)

        for name in table_names_known - table_names_given:
            self.log_warn(F'table name known but not given: {name}')
        for name in table_names_given - table_names_known:
            self.log_warn(F'table name given but not known: {name}')

        processed_table_data: dict[str, list[dict[str, str]]] = {}
        postprocessing: list[_ScriptItem] = []

        for table_name, table in tables.items():
            if table_stream(table_name) is None:
                continue
            processed = []
            info = list(table.values())
            keys = list(table.keys())
            temp = [k.strip('_') for k in keys]
            if len(set(keys)) == len(set(temp)):
                keys = temp
            for r, row in enumerate(stream_to_rows(stream(table_name), column_formats(table))):
                values = []
                for index, value in enumerate(row):
                    vt = info[index].type
                    if vt is MsiType.Long:
                        if value != 0:
                            value -= 0x80000000
                    elif vt is MsiType.Short:
                        if value != 0:
                            value -= 0x8000
                    elif value in strings:
                        value = strings.ref(value)
                    elif not info[index].is_integer:
                        value = ''
                    values.append(value)
                entry = dict(zip(keys, values))
                einfo = {t: i for t, i in zip(keys, info)}
                if table_name == 'MsiFileHash':
                    entry['Hash'] = struct.pack(
                        '<IIII',
                        row[2] ^ 0x80000000,
                        row[3] ^ 0x80000000,
                        row[4] ^ 0x80000000,
                        row[5] ^ 0x80000000,
                    ).hex()
                if table_name == 'CustomAction':
                    code = row[1] & 0x3F
                    try:
                        entry['Comment'] = self._CUSTOM_ACTION_TYPES[code]
                    except LookupError:
                        pass
                    t = einfo.get('Target')
                    c = {0x25: 'js', 0x26: 'vbs', 0x33: None}
                    if code in c and t and not t.is_integer:
                        postprocessing.append(_ScriptItem(r, c[code]))
                processed.append(entry)
            if processed:
                processed_table_data[table_name] = processed

        streams = [result for result in streams if result not in read]

        if ca := processed_table_data.get('CustomAction'):
            properties, files, components = _formatting_tables(processed_table_data)
            for item in postprocessing:
                entry = ca[item.row_index]
                try:
                    action: str = entry['Action']
                    target: str = entry['Target']
                except KeyError:
                    continue
                root = F'Action/{action}'
                if item.extension:
                    action = F'{root}.{item.extension}'
                    streams.append(UnpackResult(action, target.encode(self.codec)))
                    continue
                target = _format(target, properties, files, components)
                parts = [part.partition('\x02') for part in target.split('\x01')]
                if not all(part[1] == '\x02' for part in parts):
                    continue
                for name, _, script in parts:
                    if not name.lower().startswith('script'):
                        continue
                    if not script:
                        continue
                    action = F'{root}.{name}'
                    streams.append(UnpackResult(action, script.encode(self.codec)))

        for ignored_stream in [
            'SummaryInformation',
            'DocumentSummaryInformation',
            'DigitalSignature',
            'MsiDigitalSignatureEx'
        ]:
            path = F'[5]{ignored_stream}'
            summary = [result for result in streams if result.path == path]
            streams = [result for result in streams if result.path != path]
            for result in summary:
                result.path = F'Meta/{ignored_stream}'
                yield result

        inconsistencies = 0
        w1 = len(str(len(strings)))
        w2 = len(str(max(max(strings.computed_ref_count), max(strings.provided_ref_count))))
        for k in range(len(strings)):
            c = strings.computed_ref_count[k]
            p = strings.provided_ref_count[k]
            if c != p and not self.log_debug(F'string {k:0{w1}d} reference count computed={c:0{w2}d} provided={p:0{w2}d}'):
                inconsistencies += 1
        if inconsistencies:
            self.log_info(F'found {inconsistencies} incorrect string reference counts')

        def fix_msi_path(path: str):
            prefix, dot, name = path.partition('.')
            if dot == '.' and prefix in processed_table_data:
                path = F'{prefix}/{name}'
            return path

        if self.args.nocab:
            cabs: list[UnpackResult] = []
        else:
            def _iscab(path):
                return media_info and any(item.get('Cabinet', '') == F'#{path}' for item in media_info)
            media_info = processed_table_data.get('Media', [])
            cabs = [item for item in streams if _iscab(item.path)]
            for cab in cabs:
                self.log_info(F'found cab file: {cab.path}')
        if cabs:
            directories = _install_directories(processed_table_data)
            component_directory: dict[str, str] = {}
            for component_info in processed_table_data.get('Component', []):
                try:
                    component_directory[component_info['Component']] = component_info['Directory']
                except KeyError:
                    continue
            separator = self._get_path_separator()
            file_paths: dict[str, str] = {}

            for file_info in processed_table_data.get('File', []):
                try:
                    src_name = file_info['File']
                    dst_name = file_info['FileName']
                    component = file_info['Component']
                except KeyError:
                    continue
                _, _, long = dst_name.partition('|')
                dst_name = long or dst_name
                directory = component_directory.get(component)
                segments = directories.get(directory, ()) if directory else ()
                file_paths[src_name] = separator.join((*segments, dst_name))

            for cab in cabs:
                path = cab.path
                try:
                    _cabinet = Cabinet(memoryview(cab.get_data()))
                    unpacked = _cabinet.process().get_files()
                except Exception as e:
                    self.log_info(F'unable to extract embedded cab file: {e!s}')
                    continue
                base, dot, ext = path.rpartition('.')
                if dot == '.' and ext.lower() == 'cab':
                    path = base
                else:
                    cab.path = F'{path}.cab'
                for result in unpacked:
                    sub_path = file_paths.get(result.name, result.name)
                    sub_path = separator.join((path, sub_path))
                    streams.append(UnpackResult(sub_path, result.extractor()))

        for item in streams:
            item.path = fix_msi_path(item.path)
        ds = UnpackResult(self._SYNTHETIC_STREAMS_FILENAME, json.dumps(processed_table_data))
        streams.append(ds)

        from refinery.units.formats.csv import json_to_csv

        for key, jd in processed_table_data.items():
            sk = key.strip('_')
            if sk not in processed_table_data:
                key = sk
            try:
                tbl = UnpackResult(F'{self._SYNTHETIC_STREAMS_TOPLEVEL}/{key}.csv', json_to_csv(jd))
            except Exception:
                continue
            streams.append(tbl)

        streams.sort(key=lambda item: item.path)
        yield from streams

    @classmethod
    def _get_msi_from_overlay(cls, data: buf) -> buf:
        if is_likely_pe(data):
            from refinery.units.formats.pe import get_pe_size
            view = memoryview(data)
            overlay = view[get_pe_size(data):]
            if is_likely_msi(overlay):
                return overlay
            if (start := buffer_offset(overlay, _NEGATED_MSI_SIGNATURE, 0, 0x1000)) >= 0:
                if (nulls := buffer_offset(overlay, bytes(8), start, start + 0x1000)) >= 0:
                    from refinery.units.blockwise.neg import neg
                    decoded = overlay[start:nulls] | neg | bytearray
                    decoded.extend(overlay[nulls:])
                    if is_likely_msi(decoded):
                        return decoded
        return data

    @classmethod
    def handles(cls, data):
        if is_likely_msi(data):
            return True
        if is_likely_pe(data):
            from refinery.units.formats.pe import get_pe_size
            view = memoryview(data)
            overlay = view[get_pe_size(data):]
            if is_likely_msi(overlay):
                return True
            if overlay[:8] == _NEGATED_MSI_SIGNATURE:
                return True
        return False


if _d := xtmsi.__doc__:
    xtmsi.__doc__ = _d % xtmsi._SYNTHETIC_STREAMS_FILENAME

Classes

class MsiType (*args, **kwds)

Known data types for MSI table cell entries.

Expand source code Browse git

class MsiType(enum.IntEnum):
    """
    Known data types for MSI table cell entries.
    """

    Long = 0x104
    Short = 0x502
    Binary = 0x900
    String = 0xD00
    StringLocalized = 0xF00
    Unknown = 0

    def __str__(self):
        return self.name

Ancestors

enum.IntEnum
builtins.int
enum.ReprEnum
enum.Enum

Class variables

var Long: The type of the None singleton.
var Short: The type of the None singleton.
var Binary: The type of the None singleton.
var String: The type of the None singleton.
var StringLocalized: The type of the None singleton.
var Unknown: The type of the None singleton.

class MSITableColumnInfo (number, attributes)

Represents information about an MSI table column. See also: https://doxygen.reactos.org/db/de4/msipriv_8h.html

Expand source code Browse git

class MSITableColumnInfo(NamedTuple):
    """
    Represents information about an MSI table column. See also:
    https://doxygen.reactos.org/db/de4/msipriv_8h.html
    """
    number: int
    attributes: int

    @property
    def type(self) -> MsiType:
        try:
            if self.is_integer:
                return MsiType(self.attributes & 0xFFF)
            else:
                return MsiType(self.attributes & 0xF00)
        except Exception:
            return MsiType.Unknown

    @property
    def is_integer(self) -> bool:
        return self.attributes & 0x0F00 < 0x800

    @property
    def is_key(self) -> bool:
        return self.attributes & 0x2000 == 0x2000

    @property
    def is_nullable(self) -> bool:
        return self.attributes & 0x1000 == 0x1000

    @property
    def length(self) -> int:
        vt = self.type
        if vt is MsiType.Long:
            return 4
        if vt is MsiType.Short:
            return 2
        return self.attributes & 0xFF

    @property
    def struct_format(self) -> str:
        vt = self.type
        if vt is MsiType.Long:
            return 'I'
        elif vt is MsiType.Short:
            return 'H'
        else:
            return 'H'

Ancestors

builtins.tuple

Instance variables

var number

Alias for field number 0

Expand source code Browse git

class MSITableColumnInfo(NamedTuple):
    """
    Represents information about an MSI table column. See also:
    https://doxygen.reactos.org/db/de4/msipriv_8h.html
    """
    number: int
    attributes: int

    @property
    def type(self) -> MsiType:
        try:
            if self.is_integer:
                return MsiType(self.attributes & 0xFFF)
            else:
                return MsiType(self.attributes & 0xF00)
        except Exception:
            return MsiType.Unknown

    @property
    def is_integer(self) -> bool:
        return self.attributes & 0x0F00 < 0x800

    @property
    def is_key(self) -> bool:
        return self.attributes & 0x2000 == 0x2000

    @property
    def is_nullable(self) -> bool:
        return self.attributes & 0x1000 == 0x1000

    @property
    def length(self) -> int:
        vt = self.type
        if vt is MsiType.Long:
            return 4
        if vt is MsiType.Short:
            return 2
        return self.attributes & 0xFF

    @property
    def struct_format(self) -> str:
        vt = self.type
        if vt is MsiType.Long:
            return 'I'
        elif vt is MsiType.Short:
            return 'H'
        else:
            return 'H'

var attributes

Alias for field number 1

Expand source code Browse git

class MSITableColumnInfo(NamedTuple):
    """
    Represents information about an MSI table column. See also:
    https://doxygen.reactos.org/db/de4/msipriv_8h.html
    """
    number: int
    attributes: int

    @property
    def type(self) -> MsiType:
        try:
            if self.is_integer:
                return MsiType(self.attributes & 0xFFF)
            else:
                return MsiType(self.attributes & 0xF00)
        except Exception:
            return MsiType.Unknown

    @property
    def is_integer(self) -> bool:
        return self.attributes & 0x0F00 < 0x800

    @property
    def is_key(self) -> bool:
        return self.attributes & 0x2000 == 0x2000

    @property
    def is_nullable(self) -> bool:
        return self.attributes & 0x1000 == 0x1000

    @property
    def length(self) -> int:
        vt = self.type
        if vt is MsiType.Long:
            return 4
        if vt is MsiType.Short:
            return 2
        return self.attributes & 0xFF

    @property
    def struct_format(self) -> str:
        vt = self.type
        if vt is MsiType.Long:
            return 'I'
        elif vt is MsiType.Short:
            return 'H'
        else:
            return 'H'

var type

Expand source code Browse git

@property
def type(self) -> MsiType:
    try:
        if self.is_integer:
            return MsiType(self.attributes & 0xFFF)
        else:
            return MsiType(self.attributes & 0xF00)
    except Exception:
        return MsiType.Unknown

var is_integer

Expand source code Browse git

@property
def is_integer(self) -> bool:
    return self.attributes & 0x0F00 < 0x800

var is_key

Expand source code Browse git

@property
def is_key(self) -> bool:
    return self.attributes & 0x2000 == 0x2000

var is_nullable

Expand source code Browse git

@property
def is_nullable(self) -> bool:
    return self.attributes & 0x1000 == 0x1000

var length

Expand source code Browse git

@property
def length(self) -> int:
    vt = self.type
    if vt is MsiType.Long:
        return 4
    if vt is MsiType.Short:
        return 2
    return self.attributes & 0xFF

var struct_format

Expand source code Browse git

@property
def struct_format(self) -> str:
    vt = self.type
    if vt is MsiType.Long:
        return 'I'
    elif vt is MsiType.Short:
        return 'H'
    else:
        return 'H'

class MSIStringData (string_data, string_pool)

Expand source code Browse git

class MSIStringData:
    def __init__(self, string_data: buf, string_pool: buf):
        data = StructReader(string_data)
        pool = StructReader(string_pool)
        self.strings: list[bytes] = []
        self.provided_ref_count: list[int] = []
        self.computed_ref_count: list[int] = []
        self.codepage = pool.u16()
        self._unknown = pool.u16()
        while not pool.eof:
            size = pool.u16()
            rc = pool.u16()
            if size == 0 and rc != 0:
                size = pool.u32()
            string = data.read_bytes(size)
            self.strings.append(string)
            self.provided_ref_count.append(rc)
            self.computed_ref_count.append(0)

    @cached_property
    def codec(self):
        try:
            return codecs.lookup(F'cp{self.codepage}').name
        except Exception:
            xtmsi.log_info('failed looking up codec', self.codepage)
            return 'latin1'

    def __len__(self):
        return len(self.strings)

    def __iter__(self):
        yield from range(1, len(self) + 1)

    def __contains__(self, index):
        return 0 < index <= len(self)

    def ref(self, index: int, increment=True) -> str:
        assert index > 0
        index -= 1
        if increment:
            self.computed_ref_count[index] += 1
        string = self.strings[index]
        try:
            return string.decode(self.codec)
        except UnicodeDecodeError:
            return string.decode('latin1')

Instance variables

var codec

Expand source code Browse git

@cached_property
def codec(self):
    try:
        return codecs.lookup(F'cp{self.codepage}').name
    except Exception:
        xtmsi.log_info('failed looking up codec', self.codepage)
        return 'latin1'

Methods

def ref(self, index, increment=True)

Expand source code Browse git

def ref(self, index: int, increment=True) -> str:
    assert index > 0
    index -= 1
    if increment:
        self.computed_ref_count[index] += 1
    string = self.strings[index]
    try:
        return string.decode(self.codec)
    except UnicodeDecodeError:
        return string.decode('latin1')

class xtmsi (*paths, nocab=False, path=b'path', regex=False, exact=False, fuzzy=0, drop_path=False, join_path=False, list=False, exclude=None)

Extract files and metadata from Microsoft Installer (MSI) archives.

This unit extracts items with an associated virtual path from a container; each extracted item is emitted as a separate chunk with a corresponding meta variable named "path".

Positional arguments to PathExtractorUnit are patterns to filter the extracted items. Use the -x flag to add an exclusion pattern. To extract all files with a foo or bar extension, but none that has the word "temp" in its path:

PathExtractorUnit .foo .bar -x temp

To view only the paths of all chunks, use the listing switch:

emit data | ... | PathExtractorUnit -l

Otherwise, extracted items are written to the standard output port and usually require a frame to properly process. In order to dump all extracted data to disk, the following pipeline can be used:

emit data | ... | PathExtractorUnit [| dump extracted/{path} ]

The value {path} is a placeholder which is substituted by the virtual path of the extracted item. When several items have the same path, each of them gets a number appended to its path in the order in which PathExtractorUnit lists them, and a pattern that matches the path before the number selects all of them. When using PathExtractorUnit to unpack a file on disk, the following pattern can be useful:

ef pack.bin [| PathExtractorUnit -j | d2p ]

The unit ef is also a path extractor. By specifying -j (or --join), the paths of extracted items are combined. Here, d2p is a shortcut for dump {path}. It deconflicts the joined paths with the local file system: If pack.bin contains items one.txt and two.txt, the following local file tree would be the result:

pack.bin
pack/one.txt
pack/two.txt

Finally, the -d (or --drop) switch can be used to not create (or alter) the path metadata at all, which is useful in cases where path metadata from a previous unit should be preserved.

The synthetic file MsiTables.json contains parsed MSI table information, similar to the output of the Orca tool. Binary streams are placed in a virtual folder called "Binary", and extracted scripts from custom actions are separately extracted in a virtual folder named "Action". Files from embedded cabinets are extracted at their install path, which the File, Component, and Directory tables give for each of them.

Expand source code Browse git

class xtmsi(xtdoc, docs=(
    'Extract files and metadata from Microsoft Installer (MSI) archives.'
    '{p}{PathExtractorUnit}{p}{0}'
)):
    """
    The synthetic file %s contains parsed MSI table information, similar to the output of the Orca
    tool. Binary streams are placed in a virtual folder called "Binary", and extracted scripts from
    custom actions are separately extracted in a virtual folder named "Action". Files from embedded
    cabinets are extracted at their install path, which the File, Component, and Directory tables
    give for each of them.
    """

    _SYNTHETIC_STREAMS_FILENAME = 'MsiTables.json'
    _SYNTHETIC_STREAMS_TOPLEVEL = 'MsiTables'

    # https://learn.microsoft.com/en-us/windows/win32/msi/summary-list-of-all-custom-action-types
    _CUSTOM_ACTION_TYPES = {
        0x01: 'DLL file stored in a Binary table stream.',
        0x02: 'EXE file stored in a Binary table stream.',
        0x05: 'JScript file stored in a Binary table stream.',
        0x06: 'VBScript file stored in a Binary table stream.',
        0x11: 'DLL file that is installed with a product.',
        0x12: 'EXE file that is installed with a product.',
        0x13: 'Displays a specified error message and returns failure, terminating the installation.',
        0x15: 'JScript file that is installed with a product.',
        0x16: 'VBScript file that is installed with a product.',
        0x22: 'EXE file having a path referencing a directory.',
        0x23: 'Directory set with formatted text.',
        0x25: 'JScript text stored in this sequence table.',
        0x26: 'VBScript text stored in this sequence table.',
        0x32: 'EXE file having a path specified by a property value.',
        0x33: 'Property set with formatted text.',
        0x35: 'JScript text specified by a property value.',
        0x36: 'VBScript text specified by a property value.',
    }

    def __init__(
        self, *paths,
        nocab: Param[bool, Arg.Switch('-N', help='Do not list and extract embedded CAB archives.')] = False,
        **keywords,
    ):
        super().__init__(*paths, nocab=nocab, **keywords)

    def unpack(self, data: buf):
        document = OleFile(self._get_msi_from_overlay(data))
        streams: list[UnpackResult] = []
        streams_by_entry: dict[DirectoryEntry, UnpackResult] = {}
        read: set[UnpackResult] = set()

        for entry, result in self._streams(document):
            streams.append(result)
            streams_by_entry[entry] = result

        def table_stream(table: str) -> UnpackResult | None:
            if (entry := document.find(msi_table_stream_name(table))) is None:
                return None
            if (result := streams_by_entry.get(entry)) is None or result in read:
                return None
            return result

        def stream(table: str):
            if (result := table_stream(table)) is None:
                raise KeyError(F'table stream not found: {table}')
            read.add(result)
            content = result.get_data()
            result.release()
            return content

        def column_formats(table: dict[str, MSITableColumnInfo]) -> str:
            return ''.join(v.struct_format for v in table.values())

        def stream_to_rows(data: buf, row_format: str):
            row_size = struct.calcsize(F'<{row_format}')
            row_count = int(len(data) / row_size)
            reader = StructReader(data)
            columns = [reader.read_struct(F'<{sc * row_count}') for sc in row_format]
            for i in range(row_count):
                yield [int(c[i]) for c in columns]

        tables: dict[str, dict[str, MSITableColumnInfo]] = collections.defaultdict(collections.OrderedDict)
        strings = MSIStringData(stream('_StringData'), stream('_StringPool'))

        for tbl_name_id, col_number, col_name_id, col_attributes in stream_to_rows(stream('_Columns'), 'HHHH'):
            tbl_name = strings.ref(tbl_name_id)
            col_name = strings.ref(col_name_id)
            tables[tbl_name][col_name] = MSITableColumnInfo(col_number, col_attributes)

        table_names_given = {strings.ref(k) for k in chunks.unpack(stream('_Tables'), 2, False)}
        table_names_known = set(tables)

        for name in table_names_known - table_names_given:
            self.log_warn(F'table name known but not given: {name}')
        for name in table_names_given - table_names_known:
            self.log_warn(F'table name given but not known: {name}')

        processed_table_data: dict[str, list[dict[str, str]]] = {}
        postprocessing: list[_ScriptItem] = []

        for table_name, table in tables.items():
            if table_stream(table_name) is None:
                continue
            processed = []
            info = list(table.values())
            keys = list(table.keys())
            temp = [k.strip('_') for k in keys]
            if len(set(keys)) == len(set(temp)):
                keys = temp
            for r, row in enumerate(stream_to_rows(stream(table_name), column_formats(table))):
                values = []
                for index, value in enumerate(row):
                    vt = info[index].type
                    if vt is MsiType.Long:
                        if value != 0:
                            value -= 0x80000000
                    elif vt is MsiType.Short:
                        if value != 0:
                            value -= 0x8000
                    elif value in strings:
                        value = strings.ref(value)
                    elif not info[index].is_integer:
                        value = ''
                    values.append(value)
                entry = dict(zip(keys, values))
                einfo = {t: i for t, i in zip(keys, info)}
                if table_name == 'MsiFileHash':
                    entry['Hash'] = struct.pack(
                        '<IIII',
                        row[2] ^ 0x80000000,
                        row[3] ^ 0x80000000,
                        row[4] ^ 0x80000000,
                        row[5] ^ 0x80000000,
                    ).hex()
                if table_name == 'CustomAction':
                    code = row[1] & 0x3F
                    try:
                        entry['Comment'] = self._CUSTOM_ACTION_TYPES[code]
                    except LookupError:
                        pass
                    t = einfo.get('Target')
                    c = {0x25: 'js', 0x26: 'vbs', 0x33: None}
                    if code in c and t and not t.is_integer:
                        postprocessing.append(_ScriptItem(r, c[code]))
                processed.append(entry)
            if processed:
                processed_table_data[table_name] = processed

        streams = [result for result in streams if result not in read]

        if ca := processed_table_data.get('CustomAction'):
            properties, files, components = _formatting_tables(processed_table_data)
            for item in postprocessing:
                entry = ca[item.row_index]
                try:
                    action: str = entry['Action']
                    target: str = entry['Target']
                except KeyError:
                    continue
                root = F'Action/{action}'
                if item.extension:
                    action = F'{root}.{item.extension}'
                    streams.append(UnpackResult(action, target.encode(self.codec)))
                    continue
                target = _format(target, properties, files, components)
                parts = [part.partition('\x02') for part in target.split('\x01')]
                if not all(part[1] == '\x02' for part in parts):
                    continue
                for name, _, script in parts:
                    if not name.lower().startswith('script'):
                        continue
                    if not script:
                        continue
                    action = F'{root}.{name}'
                    streams.append(UnpackResult(action, script.encode(self.codec)))

        for ignored_stream in [
            'SummaryInformation',
            'DocumentSummaryInformation',
            'DigitalSignature',
            'MsiDigitalSignatureEx'
        ]:
            path = F'[5]{ignored_stream}'
            summary = [result for result in streams if result.path == path]
            streams = [result for result in streams if result.path != path]
            for result in summary:
                result.path = F'Meta/{ignored_stream}'
                yield result

        inconsistencies = 0
        w1 = len(str(len(strings)))
        w2 = len(str(max(max(strings.computed_ref_count), max(strings.provided_ref_count))))
        for k in range(len(strings)):
            c = strings.computed_ref_count[k]
            p = strings.provided_ref_count[k]
            if c != p and not self.log_debug(F'string {k:0{w1}d} reference count computed={c:0{w2}d} provided={p:0{w2}d}'):
                inconsistencies += 1
        if inconsistencies:
            self.log_info(F'found {inconsistencies} incorrect string reference counts')

        def fix_msi_path(path: str):
            prefix, dot, name = path.partition('.')
            if dot == '.' and prefix in processed_table_data:
                path = F'{prefix}/{name}'
            return path

        if self.args.nocab:
            cabs: list[UnpackResult] = []
        else:
            def _iscab(path):
                return media_info and any(item.get('Cabinet', '') == F'#{path}' for item in media_info)
            media_info = processed_table_data.get('Media', [])
            cabs = [item for item in streams if _iscab(item.path)]
            for cab in cabs:
                self.log_info(F'found cab file: {cab.path}')
        if cabs:
            directories = _install_directories(processed_table_data)
            component_directory: dict[str, str] = {}
            for component_info in processed_table_data.get('Component', []):
                try:
                    component_directory[component_info['Component']] = component_info['Directory']
                except KeyError:
                    continue
            separator = self._get_path_separator()
            file_paths: dict[str, str] = {}

            for file_info in processed_table_data.get('File', []):
                try:
                    src_name = file_info['File']
                    dst_name = file_info['FileName']
                    component = file_info['Component']
                except KeyError:
                    continue
                _, _, long = dst_name.partition('|')
                dst_name = long or dst_name
                directory = component_directory.get(component)
                segments = directories.get(directory, ()) if directory else ()
                file_paths[src_name] = separator.join((*segments, dst_name))

            for cab in cabs:
                path = cab.path
                try:
                    _cabinet = Cabinet(memoryview(cab.get_data()))
                    unpacked = _cabinet.process().get_files()
                except Exception as e:
                    self.log_info(F'unable to extract embedded cab file: {e!s}')
                    continue
                base, dot, ext = path.rpartition('.')
                if dot == '.' and ext.lower() == 'cab':
                    path = base
                else:
                    cab.path = F'{path}.cab'
                for result in unpacked:
                    sub_path = file_paths.get(result.name, result.name)
                    sub_path = separator.join((path, sub_path))
                    streams.append(UnpackResult(sub_path, result.extractor()))

        for item in streams:
            item.path = fix_msi_path(item.path)
        ds = UnpackResult(self._SYNTHETIC_STREAMS_FILENAME, json.dumps(processed_table_data))
        streams.append(ds)

        from refinery.units.formats.csv import json_to_csv

        for key, jd in processed_table_data.items():
            sk = key.strip('_')
            if sk not in processed_table_data:
                key = sk
            try:
                tbl = UnpackResult(F'{self._SYNTHETIC_STREAMS_TOPLEVEL}/{key}.csv', json_to_csv(jd))
            except Exception:
                continue
            streams.append(tbl)

        streams.sort(key=lambda item: item.path)
        yield from streams

    @classmethod
    def _get_msi_from_overlay(cls, data: buf) -> buf:
        if is_likely_pe(data):
            from refinery.units.formats.pe import get_pe_size
            view = memoryview(data)
            overlay = view[get_pe_size(data):]
            if is_likely_msi(overlay):
                return overlay
            if (start := buffer_offset(overlay, _NEGATED_MSI_SIGNATURE, 0, 0x1000)) >= 0:
                if (nulls := buffer_offset(overlay, bytes(8), start, start + 0x1000)) >= 0:
                    from refinery.units.blockwise.neg import neg
                    decoded = overlay[start:nulls] | neg | bytearray
                    decoded.extend(overlay[nulls:])
                    if is_likely_msi(decoded):
                        return decoded
        return data

    @classmethod
    def handles(cls, data):
        if is_likely_msi(data):
            return True
        if is_likely_pe(data):
            from refinery.units.formats.pe import get_pe_size
            view = memoryview(data)
            overlay = view[get_pe_size(data):]
            if is_likely_msi(overlay):
                return True
            if overlay[:8] == _NEGATED_MSI_SIGNATURE:
                return True
        return False

Ancestors

Subclasses

xtmsi

Methods

def unpack(self, data)

Expand source code Browse git

def unpack(self, data: buf):
    document = OleFile(self._get_msi_from_overlay(data))
    streams: list[UnpackResult] = []
    streams_by_entry: dict[DirectoryEntry, UnpackResult] = {}
    read: set[UnpackResult] = set()

    for entry, result in self._streams(document):
        streams.append(result)
        streams_by_entry[entry] = result

    def table_stream(table: str) -> UnpackResult | None:
        if (entry := document.find(msi_table_stream_name(table))) is None:
            return None
        if (result := streams_by_entry.get(entry)) is None or result in read:
            return None
        return result

    def stream(table: str):
        if (result := table_stream(table)) is None:
            raise KeyError(F'table stream not found: {table}')
        read.add(result)
        content = result.get_data()
        result.release()
        return content

    def column_formats(table: dict[str, MSITableColumnInfo]) -> str:
        return ''.join(v.struct_format for v in table.values())

    def stream_to_rows(data: buf, row_format: str):
        row_size = struct.calcsize(F'<{row_format}')
        row_count = int(len(data) / row_size)
        reader = StructReader(data)
        columns = [reader.read_struct(F'<{sc * row_count}') for sc in row_format]
        for i in range(row_count):
            yield [int(c[i]) for c in columns]

    tables: dict[str, dict[str, MSITableColumnInfo]] = collections.defaultdict(collections.OrderedDict)
    strings = MSIStringData(stream('_StringData'), stream('_StringPool'))

    for tbl_name_id, col_number, col_name_id, col_attributes in stream_to_rows(stream('_Columns'), 'HHHH'):
        tbl_name = strings.ref(tbl_name_id)
        col_name = strings.ref(col_name_id)
        tables[tbl_name][col_name] = MSITableColumnInfo(col_number, col_attributes)

    table_names_given = {strings.ref(k) for k in chunks.unpack(stream('_Tables'), 2, False)}
    table_names_known = set(tables)

    for name in table_names_known - table_names_given:
        self.log_warn(F'table name known but not given: {name}')
    for name in table_names_given - table_names_known:
        self.log_warn(F'table name given but not known: {name}')

    processed_table_data: dict[str, list[dict[str, str]]] = {}
    postprocessing: list[_ScriptItem] = []

    for table_name, table in tables.items():
        if table_stream(table_name) is None:
            continue
        processed = []
        info = list(table.values())
        keys = list(table.keys())
        temp = [k.strip('_') for k in keys]
        if len(set(keys)) == len(set(temp)):
            keys = temp
        for r, row in enumerate(stream_to_rows(stream(table_name), column_formats(table))):
            values = []
            for index, value in enumerate(row):
                vt = info[index].type
                if vt is MsiType.Long:
                    if value != 0:
                        value -= 0x80000000
                elif vt is MsiType.Short:
                    if value != 0:
                        value -= 0x8000
                elif value in strings:
                    value = strings.ref(value)
                elif not info[index].is_integer:
                    value = ''
                values.append(value)
            entry = dict(zip(keys, values))
            einfo = {t: i for t, i in zip(keys, info)}
            if table_name == 'MsiFileHash':
                entry['Hash'] = struct.pack(
                    '<IIII',
                    row[2] ^ 0x80000000,
                    row[3] ^ 0x80000000,
                    row[4] ^ 0x80000000,
                    row[5] ^ 0x80000000,
                ).hex()
            if table_name == 'CustomAction':
                code = row[1] & 0x3F
                try:
                    entry['Comment'] = self._CUSTOM_ACTION_TYPES[code]
                except LookupError:
                    pass
                t = einfo.get('Target')
                c = {0x25: 'js', 0x26: 'vbs', 0x33: None}
                if code in c and t and not t.is_integer:
                    postprocessing.append(_ScriptItem(r, c[code]))
            processed.append(entry)
        if processed:
            processed_table_data[table_name] = processed

    streams = [result for result in streams if result not in read]

    if ca := processed_table_data.get('CustomAction'):
        properties, files, components = _formatting_tables(processed_table_data)
        for item in postprocessing:
            entry = ca[item.row_index]
            try:
                action: str = entry['Action']
                target: str = entry['Target']
            except KeyError:
                continue
            root = F'Action/{action}'
            if item.extension:
                action = F'{root}.{item.extension}'
                streams.append(UnpackResult(action, target.encode(self.codec)))
                continue
            target = _format(target, properties, files, components)
            parts = [part.partition('\x02') for part in target.split('\x01')]
            if not all(part[1] == '\x02' for part in parts):
                continue
            for name, _, script in parts:
                if not name.lower().startswith('script'):
                    continue
                if not script:
                    continue
                action = F'{root}.{name}'
                streams.append(UnpackResult(action, script.encode(self.codec)))

    for ignored_stream in [
        'SummaryInformation',
        'DocumentSummaryInformation',
        'DigitalSignature',
        'MsiDigitalSignatureEx'
    ]:
        path = F'[5]{ignored_stream}'
        summary = [result for result in streams if result.path == path]
        streams = [result for result in streams if result.path != path]
        for result in summary:
            result.path = F'Meta/{ignored_stream}'
            yield result

    inconsistencies = 0
    w1 = len(str(len(strings)))
    w2 = len(str(max(max(strings.computed_ref_count), max(strings.provided_ref_count))))
    for k in range(len(strings)):
        c = strings.computed_ref_count[k]
        p = strings.provided_ref_count[k]
        if c != p and not self.log_debug(F'string {k:0{w1}d} reference count computed={c:0{w2}d} provided={p:0{w2}d}'):
            inconsistencies += 1
    if inconsistencies:
        self.log_info(F'found {inconsistencies} incorrect string reference counts')

    def fix_msi_path(path: str):
        prefix, dot, name = path.partition('.')
        if dot == '.' and prefix in processed_table_data:
            path = F'{prefix}/{name}'
        return path

    if self.args.nocab:
        cabs: list[UnpackResult] = []
    else:
        def _iscab(path):
            return media_info and any(item.get('Cabinet', '') == F'#{path}' for item in media_info)
        media_info = processed_table_data.get('Media', [])
        cabs = [item for item in streams if _iscab(item.path)]
        for cab in cabs:
            self.log_info(F'found cab file: {cab.path}')
    if cabs:
        directories = _install_directories(processed_table_data)
        component_directory: dict[str, str] = {}
        for component_info in processed_table_data.get('Component', []):
            try:
                component_directory[component_info['Component']] = component_info['Directory']
            except KeyError:
                continue
        separator = self._get_path_separator()
        file_paths: dict[str, str] = {}

        for file_info in processed_table_data.get('File', []):
            try:
                src_name = file_info['File']
                dst_name = file_info['FileName']
                component = file_info['Component']
            except KeyError:
                continue
            _, _, long = dst_name.partition('|')
            dst_name = long or dst_name
            directory = component_directory.get(component)
            segments = directories.get(directory, ()) if directory else ()
            file_paths[src_name] = separator.join((*segments, dst_name))

        for cab in cabs:
            path = cab.path
            try:
                _cabinet = Cabinet(memoryview(cab.get_data()))
                unpacked = _cabinet.process().get_files()
            except Exception as e:
                self.log_info(F'unable to extract embedded cab file: {e!s}')
                continue
            base, dot, ext = path.rpartition('.')
            if dot == '.' and ext.lower() == 'cab':
                path = base
            else:
                cab.path = F'{path}.cab'
            for result in unpacked:
                sub_path = file_paths.get(result.name, result.name)
                sub_path = separator.join((path, sub_path))
                streams.append(UnpackResult(sub_path, result.extractor()))

    for item in streams:
        item.path = fix_msi_path(item.path)
    ds = UnpackResult(self._SYNTHETIC_STREAMS_FILENAME, json.dumps(processed_table_data))
    streams.append(ds)

    from refinery.units.formats.csv import json_to_csv

    for key, jd in processed_table_data.items():
        sk = key.strip('_')
        if sk not in processed_table_data:
            key = sk
        try:
            tbl = UnpackResult(F'{self._SYNTHETIC_STREAMS_TOPLEVEL}/{key}.csv', json_to_csv(jd))
        except Exception:
            continue
        streams.append(tbl)

    streams.sort(key=lambda item: item.path)
    yield from streams

Inherited members

xtdoc:
- CustomJoinBehaviour
- CustomPathSeparator
- FilterEverything
- Requires
- act
- assemble
- carved
- close
- codec
- console
- filter
- finish
- handles
- is_quiet
- is_reversible
- isatty
- labelled
- leniency
- log_always
- log_debug
- log_detach
- log_fail
- log_info
- log_level
- log_warn
- logger
- name
- nozzle
- optional_dependencies
- process
- read
- read1
- required_dependencies
- reset
- reverse
- run
- source
- superinit