Интеграция XRay (VLESS) в телеграм бота используя API Marzban
Новый протокол подключения
Ряд последних событий тонко намекает на то, что использование VPN в РФ с каждым днем будет только усложняться. К очередной блокировке нужно быть готовым заранее, поэтому добавим новый протокол VLESS на базе ядра XRay (V2Ray) в телеграм бота.
Полученный результат можно посмотреть в этом боте:
Немного теории
V2Ray, V2Fly, XRay (VMess, VLESS, XTLS)
XRay - это форк V2Fly, когда некоторые разработчики из-за ряда разногласий с остальным сообществом ушли из проекта V2Fly и продолжили развивать код параллельно под названием XRay, придумав ему слоган “Penetrates everything”, что очень недалеко от правды. Формат конфигурационных файлов остался прежним, но при этом новая реализация считается более эффективной в плане производительности, а самое главное - разработчики добавили туда несколько очень крутых фич, направленных в том числе на снижение детектируемости подключений на DPI (например, с помощью выявления TLS-in-TLS), таких как XTLS, речь о которых пойдет ниже.
V2Ray/XRay - это не протокол, а, можно сказать, фреймворк - разные протоколы с разными транспортами и расширениями под одной крышей в одном приложении. Идея простая: что клиент, что сервер - это один бинарник. В конфигурации задаются inbounds (обработчики входящих подключений) и outbound (обработчики исходящих подключений).
На клиенте inbound обычно будет работать как HTTP- или SOCKS-прокси сервер, принимая подключения от браузеров и других программ, а outbound будет настроен как клиент какого-нибудь прокси-протокола для подключения к удаленному серверу.
VLESS - это более новый протокол. В отличие от VMess он не предусматривает механизма шифрования (подразумевается, что шифрование должно производиться нижележащим транспортным протоколом, например TLS), а только проверку “свой/чужой” и паддинг данных (изменение размеров пакетов для затруднения детектирования паттернов траффика).
В протоколе исправлен ряд уязвимостей старого VMess, и он активно развивается - например, автор планирует добавить поддержку компрессии алгоритмом Zstd - не сколько для производительности, сколько для затруднения анализа “снаружи”. При этом, при установлении соединения (хендшейке) клиент и сервер обмениваются версией протокола и списком поддерживаемых фич, то есть при дальнейшем развитии должна сохраняться обратная совместимость. В общем и целом, на сегодняшний день это самый свежий и прогрессивный протокол.
Источник: https://habr.com/ru/articles/727868/
Интеграция
Далее я покажу на базовом примере, как можно организовать выдачу ключей для подключения через VLESS.
Допустим у вас есть телеграм бот, в котором вы генерируете ключи для ваших пользователей, ваш код выглядит примерно так.
Ниже очень упрощенный пример, если у вас уже есть бот - вы сможете адаптировать его под свою реализацию.
class MarzbanBackend:
def __init__(self):
self.session = requests.Session()
self.headers = {"accept": "application/json"}
self.base_url = S.MARZBAN_API_URL
if not self.headers.get("Authorization"):
self.authorize()
def _get(self, path: str) -> dict:
url = f"{self.base_url}/{path}"
response = self.session.request("GET", url, verify=False, headers=self.headers)
if response.status_code == 200:
return response.json()
def _post(self, path: str, data=None) -> dict:
url = f"{self.base_url}/{path}"
if not path == "api/admin/token":
data = json.dumps(data)
response = self.session.request("POST", url, verify=False, headers=self.headers, data=data)
if response.status_code == 201 or response.status_code == 200:
return response.json()
def _put(self, path: str, data=None) -> dict:
url = f"{self.base_url}/{path}"
json_data = json.dumps(data)
response = self.session.put(url, verify=False, headers=self.headers, data=json_data)
if response.status_code == 200:
logger.info(f"cmd xray PUT {path}, data: {data}")
return response.json()
else:
logger.error(f"cmd xray PUT not 200 status_code! {path}, data: {data}")
def authorize(self) -> None:
data = {
"username": S.MARZBAN_USER,
"password": S.MARZBAN_PASSWORD
}
response = self._post("api/admin/token", data=data)
token = response.get("access_token")
self.headers["Authorization"] = f"Bearer {token}"
def create_user(self, name: str) -> dict:
data = {
"username": name,
"proxies": {
"vless": {
"flow": "xtls-rprx-vision",
},
},
"inbounds": {
"vless": ["VLESS TCP REALITY"],
},
"data_limit": 15 * 1024 * 1024 * 1024,
"data_limit_reset_strategy": "day",
}
response = self._post("api/user", data=data)
return response
def get_user(self, name: str) -> dict:
response = self._get(f"api/user/{name}")
user = response.get("username")
status = response.get("status")
logger.info(f"get user: {user}, status: {status}")
return response
def disable_user(self, name: str) -> dict:
data = {
"status": "disabled"
}
response = self._put(f"api/user/{name}", data=data)
if response:
logger.info(f"Disable xray user: {name} success, {response.get('username', 'unknown username')}")
check = self.get_user(name)
time.sleep(0.25)
if check.get("status") != data.get("status"):
logger.error(f"After disable user {name}, user is not disabled!")
return response
else:
logger.warning(f"xray user {name} not found")
return {}
def enable_user(self, name: str) -> dict:
data = {
"status": "active"
}
response = self._put(f"api/user/{name}", data=data)
if response:
logger.info(f"Enable xray user: {name} success, {response.get('username', 'unknown username')}")
return response
else:
logger.warning(f"xray user {name} not found")
return {}
Хотелось бы обратить ваше внимание на функции create_user() и disable_user(), рассмотрим каждую из них подробнее.
В функцию create_user() я не передаю параметр expire, так как в боте используется несколько протоколов и уже существует логика управления пользователями через включение-выключение доступа (функции enable_user(), disable_user()).
Но если у вас новый бот и вы не используете другие протоколы - управление через expire выглядит хорошим вариантом.
В функции disable_user() реализована дополнительная логика на проверку отключения в следующем виде:
time.sleep(0.25)
check = self.get_user(name)
if check.get("status") != data.get("status"):
logger.error(f"After disable user {name}, user is not disabled!")
В данный момент сталкиваюсь со случаями, в которых успешно меняется статус, но в панели Marzban пользователь остается со статусом “active”.
Пока этот вариант используется для отладки, если баг будет сохраняться - можно будет сделать рекурсивный вариант внутри функции.
После создания пользователя необходимо отдать ему ссылку на подписку, которая формируется функцией create_user(), ключ: subscription_url, если вы не планируете полностью использовать API Marzban для доступа к данным сервера - то результат (данные о пользователях) лучше сохранять в своё отдельно хранилище, например postgresql, с которой взаимодействует бот.
Итог
Результаты подключения протокола VLESS можете посмотреть в моём боте.
Cсылка: https://t.me/artydev_wg_payment_bot?start=xray
Также рекомендую: Marzban - добавить подписку в 1 клик
На пиво
Если данный материал оказался вам полезен - готов принять ваши копеечки :)
в крипте:
ETH (ERC20): 0xcdc3231527a1ad105d527678ccbcf5e827747e7b
TON: UQAiIMLC2_j9tPlmQakdbz2Zh0rkTHH7tK2RTcO3rYAkr8QV
в рублях: https://pay.cloudtips.ru/p/2a3d8e06
Большое спасибо всем за внимание! Если вам интересны подобные рассуждения - подписывайтесь на мой канал artydev & Co.