path: root/python/eme/gen-eme-voucher.py

#!/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()