104 lines
4.7 KiB
Python
104 lines
4.7 KiB
Python
import base64
|
||
import secrets
|
||
import urllib.parse
|
||
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
|
||
|
||
from .runner import Result
|
||
from .server import Server
|
||
|
||
MAIN_INBOUND = 'main'
|
||
CF_WARP = 'cf_warp'
|
||
"""Исходящий прокси через Cloudflare WARP."""
|
||
PROXY_SCHEMES = {'socks': 'socks', 'socks5': 'socks', 'socks5h': 'socks', 'http': 'http', 'https': 'http'}
|
||
|
||
|
||
class XuiPanel(Server):
|
||
"""Сервер с панелью 3x-ui."""
|
||
|
||
def install(
|
||
self, username: str, password: str, panel_port: int, web_base_path: str, sub_port: int, sub_path: str,
|
||
domain: str = '', ssl: bool = True,
|
||
) -> Result:
|
||
"""Ставит панель, включает подписку на sub_port по пути /sub_path/ и DNS-over-TLS к Cloudflare.
|
||
|
||
Сертификат Let's Encrypt на domain, без него - на ip; не удалось выпустить - ошибка. ssl=False - панель на http,
|
||
для серверов без публичного адреса. Открывает порты 80 (acme), 443, panel_port и sub_port.
|
||
Возвращает data: {'api_token', 'panel_host' (домен или ip из сертификата), 'ssl'}.
|
||
"""
|
||
return self.run(
|
||
'xui',
|
||
username=username,
|
||
password=password,
|
||
panel_port=panel_port,
|
||
web_base_path=web_base_path,
|
||
sub_port=sub_port,
|
||
sub_path=sub_path,
|
||
domain=domain,
|
||
ssl=ssl,
|
||
)
|
||
|
||
def issue_certificate(self, domain: str = '') -> Result:
|
||
"""Выпускает сертификат панели и подписки на domain, без него - на ip. Возвращает data: {'panel_host'}.
|
||
|
||
Так домен ставится или меняется после установки. Домен должен указывать на сервер или его маскировку.
|
||
"""
|
||
return self.run('xui_cert', domain=domain)
|
||
|
||
def set_outbound(self, proxy: str = '') -> Result:
|
||
"""Весь исходящий трафик клиентов через прокси. Перед применением прокси проверяется панелью.
|
||
|
||
proxy - socks5://user:password@host:port, http(s)://..., CF_WARP или пустая строка, чтобы выходить напрямую.
|
||
"""
|
||
warp_keys = generate_wireguard_keys() if proxy == CF_WARP else None
|
||
proxy_outbound = parse_proxy(proxy) if proxy and proxy != CF_WARP else None
|
||
return self.run('xui_outbound', proxy_outbound=proxy_outbound, warp_keys=warp_keys)
|
||
|
||
def create_inbound(self, remark: str = MAIN_INBOUND, target: str = '') -> Result:
|
||
"""Создает подключение vless + xhttp + reality на 443 порту, uTLS firefox, сброс трафика ежемесячно.
|
||
|
||
target - цель reality (host:port) или список через запятую. Без него панель выбирает из своих кандидатов.
|
||
Уже созданное подключение с тем же remark не меняется.
|
||
Возвращает data: {'inbound_id', 'port', 'target', 'server_name', 'public_key', 'short_ids', 'path'}.
|
||
"""
|
||
return self.run(
|
||
'xui_inbound',
|
||
remark=remark,
|
||
target=target,
|
||
short_ids=generate_short_ids(),
|
||
xhttp_path=f'/{secrets.token_hex(8)}',
|
||
)
|
||
|
||
|
||
def generate_short_ids(count: int = 4) -> list[str]:
|
||
"""shortId reality: hex длиной от 2 до 16 символов."""
|
||
return [secrets.token_hex(secrets.randbelow(8) + 1) for _ in range(count)]
|
||
|
||
|
||
def parse_proxy(url: str) -> dict:
|
||
"""outbound Xray для прокси вида socks5://user:password@host:port или http(s)://... ValueError, если не разобрать."""
|
||
parts = urllib.parse.urlsplit(url)
|
||
protocol = PROXY_SCHEMES.get(parts.scheme)
|
||
if not protocol or not parts.hostname or not parts.port:
|
||
raise ValueError(f'expected socks5://user:password@host:port or http(s)://..., got {parts.scheme}://...')
|
||
server = {'address': parts.hostname, 'port': parts.port}
|
||
if parts.username:
|
||
server['users'] = [{
|
||
'user': urllib.parse.unquote(parts.username),
|
||
'pass': urllib.parse.unquote(parts.password or ''),
|
||
}]
|
||
outbound = {'tag': 'proxy', 'protocol': protocol, 'settings': {'servers': [server]}}
|
||
if parts.scheme == 'https':
|
||
outbound['streamSettings'] = {'security': 'tls', 'tlsSettings': {'serverName': parts.hostname}}
|
||
return outbound
|
||
|
||
|
||
def generate_wireguard_keys() -> dict:
|
||
key = X25519PrivateKey.generate()
|
||
private = key.private_bytes(
|
||
serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption(),
|
||
)
|
||
public = key.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
|
||
return {'private_key': base64.b64encode(private).decode(), 'public_key': base64.b64encode(public).decode()}
|