Spaces:
Build error
Build error
File size: 6,381 Bytes
46c28e3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | import eventlet
eventlet.monkey_patch()
from flask import Flask, render_template, request, jsonify
from flask_socketio import SocketIO, emit
import os, time, subprocess, shutil, json, threading, random
from agents import Agents # استيراد الوكلاء الذين أنشأناهم للتو
app = Flask(__name__)
app.config['SECRET_KEY'] = 'emmanuel_scripts_buster'
socketio = SocketIO(app, cors_allowed_origins="*", async_mode='eventlet')
UPLOAD_FOLDER = '/tmp/apk_analysis'
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
MAX_ATTEMPTS = 10000 # عدد المحاولات
# ------------------ واجهات التوجيه ------------------
@app.route('/')
def index():
return render_template_string(HTML_TEMPLATE) # HTML_TEMPLATE سنعرفه لاحقاً
@app.route('/upload', methods=['POST'])
def upload_file():
if 'file' not in request.files: return jsonify({"error": "لم يتم رفع ملف"})
file = request.files['file']
if file.filename == '': return jsonify({"error": "اسم الملف فارغ"})
filepath = os.path.join(UPLOAD_FOLDER, file.filename)
file.save(filepath)
socketio.emit('log_update', {'data': f'[✅] تم رفع الملف: {file.filename} (الحجم: ~180MB)'})
# تشغيل الخيط الرئيسي للتحليل
thread = threading.Thread(target=run_full_analysis, args=(filepath,))
thread.start()
return jsonify({"message": "بدء التحليل الشامل..."})
# ------------------ محرك التحليل المتقدم ------------------
def run_full_analysis(apk_path):
socketio.emit('status_update', {'status': 'decompiling'})
output_dir = apk_path.replace('.apk', '_decomp')
# 1. محاولة فك الضغط (APKTool)
for attempt in range(MAX_ATTEMPTS):
try:
socketio.emit('log_update', {'data': f'[Loop {attempt+1}] محاولة فك الضغط عبر APKTool...'})
subprocess.run(['apktool', 'd', apk_path, '-o', output_dir, '-f'], check=True, capture_output=True, text=True)
socketio.emit('log_update', {'data': '[✅ APKTool] فك الضغط بنجاح!'})
break
except Exception as e:
socketio.emit('log_update', {'data': f'[⚠️ APKTool] فشل، إعادة المحاولة... ({e})'})
if attempt == MAX_ATTEMPTS - 1:
socketio.emit('log_update', {'data': '[❌ APKTool] فشل بعد 10000 محاولة، توقف.'})
return
# 2. البحث عن النصوص في الملفات الثنائية (Strings) وخارجها
socketio.emit('log_update', {'data': '[👁️ Agent 0] بحث مركزي عن النصوص (ONE STATE, VIP ACCESS, ENTRAR)...'})
search_results = subprocess.run(f'grep -rnw "{output_dir}" -e "ONE STATE" -e "VIP ACCESS" -e "ENTRAR"', shell=True, capture_output=True, text=True)
socketio.emit('log_update', {'data': f'[📄] نتائج البحث:\n{search_results.stdout if search_results.stdout else "لم نجد بعد"}'})
# 3. تحليل الـ SO (المكتبات الثنائية)
lib_dir = os.path.join(output_dir, 'lib')
if os.path.exists(lib_dir):
socketio.emit('log_update', {'data': '[🧬] فحص ملفات .so (تحليل البايت كود الثنائي)...'})
for root, dirs, files in os.walk(lib_dir):
for file in files:
if file.endswith('.so'):
so_path = os.path.join(root, file)
socketio.emit('log_update', {'data': f'[SO] معالجة: {file}'})
# إذا كان ملف مشفر، نبدأ بـ Ghidra headless analysis
# مثال: subprocess.run(['/opt/ghidra/support/analyzeHeadless', so_path...])
# بالنسبة للمحاولات الـ 10000، نضع دالة في حلقة تحاول فك التشفير بطرق مختلفة حتى تنجح
# 4. تفعيل الوكيل الذكي 1 (تحليل Smali)
socketio.emit('status_update', {'status': 'ai_analysis'})
agents = Agents()
ai_response_raw = agents.agent_1_binary_analyzer(output_dir)
socketio.emit('log_update', {'data': f'[🤖 Agent 1 - Groq] تقرير التحليل الأولي:\n{ai_response_raw}'})
try:
ai_response = json.loads(ai_response_raw) # تحويل JSON
if ai_response['found']:
patch_filename = os.path.join(output_dir, ai_response['filename'])
# قراءة محتوى الملف الأصلي لإرساله للوكيل 3
with open(patch_filename, 'r', encoding='utf-8', errors='ignore') as f:
original_smali = f.read()
socketio.emit('log_update', {'data': f'[📝] تأهيل لتعديل الملف: {patch_filename}'})
# 5. تفعيل الوكيل الذكي 3 (كتابة التعديل)
patched_code = agents.agent_3_patcher(ai_response_raw, original_smali)
socketio.emit('log_update', {'data': f'[🤖 Agent 3 - Groq] الكود المعدل المُقترح:\n{patched_code}'})
# تعديل الملف الحقيقي (افتراضيًا نقوم بتنفيذ كود الوكيل 3)
if "```smali" in patched_code:
# استخراج الكود ونقله للملف (تطبيق عملي)
pass
except Exception as e:
socketio.emit('log_update', {'data': f'[❌] خطأ في معالجة نتيجة الذكاء الاصطناعي: {e}'})
# 6. إعادة بناء الـ APK المعدل (Patch APK)
socketio.emit('status_update', {'status': 'building'})
output_apk = '/tmp/final_patched_game.apk'
try:
subprocess.run(['apktool', 'b', output_dir, '-o', output_apk], check=True)
socketio.emit('log_update', {'data': f'[✅] تم إعادة بناء الـ APK في: {output_apk}'})
socketio.emit('status_update', {'status': 'complete', 'download_url': '/download_patched'})
except Exception as e:
socketio.emit('log_update', {'data': f'[❌] فشل عملية إعادة البناء: {e}'})
@app.route('/download_patched')
def download_file():
return send_file('/tmp/final_patched_game.apk', as_attachment=True)
if __name__ == '__main__':
socketio.run(app, host='0.0.0.0', port=7860) |