mirror of
https://github.com/qtmleap/revkit.git
synced 2026-10-26 02:31:51 +01:00
feat(frida,mitmproxy): Fridaフック改善とmitmproxyキャプチャスクリプト追加
- hook_netflix_ios.js: readVec修正 (toUInt32), aesCbcEncryptDecrypt独立フック, key_b64キャプチャ対応, hookCrypto有効化 - hook_appboot_bypass.js: appboot SSLピンニングバイパス (Frida版) - hook_appboot_openssl_bypass.js: OpenSSL C関数バイパス - hook_crash_trace.js: クラッシュ時スタックトレースキャプチャ - netflix_ios_capture.py: mitmproxyアドオン (TLSパススルー, Netflix通信キャプチャ) - msl_decoder.py: MSL CBOR/JSONデコーダー Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
tkgstrator
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
80c6f00408
commit
6b006d47f9
@@ -0,0 +1,468 @@
|
||||
"""Netflix MSL CBOR Decoder
|
||||
|
||||
iOS Netflix は MSL プロトコルに CBOR (RFC 7049) エンコーディングを使用する。
|
||||
CBOR メッセージは数値キーを使い、値自体もネストされた CBOR バイト列を含む。
|
||||
このモジュールは raw バイナリを受け取り、人間が読める Python dict に変換する。
|
||||
|
||||
使い方::
|
||||
|
||||
from msl_decoder import decode_msl_message
|
||||
|
||||
with open("req_appboot.bin", "rb") as f:
|
||||
result = decode_msl_message(f.read())
|
||||
|
||||
import json
|
||||
print(json.dumps(result, indent=2, ensure_ascii=False))
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import gzip
|
||||
import io
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
import cbor2
|
||||
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
# Numeric key → human-readable name mappings
|
||||
#
|
||||
# These come from analysis of captured iOS Netflix MSL traffic.
|
||||
# CBOR uses compact integer keys instead of JSON string keys.
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
|
||||
# Top-level MSL message keys
|
||||
MSL_TOP_KEYS: dict[int, str] = {
|
||||
16: "signature",
|
||||
32: "header",
|
||||
33: "payload",
|
||||
34: "entityauthdata",
|
||||
}
|
||||
|
||||
# Header fields (inside key 32)
|
||||
MSL_HEADER_KEYS: dict[int, str] = {
|
||||
10: "mastertoken",
|
||||
11: "sender_numeric",
|
||||
12: "messageid",
|
||||
13: "timestamp",
|
||||
14: "sequence_number",
|
||||
15: "capabilities",
|
||||
16: "renewable",
|
||||
94: "handshake_options",
|
||||
95: "handshake",
|
||||
}
|
||||
|
||||
# Capabilities sub-fields (inside header → 15)
|
||||
MSL_CAPABILITIES_KEYS: dict[int, str] = {
|
||||
10: "mastertoken",
|
||||
11: "sender_numeric",
|
||||
12: "messageid",
|
||||
13: "timestamp",
|
||||
14: "sequence_number",
|
||||
94: "handshake_options",
|
||||
95: "handshake",
|
||||
}
|
||||
|
||||
# Payload / key exchange fields (inside key 33)
|
||||
MSL_PAYLOAD_KEYS: dict[int, str] = {
|
||||
6: "ciphertext_or_scheme",
|
||||
7: "iv_or_keydata",
|
||||
8: "keyid",
|
||||
9: "sha256_or_hmac",
|
||||
}
|
||||
|
||||
# Entity auth data (inside key 34)
|
||||
MSL_ENTITY_AUTH_KEYS: dict[int, str] = {
|
||||
30: "scheme",
|
||||
35: "authdata",
|
||||
}
|
||||
|
||||
# Entity auth → authdata sub-fields
|
||||
MSL_AUTHDATA_KEYS: dict[int, str] = {
|
||||
50: "device_key_data",
|
||||
80: "esn_prefix",
|
||||
81: "esn",
|
||||
}
|
||||
|
||||
# String keys that already appear with names (not remapped)
|
||||
_STRING_KEY_PASSTHROUGH = {
|
||||
"apphmac",
|
||||
"appid",
|
||||
"appkeyversion",
|
||||
"devicetoken",
|
||||
}
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
# CBOR-aware recursive decoder
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def _try_cbor_decode(raw: bytes) -> Any | None:
|
||||
"""Try to decode bytes as CBOR. Returns None on failure."""
|
||||
try:
|
||||
return cbor2.loads(raw)
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _try_utf8(raw: bytes) -> str | None:
|
||||
"""Try to decode bytes as UTF-8 text."""
|
||||
try:
|
||||
return raw.decode("utf-8")
|
||||
except (UnicodeDecodeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def _bytes_to_value(raw: bytes, max_hex_len: int = 512) -> Any:
|
||||
"""Convert bytes to best human-readable representation.
|
||||
|
||||
Priority:
|
||||
1. Nested CBOR → recursively decode
|
||||
2. Valid UTF-8 string
|
||||
3. Hex string (if short enough)
|
||||
4. Truncated hex summary
|
||||
"""
|
||||
# 1. Try nested CBOR
|
||||
inner = _try_cbor_decode(raw)
|
||||
if inner is not None and not isinstance(inner, (int, float, bool)):
|
||||
# Avoid interpreting short byte sequences as bare CBOR integers
|
||||
return _decode_value(inner)
|
||||
|
||||
# 2. Try UTF-8
|
||||
text = _try_utf8(raw)
|
||||
if text is not None:
|
||||
return text
|
||||
|
||||
# 3. Hex encode
|
||||
if len(raw) <= max_hex_len:
|
||||
return raw.hex()
|
||||
return f"<bytes:{len(raw)}>{raw[:64].hex()}..."
|
||||
|
||||
|
||||
def _decode_value(obj: Any) -> Any:
|
||||
"""Recursively decode a CBOR-decoded Python value into JSON-safe form."""
|
||||
if isinstance(obj, bytes):
|
||||
return _bytes_to_value(obj)
|
||||
|
||||
if isinstance(obj, dict):
|
||||
result: dict[str, Any] = {}
|
||||
for k, v in obj.items():
|
||||
key_str = str(k)
|
||||
result[key_str] = _decode_value(v)
|
||||
return result
|
||||
|
||||
if isinstance(obj, list):
|
||||
return [_decode_value(item) for item in obj]
|
||||
|
||||
if isinstance(obj, cbor2.CBORTag):
|
||||
return {"__cbor_tag__": obj.tag, "value": _decode_value(obj.value)}
|
||||
|
||||
# Handle cbor2 break marker
|
||||
type_name = type(obj).__name__
|
||||
if "break_marker" in type_name:
|
||||
return "__CBOR_BREAK__"
|
||||
|
||||
# int, float, bool, str, None are already JSON-safe
|
||||
return obj
|
||||
|
||||
|
||||
def _remap_keys(obj: Any, key_map: dict[int, str] | None = None) -> Any:
|
||||
"""Apply human-readable key names to a decoded dict."""
|
||||
if not isinstance(obj, dict) or not key_map:
|
||||
return obj
|
||||
|
||||
result: dict[str, Any] = {}
|
||||
for k, v in obj.items():
|
||||
try:
|
||||
int_key = int(k)
|
||||
name = key_map.get(int_key, k)
|
||||
except (ValueError, TypeError):
|
||||
name = k
|
||||
result[name] = v
|
||||
return result
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
# High-level MSL message mapping
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def _map_authdata(authdata: Any) -> Any:
|
||||
"""Apply known field name mappings to entity auth data."""
|
||||
if not isinstance(authdata, dict):
|
||||
return authdata
|
||||
|
||||
result: dict[str, Any] = {}
|
||||
for k, v in authdata.items():
|
||||
try:
|
||||
int_key = int(k)
|
||||
name = MSL_AUTHDATA_KEYS.get(int_key, k)
|
||||
except (ValueError, TypeError):
|
||||
name = k
|
||||
result[name] = v
|
||||
return result
|
||||
|
||||
|
||||
def _map_entity_auth(entity_auth: Any) -> Any:
|
||||
"""Map entity auth fields (key 34)."""
|
||||
if not isinstance(entity_auth, dict):
|
||||
return entity_auth
|
||||
|
||||
mapped = _remap_keys(entity_auth, MSL_ENTITY_AUTH_KEYS)
|
||||
|
||||
if "authdata" in mapped and isinstance(mapped["authdata"], dict):
|
||||
mapped["authdata"] = _map_authdata(mapped["authdata"])
|
||||
|
||||
return mapped
|
||||
|
||||
|
||||
def _map_header(header: Any) -> Any:
|
||||
"""Map header fields (key 32)."""
|
||||
if not isinstance(header, dict):
|
||||
return header
|
||||
return _remap_keys(header, MSL_HEADER_KEYS)
|
||||
|
||||
|
||||
def _map_payload(payload: Any) -> Any:
|
||||
"""Map payload / key exchange fields (key 33)."""
|
||||
if not isinstance(payload, dict):
|
||||
return payload
|
||||
return _remap_keys(payload, MSL_PAYLOAD_KEYS)
|
||||
|
||||
|
||||
def _apply_msl_field_names(decoded: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Apply all known MSL field name mappings to a decoded message."""
|
||||
result: dict[str, Any] = {}
|
||||
|
||||
for k, v in decoded.items():
|
||||
try:
|
||||
int_key = int(k)
|
||||
name = MSL_TOP_KEYS.get(int_key, k)
|
||||
except (ValueError, TypeError):
|
||||
name = k
|
||||
|
||||
if name == "header":
|
||||
v = _map_header(v)
|
||||
elif name == "payload":
|
||||
v = _map_payload(v)
|
||||
elif name == "entityauthdata":
|
||||
v = _map_entity_auth(v)
|
||||
|
||||
result[name] = v
|
||||
|
||||
return result
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
# Format detection
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def _is_gzip(data: bytes) -> bool:
|
||||
"""Check for gzip magic bytes."""
|
||||
return len(data) >= 2 and data[0] == 0x1F and data[1] == 0x8B
|
||||
|
||||
|
||||
def _is_cbor(data: bytes) -> bool:
|
||||
"""Heuristic: CBOR self-describe tag (0xD9D9F7) or common CBOR map prefix."""
|
||||
if len(data) < 2:
|
||||
return False
|
||||
# Self-describe tag
|
||||
if data[:3] == b"\xd9\xd9\xf7":
|
||||
return True
|
||||
# Major type 5 (map) with various lengths
|
||||
first = data[0]
|
||||
major = first >> 5
|
||||
return major == 5 # map
|
||||
|
||||
|
||||
def _is_json(data: bytes) -> bool:
|
||||
"""Check if data looks like JSON."""
|
||||
stripped = data.lstrip()
|
||||
return len(stripped) > 0 and stripped[0:1] in (b"{", b"[")
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
# Streaming CBOR decoder (for responses with multiple items)
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def _decode_cbor_stream(data: bytes) -> list[Any]:
|
||||
"""Decode potentially concatenated CBOR items from a byte stream."""
|
||||
buf = io.BytesIO(data)
|
||||
items: list[Any] = []
|
||||
decoder = cbor2.CBORDecoder(buf)
|
||||
while buf.tell() < len(data):
|
||||
try:
|
||||
item = decoder.decode()
|
||||
items.append(item)
|
||||
except Exception:
|
||||
break
|
||||
return items
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
# Public API
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def decode_msl_message(
|
||||
raw: bytes,
|
||||
*,
|
||||
apply_names: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
"""Decode a Netflix MSL message from raw bytes.
|
||||
|
||||
Handles:
|
||||
- Gzip-compressed data (responses are often gzipped)
|
||||
- CBOR-encoded messages (iOS)
|
||||
- JSON-encoded messages (Chrome / Android)
|
||||
- URL-encoded form data (iosui endpoints)
|
||||
- Nested CBOR within byte string values
|
||||
- Encrypted payloads (returned as hex strings)
|
||||
|
||||
Args:
|
||||
raw: Raw bytes from the captured request/response body.
|
||||
apply_names: If True, map numeric CBOR keys to human-readable names.
|
||||
|
||||
Returns:
|
||||
A dict with keys:
|
||||
- ``format``: ``"cbor"``, ``"json"``, or ``"unknown"``
|
||||
- ``compressed``: bool, whether gzip was detected
|
||||
- ``messages``: list of decoded message dicts
|
||||
"""
|
||||
if not raw:
|
||||
return {"format": "unknown", "compressed": False, "messages": []}
|
||||
|
||||
compressed = _is_gzip(raw)
|
||||
data = raw
|
||||
|
||||
if compressed:
|
||||
try:
|
||||
data = gzip.decompress(raw)
|
||||
except Exception:
|
||||
return {
|
||||
"format": "unknown",
|
||||
"compressed": True,
|
||||
"messages": [],
|
||||
"error": "gzip decompression failed",
|
||||
}
|
||||
|
||||
# ── Try CBOR ──
|
||||
if _is_cbor(data):
|
||||
items = _decode_cbor_stream(data)
|
||||
if items:
|
||||
messages = []
|
||||
for item in items:
|
||||
decoded = _decode_value(item)
|
||||
if apply_names and isinstance(decoded, dict):
|
||||
decoded = _apply_msl_field_names(decoded)
|
||||
messages.append(decoded)
|
||||
return {
|
||||
"format": "cbor",
|
||||
"compressed": compressed,
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
# ── Try JSON (Chrome / Android MSL) ──
|
||||
if _is_json(data):
|
||||
try:
|
||||
text = data.decode("utf-8", errors="replace")
|
||||
except Exception:
|
||||
text = ""
|
||||
|
||||
# MSL JSON can be newline-delimited (header + payload chunks)
|
||||
messages = []
|
||||
for line in text.split("\n"):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
try:
|
||||
obj = json.loads(line)
|
||||
if isinstance(obj, dict):
|
||||
messages.append(obj)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
|
||||
if not messages:
|
||||
# Try as single JSON object
|
||||
try:
|
||||
single = json.loads(text)
|
||||
if isinstance(single, dict):
|
||||
messages = [single]
|
||||
elif isinstance(single, list):
|
||||
messages = [{"__array__": single}]
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
if messages:
|
||||
return {
|
||||
"format": "json",
|
||||
"compressed": compressed,
|
||||
"messages": messages,
|
||||
}
|
||||
|
||||
# ── Fallback: try CBOR anyway (no self-describe tag) ──
|
||||
try:
|
||||
items = _decode_cbor_stream(data)
|
||||
if items:
|
||||
messages = []
|
||||
for item in items:
|
||||
decoded = _decode_value(item)
|
||||
if apply_names and isinstance(decoded, dict):
|
||||
decoded = _apply_msl_field_names(decoded)
|
||||
messages.append(decoded)
|
||||
return {
|
||||
"format": "cbor",
|
||||
"compressed": compressed,
|
||||
"messages": messages,
|
||||
}
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# ── Unknown format ──
|
||||
return {
|
||||
"format": "unknown",
|
||||
"compressed": compressed,
|
||||
"messages": [],
|
||||
"raw_preview": data[:256].hex(),
|
||||
}
|
||||
|
||||
|
||||
def decode_msl_file(path: str, *, apply_names: bool = True) -> dict[str, Any]:
|
||||
"""Convenience: decode from a file path."""
|
||||
with open(path, "rb") as f:
|
||||
return decode_msl_message(f.read(), apply_names=apply_names)
|
||||
|
||||
|
||||
def decode_to_json(raw: bytes, *, apply_names: bool = True, indent: int = 2) -> str:
|
||||
"""Decode and return a JSON string."""
|
||||
result = decode_msl_message(raw, apply_names=apply_names)
|
||||
return json.dumps(result, indent=indent, ensure_ascii=False, default=str)
|
||||
|
||||
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
# CLI
|
||||
# ════════════════════════════════════════════════════════════════
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Command-line entry point for decoding MSL binary files."""
|
||||
import sys
|
||||
|
||||
if len(sys.argv) < 2:
|
||||
print(f"Usage: {sys.argv[0]} <file.bin> [file2.bin ...]", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
for path in sys.argv[1:]:
|
||||
if len(sys.argv) > 2:
|
||||
print(f"=== {path} ===")
|
||||
result = decode_msl_file(path)
|
||||
print(json.dumps(result, indent=2, ensure_ascii=False, default=str))
|
||||
if len(sys.argv) > 2:
|
||||
print()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user