SickGear/lib/hachoir/parser/archive/rpm.py

"""
RPM archive parser.

Author: Victor Stinner, 1st December 2005.
"""

from hachoir.parser import Parser
from hachoir.field import (FieldSet, ParserError,
                               UInt8, UInt16, UInt32, UInt64, Enum,
                               NullBytes, Bytes, RawBytes, SubFile,
                               Character, CString, String)
from hachoir.core.endian import BIG_ENDIAN
from hachoir.parser.archive.gzip_parser import GzipParser
from hachoir.parser.archive.bzip2_parser import Bzip2Parser


class ItemContent(FieldSet):
    format_type = {
        0: UInt8,
        1: Character,
        2: UInt8,
        3: UInt16,
        4: UInt32,
        5: UInt64,
        6: CString,
        7: RawBytes,
        8: CString,
        9: CString
    }

    def __init__(self, parent, name, item):
        FieldSet.__init__(self, parent, name, item.description)
        self.related_item = item
        self._name = "content_%s" % item.name

    def createFields(self):
        item = self.related_item
        type = item["type"].value

        cls = self.format_type[type]
        count = item["count"].value
        if cls is RawBytes:  # or type == 8:
            if cls is RawBytes:
                args = (self, "value", count)
            else:
                args = (self, "value")  # cls is CString
            count = 1
        else:
            if 1 < count:
                args = (self, "value[]")
            else:
                args = (self, "value")
        for index in range(count):
            yield cls(*args)


class Item(FieldSet):
    type_name = {
        0: "NULL",
        1: "CHAR",
        2: "INT8",
        3: "INT16",
        4: "INT32",
        5: "INT64",
        6: "CSTRING",
        7: "BIN",
        8: "CSTRING_ARRAY",
        9: "CSTRING?"
    }
    tag_name = {
        1000: "File size",
        1001: "(Broken) MD5 signature",
        1002: "PGP 2.6.3 signature",
        1003: "(Broken) MD5 signature",
        1004: "MD5 signature",
        1005: "GnuPG signature",
        1006: "PGP5 signature",
        1007: "Uncompressed payload size (bytes)",
        256 + 8: "Broken SHA1 header digest",
        256 + 9: "Broken SHA1 header digest",
        256 + 13: "Broken SHA1 header digest",
        256 + 11: "DSA header signature",
        256 + 12: "RSA header signature"
    }

    def __init__(self, parent, name, description=None, tag_name_dict=None):
        FieldSet.__init__(self, parent, name, description)
        if tag_name_dict is None:
            tag_name_dict = Item.tag_name
        self.tag_name_dict = tag_name_dict

    def createFields(self):
        yield Enum(UInt32(self, "tag", "Tag"), self.tag_name_dict)
        yield Enum(UInt32(self, "type", "Type"), Item.type_name)
        yield UInt32(self, "offset", "Offset")
        yield UInt32(self, "count", "Count")

    def createDescription(self):
        return "Item: %s (%s)" % (self["tag"].display, self["type"].display)


class ItemHeader(Item):
    tag_name = {
        61: "Current image",
        62: "Signatures",
        63: "Immutable",
        64: "Regions",
        100: "I18N string locales",
        1000: "Name",
        1001: "Version",
        1002: "Release",
        1003: "Epoch",
        1004: "Summary",
        1005: "Description",
        1006: "Build time",
        1007: "Build host",
        1008: "Install time",
        1009: "Size",
        1010: "Distribution",
        1011: "Vendor",
        1012: "Gif",
        1013: "Xpm",
        1014: "Licence",
        1015: "Packager",
        1016: "Group",
        1017: "Changelog",
        1018: "Source",
        1019: "Patch",
        1020: "Url",
        1021: "OS",
        1022: "Arch",
        1023: "Prein",
        1024: "Postin",
        1025: "Preun",
        1026: "Postun",
        1027: "Old filenames",
        1028: "File sizes",
        1029: "File states",
        1030: "File modes",
        1031: "File uids",
        1032: "File gids",
        1033: "File rdevs",
        1034: "File mtimes",
        1035: "File MD5s",
        1036: "File link to's",
        1037: "File flags",
        1038: "Root",
        1039: "File username",
        1040: "File groupname",
        1043: "Icon",
        1044: "Source rpm",
        1045: "File verify flags",
        1046: "Archive size",
        1047: "Provide name",
        1048: "Require flags",
        1049: "Require name",
        1050: "Require version",
        1051: "No source",
        1052: "No patch",
        1053: "Conflict flags",
        1054: "Conflict name",
        1055: "Conflict version",
        1056: "Default prefix",
        1057: "Build root",
        1058: "Install prefix",
        1059: "Exclude arch",
        1060: "Exclude OS",
        1061: "Exclusive arch",
        1062: "Exclusive OS",
        1064: "RPM version",
        1065: "Trigger scripts",
        1066: "Trigger name",
        1067: "Trigger version",
        1068: "Trigger flags",
        1069: "Trigger index",
        1079: "Verify script",
        # TODO: Finish the list (id 1070..1162 using rpm library source code)
    }

    def __init__(self, parent, name, description=None):
        Item.__init__(self, parent, name, description, self.tag_name)


class PropertySet(FieldSet):

    def __init__(self, parent, name, *args):
        FieldSet.__init__(self, parent, name, *args)
        self._size = self["content_item[1]"].address + self["size"].value * 8

    def createFields(self):
        # Read chunk header
        yield Bytes(self, "signature", 3, r"Property signature (\x8E\xAD\xE8)")
        if self["signature"].value != b"\x8E\xAD\xE8":
            raise ParserError("Invalid property signature")
        yield UInt8(self, "version", "Signature version")
        yield NullBytes(self, "reserved", 4, "Reserved")
        yield UInt32(self, "count", "Count")
        yield UInt32(self, "size", "Size")

        # Read item header
        items = []
        for i in range(0, self["count"].value):
            item = ItemHeader(self, "item[]")
            yield item
            items.append(item)

        # Sort items by their offset
        items.sort(key=lambda field: field["offset"].value)

        # Read item content
        start = self.current_size // 8
        for item in items:
            offset = item["offset"].value
            diff = offset - (self.current_size // 8 - start)
            if 0 < diff:
                yield NullBytes(self, "padding[]", diff)
            yield ItemContent(self, "content[]", item)
        size = start + self["size"].value - self.current_size // 8
        if 0 < size:
            yield NullBytes(self, "padding[]", size)


class RpmFile(Parser):
    PARSER_TAGS = {
        "id": "rpm",
        "category": "archive",
        "file_ext": ("rpm",),
        "mime": ("application/x-rpm",),
        # file header + checksum + content header
        "min_size": (96 + 16 + 16) * 8,
        "magic": ((b'\xED\xAB\xEE\xDB', 0),),
        "description": "RPM package"
    }
    TYPE_NAME = {
        0: "Binary",
        1: "Source"
    }
    endian = BIG_ENDIAN

    def validate(self):
        if self["signature"].value != b'\xED\xAB\xEE\xDB':
            return "Invalid signature"
        if self["major_ver"].value != 3:
            return "Unknown major version (%u)" % self["major_ver"].value
        if self["type"].value not in self.TYPE_NAME:
            return "Invalid RPM type"
        return True

    def createFields(self):
        yield Bytes(self, "signature", 4, r"RPM file signature (\xED\xAB\xEE\xDB)")
        yield UInt8(self, "major_ver", "Major version")
        yield UInt8(self, "minor_ver", "Minor version")
        yield Enum(UInt16(self, "type", "RPM type"), RpmFile.TYPE_NAME)
        yield UInt16(self, "architecture", "Architecture")
        yield String(self, "name", 66, "Archive name", strip="\0", charset="ASCII")
        yield UInt16(self, "os", "OS")
        yield UInt16(self, "signature_type", "Type of signature")
        yield NullBytes(self, "reserved", 16, "Reserved")
        yield PropertySet(self, "checksum", "Checksum (signature)")
        yield PropertySet(self, "header", "Header")

        if self._size is None:  # TODO: is it possible to handle piped input?
            raise NotImplementedError

        size = (self._size - self.current_size) // 8
        if size:
            if 3 <= size and self.stream.readBytes(self.current_size, 3) == b"BZh":
                yield SubFile(self, "content", size, "bzip2 content", parser=Bzip2Parser)
            else:
                yield SubFile(self, "content", size, "gzip content", parser=GzipParser)