gecko-dev/python/eme/gen-eme-voucher.py

634 строки
20 KiB
Python

#!/usr/bin/env python2.7
#
# Copyright 2014 Adobe Systems Incorporated. All Rights Reserved.
#
# Adobe permits you to use, modify, and distribute this file in accordance
# with the terms of the Mozilla Public License, v 2.0 accompanying it. If
# a copy of the MPL was not distributed with this file, You can obtain one
# at http://mozilla.org/MPL/2.0/.
#
# Creates an Adobe Access signed voucher for x32/x64 windows executables
# Notes: This is currently python2.7 due to mozilla build system requirements
from __future__ import print_function
import argparse, bitstring, pprint, hashlib, os, subprocess, sys, tempfile, macholib, macholib.MachO
from pyasn1.codec.der import encoder as der_encoder
from pyasn1.type import univ, namedtype, namedval, constraint
# Defined in WinNT.h from the Windows SDK
IMAGE_SCN_MEM_EXECUTE = 0x20000000
IMAGE_REL_BASED_HIGHLOW = 3
IMAGE_REL_BASED_DIR64 = 10
# CodeSectionDigest ::= SEQUENCE {
# offset INTEGER -- section's file offset in the signed binary
# digestAlgorithm OBJECT IDENTIFIER -- algorithm identifier for the hash value below. For now only supports SHA256.
# digestValue OCTET STRING -- hash value of the TEXT segment.
# }
class CodeSectionDigest(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('offset', univ.Integer()),
namedtype.NamedType('digestAlgorithm', univ.ObjectIdentifier()),
namedtype.NamedType('digest', univ.OctetString()))
# CodeSegmentDigest ::= SEQUENCE {
# offset INTEGER -- TEXT segment's file offset in the signed binary
# codeSectionDigests SET OF CodeSectionDigests
# }
class SetOfCodeSectionDigest(univ.SetOf):
componentType = CodeSectionDigest()
class CodeSegmentDigest(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('offset', univ.Integer()),
namedtype.NamedType('codeSectionDigests', SetOfCodeSectionDigest()))
# ArchitectureDigest ::= SEQUENCE {
# cpuType ENUMERATED CpuType
# cpuSubType ENUMERATED CpuSubType
# CodeSegmentDigests SET OF CodeSegmentDigests
# }
class SetOfCodeSegmentDigest(univ.SetOf):
componentType = CodeSegmentDigest()
class CPUType(univ.Enumerated):
namedValues = namedval.NamedValues(
('IMAGE_FILE_MACHINE_I386', 0x14c),
('IMAGE_FILE_MACHINE_AMD64',0x8664 ),
('MACHO_CPU_TYPE_I386',0x7 ),
('MACHO_CPU_TYPE_X86_64',0x1000007 ),
)
subtypeSpec = univ.Enumerated.subtypeSpec + \
constraint.SingleValueConstraint(0x14c, 0x8664, 0x7, 0x1000007)
class CPUSubType(univ.Enumerated):
namedValues = namedval.NamedValues(
('IMAGE_UNUSED', 0x0),
('CPU_SUBTYPE_X86_ALL', 0x3),
('CPU_SUBTYPE_X86_64_ALL', 0x80000003)
)
subtypeSpec = univ.Enumerated.subtypeSpec + \
constraint.SingleValueConstraint(0, 0x3, 0x80000003)
class ArchitectureDigest(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('cpuType', CPUType()),
namedtype.NamedType('cpuSubType', CPUSubType()),
namedtype.NamedType('CodeSegmentDigests', SetOfCodeSegmentDigest())
)
# ApplicationDigest ::= SEQUENCE {
# version INTEGER
# digests SET OF ArchitectureDigest
# }
class SetOfArchitectureDigest(univ.SetOf):
componentType = ArchitectureDigest()
class ApplicationDigest(univ.Sequence):
componentType = namedtype.NamedTypes(
namedtype.NamedType('version', univ.Integer()),
namedtype.NamedType('digests', SetOfArchitectureDigest())
)
def meets_requirements(items, requirements):
for r in requirements:
for n, v in r.items():
if n not in items or items[n] != v: return False
return True
# return total number of bytes read from items_in excluding leaves
# TODO: research replacing this with the python built-in struct module
def parse_items(stream, items_in, items_out):
bits_read = 0
total_bits_read = 0
for item in items_in:
name = item[0]
t = item[1]
bits = 1 if ":" not in t else int(t[t.index(":") + 1:])
if ":" in t and t.find("bytes") >= 0:
bits = bits * 8
if len(item) == 2:
items_out[name] = stream.read(t)
bits_read += bits
total_bits_read += bits
elif len(item) == 3 or len(item) == 4:
requirements = list(filter(lambda x: isinstance(x, dict), item[2]))
sub_items = list(filter(lambda x: isinstance(x, tuple), item[2]))
if not meets_requirements(items_out, requirements): continue
# has sub-items based on length
items_out[name] = stream.read(t)
bits_read += bits
total_bits_read += bits
if len(item) == 4:
bit_length = items_out[name] * 8
if bit_length > 0:
sub_read, sub_total_read = parse_items(stream, sub_items, items_out)
bit_length -= sub_read
total_bits_read += sub_total_read
if bit_length > 0:
items_out[item[3]] = stream.read('bits:' + str(bit_length))
bits_read += bit_length
total_bits_read += bit_length
else:
raise Exception("unrecognized item" + pprint.pformat(item))
return bits_read, total_bits_read
# macho stuff
# Constant for the magic field of the mach_header (32-bit architectures)
MH_MAGIC =0xfeedface # the mach magic number
MH_CIGAM =0xcefaedfe # NXSwapInt(MH_MAGIC)
MH_MAGIC_64 =0xfeedfacf # the the 64-bit mach magic number
MH_CIGAM_64 =0xcffaedfe # NXSwapInt(MH_MAGIC_64)
FAT_CIGAM = 0xbebafeca
FAT_MAGIC = 0xcafebabe
LC_SEGMENT = 0x1
LC_SEGMENT_64 = 0x19 # 64-bit segment of this file to be
# TODO: perhaps switch to pefile module when it officially supports python3
class SectionHeader:
def __init__(self, stream):
items = [
('Name', 'bytes:8'),
('VirtualSize', 'uintle:32'),
('VirtualAddress', 'uintle:32'),
('SizeOfRawData', 'uintle:32'),
('PointerToRawData', 'uintle:32'),
('PointerToRelocations', 'uintle:32'),
('PointerToLineNumber', 'uintle:32'),
('NumberOfRelocations', 'uintle:16'),
('NumberOfLineNumbers', 'uintle:16'),
('Characteristics', 'uintle:32')
]
self.items = dict()
self.relocs = dict()
_, self.bits_read = parse_items(stream, items, self.items)
self.sectionName = self.items['Name'].decode('utf-8')
self.offset = self.items['PointerToRawData']
COFF_DATA_DIRECTORY_TYPES = [
"Export Table",
"Import Table",
"Resource Table",
"Exception Table",
"Certificate Tble",
"Base Relocation Table",
"Debug",
"Architecture",
"Global Ptr",
"TLS Table",
"Load Config Table",
"Bound Import",
"IAT",
"Delay Import Descriptor",
"CLR Runtime Header",
"Reserved",
]
def chained_safe_get(obj, names, default=None):
if obj is None: return default
for n in names:
if n in obj:
obj = obj[n]
else:
return default
return obj
class OptionalHeader:
def __init__(self, stream, size):
self.items = {}
items = []
if size:
items += [
('Magic', 'uintle:16'),
('MajorLinkerVersion', 'uintle:8'),
('MinorLinkerVersion', 'uintle:8'),
('SizeOfCode', 'uintle:32'),
('SizeOfInitializedData', 'uintle:32'),
('SizeOfUninitializedData', 'uintle:32'),
('AddressOfEntryPoint', 'uintle:32'),
('BaseOfCode', 'uintle:32'),
]
_, self.bits_read = parse_items(stream, items, self.items)
items = []
if self.items['Magic'] == 0x10b: # PE32
items += [('BaseOfData', 'uintle:32')]
address_size = 'uintle:64' if self.items['Magic'] == 0x20b else 'uintle:32'
items += [
('ImageBase', address_size),
('SectionAlignment', 'uintle:32'),
('FileAlignment', 'uintle:32'),
('MajorOperatingSystemVersion', 'uintle:16'),
('MinorOperatingSystemVersion', 'uintle:16'),
('MajorImageVersion', 'uintle:16'),
('MinorImageVersion', 'uintle:16'),
('MajorSubsystemVersion', 'uintle:16'),
('MinorSubsystemVersion', 'uintle:16'),
('Win32VersionValue', 'uintle:32'),
('SizeOfImage', 'uintle:32'),
('SizeOfHeaders', 'uintle:32'),
('CheckSum', 'uintle:32'),
('Subsystem', 'uintle:16'),
('DllCharacteristics', 'uintle:16'),
('SizeOfStackReserve', address_size),
('SizeOfStackCommit', address_size),
('SizeOfHeapReserve', address_size),
('SizeOfHeapCommit', address_size),
('LoaderFlags', 'uintle:32'),
('NumberOfRvaAndSizes', 'uintle:32'),
]
if size > 28:
_, bits_read = parse_items(stream, items, self.items)
self.bits_read += bits_read
if 'NumberOfRvaAndSizes' in self.items:
index = 0
self.items['Data Directories'] = dict()
while self.bits_read / 8 < size:
d = self.items['Data Directories'][COFF_DATA_DIRECTORY_TYPES[index]] = dict()
_, bits_read = parse_items(stream, [('VirtualAddress', 'uintle:32'), ('Size', 'uintle:32')], d)
self.bits_read += bits_read
index += 1
class COFFFileHeader:
def __init__(self, stream):
self.items = {}
self.section_headers = []
items = [
('Machine', 'uintle:16'),
('NumberOfSections', 'uintle:16'),
('TimeDateStamp', 'uintle:32'),
('PointerToSymbolTable', 'uintle:32'),
('NumberOfSymbols', 'uintle:32'),
('SizeOfOptionalHeader', 'uintle:16'),
('Characteristics', 'uintle:16')
]
_, self.bits_read = parse_items(stream, items, self.items)
self.OptionalHeader = OptionalHeader(stream, self.items['SizeOfOptionalHeader'])
self.bits_read += self.OptionalHeader.bits_read
# start reading section headers
num_sections = self.items['NumberOfSections']
while num_sections > 0 :
section_header = SectionHeader(stream)
self.bits_read += section_header.bits_read
self.section_headers.append(section_header)
num_sections -= 1
self.section_headers.sort(key=lambda header: header.offset)
# Read Relocations
self.process_relocs(stream)
def process_relocs(self, stream):
reloc_table = chained_safe_get(self.OptionalHeader.items, ['Data Directories', 'Base Relocation Table'])
if reloc_table is None: return
orig_pos = stream.bitpos
_, stream.bytepos = self.get_rva_section(reloc_table['VirtualAddress'])
end_pos = stream.bitpos + reloc_table['Size'] * 8
while stream.bitpos < end_pos:
page_rva = stream.read('uintle:32')
block_size = stream.read('uintle:32')
for i in range(0, int((block_size - 8) / 2)):
data = stream.read('uintle:16')
typ = data >> 12
offset = data & 0xFFF
if offset == 0 and i > 0: continue
assert(typ == IMAGE_REL_BASED_HIGHLOW or typ == IMAGE_REL_BASED_DIR64)
cur_pos = stream.bitpos
sh, value_bytepos = self.get_rva_section(page_rva + offset)
stream.bytepos = value_bytepos
value = stream.read('uintle:32' if typ == IMAGE_REL_BASED_HIGHLOW else 'uintle:64')
# remove BaseAddress
value -= self.OptionalHeader.items['ImageBase']
bit_size = (4 if typ == IMAGE_REL_BASED_HIGHLOW else 8) * 8
stream.overwrite(bitstring.BitArray(uint=value, length=bit_size), pos=value_bytepos * 8)
stream.pos = cur_pos
stream.bitpos = orig_pos
def get_rva_section(self, rva):
for sh in self.section_headers:
if rva < sh.items['VirtualAddress'] or rva >= sh.items['VirtualAddress'] + sh.items['VirtualSize']:
continue
file_pointer = rva - sh.items['VirtualAddress'] + sh.items['PointerToRawData']
return sh, file_pointer
raise Exception('Could not match RVA to section')
def create_temp_file(suffix=""):
fd, path = tempfile.mkstemp(suffix=suffix)
os.close(fd)
return path
class ExpandPath(argparse.Action):
def __call__(self, parser, namespace, values, option_string=None):
setattr(namespace, self.dest, os.path.abspath(os.path.expanduser(values)))
# this does a naming trick since windows doesn't allow multiple usernames for the same server
def get_password(service_name, user_name):
try:
import keyring
# windows doesn't allow multiple usernames for the same server, argh
if sys.platform == "win32":
password = keyring.get_password(service_name + "-" + user_name, user_name)
else:
password = keyring.get_password(service_name, user_name)
return password
except:
# This allows for manual testing where you do not wish to cache the password on the system
print("Missing keyring module...getting password manually")
return None
def openssl_cmd(app_args, args, password_in, password_out):
password = get_password(app_args.password_service, app_args.password_user) if (password_in or password_out) else None
env = None
args = [app_args.openssl_path] + args
if password is not None:
env = os.environ.copy()
env["COFF_PW"] = password
if password_in: args += ["-passin", "env:COFF_PW"]
if password_out: args += ["-passout", "env:COFF_PW", "-password", "env:COFF_PW"]
subprocess.check_call(args, env=env)
def processMachoBinary(filename):
outDict = dict()
outDict['result'] = False
setOfArchDigests = SetOfArchitectureDigest()
archDigestIdx = 0
parsedMacho = macholib.MachO.MachO(filename)
for header in parsedMacho.headers :
arch_digest = ArchitectureDigest()
lc_segment = LC_SEGMENT
arch_digest.setComponentByName('cpuType', CPUType(header.header.cputype))
arch_digest.setComponentByName('cpuSubType', CPUSubType(header.header.cpusubtype))
if header.header.cputype == 0x1000007:
lc_segment = LC_SEGMENT_64
segment_commands = list(filter(lambda x: x[0].cmd == lc_segment, header.commands))
text_segment_commands = list(filter(lambda x: x[1].segname.decode("utf-8").startswith("__TEXT"), segment_commands))
code_segment_digests = SetOfCodeSegmentDigest()
code_segment_idx = 0
for text_command in text_segment_commands:
codeSegmentDigest = CodeSegmentDigest()
codeSegmentDigest.setComponentByName('offset', text_command[1].fileoff)
sectionDigestIdx = 0
set_of_digest = SetOfCodeSectionDigest()
for section in text_command[2]:
digester = hashlib.sha256()
digester.update(section.section_data)
digest = digester.digest()
code_section_digest = CodeSectionDigest()
code_section_digest.setComponentByName('offset', section.offset)
code_section_digest.setComponentByName('digestAlgorithm', univ.ObjectIdentifier('2.16.840.1.101.3.4.2.1'))
code_section_digest.setComponentByName('digest', univ.OctetString(digest))
set_of_digest.setComponentByPosition(sectionDigestIdx, code_section_digest)
sectionDigestIdx += 1
codeSegmentDigest.setComponentByName('codeSectionDigests', set_of_digest)
code_segment_digests.setComponentByPosition(code_segment_idx, codeSegmentDigest)
code_segment_idx += 1
arch_digest.setComponentByName('CodeSegmentDigests', code_segment_digests)
setOfArchDigests.setComponentByPosition(archDigestIdx, arch_digest)
archDigestIdx += 1
outDict['result'] = True
if outDict['result']:
appDigest = ApplicationDigest()
appDigest.setComponentByName('version', 1)
appDigest.setComponentByName('digests', setOfArchDigests)
outDict['digest'] = appDigest
return outDict
def processCOFFBinary(stream):
outDict = dict()
outDict['result'] = False
# find the COFF header.
# skip forward past the MSDOS stub header to 0x3c.
stream.bytepos = 0x3c
# read 4 bytes, this is the file offset of the PE signature.
pe_sig_offset = stream.read('uintle:32')
stream.bytepos = pe_sig_offset
# read 4 bytes, make sure it's a PE signature.
signature = stream.read('uintle:32')
if signature != 0x00004550:
return outDict
# after signature is the actual COFF file header.
coff_header = COFFFileHeader(stream)
arch_digest = ArchitectureDigest()
if coff_header.items['Machine'] == 0x14c:
arch_digest.setComponentByName('cpuType', CPUType('IMAGE_FILE_MACHINE_I386'))
elif coff_header.items['Machine'] == 0x8664:
arch_digest.setComponentByName('cpuType', CPUType('IMAGE_FILE_MACHINE_AMD64'))
arch_digest.setComponentByName('cpuSubType', CPUSubType('IMAGE_UNUSED'))
text_section_headers = list(filter(lambda x: (x.items['Characteristics'] & IMAGE_SCN_MEM_EXECUTE) == IMAGE_SCN_MEM_EXECUTE, coff_header.section_headers))
code_segment_digests = SetOfCodeSegmentDigest()
code_segment_idx = 0
for code_sect_header in text_section_headers:
stream.bytepos = code_sect_header.offset
code_sect_bytes = stream.read('bytes:' + str(code_sect_header.items['VirtualSize']))
digester = hashlib.sha256()
digester.update(code_sect_bytes)
digest = digester.digest()
# with open('segment_' + str(code_sect_header.offset) + ".bin", 'wb') as f:
# f.write(code_sect_bytes)
code_section_digest = CodeSectionDigest()
code_section_digest.setComponentByName('offset', code_sect_header.offset)
code_section_digest.setComponentByName('digestAlgorithm', univ.ObjectIdentifier('2.16.840.1.101.3.4.2.1'))
code_section_digest.setComponentByName('digest', univ.OctetString(digest))
set_of_digest = SetOfCodeSectionDigest()
set_of_digest.setComponentByPosition(0, code_section_digest)
codeSegmentDigest = CodeSegmentDigest()
codeSegmentDigest.setComponentByName('offset', code_sect_header.offset)
codeSegmentDigest.setComponentByName('codeSectionDigests', set_of_digest)
code_segment_digests.setComponentByPosition(code_segment_idx, codeSegmentDigest)
code_segment_idx += 1
arch_digest.setComponentByName('CodeSegmentDigests', code_segment_digests)
setOfArchDigests = SetOfArchitectureDigest()
setOfArchDigests.setComponentByPosition(0, arch_digest)
appDigest = ApplicationDigest()
appDigest.setComponentByName('version', 1)
appDigest.setComponentByName('digests', setOfArchDigests)
outDict['result'] = True
outDict['digest'] = appDigest
return outDict
def main():
parser = argparse.ArgumentParser(description='PE/COFF Signer')
parser.add_argument('-input', action=ExpandPath, required=True, help="File to parse.")
parser.add_argument('-output', action=ExpandPath, required=True, help="File to write to.")
parser.add_argument('-openssl_path', action=ExpandPath, help="Path to OpenSSL to create signed voucher")
parser.add_argument('-signer_pfx', action=ExpandPath, help="Path to certificate to use to sign voucher. Must contain full certificate chain.")
parser.add_argument('-password_service', help="Name of Keyring/Wallet service/host")
parser.add_argument('-password_user', help="Name of Keyring/Wallet user name")
parser.add_argument('-verbose', action='store_true', help="Verbose output.")
app_args = parser.parse_args()
# to simplify relocation handling we use a mutable BitStream so we can remove
# the BaseAddress from each relocation
stream = bitstring.BitStream(filename=app_args.input)
dict = processCOFFBinary(stream)
if dict['result'] == False:
dict = processMachoBinary(app_args.input)
if dict['result'] == False:
raise Exception("Invalid File")
binaryDigest = der_encoder.encode(dict['digest'])
with open(app_args.output, 'wb') as f:
f.write(binaryDigest)
# sign with openssl if specified
if app_args.openssl_path is not None:
assert app_args.signer_pfx is not None
out_base, out_ext = os.path.splitext(app_args.output)
signed_path = out_base + ".signed" + out_ext
# http://stackoverflow.com/questions/12507277/how-to-fix-unable-to-write-random-state-in-openssl
temp_files = []
if sys.platform == "win32" and "RANDFILE" not in os.environ:
temp_file = create_temp_file()
temp_files += [temp_file]
os.environ["RANDFILE"] = temp_file
try:
# create PEM from PFX
pfx_pem_path = create_temp_file(".pem")
temp_files += [pfx_pem_path]
print("Extracting PEM from PFX to:" + pfx_pem_path)
openssl_cmd(app_args, ["pkcs12", "-in", app_args.signer_pfx, "-out", pfx_pem_path], True, True)
# extract CA certs
pfx_cert_path = create_temp_file(".cert")
temp_files += [pfx_cert_path]
print("Extracting cert from PFX to:" + pfx_cert_path)
openssl_cmd(app_args, ["pkcs12", "-in", app_args.signer_pfx, "-cacerts", "-nokeys", "-out", pfx_cert_path], True, False)
# we embed the public keychain for client validation
openssl_cmd(app_args, ["cms", "-sign", "-nodetach", "-md", "sha256", "-binary", "-in", app_args.output, "-outform", "der", "-out", signed_path, "-signer", pfx_pem_path, "-certfile", pfx_cert_path], True, False)
finally:
for t in temp_files:
if "RANDFILE" in os.environ and t == os.environ["RANDFILE"]:
del os.environ["RANDFILE"]
os.unlink(t)
if __name__ == '__main__':
main()