#!/usr/bin/env python3
"""
PCA Phobos — Web Panel for Phobos (Obfuscated WireGuard VPN)
Management panel: clients, sessions, labels, subscriptions, Telegram alerts.
"""
import json, os, subprocess, threading, time, secrets, hashlib, re
from datetime import datetime, timedelta
from pathlib import Path
from flask import Flask, request, redirect, url_for, session, make_response
app = Flask(__name__)
PHOBOS_DIR = "/opt/Phobos"
CLIENTS_DIR = f"{PHOBOS_DIR}/clients"
SERVER_ENV = f"{PHOBOS_DIR}/server/server.env"
PANEL_DIR = "/opt/phobos-panel"
SETTINGS_FILE = f"{PANEL_DIR}/settings.json"
SECRET_FILE = f"{PANEL_DIR}/.secret_key"
SERVER_IP = subprocess.getoutput("curl -s https://api.ipify.org 2>/dev/null || hostname -I | awk '{print $1}'").strip()
os.makedirs(PANEL_DIR, exist_ok=True)
if os.path.exists(SECRET_FILE):
app.secret_key = open(SECRET_FILE).read().strip()
else:
app.secret_key = secrets.token_hex(32)
with open(SECRET_FILE, "w") as f:
f.write(app.secret_key)
DEFAULT_SETTINGS = {
"admin_pass": "OcAdmin2026!",
"tg_bot_token": "",
"tg_chat_id": "",
"monitor_interval": 30,
"labels": {},
"subscriptions": {}
}
def load_settings():
if os.path.exists(SETTINGS_FILE):
with open(SETTINGS_FILE) as f:
s = json.load(f)
for k, v in DEFAULT_SETTINGS.items():
s.setdefault(k, v)
return s
return dict(DEFAULT_SETTINGS)
def save_settings(s):
with open(SETTINGS_FILE, "w") as f:
json.dump(s, f, indent=2, ensure_ascii=False)
def tg_send(text):
s = load_settings()
token, chat = s.get("tg_bot_token", ""), s.get("tg_chat_id", "")
if not token or not chat:
return
try:
import urllib.request
url = f"https://api.telegram.org/bot{token}/sendMessage"
data = json.dumps({"chat_id": chat, "text": text, "parse_mode": "HTML"}).encode()
req = urllib.request.Request(url, data=data, headers={"Content-Type": "application/json"})
urllib.request.urlopen(req, timeout=10)
except Exception:
pass
def get_wg_peers():
"""Parse `wg show wg0` to get active peers with transfer/handshake info."""
try:
out = subprocess.check_output(["wg", "show", "wg0"], text=True, timeout=5)
except Exception:
return {}
peers = {}
current_pub = None
for line in out.split("\n"):
line = line.strip()
if line.startswith("peer:"):
current_pub = line.split("peer:")[1].strip()
peers[current_pub] = {}
elif current_pub and ":" in line:
key, val = line.split(":", 1)
peers[current_pub][key.strip()] = val.strip()
return peers
def get_clients():
"""Read all clients from /opt/Phobos/clients/*/metadata.json."""
clients = []
clients_path = Path(CLIENTS_DIR)
if not clients_path.exists():
return clients
for d in sorted(clients_path.iterdir()):
meta_file = d / "metadata.json"
if meta_file.exists():
try:
with open(meta_file) as f:
meta = json.load(f)
meta["_dir"] = str(d)
clients.append(meta)
except Exception:
pass
return clients
def get_active_sessions():
"""Combine WG peers with client metadata to build session list."""
peers = get_wg_peers()
clients = get_clients()
pub_to_client = {}
for c in clients:
pub_to_client[c.get("public_key", "")] = c
sessions = []
for pub_key, info in peers.items():
handshake = info.get("latest handshake", "")
if not handshake:
continue
client = pub_to_client.get(pub_key, {})
client_id = client.get("client_id", "unknown")
tunnel_ip = client.get("tunnel_ip_v4", "")
endpoint = info.get("endpoint", "")
real_ip = endpoint.split(":")[0] if endpoint else ""
rx = info.get("transfer", "")
rx_bytes = rx.split("received,")[0].strip() if "received," in rx else ""
tx_bytes = rx.split("received,")[1].strip().replace("sent", "").strip() if "received," in rx else ""
sessions.append({
"client_id": client_id,
"public_key": pub_key,
"tunnel_ip": tunnel_ip,
"real_ip": real_ip,
"endpoint": endpoint,
"handshake": handshake,
"rx": rx_bytes,
"tx": tx_bytes,
})
return sessions
def is_peer_online(handshake_str):
"""Check if peer had a handshake within last 3 minutes."""
try:
parts = handshake_str.split(",")
total_seconds = 0
for p in parts:
p = p.strip()
if "minute" in p:
total_seconds += int(re.search(r"(\d+)", p).group(1)) * 60
elif "second" in p:
total_seconds += int(re.search(r"(\d+)", p).group(1))
elif "hour" in p:
total_seconds += int(re.search(r"(\d+)", p).group(1)) * 3600
return total_seconds < 180
except Exception:
return False
def kick_peer(public_key):
"""Remove and re-add peer to force disconnect."""
try:
out = subprocess.check_output(["wg", "show", "wg0"], text=True, timeout=5)
allowed = ""
found = False
for line in out.split("\n"):
if line.strip().startswith("peer:") and public_key in line:
found = True
elif found and "allowed ips:" in line:
allowed = line.split("allowed ips:")[1].strip()
break
subprocess.run(["wg", "set", "wg0", "peer", public_key, "remove"], timeout=5)
if allowed:
subprocess.run(["wg", "set", "wg0", "peer", public_key, "allowed-ips", allowed], timeout=5)
return True
except Exception:
return False
def check_expiry():
"""Check subscription expiry, lock expired clients."""
s = load_settings()
subs = s.get("subscriptions", {})
today = datetime.now().date()
changed = False
for client_id, info in list(subs.items()):
if not info.get("expiry"):
continue
try:
exp_date = datetime.strptime(info["expiry"], "%Y-%m-%d").date()
except ValueError:
continue
days_left = (exp_date - today).days
if days_left <= 0 and not info.get("locked"):
info["locked"] = True
changed = True
kick_client_by_id(client_id)
tg_send(f"⛔ {client_id} — подписка истекла! Клиент заблокирован.")
elif days_left == 3 and not info.get("warn3"):
info["warn3"] = True
changed = True
tg_send(f"⚠️ {client_id} — подписка истекает через 3 дня ({info['expiry']})")
elif days_left == 1 and not info.get("warn1"):
info["warn1"] = True
changed = True
tg_send(f"⚠️ {client_id} — подписка истекает ЗАВТРА ({info['expiry']})")
if changed:
save_settings(s)
def kick_client_by_id(client_id):
"""Find client's public key and kick them."""
clients = get_clients()
for c in clients:
if c.get("client_id") == client_id:
kick_peer(c.get("public_key", ""))
return True
return False
prev_session_keys = None
def session_monitor():
"""Background thread: monitor sessions, send Telegram alerts."""
global prev_session_keys
while True:
try:
s = load_settings()
interval = s.get("monitor_interval", 30)
sessions = get_active_sessions()
current_keys = set()
for sess in sessions:
if is_peer_online(sess.get("handshake", "")):
key = (sess["client_id"], sess["real_ip"])
current_keys.add(key)
if prev_session_keys is not None:
labels = s.get("labels", {})
for key in current_keys - prev_session_keys:
client_id, real_ip = key
label = labels.get(real_ip, "")
name = f"{label} ({real_ip})" if label else real_ip
tg_send(f"🟢 {client_id} подключился — {name}")
for key in prev_session_keys - current_keys:
client_id, real_ip = key
label = labels.get(real_ip, "")
name = f"{label} ({real_ip})" if label else real_ip
tg_send(f"🔴 {client_id} отключился — {name}")
prev_session_keys = current_keys
check_expiry()
time.sleep(interval)
except Exception:
time.sleep(30)
monitor_thread = threading.Thread(target=session_monitor, daemon=True)
monitor_thread.start()
PAGE = """
Phobos VPN Panel
CONTENT
"""
def render(content):
return PAGE.replace("CONTENT", content)
@app.route("/login", methods=["GET", "POST"])
def login():
s = load_settings()
msg = ""
if request.method == "POST":
if request.form.get("password") == s["admin_pass"]:
session["auth"] = True
return redirect(url_for("dashboard"))
msg = 'Неверный пароль
'
html = f"""
Вход в панель
{msg}
"""
return render(html)
@app.route("/logout")
def logout():
session.clear()
return redirect(url_for("login"))
def auth_required(f):
from functools import wraps
@wraps(f)
def decorated(*args, **kwargs):
if not session.get("auth"):
return redirect(url_for("login"))
return f(*args, **kwargs)
return decorated
@app.route("/")
@auth_required
def dashboard():
return redirect(url_for("sessions_page"))
@app.route("/sessions")
@auth_required
def sessions_page():
s = load_settings()
labels = s.get("labels", {})
sessions = get_active_sessions()
rows = ""
online_count = 0
for sess in sessions:
online = is_peer_online(sess.get("handshake", ""))
if online:
online_count += 1
status = 'Online' if online else 'Offline'
real_ip = sess["real_ip"]
label = labels.get(real_ip, "")
label_display = f"{label} " if label else ""
rows += f"""
| {sess['client_id']} |
{sess['tunnel_ip']} |
{label_display}{real_ip} |
{sess['handshake']} |
{sess['rx']} / {sess['tx']} |
{status} |
Kick |
"""
if not rows:
rows = '| Нет активных сессий |
'
html = f"""
Активные сессии
| Клиент | VPN IP | Real IP | Handshake | RX / TX | Статус | |
|---|
{rows}
"""
return render(html)
@app.route("/kick/")
@auth_required
def kick(pub_key):
kick_peer(pub_key)
return redirect(url_for("sessions_page"))
@app.route("/clients", methods=["GET", "POST"])
@auth_required
def clients_page():
s = load_settings()
subs = s.get("subscriptions", {})
msg = ""
if request.method == "POST":
action = request.form.get("action")
if action == "add":
name = request.form.get("name", "").strip()
if name and re.match(r"^[a-zA-Z0-9_-]+$", name):
try:
out = subprocess.check_output(
[f"{PHOBOS_DIR}/repo/server/scripts/phobos-client.sh", "add", name],
text=True, timeout=30, stderr=subprocess.STDOUT,
env={**os.environ, **_load_server_env()}
)
msg = f'Клиент {name} создан
'
except subprocess.CalledProcessError as e:
msg = f'{e.output}
'
else:
msg = 'Имя: буквы, цифры, _ и -
'
elif action == "delete":
client_id = request.form.get("client_id", "").strip()
if client_id:
try:
out = subprocess.check_output(
[f"{PHOBOS_DIR}/repo/server/scripts/phobos-client.sh", "remove", client_id],
text=True, timeout=15, stderr=subprocess.STDOUT,
env={**os.environ, **_load_server_env()}
)
msg = f'Клиент {client_id} удалён
'
except subprocess.CalledProcessError as e:
msg = f'{e.output}
'
elif action == "set_expiry":
client_id = request.form.get("client_id", "").strip()
expiry = request.form.get("expiry", "").strip()
if client_id:
if client_id not in subs:
subs[client_id] = {}
subs[client_id]["expiry"] = expiry
subs[client_id].pop("locked", None)
subs[client_id].pop("warn3", None)
subs[client_id].pop("warn1", None)
save_settings(s)
msg = f'Срок для {client_id} обновлён
'
clients = get_clients()
peers = get_wg_peers()
online_pubs = set()
for pub, info in peers.items():
if is_peer_online(info.get("latest handshake", "")):
online_pubs.add(pub)
rows = ""
today = datetime.now().date()
for c in clients:
cid = c.get("client_id", "")
pub = c.get("public_key", "")
ip = c.get("tunnel_ip_v4", "")
created = c.get("created_at", "")[:10]
is_online = pub in online_pubs
status = 'Online' if is_online else 'Offline'
sub = subs.get(cid, {})
expiry = sub.get("expiry", "")
locked = sub.get("locked", False)
expiry_badge = ""
if expiry:
try:
exp_date = datetime.strptime(expiry, "%Y-%m-%d").date()
days = (exp_date - today).days
if locked:
expiry_badge = f'⛔ истёк'
elif days <= 1:
expiry_badge = f'⚠️ {days}д'
elif days <= 3:
expiry_badge = f'{days}д'
else:
expiry_badge = f'{days}д'
except ValueError:
pass
rows += f"""
| {cid} |
{ip} |
{created} |
{status} |
|
|
"""
html = f"""
{msg}
Клиенты VPN
| Клиент | VPN IP | Создан | Статус | Подписка | |
|---|
{rows}
"""
return render(html)
@app.route("/labels", methods=["GET", "POST"])
@auth_required
def labels_page():
s = load_settings()
msg = ""
if request.method == "POST":
action = request.form.get("action")
if action == "add":
ip = request.form.get("ip", "").strip()
label = request.form.get("label", "").strip()
if ip and label:
s["labels"][ip] = label
save_settings(s)
msg = f'Метка добавлена: {ip} → {label}
'
elif action == "delete":
ip = request.form.get("ip", "").strip()
s["labels"].pop(ip, None)
save_settings(s)
msg = f'Метка удалена
'
labels = s.get("labels", {})
rows = ""
for ip, label in sorted(labels.items()):
rows += f"""
| {ip} | {label} |
|
"""
html = f"""
{msg}
Метки по Real IP
Привяжите понятное имя (квартира, офис, дача) к внешнему IP адресу роутера. Метка отображается в таблице сессий и в Telegram уведомлениях вместо голого IP.
"""
return render(html)
@app.route("/settings", methods=["GET", "POST"])
@auth_required
def settings_page():
s = load_settings()
msg = ""
if request.method == "POST":
new_pass = request.form.get("admin_pass", "").strip()
if new_pass:
s["admin_pass"] = new_pass
s["tg_bot_token"] = request.form.get("tg_bot_token", "").strip()
s["tg_chat_id"] = request.form.get("tg_chat_id", "").strip()
try:
s["monitor_interval"] = max(10, int(request.form.get("monitor_interval", 30)))
except ValueError:
pass
save_settings(s)
msg = 'Настройки сохранены
'
html = f"""
{msg}
Информация о сервере
| VPS IP | {SERVER_IP} |
| WireGuard порт | 51820 (localhost) |
| Обфускатор порты | {_load_server_env().get('OBFUSCATOR_PORTS', '2083,5443,993')} |
| Панель порт | 8443 |
| Phobos клиенты | {CLIENTS_DIR} |
Установка на роутер
Keenetic/Netcraze с Entware — выполнить по SSH на роутере:
{_render_install_commands()}
"""
return render(html)
TOKENS_FILE = f"{PHOBOS_DIR}/tokens/tokens.json"
def _render_install_commands():
"""Read tokens and render install commands for each client."""
try:
if os.path.exists(TOKENS_FILE):
with open(TOKENS_FILE) as f:
tokens = json.load(f)
else:
tokens = []
except Exception:
tokens = []
if not tokens:
return 'Нет активных токенов. Добавьте клиента.
'
lines = ""
for t in tokens:
client = t.get("client", "?")
token = t.get("token", "")
expires = t.get("expires", 0)
exp_str = datetime.fromtimestamp(expires).strftime("%Y-%m-%d %H:%M") if expires else "?"
cmd = f"wget -O - http://{SERVER_IP}/init/{token}.sh | sh"
lines += f"""
{client}
(до {exp_str})
{cmd}
"""
return lines
SERVERS_FILE = f"{PANEL_DIR}/servers.json"
def load_servers():
if os.path.exists(SERVERS_FILE):
with open(SERVERS_FILE) as f:
return json.load(f)
return []
def save_servers(servers):
with open(SERVERS_FILE, "w") as f:
json.dump(servers, f, indent=2)
def check_server_health(server):
"""Check secondary server health via API."""
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=5)
return json.loads(resp.read())
except Exception as e:
return {"status": "error", "error": str(e)}
def sync_peer_to_server(server, public_key, allowed_ips, action="add"):
"""Add or remove peer on secondary server."""
try:
import urllib.request
url = f"http://{server['ip']}:8444/api/peers/{action}"
data = json.dumps({"public_key": public_key, "allowed_ips": allowed_ips}).encode()
req = urllib.request.Request(url, data=data, headers={
"Content-Type": "application/json",
"X-API-Key": server.get("api_key", "")
})
resp = urllib.request.urlopen(req, timeout=10)
return json.loads(resp.read())
except Exception:
return {"status": "error"}
def sync_peer_to_all_servers(public_key, allowed_ips, action="add"):
"""Sync peer to all secondary servers."""
servers = load_servers()
for srv in servers:
if srv.get("enabled", True):
sync_peer_to_server(srv, public_key, allowed_ips, action)
@app.route("/api/servers/register", methods=["POST"])
def api_register_server():
s = load_settings()
api_key = request.headers.get("X-API-Key", "")
if api_key != s.get("server_api_key", ""):
return json.dumps({"error": "unauthorized"}), 401, {"Content-Type": "application/json"}
data = request.json
servers = load_servers()
ip = data.get("ip", "")
for srv in servers:
if srv["ip"] == ip:
srv.update(data)
srv["last_seen"] = datetime.now().isoformat()
save_servers(servers)
return json.dumps({"status": "updated"}), 200, {"Content-Type": "application/json"}
data["last_seen"] = datetime.now().isoformat()
data["enabled"] = True
data["api_key"] = api_key
servers.append(data)
save_servers(servers)
return json.dumps({"status": "registered"}), 200, {"Content-Type": "application/json"}
@app.route("/servers", methods=["GET", "POST"])
@auth_required
def servers_page():
s = load_settings()
servers = load_servers()
msg = ""
if request.method == "POST":
action = request.form.get("action")
if action == "add":
ip = request.form.get("ip", "").strip()
api_key = request.form.get("api_key", "").strip()
if ip:
servers.append({
"ip": ip,
"api_key": api_key,
"enabled": True,
"last_seen": "",
"wg_public_key": "",
"obfuscator_key": "",
"ports": ""
})
save_servers(servers)
msg = f'Сервер {ip} добавлен
'
elif action == "remove":
ip = request.form.get("ip", "").strip()
servers = [s for s in servers if s.get("ip") != ip]
save_servers(servers)
msg = f'Сервер удалён
'
elif action == "sync":
ip = request.form.get("ip", "").strip()
srv = next((s for s in servers if s["ip"] == ip), None)
if srv:
clients = get_clients()
synced = 0
for c in clients:
pub = c.get("public_key", "")
tip = c.get("tunnel_ip_v4", "")
if pub and tip:
res = sync_peer_to_server(srv, pub, f"{tip}/32", "add")
if res.get("status") == "ok":
synced += 1
msg = f'Синхронизировано {synced} клиентов на {ip}
'
elif action == "fetch_info":
ip = request.form.get("ip", "").strip()
for srv in servers:
if srv["ip"] == ip:
try:
import urllib.request
url = f"http://{ip}:8444/api/info"
req = urllib.request.Request(url, headers={"X-API-Key": srv.get("api_key", "")})
resp = urllib.request.urlopen(req, timeout=5)
info = json.loads(resp.read())
srv["wg_public_key"] = info.get("wg_public_key", "")
srv["obfuscator_key"] = info.get("obfuscator_key", "")
srv["ports"] = ",".join(info.get("ports", []))
save_servers(servers)
msg = f'Инфо получено от {ip}
'
except Exception as e:
msg = f'Ошибка: {e}
'
elif action == "set_api_key":
new_key = request.form.get("server_api_key", "").strip()
if new_key:
s["server_api_key"] = new_key
save_settings(s)
msg = 'API ключ обновлён
'
# Build server rows
rows = ""
for srv in servers:
ip = srv.get("ip", "")
health = check_server_health(srv)
status_class = "badge-on" if health.get("status") == "ok" else "badge-off"
status_text = "Online" if health.get("status") == "ok" else "Offline"
peers = health.get("peers", "?")
ports = srv.get("ports", "?")
rows += f"""
| {ip} |
{ports} |
{peers} |
{status_text} |
|
"""
api_key = s.get("server_api_key", "")
html = f"""
{msg}
API ключ для secondary серверов
Единый ключ аутентификации для связи с резервными серверами. Укажите его при установке secondary сервера.
Серверы VPN
Резервные VPN серверы для автоматического переключения. Роутеры с health monitor автоматически переходят на backup если основной сервер недоступен. Кнопка Sync — синхронизирует всех клиентов на выбранный сервер, Info — запрашивает данные сервера.
| IP | Порты | Peers | Статус | |
|---|
{rows}
Установка secondary сервера
Одна команда разворачивает WireGuard + обфускатор + мини-API на новом VPS. Сервер автоматически зарегистрируется в панели. Выполнить по SSH на новом VPS:
MAIN_SERVER={SERVER_IP} MAIN_API_KEY={api_key} bash <(curl -fsSL https://raw.githubusercontent.com/andrey271192/PCA_Phobos/main/server/secondary-setup.sh)
"""
return render(html)
def _load_server_env():
"""Load server.env as dict for subprocess env."""
env = {}
if os.path.exists(SERVER_ENV):
with open(SERVER_ENV) as f:
for line in f:
line = line.strip()
if "=" in line and not line.startswith("#"):
k, v = line.split("=", 1)
env[k] = v
return env
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8443, debug=False)