""" APK Deep Diff Analyzer - Main Analysis Engine """ import os import zipfile import re from typing import Dict, Any, List, Tuple, Set from concurrent.futures import ThreadPoolExecutor, as_completed from utils import ( get_file_hash, get_file_size_formatted, extract_apk, list_files_recursive, get_zip_entries, calculate_similarity, create_temp_dir, cleanup_temp_dir, get_apk_info ) from manifest_parser import compare_manifests, ManifestAnalyzer from dex_analyzer import analyze_all_dex, find_dex_files, DexAnalyzer from signature_analyzer import compare_signatures, SignatureAnalyzer class APKDeepAnalyzer: """Main engine for deep APK analysis and comparison.""" def __init__(self, orig_apk: str, mod_apk: str, progress_callback=None): self.orig_apk = orig_apk self.mod_apk = mod_apk self.progress_callback = progress_callback self.orig_temp_dir = None self.mod_temp_dir = None self.results = {} def _update_progress(self, stage: int, message: str = ""): """Update progress via callback.""" if self.progress_callback: self.progress_callback(stage, message) def analyze(self) -> Dict[str, Any]: """Run complete analysis pipeline.""" try: self._update_progress(0, "جاري رفع الملفات...") # Stage 1: Validate APKs self._update_progress(1, "التحقق من صحة ملفات APK...") if not self._validate_apk(self.orig_apk): raise ValueError("ملف APK الأصلي غير صالح (Invalid Original APK)") if not self._validate_apk(self.mod_apk): raise ValueError("ملف APK المعدل غير صالح (Invalid Modified APK)") # Stage 2: Extract APKs self._update_progress(2, "فك ضغط ملفات APK...") self.orig_temp_dir = create_temp_dir("orig_") self.mod_temp_dir = create_temp_dir("mod_") extract_apk(self.orig_apk, self.orig_temp_dir) extract_apk(self.mod_apk, self.mod_temp_dir) # Stage 3: ZIP structure analysis self._update_progress(3, "تحليل بنية ZIP...") orig_zip = get_zip_entries(self.orig_apk) mod_zip = get_zip_entries(self.mod_apk) # Stage 4: File comparison self._update_progress(4, "مقارنة الملفات...") file_comparison = self._compare_files() # Stage 5: Manifest analysis self._update_progress(5, "تحليل AndroidManifest.xml...") manifest_comparison = self._analyze_manifest() # Stage 6: DEX analysis self._update_progress(6, "تحليل ملفات DEX...") dex_analysis = self._analyze_dex() # Stage 7: Native libraries self._update_progress(7, "تحليل المكتبات الأصلية...") native_libs = self._analyze_native_libraries() # Stage 8: Resources self._update_progress(8, "تحليل الموارد...") resources = self._analyze_resources() # Stage 9: Signature self._update_progress(9, "تحليل التوقيع...") signature_comparison = compare_signatures(self.orig_apk, self.mod_apk) # Stage 10: Strings & URLs self._update_progress(10, "تحليل النصوص والروابط...") strings_analysis = self._analyze_strings() # Stage 11: Suspicious changes self._update_progress(11, "البحث عن تغييرات مشبوهة...") suspicious = self._detect_suspicious( manifest_comparison, dex_analysis, native_libs, signature_comparison, file_comparison ) # Compile results self.results = { "original_info": get_apk_info(self.orig_apk), "modified_info": get_apk_info(self.mod_apk), "zip_comparison": { "original_entries": len(orig_zip), "modified_entries": len(mod_zip), }, "file_comparison": file_comparison, "manifest_comparison": manifest_comparison, "dex_analysis": dex_analysis, "native_libraries": native_libs, "resources": resources, "signature_comparison": signature_comparison, "strings_analysis": strings_analysis, "suspicious_changes": suspicious, } self._update_progress(12, "اكتمل التحليل!") return self.results except Exception as e: self.results = {"error": str(e)} return self.results finally: self._cleanup() def _validate_apk(self, apk_path: str) -> bool: """Validate APK file.""" if not os.path.exists(apk_path): return False if not zipfile.is_zipfile(apk_path): return False try: with zipfile.ZipFile(apk_path, 'r') as zf: # Check for AndroidManifest.xml if "AndroidManifest.xml" not in zf.namelist(): return False # Check for at least one DEX file has_dex = any(name.endswith('.dex') for name in zf.namelist()) if not has_dex: return False return True except Exception: return False def _compare_files(self) -> Dict[str, Any]: """Compare all files between two extracted APKs.""" orig_files = list_files_recursive(self.orig_temp_dir) mod_files = list_files_recursive(self.mod_temp_dir) orig_map = {path: (size, sha) for path, size, sha in orig_files} mod_map = {path: (size, sha) for path, size, sha in mod_files} all_paths = set(orig_map.keys()) | set(mod_map.keys()) comparison = { "added": [], "removed": [], "modified": [], "unchanged": [], "details": [], } for path in sorted(all_paths): in_orig = path in orig_map in_mod = path in mod_map if in_orig and not in_mod: comparison["removed"].append(path) comparison["details"].append({ "path": path, "original_size": orig_map[path][0], "modified_size": 0, "original_sha256": orig_map[path][1], "modified_sha256": "", "status": "Removed", }) elif in_mod and not in_orig: comparison["added"].append(path) comparison["details"].append({ "path": path, "original_size": 0, "modified_size": mod_map[path][0], "original_sha256": "", "modified_sha256": mod_map[path][1], "status": "Added", }) else: orig_size, orig_sha = orig_map[path] mod_size, mod_sha = mod_map[path] if orig_sha == mod_sha: comparison["unchanged"].append(path) status = "Unchanged" else: comparison["modified"].append(path) status = "Modified" comparison["details"].append({ "path": path, "original_size": orig_size, "modified_size": mod_size, "original_sha256": orig_sha, "modified_sha256": mod_sha, "status": status, "similarity": calculate_similarity(orig_size, mod_size, orig_sha, mod_sha), }) comparison["summary"] = { "total": len(all_paths), "added": len(comparison["added"]), "removed": len(comparison["removed"]), "modified": len(comparison["modified"]), "unchanged": len(comparison["unchanged"]), } return comparison def _analyze_manifest(self) -> Dict[str, Any]: """Analyze and compare AndroidManifest.xml.""" orig_manifest = os.path.join(self.orig_temp_dir, "AndroidManifest.xml") mod_manifest = os.path.join(self.mod_temp_dir, "AndroidManifest.xml") if os.path.exists(orig_manifest) and os.path.exists(mod_manifest): return compare_manifests(orig_manifest, mod_manifest) return {"error": "AndroidManifest.xml not found"} def _analyze_dex(self) -> Dict[str, Any]: """Analyze DEX files.""" orig_dex = analyze_all_dex(self.orig_temp_dir) mod_dex = analyze_all_dex(self.mod_temp_dir) # Compare DEX files orig_files = set(orig_dex["dex_files"].keys()) mod_files = set(mod_dex["dex_files"].keys()) dex_comparison = { "original_count": orig_dex["dex_count"], "modified_count": mod_dex["dex_count"], "added": sorted(list(mod_files - orig_files)), "removed": sorted(list(orig_files - mod_files)), "common": sorted(list(orig_files & mod_files)), } # Compare common DEX files for fname in dex_comparison["common"]: orig_header = orig_dex["dex_files"][fname].get("header", {}) mod_header = mod_dex["dex_files"][fname].get("header", {}) if orig_header != mod_header: dex_comparison["modified"] = dex_comparison.get("modified", []) + [fname] return { "dex_count": max(orig_dex["dex_count"], mod_dex["dex_count"]), "dex_files": {**orig_dex["dex_files"], **mod_dex["dex_files"]}, "comparison": dex_comparison, "all_strings": list(set(orig_dex["all_strings"]) | set(mod_dex["all_strings"])), "all_urls": list(set(orig_dex["all_urls"]) | set(mod_dex["all_urls"])), "all_domains": list(set(orig_dex["all_domains"]) | set(mod_dex["all_domains"])), "all_ips": list(set(orig_dex["all_ips"]) | set(mod_dex["all_ips"])), "all_suspicious": list(set(orig_dex["all_suspicious"]) | set(mod_dex["all_suspicious"])), } def _analyze_native_libraries(self) -> Dict[str, Any]: """Analyze native libraries.""" def get_libs(temp_dir: str) -> Dict[str, Dict[str, Any]]: libs = {} lib_dir = os.path.join(temp_dir, "lib") if os.path.exists(lib_dir): for arch in os.listdir(lib_dir): arch_dir = os.path.join(lib_dir, arch) if os.path.isdir(arch_dir): for f in os.listdir(arch_dir): if f.endswith('.so'): filepath = os.path.join(arch_dir, f) libs[f"{arch}/{f}"] = { "path": filepath, "size": os.path.getsize(filepath), "sha256": get_file_hash(filepath), "architecture": arch, } return libs orig_libs = get_libs(self.orig_temp_dir) mod_libs = get_libs(self.mod_temp_dir) all_keys = set(orig_libs.keys()) | set(mod_libs.keys()) result = { "original": list(orig_libs.keys()), "modified": list(mod_libs.keys()), "added": [], "removed": [], "modified": [], "details": [], } for key in sorted(all_keys): if key in mod_libs and key not in orig_libs: result["added"].append(key) result["details"].append({ "name": key, "status": "Added", "modified_size": mod_libs[key]["size"], }) elif key in orig_libs and key not in mod_libs: result["removed"].append(key) result["details"].append({ "name": key, "status": "Removed", "original_size": orig_libs[key]["size"], }) else: if orig_libs[key]["sha256"] != mod_libs[key]["sha256"]: result["modified"].append(key) status = "Modified" else: status = "Unchanged" result["details"].append({ "name": key, "status": status, "original_size": orig_libs[key]["size"], "modified_size": mod_libs[key]["size"], "original_sha256": orig_libs[key]["sha256"], "modified_sha256": mod_libs[key]["sha256"], }) return result def _analyze_resources(self) -> Dict[str, Any]: """Analyze resources (res, assets, resources.arsc).""" def analyze_dir(temp_dir: str, dir_name: str) -> Dict[str, Any]: result = {"files": [], "count": 0} target = os.path.join(temp_dir, dir_name) if os.path.exists(target): for root, dirs, files in os.walk(target): for f in files: filepath = os.path.join(root, f) rel = os.path.relpath(filepath, target) result["files"].append({ "name": f"{dir_name}/{rel}", "size": os.path.getsize(filepath), "sha256": get_file_hash(filepath), }) result["count"] = len(result["files"]) return result orig_res = analyze_dir(self.orig_temp_dir, "res") mod_res = analyze_dir(self.mod_temp_dir, "res") orig_assets = analyze_dir(self.orig_temp_dir, "assets") mod_assets = analyze_dir(self.mod_temp_dir, "assets") # resources.arsc orig_arsc = os.path.join(self.orig_temp_dir, "resources.arsc") mod_arsc = os.path.join(self.mod_temp_dir, "resources.arsc") arsc_info = { "original_exists": os.path.exists(orig_arsc), "modified_exists": os.path.exists(mod_arsc), "original_size": os.path.getsize(orig_arsc) if os.path.exists(orig_arsc) else 0, "modified_size": os.path.getsize(mod_arsc) if os.path.exists(mod_arsc) else 0, } return { "res": {"original": orig_res, "modified": mod_res}, "assets": {"original": orig_assets, "modified": mod_assets}, "resources_arsc": arsc_info, } def _analyze_strings(self) -> Dict[str, Any]: """Extract and compare strings from both APKs.""" # Collect all strings from file comparison orig_strings = set() mod_strings = set() # Extract strings from text files def extract_strings_from_dir(temp_dir: str) -> Set[str]: strings = set() for root, dirs, files in os.walk(temp_dir): for f in files: if f.endswith(('.xml', '.json', '.txt', '.properties', '.cfg', '.ini')): try: filepath = os.path.join(root, f) with open(filepath, 'r', encoding='utf-8', errors='ignore') as file: content = file.read() # Extract URLs urls = re.findall(r'https?://[^\s<>"{}|\^`\[\]]+', content, re.IGNORECASE) strings.update(urls) # Extract domains domains = re.findall(r'(?:[a-zA-Z0-9](?:[a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}', content, re.IGNORECASE) strings.update(domains) # Extract IPs ips = re.findall(r'\b(?:(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(?:25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\b', content) strings.update(ips) except: pass return strings orig_strings = extract_strings_from_dir(self.orig_temp_dir) mod_strings = extract_strings_from_dir(self.mod_temp_dir) return { "added": sorted(list(mod_strings - orig_strings)), "removed": sorted(list(orig_strings - mod_strings)), "common": sorted(list(orig_strings & mod_strings)), } def _detect_suspicious(self, manifest: Dict, dex: Dict, native: Dict, signature: Dict, files: Dict) -> List[str]: """Detect potentially suspicious changes.""" suspicious = [] # New permissions perms = manifest.get("changes", {}).get("permissions", {}) dangerous_perms = [ "android.permission.SEND_SMS", "android.permission.READ_SMS", "android.permission.RECEIVE_SMS", "android.permission.CALL_PHONE", "android.permission.READ_PHONE_STATE", "android.permission.ACCESS_FINE_LOCATION", "android.permission.RECORD_AUDIO", "android.permission.CAMERA", "android.permission.WRITE_EXTERNAL_STORAGE", "android.permission.READ_EXTERNAL_STORAGE", "android.permission.SYSTEM_ALERT_WINDOW", "android.permission.BIND_ACCESSIBILITY_SERVICE", "android.permission.REQUEST_INSTALL_PACKAGES", ] for perm in perms.get("added", []): if any(dp in perm for dp in dangerous_perms): suspicious.append(f"صلاحية خطيرة مضافة: {perm}") # Debug enabled pkg_changes = manifest.get("changes", {}).get("package_info", {}) if "debuggable" in pkg_changes: debug_orig = str(pkg_changes["debuggable"].get("original", "")).lower() debug_mod = str(pkg_changes["debuggable"].get("modified", "")).lower() if debug_orig in ["false", ""] and debug_mod == "true": suspicious.append("تم تفعيل وضع التصحيح (debuggable=true)") # Cleartext traffic if "usesCleartextTraffic" in pkg_changes: suspicious.append("تم تغيير إعدادات نقل البيانات غير المشفرة") # New services/receivers svc_changes = manifest.get("changes", {}).get("services", {}) if svc_changes.get("added"): for svc in svc_changes["added"]: suspicious.append(f"خدمة جديدة (Service): {svc.get('name', 'Unknown')}") rec_changes = manifest.get("changes", {}).get("receivers", {}) if rec_changes.get("added"): for rec in rec_changes["added"]: suspicious.append(f"مستقبل جديد (Receiver): {rec.get('name', 'Unknown')}") # New native libraries if native.get("added"): for lib in native["added"]: suspicious.append(f"مكتبة Native جديدة: {lib}") # Signature changed if signature.get("signature_changed"): suspicious.append("تغيير في توقيع APK (Signature Changed)") # Suspicious APIs in DEX suspicious_apis = dex.get("all_suspicious", []) if suspicious_apis: suspicious.append(f"تم العثور على {len(suspicious_apis)} استدعاء API مشبوه في DEX") # New URLs/domains strings = self._analyze_strings() new_urls = strings.get("added", []) suspicious_domains = ['.tk', '.ml', '.ga', '.cf', '.gq'] for url in new_urls: if any(sd in url for sd in suspicious_domains): suspicious.append(f"نطاق مشبوه جديد: {url}") return suspicious def _cleanup(self): """Clean up temporary files.""" if self.orig_temp_dir: cleanup_temp_dir(self.orig_temp_dir) if self.mod_temp_dir: cleanup_temp_dir(self.mod_temp_dir)