Update: 2026-07-05 22:59:19
This commit is contained in:
@@ -0,0 +1,206 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
# ==============================================================================
|
||||
# Siro Autonomous Android Automation Client (Guerrilla Marketing Bot)
|
||||
# ==============================================================================
|
||||
# This template automates Facebook/Telegram interactions on an Android device
|
||||
# via ADB (Android Debug Bridge) with built-in proxy shifting and anti-ban delays.
|
||||
#
|
||||
# Requirements:
|
||||
# 1. Python 3.x
|
||||
# 2. Android device with USB Debugging enabled, connected to PC.
|
||||
# 3. ADB installed and added to System PATH.
|
||||
# 4. Target apps (Facebook/Telegram) installed and logged in.
|
||||
# ==============================================================================
|
||||
|
||||
import os
|
||||
import sys
|
||||
import time
|
||||
import random
|
||||
import subprocess
|
||||
import requests
|
||||
|
||||
# --- Configuration ---
|
||||
SERVER_URL = "http://jordan-siro.intaleqapp.com/backend/marketing_engine/social_worker.php"
|
||||
BOT_TOKEN = "YOUR_SECRET_BOT_TOKEN" # Must match headers in social_worker.php
|
||||
PLATFORM = "facebook" # 'facebook' or 'telegram'
|
||||
CHECK_INTERVAL_SECONDS = 60 # Cooldown between checking for new tasks
|
||||
|
||||
headers = {
|
||||
"X-Bot-Token": BOT_TOKEN
|
||||
}
|
||||
|
||||
# --- ADB Helper Functions ---
|
||||
def run_adb(cmd):
|
||||
"""Run an ADB command and return output."""
|
||||
try:
|
||||
result = subprocess.run(f"adb {cmd}", shell=True, capture_output=True, text=True, timeout=15)
|
||||
return result.stdout.strip()
|
||||
except subprocess.TimeoutExpired:
|
||||
print("[-] ADB command timed out.")
|
||||
return ""
|
||||
|
||||
def set_android_proxy(ip, port, username=None, password=None):
|
||||
"""Dynamically set global HTTP proxy on Android device."""
|
||||
if not ip or not port:
|
||||
print("[+] Clearing Android proxy...")
|
||||
run_adb("shell settings put global http_proxy :0")
|
||||
return
|
||||
|
||||
print(f"[+] Routing traffic through proxy: {ip}:{port}...")
|
||||
# Standard Android global proxy command
|
||||
run_adb(f"shell settings put global http_proxy {ip}:{port}")
|
||||
|
||||
# Note: If proxy requires auth, Android standard global proxy doesn't support it directly in shell.
|
||||
# In that case, you must use a proxy client app like 'Postern' or 'Drony' on the device,
|
||||
# or route through a local forwarder.
|
||||
time.sleep(3)
|
||||
|
||||
def adb_click(x, y):
|
||||
"""Click on coordinates (x, y)."""
|
||||
run_adb(f"shell input tap {x} {y}")
|
||||
time.sleep(random.uniform(1.0, 2.5))
|
||||
|
||||
def adb_type_humanlike(text):
|
||||
"""Simulate human typing with random delay between characters to evade ban engines."""
|
||||
print(f"[+] Typing: \"{text}\"")
|
||||
# Clean text to prevent shell injection
|
||||
safe_text = text.replace('"', '\\"').replace('$', '\\$').replace('`', '\\`').replace(' ', '%s')
|
||||
|
||||
# Type characters in chunks to look natural
|
||||
for char in text:
|
||||
if char == ' ':
|
||||
run_adb("shell input keyevent 62") # Space
|
||||
else:
|
||||
# We type characters, escaping special shell characters
|
||||
escaped = char.replace('"', '\\"').replace('$', '\\$').replace('`', '\\`').replace("'", "\\'")
|
||||
run_adb(f"shell input text '{escaped}'")
|
||||
time.sleep(random.uniform(0.05, 0.25)) # Typing speed variability
|
||||
|
||||
time.sleep(random.uniform(1.0, 2.0))
|
||||
|
||||
def adb_scroll_down():
|
||||
"""Perform a human-like swipe down."""
|
||||
# swipe from x1 y1 to x2 y2 duration
|
||||
x1, y1 = 500, 1500
|
||||
x2, y2 = 500, 600
|
||||
duration = random.randint(300, 800)
|
||||
run_adb(f"shell input swipe {x1} {y1} {x2} {y2} {duration}")
|
||||
time.sleep(random.uniform(1.5, 3.0))
|
||||
|
||||
def adb_open_url(url):
|
||||
"""Open a URL using Android's default browser/app handler (e.g. opens deep links)."""
|
||||
print(f"[+] Launching URL: {url}")
|
||||
run_adb(f"shell am start -a android.intent.action.VIEW -d '{url}'")
|
||||
time.sleep(random.uniform(5.0, 8.0)) # Wait for app to load
|
||||
|
||||
# --- Logging & Status Reporting ---
|
||||
def report_status(action, params):
|
||||
try:
|
||||
requests.post(f"{SERVER_URL}?action={action}", data=params, headers=headers, timeout=10)
|
||||
except Exception as e:
|
||||
print(f"[-] Status report connection error: {e}")
|
||||
|
||||
# --- Main Task Loop ---
|
||||
def process_task():
|
||||
print("[*] Checking for pending tasks...")
|
||||
try:
|
||||
res = requests.get(f"{SERVER_URL}?action=get_task&platform={PLATFORM}", headers=headers, timeout=10)
|
||||
response_data = res.json()
|
||||
except Exception as e:
|
||||
print(f"[-] Connection failed: {e}")
|
||||
return
|
||||
|
||||
if response_data.get("status") != "success" or not response_data.get("data"):
|
||||
print("[*] No tasks scheduled at this moment.")
|
||||
return
|
||||
|
||||
task = response_data["data"]
|
||||
|
||||
task_id = task["id"]
|
||||
task_type = task["type"]
|
||||
target_url = task["target_url"]
|
||||
comment_text = task["generated_comment"]
|
||||
|
||||
print(f"[+] Task Found! ID: {task_id} | Type: {task_type}")
|
||||
print(f"[+] Account to use: {task.get('bot_username')}")
|
||||
|
||||
# 1. Shifting IP via Proxy Configuration
|
||||
set_android_proxy(
|
||||
task.get("proxy_ip"),
|
||||
task.get("proxy_port"),
|
||||
task.get("proxy_username"),
|
||||
task.get("proxy_password")
|
||||
)
|
||||
|
||||
success = False
|
||||
error_msg = "Unknown execution error"
|
||||
|
||||
try:
|
||||
# 2. Evading Automated Scanners via Random Pre-sleep
|
||||
pre_delay = random.randint(5, 15)
|
||||
print(f"[+] Simulating human delay before opening app... sleeping {pre_delay}s")
|
||||
time.sleep(pre_delay)
|
||||
|
||||
# 3. Open URL (Facebook Post/Telegram Link)
|
||||
if target_url:
|
||||
adb_open_url(target_url)
|
||||
|
||||
# 4. Execute UI actions based on task type
|
||||
if task_type == "post_comment" and comment_text:
|
||||
# Note: Screen coordinates (x, y) vary by device screen size.
|
||||
# You must customize these coordinates based on your device profile.
|
||||
print("[*] Performing UI interactions to comment...")
|
||||
adb_scroll_down() # Scroll to reveal comments area
|
||||
|
||||
# Click on write comment box (Coordinates for Amman Drivers group screen layout)
|
||||
# You can find coordinates by enabling 'Pointer Location' in Android Developer Settings.
|
||||
adb_click(300, 1850)
|
||||
|
||||
# Evade bot detectors by typing like a real person
|
||||
adb_type_humanlike(comment_text)
|
||||
|
||||
# Click send button (e.g. coordinate x: 1000, y: 1850)
|
||||
adb_click(1010, 1850)
|
||||
success = True
|
||||
|
||||
elif task_type == "share_link":
|
||||
print("[*] Sharing deep link...")
|
||||
# Click Share coordinates, type message, and click post
|
||||
# Example coordinates:
|
||||
adb_click(900, 1200) # Share button
|
||||
adb_click(500, 1500) # Write post option
|
||||
adb_type_humanlike(comment_text)
|
||||
adb_click(1000, 150) # Post button
|
||||
success = True
|
||||
|
||||
else:
|
||||
error_msg = f"Task type {task_type} is not yet supported in Android ADB Client"
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"Automation failure: {str(e)}"
|
||||
print(f"[-] Error: {error_msg}")
|
||||
|
||||
# 5. Clear proxy setting to keep device clean
|
||||
set_android_proxy(None, None)
|
||||
|
||||
# 6. Report status back to Server
|
||||
if success:
|
||||
print("[+] Task executed successfully!")
|
||||
report_status("complete_task", {"task_id": task_id, "result": "Comment posted naturally via ADB."})
|
||||
else:
|
||||
print(f"[-] Task failed: {error_msg}")
|
||||
report_status("fail_task", {"task_id": task_id, "error_message": error_msg})
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("[*] Starting Siro Guerrilla Marketing Automation Bot...")
|
||||
# Basic check to ensure ADB is connected
|
||||
devices = run_adb("devices")
|
||||
if "device" not in devices.split("\n")[1:]:
|
||||
print("[-] Critical Error: No Android device detected via ADB! Please connect device and enable USB Debugging.")
|
||||
sys.exit(1)
|
||||
|
||||
print("[+] Android device connected. Starting automation loop...")
|
||||
while True:
|
||||
process_task()
|
||||
time.sleep(CHECK_INTERVAL_SECONDS)
|
||||
Reference in New Issue
Block a user