Update app.py

app.py

@@ -1,34 +1,40 @@
 import os
 import re
 import json
+import shutil
+import zipfile
+import base64
 import asyncio
 from datetime import datetime
 from typing import List, Dict, Any
 
-from fastapi import FastAPI, UploadFile, File, Request, WebSocket, WebSocketDisconnect
+from fastapi import FastAPI, UploadFile, File, Form, Request, WebSocket, WebSocketDisconnect
 from fastapi.responses import HTMLResponse, FileResponse
 from fastapi.templating import Jinja2Templates
 
-app = FastAPI(title="Multi-SO Binary String Extractor")
+app = FastAPI(title="Decompiled APK ZIP Analyzer")
 
-WORKSPACE_DIR = "/tmp/so_workspace"
-OUTPUT_DIR = "/tmp/so_output"
+WORKSPACE_DIR = "/tmp/zip_workspace"
+OUTPUT_DIR = "/tmp/zip_output"
 os.makedirs(WORKSPACE_DIR, exist_ok=True)
 os.makedirs(OUTPUT_DIR, exist_ok=True)
 
 templates = Jinja2Templates(directory="templates")
 
-# الكلمات المفتاحية المطلوبة للبحث
-TARGET_KEYWORDS = [
-    "https://stuartmods.xyz",
-    "discord",
-    "discord.gg",
-    "stuartmods.xyz",
-    "stuartmods",
-    "StURTMODS",
-    ".xyz"
+# تعابير نمطية لفك الأنماط (Regex Patterns)
+URL_PATTERN = re.compile(r'https?://[^\s<>"{}|\\^`]+|www\.[^\s<>"{}|\\^`]+')
+EMAIL_PATTERN = re.compile(r'[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}')
+BASE64_PATTERN = re.compile(r'(?:[A-Za-z0-9+/]{4}){8,}(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?')
+HEX_PATTERN = re.compile(r'\b[0-9a-fA-F]{32,}\b')
+
+# تحديد الأولوية بناءً على الكلمات المفتاحية
+HIGH_PRIORITY_KEYWORDS = [
+    "discord.gg", "discord.com/invite", "stuartmods.xyz", "stuartmods",
+    "license", "activation", "verify", "key", "password", "token"
 ]
 
+MEDIUM_PRIORITY_KEYWORDS = ["api.", "http://", "https://", "secret", "auth"]
+
 class ConnectionManager:
     def __init__(self):
         self.active_connections: List[WebSocket] = []
@@ -41,11 +47,12 @@ class ConnectionManager:
         if websocket in self.active_connections:
             self.active_connections.remove(websocket)
 
-    async def broadcast_log(self, level: str, message: str):
+    async def broadcast_log(self, level: str, message: str, progress: int = 0):
         log_entry = {
             "timestamp": datetime.now().strftime("%H:%M:%S"),
             "level": level,
-            "message": message
+            "message": message,
+            "progress": progress
         }
         for connection in self.active_connections:
             try:
@@ -55,20 +62,54 @@ class ConnectionManager:
 
 manager = ConnectionManager()
 
-def extract_strings_from_binary(file_path: str, min_length: int = 3) -> List[str]:
-    """
-    استخراج كافة النصوص القابلة للقراءة (ASCII & UTF-8) من الملف الثنائي .so
-    """
-    strings_found = []
-    with open(file_path, "rb") as f:
-        content = f.read()
-        pattern = re.compile(rb'[\x20-\x7e]{' + str(min_length).encode() + rb',}')
-        for match in pattern.finditer(content):
+def clean_directories():
+    """تنظيف المجلدات المؤقتة لمنع الامتلاء"""
+    for folder in [WORKSPACE_DIR, OUTPUT_DIR]:
+        for filename in os.listdir(folder):
+            file_path = os.path.join(folder, filename)
             try:
-                strings_found.append(match.group().decode("ascii"))
-            except UnicodeDecodeError:
-                pass
-    return strings_found
+                if os.path.isfile(file_path) or os.path.islink(file_path):
+                    os.unlink(file_path)
+                elif os.path.isdir(file_path):
+                    shutil.rmtree(file_path)
+            except Exception as e:
+                print(f"Error cleaning {file_path}: {e}")
+
+def determine_severity(match_text: str) -> str:
+    """تحديد مستوى خطورة/أهمية النتيجة"""
+    text_lower = match_text.lower()
+    for kw in HIGH_PRIORITY_KEYWORDS:
+        if kw in text_lower:
+            return "HIGH"
+    for kw in MEDIUM_PRIORITY_KEYWORDS:
+        if kw in text_lower:
+            return "MEDIUM"
+    return "LOW"
+
+def extract_snippet(text: str, start_idx: int, end_idx: int, margin: int = 30) -> str:
+    """استخراج مقتطف نصي حول النتيجة مع سياقها"""
+    start = max(0, start_idx - margin)
+    end = min(len(text), end_idx + margin)
+    snippet = text[start:end].replace('\n', ' ').replace('\r', '')
+    return f"...{snippet}..."
+
+def decode_base64_strings(text: str) -> List[Dict[str, str]]:
+    """البحث عن نصوص Base64 ومحاولة فك تشفيرها"""
+    decoded_results = []
+    matches = BASE64_PATTERN.findall(text)
+    for b64_str in set(matches[:20]): # الحد الأقصى 20 لكل ملف للأداء
+        try:
+            decoded_bytes = base64.b64decode(b64_str)
+            decoded_text = decoded_bytes.decode('utf-8', errors='ignore')
+            # التأكد من أن النص المفكوك يحتوي على أحرف قابلة للقراءة
+            if len(decoded_text.strip()) > 6 and any(c.isalnum() for c in decoded_text):
+                decoded_results.append({
+                    "original": b64_str,
+                    "decoded": decoded_text
+                })
+        except Exception:
+            pass
+    return decoded_results
 
 @app.get("/", response_class=HTMLResponse)
 async def serve_index(request: Request):
@@ -83,75 +124,209 @@ async def websocket_logs(websocket: WebSocket):
     except WebSocketDisconnect:
         manager.disconnect(websocket)
 
-@app.post("/analyze-so")
-async def analyze_multiple_so(files: List[UploadFile] = File(...)):
-    so_files = [f for f in files if f.filename.endswith(".so")]
-    if not so_files:
-        return {"status": "error", "message": "الرجاء اختيار ملفات بصيغة .so فقط"}
-
-    await manager.broadcast_log("INFO", f"بدء استقبال {len(so_files)} ملف/ملفات .so...")
-
-    all_file_results = []
-    overall_keyword_summary = {kw: 0 for kw in TARGET_KEYWORDS}
-    total_matches_all_files = 0
-
-    for file in so_files:
-        file_path = os.path.join(WORKSPACE_DIR, file.filename)
-        content = await file.read()
-        with open(file_path, "wb") as buffer:
-            buffer.write(content)
-
-        file_size_mb = round(len(content) / (1024 * 1024), 2)
-        await manager.broadcast_log("INFO", f"جاري فحص: {file.filename} ({file_size_mb} MB)...")
-
-        # استخراج النصوص
-        extracted_strings = extract_strings_from_binary(file_path)
-        matches_by_kw = {kw: [] for kw in TARGET_KEYWORDS}
-        file_matches_count = 0
-
-        for s in extracted_strings:
-            for kw in TARGET_KEYWORDS:
-                if kw.lower() in s.lower():
-                    matches_by_kw[kw].append(s)
-                    file_matches_count += 1
-                    overall_keyword_summary[kw] += 1
-
-        total_matches_all_files += file_matches_count
-
-        await manager.broadcast_log(
-            "SUCCESS" if file_matches_count > 0 else "INFO",
-            f"الملف {file.filename}: تم استخراج {len(extracted_strings)} نص والعثور على {file_matches_count} تطابق."
-        )
-
-        all_file_results.append({
-            "filename": file.filename,
-            "file_size": f"{file_size_mb} MB",
-            "total_strings": len(extracted_strings),
-            "file_matches_count": file_matches_count,
-            "matches": matches_by_kw
-        })
-
-    await manager.broadcast_log("SUCCESS", f"اكتمل الفحص بالكامل! إجمالي التطابقات لجميع الملفات: {total_matches_all_files}")
-
-    # حفظ النتيجة في ملف JSON شامل
-    result_data = {
-        "total_files_scanned": len(so_files),
-        "total_matches_all_files": total_matches_all_files,
-        "overall_keyword_summary": overall_keyword_summary,
-        "files_details": all_file_results
+@app.post("/analyze-zip")
+async def analyze_zip(
+    file: UploadFile = File(...),
+    keywords: str = Form("discord.gg, discord.com/invite, stuartmods.xyz, stuartmods, license, activation, verify, key, password")
+):
+    if not file.filename.endswith(".zip"):
+        return {"status": "error", "message": "الرجاء رفع ملف بصيغة .zip فقط"}
+
+    clean_directories()
+
+    zip_path = os.path.join(WORKSPACE_DIR, "target.zip")
+    extract_folder = os.path.join(WORKSPACE_DIR, "extracted")
+    os.makedirs(extract_folder, exist_ok=True)
+
+    # 1. حفظ ملف الـ ZIP
+    await manager.broadcast_log("INFO", f"جاري استقبال الملف: {file.filename}...", 5)
+    with open(zip_path, "wb") as buffer:
+        while chunk := await file.read(1024 * 1024):
+            buffer.write(chunk)
+
+    # 2. فك الضغط
+    await manager.broadcast_log("INFO", "جاري فك ضغط المحتويات...", 15)
+    try:
+        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
+            zip_file_list = zip_ref.namelist()
+            zip_ref.extractall(extract_folder)
+        await manager.broadcast_log("SUCCESS", f"تم فك الضغط بنجاح! إجمالي الملفات الداخلية: {len(zip_file_list)}", 25)
+    except Exception as e:
+        return {"status": "error", "message": f"فشل فك ضغط الملف: {str(e)}"}
+
+    # 3. معالجة قائمة الكلمات المفتاحية من المستخدم
+    user_keywords = [k.strip().lower() for k in keywords.split(",") if k.strip()]
+
+    # 4. تجميع الملفات المستهدفة
+    target_extensions = ('.java', '.smali', '.xml', '.json', '.txt', '.html', '.properties', '.so')
+    files_to_scan = []
+
+    for root, _, files in os.walk(extract_folder):
+        for f in files:
+            if f.endswith(target_extensions):
+                files_to_scan.append(os.path.join(root, f))
+
+    total_files = len(files_to_scan)
+    await manager.broadcast_log("INFO", f"بدء الفحص العميق لـ {total_files} ملف مخصص...", 30)
+
+    results_by_file = []
+    summary_stats = {
+        "total_files_scanned": total_files,
+        "total_matches": 0,
+        "high_severity_count": 0,
+        "medium_severity_count": 0,
+        "low_severity_count": 0,
+        "keywords_found": {kw: 0 for kw in user_keywords}
     }
 
-    report_path = os.path.join(OUTPUT_DIR, "scan_result.json")
-    with open(report_path, "w", encoding="utf-8") as f:
-        json.dump(result_data, f, ensure_ascii=False, indent=2)
+    # 5. الفحص المباشر والسريع للملفات
+    for idx, file_full_path in enumerate(files_to_scan):
+        relative_path = os.path.relpath(file_full_path, extract_folder)
+        
+        # بث التقدم كل 20 ملف أو عند آخر ملف
+        if idx % 25 == 0 or idx == total_files - 1:
+            progress_pct = 30 + int((idx / total_files) * 60)
+            await manager.broadcast_log(
+                "INFO", 
+                f"جاري فحص الملف ({idx + 1}/{total_files}): {relative_path}", 
+                progress_pct
+            )
+            await asyncio.sleep(0.001)  # للسماح برفع السجلات عبر Event Loop
+
+        try:
+            with open(file_full_path, 'r', encoding='utf-8', errors='ignore') as f:
+                content = f.read()
+
+            file_matches = []
+
+            # أ) البحث عن الكلمات المفتاحية
+            content_lower = content.lower()
+            for kw in user_keywords:
+                if kw in content_lower:
+                    for match in re.finditer(re.escape(kw), content_lower):
+                        start, end = match.span()
+                        snippet = extract_snippet(content, start, end)
+                        severity = determine_severity(kw)
+                        
+                        file_matches.append({
+                            "type": "KEYWORD",
+                            "match": kw,
+                            "snippet": snippet,
+                            "severity": severity
+                        })
+                        summary_stats["keywords_found"][kw] += 1
+                        summary_stats[f"{severity.lower()}_severity_count"] += 1
+
+            # ب) استخراج الروابط URLs
+            urls = URL_PATTERN.findall(content)
+            for url in set(urls[:15]): # تجنب التكرار المفرط
+                severity = "HIGH" if "discord" in url.lower() or "stuartmods" in url.lower() else "MEDIUM"
+                file_matches.append({
+                    "type": "URL",
+                    "match": url,
+                    "snippet": url,
+                    "severity": severity
+                })
+                summary_stats[f"{severity.lower()}_severity_count"] += 1
+
+            # ج) استخراج البريد الإلكتروني Emails
+            emails = EMAIL_PATTERN.findall(content)
+            for email in set(emails[:10]):
+                file_matches.append({
+                    "type": "EMAIL",
+                    "match": email,
+                    "snippet": email,
+                    "severity": "MEDIUM"
+                })
+                summary_stats["medium_severity_count"] += 1
+
+            # د) فك نصوص Base64
+            decoded_b64 = decode_base64_strings(content)
+            for b64 in decoded_b64:
+                file_matches.append({
+                    "type": "BASE64_DECODED",
+                    "match": f"{b64['original']} -> {b64['decoded']}",
+                    "snippet": b64['decoded'],
+                    "severity": "MEDIUM"
+                })
+                summary_stats["medium_severity_count"] += 1
+
+            if file_matches:
+                summary_stats["total_matches"] += len(file_matches)
+                results_by_file.append({
+                    "file": relative_path,
+                    "matches_count": len(file_matches),
+                    "matches": file_matches
+                })
+
+        except Exception as e:
+            pass
+
+    await manager.broadcast_log("SUCCESS", f"اكتمل الفحص! تم العثور على {summary_stats['total_matches']} نتيجة.", 95)
+
+    # 6. حفظ التقارير (JSON و HTML)
+    report_data = {
+        "summary": summary_stats,
+        "details": results_by_file
+    }
+
+    json_report_path = os.path.join(OUTPUT_DIR, "analysis_report.json")
+    with open(json_report_path, "w", encoding="utf-8") as f:
+        json.dump(report_data, f, ensure_ascii=False, indent=2)
+
+    # توليد التقرير الـ HTML المستقل
+    generate_html_report(report_data, os.path.join(OUTPUT_DIR, "analysis_report.html"))
+
+    await manager.broadcast_log("SUCCESS", "تم إنشاء تقارير JSON و HTML جاهزة للتحميل.", 100)
+
+    return {"status": "success", "data": report_data}
+
+def generate_html_report(data: Dict[str, Any], save_path: str):
+    """إنشاء ملف HTML تفاعلي مستقل كتقرير قابل للتحميل"""
+    html_content = f"""<!DOCTYPE html>
+<html lang="ar" dir="rtl">
+<head>
+    <meta charset="UTF-8">
+    <title>APK Analysis Report</title>
+    <style>
+        body {{ font-family: system-ui, sans-serif; background: #0f172a; color: #f8fafc; padding: 20px; }}
+        .card {{ background: #1e293b; padding: 15px; border-radius: 8px; margin-bottom: 15px; border: 1px solid #334155; }}
+        h1 {{ color: #38bdf8; font-size: 1.4rem; }}
+        .badge {{ padding: 3px 8px; border-radius: 4px; font-weight: bold; font-size: 0.8rem; }}
+        .HIGH {{ background: #ef4444; color: white; }}
+        .MEDIUM {{ background: #f59e0b; color: black; }}
+        .LOW {{ background: #10b981; color: white; }}
+        .match-item {{ background: #020617; padding: 8px; border-radius: 4px; margin-top: 6px; font-family: monospace; font-size: 0.85rem; word-break: break-all; }}
+    </style>
+</head>
+<body>
+    <h1>تقرير فحص ملف APK المفكك</h1>
+    <div class="card">
+        <h3>الملفات المفحوصة: {data['summary']['total_files_scanned']} | إجمالي النتائج: {data['summary']['total_matches']}</h3>
+        <p>عالية الأهمية (HIGH): <span class="badge HIGH">{data['summary']['high_severity_count']}</span> | متوسطة (MEDIUM): <span class="badge MEDIUM">{data['summary']['medium_severity_count']}</span></p>
+    </div>
+"""
+    for file_res in data['details']:
+        html_content += f"""<div class="card">
+            <h4>📁 {file_res['file']} ({file_res['matches_count']} نتائج)</h4>"""
+        for m in file_res['matches']:
+            html_content += f"""<div class="match-item">
+                <span class="badge {m['severity']}">{m['severity']}</span> <b>[{m['type']}]</b>: {m['snippet']}
+            </div>"""
+        html_content += "</div>"
+
+    html_content += "</body></html>"
 
-    return {"status": "success", "data": result_data}
+    with open(save_path, "w", encoding="utf-8") as f:
+        f.write(html_content)
 
 @app.get("/download-report")
-async def download_report():
-    report_path = os.path.join(OUTPUT_DIR, "scan_result.json")
-    if os.path.exists(report_path):
-        return FileResponse(path=report_path, filename="multi_so_scan_report.json", media_type="application/json")
+async def download_report(fmt: str = "json"):
+    filename = "analysis_report.json" if fmt == "json" else "analysis_report.html"
+    file_path = os.path.join(OUTPUT_DIR, filename)
+    if os.path.exists(file_path):
+        media_type = "application/json" if fmt == "json" else "text/html"
+        return FileResponse(path=file_path, filename=filename, media_type=media_type)
     return {"error": "الملف غير موجود"}
 
 if __name__ == "__main__":