astelm
/
tg-ws-proxy
mirror of https://github.com/Flowseal/tg-ws-proxy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge upstream/main into android_migration

This commit is contained in:
Dark_Avery 2026-03-28 22:21:31 +03:00
parent e6cceac19f 0d11062c92
commit e50ea25d91
16 changed files with 1710 additions and 672 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
.dockerignore Normal file
View File

@ -0,0 +1,28 @@
.git
.github
.gitignore
__pycache__/
*.py[cod]
*.pyo
*.egg-info/
.pytest_cache/
.mypy_cache/
.ruff_cache/
.venv/
venv/
dist/
build/
packaging/
windows.py
icon.ico
*.spec
*.spec.bak
*.manifest
*.log
.vscode/
.idea/
*.swp
*.swo
.DS_Store
Thumbs.db
Desktop.ini

45
Dockerfile Normal file
View File

@ -0,0 +1,45 @@
# syntax=docker/dockerfile:1.7
FROM python:3.12-slim AS builder
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PIP_DISABLE_PIP_VERSION_CHECK=1 \
PIP_NO_CACHE_DIR=1 \
VIRTUAL_ENV=/opt/venv
RUN apt-get update \
&& apt-get install -y --no-install-recommends build-essential cargo libffi-dev libssl-dev \
&& python -m venv "$VIRTUAL_ENV" \
&& "$VIRTUAL_ENV/bin/pip" install --upgrade pip setuptools wheel \
&& rm -rf /var/lib/apt/lists/*
WORKDIR /app
RUN "$VIRTUAL_ENV/bin/pip" install cryptography==46.0.5
FROM python:3.12-slim AS runtime
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
PATH=/opt/venv/bin:$PATH \
TG_WS_PROXY_HOST=0.0.0.0 \
TG_WS_PROXY_PORT=1080 \
TG_WS_PROXY_DC_IPS="2:149.154.167.220 4:149.154.167.220"
RUN apt-get update \
&& apt-get install -y --no-install-recommends tini ca-certificates \
&& rm -rf /var/lib/apt/lists/* \
&& groupadd --system app \
&& useradd --system --gid app --create-home --home-dir /home/app app
WORKDIR /app
COPY --from=builder /opt/venv /opt/venv
COPY proxy ./proxy
COPY README.md LICENSE ./
USER app
EXPOSE 1080/tcp
ENTRYPOINT ["/usr/bin/tini", "--", "/bin/sh", "-lc", "set -eu; args=\"--host ${TG_WS_PROXY_HOST} --port ${TG_WS_PROXY_PORT}\"; for dc in ${TG_WS_PROXY_DC_IPS}; do args=\"$args --dc-ip $dc\"; done; exec python -u proxy/tg_ws_proxy.py $args \"$@\"", "--"]
CMD []

13
README.md
View File

@ -40,10 +40,12 @@ Telegram Desktop → SOCKS5 (127.0.0.1:1080) → TG WS Proxy → WSS → Telegra
- **Открыть в Telegram** — автоматически настроить прокси через `tg://socks` ссылку
- **Перезапустить прокси** — перезапуск без выхода из приложения
- **Настройки...** — GUI-редактор конфигурации
- **Настройки...** — GUI-редактор конфигурации (в т.ч. версия приложения, опциональная проверка обновлений с GitHub)
- **Открыть логи** — открыть файл логов
- **Выход** — остановить прокси и закрыть приложение
При первом запуске после старта может появиться запрос об открытии страницы релиза, если на GitHub вышла новая версия (отключается в настройках).
### macOS
Перейдите на [страницу релизов](https://github.com/Flowseal/tg-ws-proxy/releases) и скачайте **`TgWsProxy_macos_universal.dmg`** — универсальная сборка для Apple Silicon и Intel.
@ -254,15 +256,22 @@ Tray-приложение хранит данные в:
```json
{
"host": "127.0.0.1",
"port": 1080,
"dc_ip": [
"2:149.154.167.220",
"4:149.154.167.220"
],
"verbose": false
"verbose": false,
"buf_kb": 256,
"pool_size": 4,
"log_max_mb": 5.0,
"check_updates": true
}
```
Ключ **`check_updates`** — при `true` при запросе к GitHub сравнивается версия с последним релизом (только уведомление и ссылка на страницу загрузки). На Windows в конфиге может быть **`autostart`** (автозапуск при входе в систему).
## Автоматическая сборка
Проект содержит спецификации PyInstaller ([`packaging/windows.spec`](packaging/windows.spec), [`packaging/macos.spec`](packaging/macos.spec), [`packaging/linux.spec`](packaging/linux.spec)) и GitHub Actions workflow ([`.github/workflows/build.yml`](.github/workflows/build.yml)) для автоматической сборки.

BIN
icon.ico
View File

Binary file not shown.
Side by Side Swipe Overlay

Before

Width:  |  Height:  |  Size: 5.7 KiB

After

Width:  |  Height:  |  Size: 22 KiB

431
linux.py
View File

@ -7,6 +7,7 @@ import os
import subprocess
import sys
import threading
import webbrowser
import time
from pathlib import Path
from typing import Optional
@ -19,6 +20,23 @@ from PIL import Image, ImageDraw, ImageFont
import proxy.tg_ws_proxy as tg_ws_proxy
from proxy.app_runtime import ProxyAppRuntime
from proxy import __version__
from utils.default_config import default_tray_config
from ui.ctk_tray_ui import (
install_tray_config_buttons,
install_tray_config_form,
populate_first_run_window,
tray_settings_scroll_and_footer,
validate_config_form,
)
from ui.ctk_theme import (
CONFIG_DIALOG_FRAME_PAD,
CONFIG_DIALOG_SIZE,
FIRST_RUN_SIZE,
create_ctk_root,
ctk_theme_for_platform,
main_content_frame,
)
APP_NAME = "TgWsProxy"
APP_DIR = Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config")) / APP_NAME
@ -28,15 +46,7 @@ FIRST_RUN_MARKER = APP_DIR / ".first_run_done"
IPV6_WARN_MARKER = APP_DIR / ".ipv6_warned"
DEFAULT_CONFIG = {
"port": 1080,
"host": "127.0.0.1",
"dc_ip": ["2:149.154.167.220", "4:149.154.167.220"],
"verbose": False,
"log_max_mb": 5,
"buf_kb": 256,
"pool_size": 4,
}
DEFAULT_CONFIG = default_tray_config()
_tray_icon: Optional[object] = None
@ -225,9 +235,52 @@ def _show_info(text: str, title: str = "TG WS Proxy"):
root.destroy()
def _ask_yes_no_dialog(text: str, title: str = "TG WS Proxy") -> bool:
import tkinter as _tk
from tkinter import messagebox as _mb
root = _tk.Tk()
root.withdraw()
try:
root.attributes("-topmost", True)
except Exception:
pass
r = _mb.askyesno(title, text, parent=root)
root.destroy()
return bool(r)
def _maybe_notify_update_async():
def _work():
time.sleep(1.5)
if _exiting:
return
if not _config.get("check_updates", True):
return
try:
from utils.update_check import RELEASES_PAGE_URL, get_status, run_check
run_check(__version__)
st = get_status()
if not st.get("has_update"):
return
url = (st.get("html_url") or "").strip() or RELEASES_PAGE_URL
ver = st.get("latest") or "?"
text = (
f"Доступна новая версия: {ver}\n\n"
f"Открыть страницу релиза в браузере?"
)
if _ask_yes_no_dialog(text, "TG WS Proxy — обновление"):
webbrowser.open(url)
except Exception as exc:
log.debug("Update check failed: %s", exc)
threading.Thread(target=_work, daemon=True, name="update-check").start()
def _on_open_in_telegram(icon=None, item=None):
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
url = f"tg://socks?server=127.0.0.1&port={port}"
url = f"tg://socks?server={host}&port={port}"
log.info("Copying %s", url)
try:
@ -256,192 +309,36 @@ def _edit_config_dialog():
cfg = dict(_config)
ctk.set_appearance_mode("light")
ctk.set_default_color_theme("blue")
theme = ctk_theme_for_platform()
w, h = CONFIG_DIALOG_SIZE
root = ctk.CTk()
root.title("TG WS Proxy — Настройки")
root.resizable(False, False)
root.attributes("-topmost", True)
icon_img = _load_icon()
if icon_img:
from PIL import ImageTk
_photo = ImageTk.PhotoImage(icon_img.resize((64, 64)))
root.iconphoto(False, _photo)
TG_BLUE = "#3390ec"
TG_BLUE_HOVER = "#2b7cd4"
BG = "#ffffff"
FIELD_BG = "#f0f2f5"
FIELD_BORDER = "#d6d9dc"
TEXT_PRIMARY = "#000000"
TEXT_SECONDARY = "#707579"
FONT_FAMILY = "Sans"
w, h = 420, 540
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
root.geometry(f"{w}x{h}+{(sw - w) // 2}+{(sh - h) // 2}")
root.configure(fg_color=BG)
frame = ctk.CTkFrame(root, fg_color=BG, corner_radius=0)
frame.pack(fill="both", expand=True, padx=24, pady=20)
# Host
ctk.CTkLabel(
frame,
text="IP-адрес прокси",
font=(FONT_FAMILY, 13),
text_color=TEXT_PRIMARY,
anchor="w",
).pack(anchor="w", pady=(0, 4))
host_var = ctk.StringVar(value=cfg.get("host", "127.0.0.1"))
host_entry = ctk.CTkEntry(
frame,
textvariable=host_var,
width=200,
height=36,
font=(FONT_FAMILY, 13),
corner_radius=10,
fg_color=FIELD_BG,
border_color=FIELD_BORDER,
border_width=1,
text_color=TEXT_PRIMARY,
root = create_ctk_root(
ctk,
title="TG WS Proxy — Настройки",
width=w,
height=h,
theme=theme,
after_create=_apply_linux_ctk_window_icon,
)
host_entry.pack(anchor="w", pady=(0, 12))
# Port
ctk.CTkLabel(
frame,
text="Порт прокси",
font=(FONT_FAMILY, 13),
text_color=TEXT_PRIMARY,
anchor="w",
).pack(anchor="w", pady=(0, 4))
port_var = ctk.StringVar(value=str(cfg.get("port", 1080)))
port_entry = ctk.CTkEntry(
frame,
textvariable=port_var,
width=120,
height=36,
font=(FONT_FAMILY, 13),
corner_radius=10,
fg_color=FIELD_BG,
border_color=FIELD_BORDER,
border_width=1,
text_color=TEXT_PRIMARY,
fpx, fpy = CONFIG_DIALOG_FRAME_PAD
frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy)
scroll, footer = tray_settings_scroll_and_footer(ctk, frame, theme)
widgets = install_tray_config_form(
ctk, scroll, theme, cfg, DEFAULT_CONFIG,
show_autostart=False,
)
port_entry.pack(anchor="w", pady=(0, 12))
# DC-IP mappings
ctk.CTkLabel(
frame,
text="DC → IP маппинги (по одному на строку, формат DC:IP)",
font=(FONT_FAMILY, 13),
text_color=TEXT_PRIMARY,
anchor="w",
).pack(anchor="w", pady=(0, 4))
dc_textbox = ctk.CTkTextbox(
frame,
width=370,
height=120,
font=("Monospace", 12),
corner_radius=10,
fg_color=FIELD_BG,
border_color=FIELD_BORDER,
border_width=1,
text_color=TEXT_PRIMARY,
)
dc_textbox.pack(anchor="w", pady=(0, 12))
dc_textbox.insert("1.0", "\n".join(cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"])))
# Verbose
verbose_var = ctk.BooleanVar(value=cfg.get("verbose", False))
ctk.CTkCheckBox(
frame,
text="Подробное логирование (verbose)",
variable=verbose_var,
font=(FONT_FAMILY, 13),
text_color=TEXT_PRIMARY,
fg_color=TG_BLUE,
hover_color=TG_BLUE_HOVER,
corner_radius=6,
border_width=2,
border_color=FIELD_BORDER,
).pack(anchor="w", pady=(0, 8))
# Advanced: buf_kb, pool_size, log_max_mb
adv_frame = ctk.CTkFrame(frame, fg_color="transparent")
adv_frame.pack(anchor="w", fill="x", pady=(4, 8))
for col, (lbl, key, w_) in enumerate([
("Буфер (KB, 256 default)", "buf_kb", 120),
("WS пулов (4 default)", "pool_size", 120),
("Log size (MB, 5 def)", "log_max_mb", 120),
]):
col_frame = ctk.CTkFrame(adv_frame, fg_color="transparent")
col_frame.pack(side="left", padx=(0, 10))
ctk.CTkLabel(col_frame, text=lbl, font=(FONT_FAMILY, 11),
text_color=TEXT_SECONDARY, anchor="w").pack(anchor="w")
ctk.CTkEntry(col_frame, width=w_, height=30, font=(FONT_FAMILY, 12),
corner_radius=8, fg_color=FIELD_BG,
border_color=FIELD_BORDER, border_width=1,
text_color=TEXT_PRIMARY,
textvariable=ctk.StringVar(
value=str(cfg.get(key, DEFAULT_CONFIG[key]))
)).pack(anchor="w")
_adv_entries = list(adv_frame.winfo_children())
_adv_keys = ["buf_kb", "pool_size", "log_max_mb"]
def on_save():
import socket as _sock
host_val = host_var.get().strip()
try:
_sock.inet_aton(host_val)
except OSError:
_show_error("Некорректный IP-адрес.")
merged = validate_config_form(
widgets, DEFAULT_CONFIG, include_autostart=False)
if isinstance(merged, str):
_show_error(merged)
return
try:
port_val = int(port_var.get().strip())
if not (1 <= port_val <= 65535):
raise ValueError
except ValueError:
_show_error("Порт должен быть числом 1-65535")
return
lines = [
l.strip()
for l in dc_textbox.get("1.0", "end").strip().splitlines()
if l.strip()
]
try:
tg_ws_proxy.parse_dc_ip_list(lines)
except ValueError as e:
_show_error(str(e))
return
new_cfg = {
"host": host_val,
"port": port_val,
"dc_ip": lines,
"verbose": verbose_var.get(),
}
for i, key in enumerate(_adv_keys):
col_frame = _adv_entries[i]
entry = col_frame.winfo_children()[1]
try:
val = float(entry.get().strip())
if key in ("buf_kb", "pool_size"):
val = int(val)
new_cfg[key] = val
except ValueError:
new_cfg[key] = DEFAULT_CONFIG[key]
new_cfg = merged
save_config(new_cfg)
_config.update(new_cfg)
log.info("Config saved: %s", new_cfg)
@ -463,21 +360,18 @@ def _edit_config_dialog():
def on_cancel():
root.destroy()
btn_frame = ctk.CTkFrame(frame, fg_color="transparent")
btn_frame.pack(fill="x", pady=(20, 0))
ctk.CTkButton(btn_frame, text="Сохранить", height=38,
font=(FONT_FAMILY, 14, "bold"), corner_radius=10,
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
text_color="#ffffff",
command=on_save).pack(side="left", fill="x", expand=True, padx=(0, 8))
ctk.CTkButton(btn_frame, text="Отмена", height=38,
font=(FONT_FAMILY, 14), corner_radius=10,
fg_color=FIELD_BG, hover_color=FIELD_BORDER,
text_color=TEXT_PRIMARY, border_width=1,
border_color=FIELD_BORDER,
command=on_cancel).pack(side="right", fill="x", expand=True)
install_tray_config_buttons(
ctk, footer, theme, on_save=on_save, on_cancel=on_cancel)
root.mainloop()
try:
root.mainloop()
finally:
import tkinter as tk
try:
if root.winfo_exists():
root.destroy()
except tk.TclError:
pass
def _on_open_logs(icon=None, item=None):
@ -525,128 +419,41 @@ def _show_first_run():
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
tg_url = f"tg://socks?server={host}&port={port}"
if ctk is None:
FIRST_RUN_MARKER.touch()
return
ctk.set_appearance_mode("light")
ctk.set_default_color_theme("blue")
theme = ctk_theme_for_platform()
w, h = FIRST_RUN_SIZE
TG_BLUE = "#3390ec"
TG_BLUE_HOVER = "#2b7cd4"
BG = "#ffffff"
FIELD_BG = "#f0f2f5"
FIELD_BORDER = "#d6d9dc"
TEXT_PRIMARY = "#000000"
TEXT_SECONDARY = "#707579"
FONT_FAMILY = "Sans"
root = ctk.CTk()
root.title("TG WS Proxy")
root.resizable(False, False)
root.attributes("-topmost", True)
icon_img = _load_icon()
if icon_img:
from PIL import ImageTk
_photo = ImageTk.PhotoImage(icon_img.resize((64, 64)))
root.iconphoto(False, _photo)
w, h = 520, 440
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
root.geometry(f"{w}x{h}+{(sw - w) // 2}+{(sh - h) // 2}")
root.configure(fg_color=BG)
frame = ctk.CTkFrame(root, fg_color=BG, corner_radius=0)
frame.pack(fill="both", expand=True, padx=28, pady=24)
title_frame = ctk.CTkFrame(frame, fg_color="transparent")
title_frame.pack(anchor="w", pady=(0, 16), fill="x")
# Blue accent bar
accent_bar = ctk.CTkFrame(
title_frame, fg_color=TG_BLUE, width=4, height=32, corner_radius=2
)
accent_bar.pack(side="left", padx=(0, 12))
ctk.CTkLabel(
title_frame,
text="Прокси запущен и работает в системном трее",
font=(FONT_FAMILY, 17, "bold"),
text_color=TEXT_PRIMARY,
).pack(side="left")
# Info sections
sections = [
("Как подключить Telegram Desktop:", True),
(" Автоматически:", True),
(f" ПКМ по иконке в трее → «Открыть в Telegram»", False),
(f" Или ссылка: {tg_url}", False),
("\n Вручную:", True),
(" Настройки → Продвинутые → Тип подключения → Прокси", False),
(f" SOCKS5 → {host} : {port} (без логина/пароля)", False),
]
for text, bold in sections:
weight = "bold" if bold else "normal"
ctk.CTkLabel(
frame,
text=text,
font=(FONT_FAMILY, 13, weight),
text_color=TEXT_PRIMARY,
anchor="w",
justify="left",
).pack(anchor="w", pady=1)
# Spacer
ctk.CTkFrame(frame, fg_color="transparent", height=16).pack()
# Separator
ctk.CTkFrame(frame, fg_color=FIELD_BORDER, height=1, corner_radius=0).pack(
fill="x", pady=(0, 12)
root = create_ctk_root(
ctk,
title="TG WS Proxy",
width=w,
height=h,
theme=theme,
after_create=_apply_linux_ctk_window_icon,
)
# Checkbox
auto_var = ctk.BooleanVar(value=True)
ctk.CTkCheckBox(
frame,
text="Открыть прокси в Telegram сейчас",
variable=auto_var,
font=(FONT_FAMILY, 13),
text_color=TEXT_PRIMARY,
fg_color=TG_BLUE,
hover_color=TG_BLUE_HOVER,
corner_radius=6,
border_width=2,
border_color=FIELD_BORDER,
).pack(anchor="w", pady=(0, 16))
def on_ok():
def on_done(open_tg: bool):
FIRST_RUN_MARKER.touch()
open_tg = auto_var.get()
root.destroy()
if open_tg:
_on_open_in_telegram()
ctk.CTkButton(
frame,
text="Начать",
width=180,
height=42,
font=(FONT_FAMILY, 15, "bold"),
corner_radius=10,
fg_color=TG_BLUE,
hover_color=TG_BLUE_HOVER,
text_color="#ffffff",
command=on_ok,
).pack(pady=(0, 0))
populate_first_run_window(
ctk, root, theme, host=host, port=port, on_done=on_done)
root.protocol("WM_DELETE_WINDOW", on_ok)
root.mainloop()
try:
root.mainloop()
finally:
import tkinter as tk
try:
if root.winfo_exists():
root.destroy()
except tk.TclError:
pass
def _has_ipv6_enabled() -> bool:
@ -722,7 +529,7 @@ def run_tray():
setup_logging(_config.get("verbose", False),
log_max_mb=_config.get("log_max_mb", DEFAULT_CONFIG["log_max_mb"]))
log.info("TG WS Proxy tray app starting")
log.info("TG WS Proxy версия %s, tray app starting", __version__)
log.info("Config: %s", _config)
log.info("Log file: %s", LOG_FILE)
@ -738,6 +545,8 @@ def run_tray():
start_proxy()
_maybe_notify_update_async()
_show_first_run()
_check_ipv6_warning()

83
macos.py
View File

@ -30,6 +30,8 @@ except ImportError:
import proxy.tg_ws_proxy as tg_ws_proxy
from proxy.app_runtime import ProxyAppRuntime
from proxy import __version__
from utils.default_config import default_tray_config
APP_NAME = "TgWsProxy"
APP_DIR = Path.home() / "Library" / "Application Support" / APP_NAME
@ -39,15 +41,7 @@ FIRST_RUN_MARKER = APP_DIR / ".first_run_done"
IPV6_WARN_MARKER = APP_DIR / ".ipv6_warned"
MENUBAR_ICON_PATH = APP_DIR / "menubar_icon.png"
DEFAULT_CONFIG = {
"port": 1080,
"host": "127.0.0.1",
"dc_ip": ["2:149.154.167.220", "4:149.154.167.220"],
"verbose": False,
"log_max_mb": 5,
"buf_kb": 256,
"pool_size": 4,
}
DEFAULT_CONFIG = default_tray_config()
_app: Optional[object] = None
_config: dict = {}
@ -264,8 +258,9 @@ def restart_proxy():
# Menu callbacks
def _on_open_in_telegram(_=None):
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
url = f"tg://socks?server=127.0.0.1&port={port}"
url = f"tg://socks?server={host}&port={port}"
log.info("Opening %s", url)
try:
result = subprocess.call(['open', url])
@ -333,6 +328,55 @@ def _on_edit_config(_=None):
threading.Thread(target=_edit_config_dialog, daemon=True).start()
def _check_updates_menu_title() -> str:
on = bool(_config.get("check_updates", True))
return (
"✓ Проверять обновления при запуске"
if on
else "Проверять обновления при запуске (выкл)"
)
def _toggle_check_updates(_=None):
global _config
_config["check_updates"] = not bool(_config.get("check_updates", True))
save_config(_config)
if _app is not None:
_app._check_updates_item.title = _check_updates_menu_title()
def _on_open_release_page(_=None):
from utils.update_check import RELEASES_PAGE_URL
webbrowser.open(RELEASES_PAGE_URL)
def _maybe_notify_update_async():
def _work():
time.sleep(1.5)
if _exiting:
return
if not _config.get("check_updates", True):
return
try:
from utils.update_check import RELEASES_PAGE_URL, get_status, run_check
run_check(__version__)
st = get_status()
if not st.get("has_update"):
return
url = (st.get("html_url") or "").strip() or RELEASES_PAGE_URL
ver = st.get("latest") or "?"
if _ask_yes_no(
f"Доступна новая версия: {ver}\n\n"
f"Открыть страницу релиза в браузере?",
"TG WS Proxy — обновление",
):
webbrowser.open(url)
except Exception as exc:
log.debug("Update check failed: %s", exc)
threading.Thread(target=_work, daemon=True, name="update-check").start()
# Settings via native macOS dialogs
def _edit_config_dialog():
cfg = load_config()
@ -523,6 +567,15 @@ class TgWsProxyApp(_TgWsProxyAppBase):
self._logs_item = rumps.MenuItem(
"Открыть логи",
callback=_on_open_logs)
self._release_page_item = rumps.MenuItem(
"Страница релиза на GitHub…",
callback=_on_open_release_page)
self._check_updates_item = rumps.MenuItem(
_check_updates_menu_title(),
callback=_toggle_check_updates)
self._version_item = rumps.MenuItem(
f"Версия {__version__}",
callback=lambda _: None)
super().__init__(
"TG WS Proxy",
@ -535,6 +588,11 @@ class TgWsProxyApp(_TgWsProxyAppBase):
self._restart_item,
self._settings_item,
self._logs_item,
None,
self._release_page_item,
self._check_updates_item,
None,
self._version_item,
])
def update_menu_title(self):
@ -552,7 +610,7 @@ def run_menubar():
setup_logging(_config.get("verbose", False),
log_max_mb=_config.get("log_max_mb", DEFAULT_CONFIG["log_max_mb"]))
log.info("TG WS Proxy menubar app starting")
log.info("TG WS Proxy версия %s, menubar app starting", __version__)
log.info("Config: %s", _config)
log.info("Log file: %s", LOG_FILE)
@ -567,6 +625,9 @@ def run_menubar():
return
start_proxy()
_maybe_notify_update_async()
_show_first_run()
_check_ipv6_warning()

246
proxy/tg_ws_proxy.py
View File

@ -4,6 +4,7 @@ import argparse
import asyncio
import base64
import logging
from collections import deque
import logging.handlers
import os
import socket as _socket
@ -145,7 +146,14 @@ _st_Q = struct.Struct('>Q')
_st_I_net = struct.Struct('!I')
_st_Ih = struct.Struct('<Ih')
_st_I_le = struct.Struct('<I')
_VALID_PROTOS = frozenset((0xEFEFEFEF, 0xEEEEEEEE, 0xDDDDDDDD))
_PROTO_ABRIDGED = 0xEFEFEFEF
_PROTO_INTERMEDIATE = 0xEEEEEEEE
_PROTO_PADDED_INTERMEDIATE = 0xDDDDDDDD
_VALID_PROTOS = frozenset((
_PROTO_ABRIDGED,
_PROTO_INTERMEDIATE,
_PROTO_PADDED_INTERMEDIATE,
))
class RawWebSocket:
@ -382,11 +390,7 @@ def _is_http_transport(data: bytes) -> bool:
data[:5] == b'HEAD ' or data[:8] == b'OPTIONS ')
def _dc_from_init(data: bytes) -> Tuple[Optional[int], bool]:
"""
Extract DC ID from the 64-byte MTProto obfuscation init packet.
Returns (dc_id, is_media).
"""
def _dc_from_init(data: bytes, *, return_proto: bool = False):
try:
key = bytes(data[8:40])
iv = bytes(data[40:56])
@ -400,11 +404,14 @@ def _dc_from_init(data: bytes) -> Tuple[Optional[int], bool]:
if proto in _VALID_PROTOS:
dc = abs(dc_raw)
if 1 <= dc <= 5 or dc == 203:
return dc, (dc_raw < 0)
return (
(dc, (dc_raw < 0), proto)
if return_proto else (dc, (dc_raw < 0))
)
return (None, False, proto) if return_proto else (None, False)
except Exception as exc:
log.debug("DC extraction failed: %s", exc)
return None, False
return (None, False, None) if return_proto else (None, False)
def _patch_init_dc(data: bytes, dc: int) -> bytes:
"""
@ -435,54 +442,103 @@ def _patch_init_dc(data: bytes, dc: int) -> bytes:
class _MsgSplitter:
"""
Splits client TCP data into individual MTProto abridged-protocol
messages so each can be sent as a separate WebSocket frame.
Splits client TCP data into individual MTProto transport packets so
each can be sent as a separate WebSocket frame.
The Telegram WS relay processes one MTProto message per WS frame.
Mobile clients batches multiple messages in a single TCP write (e.g.
msgs_ack + req_DH_params). If sent as one WS frame, the relay
only processes the first message — DH handshake never completes.
Some mobile clients coalesce multiple MTProto packets into one TCP
write, and TCP reads may also cut a packet in half. Keep a rolling
buffer so incomplete packets are not forwarded as standalone frames.
"""
def __init__(self, init_data: bytes):
__slots__ = ('_dec', '_proto', '_cipher_buf', '_plain_buf', '_disabled')
def __init__(self, init_data: bytes, proto: Optional[int] = None):
if proto is None:
_, _, proto = _dc_from_init(init_data, return_proto=True)
key_raw = bytes(init_data[8:40])
iv = bytes(init_data[40:56])
self._dec = create_aes_ctr_transform(key_raw, iv)
self._dec.update(b'\x00' * 64) # skip init packet
self._proto = proto
self._cipher_buf = bytearray()
self._plain_buf = bytearray()
self._disabled = False
def split(self, chunk: bytes) -> List[bytes]:
"""Decrypt to find message boundaries, return split ciphertext."""
plain = self._dec.update(chunk)
boundaries = []
pos = 0
plain_len = len(plain)
while pos < plain_len:
first = plain[pos]
if first == 0x7f:
if pos + 4 > plain_len:
break
msg_len = (
_st_I_le.unpack_from(plain, pos + 1)[0] & 0xFFFFFF
) * 4
pos += 4
else:
msg_len = first * 4
pos += 1
if msg_len == 0 or pos + msg_len > plain_len:
break
pos += msg_len
boundaries.append(pos)
if len(boundaries) <= 1:
"""Decrypt to find packet boundaries, return complete ciphertext packets."""
if not chunk:
return []
if self._disabled:
return [chunk]
self._cipher_buf.extend(chunk)
self._plain_buf.extend(self._dec.update(chunk))
parts = []
prev = 0
for b in boundaries:
parts.append(chunk[prev:b])
prev = b
if prev < len(chunk):
parts.append(chunk[prev:])
while self._cipher_buf:
packet_len = self._next_packet_len()
if packet_len is None:
break
if packet_len <= 0:
parts.append(bytes(self._cipher_buf))
self._cipher_buf.clear()
self._plain_buf.clear()
self._disabled = True
break
parts.append(bytes(self._cipher_buf[:packet_len]))
del self._cipher_buf[:packet_len]
del self._plain_buf[:packet_len]
return parts
def flush(self) -> List[bytes]:
if not self._cipher_buf:
return []
tail = bytes(self._cipher_buf)
self._cipher_buf.clear()
self._plain_buf.clear()
return [tail]
def _next_packet_len(self) -> Optional[int]:
if not self._plain_buf:
return None
if self._proto == _PROTO_ABRIDGED:
return self._next_abridged_len()
if self._proto in (_PROTO_INTERMEDIATE, _PROTO_PADDED_INTERMEDIATE):
return self._next_intermediate_len()
return 0
def _next_abridged_len(self) -> Optional[int]:
first = self._plain_buf[0]
if first in (0x7F, 0xFF):
if len(self._plain_buf) < 4:
return None
payload_len = int.from_bytes(self._plain_buf[1:4], 'little') * 4
header_len = 4
else:
payload_len = (first & 0x7F) * 4
header_len = 1
if payload_len <= 0:
return 0
packet_len = header_len + payload_len
if len(self._plain_buf) < packet_len:
return None
return packet_len
def _next_intermediate_len(self) -> Optional[int]:
if len(self._plain_buf) < 4:
return None
payload_len = _st_I_le.unpack_from(self._plain_buf, 0)[0] & 0x7FFFFFFF
if payload_len <= 0:
return 0
packet_len = 4 + payload_len
if len(self._plain_buf) < packet_len:
return None
return packet_len
def _ws_domains(dc: int, is_media) -> List[str]:
dc = _DC_OVERRIDES.get(dc, dc)
@ -505,12 +561,15 @@ class Stats:
self.pool_misses = 0
def summary(self) -> str:
pool_total = self.pool_hits + self.pool_misses
pool_s = (
f"{self.pool_hits}/{pool_total}" if pool_total else "n/a")
return (f"total={self.connections_total} ws={self.connections_ws} "
f"tcp_fb={self.connections_tcp_fallback} "
f"http_skip={self.connections_http_rejected} "
f"pass={self.connections_passthrough} "
f"err={self.ws_errors} "
f"pool={self.pool_hits}/{self.pool_hits+self.pool_misses} "
f"pool={pool_s} "
f"up={_human_bytes(self.bytes_up)} "
f"down={_human_bytes(self.bytes_down)}")
@ -535,7 +594,7 @@ def get_stats_snapshot() -> Dict[str, int]:
class _WsPool:
def __init__(self):
self._idle: Dict[Tuple[int, bool], list] = {}
self._idle: Dict[Tuple[int, bool], deque] = {}
self._refilling: Set[Tuple[int, bool]] = set()
async def get(self, dc: int, is_media: bool,
@ -544,9 +603,12 @@ class _WsPool:
key = (dc, is_media)
now = time.monotonic()
bucket = self._idle.get(key, [])
bucket = self._idle.get(key)
if bucket is None:
bucket = deque()
self._idle[key] = bucket
while bucket:
ws, created = bucket.pop(0)
ws, created = bucket.popleft()
age = now - created
if age > _WS_POOL_MAX_AGE or ws._closed:
asyncio.create_task(self._quiet_close(ws))
@ -570,7 +632,7 @@ class _WsPool:
async def _refill(self, key, target_ip, domains):
dc, is_media = key
try:
bucket = self._idle.setdefault(key, [])
bucket = self._idle.setdefault(key, deque())
needed = _WS_POOL_SIZE - len(bucket)
if needed <= 0:
return
@ -646,6 +708,10 @@ async def _bridge_ws(reader, writer, ws: RawWebSocket, label,
while True:
chunk = await reader.read(65536)
if not chunk:
if splitter:
tail = splitter.flush()
if tail:
await ws.send(tail[0])
break
n = len(chunk)
_stats.bytes_up += n
@ -653,6 +719,8 @@ async def _bridge_ws(reader, writer, ws: RawWebSocket, label,
up_packets += 1
if splitter:
parts = splitter.split(chunk)
if not parts:
continue
if len(parts) > 1:
await ws.send_batch(parts)
else:
@ -913,14 +981,14 @@ async def _handle_client(reader, writer):
return
# -- Extract DC ID --
dc, is_media = _dc_from_init(init)
init_patched = False
dc, is_media, proto = _dc_from_init(init)
init_patched = False
# Android (may be ios too) with useSecret=0 has random dc_id bytes — patch it
if dc is None and dst in _IP_TO_DC:
dc, is_media = _IP_TO_DC.get(dst)
if dc in _dc_opt:
init = _patch_init_dc(init, dc if is_media else -dc)
init = _patch_init_dc(init, -dc if is_media else dc)
init_patched = True
if dc is None or dc not in _dc_opt:
@ -1022,9 +1090,12 @@ async def _handle_client(reader, writer):
_stats.connections_ws += 1
splitter = None
if init_patched:
# Turning splitter on for mobile clients or media-connections, so as the big files don't get fragmented by the TCP socket.
if proto is not None and (init_patched or is_media or proto != _PROTO_INTERMEDIATE):
try:
splitter = _MsgSplitter(init)
splitter = _MsgSplitter(init, proto)
log.debug("[%s] MsgSplitter activated for proto 0x%08X", label, proto)
except Exception:
pass
@ -1044,6 +1115,11 @@ async def _handle_client(reader, writer):
log.debug("[%s] cancelled", label)
except ConnectionResetError:
log.debug("[%s] connection reset", label)
except OSError as exc:
if getattr(exc, 'winerror', None) == 1236:
log.debug("[%s] connection aborted by local system", label)
else:
log.error("[%s] unexpected os error: %s", label, exc)
except Exception as exc:
log.error("[%s] unexpected: %s", label, exc)
finally:
@ -1087,34 +1163,50 @@ async def _run(port: int, dc_opt: Dict[int, Optional[str]],
log.info("=" * 60)
async def log_stats():
while True:
await asyncio.sleep(60)
bl = ', '.join(
f'DC{d}{"m" if m else ""}'
for d, m in sorted(_ws_blacklist)) or 'none'
log.info("stats: %s | ws_bl: %s", _stats.summary(), bl)
try:
while True:
await asyncio.sleep(60)
bl = ', '.join(
f'DC{d}{"m" if m else ""}'
for d, m in sorted(_ws_blacklist)) or 'none'
log.info("stats: %s | ws_bl: %s", _stats.summary(), bl)
except asyncio.CancelledError:
raise
asyncio.create_task(log_stats())
log_stats_task = asyncio.create_task(log_stats())
await _ws_pool.warmup(dc_opt)
if stop_event:
async def wait_stop():
await stop_event.wait()
server.close()
me = asyncio.current_task()
for task in list(asyncio.all_tasks()):
if task is not me:
task.cancel()
try:
await server.wait_closed()
except asyncio.CancelledError:
pass
asyncio.create_task(wait_stop())
async with server:
try:
async with server:
if stop_event:
serve_task = asyncio.create_task(server.serve_forever())
stop_task = asyncio.create_task(stop_event.wait())
done, _pending = await asyncio.wait(
(serve_task, stop_task),
return_when=asyncio.FIRST_COMPLETED,
)
if stop_task in done:
server.close()
await server.wait_closed()
if not serve_task.done():
serve_task.cancel()
try:
await serve_task
except asyncio.CancelledError:
pass
else:
stop_task.cancel()
try:
await stop_task
except asyncio.CancelledError:
pass
else:
await server.serve_forever()
finally:
log_stats_task.cancel()
try:
await server.serve_forever()
await log_stats_task
except asyncio.CancelledError:
pass
_server_instance = None

2
pyproject.toml
View File

@ -62,7 +62,7 @@ Source = "https://github.com/Flowseal/tg-ws-proxy"
Issues = "https://github.com/Flowseal/tg-ws-proxy/issues"
[tool.hatch.build.targets.wheel]
packages = ["proxy"]
packages = ["proxy", "ui", "utils"]
[tool.hatch.build.force-include]
"windows.py" = "windows.py"

4
ui/__init__.py Normal file
View File

@ -0,0 +1,4 @@
"""
Интерфейс tray (CustomTkinter): тема, диалоги настроек, подсказки.
Ядро прокси — пакет `proxy`.
"""

112
ui/ctk_theme.py Normal file
View File

@ -0,0 +1,112 @@
"""
Общая светлая тема и фабрика окон CustomTkinter для tray-приложений (Windows / Linux).
Цвета и отступы задаются в одном месте — правки темы не дублируются по платформам.
"""
from __future__ import annotations
import sys
import tkinter
from dataclasses import dataclass
from typing import Any, Callable, Optional, Tuple
_tk_variable_del_guard_installed = False
def _install_tkinter_variable_del_guard() -> None:
"""
Убирает «Exception ignored» при выходе процесса: Tcl уже разрушен, а GC ещё
вызывает Variable.__del__ (StringVar и т.д.) — напр. окно CTk в фоновом потоке.
"""
global _tk_variable_del_guard_installed
if _tk_variable_del_guard_installed:
return
_orig = tkinter.Variable.__del__
def _safe_variable_del(self: Any, _orig: Any = _orig) -> None:
try:
_orig(self)
except (RuntimeError, tkinter.TclError):
pass
tkinter.Variable.__del__ = _safe_variable_del # type: ignore[assignment]
_tk_variable_del_guard_installed = True
# Размеры и отступы (единые для диалогов настроек и первого запуска)
CONFIG_DIALOG_SIZE: Tuple[int, int] = (460, 560)
CONFIG_DIALOG_FRAME_PAD: Tuple[int, int] = (20, 14)
FIRST_RUN_SIZE: Tuple[int, int] = (520, 440)
FIRST_RUN_FRAME_PAD: Tuple[int, int] = (28, 24)
@dataclass(frozen=True)
class CtkTheme:
"""Палитра Telegram-style и семейства шрифтов для UI и моноширинного текста."""
tg_blue: str = "#3390ec"
tg_blue_hover: str = "#2b7cd4"
bg: str = "#ffffff"
field_bg: str = "#f0f2f5"
field_border: str = "#d6d9dc"
text_primary: str = "#000000"
text_secondary: str = "#707579"
ui_font_family: str = "Sans"
mono_font_family: str = "Monospace"
def ctk_theme_for_platform() -> CtkTheme:
if sys.platform == "win32":
return CtkTheme(ui_font_family="Segoe UI", mono_font_family="Consolas")
return CtkTheme()
def apply_ctk_appearance(ctk: Any) -> None:
ctk.set_appearance_mode("light")
ctk.set_default_color_theme("blue")
def center_ctk_geometry(root: Any, width: int, height: int) -> None:
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
root.geometry(f"{width}x{height}+{(sw - width) // 2}+{(sh - height) // 2}")
def create_ctk_root(
ctk: Any,
*,
title: str,
width: int,
height: int,
theme: CtkTheme,
topmost: bool = True,
after_create: Optional[Callable[[Any], None]] = None,
) -> Any:
"""
Создаёт CTk: глобальная тема, заголовок, без ресайза, по центру экрана, фон из палитры.
after_create — опционально: установка иконки окна (различается по ОС).
"""
_install_tkinter_variable_del_guard()
apply_ctk_appearance(ctk)
root = ctk.CTk()
root.title(title)
root.resizable(False, False)
if topmost:
root.attributes("-topmost", True)
center_ctk_geometry(root, width, height)
root.configure(fg_color=theme.bg)
if after_create:
after_create(root)
return root
def main_content_frame(
ctk: Any,
root: Any,
theme: CtkTheme,
*,
padx: int,
pady: int,
) -> Any:
frame = ctk.CTkFrame(root, fg_color=theme.bg, corner_radius=0)
frame.pack(fill="both", expand=True, padx=padx, pady=pady)
return frame

114
ui/ctk_tooltip.py Normal file
View File

@ -0,0 +1,114 @@
"""
Всплывающие подсказки для CustomTkinter / tk: задержка, Toplevel без рамки, wrap.
"""
from __future__ import annotations
import tkinter as tk
from typing import Any, List, Optional
class CtkTooltip:
"""Показ текста при наведении на виджет."""
def __init__(
self,
widget: Any,
text: str,
*,
delay_ms: int = 450,
wraplength: int = 320,
) -> None:
self.widget = widget
self.text = text
self.delay_ms = delay_ms
self.wraplength = wraplength
self._after_id: Optional[str] = None
self._tip: Optional[tk.Toplevel] = None
widget.bind("<Enter>", self._schedule, add="+")
widget.bind("<Leave>", self._hide, add="+")
widget.bind("<Button>", self._hide, add="+")
widget.bind("<Destroy>", self._on_destroy, add="+")
def _schedule(self, _event: Any = None) -> None:
self._cancel_after()
self._after_id = self.widget.after(self.delay_ms, self._show)
def _cancel_after(self) -> None:
if self._after_id is not None:
try:
self.widget.after_cancel(self._after_id)
except Exception:
pass
self._after_id = None
def _show(self) -> None:
self._after_id = None
if self._tip is not None:
return
try:
if not self.widget.winfo_exists():
return
except Exception:
return
tw = tk.Toplevel(self.widget.winfo_toplevel())
tw.wm_overrideredirect(True)
try:
tw.wm_attributes("-topmost", True)
except Exception:
pass
tw.configure(bg="#2b2b2b")
lbl = tk.Label(
tw,
text=self.text,
justify="left",
wraplength=self.wraplength,
background="#2b2b2b",
foreground="#f0f0f0",
relief="flat",
borderwidth=0,
padx=10,
pady=8,
font=("Segoe UI", 10) if _is_windows() else None,
)
lbl.pack()
x = self.widget.winfo_rootx() + 12
y = self.widget.winfo_rooty() + self.widget.winfo_height() + 4
tw.wm_geometry(f"+{x}+{y}")
self._tip = tw
def _hide(self, _event: Any = None) -> None:
self._cancel_after()
if self._tip is not None:
try:
self._tip.destroy()
except Exception:
pass
self._tip = None
def _on_destroy(self, _event: Any = None) -> None:
self._hide()
def _is_windows() -> bool:
import sys
return sys.platform == "win32"
def attach_ctk_tooltip(
widget: Any,
text: str,
*,
delay_ms: int = 450,
wraplength: int = 320,
) -> None:
"""Повесить подсказку на виджет (CTk или tk)."""
CtkTooltip(widget, text, delay_ms=delay_ms, wraplength=wraplength)
def attach_tooltip_to_widgets(widgets: List[Any], text: str, **kwargs: Any) -> None:
"""Одна и та же подсказка на несколько виджетов (подпись + поле)."""
for w in widgets:
attach_ctk_tooltip(w, text, **kwargs)

579
ui/ctk_tray_ui.py Normal file
View File

@ -0,0 +1,579 @@
"""
Общая разметка CustomTkinter для tray (Windows / Linux): настройки и первый запуск.
Логика сохранения и колбэки остаются в платформенных модулях.
"""
from __future__ import annotations
import webbrowser
from dataclasses import dataclass
from typing import Any, Callable, Dict, List, Optional, Tuple, Union
import proxy.tg_ws_proxy as tg_ws_proxy
from proxy import __version__
from utils.update_check import RELEASES_PAGE_URL, get_status
from ui.ctk_theme import (
FIRST_RUN_FRAME_PAD,
CtkTheme,
main_content_frame,
)
from ui.ctk_tooltip import attach_ctk_tooltip, attach_tooltip_to_widgets
# Подсказки для формы настроек (новые пользователи)
_TIP_HOST = (
"Адрес, на котором прокси принимает SOCKS5-подключения.\n"
"Обычно 127.0.0.1 — локальная сеть, 0.0.0.0 - все интерфейсы"
)
_TIP_PORT = (
"Порт SOCKS5. В Telegram Desktop в настройках прокси должен быть "
"указан тот же порт"
)
_TIP_DC = (
"Соответствие номера датацентра Telegram (DC) и IP-адреса сервера.\n"
"Каждая строка: «номер:IP», например 2:149.154.167.220. "
"Прокси по этим правилам направляет трафик к нужным серверам Telegram"
)
_TIP_VERBOSE = (
"Если включено, в файл логов пишется больше подробностей — "
"необходимо при поиске неполадок"
)
_TIP_BUF_KB = (
"Размер буфера приёма/передачи в килобайтах.\n"
"Больше значение — больше выделение памяти на сокет"
)
_TIP_POOL = (
"Сколько параллельных WebSocket-сессий к одному датацентру можно держать.\n"
"Увеличение может помочь при высокой нагрузке"
)
_TIP_LOG_MB = (
"Максимальный размер файла лога; при достижении лимита файл перезаписывается"
)
_TIP_AUTOSTART = (
"Запускать TG WS Proxy при входе в Windows. "
"Если вы переместите программу в другую папку, автозапуск сбросится"
)
_TIP_CHECK_UPDATES = (
"При запуске проверять наличие обновлений"
)
_TIP_SAVE = "Сохранить настройки"
_TIP_CANCEL = "Закрыть окно без сохранения изменений"
# Внутренняя ширина полей относительно ширины окна настроек (см. CONFIG_DIALOG_SIZE)
_CONFIG_FORM_INNER_WIDTH = 396
def tray_settings_scroll_and_footer(
ctk: Any,
content_parent: Any,
theme: CtkTheme,
) -> Tuple[Any, Any]:
"""
Нижняя панель под кнопки и прокручиваемая область для формы (форма не обрезает кнопки).
Возвращает (scroll_frame, footer_frame).
"""
footer = ctk.CTkFrame(content_parent, fg_color=theme.bg)
footer.pack(side="bottom", fill="x")
scroll = ctk.CTkScrollableFrame(
content_parent,
fg_color=theme.bg,
corner_radius=0,
scrollbar_button_color=theme.field_border,
scrollbar_button_hover_color=theme.text_secondary,
)
scroll.pack(fill="both", expand=True)
return scroll, footer
def _config_section(
ctk: Any,
parent: Any,
theme: CtkTheme,
title: str,
*,
bottom_spacer: int = 6,
) -> Any:
"""Заголовок секции и карточка с рамкой для группировки полей."""
wrap = ctk.CTkFrame(parent, fg_color="transparent")
wrap.pack(fill="x", pady=(0, bottom_spacer))
ctk.CTkLabel(
wrap,
text=title,
font=(theme.ui_font_family, 12, "bold"),
text_color=theme.text_primary,
anchor="w",
).pack(anchor="w", pady=(0, 2))
card = ctk.CTkFrame(
wrap,
fg_color=theme.field_bg,
corner_radius=10,
border_width=1,
border_color=theme.field_border,
)
card.pack(fill="x")
inner = ctk.CTkFrame(card, fg_color="transparent")
inner.pack(fill="x", padx=10, pady=8)
return inner
@dataclass
class TrayConfigFormWidgets:
host_var: Any
port_var: Any
dc_textbox: Any
verbose_var: Any
adv_entries: List[Any]
adv_keys: Tuple[str, ...]
autostart_var: Optional[Any]
check_updates_var: Optional[Any]
def install_tray_config_form(
ctk: Any,
frame: Any,
theme: CtkTheme,
cfg: dict,
default_config: dict,
*,
show_autostart: bool = False,
autostart_value: bool = False,
) -> TrayConfigFormWidgets:
"""Поля настроек прокси внутри уже созданного `frame`."""
header = ctk.CTkFrame(frame, fg_color="transparent")
header.pack(fill="x", pady=(0, 2))
ctk.CTkLabel(
header,
text="Настройки прокси",
font=(theme.ui_font_family, 17, "bold"),
text_color=theme.text_primary,
anchor="w",
).pack(side="left")
ctk.CTkLabel(
header,
text=f"v{__version__}",
font=(theme.ui_font_family, 12),
text_color=theme.text_secondary,
anchor="e",
).pack(side="right")
inner_w = _CONFIG_FORM_INNER_WIDTH
conn = _config_section(ctk, frame, theme, "Подключение SOCKS5")
host_row = ctk.CTkFrame(conn, fg_color="transparent")
host_row.pack(fill="x")
host_col = ctk.CTkFrame(host_row, fg_color="transparent")
host_col.pack(side="left", fill="x", expand=True, padx=(0, 10))
host_lbl = ctk.CTkLabel(
host_col,
text="IP-адрес",
font=(theme.ui_font_family, 12),
text_color=theme.text_secondary,
anchor="w",
)
host_lbl.pack(anchor="w", pady=(0, 2))
host_var = ctk.StringVar(value=cfg.get("host", default_config["host"]))
host_entry = ctk.CTkEntry(
host_col,
textvariable=host_var,
width=160,
height=36,
font=(theme.ui_font_family, 13),
corner_radius=10,
fg_color=theme.bg,
border_color=theme.field_border,
border_width=1,
text_color=theme.text_primary,
)
host_entry.pack(fill="x", pady=(0, 0))
attach_tooltip_to_widgets([host_lbl, host_entry, host_col], _TIP_HOST)
port_col = ctk.CTkFrame(host_row, fg_color="transparent")
port_col.pack(side="left")
port_lbl = ctk.CTkLabel(
port_col,
text="Порт",
font=(theme.ui_font_family, 12),
text_color=theme.text_secondary,
anchor="w",
)
port_lbl.pack(anchor="w", pady=(0, 2))
port_var = ctk.StringVar(value=str(cfg.get("port", default_config["port"])))
port_entry = ctk.CTkEntry(
port_col,
textvariable=port_var,
width=100,
height=36,
font=(theme.ui_font_family, 13),
corner_radius=10,
fg_color=theme.bg,
border_color=theme.field_border,
border_width=1,
text_color=theme.text_primary,
)
port_entry.pack(anchor="w")
attach_tooltip_to_widgets([port_lbl, port_entry, port_col], _TIP_PORT)
dc_inner = _config_section(ctk, frame, theme, "Датацентры Telegram (DC → IP)")
dc_lbl = ctk.CTkLabel(
dc_inner,
text="По одному правилу на строку, формат: номер:IP",
font=(theme.ui_font_family, 11),
text_color=theme.text_secondary,
anchor="w",
)
dc_lbl.pack(anchor="w", pady=(0, 4))
dc_textbox = ctk.CTkTextbox(
dc_inner,
width=inner_w,
height=88,
font=(theme.mono_font_family, 12),
corner_radius=10,
fg_color=theme.bg,
border_color=theme.field_border,
border_width=1,
text_color=theme.text_primary,
)
dc_textbox.pack(fill="x")
dc_textbox.insert("1.0", "\n".join(cfg.get("dc_ip", default_config["dc_ip"])))
attach_tooltip_to_widgets([dc_lbl, dc_textbox], _TIP_DC)
log_inner = _config_section(ctk, frame, theme, "Логи и производительность")
verbose_var = ctk.BooleanVar(value=cfg.get("verbose", False))
verbose_cb = ctk.CTkCheckBox(
log_inner,
text="Подробное логирование (verbose)",
variable=verbose_var,
font=(theme.ui_font_family, 13),
text_color=theme.text_primary,
fg_color=theme.tg_blue,
hover_color=theme.tg_blue_hover,
corner_radius=6,
border_width=2,
border_color=theme.field_border,
)
verbose_cb.pack(anchor="w", pady=(0, 6))
attach_ctk_tooltip(verbose_cb, _TIP_VERBOSE)
adv_frame = ctk.CTkFrame(log_inner, fg_color="transparent")
adv_frame.pack(fill="x")
adv_rows = [
("Буфер, КБ (по умолчанию 256)", "buf_kb", _TIP_BUF_KB),
("Пул WebSocket-сессий (по умолчанию 4)", "pool_size", _TIP_POOL),
("Макс. размер лога, МБ (по умолчанию 5)", "log_max_mb", _TIP_LOG_MB),
]
for lbl, key, tip in adv_rows:
col_frame = ctk.CTkFrame(adv_frame, fg_color="transparent")
col_frame.pack(fill="x", pady=(0, 0 if key == "log_max_mb" else 5))
adv_l = ctk.CTkLabel(
col_frame,
text=lbl,
font=(theme.ui_font_family, 11),
text_color=theme.text_secondary,
anchor="w",
)
adv_l.pack(anchor="w", pady=(0, 2))
adv_e = ctk.CTkEntry(
col_frame,
width=inner_w,
height=32,
font=(theme.ui_font_family, 13),
corner_radius=8,
fg_color=theme.bg,
border_color=theme.field_border,
border_width=1,
text_color=theme.text_primary,
textvariable=ctk.StringVar(
value=str(cfg.get(key, default_config[key]))
),
)
adv_e.pack(fill="x")
attach_tooltip_to_widgets([adv_l, adv_e, col_frame], tip)
adv_entries = list(adv_frame.winfo_children())
adv_keys = ("buf_kb", "pool_size", "log_max_mb")
upd_inner = _config_section(ctk, frame, theme, "Обновления")
st = get_status()
check_updates_var = ctk.BooleanVar(
value=bool(
cfg.get("check_updates", default_config.get("check_updates", True))
)
)
upd_cb = ctk.CTkCheckBox(
upd_inner,
text="Проверять обновления при запуске",
variable=check_updates_var,
font=(theme.ui_font_family, 13),
text_color=theme.text_primary,
fg_color=theme.tg_blue,
hover_color=theme.tg_blue_hover,
corner_radius=6,
border_width=2,
border_color=theme.field_border,
)
upd_cb.pack(anchor="w", pady=(0, 6))
attach_ctk_tooltip(upd_cb, _TIP_CHECK_UPDATES)
if st.get("error"):
upd_status = "Не удалось связаться с GitHub. Проверьте сеть."
elif not st.get("checked"):
upd_status = "Статус появится после фоновой проверки при запуске."
elif st.get("has_update") and st.get("latest"):
upd_status = (
f"На GitHub доступна версия {st['latest']} "
f"(у вас {__version__})."
)
elif st.get("ahead_of_release") and st.get("latest"):
upd_status = (
f"У вас {__version__} — новее последнего релиза на GitHub "
f"({st['latest']})."
)
else:
upd_status = "Установлена последняя известная версия с GitHub."
ctk.CTkLabel(
upd_inner,
text=upd_status,
font=(theme.ui_font_family, 11),
text_color=theme.text_secondary,
anchor="w",
justify="left",
wraplength=inner_w,
).pack(anchor="w", pady=(0, 8))
rel_url = (st.get("html_url") or "").strip() or RELEASES_PAGE_URL
open_rel_btn = ctk.CTkButton(
upd_inner,
text="Открыть страницу релиза",
height=32,
font=(theme.ui_font_family, 13),
corner_radius=8,
fg_color=theme.field_bg,
hover_color=theme.field_border,
text_color=theme.text_primary,
border_width=1,
border_color=theme.field_border,
command=lambda u=rel_url: webbrowser.open(u),
)
open_rel_btn.pack(anchor="w")
autostart_var = None
if show_autostart:
sys_inner = _config_section(
ctk, frame, theme, "Запуск Windows", bottom_spacer=4
)
autostart_var = ctk.BooleanVar(value=autostart_value)
as_cb = ctk.CTkCheckBox(
sys_inner,
text="Автозапуск при включении компьютера",
variable=autostart_var,
font=(theme.ui_font_family, 13),
text_color=theme.text_primary,
fg_color=theme.tg_blue,
hover_color=theme.tg_blue_hover,
corner_radius=6,
border_width=2,
border_color=theme.field_border,
)
as_cb.pack(anchor="w", pady=(0, 4))
as_hint = ctk.CTkLabel(
sys_inner,
text="Если переместить программу в другую папку, запись автозапуска может сброситься.",
font=(theme.ui_font_family, 11),
text_color=theme.text_secondary,
anchor="w",
justify="left",
wraplength=inner_w,
)
as_hint.pack(anchor="w")
attach_tooltip_to_widgets([as_cb, as_hint], _TIP_AUTOSTART)
return TrayConfigFormWidgets(
host_var=host_var,
port_var=port_var,
dc_textbox=dc_textbox,
verbose_var=verbose_var,
adv_entries=adv_entries,
adv_keys=adv_keys,
autostart_var=autostart_var,
check_updates_var=check_updates_var,
)
def merge_adv_from_form(
widgets: TrayConfigFormWidgets,
base: Dict[str, Any],
default_config: dict,
) -> None:
"""Дополняет base значениями buf_kb / pool_size / log_max_mb (in-place)."""
for i, key in enumerate(widgets.adv_keys):
col_frame = widgets.adv_entries[i]
entry = col_frame.winfo_children()[1]
try:
val = float(entry.get().strip())
if key in ("buf_kb", "pool_size"):
val = int(val)
base[key] = val
except ValueError:
base[key] = default_config[key]
def validate_config_form(
widgets: TrayConfigFormWidgets,
default_config: dict,
*,
include_autostart: bool,
) -> Union[dict, str]:
"""
Возвращает словарь полей конфига или строку ошибки для показа пользователю.
"""
import socket as _sock
host_val = widgets.host_var.get().strip()
try:
_sock.inet_aton(host_val)
except OSError:
return "Некорректный IP-адрес."
try:
port_val = int(widgets.port_var.get().strip())
if not (1 <= port_val <= 65535):
raise ValueError
except ValueError:
return "Порт должен быть числом 1-65535"
lines = [
l.strip()
for l in widgets.dc_textbox.get("1.0", "end").strip().splitlines()
if l.strip()
]
try:
tg_ws_proxy.parse_dc_ip_list(lines)
except ValueError as e:
return str(e)
new_cfg: Dict[str, Any] = {
"host": host_val,
"port": port_val,
"dc_ip": lines,
"verbose": widgets.verbose_var.get(),
}
if include_autostart:
new_cfg["autostart"] = (
widgets.autostart_var.get()
if widgets.autostart_var is not None
else False
)
merge_adv_from_form(widgets, new_cfg, default_config)
if widgets.check_updates_var is not None:
new_cfg["check_updates"] = bool(widgets.check_updates_var.get())
return new_cfg
def install_tray_config_buttons(
ctk: Any,
frame: Any,
theme: CtkTheme,
*,
on_save: Callable[[], None],
on_cancel: Callable[[], None],
) -> None:
ctk.CTkFrame(
frame,
fg_color=theme.field_border,
height=1,
corner_radius=0,
).pack(fill="x", pady=(4, 10))
btn_frame = ctk.CTkFrame(frame, fg_color="transparent")
btn_frame.pack(fill="x", pady=(0, 0))
save_btn = ctk.CTkButton(
btn_frame, text="Сохранить", height=38,
font=(theme.ui_font_family, 14, "bold"), corner_radius=10,
fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
text_color="#ffffff",
command=on_save)
save_btn.pack(side="left", fill="x", expand=True, padx=(0, 8))
attach_ctk_tooltip(save_btn, _TIP_SAVE)
cancel_btn = ctk.CTkButton(
btn_frame, text="Отмена", height=38,
font=(theme.ui_font_family, 14), corner_radius=10,
fg_color=theme.field_bg, hover_color=theme.field_border,
text_color=theme.text_primary, border_width=1,
border_color=theme.field_border,
command=on_cancel)
cancel_btn.pack(side="right", fill="x", expand=True)
attach_ctk_tooltip(cancel_btn, _TIP_CANCEL)
def populate_first_run_window(
ctk: Any,
root: Any,
theme: CtkTheme,
*,
host: str,
port: int,
on_done: Callable[[bool], None],
) -> None:
"""
Содержимое окна первого запуска. on_done(open_in_telegram) — по «Начать» и по закрытию окна.
"""
tg_url = f"tg://socks?server={host}&port={port}"
fpx, fpy = FIRST_RUN_FRAME_PAD
frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy)
title_frame = ctk.CTkFrame(frame, fg_color="transparent")
title_frame.pack(anchor="w", pady=(0, 16), fill="x")
accent_bar = ctk.CTkFrame(title_frame, fg_color=theme.tg_blue,
width=4, height=32, corner_radius=2)
accent_bar.pack(side="left", padx=(0, 12))
ctk.CTkLabel(title_frame, text="Прокси запущен и работает в системном трее",
font=(theme.ui_font_family, 17, "bold"),
text_color=theme.text_primary).pack(side="left")
sections = [
("Как подключить Telegram Desktop:", True),
(" Автоматически:", True),
(" ПКМ по иконке в трее → «Открыть в Telegram»", False),
(f" Или ссылка: {tg_url}", False),
("\n Вручную:", True),
(" Настройки → Продвинутые → Тип подключения → Прокси", False),
(f" SOCKS5 → {host} : {port} (без логина/пароля)", False),
]
for text, bold in sections:
weight = "bold" if bold else "normal"
ctk.CTkLabel(frame, text=text,
font=(theme.ui_font_family, 13, weight),
text_color=theme.text_primary,
anchor="w", justify="left").pack(anchor="w", pady=1)
ctk.CTkFrame(frame, fg_color="transparent", height=16).pack()
ctk.CTkFrame(frame, fg_color=theme.field_border, height=1,
corner_radius=0).pack(fill="x", pady=(0, 12))
auto_var = ctk.BooleanVar(value=True)
ctk.CTkCheckBox(frame, text="Открыть прокси в Telegram сейчас",
variable=auto_var, font=(theme.ui_font_family, 13),
text_color=theme.text_primary,
fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
corner_radius=6, border_width=2,
border_color=theme.field_border).pack(anchor="w", pady=(0, 16))
def on_ok():
on_done(auto_var.get())
ctk.CTkButton(frame, text="Начать", width=180, height=42,
font=(theme.ui_font_family, 15, "bold"), corner_radius=10,
fg_color=theme.tg_blue, hover_color=theme.tg_blue_hover,
text_color="#ffffff",
command=on_ok).pack(pady=(0, 0))
root.protocol("WM_DELETE_WINDOW", on_ok)

5
utils/__init__.py Normal file
View File

@ -0,0 +1,5 @@
"""Вспомогательные утилиты (проверка релизов и т.п.)."""
from utils.update_check import RELEASES_PAGE_URL, get_status, run_check
__all__ = ["RELEASES_PAGE_URL", "get_status", "run_check"]

27
utils/default_config.py Normal file
View File

@ -0,0 +1,27 @@
"""
Общие значения по умолчанию для tray-приложений (Windows / Linux / macOS).
Единственное отличие по платформе — ключ autostart только на Windows.
"""
from __future__ import annotations
import sys
from typing import Any, Dict
_TRAY_DEFAULTS_COMMON: Dict[str, Any] = {
"port": 1080,
"host": "127.0.0.1",
"dc_ip": ["2:149.154.167.220", "4:149.154.167.220"],
"verbose": False,
"check_updates": True,
"log_max_mb": 5,
"buf_kb": 256,
"pool_size": 4,
}
def default_tray_config() -> Dict[str, Any]:
"""Новая копия конфига по умолчанию для текущей ОС."""
cfg = dict(_TRAY_DEFAULTS_COMMON)
if sys.platform == "win32":
cfg["autostart"] = False
return cfg

223
utils/update_check.py Normal file
View File

@ -0,0 +1,223 @@
"""
Минимальная проверка новой версии через GitHub Releases API (без сторонних зависимостей).
Ограничение частоты запросов: не чаще одного раза в час на машину (кэш в каталоге
данных приложения). Поддерживается If-None-Match (ETag) для ответа 304.
"""
from __future__ import annotations
import json
import os
import sys
import time
from itertools import zip_longest
from pathlib import Path
from typing import Any, Dict, Optional, Tuple
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
REPO = "Flowseal/tg-ws-proxy"
RELEASES_LATEST_API = f"https://api.github.com/repos/{REPO}/releases/latest"
RELEASES_PAGE_URL = f"https://github.com/{REPO}/releases/latest"
# Не чаще одного полного запроса к API в час (без учёта 304 с тем же ETag).
_MIN_FETCH_INTERVAL_SEC = 3600.0
_state: Dict[str, Any] = {
"checked": False,
"has_update": False,
"ahead_of_release": False,
"latest": None,
"html_url": None,
"error": None,
}
def _cache_file() -> Optional[Path]:
try:
if sys.platform == "win32":
root = Path(os.environ.get("APPDATA", str(Path.home()))) / "TgWsProxy"
elif sys.platform == "darwin":
root = Path.home() / "Library/Application Support/TgWsProxy"
else:
xdg = os.environ.get("XDG_CONFIG_HOME")
root = (Path(xdg).expanduser() if xdg else Path.home() / ".config") / "TgWsProxy"
root.mkdir(parents=True, exist_ok=True)
return root / ".update_check_cache.json"
except OSError:
return None
def _load_cache(path: Optional[Path]) -> Dict[str, Any]:
if not path or not path.is_file():
return {}
try:
return json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {}
def _save_cache(path: Optional[Path], data: Dict[str, Any]) -> None:
if not path:
return
try:
path.write_text(json.dumps(data), encoding="utf-8")
except OSError:
pass
def _parse_version_tuple(s: str) -> tuple:
s = (s or "").strip().lstrip("vV")
if not s:
return (0,)
parts = []
for seg in s.split("."):
digits = "".join(c for c in seg if c.isdigit())
if digits:
try:
parts.append(int(digits))
except ValueError:
parts.append(0)
else:
parts.append(0)
return tuple(parts) if parts else (0,)
def _version_gt(a: str, b: str) -> bool:
"""True, если версия a новее b (простое сравнение по сегментам)."""
ta = _parse_version_tuple(a)
tb = _parse_version_tuple(b)
for x, y in zip_longest(ta, tb, fillvalue=0):
if x > y:
return True
if x < y:
return False
return False
def _apply_release_tag(
tag: str, html_url: str, current_version: str,
) -> None:
global _state
if not tag:
_state["has_update"] = False
_state["ahead_of_release"] = False
_state["latest"] = None
_state["html_url"] = html_url.strip() or RELEASES_PAGE_URL
return
latest_clean = tag.lstrip("vV")
cur = (current_version or "").strip().lstrip("vV")
_state["latest"] = latest_clean
_state["html_url"] = html_url.strip() or RELEASES_PAGE_URL
_state["has_update"] = _version_gt(latest_clean, cur)
_state["ahead_of_release"] = bool(latest_clean) and _version_gt(
cur, latest_clean
)
def fetch_latest_release(
timeout: float = 12.0,
etag: Optional[str] = None,
) -> Tuple[Optional[dict], Optional[str], int]:
"""
GET releases/latest. Возвращает (data или None при 304, etag или None, HTTP-код).
"""
headers = {
"Accept": "application/vnd.github+json",
"User-Agent": "tg-ws-proxy-update-check",
}
if etag:
headers["If-None-Match"] = etag
req = Request(
RELEASES_LATEST_API,
headers=headers,
method="GET",
)
try:
with urlopen(req, timeout=timeout) as resp:
code = getattr(resp, "status", None) or resp.getcode()
new_etag = resp.headers.get("ETag")
raw = resp.read().decode("utf-8", errors="replace")
return json.loads(raw), new_etag, int(code)
except HTTPError as e:
if e.code == 304:
hdrs = e.headers
new_etag = hdrs.get("ETag") if hdrs else None
return None, new_etag or etag, 304
raise
def run_check(current_version: str) -> None:
"""Запрашивает последний релиз и обновляет внутреннее состояние."""
global _state
_state["checked"] = True
_state["error"] = None
cache_path = _cache_file()
cache = _load_cache(cache_path)
now = time.time()
last_attempt = float(cache.get("last_attempt_at") or 0)
if last_attempt and (now - last_attempt) < _MIN_FETCH_INTERVAL_SEC:
tag = (cache.get("tag_name") or "").strip()
if tag:
_apply_release_tag(tag, cache.get("html_url") or "", current_version)
return
err = cache.get("last_error")
_state["error"] = (
err if err else "Проверка обновлений отложена (интервал между запросами)."
)
_state["has_update"] = False
_state["ahead_of_release"] = False
_state["latest"] = None
_state["html_url"] = RELEASES_PAGE_URL
return
etag = (cache.get("etag") or "").strip() or None
try:
data, new_etag, code = fetch_latest_release(etag=etag)
cache["last_attempt_at"] = now
if code == 304:
tag = (cache.get("tag_name") or "").strip()
url = (cache.get("html_url") or "").strip() or RELEASES_PAGE_URL
_apply_release_tag(tag, url, current_version)
if new_etag:
cache["etag"] = new_etag
_save_cache(cache_path, cache)
return
assert data is not None
tag = (data.get("tag_name") or "").strip()
html_url = (data.get("html_url") or "").strip() or RELEASES_PAGE_URL
if not tag:
_state["has_update"] = False
_state["ahead_of_release"] = False
_state["latest"] = None
_state["html_url"] = html_url
else:
_apply_release_tag(tag, html_url, current_version)
if new_etag:
cache["etag"] = new_etag
cache["tag_name"] = tag
cache["html_url"] = html_url
cache.pop("last_error", None)
_save_cache(cache_path, cache)
except (HTTPError, URLError, OSError, TimeoutError, ValueError, json.JSONDecodeError) as e:
cache["last_attempt_at"] = now
msg = str(e)
if isinstance(e, HTTPError) and e.code == 403:
msg = (
"GitHub API вернул 403 (лимит или доступ). Повторите позже."
)
cache["last_error"] = msg
_save_cache(cache_path, cache)
_state["error"] = msg
_state["has_update"] = False
_state["ahead_of_release"] = False
_state["latest"] = None
_state["html_url"] = RELEASES_PAGE_URL
def get_status() -> Dict[str, Any]:
"""Снимок состояния после run_check (для подписей в настройках)."""
return dict(_state)

470
windows.py
View File

@ -1,6 +1,7 @@
from __future__ import annotations
import ctypes
import ipaddress
import json
import logging
import logging.handlers
@ -11,16 +12,49 @@ import sys
import threading
import time
import webbrowser
import pyperclip
import ipaddress
from pathlib import Path
from typing import Optional
import pystray
import customtkinter as ctk
from PIL import Image, ImageDraw, ImageFont
try:
import pyperclip
except ImportError:
pyperclip = None
try:
import pystray
except ImportError:
pystray = None
try:
import customtkinter as ctk
except ImportError:
ctk = None
try:
from PIL import Image, ImageDraw, ImageFont
except ImportError:
Image = ImageDraw = ImageFont = None
import proxy.tg_ws_proxy as tg_ws_proxy
from proxy.app_runtime import ProxyAppRuntime
from proxy import __version__
from utils.default_config import default_tray_config
from ui.ctk_tray_ui import (
install_tray_config_buttons,
install_tray_config_form,
populate_first_run_window,
tray_settings_scroll_and_footer,
validate_config_form,
)
from ui.ctk_theme import (
CONFIG_DIALOG_FRAME_PAD,
CONFIG_DIALOG_SIZE,
FIRST_RUN_SIZE,
create_ctk_root,
ctk_theme_for_platform,
main_content_frame,
)
IS_FROZEN = bool(getattr(sys, "frozen", False))
@ -33,16 +67,7 @@ FIRST_RUN_MARKER = APP_DIR / ".first_run_done"
IPV6_WARN_MARKER = APP_DIR / ".ipv6_warned"
DEFAULT_CONFIG = {
"port": 1080,
"host": "127.0.0.1",
"dc_ip": ["2:149.154.167.220", "4:149.154.167.220"],
"verbose": False,
"autostart": False,
"log_max_mb": 5,
"buf_kb": 256,
"pool_size": 4,
}
DEFAULT_CONFIG = default_tray_config()
_tray_icon: Optional[object] = None
@ -60,6 +85,15 @@ _runtime = ProxyAppRuntime(
CONFIG_FILE = _runtime.config_file
LOG_FILE = _runtime.log_file
_user32 = ctypes.windll.user32
_user32.MessageBoxW.argtypes = [
ctypes.c_void_p,
ctypes.c_wchar_p,
ctypes.c_wchar_p,
ctypes.c_uint,
]
_user32.MessageBoxW.restype = ctypes.c_int
def _same_process(lock_meta: dict, proc: psutil.Process) -> bool:
try:
@ -70,9 +104,18 @@ def _same_process(lock_meta: dict, proc: psutil.Process) -> bool:
except Exception:
return False
try:
for arg in proc.cmdline():
if "windows.py" in arg:
return True
except Exception:
pass
frozen = bool(getattr(sys, "frozen", False))
if frozen:
return os.path.basename(sys.executable) == proc.name()
return (
os.path.basename(sys.executable).lower() == proc.name().lower()
)
return False
@ -255,16 +298,61 @@ def restart_proxy():
def _show_error(text: str, title: str = "TG WS Proxy — Ошибка"):
ctypes.windll.user32.MessageBoxW(0, text, title, 0x10)
_user32.MessageBoxW(None, text, title, 0x10)
def _show_info(text: str, title: str = "TG WS Proxy"):
ctypes.windll.user32.MessageBoxW(0, text, title, 0x40)
_user32.MessageBoxW(None, text, title, 0x40)
def _ask_open_release_page(latest_version: str, url: str) -> bool:
"""Win32 Yes/No: открыть страницу релиза."""
MB_YESNO = 0x4
MB_ICONQUESTION = 0x20
IDYES = 6
text = (
f"Доступна новая версия: {latest_version}\n\n"
f"Открыть страницу релиза в браузере?"
)
r = _user32.MessageBoxW(
None,
text,
"TG WS Proxy — обновление",
MB_YESNO | MB_ICONQUESTION,
)
return r == IDYES
def _maybe_notify_update_async():
"""
Фоновая проверка GitHub Releases и уведомление (не блокирует трей).
"""
def _work():
time.sleep(1.5)
if _exiting:
return
if not _config.get("check_updates", True):
return
try:
from utils.update_check import RELEASES_PAGE_URL, get_status, run_check
run_check(__version__)
st = get_status()
if not st.get("has_update"):
return
url = (st.get("html_url") or "").strip() or RELEASES_PAGE_URL
ver = st.get("latest") or "?"
if _ask_open_release_page(str(ver), url):
webbrowser.open(url)
except Exception as exc:
log.debug("Update check failed: %s", exc)
threading.Thread(target=_work, daemon=True, name="update-check").start()
def _on_open_in_telegram(icon=None, item=None):
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
url = f"tg://socks?server=127.0.0.1&port={port}"
url = f"tg://socks?server={host}&port={port}"
log.info("Opening %s", url)
try:
result = webbrowser.open(url)
@ -272,6 +360,11 @@ def _on_open_in_telegram(icon=None, item=None):
raise RuntimeError("webbrowser.open returned False")
except Exception:
log.info("Browser open failed, copying to clipboard")
if pyperclip is None:
_show_error(
"Не удалось открыть Telegram автоматически.\n\n"
f"Установите пакет pyperclip для копирования в буфер или откройте вручную:\n{url}")
return
try:
pyperclip.copy(url)
_show_info(
@ -299,165 +392,53 @@ def _edit_config_dialog():
cfg = dict(_config)
cfg["autostart"] = is_autostart_enabled()
# Make sure that the autostart key is removed if autostart
# Make sure that the autostart key is removed if autostart
# is disabled, even if the executable file is moved.
if _supports_autostart() and not cfg["autostart"]:
set_autostart_enabled(False)
ctk.set_appearance_mode("light")
ctk.set_default_color_theme("blue")
root = ctk.CTk()
root.title("TG WS Proxy — Настройки")
root.resizable(False, False)
root.attributes("-topmost", True)
icon_path = str(Path(__file__).parent / "icon.ico")
root.iconbitmap(icon_path)
TG_BLUE = "#3390ec"
TG_BLUE_HOVER = "#2b7cd4"
BG = "#ffffff"
FIELD_BG = "#f0f2f5"
FIELD_BORDER = "#d6d9dc"
TEXT_PRIMARY = "#000000"
TEXT_SECONDARY = "#707579"
FONT_FAMILY = "Segoe UI"
w, h = 420, 540
if _supports_autostart():
h += 70
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
root.geometry(f"{w}x{h}+{(sw-w)//2}+{(sh-h)//2}")
root.configure(fg_color=BG)
frame = ctk.CTkFrame(root, fg_color=BG, corner_radius=0)
frame.pack(fill="both", expand=True, padx=24, pady=20)
# Host
ctk.CTkLabel(frame, text="IP-адрес прокси",
font=(FONT_FAMILY, 13), text_color=TEXT_PRIMARY,
anchor="w").pack(anchor="w", pady=(0, 4))
host_var = ctk.StringVar(value=cfg.get("host", "127.0.0.1"))
host_entry = ctk.CTkEntry(frame, textvariable=host_var, width=200, height=36,
font=(FONT_FAMILY, 13), corner_radius=10,
fg_color=FIELD_BG, border_color=FIELD_BORDER,
border_width=1, text_color=TEXT_PRIMARY)
host_entry.pack(anchor="w", pady=(0, 12))
# Port
ctk.CTkLabel(frame, text="Порт прокси",
font=(FONT_FAMILY, 13), text_color=TEXT_PRIMARY,
anchor="w").pack(anchor="w", pady=(0, 4))
port_var = ctk.StringVar(value=str(cfg.get("port", 1080)))
port_entry = ctk.CTkEntry(frame, textvariable=port_var, width=120, height=36,
font=(FONT_FAMILY, 13), corner_radius=10,
fg_color=FIELD_BG, border_color=FIELD_BORDER,
border_width=1, text_color=TEXT_PRIMARY)
port_entry.pack(anchor="w", pady=(0, 12))
# DC-IP mappings
ctk.CTkLabel(frame, text="DC → IP маппинги (по одному на строку, формат DC:IP)",
font=(FONT_FAMILY, 13), text_color=TEXT_PRIMARY,
anchor="w").pack(anchor="w", pady=(0, 4))
dc_textbox = ctk.CTkTextbox(frame, width=370, height=120,
font=("Consolas", 12), corner_radius=10,
fg_color=FIELD_BG, border_color=FIELD_BORDER,
border_width=1, text_color=TEXT_PRIMARY)
dc_textbox.pack(anchor="w", pady=(0, 12))
dc_textbox.insert("1.0", "\n".join(cfg.get("dc_ip", DEFAULT_CONFIG["dc_ip"])))
# Verbose
verbose_var = ctk.BooleanVar(value=cfg.get("verbose", False))
ctk.CTkCheckBox(frame, text="Подробное логирование (verbose)",
variable=verbose_var, font=(FONT_FAMILY, 13),
text_color=TEXT_PRIMARY,
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
corner_radius=6, border_width=2,
border_color=FIELD_BORDER).pack(anchor="w", pady=(0, 8))
# Advanced: buf_kb, pool_size, log_max_mb
adv_frame = ctk.CTkFrame(frame, fg_color="transparent")
adv_frame.pack(anchor="w", fill="x", pady=(4, 8))
for col, (lbl, key, w_) in enumerate([
("Буфер (KB, 256 default)", "buf_kb", 120),
("WS пулов (4 default)", "pool_size", 120),
("Log size (MB, 5 def)", "log_max_mb", 120),
]):
col_frame = ctk.CTkFrame(adv_frame, fg_color="transparent")
col_frame.pack(side="left", padx=(0, 10))
ctk.CTkLabel(col_frame, text=lbl, font=(FONT_FAMILY, 11),
text_color=TEXT_SECONDARY, anchor="w").pack(anchor="w")
ctk.CTkEntry(col_frame, width=w_, height=30, font=(FONT_FAMILY, 12),
corner_radius=8, fg_color=FIELD_BG,
border_color=FIELD_BORDER, border_width=1,
text_color=TEXT_PRIMARY,
textvariable=ctk.StringVar(
value=str(cfg.get(key, DEFAULT_CONFIG[key]))
)).pack(anchor="w")
_adv_entries = list(adv_frame.winfo_children())
_adv_keys = ["buf_kb", "pool_size", "log_max_mb"]
autostart_var = None
theme = ctk_theme_for_platform()
w, h = CONFIG_DIALOG_SIZE
if _supports_autostart():
autostart_var = ctk.BooleanVar(value=cfg["autostart"])
ctk.CTkCheckBox(frame, text="Автозапуск при включении Windows",
variable=autostart_var, font=(FONT_FAMILY, 13),
text_color=TEXT_PRIMARY,
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
corner_radius=6, border_width=2,
border_color=FIELD_BORDER).pack(anchor="w", pady=(0, 8))
ctk.CTkLabel(frame, text="При перемещении файла или открытии из другой папки\nавтозапуск будет сброшен",
font=(FONT_FAMILY, 13), text_color=TEXT_SECONDARY,
anchor="w", justify="left").pack(anchor="w", pady=(0, 8))
h += 100
icon_path = str(Path(__file__).parent / "icon.ico")
root = create_ctk_root(
ctk,
title="TG WS Proxy — Настройки",
width=w,
height=h,
theme=theme,
after_create=lambda r: r.iconbitmap(icon_path),
)
fpx, fpy = CONFIG_DIALOG_FRAME_PAD
frame = main_content_frame(ctk, root, theme, padx=fpx, pady=fpy)
scroll, footer = tray_settings_scroll_and_footer(ctk, frame, theme)
widgets = install_tray_config_form(
ctk,
scroll,
theme,
cfg,
DEFAULT_CONFIG,
show_autostart=_supports_autostart(),
autostart_value=cfg.get("autostart", False),
)
def on_save():
import socket as _sock
host_val = host_var.get().strip()
try:
_sock.inet_aton(host_val)
except OSError:
_show_error("Некорректный IP-адрес.")
merged = validate_config_form(
widgets,
DEFAULT_CONFIG,
include_autostart=_supports_autostart(),
)
if isinstance(merged, str):
_show_error(merged)
return
try:
port_val = int(port_var.get().strip())
if not (1 <= port_val <= 65535):
raise ValueError
except ValueError:
_show_error("Порт должен быть числом 1-65535")
return
lines = [l.strip() for l in dc_textbox.get("1.0", "end").strip().splitlines()
if l.strip()]
try:
tg_ws_proxy.parse_dc_ip_list(lines)
except ValueError as e:
_show_error(str(e))
return
new_cfg = {
"host": host_val,
"port": port_val,
"dc_ip": lines,
"verbose": verbose_var.get(),
"autostart": (autostart_var.get() if autostart_var is not None else False),
}
for i, key in enumerate(_adv_keys):
col_frame = _adv_entries[i]
entry = col_frame.winfo_children()[1]
try:
val = float(entry.get().strip())
if key in ("buf_kb", "pool_size"):
val = int(val)
new_cfg[key] = val
except ValueError:
new_cfg[key] = DEFAULT_CONFIG[key]
new_cfg = merged
save_config(new_cfg)
_config.update(new_cfg)
log.info("Config saved: %s", new_cfg)
@ -467,6 +448,8 @@ def _edit_config_dialog():
_tray_icon.menu = _build_menu()
# Win32 MessageBox из того же потока, что и mainloop CTk, блокирует обработку Tcl/Tk
# и даёт зависание; tkinter.messagebox согласован с циклом окна.
from tkinter import messagebox
if messagebox.askyesno("Перезапустить?",
"Настройки сохранены.\n\n"
@ -480,21 +463,18 @@ def _edit_config_dialog():
def on_cancel():
root.destroy()
btn_frame = ctk.CTkFrame(frame, fg_color="transparent")
btn_frame.pack(fill="x", pady=(20, 0))
ctk.CTkButton(btn_frame, text="Сохранить", height=38,
font=(FONT_FAMILY, 14, "bold"), corner_radius=10,
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
text_color="#ffffff",
command=on_save).pack(side="left", fill="x", expand=True, padx=(0, 8))
ctk.CTkButton(btn_frame, text="Отмена", height=38,
font=(FONT_FAMILY, 14), corner_radius=10,
fg_color=FIELD_BG, hover_color=FIELD_BORDER,
text_color=TEXT_PRIMARY, border_width=1,
border_color=FIELD_BORDER,
command=on_cancel).pack(side="right", fill="x", expand=True)
install_tray_config_buttons(
ctk, footer, theme, on_save=on_save, on_cancel=on_cancel)
root.mainloop()
try:
root.mainloop()
finally:
import tkinter as tk
try:
if root.winfo_exists():
root.destroy()
except tk.TclError:
pass
def _on_open_logs(icon=None, item=None):
@ -530,101 +510,41 @@ def _show_first_run():
host = _config.get("host", DEFAULT_CONFIG["host"])
port = _config.get("port", DEFAULT_CONFIG["port"])
tg_url = f"tg://socks?server={host}&port={port}"
if ctk is None:
FIRST_RUN_MARKER.touch()
return
ctk.set_appearance_mode("light")
ctk.set_default_color_theme("blue")
TG_BLUE = "#3390ec"
TG_BLUE_HOVER = "#2b7cd4"
BG = "#ffffff"
FIELD_BG = "#f0f2f5"
FIELD_BORDER = "#d6d9dc"
TEXT_PRIMARY = "#000000"
TEXT_SECONDARY = "#707579"
FONT_FAMILY = "Segoe UI"
root = ctk.CTk()
root.title("TG WS Proxy")
root.resizable(False, False)
root.attributes("-topmost", True)
theme = ctk_theme_for_platform()
icon_path = str(Path(__file__).parent / "icon.ico")
root.iconbitmap(icon_path)
w, h = FIRST_RUN_SIZE
root = create_ctk_root(
ctk,
title="TG WS Proxy",
width=w,
height=h,
theme=theme,
after_create=lambda r: r.iconbitmap(icon_path),
)
w, h = 520, 440
sw = root.winfo_screenwidth()
sh = root.winfo_screenheight()
root.geometry(f"{w}x{h}+{(sw-w)//2}+{(sh-h)//2}")
root.configure(fg_color=BG)
frame = ctk.CTkFrame(root, fg_color=BG, corner_radius=0)
frame.pack(fill="both", expand=True, padx=28, pady=24)
title_frame = ctk.CTkFrame(frame, fg_color="transparent")
title_frame.pack(anchor="w", pady=(0, 16), fill="x")
# Blue accent bar
accent_bar = ctk.CTkFrame(title_frame, fg_color=TG_BLUE,
width=4, height=32, corner_radius=2)
accent_bar.pack(side="left", padx=(0, 12))
ctk.CTkLabel(title_frame, text="Прокси запущен и работает в системном трее",
font=(FONT_FAMILY, 17, "bold"),
text_color=TEXT_PRIMARY).pack(side="left")
# Info sections
sections = [
("Как подключить Telegram Desktop:", True),
(" Автоматически:", True),
(f" ПКМ по иконке в трее → «Открыть в Telegram»", False),
(f" Или ссылка: {tg_url}", False),
("\n Вручную:", True),
(" Настройки → Продвинутые → Тип подключения → Прокси", False),
(f" SOCKS5 → {host} : {port} (без логина/пароля)", False),
]
for text, bold in sections:
weight = "bold" if bold else "normal"
ctk.CTkLabel(frame, text=text,
font=(FONT_FAMILY, 13, weight),
text_color=TEXT_PRIMARY,
anchor="w", justify="left").pack(anchor="w", pady=1)
# Spacer
ctk.CTkFrame(frame, fg_color="transparent", height=16).pack()
# Separator
ctk.CTkFrame(frame, fg_color=FIELD_BORDER, height=1,
corner_radius=0).pack(fill="x", pady=(0, 12))
# Checkbox
auto_var = ctk.BooleanVar(value=True)
ctk.CTkCheckBox(frame, text="Открыть прокси в Telegram сейчас",
variable=auto_var, font=(FONT_FAMILY, 13),
text_color=TEXT_PRIMARY,
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
corner_radius=6, border_width=2,
border_color=FIELD_BORDER).pack(anchor="w", pady=(0, 16))
def on_ok():
def on_done(open_tg: bool):
FIRST_RUN_MARKER.touch()
open_tg = auto_var.get()
root.destroy()
if open_tg:
_on_open_in_telegram()
ctk.CTkButton(frame, text="Начать", width=180, height=42,
font=(FONT_FAMILY, 15, "bold"), corner_radius=10,
fg_color=TG_BLUE, hover_color=TG_BLUE_HOVER,
text_color="#ffffff",
command=on_ok).pack(pady=(0, 0))
populate_first_run_window(
ctk, root, theme, host=host, port=port, on_done=on_done)
root.protocol("WM_DELETE_WINDOW", on_ok)
root.mainloop()
try:
root.mainloop()
finally:
import tkinter as tk
try:
if root.winfo_exists():
root.destroy()
except tk.TclError:
pass
def _has_ipv6_enabled() -> bool:
@ -633,8 +553,15 @@ def _has_ipv6_enabled() -> bool:
addrs = _sock.getaddrinfo(_sock.gethostname(), None, _sock.AF_INET6)
for addr in addrs:
ip = addr[4][0]
if ip and not ip.startswith('::1') and not ip.startswith('fe80::1'):
return True
if not ip or ip.startswith("::1"):
continue
try:
if ipaddress.IPv6Address(ip).is_link_local:
continue
except ValueError:
if ip.startswith("fe80:"):
continue
return True
except Exception:
pass
try:
@ -699,13 +626,14 @@ def run_tray():
setup_logging(_config.get("verbose", False),
log_max_mb=_config.get("log_max_mb", DEFAULT_CONFIG["log_max_mb"]))
log.info("TG WS Proxy tray app starting")
log.info("TG WS Proxy версия %s, tray app starting", __version__)
log.info("Config: %s", _config)
log.info("Log file: %s", LOG_FILE)
if pystray is None or Image is None:
log.error("pystray or Pillow not installed; "
"running in console mode")
if pystray is None or Image is None or ctk is None:
log.error(
"pystray, Pillow or customtkinter not installed; "
"running in console mode")
start_proxy()
try:
while True:
@ -716,6 +644,8 @@ def run_tray():
start_proxy()
_maybe_notify_update_async()
_show_first_run()
_check_ipv6_warning()