Update app.py

app.py

@@ -8,6 +8,7 @@ import shutil
 import zipfile
 import base64
 import asyncio
+import time
 import subprocess
 from datetime import datetime
 from typing import List, Dict, Any, Tuple
@@ -25,14 +26,12 @@ os.makedirs(OUTPUT_DIR, exist_ok=True)
 
 templates = Jinja2Templates(directory="templates")
 
-# الامتدادات المظلمة المستبعدة توفيراً للوقت
 IGNORED_EXTENSIONS = {
     '.png', '.jpg', '.jpeg', '.gif', '.bmp', '.webp', '.ico',
     '.mp4', '.avi', '.mkv', '.mov', '.flv', '.wmv', '.mp3', '.wav', '.ogg',
     '.ttf', '.otf', '.woff', '.woff2', '.eot', '.pdf'
 }
 
-# المجلدات الواهية المستبعدة (تصاميم وأيقونات لا تحتوي كوداً)
 IGNORED_SUBDIRS = [
     'res/drawable', 'res/color', 'res/anim', 'res/mipmap', 'res/values-'
 ]
@@ -67,7 +66,7 @@ class ConnectionManager:
             "message": message,
             "progress": progress
         }
-        for connection in self.active_connections:
+        for connection in list(self.active_connections):
             try:
                 await connection.send_json(log_entry)
             except Exception:
@@ -85,26 +84,18 @@ def clean_workspace():
                 elif os.path.isdir(file_path):
                     shutil.rmtree(file_path)
             except Exception as e:
-                print(f"Cleanup error for {file_path}: {e}")
-
-def calculate_entropy(data: bytes) -> float:
-    if not data:
-        return 0.0
-    entropy = 0.0
-    length = len(data)
-    byte_counts = [0] * 256
-    for b in data:
-        byte_counts[b] += 1
-    for count in byte_counts:
-        if count == 0:
-            continue
-        p = count / length
-        entropy -= p * math.log2(p)
-    return entropy
+                pass
+
+def extract_zip_sync(zip_path: str, extract_to: str) -> List[str]:
+    """فك ضغط الملف بمسار مستقل عن Event Loop لمنع التجمّد"""
+    with zipfile.ZipFile(zip_path, 'r') as zip_ref:
+        file_list = zip_ref.namelist()
+        zip_ref.extractall(extract_to)
+    return file_list
 
 def extract_strings_from_binary_file(file_path: str) -> str:
     try:
-        res = subprocess.run(['strings', '-n', '4', file_path], capture_output=True, text=True, errors='ignore')
+        res = subprocess.run(['strings', '-n', '4', file_path], capture_output=True, text=True, errors='ignore', timeout=5)
         return res.stdout
     except Exception:
         try:
@@ -115,36 +106,10 @@ def extract_strings_from_binary_file(file_path: str) -> str:
         except Exception:
             return ""
 
-def attempt_decompression_and_xor(data: bytes) -> List[Tuple[str, str]]:
-    extracted_texts = []
-    try:
-        decompressed = zlib.decompress(data)
-        extracted_texts.append(("Zlib Decompressed", decompressed.decode('utf-8', errors='ignore')))
-    except Exception:
-        pass
-
-    try:
-        decompressed = gzip.decompress(data)
-        extracted_texts.append(("Gzip Decompressed", decompressed.decode('utf-8', errors='ignore')))
-    except Exception:
-        pass
-
-    for key in XOR_KEYS:
-        try:
-            xor_bytes = bytes([b ^ key for b in data[:50000]])
-            decoded = xor_bytes.decode('utf-8', errors='ignore')
-            if any(k in decoded.lower() for k in ['http', 'discord', 'key', 'license', 'stuart']):
-                extracted_texts.append((f"XOR Key (0x{key:02X})", decoded))
-        except Exception:
-            pass
-
-    return extracted_texts
-
 def extract_snippet(text: str, start: int, end: int, margin: int = 50) -> str:
     s_start = max(0, start - margin)
     s_end = min(len(text), end + margin)
-    snippet = text[s_start:s_end].replace('\n', ' ').replace('\r', ' ')
-    return f"...{snippet}..."
+    return f"...{text[s_start:s_end].replace('\n', ' ').replace('\r', ' ')}..."
 
 def classify_severity(match_type: str, match_value: str) -> str:
     val_lower = match_value.lower()
@@ -180,22 +145,23 @@ async def analyze_zip(file: UploadFile = File(...)):
     extract_folder = os.path.join(WORKSPACE_DIR, "extracted")
     os.makedirs(extract_folder, exist_ok=True)
 
-    await manager.broadcast_log("INFO", f"استقبال الملف على الخادم: {file.filename}...", 5)
+    # 1. حفظ الملف بسلاسة لتجنب احتكار الذاكرة
+    await manager.broadcast_log("INFO", f"جاري استلام الملف: {file.filename}...", 5)
     with open(zip_path, "wb") as buffer:
-        while chunk := await file.read(1024 * 1024):
+        while chunk := await file.read(2 * 1024 * 1024):  # قراءة بإنتاجية 2MB
             buffer.write(chunk)
 
-    await manager.broadcast_log("INFO", "جاري فك ضغط الأرشيف...", 15)
+    # 2. فك الضغط بـ Thread خفي لمنع تجمّد السيرفر عند 12%
+    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)
+        zip_file_list = await asyncio.to_thread(extract_zip_sync, zip_path, 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. تجميع الملفات المفلترة
     files_to_process = []
-    max_file_size = 50 * 1024 * 1024
+    max_file_size = 30 * 1024 * 1024  # حد أقصى 30MB للملف الواحد لتفادي البطء
 
     for root, _, files in os.walk(extract_folder):
         for f in files:
@@ -206,7 +172,6 @@ async def analyze_zip(file: UploadFile = File(...)):
             full_path = os.path.join(root, f)
             rel_p = os.path.relpath(full_path, extract_folder).replace('\\', '/')
 
-            # استبعاد مجلدات الموارد غير البرمجية لتسريع الفحص
             if any(rel_p.startswith(sub) for sub in IGNORED_SUBDIRS):
                 continue
 
@@ -224,16 +189,22 @@ async def analyze_zip(file: UploadFile = File(...)):
         "medium_priority_count": 0,
         "low_priority_count": 0,
         "matches_by_type": {
-            "DISCORD": 0, "URL": 0, "EMAIL": 0, "KEYWORD": 0, "BASE64_DECODED": 0, "UNPACKED_XOR": 0
+            "DISCORD": 0, "URL": 0, "EMAIL": 0, "KEYWORD": 0, "BASE64_DECODED": 0
         }
     }
 
+    last_log_time = time.time()
+
+    # 4. مسح وتسريع الفحص
     for idx, file_path in enumerate(files_to_process):
         rel_path = os.path.relpath(file_path, extract_folder)
 
-        if idx % 15 == 0 or idx == total_files - 1:
+        # إرسال تحديث السجل كل 200 ملي ثانية فقط لحماية المتصفح من التجمد
+        current_time = time.time()
+        if (current_time - last_log_time > 0.2) or idx == total_files - 1:
             progress = 30 + int((idx / total_files) * 65)
-            await manager.broadcast_log("INFO", f"جاري تفتيش الملف ({idx + 1}/{total_files}): {rel_path}", progress)
+            await manager.broadcast_log("INFO", f"تفتيش ({idx + 1}/{total_files}): {rel_path}", progress)
+            last_log_time = current_time
             await asyncio.sleep(0.001)
 
         file_findings = []
@@ -242,7 +213,6 @@ async def analyze_zip(file: UploadFile = File(...)):
             with open(file_path, 'rb') as f:
                 raw_bytes = f.read()
 
-            file_entropy = calculate_entropy(raw_bytes)
             is_binary = b'\x00' in raw_bytes[:1024] or rel_path.endswith(('.so', '.dex', '.dat'))
 
             if is_binary:
@@ -288,7 +258,7 @@ async def analyze_zip(file: UploadFile = File(...)):
                 stats[f"{sev.lower()}_priority_count"] += 1
 
             # Base64
-            for b64_match in BASE64_PATTERN.findall(content_text)[:10]:
+            for b64_match in BASE64_PATTERN.findall(content_text)[:5]:
                 try:
                     decoded = base64.b64decode(b64_match).decode('utf-8', errors='ignore')
                     if any(k in decoded.lower() for k in ['http', 'discord', 'key', 'stuart', 'license', 'verify']):
@@ -303,20 +273,6 @@ async def analyze_zip(file: UploadFile = File(...)):
                 except Exception:
                     pass
 
-            if file_entropy > 7.0:
-                unpacked = attempt_decompression_and_xor(raw_bytes)
-                for method, text_data in unpacked:
-                    for match in DISCORD_PATTERN.finditer(text_data):
-                        val = match.group()
-                        file_findings.append({
-                            "type": "UNPACKED_XOR",
-                            "value": f"[{method}] {val}",
-                            "snippet": f"مستخرج عبر {method}: {val}",
-                            "severity": "HIGH"
-                        })
-                        stats["matches_by_type"]["UNPACKED_XOR"] += 1
-                        stats["high_priority_count"] += 1
-
             if file_findings:
                 stats["total_matches"] += len(file_findings)
                 findings_by_file.append({
@@ -325,7 +281,7 @@ async def analyze_zip(file: UploadFile = File(...)):
                     "results": file_findings
                 })
 
-        except Exception as e:
+        except Exception:
             pass
 
     await manager.broadcast_log("SUCCESS", f"اكتمل الفحص بنجاح! إجمالي التطابقات المكتشفة: {stats['total_matches']}", 100)