path: root/files/TG_bot_VPN-main/utils/generate_vpn_keys.py

import base64
import os
import json
import uuid
import tempfile
import paramiko
import qrcode
import secrets
import random
from cryptography.hazmat.primitives.asymmetric import x25519
from cryptography.hazmat.primitives import serialization
from loader import app_logger
from database.models import VPNKey, Server
import copy
from config_data.config import (
    DEFAULT_SERVER_USER,
    DEFAULT_SERVER_PASSWORD,
    XRAY_CONFIG_PATH,
    QR_CODE_DIR,
    XRAY_REALITY_FINGERPRINT,
    DOMAINS_LIST,
)

def generate_x25519_keys_base64() -> dict:
    """
    Генерирует пару ключей X25519 и возвращает их в base64-формате,
    что соответствует требованиям Xray Reality.
    """
    private_key = x25519.X25519PrivateKey.generate()
    public_key = private_key.public_key()
    private_bytes = private_key.private_bytes(
         encoding=serialization.Encoding.Raw,
         format=serialization.PrivateFormat.Raw,
         encryption_algorithm=serialization.NoEncryption()
    )
    public_bytes = public_key.public_bytes(
         encoding=serialization.Encoding.Raw,
         format=serialization.PublicFormat.Raw
    )
    private_b64 = base64.b64encode(private_bytes).decode('utf-8')
    public_b64 = base64.b64encode(public_bytes).decode('utf-8')
    return {"private": private_b64, "public": public_b64}


def remote_generate_xray_keys(server_obj: Server) -> dict:
    cmd = "xray x25519"
    output = execute_ssh_command(
        ip=server_obj.ip_address,
        username=DEFAULT_SERVER_USER,
        password=DEFAULT_SERVER_PASSWORD,
        command=cmd,
        timeout=30
    )
    keys = {}
    for line in output.splitlines():
        if line.lower().startswith("private"):
            keys["private"] = line.split(":", 1)[1].strip()
        elif line.lower().startswith("public"):
            keys["public"] = line.split(":", 1)[1].strip()
    if "private" in keys and "public" in keys:
        return keys
    else:
        app_logger.error("Не удалось сгенерировать ключи через 'xray x25519' на сервере.")
        return None


def ensure_reality_params(config_template: dict, server_obj: Server) -> tuple:
    # Выбираем случайный домен из DOMAINS_LIST
    domain = random.choice(DOMAINS_LIST)
    # Генерируем ключи (приватный и публичный) на сервере
    keys = remote_generate_xray_keys(server_obj)
    if not keys:
        raise Exception("Удаленная генерация ключей не удалась")
    # Генерируем список shortIds (например, 6 значений)
    short_ids = [secrets.token_hex(4) for _ in range(6)]

    # Создаем глубокую копию шаблона, чтобы не изменять оригинал
    config = copy.deepcopy(config_template)

    # Обновляем параметры в секции realitySettings
    reality = config["inbounds"][0]["streamSettings"]["realitySettings"]
    reality["dest"] = f"{domain}:443"
    reality["serverNames"] = [domain]
    reality["privateKey"] = keys["private"]
    reality["publicKey"] = keys["public"]    # <-- Добавляем поле publicKey
    reality["shortIds"] = short_ids

    # Обновляем домены в routing.rules (если указаны)
    for rule in config.get("routing", {}).get("rules", []):
        if "domain" in rule:
            rule["domain"] = [domain]

    # Возвращаем обновленный конфиг и публичный ключ для формирования VLESS-ссылки
    return config, keys["public"]


SECURE_XRAY_CONFIG = {
    "log": {
        "loglevel": "debug"
    },
    "inbounds": [
        {
            "listen": "0.0.0.0",
            "port": 443,
            "protocol": "vless",
            "settings": {
                "clients": [],
                "decryption": "none"
            },
            "streamSettings": {
                "network": "tcp",
                "security": "reality",
                "realitySettings": {
                    "dest": "",
                    "serverNames": [],
                    "privateKey": "",
                    "shortIds": [],
                    "publicKey": ""
                }
            },
            "sniffing": {
                "enabled": True,
                "destOverride": ["http", "tls", "quic"]
            }
        }
    ],
    "outbounds": [
        {
            "protocol": "freedom",
            "tag": "direct"
        },
        {
            "protocol": "blackhole",
            "tag": "block"
        }
    ]
}


def execute_ssh_command(ip: str, username: str, password: str, command: str, timeout: int = 60) -> str:
    client = paramiko.SSHClient()
    try:
        client.load_system_host_keys()
        client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        client.connect(
            hostname=ip,
            username=username,
            password=password,
            timeout=10,
            allow_agent=False,
            look_for_keys=False
        )
        stdin, stdout, stderr = client.exec_command(command, timeout=timeout, get_pty=True)
        output = stdout.read().decode('utf-8').strip()
        err = stderr.read().decode('utf-8').strip()
        if err:
            app_logger.error(f"Ошибка выполнения команды на {ip}: {err}")
        return output
    except Exception as ex:
        app_logger.error(f"Ошибка при подключении к {ip}: {ex}")
        return ""
    finally:
        client.close()




def setup_server(server_obj: Server) -> bool:
    """
    Настраивает сервер для работы VPN.
    Если возникают какие-то ошибки при установке пакетов -
    https://timeweb.cloud/docs/unix-guides/troubleshooting-unix/ustranenie-oshibki-could-not-get-lock-var-lib-dpkg-lock

    Алгоритм:
      1. Подключение по SSH с правами администратора.
      2. Проверка наличия DEFAULT_SERVER_USER. Если отсутствует, создаём пользователя и устанавливаем пароль.
         Для установки пароля используется команда chpasswd, что исключает необходимость интерактивного ввода.
      3. С использованием SFTP загружается обновлённый и безопасный конфиг Xray.
      4. Перезапуск службы Xray.

    :param server_obj: объект Server с данными для подключения.
    :return: True, если настройка прошла успешно, иначе False.
    """
    try:
        # Подключаемся к серверу под администратором
        with paramiko.SSHClient() as ssh_admin:
            ssh_admin.load_system_host_keys()
            ssh_admin.set_missing_host_key_policy(paramiko.AutoAddPolicy())
            ssh_admin.connect(
                hostname=server_obj.ip_address,
                username=server_obj.username,
                password=server_obj.password,
                timeout=10,
                allow_agent=False,
                look_for_keys=False
            )
            # Проверяем наличие пользователя DEFAULT_SERVER_USER
            check_cmd = f"id {DEFAULT_SERVER_USER}"
            stdin, stdout, stderr = ssh_admin.exec_command(check_cmd, timeout=30)
            error_output = stderr.read().decode('utf-8').lower()
            if "no such user" in error_output:
                app_logger.info(
                    f"Пользователь {DEFAULT_SERVER_USER} не найден на сервере {server_obj.location}. Создаем его.")
                # Создаем пользователя
                ssh_admin.exec_command(f"useradd -m -s /bin/bash -G sudo {DEFAULT_SERVER_USER}", timeout=20)
                # Устанавливаем пароль с помощью chpasswd (без интерактивного ввода)
                passwd_cmd = f'echo "{DEFAULT_SERVER_USER}:{DEFAULT_SERVER_PASSWORD}" | chpasswd'
                ssh_admin.exec_command(passwd_cmd, timeout=20)
                app_logger.info(f"Пользователь {DEFAULT_SERVER_USER} создан и пароль установлен.")
            else:
                app_logger.info(f"Пользователь {DEFAULT_SERVER_USER} уже существует")
        update_jq_cmd = (
            f"echo {DEFAULT_SERVER_PASSWORD} | sudo -S apt update && "
            f"echo {DEFAULT_SERVER_PASSWORD} | sudo -S apt upgrade -y && "
            f"echo {DEFAULT_SERVER_PASSWORD} | sudo -S apt install -y jq"
        )
        execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=update_jq_cmd,
            timeout=300
        )
        app_logger.info(f"Системные компоненты обновлены, зависимости установлены.")
        xray_check_cmd = "which xray"
        xray_installed = execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=xray_check_cmd,
            timeout=30
        )
        if not xray_installed:
            app_logger.info("Xray не найден, начинаю установку с использованием официального скрипта.")
            install_xray_cmd = (
                "curl -O https://raw.githubusercontent.com/XTLS/Xray-install/main/install-release.sh && "
                f"echo {DEFAULT_SERVER_PASSWORD} | sudo -S bash install-release.sh"
            )
            output_install_xray = execute_ssh_command(
                ip=server_obj.ip_address,
                username=DEFAULT_SERVER_USER,
                password=DEFAULT_SERVER_PASSWORD,
                command=install_xray_cmd,
                timeout=300
            )
            app_logger.info(f"Установка Xray выполнена успешно. Вывод: {output_install_xray}")
        else:
            app_logger.info("Xray уже установлен. Сервер настроен")
            return True

        # Разрешаем 443 порт
        allow_443_port_command = (
            f"echo {DEFAULT_SERVER_PASSWORD} | sudo -S ufw allow 443/tcp && "
            f"echo {DEFAULT_SERVER_PASSWORD} | sudo -S netstat -tulpn | grep ':443'"
        )
        port_output = execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=allow_443_port_command,
            timeout=300
        )
        app_logger.info(f"Порт 443 разрешен. Проверка: {port_output}")
        # Подключаемся для загрузки конфигурации Xray
        ssh = paramiko.SSHClient()
        ssh.load_system_host_keys()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(
            hostname=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            timeout=10,
            allow_agent=False,
            look_for_keys=False
        )
        sftp = ssh.open_sftp()

        # заполняем конфигурацию данными
        final_config, public_key = ensure_reality_params(SECURE_XRAY_CONFIG, server_obj)

        # Сохраняем public key в модель сервера
        server_obj.public_key = public_key
        server_obj.save()

        # Сохраняем локально безопасный конфиг Xray
        local_config_path = os.path.join(tempfile.gettempdir(), f"secure_xray_config_{server_obj.id}.json")
        with open(local_config_path, "w") as f:
            json.dump(final_config, f, indent=2)
        # Загружаем конфиг на сервер

        temp_remote_path = f"/tmp/secure_xray_config_{server_obj.id}.json"
        sftp.put(local_config_path, temp_remote_path)
        app_logger.info(f"Конфигурационный файл загружен во временную директорию: {temp_remote_path}")

        # Перемещаем файл с помощью sudo, чтобы обойти ограничение прав доступа.
        move_cmd = f'echo {DEFAULT_SERVER_PASSWORD} | sudo -S mv {temp_remote_path} {XRAY_CONFIG_PATH}'
        move_output = execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=move_cmd,
            timeout=30
        )
        app_logger.info(f"Файл конфигурации перемещён в {XRAY_CONFIG_PATH}. Вывод: {move_output}")
        sftp.close()
        ssh.close()

        # Проверяем конфиг файл на валидность
        test_config = f'echo {DEFAULT_SERVER_PASSWORD} | sudo -S xray run -test -confdir /usr/local/etc/xray/'
        test_output = execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=test_config,
            timeout=30
        )
        app_logger.info(f"Конфиг проверен. Вывод: {test_output}")

        # Синхронизируем время на сервере
        time_syn_cmd = f'echo {DEFAULT_SERVER_PASSWORD} | sudo -S timedatectl set-ntp true'
        execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=time_syn_cmd,
            timeout=30
        )
        app_logger.info(f"Время на сервере синхронизировано.")

        # Перезапускаем службу Xray для применения конфигурации
        restart_cmd = f'echo {DEFAULT_SERVER_PASSWORD} | sudo -S systemctl restart xray'
        restart_output = execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=restart_cmd,
            timeout=30
        )
        app_logger.info(f"Служба Xray перезапущена. Вывод: {restart_output}")
        return True
    except Exception as ex:
        app_logger.error(f"Ошибка при настройке сервера {server_obj.location}: {ex}")
        return False


def generate_key(server_obj: Server) -> VPNKey | None:
    """
    Генерирует новый VPN ключ для сервера.

    Алгоритм:
      1. Генерация уникального UUID для клиента.
      2. Обновление конфигурации Xray на сервере с помощью утилиты jq (добавление нового клиента в список).
      3. Перезапуск службы Xray.
      4. Формирование VLESS-ссылки с параметрами Xray Reality.
      5. Генерация QR-кода для VLESS-ссылки.
      6. Создание записи VPNKey в базе данных.

    :param server_obj: объект Server.
    :return: объект VPNKey с рабочей VLESS‑ссылкой и путем к QR‑коду, либо None при ошибке.
    """
    try:
        # Шаг 1. Генерация UUID для нового клиента
        client_uuid = str(uuid.uuid4())
        app_logger.info(f"Сгенерирован новый UUID для клиента: {client_uuid}")

        # Шаг 2. Обновление конфигурации Xray через jq
        update_cmd = (
            f'echo {DEFAULT_SERVER_PASSWORD} | sudo -S sh -c '
            f'"jq \'.inbounds[0].settings.clients += [{{\\"id\\": \\"{client_uuid}\\", '
            f'\\"flow\\": \\"xtls-rprx-vision\\"}}]\' {XRAY_CONFIG_PATH} > {XRAY_CONFIG_PATH}.tmp && '
            f'mv {XRAY_CONFIG_PATH}.tmp {XRAY_CONFIG_PATH}"'
        )
        update_output = execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=update_cmd,
            timeout=30
        )
        app_logger.info(f"Конфигурация Xray обновлена. Вывод: {update_output}")

        # Шаг 3. Перезапуск службы Xray
        restart_cmd = f'echo {DEFAULT_SERVER_PASSWORD} | sudo -S systemctl restart xray'
        restart_output = execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=restart_cmd,
            timeout=30
        )
        app_logger.info(f"Служба Xray перезапущена. Вывод: {restart_output}")

        # Шаг 4. Формирование VLESS-ссылки
        # Получаем данные из конфиг файла
        config_content = execute_ssh_command(
            ip=server_obj.ip_address,
            username=DEFAULT_SERVER_USER,
            password=DEFAULT_SERVER_PASSWORD,
            command=f"cat {XRAY_CONFIG_PATH}",
            timeout=30
        )
        try:
            config_json = json.loads(config_content)
            # Извлекаем данные из секции realitySettings во втором inbound (индекс 1)
            reality_settings = config_json["inbounds"][0]["streamSettings"]["realitySettings"]
            # Предполагаем, что в конфиге теперь есть оба параметра: publicKey и serverNames
            server_name = reality_settings.get("serverNames", [None])[0]
            public_key = server_obj.public_key
            short_id = random.choice(reality_settings["shortIds"])
            if not server_name or not public_key or not short_id:
                raise ValueError("Недостаточно параметров Xray Reality в конфиге.")
        except Exception as e:
            app_logger.error(f"Ошибка при парсинге конфигурационного файла: {e}")
            return None

        vless_link = (
            f"vless://{client_uuid}@{server_obj.ip_address}:443?"
            f"security=reality&"
            f"encryption=none&"
            f"flow=xtls-rprx-vision&"
            f"type=tcp&"
            f"fp={XRAY_REALITY_FINGERPRINT}&"
            f"sni={server_name}&"
            f"pbk={public_key}&"
            f"sid={short_id}#GuardVPN"
        )
        app_logger.info(f"Сформирована VLESS ссылка: {vless_link}")


        # Шаг 5. Генерация QR-кода
        key_number = len(server_obj.keys) + 1 if hasattr(server_obj, "keys") else 1
        qr_code_filename = f"vpn_key_{server_obj.id}_{key_number}.png"
        qr_code_path = os.path.join(QR_CODE_DIR, qr_code_filename)
        qr = qrcode.QRCode(
            version=1,
            error_correction=qrcode.constants.ERROR_CORRECT_H,
            box_size=10,
            border=4
        )
        qr.add_data(vless_link)
        qr.make(fit=True)
        img = qr.make_image(fill_color="black", back_color="white")
        img.save(qr_code_path)
        app_logger.info(f"QR-код сгенерирован и сохранён по пути: {qr_code_path}")

        # Шаг 6. Создание записи VPNKey в БД с использованием peewee.
        vpn_key = VPNKey.create(
            server=server_obj,
            name=f"VPN Key {server_obj.location} #{key_number}",
            key=vless_link,
            qr_code=qr_code_path,
            is_valid=True
        )
        app_logger.info(f"VPN ключ успешно создан: {vpn_key.key}")
        return vpn_key
    except Exception as ex:
        app_logger.error(f"Ошибка при генерации VPN ключа для сервера {server_obj.location}: {ex}")
        return None