|
| 1 | +import base64 |
| 2 | +import io |
| 3 | +import json |
| 4 | +from pathlib import Path |
| 5 | + |
| 6 | +import requests |
| 7 | +from pyhpke import AEADId, CipherSuite, KDFId, KEMId, KEMKey |
| 8 | +from structlog import get_logger |
| 9 | + |
| 10 | +from unblob.file_utils import File, InvalidInputFormat |
| 11 | +from unblob.models import ( |
| 12 | + Extractor, |
| 13 | + ExtractResult, |
| 14 | + Handler, |
| 15 | + HandlerDoc, |
| 16 | + HandlerType, |
| 17 | + HexString, |
| 18 | + Reference, |
| 19 | + ValidChunk, |
| 20 | +) |
| 21 | + |
| 22 | +logger = get_logger() |
| 23 | + |
| 24 | +# python-aea uses enum.StrEnum, which only exists from 3.11 on, so it is declared |
| 25 | +# as a Python 3.11+ dependency and imported lazily: importing it at module level |
| 26 | +# would take the whole CLI down on 3.10. |
| 27 | +_UNSUPPORTED_PYTHON = "AEA decryption needs Python 3.11 or newer" |
| 28 | + |
| 29 | +# HPKE suite as used by Apple AEA Profile 1 |
| 30 | +_HPKE_SUITE = CipherSuite.new( |
| 31 | + KEMId.DHKEM_P256_HKDF_SHA256, KDFId.HKDF_SHA256, AEADId.AES256_GCM |
| 32 | +) |
| 33 | + |
| 34 | + |
| 35 | +# AEA auth data (WKMS key fields) is a few KB; cap generously to reject false-positive |
| 36 | +# "AEA1" matches without reading a bogus multi-GB size into memory. |
| 37 | +_MAX_AUTH_DATA_SIZE = 1 << 20 |
| 38 | + |
| 39 | + |
| 40 | +def _parse_auth_data_fields(auth_data_blob: bytes) -> dict[str, bytes]: |
| 41 | + fields = {} |
| 42 | + while auth_data_blob: |
| 43 | + if len(auth_data_blob) < 4: |
| 44 | + raise InvalidInputFormat("AEA auth data: truncated field header") |
| 45 | + field_size = int.from_bytes(auth_data_blob[:4], "little") |
| 46 | + # field_size covers the 4-byte size prefix + "key\x00value"; must fit and advance. |
| 47 | + if not 4 < field_size <= len(auth_data_blob): |
| 48 | + raise InvalidInputFormat(f"AEA auth data: invalid field size {field_size}") |
| 49 | + key, sep, value = auth_data_blob[4:field_size].partition(b"\x00") |
| 50 | + if not sep: |
| 51 | + raise InvalidInputFormat("AEA auth data: field missing NUL separator") |
| 52 | + fields[key.decode("latin-1")] = value |
| 53 | + auth_data_blob = auth_data_blob[field_size:] |
| 54 | + return fields |
| 55 | + |
| 56 | + |
| 57 | +def _unwrap_session_key(fields: dict[str, bytes]) -> bytes: |
| 58 | + if ( |
| 59 | + "com.apple.wkms.fcs-response" not in fields |
| 60 | + or "com.apple.wkms.fcs-key-url" not in fields |
| 61 | + ): |
| 62 | + raise ValueError( |
| 63 | + "AEA file does not contain WKMS key fields — cannot decrypt without a pre-shared key" |
| 64 | + ) |
| 65 | + fcs_response = json.loads(fields["com.apple.wkms.fcs-response"]) |
| 66 | + enc_request = base64.b64decode(fcs_response["enc-request"]) |
| 67 | + wrapped_key = base64.b64decode(fcs_response["wrapped-key"]) |
| 68 | + url = fields["com.apple.wkms.fcs-key-url"].decode() |
| 69 | + |
| 70 | + r = requests.get(url, timeout=10) |
| 71 | + r.raise_for_status() |
| 72 | + privkey = KEMKey.from_pem(r.text) |
| 73 | + |
| 74 | + recipient = _HPKE_SUITE.create_recipient_context(enc_request, privkey) |
| 75 | + return recipient.open(wrapped_key) |
| 76 | + |
| 77 | + |
| 78 | +class AEAExtractor(Extractor): |
| 79 | + def extract(self, inpath: Path, outdir: Path) -> ExtractResult: |
| 80 | + try: |
| 81 | + from aea import aea as aeaformat # noqa: PLC0415 |
| 82 | + except ImportError: |
| 83 | + logger.warning( |
| 84 | + "AEA: cannot decrypt — skipping extraction", reason=_UNSUPPORTED_PYTHON |
| 85 | + ) |
| 86 | + return ExtractResult(reports=[]) |
| 87 | + |
| 88 | + with inpath.open("rb") as f: |
| 89 | + header = f.read(12) |
| 90 | + auth_data_size = int.from_bytes(header[8:12], "little") |
| 91 | + auth_data_blob = f.read(auth_data_size) |
| 92 | + |
| 93 | + fields = _parse_auth_data_fields(auth_data_blob) |
| 94 | + try: |
| 95 | + symmetric_key = _unwrap_session_key(fields) |
| 96 | + except ValueError as e: |
| 97 | + logger.warning("AEA: cannot decrypt — skipping extraction", reason=str(e)) |
| 98 | + return ExtractResult(reports=[]) |
| 99 | + logger.debug("AEA session key obtained", length=len(symmetric_key)) |
| 100 | + |
| 101 | + decrypted_path = outdir / "decrypted.bin" |
| 102 | + with inpath.open("rb") as infile, decrypted_path.open("wb") as outfile: |
| 103 | + try: |
| 104 | + aeaformat.decode_stream(infile, outfile, symmetric_key=symmetric_key) |
| 105 | + except aeaformat.MACValidationError as e: |
| 106 | + logger.error( |
| 107 | + "AEA MAC validation failed — symmetric_key is likely wrong", |
| 108 | + error=str(e), |
| 109 | + ) |
| 110 | + raise |
| 111 | + except aeaformat.ParseError as e: |
| 112 | + logger.error("AEA parse error", error=str(e)) |
| 113 | + raise |
| 114 | + logger.debug( |
| 115 | + "AEA decryption complete", output_size=decrypted_path.stat().st_size |
| 116 | + ) |
| 117 | + |
| 118 | + return ExtractResult(reports=[]) |
| 119 | + |
| 120 | + |
| 121 | +class AEAHandler(Handler): |
| 122 | + NAME = "aea" |
| 123 | + PATTERNS = [HexString("41 45 41 31")] # AEA1 |
| 124 | + EXTRACTOR = AEAExtractor() |
| 125 | + |
| 126 | + DOC = HandlerDoc( |
| 127 | + name="Apple Encrypted Archive (AEA)", |
| 128 | + description="Apple Encrypted Archive (AEA) is Apple's encrypted container format used for secure firmware and OTA update distribution. Profile 1 archives use Hybrid Public Key Encryption (HPKE) with Apple's WKMS key management service to wrap a per-archive symmetric key.", |
| 129 | + handler_type=HandlerType.ARCHIVE, |
| 130 | + vendor="Apple", |
| 131 | + references=[ |
| 132 | + Reference( |
| 133 | + title="Apple Archive - Apple Developer Documentation", |
| 134 | + url="https://developer.apple.com/documentation/applearchive", |
| 135 | + ), |
| 136 | + ], |
| 137 | + limitations=[ |
| 138 | + "Decryption requires access to Apple's WKMS key management service", |
| 139 | + "Archives without WKMS fields cannot be decrypted", |
| 140 | + ], |
| 141 | + ) |
| 142 | + |
| 143 | + def calculate_chunk(self, file: File, start_offset: int) -> ValidChunk | None: |
| 144 | + # Validate the auth-data header so false-positive "AEA1" matches (common inside |
| 145 | + # large filesystems) are rejected here instead of crashing the extractor. |
| 146 | + file.seek(start_offset, io.SEEK_SET) |
| 147 | + header = file.read(12) |
| 148 | + if len(header) < 12: |
| 149 | + raise InvalidInputFormat("AEA header truncated") |
| 150 | + auth_data_size = int.from_bytes(header[8:12], "little") |
| 151 | + if auth_data_size > _MAX_AUTH_DATA_SIZE: |
| 152 | + raise InvalidInputFormat(f"AEA auth data size too large: {auth_data_size}") |
| 153 | + auth_data_blob = file.read(auth_data_size) |
| 154 | + if len(auth_data_blob) < auth_data_size: |
| 155 | + raise InvalidInputFormat("AEA auth data truncated") |
| 156 | + _parse_auth_data_fields(auth_data_blob) |
| 157 | + |
| 158 | + file.seek(0, io.SEEK_END) |
| 159 | + return ValidChunk(start_offset=start_offset, end_offset=file.tell()) |
0 commit comments