diff --git a/src/netflix_msl/ios_client.py b/src/netflix_msl/ios_client.py index 27b5b24..f336a11 100644 --- a/src/netflix_msl/ios_client.py +++ b/src/netflix_msl/ios_client.py @@ -5,12 +5,14 @@ appboot から MSL セッション鍵確立までの End-to-end フローを実 ## フロー概要 (msl_key_relationship.md §2) Phase 0 (前提): ESN → TFIT-WB-AES → enc_key_0, sign_key_0 - (tools/emulate_tfit.py または Frida キャプチャで取得) + (tools/emulate_tfit.py で導出) Phase 3: kdf_renew(PSK, enc_key_0, sign_key_0, nonce) → enc_key_1, sign_key_1 Phase 1: DH 鍵ペア生成 → appboot リクエスト構築 → POST appboot.netflix.com + 署名 = HMAC-SHA256(sign_key_0, key_request_data_bytes) レスポンス解析 → server_scheme_data (96B), server_nonce (16B) + レスポンスヘッダー → x-netflix-deviceidtoken Phase 2: DH_compute_key(server_pub_key, client_priv_key) → HMAC-SHA384(48B_KEY, 0x00 || shared_secret) → enc_key, bootstrap_key @@ -19,6 +21,27 @@ appboot から MSL セッション鍵確立までの End-to-end フローを実 Phase 5: MSL 通信 (AES-128-CBC + HMAC-SHA256) +## 実測で確定した entity_auth_data 構造 (243 リクエストから確認) + + { + 35: { + "apphmac": bytes # 初回は b""; 以降は device_id_token.encode("utf-8") + "appid": str # IOS_APPID 定数 + "appkeyversion": int # 1 + "devicetoken": bytes # 初回は b""; 以降は base64.b64decode(device_id_token) + 3: str # 完全 ESN (整数キー) + }, + 30: "MGK_APPID" + } + + - apphmac は HMAC 値ではなく deviceIdToken 文字列の UTF-8 バイト列 + - devicetoken は deviceIdToken を Base64 デコードした 216B raw bytes + - 両フィールドの出所は同一: x-netflix-deviceidtoken レスポンスヘッダー + - 初回リクエストでは両フィールドとも b"" (空バイト) + - entity_auth_data の scheme 名は "MGK_APPID" (旧 "FAIRPLAY_MGK_APPID" は誤り) + - key_request_data の identity (key 33.8) は完全 ESN のみ (サフィックスなし) + - 署名対象は key_request_data_bytes のみ (ヘッダーは含まない) + ## 制約 - key 33.6 の TFIT エンコード部 (DH 公開鍵 128B → session_region 172B) は @@ -26,97 +49,91 @@ appboot から MSL セッション鍵確立までの End-to-end フローを実 現在は session_region をコンストラクタパラメータとして受け取る。 - server_scheme_data (96B) から server DH 公開鍵の抽出は未解明。 - 現在は server_scheme_data を直接パラメータとして受け取る。 - -## TODO (Frida キャプチャが必要な値) - - - enc_key_0 / sign_key_0: TFIT エミュレーション (emulate_tfit.py) で導出可能 - - devicetoken: Frida/Tweak `AppbootKeyExtract` でキャプチャ - - apphmac: NetflixCrypto.compute_apphmac(enc_key_0, sign_key_0) で計算可能 - - device_key_data: Frida/Tweak `AppbootKeyExtract` でキャプチャ (~6,576 bytes) - - session_region: TFIT エミュレーション (emulate_tfit.py) または Frida キャプチャ - - s1, s2, s3: Frida キャプチャで取得 - - renewable / capabilities: 既知の固定値 (ios_auth_manifest_license_flow.md §2.1) - - server_dh_pub_key: server_scheme_data (96B) のフォーマット解明後に自動抽出可能 """ from __future__ import annotations +import base64 +import hashlib +import hmac as hmac_mod +import os from dataclasses import dataclass, field +import cbor2 import requests -from netflix_msl.cbor_decoder import CborMslDecoder -from netflix_msl.cbor_encoder import CborMslEncoder -from netflix_msl.constants import IOS_APPBOOT_ENDPOINT, IOS_APPID, IOS_APPKEYVERSION +from netflix_msl.constants import ( + IOS_APPBOOT_ENDPOINT, + IOS_APPID, + IOS_APPKEYVERSION, + IOS_KEY336_DEVICE_HEADER, +) from netflix_msl.crypto import NetflixCrypto, SessionKeys +# --------------------------------------------------------------------------- +# CBOR 数値キー定数 +# --------------------------------------------------------------------------- + +_ENTITY_SCHEME = 30 +_ENTITY_AUTH_DATA = 35 +_KEYEX_SCHEME = 6 +_KEYEX_KEYDATA = 7 +_KEYEX_IDENTITY = 8 +_KEYEX_NONCE = 9 +_KEY_ENTITY_AUTH = 34 +_KEY_KEY_EXCHANGE = 33 +_KEY_MESSAGE_SIG = 16 + +# key 33.6 XOR nonce 変形マスク (165 サンプル全てで一致) +_N2_MASK = bytes([0xF4, 0x1B, 0x00, 0x00, 0x00, 0x00, 0x00]) +_N3_MASK = bytes([0xF4, 0x1B, 0x00, 0x18, 0x1B, 0x00, 0x00]) +_TAIL_MASK = bytes([0xF3, 0x1C, 0x07, 0x1F]) + @dataclass class iOSAppbootParams: """appboot リクエストに必要なデバイス固有パラメータ. - 全フィールドは Frida/Tweak キャプチャまたは TFIT エミュレーションで取得する。 + enc_key_0 / sign_key_0 は tools/emulate_tfit.py で ESN から導出する。 + session_region / s1 / s2 / s3 は TFIT エミュレーション実装後に自動導出予定。 """ - # Phase 0 MGK (TFIT エミュレーションで導出可能) + # Phase 0 MGK (tools/emulate_tfit.py で導出可能) enc_key_0: bytes - """Phase 0 MGK 暗号化鍵 (16 bytes). TFIT エミュレーションで導出可能.""" + """Phase 0 MGK 暗号化鍵 (16 bytes). tools/emulate_tfit.py で導出。""" sign_key_0: bytes - """Phase 0 MGK 署名鍵 (32 bytes). TFIT エミュレーションで導出可能.""" - - # FAIRPLAY_MGK_APPID entity_auth_data - appid: str - """アプリ識別子 UUID (e.g. 'a2becfec-b286-535c-b884-903a384caee6'). - TODO: Frida キャプチャで取得が必要 (バイナリに埋め込まれている可能性あり).""" - - devicetoken: str - """デバイストークン (hex string). - TODO: Frida/Tweak AppbootKeyExtract でキャプチャ.""" - - apphmac: str - """アプリ認証 HMAC-SHA256 (hex string). - NetflixCrypto.compute_apphmac(enc_key_0, sign_key_0).hex() で計算可能.""" - - device_key_data: bytes - """デバイス固有鍵データ (~6,576 bytes). - TODO: Frida/Tweak AppbootKeyExtract でキャプチャ.""" + """Phase 0 MGK 署名鍵 (32 bytes). tools/emulate_tfit.py で導出。""" # key 33.6 session_region session_region: bytes """key 33.6 平文の [128:300] セッション領域 (172 bytes). TFIT エミュレーション (tools/emulate_tfit.py) または Frida キャプチャで取得. - TODO: emulate_tfit.py による自動導出が未実装.""" + TODO: emulate_tfit.py による自動導出が未実装。""" # key 33.6 separators s1: bytes """key 33.6 の 9B セッション固定セパレータ (pt[307:316]). - TODO: Frida キャプチャで取得が必要.""" + TODO: Frida キャプチャで取得が必要。""" s2: bytes """key 33.6 の 9B セッション固定セパレータ (pt[323:332]). - TODO: Frida キャプチャで取得が必要.""" + TODO: Frida キャプチャで取得が必要。""" s3: bytes """key 33.6 の 9B セパレータ (pt[339:348]). byte[5] は per-request counter. - TODO: Frida キャプチャで取得が必要.""" - - # MSL ヘッダー固定値 - renewable: bytes - """capabilities key 16 の固定バイト列 (44 bytes). - 既知の固定値: 010100810001012022b1205c03559bc416af500d517f2c15463fc04717f8fb38b40c5ddce4e24fe11cc01955 - TODO: キャプチャから取得した値を使用すること.""" - - capabilities: dict - """capabilities dict (key 15 の値). - {10: bytes, 11: int, 12: int, 13: int, 14: int, 94: {95: True}} - TODO: キャプチャから取得した値を使用すること.""" + TODO: Frida キャプチャで取得が必要。""" # オプション + appid: str = IOS_APPID appkeyversion: int = IOS_APPKEYVERSION - esn_prefix: str = "" - scheme_suffix: str = "_3" + + # 継続セッション用 (初回は None) + device_id_token: str | None = None + """x-netflix-deviceidtoken ヘッダー値 (Base64 文字列). + 初回リクエストは None。以降は前回レスポンスの値をセットする。 + - apphmac = device_id_token.encode("utf-8") として entity_auth_data に設定 + - devicetoken = base64.b64decode(device_id_token) として entity_auth_data に設定""" @dataclass @@ -151,21 +168,18 @@ class iOSSessionState: class iOSMslClient: """iOS Netflix MSL appboot から暗号化 MSL リクエストまでのオーケストレーター. - 使用例 (Frida/Tweak でキャプチャした値を使用): + 使用例: + + from tools.emulate_tfit import compute_mgk # または外部から値を渡す params = iOSAppbootParams( - enc_key_0=bytes.fromhex("0817065e..."), + enc_key_0=bytes.fromhex("0817065e..."), # TFIT エミュレーション結果 sign_key_0=bytes.fromhex("91f752f7..."), - appid="a2becfec-b286-535c-b884-903a384caee6", - devicetoken="0608a1b7...", # TODO: Frida キャプチャ - apphmac=NetflixCrypto.compute_apphmac(enc_key_0, sign_key_0).hex(), - device_key_data=b"...", # TODO: Frida キャプチャ (~6576B) - session_region=b"...", # TODO: TFIT エミュレーション or Frida - s1=b"...", # TODO: Frida キャプチャ - s2=b"...", # TODO: Frida キャプチャ - s3=b"...", # TODO: Frida キャプチャ - renewable=bytes.fromhex("010100..."), - capabilities={10: b"...", ...}, + session_region=b"\\x00" * 172, # TODO: TFIT エミュレーション + s1=b"\\x00" * 9, # TODO: Frida キャプチャ + s2=b"\\x00" * 9, + s3=b"\\x00" * 9, + device_id_token=None, # 初回は None ) client = iOSMslClient( @@ -174,7 +188,7 @@ class iOSMslClient: ) session = client.perform_appboot() - # session.session_keys.enc_key_1 で以降の MSL リクエストを暗号化 + # session.device_id_token を次回 iOSAppbootParams.device_id_token に設定する """ def __init__( @@ -187,7 +201,7 @@ class iOSMslClient: Args: esn: PRV ESN (e.g. "NFAPPL-02-IPHONE9=1-AD0455...") - appboot_params: デバイス固有パラメータ (Frida キャプチャ値) + appboot_params: デバイス固有パラメータ timeout: HTTP タイムアウト秒数 (default 30) """ self.esn = esn @@ -219,11 +233,12 @@ class iOSMslClient: 手順: 1. DH 鍵ペアを生成 - 2. appboot CBOR リクエストを構築 + 2. entity_auth_data + key_request_data CBOR を構築 3. POST appboot.netflix.com/{ESN_PREFIX}?keyVersion=1 4. レスポンスを解析して server_scheme_data と server_nonce を抽出 - 5. DH 共有秘密を計算 - 6. Phase 2/3 KDF で SessionKeys を導出 + 5. x-netflix-deviceidtoken ヘッダーを記録 + 6. DH 共有秘密を計算 (server DH 公開鍵抽出が未実装のため暫定) + 7. Phase 2/3 KDF で SessionKeys を導出 Returns: iOSSessionState: セッション鍵・Cookie・deviceIdToken を含む状態 @@ -239,28 +254,11 @@ class iOSMslClient: # --- Step 2: appboot CBOR リクエストを構築 --- print("[appboot] Step 2: CBOR リクエストを構築中...") - encoder = CborMslEncoder(self.crypto) - cbor_message, k9_xor_nonce, nonce_7b = encoder.build_appboot_request( - esn=self.esn, - dh_pub_key=dh_pub_key, - appid=self.params.appid, - devicetoken=self.params.devicetoken, - apphmac=self.params.apphmac, - device_key_data=self.params.device_key_data, - session_region=self.params.session_region, - s1=self.params.s1, - s2=self.params.s2, - s3=self.params.s3, - renewable=self.params.renewable, - capabilities=self.params.capabilities, - appkeyversion=self.params.appkeyversion, - esn_prefix=self.params.esn_prefix, - scheme_suffix=self.params.scheme_suffix, - ) + cbor_message, k9_xor_nonce, nonce_7b = self._build_appboot_cbor() print(f" Request size: {len(cbor_message)} bytes") # --- Step 3: POST appboot --- - esn_prefix = self._extract_esn_prefix(self.esn) + esn_prefix = _extract_esn_prefix(self.esn) url = IOS_APPBOOT_ENDPOINT + esn_prefix params = {"keyVersion": str(self.params.appkeyversion)} @@ -280,8 +278,7 @@ class iOSMslClient: # --- Step 4: レスポンスを解析 --- print("[appboot] Step 4: レスポンスを解析中...") - decoder = CborMslDecoder(self.crypto) - parsed = decoder.parse_appboot_response(resp.content) + parsed = _parse_appboot_response(resp.content) server_scheme_data = parsed.get("server_scheme_data") server_nonce = parsed.get("server_nonce") @@ -295,13 +292,13 @@ class iOSMslClient: if server_nonce: print(f" server_nonce: {server_nonce.hex()}") - # Cookie と deviceIdToken を抽出 - nfvdid_cookie = self._extract_nfvdid_cookie(resp) + # Step 5: Cookie と deviceIdToken を抽出 + nfvdid_cookie = _extract_nfvdid_cookie(resp) device_id_token = resp.headers.get("x-netflix-deviceidtoken", "") if nfvdid_cookie: print(f" nfvdid cookie: {nfvdid_cookie[:32]}...") if device_id_token: - print(f" deviceIdToken: {device_id_token[:32]}...") + print(f" deviceIdToken: {device_id_token[:40]}...") if server_scheme_data is None: raise ValueError( @@ -312,13 +309,12 @@ class iOSMslClient: "server_nonce (key 33.9) が appboot レスポンスに存在しない" ) - # --- Step 5: DH 共有秘密を計算 --- + # --- Step 6: DH 共有秘密を計算 --- # TODO: server_scheme_data (96B) から server DH 公開鍵 (128B) を抽出するロジックが未解明。 - # 現在は server_scheme_data の構造が [IV(16B)][CT(48B)][HMAC(32B)] と推定されているが、 - # 実際の DH 公開鍵の位置・抽出方法は msl_cbor_key_exchange_analysis.md §5 参照。 - # Frida フック (DH_compute_key の入力引数) で server DH 公開鍵を直接キャプチャすること。 - print("[appboot] Step 5: DH 共有秘密を計算中...") - server_dh_pub_key = self._extract_server_dh_pub_key(server_scheme_data) + # 正確な抽出方法は Frida フック (DH_compute_key の peer_pub 引数) で + # server DH 公開鍵を直接キャプチャして解明する必要がある。 + print("[appboot] Step 6: DH 共有秘密を計算中...") + server_dh_pub_key = _extract_server_dh_pub_key(server_scheme_data) dh_shared_secret = NetflixCrypto.compute_dh_shared_secret( peer_public=server_dh_pub_key, private_key=dh_priv_key, @@ -327,8 +323,8 @@ class iOSMslClient: f" dh_shared_secret: {dh_shared_secret[:16].hex()}... ({len(dh_shared_secret)}B)" ) - # --- Step 6: Phase 2/3 KDF でセッション鍵を導出 --- - print("[appboot] Step 6: セッション鍵を導出中 (Phase 3 → Phase 2)...") + # --- Step 7: Phase 2/3 KDF でセッション鍵を導出 --- + print("[appboot] Step 7: セッション鍵を導出中 (Phase 3 → Phase 2)...") session_keys = NetflixCrypto.derive_full_key_chain( enc_key_0=self.params.enc_key_0, sign_key_0=self.params.sign_key_0, @@ -358,113 +354,99 @@ class iOSMslClient: print("[appboot] 完了") return self.session_state - def build_encrypted_msl_request( - self, - payload_body: dict | list | str, - endpoint: str, - keyid: str | None = None, - compress: bool = False, - ) -> bytes: - """暗号化 MSL リクエストを構築する. + # ------------------------------------------------------------------ + # 内部: CBOR メッセージ構築 + # ------------------------------------------------------------------ - perform_appboot() 完了後にのみ呼び出せる。 - enc_key_1 / sign_key_1 でペイロードを暗号化・署名する。 + def _build_appboot_cbor(self) -> tuple[bytes, bytes, bytes]: + """appboot CBOR メッセージを構築する. - Args: - payload_body: リクエスト本体 (dict/list/str) - endpoint: MSL エンドポイント URL (ログ用) - keyid: 鍵 ID。None の場合は "{ESN}_3" を使用。 - compress: True の場合 gzip 圧縮してから暗号化 + entity_auth_data 構造 (実測フォーマット): + { + 35: { + "apphmac": bytes # 初回は b""; 以降は device_id_token.encode("utf-8") + "appid": str + "appkeyversion": int + "devicetoken": bytes # 初回は b""; 以降は base64.b64decode(device_id_token) + 3: str # 完全 ESN (整数キー) + }, + 30: "MGK_APPID" + } + + key_request_data 構造: + { + 6: bytes(352B) # key 33.6 XOR 暗号化 scheme_data + 7: bytes(0B) # master_token (空) + 8: str # identity = 完全 ESN + 9: bytes(16B) # k9_xor_nonce + } + + 署名: HMAC-SHA256(sign_key_0, key_request_data_bytes) Returns: - CBOR エンコードされた暗号化 MSL リクエスト - - Raises: - RuntimeError: perform_appboot() が未完了の場合 + (cbor_message, k9_xor_nonce, nonce_7b) """ - if self.session_state is None: - raise RuntimeError( - "perform_appboot() を先に実行してセッションを確立すること" - ) + # --- apphmac / devicetoken を決定 --- + if self.params.device_id_token: + apphmac = self.params.device_id_token.encode("utf-8") + try: + pad = 4 - len(self.params.device_id_token) % 4 + padded = self.params.device_id_token + ("=" * pad if pad != 4 else "") + devicetoken = base64.b64decode(padded) + except Exception: + devicetoken = b"" + else: + apphmac = b"" + devicetoken = b"" - encoder = CborMslEncoder(self.crypto) - payload_bytes_raw = CborMslEncoder.build_payload_json( - message_id=self._generate_message_id(), - body=payload_body, - compress=compress, + # --- entity_auth_data を構築 --- + auth_data: dict = { + "apphmac": apphmac, + "appid": self.params.appid, + "appkeyversion": self.params.appkeyversion, + "devicetoken": devicetoken, + 3: self.esn, # 整数キー + } + entity_auth_data = { + _ENTITY_AUTH_DATA: auth_data, + _ENTITY_SCHEME: "MGK_APPID", + } + + # --- key 33.9 nonce と 7B nonce を生成 --- + k9_xor_nonce = os.urandom(16) + nonce_7b = os.urandom(7) + + # --- key 33.6 scheme_data (352B) を構築 --- + scheme_data_enc = _build_key336_scheme_data( + session_region=self.params.session_region, + nonce_7b=nonce_7b, + s1=self.params.s1, + s2=self.params.s2, + s3=self.params.s3, + k9_xor_nonce=k9_xor_nonce, ) - _keyid = keyid or f"{self.esn}_3" + # --- key_request_data を構築 --- + key_request_data: dict = { + _KEYEX_SCHEME: scheme_data_enc, + _KEYEX_KEYDATA: b"", # master_token (新規セッション) + _KEYEX_IDENTITY: self.esn, # identity = 完全 ESN + _KEYEX_NONCE: k9_xor_nonce, + } + key_request_bytes = cbor2.dumps(key_request_data) - # ヘッダー構築 (entity_auth_data なし) - header_bytes = encoder.build_header( - renewable=self.params.renewable, - capabilities=self.params.capabilities, - ) + # --- 署名: HMAC-SHA256(sign_key_0, key_request_data_bytes) --- + signature = hmac_mod.new( + self.params.sign_key_0, key_request_bytes, hashlib.sha256 + ).digest() - return encoder.build_encrypted_message( - header_bytes=header_bytes, - key_request_bytes=None, - payload_plaintext=payload_bytes_raw, - keyid=_keyid, - compress=False, # build_payload_json で既に対応済み - ) - - # ------------------------------------------------------------------ - # 内部ヘルパー - # ------------------------------------------------------------------ - - @staticmethod - def _extract_esn_prefix(esn: str) -> str: - """ESN から ESN プレフィックス部分を抽出する. - - "NFAPPL-02-IPHONE9=1-AD0455..." → "NFAPPL-02-IPHONE9=1-" - """ - idx = esn.rfind("-") - return esn[: idx + 1] if idx != -1 else esn - - @staticmethod - def _extract_nfvdid_cookie(resp: requests.Response) -> str: - """Set-Cookie ヘッダーから nfvdid Cookie 値を抽出する.""" - set_cookie = resp.headers.get("Set-Cookie", "") - if "nfvdid=" in set_cookie: - start = set_cookie.index("nfvdid=") + len("nfvdid=") - end = set_cookie.find(";", start) - return set_cookie[start:end] if end != -1 else set_cookie[start:] - # requests の cookies からも試みる - nfvdid = resp.cookies.get("nfvdid", "") - return nfvdid - - @staticmethod - def _extract_server_dh_pub_key(server_scheme_data: bytes) -> bytes: - """server_scheme_data (96B) からサーバー DH 公開鍵 (128B) を抽出する. - - TODO: server_scheme_data の構造が解明されていないため、現在は暫定実装。 - msl_cbor_key_exchange_analysis.md §5 の推定構造: - [IV(16B)][CT(48B)][HMAC(32B)] - サーバー DH 公開鍵は CT を dec_key で復号した後に得られると推定。 - dec_key の由来は未解明。 - - 正確な抽出方法は Frida フック (DH_compute_key の peer_pub 引数) で - server DH 公開鍵を直接キャプチャして解明する必要がある。 - - 現在の実装: - server_scheme_data の先頭 96B をゼロパディングして 128B を返す。 - 実際のセッション確立には Frida キャプチャした値を使用すること。 - """ - # TODO: 実際の DH 公開鍵抽出ロジックをここに実装する。 - # 暫定: server_scheme_data の先頭 96B + 32B ゼロパディング - # これは正しい値ではない。Frida でキャプチャした server DH pub_key を使用すること。 - if len(server_scheme_data) >= 128: - return server_scheme_data[:128] - return server_scheme_data.ljust(128, b"\x00") - - @staticmethod - def _generate_message_id() -> int: - """MSL メッセージ ID (rand in [0, 2^52)) を生成する.""" - import random - - return random.randint(0, 2**52) + # --- トップレベルメッセージを構築 --- + message = { + _KEY_ENTITY_AUTH: cbor2.dumps(entity_auth_data), + _KEY_KEY_EXCHANGE: key_request_bytes, + _KEY_MESSAGE_SIG: signature, + } + return cbor2.dumps(message), k9_xor_nonce, nonce_7b # ------------------------------------------------------------------ # セッション状態アクセサ @@ -484,3 +466,123 @@ class iOSMslClient: def device_id_token(self) -> str: """x-netflix-deviceidtoken 値。perform_appboot() 前は空文字列。""" return self.session_state.device_id_token if self.session_state else "" + + +# --------------------------------------------------------------------------- +# モジュールレベルヘルパー +# --------------------------------------------------------------------------- + + +def _build_key336_scheme_data( + session_region: bytes, + nonce_7b: bytes, + s1: bytes, + s2: bytes, + s3: bytes, + k9_xor_nonce: bytes, +) -> bytes: + """key 33.6 scheme_data 352B を XOR 暗号化して返す. + + NetflixCrypto.build_key336_scheme_data() と同じロジック。 + """ + n = nonce_7b + n2 = bytes(a ^ b for a, b in zip(n, _N2_MASK)) + n3 = bytes(a ^ b for a, b in zip(n, _N3_MASK)) + tail = bytes(a ^ b for a, b in zip(n[:4], _TAIL_MASK)) + + plaintext = ( + IOS_KEY336_DEVICE_HEADER + session_region + n + s1 + n2 + s2 + n3 + s3 + tail + ) + assert len(plaintext) == 352, f"plaintext length {len(plaintext)} != 352" + return bytes(plaintext[i] ^ k9_xor_nonce[i % 16] for i in range(352)) + + +def _extract_esn_prefix(esn: str) -> str: + """ESN から ESN プレフィックス部分を抽出する. + + "NFAPPL-02-IPHONE9=1-AD0455..." → "NFAPPL-02-IPHONE9=1-" + """ + idx = esn.rfind("-") + return esn[: idx + 1] if idx != -1 else esn + + +def _extract_nfvdid_cookie(resp: requests.Response) -> str: + """Set-Cookie ヘッダーから nfvdid Cookie 値を抽出する.""" + set_cookie = resp.headers.get("Set-Cookie", "") + if "nfvdid=" in set_cookie: + start = set_cookie.index("nfvdid=") + len("nfvdid=") + end = set_cookie.find(";", start) + return set_cookie[start:end] if end != -1 else set_cookie[start:] + return resp.cookies.get("nfvdid", "") + + +def _extract_server_dh_pub_key(server_scheme_data: bytes) -> bytes: + """server_scheme_data (96B) からサーバー DH 公開鍵 (128B) を抽出する. + + TODO: server_scheme_data の構造が解明されていないため、現在は暫定実装。 + 正確な抽出方法は Frida フック (DH_compute_key の peer_pub 引数) で + server DH 公開鍵を直接キャプチャして解明する必要がある。 + + 現在の実装: + server_scheme_data の先頭 96B をゼロパディングして 128B を返す。 + 実際のセッション確立には Frida キャプチャした値を使用すること。 + """ + if len(server_scheme_data) >= 128: + return server_scheme_data[:128] + return server_scheme_data.ljust(128, b"\x00") + + +def _parse_appboot_response(raw: bytes) -> dict: + """appboot レスポンス CBOR を解析する. + + Returns: + { + "server_scheme_data": bytes | None, + "server_nonce": bytes | None, + "scheme_id": str | None, + "status_flag": bytes | None, + "raw": dict, + } + """ + try: + top = cbor2.loads(raw) + except Exception as e: + raise ValueError(f"CBOR デコード失敗: {e}") from e + + result: dict = { + "server_scheme_data": None, + "server_nonce": None, + "scheme_id": None, + "status_flag": None, + "raw": top, + } + + krd_raw = top.get(_KEY_KEY_EXCHANGE) + if krd_raw is None: + return result + + try: + krd = cbor2.loads(krd_raw) if isinstance(krd_raw, bytes) else krd_raw + except Exception: + return result + + if not isinstance(krd, dict): + return result + + v6 = krd.get(_KEYEX_SCHEME) + if isinstance(v6, bytes): + result["server_scheme_data"] = v6 + + v7 = krd.get(_KEYEX_KEYDATA) + if isinstance(v7, bytes): + result["status_flag"] = v7 + + v8 = krd.get(_KEYEX_IDENTITY) + if v8 is not None: + result["scheme_id"] = str(v8) + + v9 = krd.get(_KEYEX_NONCE) + if isinstance(v9, bytes): + result["server_nonce"] = v9 + + return result diff --git a/tools/test_appboot_e2e.py b/tools/test_appboot_e2e.py new file mode 100644 index 0000000..794bbed --- /dev/null +++ b/tools/test_appboot_e2e.py @@ -0,0 +1,775 @@ +#!/usr/bin/env python3 +"""test_appboot_e2e.py — iOS MSL appboot End-to-End テスト + +ESN から MSL セッション鍵確立までの完全なチェーンをテストする。 + +## テストするフロー + + Phase 0: ESN → TFIT エミュレーション → enc_key_0, sign_key_0 + Phase 3: kdf_renew(PSK, enc_key_0, sign_key_0, nonce) → enc_key_1, sign_key_1 + Phase 1: DH 鍵ペア生成 → entity_auth_data + key_request_data を構築 + → HMAC-SHA256(sign_key_0, key_request_data_bytes) で署名 + → POST appboot.netflix.com/{ESN_PREFIX}?keyVersion=1 + Phase 2: server_scheme_data / server_nonce を解析 + → deviceIdToken を x-netflix-deviceidtoken ヘッダーから抽出 + +## 実測で確定した構造 + + entity_auth_data CBOR: + { + 35: { + "apphmac": bytes # 初回は b"" (以降は compute_apphmac() 結果) + "appid": str # IOS_APPID 定数 + "appkeyversion": int # 1 + "devicetoken": bytes # 初回は b"" (以降は base64.b64decode(device_id_token)) + 3: str # 完全 ESN (整数キー) + }, + 30: "MGK_APPID" + } + + key_request_data CBOR: + { + 6: bytes(352B) # key 33.6 XOR 暗号化 scheme_data + 9: bytes(16B) # key 33.9 XOR nonce + 8: str # identity = 完全 ESN + 7: bytes(0B) # master_token (新規セッション) + } + + signature: HMAC-SHA256(sign_key_0, key_request_data_bytes) + +## 使用法 + + uv run tools/test_appboot_e2e.py + uv run tools/test_appboot_e2e.py --esn + uv run tools/test_appboot_e2e.py --no-proxy + uv run tools/test_appboot_e2e.py --device-id-token --devicetoken-hex +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import hmac as hmac_mod +import os +import subprocess +import sys + +import cbor2 +import requests + +# プロジェクトの src を Python パスに追加 +_PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, os.path.join(_PROJECT_ROOT, "src")) + +from netflix_msl.constants import ( # noqa: E402 + IOS_APPBOOT_ENDPOINT, + IOS_APPID, + IOS_APPKEYVERSION, + IOS_KDF_NONCE, + IOS_KDF_PSK, + IOS_KEY336_DEVICE_HEADER, +) +from netflix_msl.crypto import NetflixCrypto # noqa: E402 + +# --------------------------------------------------------------------------- +# 定数 +# --------------------------------------------------------------------------- + +TEST_ESN = "NFAPPL-02-IPHONE9=1-AD0455EF27D3A7B8F0872932FD9837874AF3E6F90157195BD22A8063FEB0B79E" + +# デフォルトプロキシ (mitmproxy / Proxyman) +DEFAULT_PROXY = "http://192.168.0.51:9080" + +# CBOR 数値キー定数 +ENTITY_SCHEME = 30 +ENTITY_AUTH_DATA = 35 +KEYEX_SCHEME = 6 +KEYEX_KEYDATA = 7 +KEYEX_IDENTITY = 8 +KEYEX_NONCE = 9 +KEY_ENTITY_AUTH = 34 +KEY_KEY_EXCHANGE = 33 +KEY_MESSAGE_SIG = 16 + +# key 33.6 XOR nonce / nonce 変形マスク (実測値) +_N2_MASK = bytes([0xF4, 0x1B, 0x00, 0x00, 0x00, 0x00, 0x00]) +_N3_MASK = bytes([0xF4, 0x1B, 0x00, 0x18, 0x1B, 0x00, 0x00]) +_TAIL_MASK = bytes([0xF3, 0x1C, 0x07, 0x1F]) + + +# --------------------------------------------------------------------------- +# Step 0: TFIT エミュレーション +# --------------------------------------------------------------------------- + + +def run_tfit_emulation(esn: str) -> tuple[bytes, bytes]: + """tools/emulate_tfit.py を呼び出して MGK を導出する. + + Returns: + (enc_key_0, sign_key_0) — 各 16B / 32B + """ + script = os.path.join(_PROJECT_ROOT, "tools", "emulate_tfit.py") + result = subprocess.run( + ["uv", "run", script, esn], + capture_output=True, + text=True, + cwd=_PROJECT_ROOT, + ) + if result.returncode != 0: + raise RuntimeError(f"TFIT エミュレーション失敗:\n{result.stderr}") + + enc_key_0: bytes | None = None + sign_key_0: bytes | None = None + for line in result.stdout.splitlines(): + if "MGK key" in line and "(16B)" in line: + enc_key_0 = bytes.fromhex(line.split(":")[-1].strip()) + elif "MGK vector" in line and "(32B)" in line: + sign_key_0 = bytes.fromhex(line.split(":")[-1].strip()) + + if enc_key_0 is None or sign_key_0 is None: + raise RuntimeError(f"TFIT 出力から MGK を抽出できなかった:\n{result.stdout}") + return enc_key_0, sign_key_0 + + +# --------------------------------------------------------------------------- +# Step 1: entity_auth_data の構築 (実測フォーマット) +# --------------------------------------------------------------------------- + + +def build_entity_auth_data( + esn: str, + apphmac: bytes, + devicetoken: bytes, + appid: str = IOS_APPID, + appkeyversion: int = IOS_APPKEYVERSION, +) -> dict: + """実測フォーマットの entity_auth_data dict を構築する. + + entity_auth_data CBOR 構造 (243 リクエストから確認済み): + { + 35: { + "apphmac": bytes # raw bytes (初回は b"") + "appid": str + "appkeyversion": int + "devicetoken": bytes # raw bytes 216B (初回は b"") + 3: str # 完全 ESN (整数キー 3) + }, + 30: "MGK_APPID" + } + """ + auth_data: dict = { + "apphmac": apphmac, + "appid": appid, + "appkeyversion": appkeyversion, + "devicetoken": devicetoken, + 3: esn, # 整数キー + } + return { + ENTITY_AUTH_DATA: auth_data, + ENTITY_SCHEME: "MGK_APPID", + } + + +# --------------------------------------------------------------------------- +# Step 2: key_request_data の構築 (key 33) +# --------------------------------------------------------------------------- + + +def build_key336_scheme_data( + session_region: bytes, + nonce_7b: bytes, + s1: bytes, + s2: bytes, + s3: bytes, + k9_xor_nonce: bytes, +) -> bytes: + """key 33.6 scheme_data 352B を XOR 暗号化して返す. + + NetflixCrypto.build_key336_scheme_data() と同じロジック。 + """ + n = nonce_7b + n2 = bytes(a ^ b for a, b in zip(n, _N2_MASK)) + n3 = bytes(a ^ b for a, b in zip(n, _N3_MASK)) + tail = bytes(a ^ b for a, b in zip(n[:4], _TAIL_MASK)) + + plaintext = ( + IOS_KEY336_DEVICE_HEADER # [0:128] + + session_region # [128:300] + + n # [300:307] + + s1 # [307:316] + + n2 # [316:323] + + s2 # [323:332] + + n3 # [332:339] + + s3 # [339:348] + + tail # [348:352] + ) + assert len(plaintext) == 352 + return bytes(plaintext[i] ^ k9_xor_nonce[i % 16] for i in range(352)) + + +def build_key_request_data( + session_region: bytes, + s1: bytes, + s2: bytes, + s3: bytes, + esn: str, +) -> tuple[bytes, bytes, bytes]: + """key_request_data CBOR バイト列を構築する. + + Returns: + (key_request_bytes, k9_xor_nonce, nonce_7b) + """ + k9_xor_nonce = os.urandom(16) + nonce_7b = os.urandom(7) + + scheme_data = build_key336_scheme_data( + session_region=session_region, + nonce_7b=nonce_7b, + s1=s1, + s2=s2, + s3=s3, + k9_xor_nonce=k9_xor_nonce, + ) + + key_request = { + KEYEX_SCHEME: scheme_data, + KEYEX_KEYDATA: b"", # master_token = empty (新規セッション) + KEYEX_IDENTITY: esn, # identity = 完全 ESN (サフィックスなし) + KEYEX_NONCE: k9_xor_nonce, + } + return cbor2.dumps(key_request), k9_xor_nonce, nonce_7b + + +# --------------------------------------------------------------------------- +# Step 3: appboot メッセージを構築 +# --------------------------------------------------------------------------- + + +def build_appboot_message( + entity_auth_data: dict, + key_request_bytes: bytes, + sign_key_0: bytes, +) -> bytes: + """appboot CBOR MSL メッセージを構築する. + + トップレベル構造: + { + 34: bytes ← CBOR(entity_auth_data) + 33: bytes ← key_request_data + 16: bytes ← HMAC-SHA256(sign_key_0, key_request_data_bytes) + } + + 署名の対象は key_request_data_bytes のみ (実測で確認済み)。 + """ + signature = hmac_mod.new(sign_key_0, key_request_bytes, hashlib.sha256).digest() + + return cbor2.dumps( + { + KEY_ENTITY_AUTH: cbor2.dumps(entity_auth_data), + KEY_KEY_EXCHANGE: key_request_bytes, + KEY_MESSAGE_SIG: signature, + } + ) + + +# --------------------------------------------------------------------------- +# Step 4: レスポンス解析 +# --------------------------------------------------------------------------- + + +def parse_appboot_response(raw: bytes) -> dict: + """appboot レスポンス CBOR または JSON MSL エラーを解析する. + + サーバーは 2 種類のレスポンスを返す: + - 成功 (CBOR): {33: key_response_data_bytes, 16: signature, ...} + - エラー (JSON): {"signature": "...", "errordata": ""} + + Returns: + { + "server_scheme_data": bytes | None, + "server_nonce": bytes | None, + "scheme_id": str | None, + "status_flag": bytes | None, + "raw": dict, + "error": dict | None, # MSL エラーが返った場合 + } + """ + import json as _json + + result: dict = { + "server_scheme_data": None, + "server_nonce": None, + "scheme_id": None, + "status_flag": None, + "raw": {}, + "error": None, + } + + # JSON MSL エラーレスポンスを先に試みる + if raw and raw[0] == ord("{"): + try: + envelope = _json.loads(raw) + result["raw"] = envelope + errordata_b64 = envelope.get("errordata", "") + if errordata_b64: + import base64 as _b64 + + pad = 4 - len(errordata_b64) % 4 + padded = errordata_b64 + ("=" * pad if pad != 4 else "") + try: + error_dict = _json.loads(_b64.b64decode(padded)) + result["error"] = error_dict + except Exception: + result["error"] = {"_raw": errordata_b64} + return result + except _json.JSONDecodeError: + pass + + try: + top = cbor2.loads(raw) + except Exception as e: + raise ValueError(f"CBOR デコード失敗: {e}") from e + + result: dict = { + "server_scheme_data": None, + "server_nonce": None, + "scheme_id": None, + "status_flag": None, + "raw": top, + } + + krd_raw = top.get(KEY_KEY_EXCHANGE) + if krd_raw is None: + return result + + try: + krd = cbor2.loads(krd_raw) if isinstance(krd_raw, bytes) else krd_raw + except Exception: + return result + + if not isinstance(krd, dict): + return result + + v6 = krd.get(KEYEX_SCHEME) + if isinstance(v6, bytes): + result["server_scheme_data"] = v6 + + v7 = krd.get(KEYEX_KEYDATA) + if isinstance(v7, bytes): + result["status_flag"] = v7 + + v8 = krd.get(KEYEX_IDENTITY) + if v8 is not None: + result["scheme_id"] = str(v8) + + v9 = krd.get(KEYEX_NONCE) + if isinstance(v9, bytes): + result["server_nonce"] = v9 + + return result + + +# --------------------------------------------------------------------------- +# メインフロー +# --------------------------------------------------------------------------- + + +def run_e2e_test( + esn: str, + device_id_token: str | None, + devicetoken_bytes: bytes | None, + proxy_url: str | None, + timeout: int, +) -> None: + """appboot E2E テストを実行する.""" + sep = "=" * 60 + print(sep) + print("iOS MSL appboot End-to-End テスト") + print(sep) + print(f"ESN: {esn}") + if device_id_token: + print(f"device_id_token: {device_id_token[:40]}... (継続セッション)") + else: + print("device_id_token: なし (初回リクエスト)") + print() + + # ------------------------------------------------------------------ + # Phase 0: TFIT エミュレーションで MGK を導出 + # ------------------------------------------------------------------ + print("[Phase 0] TFIT エミュレーション — MGK 導出中...") + try: + enc_key_0, sign_key_0 = run_tfit_emulation(esn) + print(f" enc_key_0: {enc_key_0.hex()}") + print(f" sign_key_0: {sign_key_0.hex()}") + except RuntimeError as e: + print(f" ERROR: {e}") + print(" TFIT エミュレーション失敗。処理を中断します。") + sys.exit(1) + + # ------------------------------------------------------------------ + # Phase 3: KDF で sign_key_1 を導出 (appboot 署名用) + # ------------------------------------------------------------------ + print() + print("[Phase 3] KDF — sign_key_0 確認 (appboot 署名に sign_key_0 を使用)...") + enc_key_1, sign_key_1 = NetflixCrypto.kdf_renew( + IOS_KDF_PSK, enc_key_0, sign_key_0, IOS_KDF_NONCE + ) + print(f" enc_key_1: {enc_key_1.hex()}") + print(f" sign_key_1: {sign_key_1.hex()}") + print() + print(" NOTE: appboot 署名は sign_key_0 を使用 (実測で確認)") + + # ------------------------------------------------------------------ + # entity_auth_data フィールドを準備 + # ------------------------------------------------------------------ + if device_id_token: + # 継続セッション: deviceIdToken から apphmac と devicetoken を設定 + apphmac = device_id_token.encode("utf-8") + if devicetoken_bytes is not None: + devicetoken = devicetoken_bytes + else: + try: + # Base64 パディング補完してデコード + pad = 4 - len(device_id_token) % 4 + padded = device_id_token + ("=" * pad if pad != 4 else "") + devicetoken = base64.b64decode(padded) + except Exception as e: + print(f" WARNING: device_id_token Base64 デコード失敗: {e}") + devicetoken = b"" + else: + # 初回リクエスト: 両フィールドを空バイトに設定 + apphmac = b"" + devicetoken = b"" + + print() + print("[entity_auth] フィールド準備:") + print( + f" apphmac: {apphmac[:32].hex() if apphmac else '(empty)'} ({len(apphmac)}B)" + ) + print( + f" devicetoken: {devicetoken[:16].hex() if devicetoken else '(empty)'} ({len(devicetoken)}B)" + ) + + # ------------------------------------------------------------------ + # Phase 1a: DH 鍵ペアを生成 + # ------------------------------------------------------------------ + print() + print("[Phase 1a] DH 鍵ペア生成中...") + dh_priv_key, dh_pub_key = NetflixCrypto.generate_dh_keypair() + print(f" DH pub_key: {dh_pub_key[:16].hex()}... ({len(dh_pub_key)}B)") + + # ------------------------------------------------------------------ + # session_region を DH 公開鍵からゼロパディングで仮構築 + # ------------------------------------------------------------------ + # TODO: 本来は TFIT-WB-AES (tools/emulate_tfit.py の session_region 導出) が必要。 + # 現在は 172B のゼロ埋めプレースホルダーを使用。 + # このため key 33.6 の内容は正しくなく、サーバーは鍵交換を拒否する。 + # Phase 2 以降の DH 鍵合意は成立しない。 + # 正式フローでは emulate_tfit.py で DH 公開鍵を TFIT-WB-AES 暗号化し + # session_region (172B) に格納する。 + session_region_placeholder = b"\x00" * 172 + + # セパレータは実測キャプチャから取得した既知の値を使用 + # (セッション固有値のため、実際の接続では Frida キャプチャが必要) + s1 = b"\x00" * 9 + s2 = b"\x00" * 9 + s3 = b"\x00" * 9 + + # ------------------------------------------------------------------ + # Phase 1b: entity_auth_data と key_request_data を構築 + # ------------------------------------------------------------------ + print() + print("[Phase 1b] CBOR メッセージを構築中...") + + entity_auth_data = build_entity_auth_data( + esn=esn, + apphmac=apphmac, + devicetoken=devicetoken, + ) + + key_request_bytes, k9_xor_nonce, nonce_7b = build_key_request_data( + session_region=session_region_placeholder, + s1=s1, + s2=s2, + s3=s3, + esn=esn, + ) + + cbor_message = build_appboot_message( + entity_auth_data=entity_auth_data, + key_request_bytes=key_request_bytes, + sign_key_0=sign_key_0, + ) + + print(f" CBOR message size: {len(cbor_message)} bytes") + print(f" k9_xor_nonce: {k9_xor_nonce.hex()}") + print(f" nonce_7b: {nonce_7b.hex()}") + + # 構築したメッセージの構造を検証 + decoded_check = cbor2.loads(cbor_message) + print(f" top-level keys: {sorted(decoded_check.keys())}") + ead_check = cbor2.loads(decoded_check[KEY_ENTITY_AUTH]) + print(f" entity_auth scheme: {ead_check[ENTITY_SCHEME]}") + krd_check = cbor2.loads(decoded_check[KEY_KEY_EXCHANGE]) + print(f" key_request identity: {krd_check[KEYEX_IDENTITY]}") + + # ------------------------------------------------------------------ + # Phase 1c: POST appboot.netflix.com + # ------------------------------------------------------------------ + esn_prefix = esn[: esn.rfind("-") + 1] + url = IOS_APPBOOT_ENDPOINT + esn_prefix + params = {"keyVersion": str(IOS_APPKEYVERSION)} + + print() + print(f"[Phase 1c] HTTP POST — {url}") + if proxy_url: + print(f" proxy: {proxy_url}") + else: + print(" proxy: なし (直接接続)") + + proxies = {"https": proxy_url, "http": proxy_url} if proxy_url else None + session = requests.Session() + session.headers.update( + { + "User-Agent": "Netflix/24 CFNetwork/1335.0.3.4 Darwin/21.6.0", + "Content-Type": "application/x-www-form-urlencoded", + "X-Netflix.APIAction": "appboot", + "X-Netflix.client.ftl.esn": esn, + "X-Netflix.Request.Attempt": "1", + "X-Netflix.Request.Client.Context": '{"appState":"foreground"}', + } + ) + + try: + resp = session.post( + url, + params=params, + data=cbor_message, + proxies=proxies, + verify=False, # mitmproxy では証明書検証をスキップ + timeout=timeout, + ) + except requests.exceptions.ConnectionError as e: + print(f" CONNECTION ERROR: {e}") + print() + print("接続失敗。サーバーに到達できませんでした。") + print("原因として以下が考えられます:") + print(" - プロキシ (mitmproxy) が起動していない") + print(" - ネットワーク接続がない") + print(" - appboot.netflix.com がブロックされている") + _print_request_summary(cbor_message, esn, apphmac, devicetoken) + return + except requests.exceptions.Timeout: + print(f" TIMEOUT (>{timeout}s)") + _print_request_summary(cbor_message, esn, apphmac, devicetoken) + return + except Exception as e: + print(f" HTTP ERROR: {type(e).__name__}: {e}") + _print_request_summary(cbor_message, esn, apphmac, devicetoken) + return + + print(f" HTTP {resp.status_code} ({len(resp.content)} bytes)") + print(f" Content-Type: {resp.headers.get('content-type', 'N/A')}") + + # deviceIdToken を抽出 + new_device_id_token = resp.headers.get("x-netflix-deviceidtoken", "") + nfstatus = resp.headers.get("X-Netflix.nfstatus", "") + if nfstatus: + print(f" X-Netflix.nfstatus: {nfstatus}") + + print() + if resp.status_code != 200: + print(f"[RESULT] HTTP {resp.status_code} — appboot 失敗") + print(f" Response body: {resp.text[:500]}") + _print_request_summary(cbor_message, esn, apphmac, devicetoken) + return + + # ------------------------------------------------------------------ + # Phase 1d: レスポンス解析 + # ------------------------------------------------------------------ + print("[Phase 1d] レスポンスを解析中...") + + if new_device_id_token: + print(f" x-netflix-deviceidtoken: {new_device_id_token[:60]}...") + try: + pad = 4 - len(new_device_id_token) % 4 + padded = new_device_id_token + ("=" * pad if pad != 4 else "") + decoded_token = base64.b64decode(padded) + print( + f" decoded: {len(decoded_token)} bytes = {decoded_token[:16].hex()}..." + ) + except Exception: + pass + else: + print(" x-netflix-deviceidtoken: なし") + + nfvdid = resp.cookies.get("nfvdid", "") + if nfvdid: + print(f" nfvdid cookie: {nfvdid[:40]}...") + + try: + parsed = parse_appboot_response(resp.content) + except ValueError as e: + print(f" レスポンス解析エラー: {e}") + print(f" raw (hex): {resp.content[:80].hex()}") + return + + # MSL エラーレスポンスの場合は表示して終了 + msl_error = parsed.get("error") + if msl_error is not None: + print(f" MSL エラーレスポンス: {msl_error}") + print() + print(sep) + print("[RESULT] appboot MSL エラー返却") + print(sep) + print( + f"STATUS: MSL エラー (errorcode={msl_error.get('errorcode', '?')}," + f" internalcode={msl_error.get('internalcode', '?')})" + ) + print() + print("原因: entity_auth_data または key_request_data が不正です。") + print(" 正しい session_region (TFIT-WB-AES 暗号化 DH 公開鍵) が必要です。") + _print_request_summary(cbor_message, esn, apphmac, devicetoken) + return + + server_scheme_data = parsed["server_scheme_data"] + server_nonce = parsed["server_nonce"] + scheme_id = parsed["scheme_id"] + + print(f" scheme_id: {scheme_id}") + if server_scheme_data: + print( + f" server_scheme_data: {server_scheme_data[:16].hex()}... ({len(server_scheme_data)}B)" + ) + else: + print(" server_scheme_data: なし") + if server_nonce: + print(f" server_nonce: {server_nonce.hex()} ({len(server_nonce)}B)") + else: + print(" server_nonce: なし") + + # ------------------------------------------------------------------ + # 結果サマリー + # ------------------------------------------------------------------ + print() + print(sep) + print("[RESULT] appboot フロー結果") + print(sep) + + success = server_scheme_data is not None and server_nonce is not None + if success: + print("STATUS: サーバーから key_response_data を受信 (鍵交換データあり)") + print() + print("取得値:") + print(f" server_scheme_data: {len(server_scheme_data)}B") + print(f" server_nonce: {server_nonce.hex()}") + print(f" scheme_id: {scheme_id}") + if new_device_id_token: + print(f" device_id_token: {new_device_id_token[:60]}...") + print() + print("次回リクエスト用コマンド:") + print( + f" uv run tools/test_appboot_e2e.py --device-id-token '{new_device_id_token}'" + ) + print() + print("NOTE: session_region が正しい TFIT 暗号化 DH 公開鍵でないため、") + print(" サーバー側の DH 計算は失敗している可能性があります。") + print( + " Phase 2 (DH 鍵合意) を完了するには emulate_tfit.py の session_region 導出が必要。" + ) + else: + print("STATUS: key_response_data が不完全 — 鍵交換未完了") + raw_top = parsed.get("raw", {}) + print( + f" raw top-level keys: {sorted(raw_top.keys()) if isinstance(raw_top, dict) else 'N/A'}" + ) + + _print_request_summary(cbor_message, esn, apphmac, devicetoken) + + +def _print_request_summary( + cbor_message: bytes, + esn: str, + apphmac: bytes, + devicetoken: bytes, +) -> None: + """送信したリクエストの概要を表示する.""" + print() + print("--- 送信リクエスト概要 ---") + print(f" CBOR size: {len(cbor_message)} bytes") + print(f" ESN: {esn}") + print( + f" apphmac: {apphmac[:8].hex() if apphmac else '(empty)'} ({len(apphmac)}B)" + ) + print( + f" devicetoken: {devicetoken[:8].hex() if devicetoken else '(empty)'} ({len(devicetoken)}B)" + ) + + +# --------------------------------------------------------------------------- +# エントリーポイント +# --------------------------------------------------------------------------- + + +def main() -> None: + parser = argparse.ArgumentParser( + description="iOS MSL appboot End-to-End テスト", + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + parser.add_argument( + "--esn", + default=TEST_ESN, + help=f"PRV ESN (default: {TEST_ESN[:40]}...)", + ) + parser.add_argument( + "--device-id-token", + default=None, + help="継続セッション用 device_id_token (x-netflix-deviceidtoken ヘッダー値, Base64)", + ) + parser.add_argument( + "--devicetoken-hex", + default=None, + help="devicetoken の生バイト列 (hex)。--device-id-token と組み合わせて使用。", + ) + parser.add_argument( + "--proxy", + default=DEFAULT_PROXY, + help=f"プロキシ URL (default: {DEFAULT_PROXY})", + ) + parser.add_argument( + "--no-proxy", + action="store_true", + help="プロキシを使用しない (直接接続)", + ) + parser.add_argument( + "--timeout", + type=int, + default=30, + help="HTTP タイムアウト秒数 (default: 30)", + ) + + args = parser.parse_args() + + proxy_url = None if args.no_proxy else args.proxy + + devicetoken_bytes: bytes | None = None + if args.devicetoken_hex: + devicetoken_bytes = bytes.fromhex(args.devicetoken_hex) + + # urllib3 の SSL 警告を抑制 + import urllib3 + + urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) + + run_e2e_test( + esn=args.esn, + device_id_token=args.device_id_token, + devicetoken_bytes=devicetoken_bytes, + proxy_url=proxy_url, + timeout=args.timeout, + ) + + +if __name__ == "__main__": + main()