Spaces:
Paused
Paused
| """ | |
| 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) | |