@@ -185,6 +185,43 @@ def _bot_status_text():
return " \n " . join ( lines )
def qr_png_bytes ( data ) :
import qrcode , io
buf = io . BytesIO ( ) ; qrcode . make ( data ) . save ( buf , format = " PNG " ) ; return buf . getvalue ( )
def tg_send_photo ( chat , png , caption = " " ) :
import urllib . request , secrets as _sec
token = load_settings ( ) . get ( " tg_bot_token " , " " )
if not token :
return None
boundary = " ----phobos " + _sec . token_hex ( 8 )
def fld ( name , val ) :
return ( " -- %s \r \n Content-Disposition: form-data; name= \" %s \" \r \n \r \n %s \r \n " % ( boundary , name , val ) ) . encode ( )
body = fld ( " chat_id " , str ( chat ) ) + fld ( " caption " , caption ) + fld ( " parse_mode " , " HTML " )
body + = ( " -- %s \r \n Content-Disposition: form-data; name= \" photo \" ; filename= \" qr.png \" \r \n Content-Type: image/png \r \n \r \n " % boundary ) . encode ( )
body + = png + ( " \r \n -- %s -- \r \n " % boundary ) . encode ( )
try :
req = urllib . request . Request ( " https://api.telegram.org/bot %s /sendPhoto " % token , data = body ,
headers = { " Content-Type " : " multipart/form-data; boundary= %s " % boundary } )
return urllib . request . urlopen ( req , timeout = 25 ) . read ( )
except Exception :
return None
def bot_send_new_client ( chat , name ) :
ok , rep = create_client_cli ( name )
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : rep , " parse_mode " : " HTML " , " reply_markup " : MAIN_KB } )
if ok :
try :
conf , link = build_phone_config ( name , " android " )
if link :
tg_send_photo ( chat , qr_png_bytes ( link ) , " 📱 Android (PhobosWG) — отсканируй QR в приложении " )
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : " phobos:// ссылка (Android): \n <code> %s </code> " % link , " parse_mode " : " HTML " } )
except Exception :
pass
def telegram_bot ( ) :
offset = 0
import urllib . request
@@ -217,23 +254,40 @@ def telegram_bot():
text = ( m . get ( " text " ) or " " ) . strip ( )
if not text :
continue
low = text . lower (
if _bot_state . get ( chat ) == " add " and not text . startswith ( " / " ) :
_bot_state . pop ( chat , None )
_ , rep = create_client_cli ( text )
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : rep , " parse_mode " : " HTML " , " reply_markup " : MAIN_KB } )
continue
toks = text . split ( )
cmd = toks [ 0 ] . spl it( " @ " ) [ 0 ] . lower ( )
arg = toks [ 1 ] if len ( toks ) > 1 else " "
text . startswith ( " ➕ " ) or cmd in ( " /add " , " /new " , " /create " ) :
# detect reply-keyboard button by its emoji prefix (NOT a command/arg )
btn = None
if not text . startswith ( " / " ) :
if text . startswith ( " ➕ " ) :
btn = " add "
elif text . startswith ( " \U0001F5D1 " ) :
btn = " del "
elif text . startsw ith ( " \U0001F4CB " ) :
btn = " list "
el if text . startswith ( " \U0001F4CA " ) :
btn = " status "
elif text . startswith ( " ❓ " ) :
btn = " help "
# conversation: awaiting a client name to create
if _bot_state . get ( chat ) == " add " :
if btn is None and not text . startswith ( " / " ) :
_bot_state . pop ( chat , None )
bot_send_new_client ( chat , text )
continue
_bot_state . pop ( chat , None ) # a button/command cancels the prompt
# parse slash command + arg (buttons carry no arg)
cmd = " "
arg = " "
if text . startswith ( " / " ) :
toks = text . split ( )
cmd = toks [ 0 ] . split ( " @ " ) [ 0 ] . lower ( )
arg = toks [ 1 ] if len ( toks ) > 1 else " "
if btn == " add " or cmd in ( " /add " , " /new " , " /create " ) :
if arg :
_ , rep = create_client_cli ( arg )
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : rep , " parse_mode " : " HTML " , " reply_markup " : MAIN_KB } )
bot_send_new_client ( chat , arg )
else :
_bot_state [ chat ] = " add "
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : " Введите имя нового клиента (латиница, цифры, _ и -): " } )
elif text . startswith ( " \U0001F5D1 " ) or cmd in ( " /del " , " /delete " , " /remove " , " /rm " ) :
elif btn == " del " or cmd in ( " /del " , " /delete " , " /remove " , " /rm " ) :
if arg :
_ , rep = delete_client_cli ( arg )
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : rep , " parse_mode " : " HTML " , " reply_markup " : MAIN_KB } )
@@ -243,14 +297,14 @@ def telegram_bot():
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : " Клиентов нет. " , " reply_markup " : MAIN_KB } )
else :
kb = { " inline_keyboard " : [ [ { " text " : " \U0001F5D1 " + c . get ( " client_id " , " " ) , " callback_data " : " del: " + c . get ( " client_id " , " " ) } ] for c in cl ] }
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : " К о г о " , " reply_markup " : kb } )
elif text . startswith ( " \U0001F4CB " ) or cmd in ( " /list " , " /ls " ) :
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : " К о г о Нажми на клиента: " , " reply_markup " : kb } )
elif btn == " list " or cmd in ( " /list " , " /ls " ) :
cl = [ c . get ( " client_id " , " " ) for c in get_clients ( ) ]
rep = ( " Клиенты ( %d ): \n " % len ( cl ) ) + " \n " . join ( " • " + c for c in cl ) if cl else " Клиентов нет. "
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : rep , " reply_markup " : MAIN_KB } )
elif text . startswith ( " \U0001F4CA " ) or cmd in ( " /status " , " /stat " ) :
elif btn == " status " or cmd in ( " /status " , " /stat " ) :
tg_api ( " sendMessage " , { " chat_id " : chat , " text " : _bot_status_text ( ) , " parse_mode " : " HTML " , " reply_markup " : MAIN_KB } )
elif text . startswith ( " ❓ " ) or cmd in ( " /help " , " /start " ) :
elif btn == " help " or cmd in ( " /help " , " /start " ) :
rep = ( " <b>PCA Phobos — бот</b> \n \n "
" Кнопки внизу или команды: \n "
" ➕ \n "
@@ -425,6 +479,7 @@ def kick_client_by_id(client_id):
prev_session_keys = None
prev_server_status = { }
_down_streak = { }
_last_fanout = 0
server_stats_cache = { }
server_handshakes_cache = { }
@@ -449,12 +504,19 @@ def get_local_stats():
def get_remote_stats ( server ) :
import urllib . request
url = f " http:// { server [ ' ip ' ] } :8444/api/health "
req = urllib . request . Request ( url , headers = { " X-API-Key " : server . get ( " api_key " , " " ) } )
data = None
for _attempt in range ( 2 ) : # retry once: smooths transient timeouts (1-worker agent)
try :
data = json . loads ( urllib . request . urlopen ( req , timeout = 6 ) . read ( ) )
break
except Exception :
data = None
try :
import urllib . request
url = f " http:// { server [ ' ip ' ] } :8444/api/health "
req = urllib . request . Request ( url , headers = { " X-API-Key " : server . get ( " api_key " , " " ) } )
resp = urllib . request . urlopen ( req , timeout = 3 )
data = json . loads ( resp . read ( ) )
if data is None :
raise Exception ( " no response " )
# cache handshakes so page renders never block on remote HTTP
server_handshakes_cache [ server . get ( " ip " , " " ) ] = data . get ( " handshakes " , { } ) or { }
# Filter peers: only count known client public keys
@@ -723,7 +785,7 @@ def push_config_to_router(client_id):
def session_monitor ( ) :
global prev_session_keys , prev_server_status , server_stats_cache , _last_fanout
global prev_session_keys , prev_server_status , server_stats_cache , _last_fanout , _down_streak
while True :
try :
s = load_settings ( )
@@ -760,19 +822,24 @@ def session_monitor():
local_stats = get_local_stats ( )
server_stats_cache [ SERVER_IP ] = local_stats
DOWN_CONFIRM = 3 # consecutive failed cycles before declaring a server DOWN
for srv in servers :
ip = srv [ " ip " ]
stats = get_remote_stats ( srv )
server_stats_cache [ ip ] =
raw = stats . get ( " u s" , " down " )
streak = _down_streak . get ( ip , 0 )
streak = streak + 1 if raw == " down " else 0
_down_streak [ ip ] = streak
# update cache on OK, or only once a DOWN is confirmed (keep last-good on transient)
if raw == " ok " or streak > = DOWN_CONFIRM :
server_stats_cache [ ip ] = stats
eff = " down " if streak > = DOWN_CONFIRM else " ok "
was_up = prev_server_status . get ( ip , " ok " )
now_status = stats [ " status " ]
if was_up == " ok " and now_status == " down " :
if was_up == " ok " and eff == " down " :
tg_send ( f " 🔴 <b>Server { ip } </b> DOWN! " )
elif was_up == " down " and now_status == " ok " :
elif was_up == " down " and eff == " ok " :
tg_send ( f " 🟢 <b>Server { ip } </b> back ONLINE " )
prev_server_status [ ip ] = now_status
prev_server_status [ ip ] = eff
# periodic config fan-out to secondaries (covers CLI-created clients
# and config drift; gated ~5 min, skips down servers)
@@ -854,7 +921,7 @@ input:focus{outline:none;border-color:#4f46e5}
CONTENT
<div class= " footer " >
<b style= " color:#a5b4fc " >PCA Phobos Panel</b> — панель управления © <a href= " https://github.com/andrey271192/PCA_Phobos " target= " _blank " rel= " noopener " >andrey271192</a><br>
Поддержать проект: <a href= " https://boosty.to/andrey27/donate " target= " _blank " rel= " noopener " >💖 Boosty</a> · <a href= " https://finance.ozon.ru/apps/sbp/ozonbankpay/019dc200-2a5d-7931-a619-782d285f6798 " target= " _blank " rel= " noopener " >💳 Ozon Bank</a> · <a href= " https://t.me/c/2474115507/78722 " target= " _blank " rel= " noopener " >✉️ Telegram</a>
Поддержать проект: <a href= " https://boosty.to/andrey27/donate " target= " _blank " rel= " noopener " >💖 Boosty</a> · <a href= " https://finance.ozon.ru/apps/sbp/ozonbankpay/019dc200-2a5d-7931-a619-782d285f6798 " target= " _blank " rel= " noopener " >💳 Ozon Bank</a> · <a href= " https://t.me/PCAdministration " target= " _blank " rel= " noopener " >✉️ Telegram</a>
<div style= " margin-top:8px;font-size:.88em;color:#475569;line-height:1.7;text-align:left;max-width:660px;margin:8px auto 0 " >
<b>Создано на основе открытых проектов:</b><br>
• <a href= " https://github.com/Ground-Zerro/Phobos " target= " _blank " rel= " noopener " >Phobos</a> — обфусцированный WireGuard VPN, автор <b>Ground_Zerro</b> · <a href= " https://boosty.to/ground_zerro " target= " _blank " rel= " noopener " >поддержать</a><br>
@@ -1317,9 +1384,12 @@ def labels_page():
def settings_page ( ) :
s = load_settings ( )
msg = " "
if request . method == " POST " and request . form . get ( " action " ) in ( " update " , " rollback " ) :
if request . form . get ( " action " ) == " update " :
if request . method == " POST " and request . form . get ( " action " ) in ( " update " , " rollback " , " check " :
_act = request . form . get ( " action " )
if _act == " update " :
_out = run_update ( request . form . get ( " target " , " " ) . strip ( ) or " main " )
elif _act == " check " :
_out = run_update ( " --check " )
else :
_out = run_update ( " --rollback " )
s = load_settings ( )
@@ -1360,9 +1430,13 @@ def settings_page():
<p> { tr ( " Установленная версия " , " Installed version " ) } : <b style= " color:#a5b4fc " > { panel_version ( ) } </b></p>
<form method= " post " class= " form-row " >
<input type= " hidden " name= " action " value= " update " >
<input type= " text " name= " target " placeholder= " main / v1.1.0 " style= " width:150px " >
<input type= " text " name= " target " placeholder= " main / beta / v1.2.3 " style= " width:150px " >
<button class= " btn btn-primary " type= " submit " onclick= " return confirm( ' { tr ( " Обновить панель и скрипты? " , " Update panel and scripts? " ) } ' ) " > { tr ( " Обновить " , " Update " ) } </button>
</form>
<form method= " post " style= " display:inline " >
<input type= " hidden " name= " action " value= " check " >
<button class= " btn btn-sm " style= " background:#0ea5e9 " type= " submit " > { tr ( " Проверить обновления " , " Check for updates " ) } </button>
</form>
<form method= " post " style= " margin-top:8px " >
<input type= " hidden " name= " action " value= " rollback " >
<button class= " btn btn-danger btn-sm " type= " submit " onclick= " return confirm( ' { tr ( " Откатить на предыдущую версию? " , " Roll back to the previous version? " ) } ' ) " > { tr ( " Откатить на предыдущую " , " Rollback " ) } </button>
