""" 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)