Validate cryptographic signature on macos macho binary

validate_macho_sig.py

#!/usr/bin/env python3
import io
import hashlib
import os
import struct
import sys
import pprint

import macholib.MachO
from macholib.mach_o import LC_CODE_SIGNATURE
import asn1crypto.x509
from asn1crypto.cms import ContentInfo, SignedData, CMSAttributes
from oscrypto import asymmetric
from certvalidator.context import ValidationContext
import certvalidator

# Apple root certificate
APPLE_ROOT = b'0\x82\x04\x040\x82\x02\xec\xa0\x03\x02\x01\x02\x02\x08\x18z\xa9\xa8\xc2\x96!\x0c0\r\x06\t*\x86H\x86\xf7\r\x01\x01\x0b\x05\x000b1\x0b0\t\x06\x03U\x04\x06\x13\x02US1\x130\x11\x06\x03U\x04\n\x13\nApple Inc.1&0$\x06\x03U\x04\x0b\x13\x1dApple Certification Authority1\x160\x14\x06\x03U\x04\x03\x13\rApple Root CA0\x1e\x17\r120201221215Z\x17\r270201221215Z0y1-0+\x06\x03U\x04\x03\x0c$Developer ID Certification Authority1&0$\x06\x03U\x04\x0b\x0c\x1dApple Certification Authority1\x130\x11\x06\x03U\x04\n\x0c\nApple Inc.1\x0b0\t\x06\x03U\x04\x06\x13\x02US0\x82\x01"0\r\x06\t*\x86H\x86\xf7\r\x01\x01\x01\x05\x00\x03\x82\x01\x0f\x000\x82\x01\n\x02\x82\x01\x01\x00\x89vO\x06[\x9aA\xee\xa5#+\x02\xa3_\xd7s?\xc05\xb0\x8b\x84\n?\x06$\x7f\xa7\x95?\xebO\x0e\x93\xaf\xb4\x0e\xd0\xc8>\xe5m\x18\xb3\x1f\xe8\x89G\xbf\xd7\t\x08\xe4\xffV\x98)\x15\xe7\x94\x9d\xb95\xa3\n\xcd\xb4\xc0\xe1\xe2`\xf4\xca\xec)xEii`k_\x8a\x92\xfc\x9e#\xe6:\xc2"\xb31O\x1c\xba\xf2\xb64YB\xee\xb0\xa9\x02\x03\x18\x91\x04\xb6\xb3x.3\x1f\x80E\rEo\xbb\x0eZ[\x7f:\xe7\xd8\x08\xd7\x0b\x0e2m\xfb\x866\xe4l\xab\xc4\x11\x8ap\x84&\xaa\x9fD\xd1\xf1\xb8\xc6{\x94\x17\x9bH\xf7\x0bX\x16\xba#\xc5\x9f\x159~\xca]\xc32_\x0f\xe0R\x7f@\xea\xbe\xac\x08d\x95[\xc9\x1a\x9c\xe5\x80\xca\x1fjD\x1cl>\xc4\xb0&\x1f\x1d\xec{\xaf^\xa0j=G\xa9X\x121? v(m\x1d\x1c\xb0\xc2N\x11i&\x8b\xcb\xd6\xd0\x11\x82\xc9N\x0f\xf1Vt\xd0\xd9\x08Kfx\xa2\xab\xac\xa7\xe2\xd2L\x87Y\xc9\x02\x03\x01\x00\x01\xa3\x81\xa60\x81\xa30\x1d\x06\x03U\x1d\x0e\x04\x16\x04\x14W\x17\xed\xa2\xcf\xdc|\x98\xa1\x10\xe0\xfc\xbe\x87-,\xf2\xe3\x17T0\x0f\x06\x03U\x1d\x13\x01\x01\xff\x04\x050\x03\x01\x01\xff0\x1f\x06\x03U\x1d#\x04\x180\x16\x80\x14+\xd0iG\x94v\t\xfe\xf4k\x8d.@\xa6\xf7GM\x7f\x08^0.\x06\x03U\x1d\x1f\x04\'0%0#\xa0!\xa0\x1f\x86\x1dhttp://crl.apple.com/root.crl0\x0e\x06\x03U\x1d\x0f\x01\x01\xff\x04\x04\x03\x02\x01\x860\x10\x06\n*\x86H\x86\xf7cd\x06\x02\x06\x04\x02\x05\x000\r\x06\t*\x86H\x86\xf7\r\x01\x01\x0b\x05\x00\x03\x82\x01\x01\x00B9tk\xa1\xdc\xc6\xa4\x8f7*\x8c\xb3\x1d\nD\xbc\x95,\x7f\xbcY\xb8\xaca\xfb\x07\x90\x922\xb9\xd4\xbf;\xc1P9jDt\xa2\xec[\x1fp\xe5\xaa\xddKl\x1c#q-_\xd1\xc5\x93\xbe\xee\x9b\x8ape\x82\x9d\x16\xe3\x1a\x10\x17\x89-\xa8\xcd\xfd\x0cxXI\x0c(\x7f3\xee\x00z\x1b\xb4v\xac\xb6\xb5\xbbO\xdf\xa8\x1b\x9d\xc8\x19\x97J\x0bVg/\xc2>\xb6\xb3\xc4\x83:\xf0wmt\xc4.#Q\xee\x9a\xa5\x03o`\xf4\xa5H\xa7\x06\xc2\xbbZ\xe2\x1f\x1fFE~\xe4\x97\xf5\'\x10\xb7 "ror\xda\xc6Pu\xc5=%\x8f]\xa3\x00\xe9\x9f6\x8cH9\x8f\xb3;\xea\x90\x80.\x95\x9a`\xf4x\xce\xf4\x0e\nS>\xa2\xfaO\xd8\x1e\xae\x84\x95\x8d2\xbcVM\x89\xe9x\x18\xe0\xac\x9aB\xbazF\x1b\x84\xa2\x89\xce\x14\xe8\x88\xd1X\x8b\xf6\xaeV\xc4,\x05*E\xaf\x0b\xd9K\xa9\x02\x0f4\xac\x88\xc7aU\x89D\xc9\'s\x07\xee\x82\xe5N\xf5p'

# SuperBlob slot IDs
cdInfoSlot = 1                                 # Info.plist
cdRequirementsSlot = 2                         # internal requirements
cdResourceDirSlot = 3                          # resource directory
cdTopDirectorySlot = 4                         # Application specific slot
cdEntitlementSlot = 5                          # embedded entitlement configuration
cdRepSpecificSlot = 6                          # for use by disk rep
cdEntitlementDERSlot = 7                       # DER representation of entitlements
cdCodeDirectorySlot = 0                        # CodeDirectory
cdAlternateCodeDirectorySlots = 0x1000         # alternate CodeDirectory array
cdAlternateCodeDirectoryLimit = 0x1005         # 5+1 hashes should be enough for everyone...
cdSignatureSlot = 0x10000                      # CMS signature
cdIdentificationSlot = 0x10001                 # identification blob (detached signatures only)
cdTicketSlot = 0x10002                         # ticket embedded in signature (DMG only)

# Apple custom OIDs used in SignerInfo
SEC_OID_APPLE_HASH_AGILITY = '1.2.840.113635.100.9.1'
SEC_OID_APPLE_HASH_AGILITY_V2 = '1.2.840.113635.100.9.2'
SEC_OID_APPLE_EXPIRATION_TIME = '1.2.840.113635.100.9.3'

# CodeDirectory Versions
earliestVersion = 0x20001                      # Earliest supported
supportsScatter = 0x20100                      # First version to support scatter option
supportsTeamID = 0x20200                       # First version to support team ID option
supportsCodeLimit64 = 0x20300                  # First version to support codeLimit64
supportsExecSegment = 0x20400                  # First version to support exec base and limit
supportsPreEncrypt = 0x20500                   # First version to support pre-encrypt hashes and runtime version

# Hashes
cd_hash = None                                 # Hash of CodeDirectory blob. This is the hash that is embedded in the CMS message
info_plist_hash = None                         # Hash of the ../Info.plist file.
ireq_hash = None                               # Hash of the internal requirements blob.
resources_hash = None                          # Hash of the ../_CodeSiguatnre/CodeResources file
app_specific_hash = None                       # Hash of application specific blob
entitlement_hash = None                        # Hash of embedded entitlements blob
disk_rep_hash = None                           # Hash of disk rep blob?? (Is this such a thing)
entitlement_der_hash = None                   # Hash of entitlements DER slot

m = macholib.MachO.MachO(sys.argv[1])
h = m.headers[0]
sigmeta = [cmd for cmd in h.commands if cmd[0].cmd == LC_CODE_SIGNATURE]
sigmeta = sigmeta[0]

with open(sys.argv[1], 'rb') as f:
    f.seek(sigmeta[1].dataoff)
    sig = f.read(sigmeta[1].datasize)

with io.BytesIO(sig) as f:
    hdr = struct.unpack('>II', f.read(8))
    assert(hdr[0] == 0xfade0cc0)

    num = struct.unpack('>I', f.read(4))[0]
    slots = []
    for slot in range(num):
        (slot_id, offset) = struct.unpack('>II', f.read(8))
        slots.append((slot_id, offset))
        
    blobs = []
    for (slot_id, offset) in slots:
        f.seek(offset)
        (blob_id, blob_size) = struct.unpack('>II', f.read(8))
        # Rewind back to offset because blob_size includes id and size
        f.seek(offset)
        blob_data = f.read(blob_size)
        blobs.append((slot_id, blob_id, blob_data))

def sort_attributes(attrs_in):
    '''
    Sort the authenticated attributes for signing by re-encoding them, asn1crypto
    takes care of the actual sorting of the set.
    '''
    attrs_out = CMSAttributes()
    for attrval in attrs_in:
        attrs_out.append(attrval)
    return attrs_out

ctx = ValidationContext(trust_roots=[APPLE_ROOT], allow_fetching=False)
validate_chain = True

for (slot_id, blob_id, blob_data) in blobs:
    if slot_id == cdSignatureSlot:
        content = ContentInfo.load(blob_data[8:]) # Skip blob id and length
        sd = content['content']
        assert(isinstance(sd, SignedData))

        print('version', sd['version'].native)
        print('digest_algorithms', [a.native for a in sd['digest_algorithms']])
        print('encap_content_info', sd['encap_content_info'].native)

        # Parse certificates.
        certs = []
        for cert in sd['certificates']:
            c = cert.chosen
            assert(isinstance(c, asn1crypto.x509.Certificate))
            certs.append(c)

        intermediates = certs[0:-1]
        end_entity_cert = certs[-1]
        # this only works after adding
        #      '1.2.840.113635.100.6.1.13', # devid_execute
        # to supported_extensions in certvalidator/validate.py
        if validate_chain:
            validator = certvalidator.CertificateValidator(end_entity_cert, intermediates, ctx)
            validator.validate_usage({'digital_signature'}, {'code_signing'})

        # Validate SignerInfos
        # Inspired by https://github.com/ralphje/signify/blob/master/signify/signerinfo.py
        public_key = asymmetric.load_public_key(end_entity_cert.public_key)
        for signerinfo in sd['signer_infos']:
            assert(isinstance(signerinfo, asn1crypto.cms.SignerInfo))

            # Check the message digest hash
            for attr in signerinfo['signed_attrs']:
                if attr['type'].native == 'message_digest':
                    digest = attr['values'][0].native
                    if cd_hash is None:
                        cd_hash = digest
                    else:
                        assert(cd_hash == digest)

            data = sort_attributes(signerinfo['signed_attrs']).dump()
            signature = signerinfo['signature'].contents
            digest_algorithm = signerinfo['digest_algorithm']['algorithm'].native
            signature_algorithm = signerinfo['signature_algorithm']['algorithm'].native 
            assert(signature_algorithm == 'rsassa_pkcs1v15')
            # raises oscrypto.errors.SignatureError on wrong signature
            asymmetric.rsa_pkcs1v15_verify(public_key, signature, data, digest_algorithm)
    elif slot_id == cdCodeDirectorySlot:
        # Hash this entire blob, including version and length.
        # The signature slot contains a signature over this hash
        blob_hash = hashlib.sha256(blob_data).digest()
        if cd_hash is None:
            cd_hash = blob_hash
        else:
            assert(cd_hash == blob_hash)

        magic, size, version, flags, hashOffset, identOffset, nSpecialSlots, nCodeSlots, ncodeLimit, hashSize, hashType, platform, pageSize, _ = struct.unpack(">IIIIIIIIIBBBBI", blob_data[0:44])
        scatterOffset = None
        teamIDOffset = None
        codeLimit64 = None
        execSegBase = None
        execSegLimit = None
        execSegFlags = None
        runtime = None
        preEncryptOffset = None

        # Some fields are bogus because of versions. Make them None
        assert(version >= earliestVersion)
        if version >= supportsScatter:
            scatterOffset = struct.unpack(">I", blob_data[44:48])
        if version >= supportsTeamID:
            teamIDOffset = struct.unpack(">I", blob_data[48:52])
        if version >= supportsCodeLimit64:
            codeLimit64 = struct.unpack(">I", blob_data[52:56])
        if version >= supportsExecSegment:
            execSegBase, execSegLimit, execSegFlags = struct.unpack(">III", blob_data[56:68])
        if version >= supportsPreEncrypt:
            runtine, preEncryptOffset = struct.unpack(">II", blob_data[68:76])

        # pageSize is log2 of the page size, so get the actual page size
        page_size = 2 ** pageSize

        # Get the position of the 0'th hash. This hash is the beginning of the code slots, the special slots are negative index
        p = hashOffset

        # Read the special slot hashes
        special_slots = []
        content_dir = os.path.split(os.path.split(os.path.abspath(sys.argv[1]))[0])[0]
        for i in range(nSpecialSlots):
            slot = blob_data[p - hashSize:p]
            p -= hashSize

            slot_num = i + 1
            if slot_num == cdInfoSlot:
                info_plist_hash = slot
                plist_path = os.path.join(content_dir, "Info.plist")
                with open(plist_path, "rb") as f:
                    plist_hash = hashlib.sha256(f.read()).digest()
                    print(plist_hash.hex(), info_plist_hash.hex())
                    assert(plist_hash == info_plist_hash)
            elif slot_num == cdRequirementsSlot:
                if ireq_hash is None:
                    ireq_hash = slot
                else:
                    assert(slot == ireq_hash)
            elif slot_num == cdResourceDirSlot:
                resources_hash = slot
                resources_path = os.path.join(content_dir, "_CodeSignature", "CodeResources")
                with open(resources_path, "rb") as f:
                    res_hash = hashlib.sha256(f.read()).digest()
                    print(res_hash.hex(), resources_hash.hex())
                    assert(res_hash == resources_hash)
            elif slot_um == cdTopDirectorySlot:
                app_specific_hash = slot
            elif slot_num == cdEntitlementSlot:
                entitlement_hash = slot
            elif slot_num == cdRepSpecificSlot:
                disk_rep_hash = slot
            elif slot_num == cdEntitlementDERSlot:
                entitlement_der_hash = slot

        # Reset p for code slots
        p = hashOffset

        # Get the code hashes and check them
        with open(sys.argv[1], "rb") as f:
            for i in range(nCodeSlots):
                slot_hash = blob_data[p:p + hashSize]
                p += hashSize

                # Check if we are at the end
                to_read = page_size
                if f.tell() + page_size >= sigmeta[1].dataoff:
                    to_read = sigmeta[1].dataoff - f.tell()

                # Hash the binary
                data = f.read(to_read)
                data_hash = hashlib.sha256(data).digest()
                print(data_hash.hex(), slot_hash.hex())
                assert(data_hash == slot_hash)

    elif slot_id == cdRequirementsSlot:
        # We don't have to do anything with this slot except check it's hash
        blob_hash = hashlib.sha256(blob_data).digest()
        if ireq_hash is None:
            ireq_hash = blob_hash
        else:
            print(ireq_hash.hex(), blob_hash.hex())
            assert(ireq_hash == blob_hash)