iOSBuildServer/backend/automation/obfuscation/obfuscate_symbols.py

4294 lines
170 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
"""
obfuscate_symbols.py — Swift 类型名符号混淆工具
用法:
python3 AutoPacking/obfuscation/obfuscate_symbols.py --scan-dir readoor/3.0 # 仅预览,不修改
python3 AutoPacking/obfuscation/obfuscate_symbols.py --scan-dir readoor/3.0 --apply # 真正替换
python3 AutoPacking/obfuscation/obfuscate_symbols.py --deobfuscate crash.log # 还原 crash 符号
"""
import os
import re
import json
import hashlib
import argparse
import sys
import secrets
import time
import shutil
import tempfile
import multiprocessing
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path
from obfuscation_runtime.canonical import (
build_canonical_symbol_tables,
build_method_mapping_from_canonical_table,
build_type_mappings_from_canonical_table,
)
from obfuscation_runtime.discovery import discover_files, prefilter_files_with_rg
from obfuscation_runtime.json_cache import IncrementalAnalysisCache, load_json_file, save_json_file
# ── 配置 ──────────────────────────────────────────────────────────────────────
# 扫描这些目录(相对于脚本所在目录)。默认改为项目根以覆盖整个仓库,
# 也可通过命令行的 --scan-dir 覆盖为更精确的子目录。
DEFAULT_SCAN_DIRS = ["."]
# 排除目录(含这些路径片段的文件跳过)
EXCLUDE_PATH_FRAGMENTS = [
".build/",
".git/",
"Pods/",
"Carthage/",
"Tests/",
"readoorTests/",
"DerivedData/",
"RDObfuscateMacro/", # 我们自己的宏包,不混淆
".framework/",
".xcframework/",
]
THIRD_PARTY_PROTOCOL_EXCLUDE_PATH_FRAGMENTS = [
frag for frag in EXCLUDE_PATH_FRAGMENTS
if frag not in {"Pods/", ".framework/", ".xcframework/"}
]
# 目录名白名单:完全匹配这些目录名则在提取阶段忽略(例如第三方/工具目录)
EXCLUDE_DIR_NAMES = ("Factorys",)
METHOD_REPLACEMENT_SKIP_DIR_NAMES = ("Pods",)
# 始终保留(白名单)的符号,完全匹配
SYMBOL_WHITELIST = {
# AppKit / UIKit 入口
"AppDelegate", "SceneDelegate",
# 常见系统协议实现类型
"Error", "Result", "Event",
# Swift 内建类型
"String", "Int", "Bool", "Double", "Float", "Data", "URL", "Date",
"Array", "Dictionary", "Set", "Optional",
# JSON 编解码保留(必须与 JSON key 名一致,混淆会导致解码失败)
"CodingKeys",
# StoreKit / 内购 相关类型(保留,避免混淆第三方/系统 IAP 协议/模型)
"SKProduct", "SKProductsRequest", "SKProductsResponse", "SKPayment", "SKPaymentQueue",
"SKPaymentTransaction", "SKPaymentTransactionObserver", "SKProductsRequestDelegate",
"SKReceiptRefreshRequest", "SKStoreProductViewController", "SKMutablePayment",
# StoreKit 2 及 App Store 相关通用类型
"Product", "Transaction", "StoreKit", "AppStore",
}
# 前缀白名单:含这些前缀的符号跳过(第三方/系统)
SYMBOL_PREFIX_WHITELIST = ("NS", "UI", "AV", "CL", "MK", "SK", "CG", "CA", "WK")
# 一些语言/运行时保留标识符,绝对不能被替换(例如 Objective-C/Swift 的 `self`/`super` 等)
RESERVED_IDENTIFIERS = {
# Obj-C instance/reference keywords
"self", "super",
# Obj-C literals / common macros
"nil", "NULL", "YES", "NO",
# Swift / JSON booleans
"true", "false",
}
# 仅混淆这些 Swift 声明关键字(保守策略,先不动 func/var
DECL_KEYWORDS = ("class", "struct", "enum", "protocol")
# 混淆后名称前缀(让 crash 还原时还能快速识别来源)
# 类型名使用 YMH方法名使用 ymh避免 Swift 导入 Obj-C 时出现
# “'YMHxxxx()' has been renamed to 'ymhxxxx()'” 提示。
TYPE_OBFUSCATED_PREFIX = "YMH"
METHOD_OBFUSCATED_PREFIX = "ymh"
OBFUSCATED_PREFIX = TYPE_OBFUSCATED_PREFIX
# 混淆哈希长度hex 字符数)。增大此值可显著降低不同运行间重复概率。
OBFUSCATED_HASH_LEN = 10
LEGACY_OBFUSCATED_HASH_LENGTHS = (6, 8, 10)
ALREADY_OBFUSCATED_PREFIXES = (TYPE_OBFUSCATED_PREFIX, METHOD_OBFUSCATED_PREFIX)
# 仅对这些文件名前缀进行混淆(基于文件名,不含扩展)
# 例如 Swift 文件 `RDWereadOAuthWebViewController.swift` 会匹配 'RD'
SWIFT_FILENAME_PREFIXES = ("RD", "RB")
# 对 Obj-C 文件匹配的前缀(例如 `STS... .m/.h`
OBJC_FILENAME_PREFIXES = ("STS", "RB")
# 仅混淆这些符号名前缀(避免在 RD 文件中误混淆第三方或通用类)
SWIFT_SYMBOL_PREFIXES = ("RD", "RB")
OBJC_SYMBOL_PREFIXES = ("STS", "RB")
# 映射表保存路径
MAP_OUTPUT_PATH = "obfuscation_map.json"
APPLY_STATE_PATH = ".obfuscate_apply_state.json"
SCRIPT_DIR = Path(__file__).resolve().parent
PROJECT_METHOD_IGNORE_FILE_PATH = str(SCRIPT_DIR / "system_methods_ignore.txt")
AUTO_DETECTED_METHOD_IGNORE_BEGIN = "# BEGIN AUTO-DETECTED-OBJC-DOT-SYNTAX"
AUTO_DETECTED_METHOD_IGNORE_END = "# END AUTO-DETECTED-OBJC-DOT-SYNTAX"
SOURCE_FILE_EXTENSIONS = (".swift", ".m", ".mm", ".h", ".pch")
# 方法映射输出dry-run
METHOD_MAP_OUTPUT_PATH = "tmp_method_map.json"
DEFAULT_WORKERS = max(1, min(8, os.cpu_count() or 4))
IOS_SYSTEM_SYMBOL_CACHE_PATH = ".ios_system_symbols_cache.json"
IOS_SYSTEM_SYMBOL_CACHE_VERSION = 2
EXTRACTION_CACHE_PATH = ".obfuscate_extraction_cache.json"
EXTRACTION_CACHE_VERSION = 3
IOS_SYSTEM_SYMBOL_SCAN_DIRS = (
"System/Library/Frameworks",
"usr/include",
)
THIRD_PARTY_PROTOCOL_ROOT_NAMES = ("Pods", "Factorys")
METHOD_CANDIDATE_PREFIXES = ("STS", "RD", "RB")
METHOD_CANDIDATE_EXTENSIONS = (".swift", ".m", ".mm", ".h")
RG_DISCOVERY_EXTENSIONS = SOURCE_FILE_EXTENSIONS
RG_CHUNK_SIZE = 256
PROJECT_SOURCE_ANALYSIS_KIND = "project_source"
THIRD_PARTY_MEMBER_ANALYSIS_KIND = "third_party_member"
_EXTRACTION_ANALYSIS_CACHE = IncrementalAnalysisCache(EXTRACTION_CACHE_PATH, EXTRACTION_CACHE_VERSION)
TYPE_DECLARATION_RG_PATTERN = r"\b(?:class|struct|enum|protocol)\s+[A-Za-z_]|@(?:interface|implementation|protocol)\s+[A-Za-z_]"
METHOD_DECLARATION_RG_PATTERN = r"\bfunc\s+[A-Za-z_]|^[[:space:]]*[+-][[:space:]]*\("
PROPERTY_DECLARATION_RG_PATTERN = r"@property\b|\b(?:var|let)\s+[A-Za-z_]"
SELECTOR_LITERAL_RG_PATTERN = r'@"[A-Za-z_][A-Za-z0-9_:@]*"'
OBJC_DOT_SYNTAX_MEMBER_RG_PATTERN = r"\.\s*[A-Za-z_]"
# 常见系统成员名兜底(即使 SDK 扫描失败,也不要混淆这些高风险名字)
COMMON_SYSTEM_MEMBER_NAMES = {
# 来源: Apple Developer Documentation (常用 UIKit / Delegate / DataSource / WebKit / UserNotifications API)
# - https://developer.apple.com/documentation/uikit/uiviewcontroller
# - https://developer.apple.com/documentation/uikit/uiview
# - https://developer.apple.com/documentation/uikit/uislider
# - https://developer.apple.com/documentation/uikit/uiscrollviewdelegate
# - https://developer.apple.com/documentation/uikit/uitableviewdelegate
# - https://developer.apple.com/documentation/uikit/uitableviewdatasource
# - https://developer.apple.com/documentation/uikit/uicollectionviewdelegate
# - https://developer.apple.com/documentation/uikit/uicollectionviewdatasource
# - https://developer.apple.com/documentation/uikit/uiapplicationdelegate
# - https://developer.apple.com/documentation/uikit/uiscenedelegate
# - https://developer.apple.com/documentation/usernotifications/unusernotificationcenterdelegate
# - https://developer.apple.com/documentation/uikit/uigesturerecognizerdelegate
# - https://developer.apple.com/documentation/webkit/wkwebview
# - https://developer.apple.com/documentation/webkit/wknavigationdelegate
# 基本 Obj-C/Swift setter/getter 名
"set",
"get",
# UIViewController lifecycle / layout
"viewDidLoad", "viewWillAppear", "viewDidAppear", "viewWillDisappear", "viewDidDisappear",
"viewWillLayoutSubviews", "viewDidLayoutSubviews", "viewSafeAreaInsetsDidChange",
"viewWillLayoutSubviews", "viewDidLayoutSubviews", "viewWillLayoutSubviews",
"didReceiveMemoryWarning", "traitCollectionDidChange", "awakeFromNib", "deinit",
"prepareForSegue", "loadView", "loadViewIfNeeded", "setNeedsUpdateProperties",
# UIView / view hierarchy
"layoutSubviews", "layoutIfNeeded", "setNeedsLayout", "setNeedsDisplay",
"addSubview", "removeFromSuperview", "willMoveToSuperview", "didMoveToSuperview",
"willMoveToWindow", "didMoveToWindow", "didAddSubview", "willRemoveSubview",
"draw", "drawRect", "setNeedsDisplay", "tintColorDidChange",
"systemLayoutSizeFittingSize", "systemLayoutSizeFittingSize:",
"systemLayoutSizeFitting", "systemLayoutSizeFitting(_:withHorizontalFittingPriority:verticalFittingPriority:)",
# UIResponder / touch handling
"touchesBegan", "touchesMoved", "touchesEnded", "touchesCancelled",
# Gesture recognizer
"gestureRecognizerShouldBegin", "gestureRecognizer", "gestureRecognizer:shouldReceiveTouch:",
"gestureRecognizer(_:shouldRecognizeSimultaneouslyWith:)",
# Scroll / Zoom delegate
"scrollViewDidScroll", "scrollViewWillBeginDragging", "scrollViewWillEndDragging",
"scrollViewDidEndDragging", "scrollViewDidEndDecelerating", "scrollViewDidEndScrollingAnimation",
"scrollViewDidZoom", "scrollViewWillBeginZooming", "viewForZooming(in:)",
"scrollViewDidChangeAdjustedContentInset",
# TableView delegate / datasource (常见签名关键词)
"tableView", "didSelectRowAt", "didDeselectRowAt", "willSelectRowAt", "willDeselectRowAt",
"cellForRowAt", "numberOfRowsInSection", "numberOfSections", "heightForRowAt", "estimatedHeightForRowAt",
"willDisplayCell", "didEndDisplayingCell", "commit", "canEditRowAt", "moveRowAt",
"accessoryButtonTappedForRowWith", "leadingSwipeActionsConfigurationForRowAt", "trailingSwipeActionsConfigurationForRowAt",
# CollectionView delegate / datasource
"collectionView", "cellForItemAt", "numberOfItemsInSection", "didSelectItemAt", "didDeselectItemAt",
"willDisplayCellForItemAt", "didEndDisplayingCellForItemAt", "targetContentOffsetForProposedContentOffset",
# App / Scene lifecycle
"application:didFinishLaunchingWithOptions:", "application:open:options:",
"applicationDidBecomeActive", "applicationWillResignActive", "applicationDidEnterBackground",
"applicationWillEnterForeground", "applicationWillTerminate",
"application:didRegisterForRemoteNotificationsWithDeviceToken:",
"application:didFailToRegisterForRemoteNotificationsWithError:",
"application:didReceiveRemoteNotification:fetchCompletionHandler:",
# Scene delegate
"scene(_:willConnectTo:options:)", "sceneDidDisconnect", "sceneWillEnterForeground",
"sceneDidBecomeActive", "sceneWillResignActive", "sceneDidEnterBackground",
# Notifications / UNUserNotificationCenterDelegate
"userNotificationCenter:didReceiveNotificationResponse:withCompletionHandler:",
"userNotificationCenter:willPresentNotification:withCompletionHandler:",
"userNotificationCenter:openSettingsFor:",
# WebKit / WKNavigationDelegate
"webView(_:decidePolicyFor:decisionHandler:)", "webView(_:didStartProvisionalNavigation:)",
"webView(_:didFinish:)", "webView(_:didFail:withError:)",
# WKWebView documented instance methods (avoid obfuscating WebKit entry points used directly in app code)
"loadHTMLString", "loadFileURL", "loadFileRequest", "loadSimulatedRequest",
"reloadFromOrigin", "stopLoading", "goBack", "goForward", "go(to:)",
"evaluateJavaScript", "callAsyncJavaScript", "takeSnapshot", "createPDF",
"createWebArchiveData", "startDownload", "resumeDownload", "setMagnification",
"pauseAllMediaPlayback", "requestMediaPlaybackState", "setAllMediaPlaybackSuspended",
"closeAllMediaPresentations", "setCameraCaptureState", "setMicrophoneCaptureState",
"find", "setMinimumViewportInset", "fetchData", "restoreData",
# UISlider documented methods and subclass override points
"setValue", "minimumTrackImage", "setMinimumTrackImage",
"maximumTrackImage", "setMaximumTrackImage", "thumbImage", "setThumbImage",
"minimumValueImageRect", "maximumValueImageRect", "trackRect", "thumbRect",
# StoreKit / In-App Purchase callbacks (Obj-C selectors / Swift method names)
"paymentQueue:updatedTransactions:", "paymentQueueRestoreCompletedTransactionsFinished",
"paymentQueue:restoreCompletedTransactionsFailedWithError:", "productsRequest:didReceiveResponse:",
"request:didFailWithError:", "paymentQueue:removedTransactions:", "paymentQueue:updatedDownloads:",
# Text input delegates
"textFieldShouldBeginEditing", "textFieldDidBeginEditing", "textFieldShouldReturn",
"textField(_:shouldChangeCharactersIn:replacementString:)",
"textViewDidChange", "textViewDidBeginEditing", "textViewDidEndEditing",
# Common focus / state restoration hooks
"encodeRestorableState", "decodeRestorableState", "applicationFinishedRestoringState",
"archivedValue",
"responds", "forwardingTarget", "respondsToSelector", "forwardingTargetForSelector:",
# SwiftUI lightweight hooks (common closure names)
"onAppear", "onDisappear",
# Project-specific API exemptions
"consigneeCreate", "consigneeUpdate",
"addRightHandleClick",
"topModel",
"topPageVc",
"modelList",
"bookTypeToItemType",
"vocabularyArray",
"letterArray",
}
# UINavigationController 常用方法/属性来源Apple Developer Documentation
COMMON_SYSTEM_MEMBER_NAMES.update({
# Properties
"topViewController", "visibleViewController", "viewControllers",
"navigationBar", "toolbar", "isNavigationBarHidden", "isToolbarHidden",
"interactivePopGestureRecognizer", "interactiveContentPopGestureRecognizer",
"hidesBarsOnTap", "hidesBarsOnSwipe", "hidesBarsWhenVerticallyCompact",
"hidesBarsWhenKeyboardAppears", "barHideOnTapGestureRecognizer", "barHideOnSwipeGestureRecognizer",
"hideShowBarDuration",
# Swift-style method bases
"pushViewController", "popViewController", "popToViewController", "popToRootViewController",
"setViewControllers", "setNavigationBarHidden", "setToolbarHidden", "show", "showDetailViewController",
# Obj-C selectors / Obj-C-style names (保守兼容)
"pushViewController:animated:", "popViewControllerAnimated:", "popToViewController:animated:",
"popToRootViewControllerAnimated:", "setViewControllers:animated:",
"setNavigationBarHidden:animated:", "setToolbarHidden:animated:",
})
IOS_SYSTEM_MEMBER_NAMES = set()
IOS_SYSTEM_SYMBOLS_LOADED = False
# If enabled, allow rewriting `#import "X.h"` lines and rename corresponding header files.
# This is risky: filenames may diverge from disk unless headers are renamed accordingly.
REWRITE_IMPORTS_ALLOWED = False
# Pods / Factorys 中声明过的第三方成员名。保守处理:这些名字一律不参与方法混淆,
# 以避免继承第三方类、实现第三方 Delegate/DataSource 时把回调或重写方法改坏。
THIRD_PARTY_PROTOCOL_METHOD_NAMES = set()
THIRD_PARTY_PROTOCOL_METHODS_LOADED = False
PROJECT_PROPERTY_MEMBER_NAMES = set()
PROJECT_PROPERTY_MEMBERS_LOADED = False
PROJECT_METHOD_IGNORE_NAMES = set()
PROJECT_METHOD_IGNORE_PATTERNS = []
OBJC_PROPERTY_DECL_RE = re.compile(r"@property\b(?P<body>.*?);", re.S)
OBJC_PROPERTY_GETTER_ATTR_RE = re.compile(r"\bgetter\s*=\s*([A-Za-z_][A-Za-z0-9_]*)")
OBJC_PROPERTY_SETTER_ATTR_RE = re.compile(r"\bsetter\s*=\s*([A-Za-z_][A-Za-z0-9_]*:)")
OBJC_SYNTHESIZE_IVAR_RE = re.compile(r"@synthesize\s+[A-Za-z_][A-Za-z0-9_]*\s*=\s*_([A-Za-z_][A-Za-z0-9_]*)\s*;")
SWIFT_INTERFACE_FUNC_RE = re.compile(r"\bfunc\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(")
SWIFT_INTERFACE_PROPERTY_RE = re.compile(r"\b(?:var|let)\s+([A-Za-z_][A-Za-z0-9_]*)\s*:")
SWIFT_PROPERTY_DECL_RE = re.compile(
r"^\s*(?:@[A-Za-z_][A-Za-z0-9_]*(?:\([^\n]*?\))?\s*)*"
r"(?:(?:public|private|fileprivate|internal|open|final|weak|unowned|lazy|static|class|override|required|optional|dynamic|mutating|nonmutating|convenience|indirect)\s+)*"
r"(?:var|let)\s+([A-Za-z_][A-Za-z0-9_]*)\b",
re.M,
)
THIRD_PARTY_PROTOCOL_NAME_RE = re.compile(r"(?:Delegate|DataSource)\b", re.I)
SWIFT_PROTOCOL_DECL_RE = re.compile(r"\bprotocol\s+([A-Za-z_][A-Za-z0-9_]*)\b[^\{]*\{")
# ── 工具函数 ──────────────────────────────────────────────────────────────────
def short_hash(name: str, salt: str) -> str:
"""生成可复现的伪随机 ID长度由 OBFUSCATED_HASH_LEN 控制。"""
h = hashlib.sha256(f"{salt}{name}".encode()).hexdigest()[:OBFUSCATED_HASH_LEN]
return h
def obfuscated_name(original: str, salt: str, nonce: int = 0, prefix: str = TYPE_OBFUSCATED_PREFIX) -> str:
# nonce 用于极端哈希冲突时重试,保证本次映射唯一
payload = original if nonce == 0 else f"{original}:{nonce}"
return f"{prefix}{short_hash(payload, salt)}"
def build_unique_mapping(symbols: list, salt: str, prefix: str = TYPE_OBFUSCATED_PREFIX) -> dict:
mapping = {}
used_obf = set()
for name in symbols:
nonce = 0
while True:
obf = obfuscated_name(name, salt, nonce, prefix=prefix)
if obf not in used_obf:
mapping[name] = obf
used_obf.add(obf)
break
nonce += 1
return mapping
def should_exclude_path(path: str) -> bool:
return any(frag in path for frag in EXCLUDE_PATH_FRAGMENTS)
def collect_swift_files(scan_dirs: list) -> list:
return discover_files(scan_dirs, (".swift",), EXCLUDE_PATH_FRAGMENTS, skip_dir_names=EXCLUDE_DIR_NAMES)
def collect_objc_files(scan_dirs: list) -> list:
return discover_files(scan_dirs, (".m", ".mm", ".h"), EXCLUDE_PATH_FRAGMENTS, skip_dir_names=EXCLUDE_DIR_NAMES)
def collect_source_files(scan_dirs: list) -> list:
"""收集所有可恢复源码文件,不受 RD/STS 文件名前缀限制。"""
return discover_files(scan_dirs, SOURCE_FILE_EXTENSIONS, EXCLUDE_PATH_FRAGMENTS)
def collect_method_candidate_files(scan_dirs: list) -> list:
"""仅收集用于方法提取的候选文件:文件名以 STS 或 RD 开头,忽略 Factorys 等目录。"""
files = discover_files(
scan_dirs,
METHOD_CANDIDATE_EXTENSIONS,
EXCLUDE_PATH_FRAGMENTS,
skip_dir_names=EXCLUDE_DIR_NAMES,
basename_prefixes=METHOD_CANDIDATE_PREFIXES,
)
return prefilter_files_with_rg(files, METHOD_DECLARATION_RG_PATTERN, chunk_size=RG_CHUNK_SIZE)
def collect_apply_target_files(scan_dirs: list, enable_objc_symbols: bool) -> list:
"""收集实际执行替换的源码文件。
定义提取仍可只看 RD/STS 前缀文件,但符号引用可能出现在任意业务源码文件中,
因此替换阶段需要覆盖全部源码文件,否则会遗漏 String+xxx 之类的引用文件。
"""
return discover_files(scan_dirs, SOURCE_FILE_EXTENSIONS, EXCLUDE_PATH_FRAGMENTS)
def collect_resource_strings(scan_dirs: list) -> set:
"""扫描 storyboard/xib 等资源,收集 class/outlet/selector 等字符串作为白名单。"""
strs = set()
pattern_attr = re.compile(r'(?:customClass|customModule|placeholder|storyboardIdentifier|reuseIdentifier|userLabel|selector)\s*=\s*"([A-Za-z_][A-Za-z0-9_:@]*)"')
pattern_tag = re.compile(r'<[^>]+\bclass\s*=\s*"([A-Za-z_][A-Za-z0-9_]*)"')
resource_files = discover_files(scan_dirs, ('.storyboard', '.xib', '.xml'), EXCLUDE_PATH_FRAGMENTS)
for full in resource_files:
try:
text = Path(full).read_text(encoding='utf-8')
except Exception:
continue
for m in pattern_attr.finditer(text):
strs.add(m.group(1))
for m in pattern_tag.finditer(text):
strs.add(m.group(1))
return strs
def _discover_third_party_protocol_roots(scan_dirs: list) -> list:
roots = set()
excluded_dir_names = {"Pods", ".git", ".build", "Carthage", "Tests", "readoorTests", "DerivedData", "RDObfuscateMacro"}
search_roots = {str(Path(os.getcwd()).resolve())}
for scan_dir in scan_dirs:
try:
resolved_scan_dir = Path(scan_dir).resolve()
except Exception:
continue
if resolved_scan_dir.exists():
search_roots.add(str(resolved_scan_dir))
for search_root in sorted(search_roots):
base_path = Path(search_root)
if not base_path.exists():
continue
if base_path.name in THIRD_PARTY_PROTOCOL_ROOT_NAMES:
roots.add(str(base_path.resolve()))
for root, dirs, _ in os.walk(base_path):
current_root = Path(root)
for directory in list(dirs):
if directory in THIRD_PARTY_PROTOCOL_ROOT_NAMES:
roots.add(str((current_root / directory).resolve()))
dirs[:] = [
directory for directory in dirs
if directory not in excluded_dir_names and directory not in THIRD_PARTY_PROTOCOL_ROOT_NAMES
]
return sorted(roots)
def _iter_protocol_source_files(root_dirs: list):
seen = set()
candidates = discover_files(root_dirs, (".h", ".swift"), THIRD_PARTY_PROTOCOL_EXCLUDE_PATH_FRAGMENTS)
prefetched = prefilter_files_with_rg(candidates, PROPERTY_DECLARATION_RG_PATTERN + "|" + METHOD_DECLARATION_RG_PATTERN)
for file_path in prefetched:
full_path = os.path.realpath(file_path)
if full_path in seen:
continue
seen.add(full_path)
yield file_path
def _extract_swift_property_members(text: str) -> set:
return {match.group(1) for match in SWIFT_PROPERTY_DECL_RE.finditer(text)}
def _empty_project_source_analysis() -> dict:
return {
"swift_symbols": [],
"objc_symbols": [],
"raw_direct_names": [],
"raw_objc_selectors": [],
"raw_swift_signatures": {},
"property_members": [],
"selector_literals": [],
}
def _empty_third_party_member_analysis() -> dict:
return {
"member_names": [],
}
def load_extraction_cache() -> dict:
return _EXTRACTION_ANALYSIS_CACHE.load_state()
def save_extraction_cache():
_EXTRACTION_ANALYSIS_CACHE.save_if_dirty()
def _build_analysis_cache_key(analysis_kind: str, path: str) -> str:
return IncrementalAnalysisCache.build_cache_key(analysis_kind, path)
def _read_text_with_hash(path: str):
return IncrementalAnalysisCache._read_text_with_hash(path)
def _get_cached_file_analysis(path: str, analysis_kind: str, analyzer):
return _EXTRACTION_ANALYSIS_CACHE.get_or_compute(path, analysis_kind, analyzer)
def _analyze_project_source_text(text: str, path: str) -> dict:
analysis = _empty_project_source_analysis()
if not text:
return analysis
if path.endswith(".swift"):
analysis["swift_symbols"] = sorted(_extract_swift_symbols_from_text(text))
direct_names, swift_signatures = _extract_swift_methods_from_text(text)
analysis["raw_direct_names"] = sorted(direct_names)
analysis["raw_swift_signatures"] = swift_signatures
analysis["property_members"] = sorted(_extract_swift_property_members(text))
return analysis
if path.endswith((".h", ".m", ".mm", ".pch")):
direct_names, objc_selectors = _extract_objc_methods_from_text(text)
analysis["objc_symbols"] = sorted(_extract_objc_symbols_from_text(text))
analysis["raw_direct_names"] = sorted(direct_names)
analysis["raw_objc_selectors"] = sorted(objc_selectors)
analysis["property_members"] = sorted(_extract_objc_property_members(text))
analysis["selector_literals"] = sorted(_collect_selector_literals_from_text(text))
return analysis
def _analyze_third_party_member_text(text: str, path: str) -> dict:
analysis = _empty_third_party_member_analysis()
if not text:
return analysis
if path.endswith(".swift"):
analysis["member_names"] = sorted(_extract_delegate_protocol_methods_from_swift_text(text))
else:
analysis["member_names"] = sorted(_extract_delegate_protocol_methods_from_objc_text(text))
return analysis
def ensure_project_property_members_loaded(scan_dirs: list) -> set:
global PROJECT_PROPERTY_MEMBERS_LOADED
if PROJECT_PROPERTY_MEMBERS_LOADED:
return PROJECT_PROPERTY_MEMBER_NAMES
property_members = set()
source_files = collect_apply_target_files(scan_dirs, enable_objc_symbols=True)
property_files = prefilter_files_with_rg(source_files, PROPERTY_DECLARATION_RG_PATTERN)
for file_path in property_files:
analysis = _get_cached_file_analysis(file_path, PROJECT_SOURCE_ANALYSIS_KIND, _analyze_project_source_text)
property_members.update(analysis.get("property_members", []))
PROJECT_PROPERTY_MEMBER_NAMES.clear()
PROJECT_PROPERTY_MEMBER_NAMES.update(property_members)
PROJECT_PROPERTY_MEMBERS_LOADED = True
save_extraction_cache()
return PROJECT_PROPERTY_MEMBER_NAMES
def _protocol_name_is_delegate_like(name: str) -> bool:
return bool(THIRD_PARTY_PROTOCOL_NAME_RE.search(name or ""))
def _extract_delegate_protocol_methods_from_objc_text(text: str) -> set:
protected = set()
protected.update(_extract_objc_property_members(text))
for method_match in OBJC_METHOD_DECL_RE.finditer(text):
declaration = method_match.group(1).strip()
selector_parts = re.findall(r'([A-Za-z_][A-Za-z0-9_]*:)', declaration)
if selector_parts:
# Obj-C 带参数方法在 Swift 中实现时通常会以首段 selector 作为 base name
# 例如 `tencentDidNotLogin:` -> `func tencentDidNotLogin(_:)`。
# 这里同时保护完整 selector 和 Swift 可见的 base name
# 避免实现第三方 Delegate/DataSource 或重写第三方类方法时被误混淆。
protected.add("".join(selector_parts))
protected.add(selector_parts[0][:-1])
continue
simple_match = re.match(r'([A-Za-z_][A-Za-z0-9_]*)', declaration)
if simple_match:
protected.add(simple_match.group(1))
return protected
def _extract_delegate_protocol_methods_from_swift_text(text: str) -> set:
protected = set(_extract_swift_property_members(text))
for method_match in SWIFT_METHOD_DECL_RE.finditer(text):
protected.add(method_match.group(1))
return protected
def ensure_third_party_protocol_methods_loaded(scan_dirs: list) -> set:
global THIRD_PARTY_PROTOCOL_METHODS_LOADED
if THIRD_PARTY_PROTOCOL_METHODS_LOADED:
return THIRD_PARTY_PROTOCOL_METHOD_NAMES
# 保守策略Pods / Factorys 里声明过的成员名全部视为受保护,
# 这样第三方基类重写、第三方 Delegate / DataSource 实现都不会进入方法映射。
protected = set()
protocol_roots = _discover_third_party_protocol_roots(scan_dirs)
for file_path in _iter_protocol_source_files(protocol_roots):
analysis = _get_cached_file_analysis(file_path, THIRD_PARTY_MEMBER_ANALYSIS_KIND, _analyze_third_party_member_text)
protected.update(analysis.get("member_names", []))
THIRD_PARTY_PROTOCOL_METHOD_NAMES.clear()
THIRD_PARTY_PROTOCOL_METHOD_NAMES.update(protected)
THIRD_PARTY_PROTOCOL_METHODS_LOADED = True
save_extraction_cache()
return THIRD_PARTY_PROTOCOL_METHOD_NAMES
def _path_contains_dir_name(path: str, dir_names: tuple) -> bool:
wanted = {name.lower() for name in dir_names}
return any(part.lower() in wanted for part in Path(path).parts)
def should_skip_method_replacements(path: str) -> bool:
return _path_contains_dir_name(path, METHOD_REPLACEMENT_SKIP_DIR_NAMES)
def _run_command_capture_stdout(args: list) -> str:
try:
completed = subprocess.run(
args,
capture_output=True,
text=True,
check=True,
)
except Exception:
return ""
return (completed.stdout or "").strip()
def _discover_ios_sdk_paths() -> list:
sdk_paths = []
env_sdk_root = os.environ.get("SDKROOT")
if env_sdk_root and os.path.exists(env_sdk_root):
sdk_paths.append(os.path.realpath(env_sdk_root))
for sdk_name in ("iphoneos", "iphonesimulator"):
sdk_path = _run_command_capture_stdout(["xcrun", "--sdk", sdk_name, "--show-sdk-path"])
if sdk_path and os.path.exists(sdk_path):
sdk_paths.append(os.path.realpath(sdk_path))
unique_paths = []
seen = set()
for sdk_path in sdk_paths:
if sdk_path in seen:
continue
seen.add(sdk_path)
unique_paths.append(sdk_path)
return unique_paths
def _load_cached_ios_system_symbols(sdk_paths: list):
payload = load_json_file(IOS_SYSTEM_SYMBOL_CACHE_PATH, {})
if not isinstance(payload, dict):
return None
cache_version = payload.get("cache_version")
cached_sdk_paths = payload.get("sdk_paths")
cached_symbols = payload.get("symbols")
if cache_version != IOS_SYSTEM_SYMBOL_CACHE_VERSION or cached_sdk_paths != sdk_paths or not isinstance(cached_symbols, list):
return None
return set(symbol for symbol in cached_symbols if isinstance(symbol, str) and symbol)
def _save_ios_system_symbols_cache(sdk_paths: list, symbols: set):
payload = {
"cache_version": IOS_SYSTEM_SYMBOL_CACHE_VERSION,
"sdk_paths": sdk_paths,
"symbols": sorted(symbols),
}
try:
save_json_file(IOS_SYSTEM_SYMBOL_CACHE_PATH, payload)
except Exception:
pass
def _iter_ios_sdk_symbol_files(sdk_path: str):
seen = set()
sdk_root = Path(sdk_path)
for relative_dir in IOS_SYSTEM_SYMBOL_SCAN_DIRS:
base_dir = sdk_root / relative_dir
if not base_dir.exists():
continue
for root, dirs, files in os.walk(base_dir):
dirs[:] = [directory for directory in dirs if not directory.endswith(".docc")]
for filename in files:
if not filename.endswith((".h", ".swiftinterface")):
continue
full_path = os.path.realpath(os.path.join(root, filename))
if full_path in seen:
continue
seen.add(full_path)
yield full_path
def _default_objc_setter_name(property_name: str) -> str:
if property_name.startswith("is") and len(property_name) > 2 and property_name[2].isupper():
setter_base = property_name[2:]
else:
setter_base = property_name[:1].upper() + property_name[1:]
return f"set{setter_base}:"
def _extract_objc_property_members(text: str) -> set:
members = set()
for match in OBJC_PROPERTY_DECL_RE.finditer(text):
body = " ".join(match.group("body").split())
if not body:
continue
name_match = re.search(r"([A-Za-z_][A-Za-z0-9_]*)\s*$", body)
if not name_match:
continue
property_name = name_match.group(1)
members.add(property_name)
getter_match = OBJC_PROPERTY_GETTER_ATTR_RE.search(body)
if getter_match:
members.add(getter_match.group(1))
else:
members.add(property_name)
if "readonly" in body:
continue
setter_match = OBJC_PROPERTY_SETTER_ATTR_RE.search(body)
if setter_match:
members.add(setter_match.group(1))
else:
members.add(_default_objc_setter_name(property_name))
for match in OBJC_SYNTHESIZE_IVAR_RE.finditer(text):
property_name = match.group(1)
members.add(property_name)
members.add(_default_objc_setter_name(property_name))
return members
def _extract_objc_sdk_members(text: str) -> set:
members = set()
members.update(_extract_objc_property_members(text))
for match in OBJC_METHOD_DECL_RE.finditer(text):
declaration = match.group(1).strip()
selector_parts = re.findall(r'([A-Za-z_][A-Za-z0-9_]*:)', declaration)
if selector_parts:
members.add("".join(selector_parts))
continue
simple_match = re.match(r'([A-Za-z_][A-Za-z0-9_]*)', declaration)
if simple_match:
members.add(simple_match.group(1))
return members
def _extract_swift_sdk_members(text: str) -> set:
members = set()
for match in SWIFT_INTERFACE_FUNC_RE.finditer(text):
members.add(match.group(1))
for match in SWIFT_INTERFACE_PROPERTY_RE.finditer(text):
members.add(match.group(1))
return members
def _extract_ios_system_symbols_from_sdk(sdk_path: str) -> set:
members = set(COMMON_SYSTEM_MEMBER_NAMES)
for file_path in _iter_ios_sdk_symbol_files(sdk_path):
try:
text = Path(file_path).read_text(encoding="utf-8", errors="ignore")
except Exception:
continue
if file_path.endswith(".swiftinterface"):
members.update(_extract_swift_sdk_members(text))
else:
members.update(_extract_objc_sdk_members(text))
return members
def ensure_ios_system_symbols_loaded(verbose: bool = True) -> set:
global IOS_SYSTEM_SYMBOLS_LOADED
if IOS_SYSTEM_SYMBOLS_LOADED:
return IOS_SYSTEM_MEMBER_NAMES
sdk_paths = _discover_ios_sdk_paths()
symbols = _load_cached_ios_system_symbols(sdk_paths)
if symbols is None:
symbols = set(COMMON_SYSTEM_MEMBER_NAMES)
for sdk_path in sdk_paths:
symbols.update(_extract_ios_system_symbols_from_sdk(sdk_path))
if sdk_paths:
_save_ios_system_symbols_cache(sdk_paths, symbols)
IOS_SYSTEM_MEMBER_NAMES.clear()
IOS_SYSTEM_MEMBER_NAMES.update(symbols)
IOS_SYSTEM_SYMBOLS_LOADED = True
if verbose:
if sdk_paths:
print(f"🛡️ 已加载 {len(IOS_SYSTEM_MEMBER_NAMES)} 个 iOS SDK 系统方法/属性名")
else:
print("⚠️ 未找到 iOS SDK当前仅使用内置系统方法保护列表")
return IOS_SYSTEM_MEMBER_NAMES
# ── 提取阶段 ─────────────────────────────────────────────────────────────────
# 匹配声明前修饰符(注解行或同行),用于过滤不应混淆的符号
UNSAFE_MODIFIERS_RE = re.compile(
r"@objc\b|@objcMembers\b|@IBAction\b|@IBOutlet\b|@IBInspectable\b"
r"|@NSManaged\b|@NSApplicationMain\b|@UIApplicationMain\b"
r"|\bopen\b|\bpublic\b"
)
# 捕获类型声明: [class|struct|enum|protocol] TypeName
DECL_RE = re.compile(
r"(?<!\w)(?:" + "|".join(DECL_KEYWORDS) + r")\s+([A-Z][a-zA-Z0-9_]+)"
)
def build_already_obfuscated_re() -> re.Pattern:
lengths = sorted(set(LEGACY_OBFUSCATED_HASH_LENGTHS))
suffix = "|".join(f"[a-f0-9]{{{length}}}" for length in lengths)
prefixes = "|".join(re.escape(prefix) for prefix in sorted(set(ALREADY_OBFUSCATED_PREFIXES), key=len, reverse=True))
return re.compile(rf"^(?:{prefixes})(?:{suffix})$")
ALREADY_OBFUSCATED_RE = build_already_obfuscated_re()
# 方法抽取正则Swift / Obj-C
SWIFT_FUNC_RE = re.compile(r"^\s*(?:@[^\n]+\s*)*(?:public|private|fileprivate|internal|open|static|class|override|final|mutating|nonmutating|async|throws|rethrows|lazy|weak|unowned|convenience|required|optional|override|protocol|extension|extension\s+|actor|struct|enum|actor\s+)?\s*func\s+([a-zA-Z_][a-zA-Z0-9_]+)")
SWIFT_METHOD_DECL_RE = re.compile(r"\bfunc\s+([A-Za-z_][A-Za-z0-9_]*)\s*\(")
SWIFT_PARAM_LABEL_RE = re.compile(
r"^\s*(?:@[A-Za-z_][A-Za-z0-9_.]*(?:\([^)]*\))?\s+)*"
r"(?:(?:inout|borrowing|consuming|isolated|__shared|__owned|sending)\s+)*"
r"(?P<label>_|[A-Za-z_][A-Za-z0-9_]*)(?:\s+(?P<local>[A-Za-z_][A-Za-z0-9_]*))?\s*$"
)
SWIFT_ARGUMENT_LABEL_RE = re.compile(r"^\s*(?P<label>_|[A-Za-z_][A-Za-z0-9_]*)\s*$")
SWIFT_SIGNATURE_KEY_RE = re.compile(r"^([A-Za-z_][A-Za-z0-9_]*)\((.*)\)$")
OBJC_SELECTOR_RE = re.compile(r"@selector\\s*\\(\\s*([A-Za-z0-9_:@]+)\\s*\\)")
OBJC_STRING_SELECTOR_RE = re.compile(r'@"([A-Za-z_][A-Za-z0-9_:@]*)"')
OBJC_METHOD_DECL_RE = re.compile(r"[+-]\s*\([^)]*\)\s*([^{;]+)")
# 方法名前缀排除(例如私有下划线或第三方前缀)
METHOD_PREFIX_EXCLUDE = ("_", "qmui")
# 常见系统/Delegate 方法名排除模式(不应当混淆)
METHOD_EXCLUDE_PATTERNS = [
# UIKit / App lifecycle
r'(?i)viewDidLoad', r'(?i)viewWillAppear', r'(?i)viewDidAppear', r'(?i)viewWillDisappear', r'(?i)viewDidDisappear',
r'(?i)viewWillLayoutSubviews', r'(?i)viewDidLayoutSubviews', r'(?i)didReceiveMemoryWarning', r'(?i)traitCollectionDidChange',
# Init / dealloc / coder
r'(?i)^init$', r'(?i)^init:$', r'(?i)^initWith', r'(?i)^dealloc$', r'(?i)awakeFromNib', r'(?i)initWithCoder', r'(?i)encodeWithCoder',
# Table/collection/scroll delegate datasource
r'(?i)tableView', r'(?i)collectionView', r'(?i)scrollView', r'(?i)numberOfSectionsInTableView', r'(?i)numberOfRowsInSection',
r'(?i)cellForRowAtIndexPath', r'(?i)didSelectRowAtIndexPath', r'(?i)canEditRowAtIndexPath', r'(?i)heightForRowAtIndexPath',
r'(?i)willDisplayCell', r'(?i)reloadData', r'(?i)scrollViewDid',
# UICollectionViewLayout / UIView constraint hooks
r'(?i)^shouldInvalidateLayout(?:$|:|\()', r'(?i)^layoutAttributesFor(?:$|[A-Z]|:|\()',
r'(?i)^invalidationContext(?:$|[A-Z]|:|\()', r'(?i)^targetContentOffset(?:$|[A-Z]|:|\()',
r'(?i)^addConstraint(?:s)?(?:$|:|\()',
# UIPickerView delegate / datasource
r'(?i)^pickerView(?:$|:|\()', r'(?i)^numberOfComponentsInPickerView(?:$|:|\()', r'(?i)^numberOfComponents$',
# UITextField / UITextView delegate
r'(?i)^textField(?:$|:|\()', r'(?i)^textFieldShould', r'(?i)^textFieldDid', r'(?i)^textFieldWill',
r'(?i)^textView(?:$|:|\()', r'(?i)^textViewShould', r'(?i)^textViewDid', r'(?i)^textViewWill',
# WebKit delegate / handler
r'(?i)^webView(?:$|:|\()', r'(?i)^userContentController(?:$|:|\()',
# Common Swift datasource base names
r'(?i)^numberOfSections$',
# Touches / gestures
r'(?i)touchesBegan', r'(?i)touchesEnded', r'(?i)touchesMoved', r'(?i)gestureRecognizer',
# App delegate / URL handling
r'(?i)application:', r'(?i)didFinishLaunchingWithOptions', r'(?i)applicationWillResignActive', r'(?i)applicationDidEnterBackground',
r'(?i)applicationWillEnterForeground', r'(?i)applicationDidBecomeActive', r'(?i)applicationWillTerminate', r'(?i)application:openURL',
# Drawing / layout
r'(?i)layoutSubviews', r'(?i)sizeThatFits', r'(?i)drawRect', r'(?i)drawRect:',
r'(?i)^systemLayoutSizeFitting(?:Size)?(?:$|:|\()',
# Obj-C runtime / selector helpers
r'(?i)^responds$', r'(?i)^forwardingTarget$',
r'(?i)respondsToSelector', r'(?i)performSelector', r'(?i)NSSelectorFromString', r'(?i)forwardingTargetForSelector',
# Swift common lifecycle
r'(?i)deinit', r'(?i)awakeFromNib', r'(?i)prepareForSegue', r'(?i)encodeRestorableStateWithCoder', r'(?i)decodeRestorableStateWithCoder',
r'(?i)^archivedValue$',
# common prefixes to avoid
r'(?i)^set[A-Z]',
# Explicitly exclude Obj-C style setter selectors like setFoo: or setFoo:withBar:
r'(?i)^set[A-Z][A-Za-z0-9_]*:',
# Third-party layout DSLs (SnapKit / Masonry) must never be obfuscated
r'(?i)^(?:make|remake|update)Constraints$',
r'(?i)^(?:mas_)?(?:make|remake|update)Constraints:$',
# Layout / view hierarchy helpers
r'(?i)didMoveToSuperview', r'(?i)didMoveToWindow', r'(?i)willMoveToSuperview', r'(?i)willMoveToWindow',
r'(?i)^setNeedsLayout$', r'(?i)^setNeedsDisplay$', r'(?i)^layoutIfNeeded$',
r'(?i)^addSubview$', r'(?i)^removeFromSuperview$', r'(?i)^bringSubviewToFront$', r'(?i)^sendSubviewToBack$', r'(?i)^insertSubview',
# Navigation / Tab controller hooks
r'(?i)navigationController', r'(?i)tabBarController',
# More scroll / gesture delegate methods
r'(?i)scrollViewDidScroll', r'(?i)scrollViewDidEndDragging', r'(?i)scrollViewWillBeginDragging',
r'(?i)gestureRecognizerShouldBegin',
# Scene / App lifecycle (SceneDelegate / AppDelegate)
r'(?i)^sceneWillEnterForeground', r'(?i)^sceneDidEnterBackground', r'(?i)^sceneWillResignActive', r'(?i)^sceneDidBecomeActive',
r'(?i)^application:didRegisterForRemoteNotificationsWithDeviceToken', r'(?i)^application:didFailToRegisterForRemoteNotificationsWithError',
r'(?i)^application:didReceiveRemoteNotification',
# UserNotifications / UNUserNotificationCenter callbacks
r'(?i)^userNotificationCenter:',
# SwiftUI lifecycle closures
r'(?i)^onAppear$', r'(?i)^onDisappear$',
# Additional common prefixes and selectors collected from Apple docs
r'(?i)^viewSafeAreaInsetsDidChange$',
r'(?i)^gestureRecognizer:',
r'(?i)^tableView:', r'(?i)^collectionView:', r'(?i)^webView:',
r'(?i)^scrollView', r'(?i)^didSelectRowAt', r'(?i)^cellForRowAt',
r'(?i)^numberOfRowsInSection', r'(?i)^numberOfItemsInSection',
r'(?i)^application:open:', r'(?i)^scene', r'(?i)^userNotificationCenter:',
# WKWebView documented high-risk API surface
r'(?i)^loadHTMLString(?:$|:|\()', r'(?i)^loadFileURL(?:$|:|\()', r'(?i)^loadFileRequest(?:$|:|\()',
r'(?i)^loadSimulatedRequest(?:$|:|\()', r'(?i)^reloadFromOrigin(?:$|:|\()', r'(?i)^stopLoading(?:$|:|\()',
r'(?i)^goBack(?:$|:|\()', r'(?i)^goForward(?:$|:|\()', r'(?i)^go\s*\((?:$|:|\()',
r'(?i)^evaluateJavaScript(?:$|:|\()', r'(?i)^callAsyncJavaScript(?:$|:|\()',
r'(?i)^takeSnapshot(?:$|:|\()', r'(?i)^createPDF(?:$|:|\()', r'(?i)^createWebArchiveData(?:$|:|\()',
r'(?i)^startDownload(?:$|:|\()', r'(?i)^resumeDownload(?:$|:|\()', r'(?i)^setMagnification(?:$|:|\()',
r'(?i)^pauseAllMediaPlayback(?:$|:|\()', r'(?i)^requestMediaPlaybackState(?:$|:|\()',
r'(?i)^setAllMediaPlaybackSuspended(?:$|:|\()', r'(?i)^closeAllMediaPresentations(?:$|:|\()',
r'(?i)^setCameraCaptureState(?:$|:|\()', r'(?i)^setMicrophoneCaptureState(?:$|:|\()',
r'(?i)^find(?:$|:|\()', r'(?i)^setMinimumViewportInset(?:$|:|\()',
r'(?i)^fetchData(?:$|:|\()', r'(?i)^restoreData(?:$|:|\()',
# UISlider documented override points and public API
r'(?i)^minimumTrackImage(?:$|:|\()', r'(?i)^setMinimumTrackImage(?:$|:|\()',
r'(?i)^maximumTrackImage(?:$|:|\()', r'(?i)^setMaximumTrackImage(?:$|:|\()',
r'(?i)^thumbImage(?:$|:|\()', r'(?i)^setThumbImage(?:$|:|\()',
r'(?i)^minimumValueImageRect(?:$|:|\()', r'(?i)^maximumValueImageRect(?:$|:|\()',
r'(?i)^trackRect(?:$|:|\()', r'(?i)^thumbRect(?:$|:|\()',
# StoreKit / In-App Purchase common callbacks and prefixes
r'(?i)^paymentQueue', r'(?i)^productsRequest', r'(?i)^productsResponse',
r'(?i)^restoreCompletedTransactions', r'(?i)^refreshReceipt', r'(?i)^SKStoreProductViewController',
]
METHOD_EXCLUDE_RE = [re.compile(p) for p in METHOD_EXCLUDE_PATTERNS]
# 支持从仓库根目录加载自定义忽略模式(每行一条正则或文字)
def _is_literal_method_ignore_entry(value: str) -> bool:
return bool(value) and all(char.isalnum() or char in '_:' for char in value)
def _load_project_method_ignore_entries() -> tuple:
names = set()
patterns = []
try:
extra_path = os.path.join(os.getcwd(), PROJECT_METHOD_IGNORE_FILE_PATH)
if os.path.exists(extra_path):
with open(extra_path, 'r', encoding='utf-8') as f:
for ln in f:
s = ln.strip()
if not s or s.startswith('#'):
continue
if _is_literal_method_ignore_entry(s):
names.add(s)
continue
try:
patterns.append(re.compile(s))
except Exception:
pass
except Exception:
pass
return names, patterns
def refresh_project_method_ignore_entries() -> set:
global PROJECT_METHOD_IGNORE_NAMES, PROJECT_METHOD_IGNORE_PATTERNS
names, patterns = _load_project_method_ignore_entries()
PROJECT_METHOD_IGNORE_NAMES.clear()
PROJECT_METHOD_IGNORE_NAMES.update(names)
PROJECT_METHOD_IGNORE_PATTERNS = patterns
return PROJECT_METHOD_IGNORE_NAMES
refresh_project_method_ignore_entries()
def is_system_method(name: str) -> bool:
if (
name in COMMON_SYSTEM_MEMBER_NAMES or
name in IOS_SYSTEM_MEMBER_NAMES or
name in THIRD_PARTY_PROTOCOL_METHOD_NAMES or
name in PROJECT_PROPERTY_MEMBER_NAMES or
name in PROJECT_METHOD_IGNORE_NAMES
):
return True
# 兼容 Obj-C selector按分段判断系统 setter/getter 等场景
if ':' in name:
selector_parts = split_objc_selector(name)
if selector_parts:
for selector_part in selector_parts:
part_name = selector_part[:-1]
if (
selector_part in IOS_SYSTEM_MEMBER_NAMES or
selector_part in THIRD_PARTY_PROTOCOL_METHOD_NAMES or
selector_part in PROJECT_PROPERTY_MEMBER_NAMES or
selector_part in PROJECT_METHOD_IGNORE_NAMES or
part_name in COMMON_SYSTEM_MEMBER_NAMES or
part_name in IOS_SYSTEM_MEMBER_NAMES or
part_name in THIRD_PARTY_PROTOCOL_METHOD_NAMES or
part_name in PROJECT_PROPERTY_MEMBER_NAMES or
part_name in PROJECT_METHOD_IGNORE_NAMES
):
return True
for rx in METHOD_EXCLUDE_RE:
if rx.search(name):
return True
for rx in PROJECT_METHOD_IGNORE_PATTERNS:
if rx.search(name):
return True
return False
def is_safe_to_rename_method(name: str, preceding_lines: str) -> bool:
# 方法名级别的安全检查(与类型类似,但允许小写开头)
if ALREADY_OBFUSCATED_RE.match(name):
return False
if name in SYMBOL_WHITELIST:
return False
if name in RESERVED_IDENTIFIERS:
return False
if name.startswith(SYMBOL_PREFIX_WHITELIST):
return False
if UNSAFE_MODIFIERS_RE.search(preceding_lines):
return False
# 避免混淆 override / dynamic / protocol 实现
if re.search(r"\\boverride\\b|\\bdynamic\\b", preceding_lines):
return False
return True
def _find_matching_delimiter(text: str, start: int, open_char: str, close_char: str) -> int:
depth = 1
index = start + 1
while index < len(text):
if text.startswith("//", index):
index = _skip_line_comment(text, index)
continue
if text.startswith("/*", index):
index = _skip_block_comment(text, index)
continue
char = text[index]
if char in ('"', "'"):
index = _skip_quoted_text(text, index)
continue
if char == open_char:
depth += 1
elif char == close_char:
depth -= 1
if depth == 0:
return index
index += 1
return -1
def _find_top_level_char(text: str, target: str, track_angles: bool = False) -> int:
paren_depth = 0
brace_depth = 0
bracket_depth = 0
angle_depth = 0
index = 0
while index < len(text):
if text.startswith("//", index):
index = _skip_line_comment(text, index)
continue
if text.startswith("/*", index):
index = _skip_block_comment(text, index)
continue
char = text[index]
if char in ('"', "'"):
index = _skip_quoted_text(text, index)
continue
if char == '(':
paren_depth += 1
elif char == ')' and paren_depth > 0:
paren_depth -= 1
elif char == '{':
brace_depth += 1
elif char == '}' and brace_depth > 0:
brace_depth -= 1
elif char == '[':
bracket_depth += 1
elif char == ']' and bracket_depth > 0:
bracket_depth -= 1
elif track_angles and char == '<':
angle_depth += 1
elif track_angles and char == '>' and angle_depth > 0:
angle_depth -= 1
elif char == target and paren_depth == 0 and brace_depth == 0 and bracket_depth == 0 and angle_depth == 0:
return index
index += 1
return -1
def _split_top_level_segments(text: str, separator: str = ',', track_angles: bool = False) -> list:
segments = []
start = 0
paren_depth = 0
brace_depth = 0
bracket_depth = 0
angle_depth = 0
index = 0
while index < len(text):
if text.startswith("//", index):
index = _skip_line_comment(text, index)
continue
if text.startswith("/*", index):
index = _skip_block_comment(text, index)
continue
char = text[index]
if char in ('"', "'"):
index = _skip_quoted_text(text, index)
continue
if char == '(':
paren_depth += 1
elif char == ')' and paren_depth > 0:
paren_depth -= 1
elif char == '{':
brace_depth += 1
elif char == '}' and brace_depth > 0:
brace_depth -= 1
elif char == '[':
bracket_depth += 1
elif char == ']' and bracket_depth > 0:
bracket_depth -= 1
elif track_angles and char == '<':
angle_depth += 1
elif track_angles and char == '>' and angle_depth > 0:
angle_depth -= 1
elif char == separator and paren_depth == 0 and brace_depth == 0 and bracket_depth == 0 and angle_depth == 0:
segments.append((text[start:index], start, index))
start = index + 1
index += 1
segments.append((text[start:], start, len(text)))
return segments
def _find_previous_identifier(text: str, start: int):
index = start - 1
while index >= 0 and text[index].isspace():
index -= 1
end = index + 1
while index >= 0 and (text[index].isalnum() or text[index] == '_'):
index -= 1
if end <= index + 1:
return None
return text[index + 1:end]
def _serialize_swift_params(params: list) -> list:
return [
{
"label": param["label"],
"has_default": param["has_default"],
"allows_trailing_closure": bool(param.get("allows_trailing_closure", False)),
}
for param in params
]
def _build_swift_signature_key(name: str, params: list) -> str:
labels = []
for param in params:
labels.append(f"{param['label']}:")
return f"{name}({''.join(labels)})"
def _split_swift_signature_key(signature_key: str):
match = SWIFT_SIGNATURE_KEY_RE.match(signature_key)
if not match:
return None, None
return match.group(1), match.group(2)
def _swift_param_type_allows_trailing_closure(type_text: str) -> bool:
return "->" in type_text
def _parse_swift_decl_parameters(params_text: str):
params = []
for segment, start, _ in _split_top_level_segments(params_text, separator=',', track_angles=True):
if not segment.strip():
continue
colon_index = _find_top_level_char(segment, ':', track_angles=True)
if colon_index == -1:
return None
prefix = segment[:colon_index]
match = SWIFT_PARAM_LABEL_RE.match(prefix)
if not match:
return None
label = match.group("label")
local_name = match.group("local")
label_span = None
local_span = None
if label != "_":
label_span = (start + match.start("label"), start + match.end("label"))
if local_name:
local_span = (start + match.start("local"), start + match.end("local"))
type_and_default = segment[colon_index + 1:]
default_index = _find_top_level_char(type_and_default, '=', track_angles=True)
has_default = default_index != -1
type_text = type_and_default if default_index == -1 else type_and_default[:default_index]
params.append({
"label": label,
"local_name": local_name,
"has_default": has_default,
"allows_trailing_closure": _swift_param_type_allows_trailing_closure(type_text),
"span": label_span,
"local_span": local_span,
})
return params
def _swift_runtime_param_name(param: dict):
local_name = param.get("local_name")
if local_name:
return local_name
label = param.get("label")
if label and label != "_":
return label
return None
def _find_swift_decl_body_start(text: str, index: int):
paren_depth = 0
bracket_depth = 0
angle_depth = 0
while index < len(text):
if text.startswith("//", index):
index = _skip_line_comment(text, index)
continue
if text.startswith("/*", index):
index = _skip_block_comment(text, index)
continue
char = text[index]
if char in ('\"', "'"):
index = _skip_quoted_text(text, index)
continue
if char == '(':
paren_depth += 1
elif char == ')' and paren_depth > 0:
paren_depth -= 1
elif char == '[':
bracket_depth += 1
elif char == ']' and bracket_depth > 0:
bracket_depth -= 1
elif char == '<':
angle_depth += 1
elif char == '>' and angle_depth > 0:
angle_depth -= 1
elif char == '{' and paren_depth == 0 and bracket_depth == 0 and angle_depth == 0:
return index
elif char == ';' and paren_depth == 0 and bracket_depth == 0 and angle_depth == 0:
return None
index += 1
return None
def _collect_swift_local_param_scopes(text: str):
scopes = []
for match in SWIFT_METHOD_DECL_RE.finditer(text):
open_paren = match.end() - 1
close_paren = _find_matching_delimiter(text, open_paren, '(', ')')
if close_paren == -1:
continue
params = _parse_swift_decl_parameters(text[open_paren + 1:close_paren])
if params is None:
continue
local_names = {
runtime_name
for runtime_name in (_swift_runtime_param_name(param) for param in params)
if runtime_name
}
if not local_names:
continue
body_start = _find_swift_decl_body_start(text, close_paren + 1)
if body_start is None:
continue
body_end = _find_matching_delimiter(text, body_start, '{', '}')
if body_end == -1:
continue
scopes.append((body_start + 1, body_end, local_names))
return scopes
def _swift_has_member_receiver(text: str, identifier_start: int) -> bool:
dot_index = identifier_start - 1
while dot_index >= 0 and text[dot_index].isspace():
dot_index -= 1
if dot_index < 0 or text[dot_index] != '.':
return False
receiver_index = dot_index - 1
while receiver_index >= 0 and text[receiver_index].isspace():
receiver_index -= 1
while receiver_index >= 0 and text[receiver_index] in ('?', '!'):
receiver_index -= 1
while receiver_index >= 0 and text[receiver_index].isspace():
receiver_index -= 1
if receiver_index < 0:
return False
receiver_char = text[receiver_index]
return receiver_char.isalnum() or receiver_char == '_' or receiver_char in (')', ']', '}')
def _swift_is_shadowed_local_param(text: str, identifier_start: int, identifier: str, local_param_scopes: list) -> bool:
if _swift_has_member_receiver(text, identifier_start):
return False
for scope_start, scope_end, local_names in local_param_scopes:
if scope_start <= identifier_start < scope_end and identifier in local_names:
return True
return False
def _extract_swift_methods_from_text(text: str) -> tuple:
direct_names = set()
swift_signatures = {}
for match in SWIFT_METHOD_DECL_RE.finditer(text):
name = match.group(1)
if name.startswith(METHOD_PREFIX_EXCLUDE):
continue
pos = match.start()
line_index = text[:pos].count('\n')
ctx_start = max(0, line_index - 5)
ctx_lines = text.splitlines()[ctx_start:line_index + 1]
ctx = "\n".join(ctx_lines)
if not is_safe_to_rename_method(name, ctx):
continue
direct_names.add(name)
open_paren = match.end() - 1
close_paren = _find_matching_delimiter(text, open_paren, '(', ')')
if close_paren == -1:
continue
params = _parse_swift_decl_parameters(text[open_paren + 1:close_paren])
if params is None:
continue
if any(param["label"] != "_" for param in params):
signature_key = _build_swift_signature_key(name, params)
swift_signatures[signature_key] = {
"kind": "swift_signature",
"params": _serialize_swift_params(params),
}
return direct_names, swift_signatures
def _extract_objc_methods_from_text(text: str) -> tuple:
direct_names = set()
objc_selectors = set()
for match in OBJC_METHOD_DECL_RE.finditer(text):
declaration = match.group(1).strip()
selector_parts = re.findall(r'([A-Za-z_][A-Za-z0-9_]*:)', declaration)
if selector_parts:
selector_name = ''.join(selector_parts)
else:
simple_match = re.match(r'([A-Za-z_][A-Za-z0-9_]*)', declaration)
if not simple_match:
continue
selector_name = simple_match.group(1)
pos = match.start()
line_index = text[:pos].count('\n')
ctx_start = max(0, line_index - 5)
ctx_lines = text.splitlines()[ctx_start:line_index + 1]
ctx = "\n".join(ctx_lines)
if selector_name.startswith(METHOD_PREFIX_EXCLUDE):
continue
if not is_safe_to_rename_method(selector_name, ctx):
continue
if ':' in selector_name:
objc_selectors.add(selector_name)
else:
direct_names.add(selector_name)
return direct_names, objc_selectors
def _extract_objc_zero_arg_method_names_from_text(text: str) -> set:
names = set()
for match in OBJC_METHOD_DECL_RE.finditer(text):
declaration = match.group(1).strip()
if ':' in declaration:
continue
simple_match = re.match(r'([A-Za-z_][A-Za-z0-9_]*)', declaration)
if simple_match:
names.add(simple_match.group(1))
return names
def _extract_objc_dot_member_accesses_from_text(text: str) -> set:
names = set()
index = 0
while index < len(text):
if text.startswith("//", index):
index = _skip_line_comment(text, index)
continue
if text.startswith("/*", index):
index = _skip_block_comment(text, index)
continue
char = text[index]
if char in ('"', "'"):
index = _skip_quoted_text(text, index)
continue
if char != '.':
index += 1
continue
prev_index = index - 1
while prev_index >= 0 and text[prev_index].isspace():
prev_index -= 1
if prev_index >= 0 and text[prev_index].isdigit():
index += 1
continue
index += 1
while index < len(text) and text[index].isspace():
index += 1
start = index
if start >= len(text) or not (text[start].isalpha() or text[start] == '_'):
continue
index += 1
while index < len(text) and (text[index].isalnum() or text[index] == '_'):
index += 1
names.add(text[start:index])
return names
def _should_auto_ignore_objc_direct_method(name: str) -> bool:
if not name:
return False
if ALREADY_OBFUSCATED_RE.match(name):
return False
if name in SYMBOL_WHITELIST:
return False
if name in RESERVED_IDENTIFIERS:
return False
if name.startswith(SYMBOL_PREFIX_WHITELIST):
return False
if (
name in COMMON_SYSTEM_MEMBER_NAMES or
name in IOS_SYSTEM_MEMBER_NAMES or
name in THIRD_PARTY_PROTOCOL_METHOD_NAMES or
name in PROJECT_PROPERTY_MEMBER_NAMES
):
return False
for rx in METHOD_EXCLUDE_RE:
if rx.search(name):
return False
return True
def _build_auto_detected_method_ignore_block(names: set) -> str:
lines = [
AUTO_DETECTED_METHOD_IGNORE_BEGIN,
"# Auto-generated: Obj-C zero-arg methods that are referenced with dot syntax.",
"# These names are excluded from direct-method obfuscation to avoid half-rewritten call sites.",
]
lines.extend(sorted(names))
lines.append(AUTO_DETECTED_METHOD_IGNORE_END)
return "\n".join(lines)
def _sync_auto_detected_method_ignore_file(detected_names: set) -> bool:
path = Path(os.getcwd()) / PROJECT_METHOD_IGNORE_FILE_PATH
if not detected_names and not path.exists():
return False
try:
existing_text = path.read_text(encoding="utf-8") if path.exists() else ""
except Exception:
existing_text = ""
block = _build_auto_detected_method_ignore_block(detected_names)
begin_marker = AUTO_DETECTED_METHOD_IGNORE_BEGIN
end_marker = AUTO_DETECTED_METHOD_IGNORE_END
begin_index = existing_text.find(begin_marker)
end_index = existing_text.find(end_marker)
if begin_index != -1 and end_index != -1 and end_index > begin_index:
end_index += len(end_marker)
prefix = existing_text[:begin_index].rstrip()
suffix = existing_text[end_index:].strip()
parts = []
if prefix:
parts.append(prefix)
parts.append(block)
if suffix:
parts.append(suffix)
new_text = "\n\n".join(parts).rstrip() + "\n"
else:
stripped_existing = existing_text.rstrip()
parts = []
if stripped_existing:
parts.append(stripped_existing)
parts.append(block)
new_text = "\n\n".join(parts).rstrip() + "\n"
if new_text == existing_text:
return False
atomic_write_text(str(path), new_text, encoding="utf-8")
return True
def _load_direct_method_aliases_from_map_files() -> dict:
aliases = {}
candidate_paths = []
default_method_map = Path(METHOD_MAP_OUTPUT_PATH)
if default_method_map.exists():
candidate_paths.append(default_method_map)
map_dir = Path("obfuscation_maps")
if map_dir.exists():
candidate_paths.extend(sorted(map_dir.glob("method_map*.json")))
seen_paths = set()
for candidate_path in candidate_paths:
resolved_path = str(candidate_path.resolve())
if resolved_path in seen_paths:
continue
seen_paths.add(resolved_path)
try:
payload = json.loads(candidate_path.read_text(encoding="utf-8"))
except Exception:
continue
if not isinstance(payload, dict):
continue
for original_name, obfuscated_name in payload.items():
if not isinstance(original_name, str) or not isinstance(obfuscated_name, str):
continue
if ':' in original_name:
continue
aliases.setdefault(original_name, set()).add(obfuscated_name)
return aliases
def sync_detected_objc_dot_syntax_method_ignores(scan_dirs: list) -> tuple:
objc_files = collect_objc_files(scan_dirs)
declaration_files = prefilter_files_with_rg(objc_files, METHOD_DECLARATION_RG_PATTERN, chunk_size=RG_CHUNK_SIZE)
dot_access_files = prefilter_files_with_rg(objc_files, OBJC_DOT_SYNTAX_MEMBER_RG_PATTERN, chunk_size=RG_CHUNK_SIZE)
zero_arg_method_names = set()
for file_path in declaration_files:
try:
text = Path(file_path).read_text(encoding="utf-8", errors="ignore")
except Exception:
continue
zero_arg_method_names.update(_extract_objc_zero_arg_method_names_from_text(text))
dot_access_names = set()
for file_path in dot_access_files:
try:
text = Path(file_path).read_text(encoding="utf-8", errors="ignore")
except Exception:
continue
dot_access_names.update(_extract_objc_dot_member_accesses_from_text(text))
detected_names = {
name for name in zero_arg_method_names & dot_access_names
if _should_auto_ignore_objc_direct_method(name)
}
direct_method_aliases = _load_direct_method_aliases_from_map_files()
for original_name, obfuscated_names in direct_method_aliases.items():
if original_name not in dot_access_names:
continue
if not _should_auto_ignore_objc_direct_method(original_name):
continue
if original_name in zero_arg_method_names or zero_arg_method_names.intersection(obfuscated_names):
detected_names.add(original_name)
file_updated = _sync_auto_detected_method_ignore_file(detected_names)
refresh_project_method_ignore_entries()
return detected_names, file_updated
def _enumerate_swift_call_shapes(params: list) -> set:
shapes = set()
def dfs(index: int, labels: list):
if index == len(params):
shapes.add(tuple(labels))
return
param = params[index]
if param.get("has_default", False):
dfs(index + 1, labels)
label = param.get("label")
if isinstance(label, str):
labels.append(label)
dfs(index + 1, labels)
labels.pop()
dfs(0, [])
return shapes
def _find_unsafe_swift_signature_bases(signature_mapping: dict) -> set:
grouped = {}
for signature_key, meta in signature_mapping.items():
base_name, _ = _split_swift_signature_key(signature_key)
if not base_name or not isinstance(meta, dict):
continue
params = meta.get("params", [])
if not isinstance(params, list):
continue
grouped.setdefault(base_name, []).append((signature_key, params))
unsafe_bases = set()
for base_name, entries in grouped.items():
seen_shapes = {}
for signature_key, params in entries:
for shape in _enumerate_swift_call_shapes(params):
previous_signature = seen_shapes.get(shape)
if previous_signature and previous_signature != signature_key:
unsafe_bases.add(base_name)
break
seen_shapes[shape] = signature_key
if base_name in unsafe_bases:
break
return unsafe_bases
def _drop_unsafe_generated_swift_signature_entries(method_mapping: dict):
swift_signature_entries = {
signature_key: meta
for signature_key, meta in method_mapping.items()
if isinstance(meta, dict) and meta.get("kind") == "swift_signature"
}
unsafe_bases = _find_unsafe_swift_signature_bases(swift_signature_entries)
if not unsafe_bases:
return method_mapping, unsafe_bases, []
filtered_mapping = {}
dropped_signatures = []
for key, value in method_mapping.items():
if key in swift_signature_entries:
base_name, _ = _split_swift_signature_key(key)
if base_name in unsafe_bases:
dropped_signatures.append(key)
continue
filtered_mapping[key] = value
return filtered_mapping, unsafe_bases, dropped_signatures
def _drop_unsafe_loaded_swift_signature_entries(swift_signature_mapping: dict):
unsafe_bases = _find_unsafe_swift_signature_bases(swift_signature_mapping)
if not unsafe_bases:
return swift_signature_mapping, unsafe_bases, []
filtered_mapping = {}
dropped_signatures = []
for signature_key, meta in swift_signature_mapping.items():
if meta.get("base_name") in unsafe_bases:
dropped_signatures.append(signature_key)
continue
filtered_mapping[signature_key] = meta
return filtered_mapping, unsafe_bases, dropped_signatures
def extract_methods_from_file(path: str) -> dict:
analysis = _get_cached_file_analysis(path, PROJECT_SOURCE_ANALYSIS_KIND, _analyze_project_source_text)
return {
"direct_names": set(analysis.get("raw_direct_names", [])),
"objc_selectors": set(analysis.get("raw_objc_selectors", [])),
"swift_signatures": dict(analysis.get("raw_swift_signatures", {})),
}
def extract_all_methods(files: list, salt: str) -> dict:
direct_methods = set()
objc_selectors = set()
for f in files:
method_data = extract_methods_from_file(f)
direct_methods.update(method_data["direct_names"])
objc_selectors.update(method_data["objc_selectors"])
direct_candidates = [
name for name in sorted(direct_methods)
if name and not ALREADY_OBFUSCATED_RE.match(name) and name not in SYMBOL_WHITELIST and not is_system_method(name)
]
selector_candidates = [
name for name in sorted(objc_selectors)
if name and not ALREADY_OBFUSCATED_RE.match(name) and name not in SYMBOL_WHITELIST and not is_system_method(name)
]
method_mapping = {}
method_mapping.update(build_unique_mapping(direct_candidates, salt, prefix=METHOD_OBFUSCATED_PREFIX))
method_mapping.update(build_unique_mapping(selector_candidates, salt, prefix=METHOD_OBFUSCATED_PREFIX))
return method_mapping
def is_safe_to_rename(name: str, preceding_lines: str) -> bool:
"""判断一个符号是否可以安全重命名"""
if ALREADY_OBFUSCATED_RE.match(name):
return False
if name in SYMBOL_WHITELIST:
return False
if name in RESERVED_IDENTIFIERS:
return False
if name.startswith(SYMBOL_PREFIX_WHITELIST):
return False
# 如果前几行有 @objc / public / open 等修饰符,跳过
if UNSAFE_MODIFIERS_RE.search(preceding_lines):
return False
return True
def _extract_swift_symbols_from_text(text: str) -> set:
lines = text.splitlines()
symbols = set()
for index, line in enumerate(lines):
for match in DECL_RE.finditer(line):
name = match.group(1)
ctx_start = max(0, index - 5)
ctx = "\n".join(lines[ctx_start:index + 1])
if is_safe_to_rename(name, ctx):
symbols.add(name)
return symbols
def extract_symbols_from_file(path: str) -> set:
analysis = _get_cached_file_analysis(path, PROJECT_SOURCE_ANALYSIS_KIND, _analyze_project_source_text)
return set(analysis.get("swift_symbols", []))
def extract_all_symbols(files: list, salt: str) -> dict:
"""返回 {原始名: 混淆名} 映射"""
all_symbols = set()
candidate_files = prefilter_files_with_rg(files, TYPE_DECLARATION_RG_PATTERN)
for f in candidate_files:
all_symbols.update(extract_symbols_from_file(f))
# 之前这里会根据 `SWIFT_SYMBOL_PREFIXES` 仅保留特定前缀的符号,
# 这会遗漏许多项目中未以特定前缀命名的业务类型。改为不过滤,
# 直接使用提取到的符号集合,后续通过白名单与修饰符检测保护不应混淆的符号。
filtered = set(all_symbols)
return build_unique_mapping(sorted(filtered), salt, prefix=TYPE_OBFUSCATED_PREFIX)
# ── Obj-C 支持 ───────────────────────────────────────────────────────────────
OBJC_INTERFACE_RE = re.compile(r"@interface\s+([A-Za-z_][A-Za-z0-9_]+)")
OBJC_IMPLEMENTATION_RE = re.compile(r"@implementation\s+([A-Za-z_][A-Za-z0-9_]+)")
OBJC_PROTOCOL_RE = re.compile(r"@protocol\s+([A-Za-z_][A-Za-z0-9_]+)")
SELECTOR_STRING_RE = re.compile(r'@"([A-Za-z_][A-Za-z0-9_:@]*)"')
OBJC_SELECTOR_PART_RE = re.compile(r'([A-Za-z_][A-Za-z0-9_]*):')
def extract_objc_symbols_from_file(path: str) -> set:
analysis = _get_cached_file_analysis(path, PROJECT_SOURCE_ANALYSIS_KIND, _analyze_project_source_text)
return set(analysis.get("objc_symbols", []))
def _extract_objc_symbols_from_text(text: str) -> set:
symbols = set()
for match in OBJC_INTERFACE_RE.finditer(text):
symbols.add(match.group(1))
for match in OBJC_IMPLEMENTATION_RE.finditer(text):
symbols.add(match.group(1))
for match in OBJC_PROTOCOL_RE.finditer(text):
symbols.add(match.group(1))
return symbols
def extract_all_objc_symbols(files: list, salt: str) -> dict:
all_symbols = set()
candidate_files = prefilter_files_with_rg(files, TYPE_DECLARATION_RG_PATTERN)
for f in candidate_files:
all_symbols.update(extract_objc_symbols_from_file(f))
# 只保留以指定前缀开头的 Obj-C 符号
filtered = [n for n in sorted(all_symbols) if any(n.startswith(p) for p in OBJC_SYMBOL_PREFIXES)]
candidates = []
for name in filtered:
if not ALREADY_OBFUSCATED_RE.match(name) and name not in SYMBOL_WHITELIST and not name.startswith(SYMBOL_PREFIX_WHITELIST):
candidates.append(name)
return build_unique_mapping(candidates, salt, prefix=TYPE_OBFUSCATED_PREFIX)
def _collect_selector_literals_from_text(text: str) -> set:
selectors = set()
for match in SELECTOR_STRING_RE.finditer(text):
value = match.group(1)
if ':' in value or re.match(r'^[a-z]', value):
selectors.add(value)
return selectors
def collect_selector_literals(files: list) -> set:
sels = set()
candidate_files = prefilter_files_with_rg(files, SELECTOR_LITERAL_RG_PATTERN)
for f in candidate_files:
analysis = _get_cached_file_analysis(f, PROJECT_SOURCE_ANALYSIS_KIND, _analyze_project_source_text)
sels.update(analysis.get("selector_literals", []))
return sels
# ── 替换阶段 ─────────────────────────────────────────────────────────────────
def build_mapping_regex(mapping: dict) -> re.Pattern:
"""构建一次性替换正则,避免 O(符号数 x 文件数) 的重复扫描。"""
if not mapping:
return re.compile(r"(?!x)x")
# 按长度降序可减少共有前缀时的回溯成本
# Exclude reserved identifiers from replacement regex to avoid accidental replacement
keys = [k for k in mapping.keys() if k not in RESERVED_IDENTIFIERS]
keys = sorted(keys, key=len, reverse=True)
body = "|".join(re.escape(k) for k in keys)
return re.compile(r"(?<![A-Za-z0-9_])(" + body + r")(?![A-Za-z0-9_])")
def build_aho_automaton(mapping: dict):
"""尝试构建 pyahocorasick 自动机以进行单次线性替换;如不可用则返回 None。"""
if not mapping:
return None
try:
import ahocorasick
except Exception:
return None
A = ahocorasick.Automaton()
for key in mapping.keys():
if key in RESERVED_IDENTIFIERS:
continue
# store key as value to identify match
A.add_word(key, key)
A.make_automaton()
return A
def _iter_aho_matches(text: str, aho) -> list:
"""返回满足词边界且不重叠的 aho 匹配结果。"""
if aho is None or not text:
return []
matches = []
text_length = len(text)
for end_index, key in aho.iter(text):
start = end_index - len(key) + 1
if start > 0 and (text[start - 1].isalnum() or text[start - 1] == '_'):
continue
if end_index + 1 < text_length and (text[end_index + 1].isalnum() or text[end_index + 1] == '_'):
continue
matches.append((start, end_index, key))
if not matches:
return []
matches.sort(key=lambda item: (item[0], -(item[1] - item[0])))
chosen = []
last_end = -1
for start, end, key in matches:
if start <= last_end:
continue
chosen.append((start, end, key))
last_end = end
return chosen
def split_objc_selector(selector: str) -> list:
return [f"{match.group(1)}:" for match in OBJC_SELECTOR_PART_RE.finditer(selector)]
def _build_swift_alias_from_objc_selector(selector: str, obfuscated_token: str):
selector_parts = split_objc_selector(selector)
if not selector_parts:
return None
base_name = selector_parts[0][:-1]
obfuscated_base_name = f"{obfuscated_token}_1"
signature_params = []
alias_params = []
for index, selector_part in enumerate(selector_parts, start=1):
label = "_" if index == 1 else selector_part[:-1]
signature_param = {
"label": label,
"has_default": False,
"allows_trailing_closure": False,
}
signature_params.append(signature_param)
alias_param = dict(signature_param)
alias_param["replacement"] = None if label == "_" else f"{obfuscated_token}_{index}"
alias_params.append(alias_param)
return {
"base_name": base_name,
"obfuscated_base_name": obfuscated_base_name,
"signature_key": _build_swift_signature_key(base_name, signature_params),
"meta": {
"base_name": base_name,
"obfuscated": obfuscated_base_name,
"params": alias_params,
},
}
def _record_alias_mapping_entry(bucket: dict, conflicts: set, key: str, value):
if key in conflicts:
return
existing = bucket.get(key)
if existing is not None and existing != value:
bucket.pop(key, None)
conflicts.add(key)
return
bucket[key] = value
def load_method_obfuscation_maps(map_path: str) -> tuple:
method_mapping = json.loads(Path(map_path).read_text(encoding='utf-8'))
direct_mapping = {}
objc_selector_mapping = {}
swift_signature_mapping = {}
objc_swift_direct_aliases = {}
objc_swift_signature_aliases = {}
objc_swift_direct_conflicts = set()
objc_swift_signature_conflicts = set()
for original, obfuscated in method_mapping.items():
# 跳过语言保留标识符(例如 Obj-C/Swift 的 `self`),避免将关键字替换为未声明标识符
if isinstance(original, str) and original in RESERVED_IDENTIFIERS:
continue
if isinstance(obfuscated, dict):
if obfuscated.get("kind") != "swift_signature":
continue
# 原生 Swift 方法只混淆基名,不再混淆参数标签。
# 这里显式忽略旧 method_map 中遗留的 swift_signature 条目,
# 避免半混淆源码再次应用时出现“基名已变、标签没跟上”的不一致。
continue
if not isinstance(obfuscated, str):
continue
if is_system_method(original):
continue
selector_parts = split_objc_selector(original)
if selector_parts:
# 如果 selector 的某一部分是保留标识符(例如 selector 包含 "self:"),跳过
parts_ok = True
for p in selector_parts:
name = p[:-1]
if name in RESERVED_IDENTIFIERS:
parts_ok = False
break
if not parts_ok:
continue
objc_selector_mapping[original] = [
f"{obfuscated}_{idx}:"
for idx in range(1, len(selector_parts) + 1)
]
swift_alias = _build_swift_alias_from_objc_selector(original, obfuscated)
if swift_alias is not None:
_record_alias_mapping_entry(
objc_swift_direct_aliases,
objc_swift_direct_conflicts,
swift_alias["base_name"],
swift_alias["obfuscated_base_name"],
)
_record_alias_mapping_entry(
objc_swift_signature_aliases,
objc_swift_signature_conflicts,
swift_alias["signature_key"],
swift_alias["meta"],
)
else:
# 直接名称(非 selector也要避免保留标识符
if original in RESERVED_IDENTIFIERS:
continue
direct_mapping[original] = obfuscated
for alias_name, alias_obfuscated in objc_swift_direct_aliases.items():
existing = direct_mapping.get(alias_name)
if existing is None or existing == alias_obfuscated:
direct_mapping[alias_name] = alias_obfuscated
for signature_key, signature_meta in objc_swift_signature_aliases.items():
existing = swift_signature_mapping.get(signature_key)
if existing is None or existing == signature_meta:
swift_signature_mapping[signature_key] = signature_meta
swift_signature_mapping, unsafe_bases, dropped_signatures = _drop_unsafe_loaded_swift_signature_entries(swift_signature_mapping)
if dropped_signatures:
preview = ", ".join(sorted(unsafe_bases)[:5])
suffix = " ..." if len(unsafe_bases) > 5 else ""
print(
f"⚠️ 跳过 {len(dropped_signatures)} 个存在重叠调用形态的 Swift 参数标签映射,"
f"涉及 {len(unsafe_bases)} 个方法基名: {preview}{suffix}"
)
return direct_mapping, objc_selector_mapping, swift_signature_mapping
def build_selector_display_mapping(objc_selector_mapping: dict) -> dict:
return {
original: "".join(parts)
for original, parts in objc_selector_mapping.items()
}
def build_swift_signature_display_mapping(direct_method_mapping: dict, swift_signature_mapping: dict) -> dict:
display = {}
for original, meta in swift_signature_mapping.items():
base_name = meta.get("base_name", "")
obfuscated_base = direct_method_mapping.get(base_name, base_name)
label_parts = []
for param in meta.get("params", []):
label = param.get("label")
if label == "_":
label_parts.append("_:")
continue
replacement = param.get("replacement", label)
label_parts.append(f"{replacement}:")
display[original] = f"{obfuscated_base}({''.join(label_parts)})"
return display
def _replace_identifier_spans(text: str, spans: list, replacements: list) -> str:
out = []
cursor = 0
for (start, end), replacement in zip(spans, replacements):
out.append(text[cursor:start])
out.append(replacement)
cursor = end
out.append(text[cursor:])
return "".join(out)
def _build_swift_decl_param_edit(param: dict, replacement: str):
label = param.get("label")
label_span = param.get("span")
if not label_span or not replacement:
return None
local_name = param.get("local_name")
local_span = param.get("local_span")
if local_name is None:
if replacement == label:
return None
return label_span, f"{replacement} {label}"
if local_span and local_name == replacement and label != replacement:
return (label_span[0], local_span[1]), replacement
if replacement == label:
return None
return label_span, replacement
def _skip_quoted_text(text: str, start: int) -> int:
quote = text[start]
index = start + 1
while index < len(text):
if text[index] == '\\':
index += 2
continue
if text[index] == quote:
return index + 1
index += 1
return len(text)
def _rewrite_swift_string_literal(text: str, start: int, rewriter) -> tuple:
quote = text[start]
index = start + 1
chunk_start = index
chunks = [quote]
while index < len(text):
if text.startswith("\\(", index):
open_paren = index + 1
close_paren = _find_matching_delimiter(text, open_paren, '(', ')')
if close_paren == -1:
break
chunks.append(text[chunk_start:index + 2])
chunks.append(rewriter(text[index + 2:close_paren]))
chunks.append(')')
index = close_paren + 1
chunk_start = index
continue
if text[index] == '\\':
index += 2
continue
if text[index] == quote:
chunks.append(text[chunk_start:index + 1])
return "".join(chunks), index + 1
index += 1
chunks.append(text[chunk_start:])
return "".join(chunks), len(text)
def _skip_line_comment(text: str, start: int) -> int:
end = text.find("\n", start)
return len(text) if end == -1 else end
def _skip_block_comment(text: str, start: int) -> int:
end = text.find("*/", start + 2)
return len(text) if end == -1 else end + 2
def _find_matching_square_bracket(text: str, start: int) -> int:
return _find_matching_delimiter(text, start, '[', ']')
def _collect_top_level_selector_parts(text: str) -> list:
parts = []
paren_depth = 0
brace_depth = 0
bracket_depth = 0
index = 0
while index < len(text):
if text.startswith("//", index):
index = _skip_line_comment(text, index)
continue
if text.startswith("/*", index):
index = _skip_block_comment(text, index)
continue
char = text[index]
if char in ('"', "'"):
index = _skip_quoted_text(text, index)
continue
if char == '(':
paren_depth += 1
index += 1
continue
if char == ')' and paren_depth > 0:
paren_depth -= 1
index += 1
continue
if char == '{':
brace_depth += 1
index += 1
continue
if char == '}' and brace_depth > 0:
brace_depth -= 1
index += 1
continue
if char == '[':
bracket_depth += 1
index += 1
continue
if char == ']' and bracket_depth > 0:
bracket_depth -= 1
index += 1
continue
if paren_depth == 0 and brace_depth == 0 and bracket_depth == 0 and (char.isalpha() or char == '_'):
start = index
index += 1
while index < len(text) and (text[index].isalnum() or text[index] == '_'):
index += 1
if index < len(text) and text[index] == ':':
parts.append((f"{text[start:index]}:", (start, index)))
index += 1
continue
continue
index += 1
return parts
def _rewrite_selector_occurrence(text: str, selector: str, replacement_parts: list, stats: dict) -> str:
replacement_names = [part[:-1] for part in replacement_parts]
full_replacement = "".join(replacement_parts)
def replace_method_decls(src: str) -> str:
chunks = []
cursor = 0
changed = 0
for match in OBJC_METHOD_DECL_RE.finditer(src):
declaration = match.group(1)
label_matches = list(OBJC_SELECTOR_PART_RE.finditer(declaration))
if not label_matches:
continue
full_selector = "".join(f"{item.group(1)}:" for item in label_matches)
if full_selector != selector or len(label_matches) != len(replacement_names):
continue
spans = [item.span(1) for item in label_matches]
new_declaration = _replace_identifier_spans(declaration, spans, replacement_names)
if new_declaration == declaration:
continue
start, end = match.span(1)
chunks.append(src[cursor:start])
chunks.append(new_declaration)
cursor = end
changed += 1
if not changed:
return src
chunks.append(src[cursor:])
stats[selector] = stats.get(selector, 0) + changed
return "".join(chunks)
def replace_selector_literals(src: str) -> str:
pattern = re.compile(rf"(@selector\s*\(\s*){re.escape(selector)}(\s*\))")
def repl_selector(match: re.Match) -> str:
stats[selector] = stats.get(selector, 0) + 1
return f"{match.group(1)}{full_replacement}{match.group(2)}"
src = pattern.sub(repl_selector, src)
string_pattern = re.compile(r'@"' + re.escape(selector) + r'"')
def repl_string(_: re.Match) -> str:
stats[selector] = stats.get(selector, 0) + 1
return f'@"{full_replacement}"'
return string_pattern.sub(repl_string, src)
def replace_message_sends(src: str) -> str:
chunks = []
index = 0
changed = 0
while index < len(src):
if src.startswith("//", index):
end = _skip_line_comment(src, index)
chunks.append(src[index:end])
index = end
continue
if src.startswith("/*", index):
end = _skip_block_comment(src, index)
chunks.append(src[index:end])
index = end
continue
char = src[index]
if char in ('"', "'"):
end = _skip_quoted_text(src, index)
chunks.append(src[index:end])
index = end
continue
if char != '[':
chunks.append(char)
index += 1
continue
end = _find_matching_square_bracket(src, index)
if end == -1:
chunks.append(src[index:])
break
segment = src[index:end + 1]
inner = replace_message_sends(segment[1:-1])
selector_parts = _collect_top_level_selector_parts(inner)
if selector_parts:
full_selector = "".join(part for part, _ in selector_parts)
if full_selector == selector and len(selector_parts) == len(replacement_names):
spans = [span for _, span in selector_parts]
inner = _replace_identifier_spans(inner, spans, replacement_names)
changed += 1
chunks.append(f"[{inner}]")
index = end + 1
if changed:
stats[selector] = stats.get(selector, 0) + changed
return "".join(chunks)
result = text
result = replace_method_decls(result)
result = replace_message_sends(result)
result = replace_selector_literals(result)
return result
def apply_objc_selector_mapping(text: str, objc_selector_mapping: dict, stats: dict) -> str:
if not objc_selector_mapping:
return text
if '@selector' not in text and ':' not in text and '@"' not in text:
return text
result = text
for selector, replacement_parts in sorted(objc_selector_mapping.items(), key=lambda item: -len(item[0])):
probe = selector.split(':', 1)[0]
if probe and probe not in result:
continue
result = _rewrite_selector_occurrence(result, selector, replacement_parts, stats)
return result
def _rewrite_objc_direct_method_occurrence(text: str, original_name: str, replacement_name: str, stats: dict) -> str:
def replace_method_decls(src: str) -> str:
chunks = []
cursor = 0
changed = 0
for match in OBJC_METHOD_DECL_RE.finditer(src):
declaration = match.group(1)
if ':' in declaration:
continue
simple_match = re.match(r'\s*([A-Za-z_][A-Za-z0-9_]*)', declaration)
if not simple_match or simple_match.group(1) != original_name:
continue
start, _ = match.span(1)
name_start = start + simple_match.start(1)
name_end = start + simple_match.end(1)
chunks.append(src[cursor:name_start])
chunks.append(replacement_name)
cursor = name_end
changed += 1
if not changed:
return src
chunks.append(src[cursor:])
stats[original_name] = stats.get(original_name, 0) + changed
return "".join(chunks)
def replace_selector_literals(src: str) -> str:
selector_pattern = re.compile(rf"(@selector\s*\(\s*){re.escape(original_name)}(\s*\))")
def repl_selector(match: re.Match) -> str:
stats[original_name] = stats.get(original_name, 0) + 1
return f"{match.group(1)}{replacement_name}{match.group(2)}"
src = selector_pattern.sub(repl_selector, src)
string_pattern = re.compile(r'@"' + re.escape(original_name) + r'"')
def repl_string(_: re.Match) -> str:
stats[original_name] = stats.get(original_name, 0) + 1
return f'@"{replacement_name}"'
return string_pattern.sub(repl_string, src)
def replace_message_sends(src: str) -> str:
chunks = []
index = 0
changed = 0
while index < len(src):
if src.startswith("//", index):
end = _skip_line_comment(src, index)
chunks.append(src[index:end])
index = end
continue
if src.startswith("/*", index):
end = _skip_block_comment(src, index)
chunks.append(src[index:end])
index = end
continue
char = src[index]
if char in ('"', "'"):
end = _skip_quoted_text(src, index)
chunks.append(src[index:end])
index = end
continue
if char != '[':
chunks.append(char)
index += 1
continue
end = _find_matching_square_bracket(src, index)
if end == -1:
chunks.append(src[index:])
break
segment = src[index:end + 1]
inner = replace_message_sends(segment[1:-1])
if not _collect_top_level_selector_parts(inner):
simple_match = re.match(r'^(?P<receiver>.+\s)(?P<name>[A-Za-z_][A-Za-z0-9_]*)\s*$', inner, re.S)
if simple_match and simple_match.group("name") == original_name:
inner = f"{simple_match.group('receiver')}{replacement_name}"
changed += 1
chunks.append(f"[{inner}]")
index = end + 1
if changed:
stats[original_name] = stats.get(original_name, 0) + changed
return "".join(chunks)
result = text
result = replace_method_decls(result)
result = replace_message_sends(result)
result = replace_selector_literals(result)
return result
def apply_objc_direct_method_mapping(text: str, direct_method_mapping: dict, stats: dict) -> str:
if not direct_method_mapping:
return text
result = text
for original_name, replacement_name in sorted(direct_method_mapping.items(), key=lambda item: -len(item[0])):
if original_name not in result:
continue
result = _rewrite_objc_direct_method_occurrence(result, original_name, replacement_name, stats)
return result
def _parse_swift_call_arguments(arguments_text: str) -> list:
arguments = []
for segment, start, _ in _split_top_level_segments(arguments_text, separator=',', track_angles=False):
if not segment.strip():
continue
colon_index = _find_top_level_char(segment, ':', track_angles=False)
if colon_index == -1:
arguments.append({
"raw_label": None,
"span": None,
})
continue
prefix = segment[:colon_index]
match = SWIFT_ARGUMENT_LABEL_RE.match(prefix)
if not match:
arguments.append({
"raw_label": None,
"span": None,
})
continue
raw_label = match.group("label")
label_span = None
if raw_label != "_":
label_span = (start + match.start("label"), start + match.end("label"))
arguments.append({
"raw_label": raw_label,
"span": label_span,
})
return arguments
def _skip_swift_call_whitespace(text: str, index: int) -> int:
while index < len(text) and text[index].isspace():
index += 1
return index
def _find_swift_call_paren(text: str, index: int):
lookahead = _skip_swift_call_whitespace(text, index)
if lookahead < len(text) and text[lookahead] in ('?', '!'):
after_marker = _skip_swift_call_whitespace(text, lookahead + 1)
if after_marker < len(text) and text[after_marker] == '(':
return after_marker
if lookahead < len(text) and text[lookahead] == '(':
return lookahead
return None
def _find_swift_trailing_closure_start(text: str, index: int):
lookahead = _skip_swift_call_whitespace(text, index)
if lookahead < len(text) and text[lookahead] == '{':
return lookahead
return None
def _swift_has_explicit_member_receiver(text: str, identifier_start: int) -> bool:
dot_index = identifier_start - 1
while dot_index >= 0 and text[dot_index].isspace():
dot_index -= 1
if dot_index < 0 or text[dot_index] != '.':
return True
receiver_index = dot_index - 1
while receiver_index >= 0 and text[receiver_index].isspace():
receiver_index -= 1
while receiver_index >= 0 and text[receiver_index] in ('?', '!'):
receiver_index -= 1
while receiver_index >= 0 and text[receiver_index].isspace():
receiver_index -= 1
if receiver_index < 0:
return False
receiver_char = text[receiver_index]
return receiver_char.isalnum() or receiver_char == '_' or receiver_char in (')', ']', '}')
def _swift_statement_prefix(text: str, identifier_start: int) -> str:
statement_start = identifier_start
while statement_start > 0:
previous = text[statement_start - 1]
if previous in '\n;{}':
break
statement_start -= 1
return text[statement_start:identifier_start].lstrip()
def _swift_trailing_brace_belongs_to_control_flow(text: str, identifier_start: int) -> bool:
prefix = _swift_statement_prefix(text, identifier_start)
return prefix.startswith(("if ", "guard ", "while ", "for ", "switch "))
def _match_swift_call_arguments(signature_params: list, call_arguments: list, has_trailing_closure: bool = False):
matches = []
def can_bind_trailing_closure(remaining_params: list) -> bool:
if not remaining_params:
return False
trailing_param = remaining_params[-1]
if not trailing_param.get("allows_trailing_closure", False):
return False
return all(param.get("has_default", False) for param in remaining_params[:-1])
def dfs(param_index: int, arg_index: int, alignment: list):
if len(matches) > 1:
return
if arg_index == len(call_arguments):
remaining_params = signature_params[param_index:]
if has_trailing_closure:
if can_bind_trailing_closure(remaining_params):
matches.append(list(alignment))
elif all(param.get("has_default", False) for param in remaining_params):
matches.append(list(alignment))
return
if param_index == len(signature_params):
return
signature_param = signature_params[param_index]
if signature_param.get("has_default", False):
dfs(param_index + 1, arg_index, alignment)
if len(matches) > 1:
return
argument = call_arguments[arg_index]
expected_label = signature_param.get("label")
raw_label = argument.get("raw_label")
is_match = (
(expected_label == "_" and raw_label is None) or
(expected_label != "_" and raw_label == expected_label)
)
if is_match:
alignment.append((arg_index, param_index))
dfs(param_index + 1, arg_index + 1, alignment)
alignment.pop()
dfs(0, 0, [])
if len(matches) != 1:
return None
return matches[0]
def _replace_swift_call_argument_labels(arguments_text: str, call_arguments: list, alignment: list, signature_params: list) -> str:
spans = []
replacements = []
for argument_index, param_index in alignment:
argument = call_arguments[argument_index]
signature_param = signature_params[param_index]
span = argument.get("span")
replacement = signature_param.get("replacement")
if span and replacement:
spans.append(span)
replacements.append(replacement)
if not spans:
return arguments_text
return _replace_identifier_spans(arguments_text, spans, replacements)
def _group_swift_signature_mapping(swift_signature_mapping: dict, direct_method_mapping: dict = None) -> dict:
grouped = {}
direct_method_mapping = direct_method_mapping or {}
for signature_key, meta in swift_signature_mapping.items():
entry = {
"signature_key": signature_key,
"base_name": meta.get("base_name"),
"params": meta.get("params", []),
}
grouped.setdefault(entry["base_name"], []).append(entry)
obfuscated_base = direct_method_mapping.get(entry["base_name"])
if obfuscated_base and obfuscated_base != entry["base_name"]:
grouped.setdefault(obfuscated_base, []).append(entry)
return grouped
def _replace_swift_method_declarations(text: str, swift_signature_mapping: dict, stats: dict) -> str:
chunks = []
cursor = 0
for match in SWIFT_METHOD_DECL_RE.finditer(text):
base_name = match.group(1)
open_paren = match.end() - 1
close_paren = _find_matching_delimiter(text, open_paren, '(', ')')
if close_paren == -1:
continue
params_text = text[open_paren + 1:close_paren]
params = _parse_swift_decl_parameters(params_text)
if params is None:
continue
signature_key = _build_swift_signature_key(base_name, params)
meta = swift_signature_mapping.get(signature_key)
if not meta:
continue
spans = []
replacements = []
for index, param in enumerate(params):
replacement = meta["params"][index].get("replacement") if index < len(meta.get("params", [])) else None
edit = _build_swift_decl_param_edit(param, replacement)
if edit:
span, replacement_text = edit
spans.append(span)
replacements.append(replacement_text)
if not spans:
continue
new_params_text = _replace_identifier_spans(params_text, spans, replacements)
chunks.append(text[cursor:open_paren + 1])
chunks.append(new_params_text)
cursor = close_paren
stats[signature_key] = stats.get(signature_key, 0) + 1
if not chunks:
return text
chunks.append(text[cursor:])
return "".join(chunks)
def _resolve_swift_call_candidate(candidates: list, call_arguments: list, has_trailing_closure: bool = False):
matches = []
for candidate in candidates:
alignment = _match_swift_call_arguments(candidate.get("params", []), call_arguments, has_trailing_closure=has_trailing_closure)
if alignment is not None:
params = candidate.get("params", [])
trailing_param_count = 1 if has_trailing_closure else 0
skipped_default_count = max(0, len(params) - len(alignment))
score = (
1 if len(params) == len(call_arguments) + trailing_param_count else 0,
-(skipped_default_count - trailing_param_count),
)
matches.append((candidate, alignment, score))
if not matches:
return None
best_score = max(match[2] for match in matches)
best_matches = [match for match in matches if match[2] == best_score]
if len(best_matches) != 1:
return None
candidate, alignment, _ = best_matches[0]
return {
"meta": candidate,
"alignment": alignment,
}
def _replace_swift_call_sites(text: str, swift_signature_mapping: dict, stats: dict, direct_method_mapping: dict = None) -> str:
grouped = _group_swift_signature_mapping(swift_signature_mapping, direct_method_mapping)
local_param_scopes = _collect_swift_local_param_scopes(text)
def rewrite(src: str, base_offset: int = 0) -> str:
chunks = []
index = 0
while index < len(src):
if src.startswith("//", index):
end = _skip_line_comment(src, index)
chunks.append(src[index:end])
index = end
continue
if src.startswith("/*", index):
end = _skip_block_comment(src, index)
chunks.append(src[index:end])
index = end
continue
char = src[index]
if char in ('"', "'"):
rewritten_literal, end = _rewrite_swift_string_literal(
src,
index,
lambda inner, offset=base_offset + index + 2: rewrite(inner, offset),
)
chunks.append(rewritten_literal)
index = end
continue
if not (char.isalpha() or char == '_'):
chunks.append(char)
index += 1
continue
start = index
index += 1
while index < len(src) and (src[index].isalnum() or src[index] == '_'):
index += 1
identifier = src[start:index]
if (
identifier not in grouped or
_find_previous_identifier(src, start) == "func" or
_swift_is_shadowed_local_param(text, base_offset + start, identifier, local_param_scopes)
):
chunks.append(src[start:index])
continue
open_paren = _find_swift_call_paren(src, index)
if open_paren is None:
chunks.append(src[start:index])
continue
close_paren = _find_matching_delimiter(src, open_paren, '(', ')')
if close_paren == -1:
chunks.append(src[start:index])
continue
inner = rewrite(src[open_paren + 1:close_paren], base_offset + open_paren + 1)
call_arguments = _parse_swift_call_arguments(inner)
has_trailing_closure = (
_find_swift_trailing_closure_start(src, close_paren + 1) is not None and
not _swift_trailing_brace_belongs_to_control_flow(src, start)
)
matched = _resolve_swift_call_candidate(grouped[identifier], call_arguments, has_trailing_closure=has_trailing_closure)
if matched is not None:
inner = _replace_swift_call_argument_labels(
inner,
call_arguments,
matched["alignment"],
matched["meta"].get("params", []),
)
signature_key = matched["meta"]["signature_key"]
stats[signature_key] = stats.get(signature_key, 0) + 1
chunks.append(src[start:open_paren + 1])
chunks.append(inner)
chunks.append(')')
index = close_paren + 1
return "".join(chunks)
return rewrite(text)
def _replace_swift_direct_method_declarations(text: str, direct_method_mapping: dict, stats: dict) -> str:
chunks = []
cursor = 0
for match in SWIFT_METHOD_DECL_RE.finditer(text):
base_name = match.group(1)
replacement = direct_method_mapping.get(base_name)
if not replacement:
continue
chunks.append(text[cursor:match.start(1)])
chunks.append(replacement)
cursor = match.end(1)
stats[base_name] = stats.get(base_name, 0) + 1
if not chunks:
return text
chunks.append(text[cursor:])
return "".join(chunks)
def _replace_swift_direct_call_sites(text: str, direct_method_mapping: dict, stats: dict) -> str:
local_param_scopes = _collect_swift_local_param_scopes(text)
def rewrite(src: str, base_offset: int = 0) -> str:
chunks = []
index = 0
while index < len(src):
if src.startswith("//", index):
end = _skip_line_comment(src, index)
chunks.append(src[index:end])
index = end
continue
if src.startswith("/*", index):
end = _skip_block_comment(src, index)
chunks.append(src[index:end])
index = end
continue
char = src[index]
if char in ('"', "'"):
rewritten_literal, end = _rewrite_swift_string_literal(
src,
index,
lambda inner, offset=base_offset + index + 2: rewrite(inner, offset),
)
chunks.append(rewritten_literal)
index = end
continue
if not (char.isalpha() or char == '_'):
chunks.append(char)
index += 1
continue
start = index
index += 1
while index < len(src) and (src[index].isalnum() or src[index] == '_'):
index += 1
identifier = src[start:index]
if (
identifier not in direct_method_mapping or
_find_previous_identifier(src, start) == "func" or
_swift_is_shadowed_local_param(text, base_offset + start, identifier, local_param_scopes)
):
chunks.append(src[start:index])
continue
open_paren = _find_swift_call_paren(src, index)
if open_paren is not None:
close_paren = _find_matching_delimiter(src, open_paren, '(', ')')
if close_paren == -1:
chunks.append(src[start:index])
continue
inner = rewrite(src[open_paren + 1:close_paren], base_offset + open_paren + 1)
chunks.append(direct_method_mapping[identifier])
chunks.append(src[index:open_paren + 1])
chunks.append(inner)
chunks.append(')')
stats[identifier] = stats.get(identifier, 0) + 1
index = close_paren + 1
continue
trailing_closure_start = _find_swift_trailing_closure_start(src, index)
if (
trailing_closure_start is not None and
_swift_has_explicit_member_receiver(src, start) and
not _swift_trailing_brace_belongs_to_control_flow(src, start)
):
chunks.append(direct_method_mapping[identifier])
chunks.append(src[index:trailing_closure_start])
stats[identifier] = stats.get(identifier, 0) + 1
index = trailing_closure_start
continue
chunks.append(src[start:index])
return "".join(chunks)
return rewrite(text)
def _replace_swift_direct_selector_references(text: str, direct_method_mapping: dict, stats: dict) -> str:
selector_ref_re = re.compile(
r"^(?P<prefix>\s*(?:(?:getter|setter)\s*:\s*)?(?:(?:[A-Za-z_][A-Za-z0-9_]*)\s*\.\s*)*)(?P<base>[A-Za-z_][A-Za-z0-9_]*)\s*$",
re.S,
)
chunks = []
index = 0
while index < len(text):
if text.startswith("//", index):
end = _skip_line_comment(text, index)
chunks.append(text[index:end])
index = end
continue
if text.startswith("/*", index):
end = _skip_block_comment(text, index)
chunks.append(text[index:end])
index = end
continue
char = text[index]
if char in ('"', "'"):
end = _skip_quoted_text(text, index)
chunks.append(text[index:end])
index = end
continue
if not text.startswith("#selector", index):
chunks.append(char)
index += 1
continue
start = index
after_keyword = index + len("#selector")
while after_keyword < len(text) and text[after_keyword].isspace():
after_keyword += 1
if after_keyword >= len(text) or text[after_keyword] != '(':
chunks.append(text[index:after_keyword])
index = after_keyword
continue
close_paren = _find_matching_delimiter(text, after_keyword, '(', ')')
if close_paren == -1:
chunks.append(text[index:])
break
inner = text[after_keyword + 1:close_paren]
new_inner = inner
match = selector_ref_re.match(inner)
if match:
base_name = match.group("base")
replacement = direct_method_mapping.get(base_name)
if replacement:
new_inner = f"{match.group('prefix')}{replacement}"
stats[base_name] = stats.get(base_name, 0) + 1
chunks.append(text[start:after_keyword + 1])
chunks.append(new_inner)
chunks.append(')')
index = close_paren + 1
return "".join(chunks)
def apply_swift_direct_method_mapping(text: str, direct_method_mapping: dict, stats: dict) -> str:
if not direct_method_mapping:
return text
result = text
result = _replace_swift_direct_method_declarations(result, direct_method_mapping, stats)
result = _replace_swift_direct_call_sites(result, direct_method_mapping, stats)
result = _replace_swift_direct_selector_references(result, direct_method_mapping, stats)
return result
def _replace_swift_selector_references(text: str, swift_signature_mapping: dict, stats: dict) -> str:
chunks = []
index = 0
selector_ref_re = re.compile(
r"^(?P<prefix>\s*(?:(?:[A-Za-z_][A-Za-z0-9_]*)\s*\.\s*)*)(?P<base>[A-Za-z_][A-Za-z0-9_]*)\((?P<labels>.*)\)\s*$",
re.S,
)
while index < len(text):
if text.startswith("//", index):
end = _skip_line_comment(text, index)
chunks.append(text[index:end])
index = end
continue
if text.startswith("/*", index):
end = _skip_block_comment(text, index)
chunks.append(text[index:end])
index = end
continue
char = text[index]
if char in ('"', "'"):
end = _skip_quoted_text(text, index)
chunks.append(text[index:end])
index = end
continue
if not text.startswith("#selector", index):
chunks.append(char)
index += 1
continue
start = index
after_keyword = index + len("#selector")
while after_keyword < len(text) and text[after_keyword].isspace():
after_keyword += 1
if after_keyword >= len(text) or text[after_keyword] != '(':
chunks.append(text[index:after_keyword])
index = after_keyword
continue
close_paren = _find_matching_delimiter(text, after_keyword, '(', ')')
if close_paren == -1:
chunks.append(text[index:])
break
inner = text[after_keyword + 1:close_paren]
new_inner = inner
match = selector_ref_re.match(inner)
if match:
base_name = match.group("base")
labels_text = match.group("labels")
label_matches = list(re.finditer(r'([A-Za-z_][A-Za-z0-9_]*|_):', labels_text))
if label_matches:
signature_key = f"{base_name}({''.join(f'{item.group(1)}:' for item in label_matches)})"
meta = swift_signature_mapping.get(signature_key)
if meta and len(label_matches) == len(meta.get("params", [])):
spans = []
replacements = []
for idx, label_match in enumerate(label_matches):
replacement = meta["params"][idx].get("replacement")
if replacement and label_match.group(1) != "_":
spans.append(label_match.span(1))
replacements.append(replacement)
if spans:
replaced_labels = _replace_identifier_spans(labels_text, spans, replacements)
new_inner = f"{match.group('prefix')}{base_name}({replaced_labels})"
stats[signature_key] = stats.get(signature_key, 0) + 1
chunks.append(text[start:after_keyword + 1])
chunks.append(new_inner)
chunks.append(')')
index = close_paren + 1
return "".join(chunks)
def apply_swift_signature_mapping(text: str, swift_signature_mapping: dict, stats: dict, direct_method_mapping: dict = None) -> str:
if not swift_signature_mapping:
return text
result = text
result = _replace_swift_method_declarations(result, swift_signature_mapping, stats)
result = _replace_swift_call_sites(result, swift_signature_mapping, stats, direct_method_mapping)
result = _replace_swift_selector_references(result, swift_signature_mapping, stats)
return result
def _record_reverse_mapping_entry(bucket: dict, key: str, value, conflicts: dict):
existing = bucket.get(key)
if existing is not None and existing != value:
conflicts.setdefault(key, set()).update({
json.dumps(existing, ensure_ascii=False, sort_keys=True),
json.dumps(value, ensure_ascii=False, sort_keys=True),
})
bucket.pop(key, None)
return
bucket[key] = value
def _build_reverse_swift_signature_entry(original_signature: str, obfuscated_meta: dict):
if obfuscated_meta.get("kind") != "swift_signature":
return None
base_name, _ = _split_swift_signature_key(original_signature)
obfuscated_token = obfuscated_meta.get("obfuscated")
if not base_name or not isinstance(obfuscated_token, str):
return None
reverse_signature_params = []
reverse_params = []
for index, param in enumerate(obfuscated_meta.get("params", []), start=1):
if not isinstance(param, dict):
return None
original_label = param.get("label")
if not isinstance(original_label, str):
return None
current_label = "_" if original_label == "_" else f"{obfuscated_token}_{index}"
has_default = bool(param.get("has_default", False))
reverse_signature_params.append({
"label": current_label,
"has_default": has_default,
"allows_trailing_closure": bool(param.get("allows_trailing_closure", False)),
})
reverse_params.append({
"label": current_label,
"has_default": has_default,
"allows_trailing_closure": bool(param.get("allows_trailing_closure", False)),
"replacement": None if original_label == "_" else original_label,
})
reverse_signature_key = _build_swift_signature_key(base_name, reverse_signature_params)
return reverse_signature_key, {
"base_name": base_name,
"obfuscated": obfuscated_token,
"params": reverse_params,
}
def load_reverse_mappings_from_files(map_files: list) -> tuple:
reverse = {}
reverse_objc_selector_mapping = {}
reverse_swift_signature_mapping = {}
conflicts = {}
for map_file in map_files:
try:
data = json.loads(Path(map_file).read_text(encoding="utf-8"))
except Exception as e:
print(f"⚠️ 跳过损坏映射文件 {map_file}: {e}")
continue
for original, obfuscated in data.items():
if isinstance(obfuscated, dict):
reverse_swift_entry = _build_reverse_swift_signature_entry(original, obfuscated)
if reverse_swift_entry is None:
continue
reverse_signature_key, reverse_signature_meta = reverse_swift_entry
_record_reverse_mapping_entry(
reverse_swift_signature_mapping,
reverse_signature_key,
reverse_signature_meta,
conflicts,
)
continue
if not isinstance(obfuscated, str):
continue
selector_parts = split_objc_selector(original)
if selector_parts:
reverse_selector = "".join(
f"{obfuscated}_{index}:"
for index in range(1, len(selector_parts) + 1)
)
_record_reverse_mapping_entry(
reverse_objc_selector_mapping,
reverse_selector,
selector_parts,
conflicts,
)
swift_alias = _build_swift_alias_from_objc_selector(original, obfuscated)
if swift_alias is not None:
_record_reverse_mapping_entry(
reverse,
swift_alias["obfuscated_base_name"],
swift_alias["base_name"],
conflicts,
)
reverse_signature_params = []
reverse_params = []
for index, param in enumerate(swift_alias["meta"].get("params", []), start=1):
original_label = param.get("label")
if not isinstance(original_label, str):
continue
current_label = "_" if original_label == "_" else f"{obfuscated}_{index}"
reverse_signature_params.append({
"label": current_label,
"has_default": False,
"allows_trailing_closure": False,
})
reverse_params.append({
"label": current_label,
"has_default": False,
"allows_trailing_closure": False,
"replacement": None if original_label == "_" else original_label,
})
reverse_signature_key = _build_swift_signature_key(swift_alias["base_name"], reverse_signature_params)
_record_reverse_mapping_entry(
reverse_swift_signature_mapping,
reverse_signature_key,
{
"base_name": swift_alias["base_name"],
"obfuscated": swift_alias["obfuscated_base_name"],
"params": reverse_params,
},
conflicts,
)
continue
_record_reverse_mapping_entry(reverse, obfuscated, original, conflicts)
# 去掉存在歧义的 token避免错误恢复
for obfuscated in conflicts:
reverse.pop(obfuscated, None)
reverse_objc_selector_mapping.pop(obfuscated, None)
reverse_swift_signature_mapping.pop(obfuscated, None)
return reverse, reverse_objc_selector_mapping, reverse_swift_signature_mapping, conflicts
def restore_source_file(path: str, reverse_mapping: dict, reverse_re: re.Pattern, reverse_objc_selector_mapping: dict = None, reverse_swift_signature_mapping: dict = None, reverse_aho=None) -> tuple:
try:
original = Path(path).read_text(encoding="utf-8")
except Exception as e:
return False, {"error": str(e)}, None, None
stats = {}
reverse_objc_selector_mapping = reverse_objc_selector_mapping or {}
reverse_swift_signature_mapping = reverse_swift_signature_mapping or {}
if reverse_aho is not None:
matches = _iter_aho_matches(original, reverse_aho)
if matches:
chunks = []
cursor = 0
for start, end, key in matches:
if cursor < start:
chunks.append(original[cursor:start])
chunks.append(reverse_mapping[key])
stats[key] = stats.get(key, 0) + 1
cursor = end + 1
if cursor < len(original):
chunks.append(original[cursor:])
result = ''.join(chunks)
else:
result = original
else:
def repl(match: re.Match) -> str:
obfuscated = match.group(1)
stats[obfuscated] = stats.get(obfuscated, 0) + 1
return reverse_mapping[obfuscated]
result = reverse_re.sub(repl, original)
if reverse_objc_selector_mapping and path.endswith((".h", ".m", ".mm", ".pch")):
result = apply_objc_selector_mapping(result, reverse_objc_selector_mapping, stats)
if reverse_swift_signature_mapping and path.endswith(".swift"):
result = apply_swift_signature_mapping(result, reverse_swift_signature_mapping, stats)
if result == original:
return False, {}, None, original
return True, stats, result, original
def transactional_write(planned_changes: dict):
if not planned_changes:
return 0
backup_dir = tempfile.mkdtemp(prefix="obf_apply_backup_")
changed_files = sorted(planned_changes.keys())
# Collect any import renames requested by planned changes (old_basename -> new_basename)
import_re = re.compile(r'#import\s+"([^"]+\.h)"')
rename_pairs = []
for path, info in planned_changes.items():
orig = info.get("original_text", "")
new = info.get("new_text", "")
orig_imports = [m.group(1) for m in import_re.finditer(orig)]
new_imports = [m.group(1) for m in import_re.finditer(new)]
# pair up imports by position; if basename differs, schedule rename
for o, n in zip(orig_imports, new_imports):
if o != n:
rename_pairs.append({"old": os.path.basename(o), "new": os.path.basename(n)})
state = {"backup_dir": backup_dir, "files": changed_files, "rename_pairs": rename_pairs}
save_apply_state(state)
for path in changed_files:
rel = path.lstrip("./")
backup_target = Path(backup_dir) / rel
backup_target.parent.mkdir(parents=True, exist_ok=True)
atomic_write_text(str(backup_target), planned_changes[path]["original_text"], encoding="utf-8")
try:
for path in changed_files:
atomic_write_text(path, planned_changes[path]["new_text"], encoding="utf-8")
# After writing changed files, perform header renames if requested.
for pair in rename_pairs:
old_name = pair.get("old")
new_name = pair.get("new")
if not old_name or not new_name:
continue
# Find candidate files by basename in workspace
for found in Path('.').rglob(old_name):
try:
# backup original header before renaming
rel = str(found).lstrip("./")
backup_target = Path(backup_dir) / rel
backup_target.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(found, backup_target)
new_path = found.with_name(new_name)
# if destination exists, back it up too
if new_path.exists():
rel_new = str(new_path).lstrip("./")
backup_target_new = Path(backup_dir) / rel_new
backup_target_new.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(new_path, backup_target_new)
# perform atomic move
Path(new_path.parent).mkdir(parents=True, exist_ok=True)
os.replace(str(found), str(new_path))
except Exception:
# If any rename fails, raise to trigger rollback
raise
clear_apply_state()
shutil.rmtree(backup_dir, ignore_errors=True)
except Exception as e:
print(f"❌ 写入过程中异常,开始回滚: {e}")
for path in changed_files:
rel = path.lstrip("./")
backup_source = Path(backup_dir) / rel
if backup_source.exists():
shutil.copy2(backup_source, path)
# Restore any renamed headers from backup if present
rpairs = state.get("rename_pairs", [])
for pair in rpairs:
old = pair.get("old")
new = pair.get("new")
if not old or not new:
continue
# If new file exists at any path, remove it and restore original from backup
for new_found in Path('.').rglob(new):
try:
# remove the moved file
new_found.unlink()
except Exception:
pass
# restore original from backup if available
for backup_file in Path(backup_dir).rglob(old):
try:
dest = Path('.') / backup_file.relative_to(backup_dir)
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(backup_file, dest)
except Exception:
pass
clear_apply_state()
shutil.rmtree(backup_dir, ignore_errors=True)
raise
return len(changed_files)
def direct_write(planned_changes: dict):
if not planned_changes:
return 0
changed_files = sorted(planned_changes.keys())
for path in changed_files:
atomic_write_text(path, planned_changes[path]["new_text"], encoding="utf-8")
return len(changed_files)
def restore_sources_from_maps(scan_dirs: list, map_dir: str):
restore_started_at = time.perf_counter()
map_files = sorted(str(p) for p in Path(map_dir).glob("*.json"))
if not map_files:
print(f"⚠️ 未在 {map_dir} 发现映射文件")
return
reverse_mapping, reverse_objc_selector_mapping, reverse_swift_signature_mapping, conflicts = load_reverse_mappings_from_files(map_files)
print(
f"🧩 加载映射文件 {len(map_files)} 个,"
f"可用反向符号 {len(reverse_mapping)} 个,"
f"Obj-C selector {len(reverse_objc_selector_mapping)} 个,"
f"Swift 签名 {len(reverse_swift_signature_mapping)}"
)
if conflicts:
print(f"⚠️ 检测到 {len(conflicts)} 个歧义 token已跳过避免错误恢复")
if not reverse_mapping and not reverse_objc_selector_mapping and not reverse_swift_signature_mapping:
print("⚠️ 没有可用于恢复的反向映射")
return
reverse_re = build_mapping_regex(reverse_mapping)
reverse_aho = build_aho_automaton(reverse_mapping)
source_files = collect_source_files(scan_dirs)
print(f"📂 准备恢复源码文件 {len(source_files)}")
planned_changes = {}
planning_started_at = time.perf_counter()
for idx, path in enumerate(source_files, start=1):
changed, stats, new_text, original_text = restore_source_file(
path,
reverse_mapping,
reverse_re,
reverse_objc_selector_mapping,
reverse_swift_signature_mapping,
reverse_aho,
)
if changed:
planned_changes[path] = {
"stats": stats,
"new_text": new_text,
"original_text": original_text,
}
if idx % 100 == 0:
print(f" ... 恢复扫描进度 {idx}/{len(source_files)}")
_log_stage_duration("源码还原规划", planning_started_at)
restored_count = transactional_write(planned_changes)
print(f"✅ 源码恢复完成,修改了 {restored_count} 个文件")
_log_stage_duration("源码还原总阶段", restore_started_at)
def _is_protected_objc_line(line: str, line_index: int) -> bool:
s = line.strip()
# 保护 #import "xxx.h",避免类名替换后头文件名失配
if s.startswith("#import") and '.h"' in s:
# Protect import lines unless the user explicitly enabled import rewriting.
return not REWRITE_IMPORTS_ALLOWED
# 保护文件头注释中的文件名(通常在前 40 行)
if line_index < 40 and ".h" in s and (s.startswith("//") or s.startswith("/*") or s.startswith("*")):
return True
if line_index < 40 and (".m" in s or ".mm" in s) and (s.startswith("//") or s.startswith("/*") or s.startswith("*")):
return True
return False
def atomic_write_text(path: str, content: str, encoding: str = "utf-8"):
"""原子写文件,避免中途中断导致半写入文件。"""
target = Path(path)
target.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile("w", delete=False, dir=str(target.parent), encoding=encoding) as tf:
tf.write(content)
temp_path = tf.name
os.replace(temp_path, str(target))
# Simple cache to skip unchanged files between runs
CACHE_PATH = ".obfuscate_cache.json"
def load_cache():
data = load_json_file(CACHE_PATH, {})
return data if isinstance(data, dict) else {}
def save_cache(cache_data: dict):
try:
save_json_file(CACHE_PATH, cache_data)
except Exception:
pass
def _format_elapsed(seconds: float) -> str:
if seconds < 60:
return f"{seconds:.2f}s"
minutes, seconds = divmod(seconds, 60)
if minutes < 60:
return f"{int(minutes)}m {seconds:.2f}s"
hours, minutes = divmod(int(minutes), 60)
return f"{hours}h {minutes}m {seconds:.2f}s"
def _log_stage_duration(label: str, started_at: float):
suffix = "" if label.endswith("耗时") else "耗时"
print(f"⏱️ {label}{suffix}: {_format_elapsed(time.perf_counter() - started_at)}")
# Globals used by multiprocessing workers (initialized in Pool initializer)
_GLOBAL_WORKER_MAPPING = None
_GLOBAL_WORKER_MAPPING_RE = None
_GLOBAL_WORKER_AHO = None
_GLOBAL_WORKER_SELECTOR_MAPPING = None
_GLOBAL_WORKER_SWIFT_SIGNATURE_MAPPING = None
_GLOBAL_WORKER_DIRECT_METHOD_MAPPING = None
def _pool_init(mapping_json_str: str, selector_mapping_json_str: str, swift_signature_mapping_json_str: str, direct_method_mapping_json_str: str):
global _GLOBAL_WORKER_MAPPING, _GLOBAL_WORKER_MAPPING_RE, _GLOBAL_WORKER_AHO, _GLOBAL_WORKER_SELECTOR_MAPPING, _GLOBAL_WORKER_SWIFT_SIGNATURE_MAPPING, _GLOBAL_WORKER_DIRECT_METHOD_MAPPING
try:
_GLOBAL_WORKER_MAPPING = json.loads(mapping_json_str)
_GLOBAL_WORKER_SELECTOR_MAPPING = json.loads(selector_mapping_json_str)
_GLOBAL_WORKER_SWIFT_SIGNATURE_MAPPING = json.loads(swift_signature_mapping_json_str)
_GLOBAL_WORKER_DIRECT_METHOD_MAPPING = json.loads(direct_method_mapping_json_str)
_GLOBAL_WORKER_AHO = build_aho_automaton(_GLOBAL_WORKER_MAPPING)
_GLOBAL_WORKER_MAPPING_RE = build_mapping_regex(_GLOBAL_WORKER_MAPPING)
except Exception:
_GLOBAL_WORKER_MAPPING = None
_GLOBAL_WORKER_AHO = None
_GLOBAL_WORKER_MAPPING_RE = None
_GLOBAL_WORKER_SELECTOR_MAPPING = None
_GLOBAL_WORKER_SWIFT_SIGNATURE_MAPPING = None
_GLOBAL_WORKER_DIRECT_METHOD_MAPPING = None
def _pool_plan_worker(path: str):
"""Worker used in multiprocessing Pool to plan file changes using global mapping/aho."""
global _GLOBAL_WORKER_MAPPING, _GLOBAL_WORKER_MAPPING_RE, _GLOBAL_WORKER_AHO, _GLOBAL_WORKER_SELECTOR_MAPPING, _GLOBAL_WORKER_SWIFT_SIGNATURE_MAPPING, _GLOBAL_WORKER_DIRECT_METHOD_MAPPING
if _GLOBAL_WORKER_MAPPING is None and not _GLOBAL_WORKER_SELECTOR_MAPPING and not _GLOBAL_WORKER_SWIFT_SIGNATURE_MAPPING and not _GLOBAL_WORKER_DIRECT_METHOD_MAPPING:
return path, False, {}, None, None
is_objc_file = path.endswith((".h", ".m", ".mm", ".pch"))
# call apply_mapping_to_file with globals
changed, stats, new_text, original_text = apply_mapping_to_file(
path,
_GLOBAL_WORKER_MAPPING,
_GLOBAL_WORKER_MAPPING_RE,
objc_selector_mapping=_GLOBAL_WORKER_SELECTOR_MAPPING,
swift_signature_mapping=_GLOBAL_WORKER_SWIFT_SIGNATURE_MAPPING,
direct_method_mapping=_GLOBAL_WORKER_DIRECT_METHOD_MAPPING,
protect_objc_headers=is_objc_file,
mapping_aho=_GLOBAL_WORKER_AHO,
)
return path, changed, stats, new_text, original_text
def save_apply_state(state: dict):
Path(APPLY_STATE_PATH).write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")
def clear_apply_state():
p = Path(APPLY_STATE_PATH)
if p.exists():
p.unlink()
def restore_from_apply_state_if_needed():
"""如果上次 apply 中断,优先恢复,避免出现部分文件被混淆。"""
p = Path(APPLY_STATE_PATH)
if not p.exists():
return
try:
state = json.loads(p.read_text(encoding="utf-8"))
except Exception as e:
print(f"⚠️ 发现损坏的 apply 状态文件,已忽略: {e}")
return
backup_dir = state.get("backup_dir", "")
files = state.get("files", [])
rename_pairs = state.get("rename_pairs", [])
if not backup_dir or not files:
clear_apply_state()
return
print("🧯 检测到上次混淆未完成,正在自动回滚文件...")
restored = 0
for file_path in files:
rel = file_path.lstrip("./")
src = Path(backup_dir) / rel
dst = Path(file_path)
if src.exists():
dst.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(src, dst)
restored += 1
print(f"✅ 已回滚 {restored}/{len(files)} 个文件")
# Attempt to restore any renamed headers recorded in state
for pair in rename_pairs:
old_name = pair.get("old")
new_name = pair.get("new")
if not old_name or not new_name:
continue
# If new files exist, remove them and restore original from backup copies if present
for new_found in Path('.').rglob(new_name):
try:
new_found.unlink()
except Exception:
pass
for backup_file in Path(backup_dir).rglob(old_name):
try:
dest = Path('.') / backup_file.relative_to(backup_dir)
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copy2(backup_file, dest)
except Exception:
pass
try:
shutil.rmtree(backup_dir, ignore_errors=True)
finally:
clear_apply_state()
def apply_mapping_to_file(path: str, mapping: dict, mapping_re: re.Pattern, objc_selector_mapping: dict = None, swift_signature_mapping: dict = None, direct_method_mapping: dict = None, protect_objc_headers: bool = False, mapping_aho=None) -> tuple:
"""
对单个文件执行符号替换。
返回 (changed: bool, stats: dict, new_text: str|None, original_text: str|None)。
使用词边界匹配,避免部分替换。
"""
try:
original = Path(path).read_text(encoding="utf-8")
except Exception as e:
return False, {"error": str(e)}, None, None
stats = {}
objc_selector_mapping = objc_selector_mapping or {}
swift_signature_mapping = swift_signature_mapping or {}
direct_method_mapping = direct_method_mapping or {}
result = original
if objc_selector_mapping and path.endswith((".h", ".m", ".mm", ".pch")):
result = apply_objc_selector_mapping(result, objc_selector_mapping, stats)
if swift_signature_mapping and path.endswith(".swift"):
result = apply_swift_signature_mapping(result, swift_signature_mapping, stats, direct_method_mapping)
if direct_method_mapping:
if path.endswith((".h", ".m", ".mm", ".pch")):
result = apply_objc_direct_method_mapping(result, direct_method_mapping, stats)
elif path.endswith(".swift"):
result = apply_swift_direct_method_mapping(result, direct_method_mapping, stats)
# Try to use Aho-Corasick automaton if available for single-pass replacement
aho = mapping_aho
if aho is None:
try:
# build_aho_automaton may be expensive; caller can prebuild and pass, but fallback here
aho = build_aho_automaton(mapping)
except Exception:
aho = None
def do_regex_replace(text: str):
def repl(match: re.Match) -> str:
orig = match.group(1)
stats[orig] = stats.get(orig, 0) + 1
return mapping[orig]
return mapping_re.sub(repl, text)
if aho is None:
# fallback to previous regex-based replacement
if protect_objc_headers:
out_lines = []
for i, line in enumerate(result.splitlines(keepends=True)):
if _is_protected_objc_line(line, i):
out_lines.append(line)
else:
out_lines.append(do_regex_replace(line))
result = "".join(out_lines)
else:
result = do_regex_replace(result)
if result == original:
return False, {}, None, original
return True, stats, result, original
# With aho: perform line-wise replacement when protect_objc_headers True to preserve header guards
def replace_with_aho_in_text(text: str):
matches = _iter_aho_matches(text, aho)
if not matches:
return text
# build new text
out = []
pos = 0
for s, e, k in matches:
if pos < s:
out.append(text[pos:s])
out.append(mapping[k])
stats[k] = stats.get(k, 0) + 1
pos = e + 1
if pos < len(text):
out.append(text[pos:])
return ''.join(out)
if protect_objc_headers:
out_lines = []
for i, line in enumerate(result.splitlines(keepends=True)):
if _is_protected_objc_line(line, i):
out_lines.append(line)
else:
out_lines.append(replace_with_aho_in_text(line))
result = ''.join(out_lines)
else:
result = replace_with_aho_in_text(result)
if result == original:
return False, {}, None, original
return True, stats, result, original
def count_mapping_occurrences_in_file(path: str, mapping_re: re.Pattern, mapping_aho=None) -> dict:
try:
text = Path(path).read_text(encoding="utf-8")
except Exception:
return {}
counts = {}
if mapping_aho is not None:
for _, _, key in _iter_aho_matches(text, mapping_aho):
counts[key] = counts.get(key, 0) + 1
return counts
for match in mapping_re.finditer(text):
key = match.group(1)
counts[key] = counts.get(key, 0) + 1
return counts
def count_mapping_occurrences(files: list, mapping_re: re.Pattern, workers: int = 1, mapping_aho=None) -> dict:
totals = {}
if workers <= 1 or len(files) < 20:
for path in files:
file_counts = count_mapping_occurrences_in_file(path, mapping_re, mapping_aho)
for key, value in file_counts.items():
totals[key] = totals.get(key, 0) + value
return totals
with ThreadPoolExecutor(max_workers=workers) as executor:
future_map = {
executor.submit(count_mapping_occurrences_in_file, path, mapping_re, mapping_aho): path
for path in files
}
for future in as_completed(future_map):
file_counts = future.result()
for key, value in file_counts.items():
totals[key] = totals.get(key, 0) + value
return totals
def plan_file_change(path: str, mapping: dict, mapping_re: re.Pattern, objc_selector_mapping: dict = None, swift_signature_mapping: dict = None, mapping_aho=None, method_direct_mapping: dict = None) -> tuple:
# Optional verbose per-file logging for diagnosing slow files during planning.
if os.environ.get("RD_OBF_VERBOSE", "0") == "1":
try:
print(f"[obf-plan] Processing: {path}")
except Exception:
pass
is_objc_file = path.endswith((".h", ".m", ".mm", ".pch"))
local_mapping = mapping
local_mapping_re = mapping_re
local_mapping_aho = mapping_aho
local_objc_selector_mapping = objc_selector_mapping
local_swift_signature_mapping = swift_signature_mapping
local_direct_method_mapping = method_direct_mapping or {}
if should_skip_method_replacements(path):
local_objc_selector_mapping = {}
local_swift_signature_mapping = {}
local_direct_method_mapping = {}
changed, stats, new_text, original_text = apply_mapping_to_file(
path,
local_mapping,
local_mapping_re,
objc_selector_mapping=local_objc_selector_mapping,
swift_signature_mapping=local_swift_signature_mapping,
direct_method_mapping=local_direct_method_mapping,
protect_objc_headers=is_objc_file,
mapping_aho=local_mapping_aho,
)
return path, changed, stats, new_text, original_text
# ── DeobfuscateCrash 还原)────────────────────────────────────────────────
def deobfuscate_crash(crash_path: str, map_path: str):
with open(map_path, encoding="utf-8") as f:
mapping = json.load(f)
# 反转映射
reverse = {v: k for k, v in mapping.items()}
crash_text = Path(crash_path).read_text(encoding="utf-8")
for obf, orig in sorted(reverse.items(), key=lambda x: -len(x[0])):
pattern = r"(?<![A-Za-z0-9_])" + re.escape(obf) + r"(?![A-Za-z0-9_])"
crash_text = re.sub(pattern, orig, crash_text)
out_path = crash_path + ".deobfuscated.txt"
Path(out_path).write_text(crash_text, encoding="utf-8")
print(f"✅ 已还原 crash 符号 → {out_path}")
def apply_archive_mode_defaults(args):
if not args.scan_dir:
args.scan_dir = DEFAULT_SCAN_DIRS
if not args.archive_mode:
if args.map_output is None:
args.map_output = MAP_OUTPUT_PATH
if args.map_methods is None:
args.map_methods = METHOD_MAP_OUTPUT_PATH
return
target_name = args.target_name or os.environ.get("TARGET_NAME") or "app"
timestamp = args.timestamp or os.environ.get("OBF_TIMESTAMP") or time.strftime("%Y%m%d_%H%M%S")
map_dir = Path(args.map_dir or "obfuscation_maps")
map_dir.mkdir(parents=True, exist_ok=True)
args.apply = True
args.enable_methods = True
if args.whitelist_json is None:
default_whitelist = Path("must_keep_summary.json")
if default_whitelist.exists():
args.whitelist_json = str(default_whitelist)
if args.map_output is None:
args.map_output = str(map_dir / f"{target_name}_{timestamp}.json")
if args.map_methods is None:
args.map_methods = str(map_dir / f"method_map_{target_name}_{timestamp}.json")
print(f"📦 Archive 模式: target={target_name}")
print(f" 类型映射: {args.map_output}")
print(f" 方法映射: {args.map_methods}")
def resolve_effective_scan_dirs(scan_dirs: list) -> list:
# 规则和替换都必须按整个项目扫描,避免 3.0 子目录运行时漏掉其他模块、Pods、Factorys 中的定义与引用。
return list(DEFAULT_SCAN_DIRS)
# ── 主流程 ────────────────────────────────────────────────────────────────────
def main():
parser = argparse.ArgumentParser(description="Swift 符号混淆工具")
parser.add_argument("--scan-dir", nargs="+", default=None,
help="兼容保留的扫描目录参数;实际混淆规则统一按整个项目扫描(默认: .")
parser.add_argument("--apply", action="store_true",
help="实际写入文件(不加此参数为 dry-run 预览)")
parser.add_argument("--archive-mode", action="store_true",
help="用于 Archive/打包阶段:自动应用混淆、生成类型和方法映射到 obfuscation_maps并尽量使用仓库默认配置。")
parser.add_argument("--target-name", default=None,
help="Archive 模式下的目标名;默认读取 TARGET_NAME 环境变量。")
parser.add_argument("--timestamp", default=None,
help="Archive 模式下的时间戳;默认读取 OBF_TIMESTAMP 或当前时间。")
parser.add_argument("--map-dir", default=None,
help="Archive 模式下映射文件输出目录(默认: obfuscation_maps")
parser.add_argument("--generate-patch-dir", metavar="PATCH_DIR", default=None,
help="生成补丁到指定目录(不直接修改源码,便于审阅)")
parser.add_argument("--map-output", default=None,
help=f"映射表输出路径(默认: {MAP_OUTPUT_PATH})")
parser.add_argument("--deobfuscate", metavar="CRASH_LOG",
help="还原 crash 日志中的符号,需配合 --map-output 指向已有映射表")
parser.add_argument("--map-input", default=MAP_OUTPUT_PATH,
help="反混淆时使用的映射表(默认同 --map-output")
parser.add_argument("--whitelist-json", metavar="JSON_FILE", default=None,
help="从 must_keep_summary.json 等文件读取额外白名单('critical'/'high' severity 符号自动加入白名单)")
parser.add_argument("--salt", default=None,
help="自定义混淆盐值(默认自动随机生成;用于保证每次打包映射不同)")
parser.add_argument("--enable-methods", dest="enable_methods", action="store_true",
help="启用方法级别混淆dry-run 仅生成方法映射)。")
parser.add_argument("--map-methods", dest="map_methods", default=None,
help="方法映射输出路径dry-run默认 tmp_method_map.json")
parser.add_argument("--fast-scan", dest="fast_scan", action="store_true",
help="启用快速扫描:优先使用 ripgrep(rg) 或 grep 批量匹配映射键,显著加速大仓库统计。")
parser.add_argument("--workers", type=int, default=DEFAULT_WORKERS,
help=f"文件级并行工作线程数(默认: {DEFAULT_WORKERS}")
objc_group = parser.add_mutually_exclusive_group()
objc_group.add_argument("--enable-objc-symbols", dest="enable_objc_symbols", action="store_true",
help="启用 Obj-C 符号混淆(默认开启)。")
objc_group.add_argument("--disable-objc-symbols", dest="enable_objc_symbols", action="store_false",
help="关闭 Obj-C 符号混淆,仅混淆 Swift 符号。")
parser.set_defaults(enable_objc_symbols=True)
parser.add_argument("--restore-source-from-map-dir", metavar="MAP_DIR", default=None,
help="根据目录下全部映射文件反向恢复源码,例如 obfuscation_maps")
parser.add_argument("--rewrite-imports", action="store_true",
help="(危险) 允许替换 `#import \"X.h\"` 行,并尝试重命名对应头文件以匹配混淆后的名称。")
parser.add_argument("--verbose-changes", action="store_true",
help="输出每个变更文件及符号命中明细Archive 模式默认只输出摘要以减少耗时。")
args = parser.parse_args()
total_started_at = time.perf_counter()
# 任何操作前先检查并恢复上次中断的 apply
restore_from_apply_state_if_needed()
apply_archive_mode_defaults(args)
# Enable risky import rewriting if requested
global REWRITE_IMPORTS_ALLOWED
if getattr(args, 'rewrite_imports', False):
REWRITE_IMPORTS_ALLOWED = True
if args.restore_source_from_map_dir:
restore_sources_from_maps(args.scan_dir, args.restore_source_from_map_dir)
_log_stage_duration("本次执行总耗时", total_started_at)
return
# ── 反混淆模式
if args.deobfuscate:
deobfuscate_crash(args.deobfuscate, args.map_input)
_log_stage_duration("本次执行总耗时", total_started_at)
return
effective_scan_dirs = resolve_effective_scan_dirs(args.scan_dir)
# 每次执行默认随机盐值,确保映射每次都不同;可用 --salt 固定以复现
run_salt = args.salt or f"run_{secrets.token_hex(8)}"
print(f"🎲 本次混淆盐值: {run_salt}")
# ── 加载外部白名单(打包时传入 must_keep_summary.json
if args.whitelist_json:
try:
wl_data = json.loads(Path(args.whitelist_json).read_text(encoding="utf-8"))
added = 0
for entry in wl_data:
sev = entry.get("severity", "low")
sym = entry.get("symbol", "")
if sym and sev in ("critical", "high"):
SYMBOL_WHITELIST.add(sym)
added += 1
print(f"🛡️ 从 {args.whitelist_json} 加载 {added} 个白名单符号critical/high")
except Exception as e:
print(f"⚠️ 加载白名单文件失败: {e}")
# ── 扫描
if args.scan_dir != effective_scan_dirs:
print(f"📂 请求扫描目录: {args.scan_dir}")
print(f"📂 实际扫描目录: {effective_scan_dirs}(规则统一按整个项目执行)")
else:
print(f"📂 扫描目录: {effective_scan_dirs}")
ensure_ios_system_symbols_loaded()
ensure_third_party_protocol_methods_loaded(effective_scan_dirs)
swift_files = collect_swift_files(effective_scan_dirs)
objc_files = collect_objc_files(effective_scan_dirs)
method_candidate_files = collect_method_candidate_files(effective_scan_dirs) if args.enable_methods else []
print(f" 找到 {len(swift_files)} 个 Swift 文件,{len(objc_files)} 个 Obj-C 文件(已排除 .build/Pods/Tests 等)")
if args.enable_methods:
print(f" 方法候选文件 {len(method_candidate_files)}rg 预筛后)")
# ── 提取符号
print("🔍 提取资源字符串以建立白名单storyboard/xib...")
resource_strings = collect_resource_strings(effective_scan_dirs)
for s in resource_strings:
SYMBOL_WHITELIST.add(s)
print("🔍 探测 Obj-C 中的选择器字面量并加入白名单...")
selector_literals = collect_selector_literals(objc_files)
for s in selector_literals:
SYMBOL_WHITELIST.add(s)
print("🔍 汇总项目属性成员白名单(支持缓存增量复用)...")
ensure_project_property_members_loaded(effective_scan_dirs)
print(f"🔍 动态探测 Obj-C 点语法 getter并同步到 {PROJECT_METHOD_IGNORE_FILE_PATH}...")
detected_dot_syntax_methods, ignore_file_updated = sync_detected_objc_dot_syntax_method_ignores(effective_scan_dirs)
print(
f" 自动忽略 {len(detected_dot_syntax_methods)} 个 Obj-C 零参方法"
+ (",已更新项目忽略列表" if ignore_file_updated else "")
)
print("🔍 提取 canonical symbol 表(按构建生成映射)...")
canonical_tables = build_canonical_symbol_tables(
swift_files,
objc_files,
method_candidate_files,
enable_objc_symbols=args.enable_objc_symbols,
prefilter_files=lambda files, pattern: prefilter_files_with_rg(files, pattern, chunk_size=RG_CHUNK_SIZE),
type_pattern=TYPE_DECLARATION_RG_PATTERN,
method_pattern=METHOD_DECLARATION_RG_PATTERN,
extract_swift_symbols=extract_symbols_from_file,
extract_objc_symbols=extract_objc_symbols_from_file,
extract_methods=extract_methods_from_file,
objc_symbol_prefixes=OBJC_SYMBOL_PREFIXES,
already_obfuscated_re=ALREADY_OBFUSCATED_RE,
symbol_whitelist=SYMBOL_WHITELIST,
symbol_prefix_whitelist=SYMBOL_PREFIX_WHITELIST,
is_system_method=is_system_method,
split_swift_signature_key=_split_swift_signature_key,
)
save_extraction_cache()
print(f"🧾 canonical digest: {canonical_tables['digest'][:12]}")
mapping_swift, mapping_objc = build_type_mappings_from_canonical_table(
canonical_tables,
run_salt,
enable_objc_symbols=args.enable_objc_symbols,
build_unique_mapping=build_unique_mapping,
type_prefix=TYPE_OBFUSCATED_PREFIX,
)
if not args.enable_objc_symbols:
print("⚠️ 已显式关闭 Obj-C 符号混淆,本次仅混淆 Swift 符号")
# 合并映射(优先保留 Swift 映射同名 -> 确保一致)
mapping = {}
mapping.update(mapping_objc)
mapping.update(mapping_swift)
# 方法映射dry-run 输出)
if args.enable_methods:
print("🔍 按本次构建盐值生成方法映射...")
mapping_methods = build_method_mapping_from_canonical_table(
canonical_tables,
run_salt,
build_unique_mapping=build_unique_mapping,
method_prefix=METHOD_OBFUSCATED_PREFIX,
)
# 移除本地生成的原生 Swift signature 条目,保留 Obj-C selector / 直接方法条目。
mapping_methods = {
k: v for k, v in mapping_methods.items()
if not (isinstance(v, dict) and v.get('kind') == 'swift_signature')
}
mapping_methods, unsafe_bases, dropped_signatures = _drop_unsafe_generated_swift_signature_entries(mapping_methods)
if dropped_signatures:
preview = ", ".join(sorted(unsafe_bases)[:5])
suffix = " ..." if len(unsafe_bases) > 5 else ""
print(
f"⚠️ 跳过 {len(dropped_signatures)} 个存在重叠调用形态的 Swift 参数标签映射,"
f"涉及 {len(unsafe_bases)} 个方法基名: {preview}{suffix}"
)
with open(args.map_methods, "w", encoding="utf-8") as mf:
json.dump(mapping_methods, mf, ensure_ascii=False, indent=2)
print(f"💾 方法映射已保存: {args.map_methods} dry-run")
print(f" 共发现 {len(mapping)} 个可混淆类型名(包含 Swift + Obj-C ")
if not mapping:
print("⚠️ 没有找到可混淆的符号,请检查扫描目录和过滤规则。")
_log_stage_duration("本次执行总耗时", total_started_at)
sys.exit(0)
# ── 预览前 30 条
print("\n── 混淆预览(前 30 条)──────────────────────────────────────")
for i, (orig, obf) in enumerate(sorted(mapping.items())):
if i >= 30:
print(f" ... 还有 {len(mapping) - 30}")
break
print(f" {orig:50s}{obf}")
# ── 保存映射表
with open(args.map_output, "w", encoding="utf-8") as f:
json.dump(mapping, f, ensure_ascii=False, indent=2)
print(f"\n💾 映射表已保存: {args.map_output} (请妥善保管,用于 crash 还原)")
if not args.apply:
print("\n⚠️ Dry-run 模式,未修改任何文件。加 --apply 参数执行实际替换。")
_log_stage_duration("本次执行总耗时", total_started_at)
return
# 如果启用了方法级别混淆并且存在方法映射文件,把方法映射合并进 mapping
method_direct_mapping = {}
objc_selector_mapping = {}
swift_signature_mapping = {}
selector_display_mapping = {}
swift_signature_display_mapping = {}
if args.enable_methods and args.map_methods and os.path.exists(args.map_methods):
try:
method_direct_mapping, objc_selector_mapping, swift_signature_mapping = load_method_obfuscation_maps(args.map_methods)
selector_display_mapping = build_selector_display_mapping(objc_selector_mapping)
swift_signature_display_mapping = build_swift_signature_display_mapping(method_direct_mapping, swift_signature_mapping)
except Exception as e:
print(f"⚠️ 加载方法映射失败: {e}")
# 统计映射键在源码中的命中次数,生成未命中报告供审查
count_stage_started_at = time.perf_counter()
try:
source_files_for_search = collect_apply_target_files(effective_scan_dirs, args.enable_objc_symbols)
occurrences = {}
keys = list(mapping.keys())
mapping_re = build_mapping_regex(mapping)
mapping_aho = build_aho_automaton(mapping)
def _fast_count_with_rg(keys_list, scan_dirs):
import subprocess
from tempfile import NamedTemporaryFile
counts = {k: 0 for k in keys_list}
keysfile = None
try:
with NamedTemporaryFile('w', delete=False, encoding='utf-8') as tf:
for k in keys_list:
tf.write(k + "\n")
keysfile = tf.name
rg = shutil.which('rg')
if rg:
# -n: show line number, -o: only matching, -F: fixed strings, -f: patterns file
cmd = [rg, '-n', '-o', '-F', '-f', keysfile, '--no-messages', '-S'] + list(scan_dirs)
out = subprocess.check_output(cmd, stderr=subprocess.DEVNULL, encoding='utf-8')
for ln in out.splitlines():
# format: path:line:matched_text
parts = ln.split(':', 2)
if len(parts) == 3:
matched = parts[2]
if matched in counts:
counts[matched] += 1
return counts
# fallback to grep if rg not available
grep = shutil.which('grep')
if grep:
cmd = [grep, '-R', '-n', '-o', '-F', '-f', keysfile] + list(scan_dirs)
out = subprocess.check_output(cmd, stderr=subprocess.DEVNULL, encoding='utf-8')
for ln in out.splitlines():
# grep format: path:line:matched_text
parts = ln.split(':', 2)
if len(parts) == 3:
matched = parts[2]
if matched in counts:
counts[matched] += 1
return counts
except subprocess.CalledProcessError:
# no matches -> return zero counts
return counts
except Exception:
return counts
finally:
if keysfile:
try:
os.unlink(keysfile)
except Exception:
pass
if args.fast_scan:
print("⚡ 使用快速扫描rg/grep统计映射键命中...")
occurrences = _fast_count_with_rg(keys, effective_scan_dirs)
else:
print(f"⚡ 使用单次扫描聚合统计映射键命中workers={args.workers}...")
occurrences = count_mapping_occurrences(source_files_for_search, mapping_re, workers=args.workers, mapping_aho=mapping_aho)
for key in keys:
occurrences.setdefault(key, 0)
unmapped = [k for k, v in occurrences.items() if v == 0]
mapped = [k for k, v in occurrences.items() if v > 0]
report = {
'total_keys': len(occurrences),
'mapped': len(mapped),
'unmapped': len(unmapped),
'unmapped_list': unmapped[:1000]
}
Path('unmapped_mappings.json').write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding='utf-8')
print(f"\n🔎 未命中报告已写入: unmapped_mappings.json{len(unmapped)} 未命中,共 {len(occurrences)} 键)")
if unmapped:
print("前 200 个未命中示例:")
for i, k in enumerate(unmapped[:200], start=1):
print(f" {i:3d}. {k}")
except Exception as e:
print(f"⚠️ 生成未命中报告出错: {e}")
finally:
_log_stage_duration("混淆统计", count_stage_started_at)
# ── 生成替换计划(不一定立即写入)
print("\n✏️ 生成替换计划(未必直接写入)...")
planning_stage_started_at = time.perf_counter()
total_changed = 0
all_targets = collect_apply_target_files(effective_scan_dirs, args.enable_objc_symbols)
mapping_re = build_mapping_regex(mapping)
mapping_aho = build_aho_automaton(mapping)
display_mapping = dict(mapping)
display_mapping.update(method_direct_mapping)
display_mapping.update(selector_display_mapping)
display_mapping.update(swift_signature_display_mapping)
planned_changes = {}
# build mapping hash for cache invalidation
mapping_hash = hashlib.sha256(
json.dumps({
'mapping': mapping,
'direct_method_mapping': method_direct_mapping,
'objc_selector_mapping': objc_selector_mapping,
'swift_signature_mapping': swift_signature_mapping,
}, sort_keys=True).encode()
).hexdigest()
cache = load_cache()
cache_files = cache.get('files', {}) if isinstance(cache.get('files', {}), dict) else {}
# determine which files need processing: changed mtime or mapping changed
files_to_process = []
for p in all_targets:
try:
mtime = os.path.getmtime(p)
except Exception:
mtime = None
prev = cache_files.get(p)
if cache.get('mapping_hash') != mapping_hash or prev != mtime:
files_to_process.append(p)
print(f"🔁 总目标文件 {len(all_targets)},需处理 {len(files_to_process)}(基于缓存)")
if args.workers > 1 and len(files_to_process) >= 20:
# Avoid returning full file contents over multiprocessing pipes in archive mode.
# Use threads and flush per-file patches to disk when --generate-patch-dir is used
# to reduce memory pressure and allow inspecting progress live.
write_patches_live = bool(args.generate_patch_dir)
patch_dir = Path(args.generate_patch_dir) if write_patches_live else None
if write_patches_live:
patch_dir.mkdir(parents=True, exist_ok=True)
manifest = {}
with ThreadPoolExecutor(max_workers=args.workers) as executor:
future_map = {
executor.submit(
plan_file_change,
path,
mapping,
mapping_re,
objc_selector_mapping,
swift_signature_mapping,
mapping_aho,
method_direct_mapping,
): path
for path in files_to_process
}
for idx, future in enumerate(as_completed(future_map), start=1):
path = future_map[future]
try:
path, changed, stats, new_text, original_text = future.result()
except Exception as e:
print(f"⚠️ 并行规划失败,文件: {path},错误: {e}")
raise
if changed:
if write_patches_live:
rel = Path(path).relative_to(Path('.'))
out_orig = patch_dir / (str(rel) + '.orig')
out_new = patch_dir / (str(rel) + '.new')
out_orig.parent.mkdir(parents=True, exist_ok=True)
try:
out_orig.write_text(original_text, encoding='utf-8')
out_new.write_text(new_text, encoding='utf-8')
except Exception:
# best-effort write
pass
manifest[str(rel)] = {'stats': stats}
else:
planned_changes[path] = {
'stats': stats,
'new_text': new_text,
}
if not args.archive_mode or args.generate_patch_dir:
planned_changes[path]['original_text'] = original_text
total_changed += 1
if idx % 50 == 0:
print(f" ... 进度 {idx}/{len(files_to_process)}")
if write_patches_live:
# finalize manifest
try:
manifest_path = patch_dir / 'manifest.json'
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8')
print(f"✅ 补丁已生成到: {patch_dir} (包含 .orig/.new 文件及 manifest.json未修改源码。")
except Exception as e:
print(f"⚠️ 无法写入补丁 manifest: {e}")
_log_stage_duration("替换规划", planning_stage_started_at)
_log_stage_duration("本次执行总耗时", total_started_at)
return
else:
for idx, path in enumerate(files_to_process, start=1):
path, changed, stats, new_text, original_text = plan_file_change(path, mapping, mapping_re, objc_selector_mapping, swift_signature_mapping, mapping_aho, method_direct_mapping)
if changed:
planned_changes[path] = {
'stats': stats,
'new_text': new_text,
}
if not args.archive_mode or args.generate_patch_dir:
planned_changes[path]['original_text'] = original_text
total_changed += 1
if idx % 50 == 0:
print(f" ... 进度 {idx}/{len(files_to_process)}")
# update cache for processed files
for p in files_to_process:
try:
cache_files[p] = os.path.getmtime(p)
except Exception:
cache_files[p] = None
cache_out = {'mapping_hash': mapping_hash, 'files': cache_files}
save_cache(cache_out)
_log_stage_duration("替换规划", planning_stage_started_at)
if total_changed == 0:
print(f"\n✅ 预览完成没有检测到要修改的文件0/{len(all_targets)})。")
print(f" 映射表: {args.map_output}")
print(" 若需实际应用,请使用 --apply")
_log_stage_duration("本次执行总耗时", total_started_at)
return
# 如果指定了补丁目录,则把 planned_changes 写到该目录用于审阅(不修改源码)
if args.generate_patch_dir:
patch_dir = Path(args.generate_patch_dir)
patch_dir.mkdir(parents=True, exist_ok=True)
manifest = {}
for path, info in planned_changes.items():
rel = Path(path).relative_to(Path('.'))
out_orig = patch_dir / (str(rel) + '.orig')
out_new = patch_dir / (str(rel) + '.new')
out_orig.parent.mkdir(parents=True, exist_ok=True)
out_orig.write_text(info['original_text'], encoding='utf-8')
out_new.write_text(info['new_text'], encoding='utf-8')
manifest[str(rel)] = {'stats': info['stats']}
manifest_path = patch_dir / 'manifest.json'
manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2), encoding='utf-8')
print(f"✅ 补丁已生成到: {patch_dir} (包含 .orig/.new 文件及 manifest.json未修改源码。")
_log_stage_duration("本次执行总耗时", total_started_at)
return
write_stage_started_at = time.perf_counter()
changed_files = sorted(planned_changes.keys())
if args.archive_mode:
direct_write(planned_changes)
else:
transactional_write(planned_changes)
_log_stage_duration("源码写入", write_stage_started_at)
if args.verbose_changes:
for path in changed_files:
print(f"{path}")
for sym, cnt in planned_changes[path]["stats"].items():
target = display_mapping.get(sym, mapping.get(sym, sym))
print(f" {sym}{target} ×{cnt}")
else:
print(f" 已省略 {len(changed_files)} 个变更文件明细;如需查看可加 --verbose-changes")
print(f"\n✅ 完成!修改了 {total_changed}/{len(all_targets)} 个文件。")
print(f" 映射表: {args.map_output}")
print(" 恢复 crash 符号: python3 AutoPacking/obfuscation/obfuscate_symbols.py --deobfuscate your.crash")
_log_stage_duration("本次执行总耗时", total_started_at)
if __name__ == "__main__":
try:
main()
except Exception:
import traceback
tb = traceback.format_exc()
sys.stderr.write("FATAL: 未捕获异常,详细信息写入 obfuscate_error.log\n")
try:
Path('obfuscate_error.log').write_text(tb, encoding='utf-8')
except Exception:
pass
sys.stderr.write(tb)
sys.exit(2)