"""
RPM archive parser.
Author: Victor Stinner, 1st December 2005.
"""
from hachoir_parser import Parser
from hachoir_core.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
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 xrange(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)
def sortRpmItem(a,b):
return int( a["offset"].value - b["offset"].value )
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 != "\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( sortRpmItem )
# 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": (u"application/x-rpm",),
"min_size": (96 + 16 + 16)*8, # file header + checksum + content header
"magic": (('\xED\xAB\xEE\xDB', 0),),
"description": "RPM package"
}
TYPE_NAME = {
0: "Binary",
1: "Source"
}
endian = BIG_ENDIAN
def validate(self):
if self["signature"].value != '\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:
yield SubFile(self, "gz_content", size, "gzip content", parser=GzipParser)
syntax highlighted by Code2HTML, v. 0.9.1