+
+
+
+
+
+ ⭐ {{ _('promo_star_cta') }}
+
+ AIVPN
+ {{ _('aivpn_subtitle') }}
+ NGINX is running
+{domain}
+{_t('share_not_found_desc', lang)}
", status_code=404) + + auth_session_key = f'share_auth_{token}' + need_password = bool(user.get('share_password_hash')) and not request.session.get(auth_session_key) + + return tpl(request, 'user_share.html', + share_user=user, + need_password=need_password, + token=token) + + +@app.post('/api/share/{token}/auth', tags=["Sharing"]) +async def api_share_auth(token: str, req: ShareAuthRequest, request: Request): + data = load_data() + user = next((u for u in data['users'] if u.get('share_token') == token), None) + if not user or not user.get('share_enabled'): + return JSONResponse({'error': 'Link expired or disabled'}, status_code=404) + + if verify_password(req.password, user.get('share_password_hash', '')): + request.session[f'share_auth_{token}'] = True + return {'status': 'success'} + else: + lang = request.cookies.get('lang', 'ru') + return JSONResponse({'error': _t('wrong_share_password', lang)}, status_code=401) + + +@app.get('/api/share/{token}/connections', tags=["Sharing"]) +async def api_share_connections(token: str, request: Request): + data = load_data() + user = next((u for u in data['users'] if u.get('share_token') == token), None) + if not user or not user.get('share_enabled'): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + + if user.get('share_password_hash'): + if not request.session.get(f'share_auth_{token}'): + return JSONResponse({'error': 'Unauthorized'}, status_code=401) + + conns = [dict(c) for c in data.get('user_connections', []) if c['user_id'] == user['id']] + for c in conns: + sid = c['server_id'] + if sid < len(data['servers']): + c['server_name'] = data['servers'][sid].get('name') or data['servers'][sid]['host'] + else: + c['server_name'] = 'Unknown' + + return {'connections': conns, 'username': user['username']} + + +@app.post('/api/share/{token}/config/{connection_id}', tags=["Sharing"]) +async def api_share_config(token: str, connection_id: str, request: Request): + data = load_data() + user = next((u for u in data['users'] if u.get('share_token') == token), None) + if not user or not user.get('share_enabled'): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + + if user.get('share_password_hash'): + if not request.session.get(f'share_auth_{token}'): + return JSONResponse({'error': 'Unauthorized'}, status_code=401) + + conn = next((c for c in data.get('user_connections', []) if c['id'] == connection_id and c['user_id'] == user['id']), None) + if not conn: + return JSONResponse({'error': 'Not found'}, status_code=404) + + try: + sid = conn['server_id'] + server = data['servers'][sid] + proto_info = server.get('protocols', {}).get(conn['protocol'], {}) + port = proto_info.get('port', '55424') + ssh = get_ssh(server) + ssh.connect() + # Use appropriate manager for the protocol + manager = get_protocol_manager(ssh, conn['protocol']) + config = _manager_call(manager, 'get_client_config', conn['protocol'], conn['client_id'], server['host'], port) + ssh.disconnect() + vpn_link = generate_vpn_link(config) if config else '' + return {'config': config, 'vpn_link': vpn_link} + except Exception as e: + logger.exception("Error getting shared config") + return JSONResponse({'error': str(e)}, status_code=500) + + +@app.post('/api/my/connections/{connection_id}/config', tags=["Self-service"]) +async def api_my_connection_config(request: Request, connection_id: str): + user = get_current_user(request) + if not user: + return JSONResponse({'error': 'Forbidden'}, status_code=403) + try: + data = load_data() + conn = next( + (c for c in data.get('user_connections', []) if c['id'] == connection_id and c['user_id'] == user['id']), + None + ) + if not conn: + return JSONResponse({'error': 'Connection not found'}, status_code=404) + sid = conn['server_id'] + if sid >= len(data['servers']): + return JSONResponse({'error': 'Server not found'}, status_code=404) + server = data['servers'][sid] + proto_info = server.get('protocols', {}).get(conn['protocol'], {}) + port = proto_info.get('port', '55424') + ssh = get_ssh(server) + ssh.connect() + # Use appropriate manager for the protocol (fixes Telemt/Xray not working for users) + manager = get_protocol_manager(ssh, conn['protocol']) + config = _manager_call(manager, 'get_client_config', conn['protocol'], conn['client_id'], server['host'], port) + ssh.disconnect() + vpn_link = generate_vpn_link(config) if config else '' + return {'config': config, 'vpn_link': vpn_link} + except Exception as e: + logger.exception("Error getting my connection config") + return JSONResponse({'error': str(e)}, status_code=500) + + +@app.get('/settings', tags=["System Templates"]) +async def settings_page(request: Request): + user = _check_admin(request) + if not user: + return RedirectResponse('/login') + data = load_data() + return tpl(request, 'settings.html', settings=data.get('settings', {}), servers=data.get('servers', []), current_version=CURRENT_VERSION) + + +@app.get('/api/settings', tags=["Settings"]) +async def api_get_settings(request: Request): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + data = load_data() + return data.get('settings', {}) + + +@app.get('/api/settings/tunnels/status', tags=["Settings"]) +async def api_tunnels_status(request: Request): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + return { + 'local_server': get_panel_local_url(request), + 'cloudflare': get_tunnel_status('cloudflare'), + 'ngrok': get_tunnel_status('ngrok'), + 'warp': get_warp_status(), + } + + +@app.post('/api/settings/tunnels/{provider}/install', tags=["Settings"]) +async def api_tunnel_install(request: Request, provider: str): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + if provider not in TUNNEL_RUNTIMES: + return JSONResponse({'error': 'Unsupported tunnel provider'}, status_code=400) + try: + await asyncio.to_thread(install_tunnel_binary, provider) + return get_tunnel_status(provider) + except Exception as e: + logger.exception(f"Error installing {provider} tunnel") + return JSONResponse({'error': str(e)}, status_code=500) + + +@app.post('/api/settings/tunnels/{provider}/start', tags=["Settings"]) +async def api_tunnel_start(request: Request, provider: str, payload: TunnelStartRequest = TunnelStartRequest()): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + if provider not in TUNNEL_RUNTIMES: + return JSONResponse({'error': 'Unsupported tunnel provider'}, status_code=400) + try: + local_url = get_panel_tunnel_target_url() + await asyncio.to_thread(start_tunnel, provider, local_url, payload.authtoken or '') + status = await wait_for_tunnel_url(provider) + if not status.get('running'): + return JSONResponse({'error': status.get('last_error') or 'Tunnel process stopped'}, status_code=500) + return status + except Exception as e: + logger.exception(f"Error starting {provider} tunnel") + return JSONResponse({'error': str(e)}, status_code=500) + + +@app.post('/api/settings/tunnels/{provider}/stop', tags=["Settings"]) +async def api_tunnel_stop(request: Request, provider: str): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + if provider not in TUNNEL_RUNTIMES: + return JSONResponse({'error': 'Unsupported tunnel provider'}, status_code=400) + try: + await asyncio.to_thread(stop_tunnel, provider) + return get_tunnel_status(provider) + except Exception as e: + logger.exception(f"Error stopping {provider} tunnel") + return JSONResponse({'error': str(e)}, status_code=500) + + +@app.delete('/api/settings/tunnels/{provider}', tags=["Settings"]) +async def api_tunnel_delete(request: Request, provider: str): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + if provider not in TUNNEL_RUNTIMES: + return JSONResponse({'error': 'Unsupported tunnel provider'}, status_code=400) + try: + await asyncio.to_thread(delete_tunnel_binary, provider) + return get_tunnel_status(provider) + except Exception as e: + logger.exception(f"Error deleting {provider} tunnel") + return JSONResponse({'error': str(e)}, status_code=500) + + +@app.post('/api/settings/warp/connect', tags=["Settings"]) +async def api_warp_connect(request: Request): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + try: + return await asyncio.to_thread(enable_warp) + except Exception as e: + logger.exception("Error connecting Cloudflare WARP") + return JSONResponse({'error': str(e)}, status_code=500) + + +@app.post('/api/settings/warp/disconnect', tags=["Settings"]) +async def api_warp_disconnect(request: Request): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + try: + return await asyncio.to_thread(disable_warp) + except Exception as e: + logger.exception("Error disconnecting Cloudflare WARP") + return JSONResponse({'error': str(e)}, status_code=500) + + +# @app.post('/api/settings/save') +# async def api_save_settings(request: Request, body: SaveSettingsRequest): +# _check_admin(request) +# data = load_data() +# data['settings'] = body.dict() +# save_data(data) + +# # Trigger sync if enabled +# if body.sync.remnawave_sync_users: +# await sync_users_with_remnawave(data) +# save_data(data) + +# return {'status': 'success'} + +@app.post('/api/settings/save', tags=["Settings"]) +async def save_settings(request: Request, payload: SaveSettingsRequest): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + data = load_data() + data['settings']['appearance'] = payload.appearance.dict() + data['settings']['sync'] = payload.sync.dict() + data['settings']['captcha'] = payload.captcha.dict() + data['settings']['telegram'] = payload.telegram.dict() + data['settings']['ssl'] = payload.ssl.dict() + save_data(data) + logger.info("Settings saved (including captcha and telegram)") + + # Handle bot start/stop based on new telegram settings + tg_cfg = payload.telegram + if tg_cfg.enabled and tg_cfg.token: + if not tg_bot.is_running(): + logger.info("Starting Telegram bot (settings save)...") + tg_bot.launch_bot(tg_cfg.token, load_data, generate_vpn_link, save_data) + else: + if tg_bot.is_running(): + logger.info("Stopping Telegram bot (settings save)...") + asyncio.create_task(tg_bot.stop_bot()) + + return {"status": "success", "bot_running": tg_bot.is_running()} + + +@app.post('/api/settings/telegram/toggle', tags=["Settings"]) +async def api_telegram_toggle(request: Request): + """Quick enable/disable of the bot without a full settings save.""" + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + data = load_data() + tg_cfg = data.get('settings', {}).get('telegram', {}) + token = tg_cfg.get('token', '') + if not token: + return JSONResponse({'error': 'Telegram token not set in settings'}, status_code=400) + + if tg_bot.is_running(): + await tg_bot.stop_bot() + tg_cfg['enabled'] = False + data['settings']['telegram'] = tg_cfg + save_data(data) + return {'status': 'stopped', 'bot_running': False} + else: + tg_bot.launch_bot(token, load_data, generate_vpn_link, save_data) + tg_cfg['enabled'] = True + data['settings']['telegram'] = tg_cfg + save_data(data) + return {'status': 'started', 'bot_running': True} + +@app.post('/api/settings/sync_now', tags=["Settings"]) +async def api_sync_now(request: Request): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + data = load_data() + count, msg = await sync_users_with_remnawave(data) + return {'status': 'success', 'count': count, 'message': msg} + + +@app.post('/api/settings/sync_delete', tags=["Settings"]) +async def api_sync_delete(request: Request): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + data = load_data() + to_delete_ids = [u['id'] for u in data['users'] if u.get('remnawave_uuid')] + if to_delete_ids: + await perform_mass_operations(delete_uids=to_delete_ids) + return {'status': 'success', 'count': len(to_delete_ids)} + + +@app.get('/api/servers/{server_id}/{protocol}/clients', tags=["Connections"]) +async def api_get_server_clients(request: Request, server_id: int, protocol: str): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + try: + data = load_data() + if server_id >= len(data['servers']): + return JSONResponse({'error': 'Server not found'}, status_code=404) + server = data['servers'][server_id] + ssh = get_ssh(server) + ssh.connect() + manager = get_protocol_manager(ssh, protocol) + clients = manager.get_clients(protocol) + ssh.disconnect() + + # Filter: only show clients that are not assigned to anyone in the panel + assigned_ids = {c['client_id'] for c in data.get('user_connections', []) if c['server_id'] == server_id and c['protocol'] == protocol} + + filtered = [] + for c in clients: + if c['clientId'] not in assigned_ids: + filtered.append({ + 'id': c['clientId'], + 'name': c.get('userData', {}).get('clientName', 'Unnamed') + }) + + return {'clients': filtered} + except Exception as e: + logger.exception("Error getting server clients") + return JSONResponse({'error': str(e)}, status_code=500) + + +@app.get('/api/settings/tokens', tags=["API Tokens"]) +async def api_list_tokens(request: Request): + """List metadata for every API token. The raw token value is never + returned by this endpoint — only its prefix and timestamps are visible + after creation, by design.""" + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + data = load_data() + users_by_id = {u['id']: u for u in data.get('users', [])} + tokens = [] + for t in data.get('api_tokens', []): + owner = users_by_id.get(t.get('user_id')) + tokens.append({ + 'id': t.get('id'), + 'name': t.get('name', ''), + 'token_prefix': t.get('token_prefix', ''), + 'created_at': t.get('created_at'), + 'last_used_at': t.get('last_used_at'), + 'owner': owner['username'] if owner else None, + 'owner_id': t.get('user_id'), + }) + return {'tokens': tokens} + + +@app.post('/api/settings/tokens', tags=["API Tokens"]) +async def api_create_token(request: Request, req: CreateApiTokenRequest): + """Issue a new bearer token. The full token value is returned **once** in + the response and never persisted in plaintext — only its SHA-256 hash is + stored, so a leaked data.json file alone cannot be used to authenticate. + Save the value at creation time; if it's lost the token must be recreated. + """ + cur = _check_admin(request) + if not cur: + return JSONResponse({'error': 'Forbidden'}, status_code=403) + name = (req.name or '').strip() + if not name: + return JSONResponse({'error': 'Token name is required'}, status_code=400) + + raw = _generate_api_token() + token_id = str(uuid.uuid4()) + # Show enough of the token in the UI to identify it later, but not enough + # to reconstruct it: the prefix + first 4 chars of the secret part. + token_prefix = raw[:len(API_TOKEN_PREFIX) + 4] + + entry = { + 'id': token_id, + 'name': name, + 'token_hash': _hash_api_token(raw), + 'token_prefix': token_prefix, + 'user_id': cur['id'], + 'created_at': datetime.now().isoformat(), + 'last_used_at': None, + } + async with DATA_LOCK: + data = load_data() + data.setdefault('api_tokens', []).append(entry) + save_data(data) + + # `token` is returned only here — subsequent reads will not see it. + return { + 'status': 'success', + 'id': token_id, + 'name': name, + 'token': raw, + 'token_prefix': token_prefix, + 'created_at': entry['created_at'], + } + + +@app.delete('/api/settings/tokens/{token_id}', tags=["API Tokens"]) +async def api_revoke_token(request: Request, token_id: str): + """Permanently revoke a token. The associated bearer value can never be + used again, even if the same name is reissued — every token has its own hash.""" + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + async with DATA_LOCK: + data = load_data() + before = len(data.get('api_tokens', [])) + data['api_tokens'] = [t for t in data.get('api_tokens', []) if t.get('id') != token_id] + if len(data['api_tokens']) == before: + return JSONResponse({'error': 'Token not found'}, status_code=404) + save_data(data) + return {'status': 'success'} + + +@app.get('/api/settings/backup/download', tags=["Settings"]) +async def api_backup_download(request: Request): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + payload = await asyncio.to_thread(export_data_dict) + content = json.dumps(payload, indent=2, ensure_ascii=False).encode('utf-8') + return StreamingResponse( + io.BytesIO(content), + media_type='application/json', + headers={'Content-Disposition': 'attachment; filename="data.json"'}, + ) + + +@app.post('/api/settings/backup/restore', tags=["Settings"]) +async def api_backup_restore(request: Request, file: UploadFile = File(...)): + if not _check_admin(request): + return JSONResponse({'error': 'Forbidden'}, status_code=403) + try: + content = await file.read() + if not content: + return JSONResponse({'error': 'Empty file'}, status_code=400) + + try: + backup_data = json.loads(content) + except json.JSONDecodeError: + return JSONResponse({'error': 'Invalid JSON format'}, status_code=400) + + # Basic structure validation + required_keys = ['servers', 'users'] + missing = [k for k in required_keys if k not in backup_data] + if missing: + return JSONResponse({'error': f'Invalid structure. Missing keys: {", ".join(missing)}'}, status_code=400) + + # Ensure types are correct + if not isinstance(backup_data['servers'], list) or not isinstance(backup_data['users'], list): + return JSONResponse({'error': 'Invalid structure: servers and users must be lists'}, status_code=400) + + # Save the new data + async with DATA_LOCK: + save_data(backup_data) + + # In a real app we might want to restart or re-init background tasks + return {'status': 'success'} + except Exception as e: + logger.exception("Error during restore") + return JSONResponse({'error': str(e)}, status_code=500) + + +if __name__ == '__main__': + data = load_data() + ssl_conf = data.get('settings', {}).get('ssl', {}) + + cert_file = ssl_conf.get('cert_path') + key_file = ssl_conf.get('key_path') + + # If text is provided, create temporary files + temp_dir = os.path.join(os.getcwd(), 'ssl_temp') + if ssl_conf.get('enabled'): + if ssl_conf.get('cert_text') or ssl_conf.get('key_text'): + if not os.path.exists(temp_dir): + os.makedirs(temp_dir) + + if ssl_conf.get('cert_text'): + cert_file = os.path.join(temp_dir, 'cert.pem') + with open(cert_file, 'w') as f: + f.write(ssl_conf['cert_text'].strip() + '\n') + + if ssl_conf.get('key_text'): + key_file = os.path.join(temp_dir, 'key.pem') + with open(key_file, 'w') as f: + f.write(ssl_conf['key_text'].strip() + '\n') + + uvicorn_kwargs = { + "app": app, + "host": "0.0.0.0", + "port": ssl_conf.get('panel_port', 5000) + } + + if ssl_conf.get('enabled') and cert_file and key_file: + if os.path.exists(cert_file) and os.path.exists(key_file): + logger.info(f"Starting panel with HTTPS enabled on domain: {ssl_conf.get('domain')} at port {uvicorn_kwargs['port']}") + uvicorn_kwargs["ssl_certfile"] = cert_file + uvicorn_kwargs["ssl_keyfile"] = key_file + else: + logger.error("SSL certificates not found at specified paths. Starting with HTTP.") + + uvicorn.run(**uvicorn_kwargs) diff --git a/db/__init__.py b/db/__init__.py new file mode 100644 index 0000000..d237706 --- /dev/null +++ b/db/__init__.py @@ -0,0 +1,29 @@ +"""Database package for Amnezia Web Panel (PostgreSQL 17).""" + +from .connection import close_pool, get_database_url, init_schema +from .store import ( + clear_tunnel_state, + ensure_db_ready, + export_data_dict, + import_from_json_file, + load_data, + load_tunnel_state, + save_data, + save_tunnel_state, + update_tunnel_state, +) + +__all__ = [ + 'clear_tunnel_state', + 'close_pool', + 'ensure_db_ready', + 'export_data_dict', + 'get_database_url', + 'import_from_json_file', + 'init_schema', + 'load_data', + 'load_tunnel_state', + 'save_data', + 'save_tunnel_state', + 'update_tunnel_state', +] diff --git a/db/connection.py b/db/connection.py new file mode 100644 index 0000000..a53ddc5 --- /dev/null +++ b/db/connection.py @@ -0,0 +1,82 @@ +"""PostgreSQL connection pool for Amnezia Web Panel.""" + +from __future__ import annotations + +import logging +import os +import threading +from pathlib import Path + +from dotenv import load_dotenv + +load_dotenv() + +logger = logging.getLogger(__name__) + +DEFAULT_DATABASE_URL = 'postgresql://amnezia:amnezia@localhost:5432/amnezia_panel' + +_pool = None +_pool_lock = threading.Lock() +_schema_ready = False + + +def get_database_url() -> str: + return os.environ.get('DATABASE_URL', DEFAULT_DATABASE_URL).strip() + + +def get_pool(): + global _pool + if _pool is not None: + return _pool + with _pool_lock: + if _pool is not None: + return _pool + from psycopg.rows import dict_row + from psycopg_pool import ConnectionPool + + url = get_database_url() + logger.info('Connecting to PostgreSQL…') + _pool = ConnectionPool( + conninfo=url, + min_size=1, + max_size=10, + kwargs={ + 'autocommit': False, + 'row_factory': dict_row, + }, + open=True, + ) + return _pool + + +def close_pool(): + global _pool, _schema_ready + with _pool_lock: + if _pool is not None: + _pool.close() + _pool = None + _schema_ready = False + + +def init_schema(): + """Create tables if they do not exist.""" + global _schema_ready + if _schema_ready: + return + schema_path = Path(__file__).with_name('schema.sql') + sql = schema_path.read_text(encoding='utf-8') + pool = get_pool() + with pool.connection() as conn: + with conn.cursor() as cur: + # psycopg3 executes one statement per execute() + for raw in sql.split(';'): + lines = [ + ln for ln in raw.splitlines() + if ln.strip() and not ln.strip().startswith('--') + ] + stmt = '\n'.join(lines).strip() + if stmt: + cur.execute(stmt) + conn.commit() + _schema_ready = True + logger.info('PostgreSQL schema ready') diff --git a/db/migrate_json.py b/db/migrate_json.py new file mode 100644 index 0000000..237e066 --- /dev/null +++ b/db/migrate_json.py @@ -0,0 +1,58 @@ +"""One-shot CLI: import legacy data.json into PostgreSQL. + +Usage: + python -m db.migrate_json [path/to/data.json] +""" + +from __future__ import annotations + +import logging +import os +import sys + +from dotenv import load_dotenv + +load_dotenv() + +logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s') +logger = logging.getLogger('migrate_json') + + +def main(argv: list[str] | None = None) -> int: + argv = list(argv if argv is not None else sys.argv[1:]) + force = '--force' in argv + paths = [a for a in argv if not a.startswith('-')] + + if getattr(sys, 'frozen', False): + app_path = os.path.dirname(sys.executable) + else: + app_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + + data_file = paths[0] if paths else os.path.join(app_path, 'data.json') + + from db.store import import_from_json_file, is_database_empty, load_data + + if not os.path.exists(data_file): + logger.error('File not found: %s', data_file) + return 1 + + if not is_database_empty(): + logger.warning('Database is not empty — import will overwrite panel tables.') + if not force: + logger.error('Re-run with --force to overwrite existing data.') + return 2 + + import_from_json_file(data_file, backup=True) + data = load_data() + logger.info( + 'Done. servers=%s users=%s connections=%s tokens=%s', + len(data['servers']), + len(data['users']), + len(data['user_connections']), + len(data['api_tokens']), + ) + return 0 + + +if __name__ == '__main__': + raise SystemExit(main()) diff --git a/db/schema.sql b/db/schema.sql new file mode 100644 index 0000000..d48a1a1 --- /dev/null +++ b/db/schema.sql @@ -0,0 +1,75 @@ +-- Amnezia Web Panel — PostgreSQL 17 schema + +CREATE TABLE IF NOT EXISTS servers ( + position INTEGER PRIMARY KEY, + name TEXT NOT NULL DEFAULT '', + host TEXT NOT NULL DEFAULT '', + ssh_port INTEGER NOT NULL DEFAULT 22, + username TEXT NOT NULL DEFAULT '', + password TEXT, + private_key TEXT, + server_info JSONB NOT NULL DEFAULT '{}'::jsonb, + protocols JSONB NOT NULL DEFAULT '{}'::jsonb +); + +CREATE TABLE IF NOT EXISTS users ( + id UUID PRIMARY KEY, + username TEXT NOT NULL UNIQUE, + password_hash TEXT NOT NULL DEFAULT '', + role TEXT NOT NULL DEFAULT 'user', + enabled BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMPTZ, + telegram_id TEXT, + email TEXT, + description TEXT, + traffic_limit BIGINT NOT NULL DEFAULT 0, + traffic_used BIGINT NOT NULL DEFAULT 0, + traffic_total BIGINT NOT NULL DEFAULT 0, + traffic_reset_strategy TEXT NOT NULL DEFAULT 'never', + last_reset_at TIMESTAMPTZ, + expiration_date TIMESTAMPTZ, + remnawave_uuid TEXT, + share_enabled BOOLEAN NOT NULL DEFAULT FALSE, + share_token TEXT, + share_password_hash TEXT +); + +CREATE TABLE IF NOT EXISTS user_connections ( + id UUID PRIMARY KEY, + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + server_id INTEGER NOT NULL DEFAULT 0, + protocol TEXT NOT NULL DEFAULT '', + client_id TEXT NOT NULL DEFAULT '', + name TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ, + last_bytes BIGINT NOT NULL DEFAULT 0 +); + +CREATE INDEX IF NOT EXISTS idx_user_connections_user_id ON user_connections(user_id); +CREATE INDEX IF NOT EXISTS idx_user_connections_server_id ON user_connections(server_id); + +CREATE TABLE IF NOT EXISTS api_tokens ( + id UUID PRIMARY KEY, + name TEXT NOT NULL DEFAULT '', + token_hash TEXT NOT NULL UNIQUE, + token_prefix TEXT NOT NULL DEFAULT '', + user_id UUID NOT NULL REFERENCES users(id) ON DELETE CASCADE, + created_at TIMESTAMPTZ, + last_used_at TIMESTAMPTZ +); + +CREATE INDEX IF NOT EXISTS idx_api_tokens_user_id ON api_tokens(user_id); +CREATE INDEX IF NOT EXISTS idx_api_tokens_hash ON api_tokens(token_hash); + +CREATE TABLE IF NOT EXISTS settings ( + id SMALLINT PRIMARY KEY DEFAULT 1 CHECK (id = 1), + data JSONB NOT NULL DEFAULT '{}'::jsonb +); + +CREATE TABLE IF NOT EXISTS tunnel_state ( + provider TEXT PRIMARY KEY, + data JSONB NOT NULL DEFAULT '{}'::jsonb +); + +INSERT INTO settings (id, data) VALUES (1, '{}'::jsonb) +ON CONFLICT (id) DO NOTHING; diff --git a/db/store.py b/db/store.py new file mode 100644 index 0000000..29a8304 --- /dev/null +++ b/db/store.py @@ -0,0 +1,413 @@ +"""Panel data store backed by PostgreSQL 17. + +Preserves the same dict shape that used to live in data.json so existing +FastAPI handlers keep working without a full rewrite. +""" + +from __future__ import annotations + +import json +import logging +import os +import shutil +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Optional +from uuid import UUID + +from psycopg.types.json import Jsonb + +from .connection import get_pool, init_schema + +logger = logging.getLogger(__name__) + +DEFAULT_SETTINGS = { + 'appearance': { + 'title': 'Amnezia', + 'logo': '❤️', + 'subtitle': 'Web Panel', + }, + 'sync': { + 'remnawave_url': '', + 'remnawave_api_key': '', + 'remnawave_sync': False, + 'remnawave_sync_users': False, + 'remnawave_create_conns': False, + 'remnawave_server_id': 0, + 'remnawave_protocol': 'awg', + }, +} + + +def _parse_ts(value: Any) -> Optional[datetime]: + if value is None or value == '': + return None + if isinstance(value, datetime): + return value + text = str(value).strip() + if not text: + return None + if text.endswith('Z'): + text = text[:-1] + '+00:00' + try: + return datetime.fromisoformat(text) + except ValueError: + return None + + +def _ts_iso(value: Any) -> Optional[str]: + if value is None: + return None + if isinstance(value, datetime): + return value.isoformat() + return str(value) + + +def _as_uuid(value: Any) -> UUID: + return UUID(str(value)) + + +def _merge_settings(raw: Optional[dict]) -> dict: + settings = json.loads(json.dumps(DEFAULT_SETTINGS)) + if not isinstance(raw, dict): + return settings + for section, defaults in DEFAULT_SETTINGS.items(): + incoming = raw.get(section) + if isinstance(incoming, dict) and isinstance(defaults, dict): + merged = dict(defaults) + merged.update(incoming) + settings[section] = merged + elif section in raw: + settings[section] = raw[section] + for key, value in raw.items(): + if key not in settings: + settings[key] = value + return settings + + +def _row_to_server(row) -> dict: + return { + 'name': row['name'] or '', + 'host': row['host'] or '', + 'ssh_port': int(row['ssh_port'] or 22), + 'username': row['username'] or '', + 'password': row['password'], + 'private_key': row['private_key'], + 'server_info': dict(row['server_info'] or {}), + 'protocols': dict(row['protocols'] or {}), + } + + +def _row_to_user(row) -> dict: + return { + 'id': str(row['id']), + 'username': row['username'], + 'password_hash': row['password_hash'] or '', + 'role': row['role'] or 'user', + 'enabled': bool(row['enabled']), + 'created_at': _ts_iso(row['created_at']), + 'telegramId': row['telegram_id'], + 'email': row['email'], + 'description': row['description'], + 'traffic_limit': int(row['traffic_limit'] or 0), + 'traffic_used': int(row['traffic_used'] or 0), + 'traffic_total': int(row['traffic_total'] or 0), + 'traffic_reset_strategy': row['traffic_reset_strategy'] or 'never', + 'last_reset_at': _ts_iso(row['last_reset_at']), + 'expiration_date': _ts_iso(row['expiration_date']), + 'remnawave_uuid': row['remnawave_uuid'], + 'share_enabled': bool(row['share_enabled']), + 'share_token': row['share_token'], + 'share_password_hash': row['share_password_hash'], + } + + +def _row_to_connection(row) -> dict: + return { + 'id': str(row['id']), + 'user_id': str(row['user_id']), + 'server_id': int(row['server_id'] or 0), + 'protocol': row['protocol'] or '', + 'client_id': row['client_id'] or '', + 'name': row['name'] or '', + 'created_at': _ts_iso(row['created_at']), + 'last_bytes': int(row['last_bytes'] or 0), + } + + +def _row_to_token(row) -> dict: + return { + 'id': str(row['id']), + 'name': row['name'] or '', + 'token_hash': row['token_hash'] or '', + 'token_prefix': row['token_prefix'] or '', + 'user_id': str(row['user_id']), + 'created_at': _ts_iso(row['created_at']), + 'last_used_at': _ts_iso(row['last_used_at']), + } + + +def load_data() -> dict: + init_schema() + pool = get_pool() + with pool.connection() as conn: + with conn.cursor() as cur: + cur.execute( + 'SELECT position, name, host, ssh_port, username, password, ' + 'private_key, server_info, protocols ' + 'FROM servers ORDER BY position ASC' + ) + servers = [_row_to_server(r) for r in cur.fetchall()] + + cur.execute( + 'SELECT id, username, password_hash, role, enabled, created_at, ' + 'telegram_id, email, description, traffic_limit, traffic_used, ' + 'traffic_total, traffic_reset_strategy, last_reset_at, ' + 'expiration_date, remnawave_uuid, share_enabled, share_token, ' + 'share_password_hash FROM users ORDER BY created_at NULLS LAST, username' + ) + users = [_row_to_user(r) for r in cur.fetchall()] + + cur.execute( + 'SELECT id, user_id, server_id, protocol, client_id, name, ' + 'created_at, last_bytes FROM user_connections ' + 'ORDER BY created_at NULLS LAST, id' + ) + user_connections = [_row_to_connection(r) for r in cur.fetchall()] + + cur.execute( + 'SELECT id, name, token_hash, token_prefix, user_id, ' + 'created_at, last_used_at FROM api_tokens ' + 'ORDER BY created_at NULLS LAST, id' + ) + api_tokens = [_row_to_token(r) for r in cur.fetchall()] + + cur.execute('SELECT data FROM settings WHERE id = 1') + settings_row = cur.fetchone() + settings = _merge_settings(settings_row['data'] if settings_row else None) + + return { + 'servers': servers, + 'users': users, + 'user_connections': user_connections, + 'api_tokens': api_tokens, + 'settings': settings, + } + + +def save_data(data: dict) -> None: + """Replace panel state in a single transaction (same semantics as rewriting data.json).""" + init_schema() + servers = data.get('servers') or [] + users = data.get('users') or [] + connections = data.get('user_connections') or [] + tokens = data.get('api_tokens') or [] + settings = _merge_settings(data.get('settings')) + + # Keep only connections whose user still exists + user_ids = {str(u.get('id')) for u in users if u.get('id')} + connections = [c for c in connections if str(c.get('user_id')) in user_ids] + tokens = [t for t in tokens if str(t.get('user_id')) in user_ids] + + pool = get_pool() + with pool.connection() as conn: + with conn.cursor() as cur: + # Order matters for FKs: children first on delete, parents first on insert + cur.execute('DELETE FROM api_tokens') + cur.execute('DELETE FROM user_connections') + cur.execute('DELETE FROM users') + cur.execute('DELETE FROM servers') + + for pos, server in enumerate(servers): + cur.execute( + 'INSERT INTO servers (position, name, host, ssh_port, username, ' + 'password, private_key, server_info, protocols) ' + 'VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)', + ( + pos, + server.get('name') or '', + server.get('host') or '', + int(server.get('ssh_port') or 22), + server.get('username') or '', + server.get('password'), + server.get('private_key'), + Jsonb(server.get('server_info') or {}), + Jsonb(server.get('protocols') or {}), + ), + ) + + for user in users: + cur.execute( + 'INSERT INTO users (' + 'id, username, password_hash, role, enabled, created_at, ' + 'telegram_id, email, description, traffic_limit, traffic_used, ' + 'traffic_total, traffic_reset_strategy, last_reset_at, ' + 'expiration_date, remnawave_uuid, share_enabled, share_token, ' + 'share_password_hash' + ') VALUES (' + '%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s' + ')', + ( + _as_uuid(user['id']), + user.get('username') or '', + user.get('password_hash') or '', + user.get('role') or 'user', + bool(user.get('enabled', True)), + _parse_ts(user.get('created_at')), + str(user['telegramId']) if user.get('telegramId') is not None else None, + user.get('email'), + user.get('description'), + int(user.get('traffic_limit') or 0), + int(user.get('traffic_used') or 0), + int(user.get('traffic_total') or 0), + user.get('traffic_reset_strategy') or 'never', + _parse_ts(user.get('last_reset_at')), + _parse_ts(user.get('expiration_date')), + user.get('remnawave_uuid'), + bool(user.get('share_enabled', False)), + user.get('share_token'), + user.get('share_password_hash'), + ), + ) + + for conn_row in connections: + cur.execute( + 'INSERT INTO user_connections (' + 'id, user_id, server_id, protocol, client_id, name, created_at, last_bytes' + ') VALUES (%s, %s, %s, %s, %s, %s, %s, %s)', + ( + _as_uuid(conn_row['id']), + _as_uuid(conn_row['user_id']), + int(conn_row.get('server_id') or 0), + conn_row.get('protocol') or '', + conn_row.get('client_id') or '', + conn_row.get('name') or '', + _parse_ts(conn_row.get('created_at')), + int(conn_row.get('last_bytes') or 0), + ), + ) + + for token in tokens: + cur.execute( + 'INSERT INTO api_tokens (' + 'id, name, token_hash, token_prefix, user_id, created_at, last_used_at' + ') VALUES (%s, %s, %s, %s, %s, %s, %s)', + ( + _as_uuid(token['id']), + token.get('name') or '', + token.get('token_hash') or '', + token.get('token_prefix') or '', + _as_uuid(token['user_id']), + _parse_ts(token.get('created_at')), + _parse_ts(token.get('last_used_at')), + ), + ) + + cur.execute( + 'INSERT INTO settings (id, data) VALUES (1, %s) ' + 'ON CONFLICT (id) DO UPDATE SET data = EXCLUDED.data', + (Jsonb(settings),), + ) + conn.commit() + + +def export_data_dict() -> dict: + """Export current DB state as the legacy data.json document.""" + return load_data() + + +def is_database_empty() -> bool: + init_schema() + pool = get_pool() + with pool.connection() as conn: + with conn.cursor() as cur: + cur.execute('SELECT COUNT(*) AS n FROM users') + users = cur.fetchone()['n'] + cur.execute('SELECT COUNT(*) AS n FROM servers') + servers = cur.fetchone()['n'] + return users == 0 and servers == 0 + + +def import_from_json_file(path: str | os.PathLike, *, backup: bool = True) -> bool: + """Import legacy data.json into Postgres. Returns True if import ran.""" + path = Path(path) + if not path.exists(): + return False + + with path.open('r', encoding='utf-8') as f: + data = json.load(f) + + if not isinstance(data, dict): + raise ValueError(f'Invalid data.json: expected object, got {type(data).__name__}') + + data.setdefault('servers', []) + data.setdefault('users', []) + data.setdefault('user_connections', []) + data.setdefault('api_tokens', []) + data.setdefault('settings', {}) + + save_data(data) + + if backup: + stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ') + backup_path = path.with_name(f'data.json.migrated.{stamp}.bak') + try: + shutil.copy2(path, backup_path) + logger.info('Backed up legacy data.json to %s', backup_path) + except OSError as e: + logger.warning('Could not backup data.json: %s', e) + + logger.info( + 'Imported data.json → PostgreSQL (%s servers, %s users, %s connections)', + len(data.get('servers', [])), + len(data.get('users', [])), + len(data.get('user_connections', [])), + ) + return True + + +def load_tunnel_state() -> dict: + init_schema() + pool = get_pool() + with pool.connection() as conn: + with conn.cursor() as cur: + cur.execute('SELECT provider, data FROM tunnel_state') + rows = cur.fetchall() + return {row['provider']: dict(row['data'] or {}) for row in rows} + + +def save_tunnel_state(state: dict) -> None: + init_schema() + pool = get_pool() + with pool.connection() as conn: + with conn.cursor() as cur: + cur.execute('DELETE FROM tunnel_state') + for provider, payload in (state or {}).items(): + cur.execute( + 'INSERT INTO tunnel_state (provider, data) VALUES (%s, %s)', + (str(provider), Jsonb(payload or {})), + ) + conn.commit() + + +def update_tunnel_state(provider: str, **updates) -> None: + state = load_tunnel_state() + provider_state = state.get(provider, {}) + provider_state.update(updates) + state[provider] = provider_state + save_tunnel_state(state) + + +def clear_tunnel_state(provider: str) -> None: + state = load_tunnel_state() + if provider in state: + state.pop(provider) + save_tunnel_state(state) + + +def ensure_db_ready(legacy_data_file: Optional[str] = None) -> None: + """Init schema and one-shot import from data.json when DB is empty.""" + init_schema() + if legacy_data_file and is_database_empty() and os.path.exists(legacy_data_file): + logger.info('Empty database — importing legacy %s', legacy_data_file) + import_from_json_file(legacy_data_file) diff --git a/docker-compose.warp.yml b/docker-compose.warp.yml new file mode 100644 index 0000000..62246a8 --- /dev/null +++ b/docker-compose.warp.yml @@ -0,0 +1,28 @@ +# Optional Docker Compose override for running Cloudflare WARP inside the panel container. +# Usage: +# docker compose -f docker-compose.yml -f docker-compose.warp.yml up -d --build +# +# Why the extra privileges: +# WARP needs a TUN device and NET_ADMIN to create/manage its tunnel interface. + +services: + amnezia_panel: + build: + context: . + dockerfile: Dockerfile.warp + image: amnezia-panel:warp-local + + cap_add: + - NET_ADMIN + - SYS_MODULE + devices: + - /dev/net/tun:/dev/net/tun + sysctls: + net.ipv4.conf.all.src_valid_mark: "1" + + volumes: + - amnezia_data:/app/data + - cloudflare_warp:/var/lib/cloudflare-warp + +volumes: + cloudflare_warp: diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..9eab514 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,50 @@ +version: '3.8' + +services: + db: + image: postgres:17 + container_name: amnezia_panel_db + environment: + POSTGRES_USER: ${POSTGRES_USER:-amnezia} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-amnezia} + POSTGRES_DB: ${POSTGRES_DB:-amnezia_panel} + volumes: + - amnezia_pgdata:/var/lib/postgresql/data + ports: + - "${POSTGRES_PORT:-5432}:5432" + restart: unless-stopped + healthcheck: + test: ["CMD-SHELL", "pg_isready -U ${POSTGRES_USER:-amnezia} -d ${POSTGRES_DB:-amnezia_panel}"] + interval: 10s + timeout: 5s + retries: 10 + start_period: 20s + + amnezia_panel: + build: + context: . + dockerfile: Dockerfile + image: amnezia-panel:local + container_name: amnezia_panel + depends_on: + db: + condition: service_healthy + ports: + - "${APP_PORT:-5000}:5000" + environment: + DATABASE_URL: postgresql://${POSTGRES_USER:-amnezia}:${POSTGRES_PASSWORD:-amnezia}@db:5432/${POSTGRES_DB:-amnezia_panel} + SECRET_KEY: ${SECRET_KEY:-} + APP_PORT: ${APP_PORT:-5000} + volumes: + - amnezia_data:/app/data + restart: unless-stopped + healthcheck: + test: ["CMD-SHELL", "python3 -c \"import socket; s=socket.socket(); s.connect(('localhost', 5000)); s.close()\""] + interval: 30s + timeout: 10s + retries: 3 + start_period: 40s + +volumes: + amnezia_data: + amnezia_pgdata: diff --git a/docker-entrypoint-warp.sh b/docker-entrypoint-warp.sh new file mode 100644 index 0000000..352d9a1 --- /dev/null +++ b/docker-entrypoint-warp.sh @@ -0,0 +1,22 @@ +#!/bin/sh +set -eu + +# Start Cloudflare WARP daemon inside the container when present. +# The panel controls it through warp-cli from the Settings page. +if command -v warp-svc >/dev/null 2>&1; then + mkdir -p /var/lib/cloudflare-warp + warp-svc >/tmp/warp-svc.log 2>&1 & + echo "$!" >/tmp/warp-svc.pid + + # Best-effort short wait: app can still start even if WARP service needs more time. + i=0 + while [ "$i" -lt 10 ]; do + if warp-cli --accept-tos status >/tmp/warp-cli-status.log 2>&1; then + break + fi + i=$((i + 1)) + sleep 1 + done +fi + +exec "$@" diff --git a/managers/__init__.py b/managers/__init__.py new file mode 100644 index 0000000..73891c1 --- /dev/null +++ b/managers/__init__.py @@ -0,0 +1,7 @@ +"""Protocol/service/SSH managers used by the web panel. + +Modules in this package are imported either directly (`from managers.ssh_manager import SSHManager`) +or lazily by name through `app.get_protocol_manager`. Keeping them in a dedicated package +makes the project root easier to scan and prevents accidental name collisions with the +generic stdlib (e.g. `socks5_manager`, `dns_manager`). +""" diff --git a/managers/adguard_manager.py b/managers/adguard_manager.py new file mode 100644 index 0000000..e5152ef --- /dev/null +++ b/managers/adguard_manager.py @@ -0,0 +1,264 @@ +""" +AdGuard Home Manager — runs adguard/adguardhome in a Docker container, +joined to the same internal `amnezia-dns-net` network the rest of the panel +uses. Two install modes: + + * 'replace' — removes the AmneziaDNS container (if present) and takes + its static IP (172.29.172.254) so VPN clients keep using + the same upstream. + * 'sidebyside' — runs alongside AmneziaDNS on a different static IP + (172.29.172.253). VPN users can hit it on demand + (e.g. via the web admin UI from inside the tunnel). + +Initial setup of AdGuard itself runs through its built-in wizard on the web +UI port — we do not try to script it (the JSON setup API is unstable across +versions and trying to drive it programmatically tends to break on upgrade). +""" + +import logging + +logger = logging.getLogger(__name__) + + +class AdguardManager: + PROTOCOL = 'adguard' + CONTAINER_NAME = 'amnezia-adguard' + IMAGE_NAME = 'adguard/adguardhome:latest' + + NETWORK_NAME = 'amnezia-dns-net' + NETWORK_SUBNET = '172.29.172.0/24' + REPLACE_IP = '172.29.172.254' # AmneziaDNS's slot + SIDEBYSIDE_IP = '172.29.172.253' # parallel to AmneziaDNS + + HOST_DIR = '/opt/amnezia/adguard' + DEFAULT_DNS_PORT = 53 + DEFAULT_WEB_PORT = 3000 + DEFAULT_DOT_PORT = 853 + DEFAULT_DOH_PORT = 443 + + def __init__(self, ssh): + self.ssh = ssh + + # ===================== STATUS ===================== + + def check_docker_installed(self): + out, _, code = self.ssh.run_command("docker --version 2>/dev/null") + if code != 0: + return False + out2, _, _ = self.ssh.run_command( + "systemctl is-active docker 2>/dev/null || service docker status 2>/dev/null" + ) + return 'active' in out2 or 'running' in out2.lower() + + def check_protocol_installed(self, protocol_type='adguard'): + out, _, _ = self.ssh.run_sudo_command( + f"docker ps -a --filter name=^{self.CONTAINER_NAME}$ --format '{{{{.Names}}}}'" + ) + return self.CONTAINER_NAME in out.strip().split('\n') + + def check_container_running(self): + out, _, _ = self.ssh.run_sudo_command( + f"docker ps --filter name=^{self.CONTAINER_NAME}$ --format '{{{{.Status}}}}'" + ) + return 'Up' in out + + def _container_ip(self): + out, _, _ = self.ssh.run_sudo_command( + f"docker inspect -f '{{{{range .NetworkSettings.Networks}}}}{{{{.IPAddress}}}} {{{{end}}}}' {self.CONTAINER_NAME} 2>/dev/null" + ) + ip = out.strip().split()[0] if out.strip() else '' + return ip + + def _detect_mode(self): + """Return 'replace' or 'sidebyside' based on the running container's IP. + Returns None if not detectable (container not running).""" + ip = self._container_ip() + if ip == self.REPLACE_IP: + return 'replace' + if ip == self.SIDEBYSIDE_IP: + return 'sidebyside' + return None + + def _container_web_port(self): + """Read the AdGuard HTTP address configured inside the container. + During the first-run wizard AdGuard writes web.listen_addresses to + AdGuardHome.yaml. If the wizard has not been completed yet, the + container is started with --web-addr 0.0.0.0:{domain}
+| ');g.push(" |
{_e(tg_id)}",
+ )
+ return
+
+ if _is_admin(panel_user):
+ await api.send_message(
+ chat_id,
+ f"👋 Hi, {_e(first_name)}!\n\n"
+ f"You are registered as {_e(panel_user.get('username'))} with Admin role.\n"
+ "Choose an action:",
+ reply_markup=_admin_main_keyboard(),
+ )
+ return
+
+ await _send_user_connections(api, chat_id, panel_user, load_data_fn, first_name=first_name)
+
+
+async def _send_user_connections(api: TelegramAPI, chat_id: int, panel_user: dict, load_data_fn: Callable, first_name: str = ""):
+ data = load_data_fn()
+ conns = [c for c in data.get("user_connections", []) if c.get("user_id") == panel_user.get("id")]
+
+ if not conns:
+ greeting = f"👋 Hi, {_e(first_name)}!\n\n" if first_name else ""
+ await api.send_message(
+ chat_id,
+ greeting + f"You are registered as {_e(panel_user.get('username'))}.\n\n"
+ "You have no connections yet. Please contact your administrator.",
+ )
+ return
+
+ kb = _build_connections_keyboard(conns, data)
+ greeting = f"👋 Hi, {_e(first_name)}!\n\n" if first_name else ""
+ await api.send_message(
+ chat_id,
+ greeting + f"You are registered as {_e(panel_user.get('username'))}.\n\n"
+ f"Your connections ({len(conns)}) — tap to get config:",
+ reply_markup=kb,
+ )
+
+
+async def _handle_refresh(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, tg_id: str, load_data_fn: Callable):
+ await api.answer_callback(callback_id, "Updated!")
+ panel_user = _find_user(load_data_fn, tg_id)
+ if not panel_user:
+ await api.edit_message(chat_id, message_id, "❌ Access denied.")
+ return
+ data = load_data_fn()
+ conns = [c for c in data.get("user_connections", []) if c.get("user_id") == panel_user.get("id")]
+ if not conns:
+ await api.edit_message(chat_id, message_id, "You have no connections.")
+ return
+ kb = _build_connections_keyboard(conns, data)
+ await api.edit_message(chat_id, message_id, f"Your connections ({len(conns)}) — tap to get config:", reply_markup=kb)
+
+
+async def _handle_get_config(api: TelegramAPI, chat_id: int, message_id: int, callback_id: str, conn_id: str, tg_id: str, load_data_fn: Callable, generate_vpn_link_fn: Callable):
+ await api.answer_callback(callback_id, "Fetching config...")
+
+ panel_user = _find_user(load_data_fn, tg_id)
+ if not panel_user:
+ await api.send_message(chat_id, "❌ Access denied.")
+ return
+
+ data = load_data_fn()
+ conn = next((c for c in data.get("user_connections", []) if c.get("id") == conn_id and (_is_admin(panel_user) or c.get("user_id") == panel_user.get("id"))), None)
+ if not conn:
+ await api.send_message(chat_id, "❌ Connection not found.")
+ return
+
+ servers = data.get("servers", [])
+ sid = conn.get("server_id")
+ if not isinstance(sid, int) or sid >= len(servers):
+ await api.send_message(chat_id, "❌ Server not found.")
+ return
+
+ await _send_config_by_client(api, chat_id, servers[sid], conn.get("protocol", "awg"), conn.get("client_id"), conn.get("name", "Connection"), generate_vpn_link_fn)
+
+
+async def _send_config_by_client(api: TelegramAPI, chat_id: int, server: dict, proto: str, client_id: str, conn_name: str, generate_vpn_link_fn: Callable):
+ loading_result = await api.send_message(chat_id, f"⏳ Fetching config for {_e(conn_name)}...")
+ loading_msg_id = loading_result.get("result", {}).get("message_id")
+ try:
+ proto_info = server.get("protocols", {}).get(proto, {})
+ port = proto_info.get("port", "55424")
+
+ def _get_cfg():
+ ssh, manager = _get_ssh_and_manager(server, proto)
+ try:
+ ssh.connect()
+ return _manager_call(manager, "get_client_config", proto, client_id, server["host"], port)
+ finally:
+ ssh.disconnect()
+
+ config = await asyncio.to_thread(_get_cfg)
+ if not config:
+ if loading_msg_id:
+ await api.edit_message(chat_id, loading_msg_id, "❌ Failed to retrieve configuration.")
+ return
+
+ if loading_msg_id:
+ await api.call("deleteMessage", chat_id=chat_id, message_id=loading_msg_id)
+
+ server_name = server.get("name") or server.get("host", "Unknown")
+ await api.send_message(chat_id, f"✅ {_e(conn_name)}\n🌐 Server: {_e(server_name)}\n🔌 Protocol: {_e(proto.upper())}")
+
+ is_link_proto = _proto_base(proto) in ("xray", "telemt")
+ if is_link_proto:
+ await api.send_message(chat_id, f"🔗 Connection link (tap to copy):\n{_e(config)}")
+ else:
+ MAX_LEN = 4000
+ if len(config) <= MAX_LEN:
+ await api.send_message(chat_id, f"📄 Configuration:\n{_e(config)}")
+ else:
+ chunks = [config[i:i + MAX_LEN] for i in range(0, len(config), MAX_LEN)]
+ for i, chunk in enumerate(chunks, 1):
+ await api.send_message(chat_id, f"📄 Configuration (part {i}/{len(chunks)}):\n{_e(chunk)}")
+
+ vpn_link = generate_vpn_link_fn(config) if config else ""
+ if vpn_link:
+ await api.send_message(chat_id, f"🔗 VPN Link (tap to copy):\n{_e(vpn_link)}")
+ filename = f"{str(conn_name).replace(' ', '_')}.conf"
+ await api.send_document(chat_id, filename=filename, content=config.encode("utf-8"), caption=f"📁 Config file: {conn_name}")
+ except Exception as e:
+ logger.exception("Bot: error getting config")
+ if loading_msg_id:
+ await api.edit_message(chat_id, loading_msg_id, f"❌ Error: {_e(e)}")
+ else:
+ await api.send_message(chat_id, f"❌ Error: {_e(e)}")
+
+
+# ----------------------------------------------------------------------- #
+# Admin handlers
+# ----------------------------------------------------------------------- #
+def _require_admin(load_data_fn: Callable, tg_id: str):
+ user = _find_user(load_data_fn, tg_id)
+ if not user or not _is_admin(user):
+ return None
+ return user
+
+
+async def _handle_add_server_command(api: TelegramAPI, msg: dict, load_data_fn: Callable, save_data_fn: Optional[Callable]):
+ chat_id = msg["chat"]["id"]
+ tg_id = str(msg["from"]["id"])
+ if not _require_admin(load_data_fn, tg_id):
+ await api.send_message(chat_id, "❌ Access denied.")
+ return
+ if not save_data_fn:
+ await api.send_message(chat_id, "❌ Saving is not available for this bot instance.")
+ return
+
+ text = msg.get("text", "")
+ parts = text.split(maxsplit=5)
+ if len(parts) < 4:
+ await api.send_message(
+ chat_id,
+ "Usage:\n"
+ "/addserver host username password [ssh_port] [name]\n\n"
+ "Example:\n"
+ "/addserver 203.0.113.10 root myPassword 22 Prod VPS\n\n"
+ "⚠️ Telegram messages are not a secrets manager. Prefer adding servers in the web panel if possible.",
+ )
+ return
+
+ host = parts[1]
+ username = parts[2]
+ password = parts[3]
+ ssh_port = 22
+ name = host
+ if len(parts) >= 5:
+ try:
+ ssh_port = int(parts[4])
+ except Exception:
+ name = parts[4]
+ if len(parts) >= 6:
+ name = parts[5] or host
+
+ data = load_data_fn()
+ data.setdefault("servers", []).append({
+ "name": name,
+ "host": host,
+ "ssh_port": ssh_port,
+ "username": username,
+ "password": password,
+ "private_key": "",
+ "protocols": {},
+ "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
+ })
+ save_data_fn(data)
+ await api.send_message(chat_id, f"✅ Server added: {_e(name)}\nHost: {_e(host)}")
+
+
+async def _admin_servers(api: TelegramAPI, chat_id: int, message_id: Optional[int], load_data_fn: Callable):
+ data = load_data_fn()
+ servers = data.get("servers", [])
+ text = f"🖥 Servers ({len(servers)})\n\nChoose a server:"
+ if message_id:
+ await api.edit_message(chat_id, message_id, text, reply_markup=_server_keyboard(data))
+ else:
+ await api.send_message(chat_id, text, reply_markup=_server_keyboard(data))
+
+
+async def _admin_users(api: TelegramAPI, chat_id: int, message_id: int, load_data_fn: Callable):
+ data = load_data_fn()
+ users = data.get("users", [])
+ await api.edit_message(chat_id, message_id, f"👤 Users ({len(users)})\n\nChoose a user:", reply_markup=_users_keyboard(data))
+
+
+async def _admin_user_detail(api: TelegramAPI, chat_id: int, message_id: int, user_id: str, load_data_fn: Callable):
+ data = load_data_fn()
+ user = next((u for u in data.get("users", []) if u.get("id") == user_id), None)
+ if not user:
+ await api.edit_message(chat_id, message_id, "❌ User not found.", reply_markup={"inline_keyboard": [[{"text": "⬅️ Users", "callback_data": "adm:users"}]]})
+ return
+ conns = [c for c in data.get("user_connections", []) if c.get("user_id") == user_id]
+ lines = [
+ f"👤 {_e(user.get('username'))}",
+ f"Role: {_e(user.get('role', 'user'))}",
+ f"Enabled: {'yes ✅' if user.get('enabled', True) else 'no 🚫'}",
+ f"Telegram ID: {_e(user.get('telegramId') or '-')}",
+ f"Email: {_e(user.get('email') or '-')}",
+ f"Connections: {len(conns)}",
+ ]
+ if user.get("description"):
+ lines.append(f"Description: {_e(user.get('description'))}")
+ rows = []
+ servers = data.get("servers", [])
+ for c in conns[:20]:
+ sid = c.get("server_id")
+ server_name = "Unknown"
+ if isinstance(sid, int) and sid < len(servers):
+ server_name = servers[sid].get("name") or servers[sid].get("host") or "Unknown"
+ rows.append([{"text": f"🔐 {c.get('name', 'Connection')} · {c.get('protocol', '').upper()} · {server_name}", "callback_data": f"cfg:{c.get('id')}"}])
+ rows.append([{"text": "⬅️ Users", "callback_data": "adm:users"}])
+ rows.append([{"text": "⬅️ Admin menu", "callback_data": "adm:menu"}])
+ await api.edit_message(chat_id, message_id, "\n".join(lines), reply_markup={"inline_keyboard": rows})
+
+
+async def _admin_server_detail(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, load_data_fn: Callable):
+ data = load_data_fn()
+ servers = data.get("servers", [])
+ if server_id < 0 or server_id >= len(servers):
+ await api.edit_message(chat_id, message_id, "❌ Server not found.")
+ return
+ server = await _refresh_server_protocol_statuses_async(servers[server_id])
+ protocols = server.get("protocols", {}) or {}
+ text = (
+ f"🖥 {_e(server.get('name') or server.get('host'))}\n"
+ f"Host: {_e(server.get('host'))}\n"
+ f"SSH: {_e(server.get('username'))}@{_e(server.get('host'))}:{_e(server.get('ssh_port', 22))}\n\n"
+ f"Protocols ({len(protocols)}):"
+ )
+ await api.edit_message(chat_id, message_id, text, reply_markup=_protocols_keyboard(server_id, server))
+
+
+async def _admin_protocol_detail(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, load_data_fn: Callable):
+ data = load_data_fn()
+ servers = data.get("servers", [])
+ if server_id < 0 or server_id >= len(servers):
+ await api.edit_message(chat_id, message_id, "❌ Server not found.")
+ return
+ server = await _refresh_server_protocol_statuses_async(servers[server_id])
+ info = (server.get("protocols", {}) or {}).get(proto)
+ if not info:
+ await api.edit_message(chat_id, message_id, "❌ Protocol not found.")
+ return
+ lines = [
+ f"🔌 {_e(_protocol_display_name(proto))}",
+ f"Server: {_e(server.get('name') or server.get('host'))}",
+ f"Status: {_protocol_status_text(info)}",
+ ]
+ for key in ("port", "container_name", "domain", "site_url", "web_port", "mode"):
+ if info.get(key) not in (None, ""):
+ lines.append(f"{_e(key)}: {_e(info.get(key))}")
+ if info.get("status_error"):
+ lines.append(f"status_error: {_e(info.get('status_error'))}")
+ await api.edit_message(chat_id, message_id, "\n".join(lines), reply_markup=_protocol_keyboard(server_id, proto, info))
+
+
+async def _admin_toggle_protocol(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, start: bool, load_data_fn: Callable):
+ await api.edit_message(chat_id, message_id, "⏳ Updating protocol container...")
+
+ def _toggle():
+ data = load_data_fn()
+ server = data["servers"][server_id]
+ ssh, manager = _get_ssh_and_manager(server, proto)
+ try:
+ ssh.connect()
+ container = (server.get("protocols", {}).get(proto, {}) or {}).get("container_name")
+ if not container:
+ # fallback: most managers expose CONTAINER_NAME for base/first instances
+ container = getattr(manager, "CONTAINER_NAME", None)
+ if not container:
+ raise RuntimeError("Container name is unknown")
+ action = "start" if start else "stop"
+ out, err, code = ssh.run_sudo_command(f"docker {action} {container}")
+ if code != 0:
+ raise RuntimeError(err or out or f"docker {action} failed")
+ return data
+ finally:
+ ssh.disconnect()
+
+ try:
+ await asyncio.to_thread(_toggle)
+ await _admin_protocol_detail(api, chat_id, message_id, server_id, proto, load_data_fn)
+ except Exception as e:
+ logger.exception("Bot admin: protocol toggle failed")
+ await api.edit_message(chat_id, message_id, f"❌ Error: {_e(e)}", reply_markup={"inline_keyboard": [[{"text": "⬅️ Protocol", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]]})
+
+
+async def _admin_clients(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, load_data_fn: Callable):
+ await api.edit_message(chat_id, message_id, "⏳ Loading connections...")
+
+ def _load_clients():
+ data = load_data_fn()
+ server = data["servers"][server_id]
+ ssh, manager = _get_ssh_and_manager(server, proto)
+ try:
+ ssh.connect()
+ return data, _manager_call(manager, "get_clients", proto)
+ finally:
+ ssh.disconnect()
+
+ try:
+ data, clients = await asyncio.to_thread(_load_clients)
+ if not clients:
+ await api.edit_message(chat_id, message_id, "👥 No connections.", reply_markup={"inline_keyboard": [[{"text": "➕ Create connection", "callback_data": _ref("add_client", {"sid": server_id, "proto": proto})}], [{"text": "⬅️ Protocol", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]]})
+ return
+ rows = []
+ conn_by_client = _connection_lookup(data, server_id, proto)
+ users_by_id = {u.get("id"): u for u in data.get("users", [])}
+ for c in clients[:40]:
+ client_id = c.get("clientId") or c.get("client_id") or c.get("id") or ""
+ conn = conn_by_client.get(client_id)
+ name = _client_display_name(c, conn)
+ traffic = ""
+ user_data = c.get("userData") or {}
+ if user_data:
+ total = (user_data.get("dataReceivedBytes") or 0) + (user_data.get("dataSentBytes") or 0)
+ traffic = f" · {_format_bytes(total)}"
+ assigned = ""
+ if conn and conn.get("user_id") in users_by_id:
+ assigned = f" · @{users_by_id[conn.get('user_id')].get('username')}"
+ c["name"] = name
+ c["assigned_user_id"] = conn.get("user_id") if conn else None
+ rows.append([{"text": f"👤 {name}{assigned}{traffic}", "callback_data": _ref("client", {"sid": server_id, "proto": proto, "client_id": client_id, "name": name, "client": c})}])
+ rows.append([{"text": "➕ Create connection", "callback_data": _ref("add_client", {"sid": server_id, "proto": proto})}])
+ rows.append([{"text": "⬅️ Protocol", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}])
+ await api.edit_message(chat_id, message_id, f"👥 {_e(_protocol_display_name(proto))} connections ({len(clients)})", reply_markup={"inline_keyboard": rows})
+ except Exception as e:
+ logger.exception("Bot admin: load clients failed")
+ await api.edit_message(chat_id, message_id, f"❌ Error: {_e(e)}", reply_markup={"inline_keyboard": [[{"text": "⬅️ Protocol", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]]})
+
+
+async def _admin_client_detail(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, client: dict):
+ client_id = client.get("clientId") or client.get("client_id") or client.get("id") or ""
+ name = _client_display_name(client)
+ user_data = client.get("userData") or {}
+ rx = user_data.get("dataReceivedBytes") or 0
+ tx = user_data.get("dataSentBytes") or 0
+ enabled = client.get("enabled")
+ if enabled is None:
+ enabled = client.get("isEnabled")
+ enabled_text = "enabled ✅" if (enabled is None or enabled) else "disabled 🚫"
+ text = (
+ f"👤 {_e(name)}\n"
+ f"Protocol: {_e(_protocol_display_name(proto))}\n"
+ f"Client ID: {_e(client_id)}\n"
+ f"Status: {enabled_text}"
+ )
+ if user_data:
+ text += f"\nTraffic: {_format_bytes(rx + tx)}\nRX: {_format_bytes(rx)} · TX: {_format_bytes(tx)}"
+ await api.edit_message(chat_id, message_id, text, reply_markup=_client_keyboard(server_id, proto, client))
+
+
+async def _admin_add_client(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, panel_user: dict, load_data_fn: Callable, save_data_fn: Optional[Callable], generate_vpn_link_fn: Callable):
+ if not save_data_fn:
+ await api.edit_message(chat_id, message_id, "❌ Saving is not available for this bot instance.")
+ return
+ _pending_inputs[str(chat_id)] = {
+ "kind": "add_client_name",
+ "sid": server_id,
+ "proto": proto,
+ "admin_user_id": panel_user.get("id"),
+ "ts": time.time(),
+ }
+ await api.edit_message(
+ chat_id,
+ message_id,
+ "➕ Create connection\n\n"
+ f"Server/protocol: {_e(_protocol_display_name(proto))}\n\n"
+ "Send the connection name in the next message.\n"
+ "Example: Ivan iPhone\n\n"
+ "Send /cancel to cancel.",
+ reply_markup={"inline_keyboard": [[{"text": "❌ Cancel", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}]]},
+ )
+
+
+async def _admin_choose_client_user(api: TelegramAPI, chat_id: int, name: str, server_id: int, proto: str, load_data_fn: Callable):
+ data = load_data_fn()
+ await api.send_message(
+ chat_id,
+ "✅ Connection name: {}\n\n"
+ "Assign this connection to a panel user?".format(_e(name)),
+ reply_markup=_assign_user_keyboard(data, server_id, proto, name),
+ )
+
+
+async def _admin_create_client(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, name: str, user_id: Optional[str], load_data_fn: Callable, save_data_fn: Optional[Callable], generate_vpn_link_fn: Callable):
+ if not save_data_fn:
+ await api.edit_message(chat_id, message_id, "❌ Saving is not available for this bot instance.")
+ return
+ await api.edit_message(chat_id, message_id, "⏳ Creating connection...")
+
+ def _create():
+ data = load_data_fn()
+ server = data["servers"][server_id]
+ proto_info = (server.get("protocols", {}) or {}).get(proto, {})
+ port = proto_info.get("port", "55424")
+ ssh, manager = _get_ssh_and_manager(server, proto)
+ try:
+ ssh.connect()
+ if _proto_base(proto) == "telemt":
+ result = manager.add_client(proto, name, server["host"], port)
+ elif _proto_base(proto) == "wireguard":
+ result = manager.add_client(name, server["host"])
+ else:
+ result = manager.add_client(proto, name, server["host"], port)
+ finally:
+ ssh.disconnect()
+ client_id = result.get("client_id") or result.get("clientId")
+ assigned_user = None
+ if user_id:
+ assigned_user = next((u for u in data.get("users", []) if u.get("id") == user_id), None)
+ if user_id and client_id:
+ data.setdefault("user_connections", []).append({
+ "id": str(uuid.uuid4()),
+ "user_id": user_id,
+ "server_id": server_id,
+ "protocol": proto,
+ "client_id": client_id,
+ "name": name,
+ "created_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
+ })
+ save_data_fn(data)
+ return server, result, client_id, assigned_user
+
+ try:
+ server, result, client_id, assigned_user = await asyncio.to_thread(_create)
+ assigned_text = f"\nAssigned to: {_e(assigned_user.get('username'))}" if assigned_user else "\nAssigned: not linked"
+ await api.edit_message(chat_id, message_id, f"✅ Connection created: {_e(name)}{assigned_text}")
+ config = result.get("config")
+ if config:
+ await _send_config_text(api, chat_id, server, proto, name, config, generate_vpn_link_fn)
+ elif client_id:
+ await _send_config_by_client(api, chat_id, server, proto, client_id, name, generate_vpn_link_fn)
+ except Exception as e:
+ logger.exception("Bot admin: add client failed")
+ await api.edit_message(chat_id, message_id, f"❌ Error: {_e(e)}", reply_markup={"inline_keyboard": [[{"text": "⬅️ Protocol", "callback_data": _ref("proto", {"sid": server_id, "proto": proto})}]]})
+
+
+async def _send_config_text(api: TelegramAPI, chat_id: int, server: dict, proto: str, conn_name: str, config: str, generate_vpn_link_fn: Callable):
+ await api.send_message(chat_id, f"✅ {_e(conn_name)}\n🌐 Server: {_e(server.get('name') or server.get('host'))}\n🔌 Protocol: {_e(proto.upper())}")
+ if _proto_base(proto) in ("xray", "telemt"):
+ await api.send_message(chat_id, f"🔗 Connection link:\n{_e(config)}")
+ else:
+ await api.send_message(chat_id, f"📄 Configuration:\n{_e(config)}")
+ vpn_link = generate_vpn_link_fn(config) if config else ""
+ if vpn_link:
+ await api.send_message(chat_id, f"🔗 VPN Link:\n{_e(vpn_link)}")
+ await api.send_document(chat_id, filename=f"{conn_name}.conf", content=config.encode("utf-8"), caption=f"📁 Config file: {conn_name}")
+
+
+async def _admin_toggle_client(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, client_id: str, enable: bool, load_data_fn: Callable):
+ await api.edit_message(chat_id, message_id, "⏳ Updating connection...")
+
+ def _toggle():
+ data = load_data_fn()
+ server = data["servers"][server_id]
+ ssh, manager = _get_ssh_and_manager(server, proto)
+ try:
+ ssh.connect()
+ return _manager_call(manager, "toggle_client", proto, client_id, enable)
+ finally:
+ ssh.disconnect()
+
+ try:
+ await asyncio.to_thread(_toggle)
+ await api.edit_message(chat_id, message_id, "✅ Updated.", reply_markup={"inline_keyboard": [[{"text": "⬅️ Connections", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}]]})
+ except Exception as e:
+ logger.exception("Bot admin: toggle client failed")
+ await api.edit_message(chat_id, message_id, f"❌ Error: {_e(e)}")
+
+
+async def _admin_remove_client(api: TelegramAPI, chat_id: int, message_id: int, server_id: int, proto: str, client_id: str, load_data_fn: Callable, save_data_fn: Optional[Callable]):
+ if not save_data_fn:
+ await api.edit_message(chat_id, message_id, "❌ Saving is not available for this bot instance.")
+ return
+ await api.edit_message(chat_id, message_id, "⏳ Removing connection...")
+
+ def _remove():
+ data = load_data_fn()
+ server = data["servers"][server_id]
+ ssh, manager = _get_ssh_and_manager(server, proto)
+ try:
+ ssh.connect()
+ _manager_call(manager, "remove_client", proto, client_id)
+ finally:
+ ssh.disconnect()
+ data["user_connections"] = [
+ c for c in data.get("user_connections", [])
+ if not (c.get("server_id") == server_id and c.get("protocol") == proto and c.get("client_id") == client_id)
+ ]
+ save_data_fn(data)
+
+ try:
+ await asyncio.to_thread(_remove)
+ await api.edit_message(chat_id, message_id, "✅ Connection removed.", reply_markup={"inline_keyboard": [[{"text": "⬅️ Connections", "callback_data": _ref("clients", {"sid": server_id, "proto": proto})}]]})
+ except Exception as e:
+ logger.exception("Bot admin: remove client failed")
+ await api.edit_message(chat_id, message_id, f"❌ Error: {_e(e)}")
+
+
+async def _handle_pending_input(api: TelegramAPI, msg: dict, load_data_fn: Callable, save_data_fn: Optional[Callable], generate_vpn_link_fn: Callable) -> bool:
+ chat_id = msg["chat"]["id"]
+ state = _pending_inputs.get(str(chat_id))
+ if not state:
+ return False
+
+ text = (msg.get("text") or "").strip()
+ if text.lower() in ("/cancel", "cancel"):
+ _pending_inputs.pop(str(chat_id), None)
+ await api.send_message(chat_id, "❌ Action cancelled.", reply_markup=_admin_main_keyboard())
+ return True
+ if text.startswith("/"):
+ _pending_inputs.pop(str(chat_id), None)
+ return False
+
+ if state.get("kind") == "add_client_name":
+ panel_user = _require_admin(load_data_fn, str(msg["from"]["id"]))
+ if not panel_user:
+ _pending_inputs.pop(str(chat_id), None)
+ await api.send_message(chat_id, "❌ Access denied.")
+ return True
+ name = text[:80].strip()
+ if not name:
+ await api.send_message(chat_id, "Name cannot be empty. Send a connection name or /cancel.")
+ return True
+ _pending_inputs.pop(str(chat_id), None)
+ await _admin_choose_client_user(api, chat_id, name, int(state.get("sid", 0)), state.get("proto", "awg"), load_data_fn)
+ return True
+
+ return False
+
+
+# ----------------------------------------------------------------------- #
+# Main polling loop and dispatcher
+# ----------------------------------------------------------------------- #
+async def _run_bot(token: str, load_data_fn: Callable, generate_vpn_link_fn: Callable, save_data_fn: Optional[Callable] = None):
+ offset = 0
+ logger.info("Telegram bot started (raw httpx polling).")
+
+ async with httpx.AsyncClient() as client:
+ api = TelegramAPI(token, client)
+
+ me = await api.call("getMe")
+ if not me.get("ok"):
+ logger.error(f"Telegram bot: invalid token or API error: {me}")
+ return
+ logger.info(f"Telegram bot logged in as @{me['result']['username']}")
+
+ while True:
+ try:
+ updates = await api.get_updates(offset=offset, timeout=25)
+ except asyncio.CancelledError:
+ logger.info("Telegram bot polling cancelled.")
+ return
+ except Exception as e:
+ logger.warning(f"Telegram bot polling error: {e}")
+ await asyncio.sleep(5)
+ continue
+
+ for update in updates:
+ offset = update["update_id"] + 1
+ try:
+ await _dispatch(api, update, load_data_fn, generate_vpn_link_fn, save_data_fn)
+ except asyncio.CancelledError:
+ return
+ except Exception as e:
+ logger.exception(f"Telegram bot: error handling update {update['update_id']}: {e}")
+
+
+async def _dispatch(api: TelegramAPI, update: dict, load_data_fn: Callable, generate_vpn_link_fn: Callable, save_data_fn: Optional[Callable] = None):
+ if "message" in update:
+ msg = update["message"]
+ text = msg.get("text", "")
+ if await _handle_pending_input(api, msg, load_data_fn, save_data_fn, generate_vpn_link_fn):
+ return
+ if text.startswith("/start") or text.startswith("/admin"):
+ await _handle_start(api, msg, load_data_fn)
+ elif text.startswith("/connections"):
+ panel_user = _find_user(load_data_fn, str(msg["from"]["id"]))
+ if not panel_user:
+ await api.send_message(msg["chat"]["id"], "❌ Access denied.")
+ else:
+ await _send_user_connections(api, msg["chat"]["id"], panel_user, load_data_fn)
+ elif text.startswith("/servers"):
+ if _require_admin(load_data_fn, str(msg["from"]["id"])):
+ await _admin_servers(api, msg["chat"]["id"], None, load_data_fn)
+ else:
+ await api.send_message(msg["chat"]["id"], "❌ Access denied.")
+ elif text.startswith("/addserver"):
+ await _handle_add_server_command(api, msg, load_data_fn, save_data_fn)
+
+ elif "callback_query" in update:
+ cq = update["callback_query"]
+ callback_id = cq["id"]
+ data_str = cq.get("data", "")
+ chat_id = cq["message"]["chat"]["id"]
+ message_id = cq["message"]["message_id"]
+ tg_id = str(cq["from"]["id"])
+
+ if data_str == "noop":
+ await api.answer_callback(callback_id)
+ return
+ if data_str == "refresh":
+ await _handle_refresh(api, chat_id, message_id, callback_id, tg_id, load_data_fn)
+ return
+ if data_str.startswith("cfg:"):
+ await _handle_get_config(api, chat_id, message_id, callback_id, data_str[4:], tg_id, load_data_fn, generate_vpn_link_fn)
+ return
+
+ panel_user = _require_admin(load_data_fn, tg_id)
+ if not panel_user:
+ await api.answer_callback(callback_id, "Access denied")
+ return
+
+ await api.answer_callback(callback_id)
+
+ if data_str == "adm:menu":
+ await api.edit_message(chat_id, message_id, "Admin menu", reply_markup=_admin_main_keyboard())
+ elif data_str == "adm:servers":
+ await _admin_servers(api, chat_id, message_id, load_data_fn)
+ elif data_str == "adm:users":
+ await _admin_users(api, chat_id, message_id, load_data_fn)
+ elif data_str == "adm:myconns":
+ await _send_user_connections(api, chat_id, panel_user, load_data_fn)
+ elif data_str == "adm:addserver_help":
+ await api.edit_message(
+ chat_id,
+ message_id,
+ "➕ Add server\n\n"
+ "Use command:\n"
+ "/addserver host username password [ssh_port] [name]\n\n"
+ "Example:\n"
+ "/addserver 203.0.113.10 root myPassword 22 Prod VPS\n\n"
+ "⚠️ Prefer the web panel for real credentials if possible.",
+ reply_markup={"inline_keyboard": [[{"text": "⬅️ Admin menu", "callback_data": "adm:menu"}]]},
+ )
+ elif data_str.startswith("srv:"):
+ await _admin_server_detail(api, chat_id, message_id, int(data_str.split(":", 1)[1]), load_data_fn)
+ else:
+ ref = _resolve_ref(data_str)
+ if not ref:
+ await api.edit_message(chat_id, message_id, "❌ Action expired. Use /start again.")
+ return
+ action = ref.get("action")
+ payload = ref.get("payload", {})
+ sid = int(payload.get("sid", 0) or 0)
+ proto = payload.get("proto", "awg")
+ if action == "user":
+ await _admin_user_detail(api, chat_id, message_id, payload.get("uid"), load_data_fn)
+ elif action == "proto":
+ await _admin_protocol_detail(api, chat_id, message_id, sid, proto, load_data_fn)
+ elif action == "toggle_proto":
+ await _admin_toggle_protocol(api, chat_id, message_id, sid, proto, bool(payload.get("start")), load_data_fn)
+ elif action == "clients":
+ await _admin_clients(api, chat_id, message_id, sid, proto, load_data_fn)
+ elif action == "client":
+ await _admin_client_detail(api, chat_id, message_id, sid, proto, payload.get("client", {}))
+ elif action == "client_cfg":
+ data = load_data_fn()
+ server = data["servers"][sid]
+ await _send_config_by_client(api, chat_id, server, proto, payload.get("client_id"), payload.get("name", "Connection"), generate_vpn_link_fn)
+ elif action == "add_client":
+ await _admin_add_client(api, chat_id, message_id, sid, proto, panel_user, load_data_fn, save_data_fn, generate_vpn_link_fn)
+ elif action == "create_client":
+ await _admin_create_client(api, chat_id, message_id, sid, proto, payload.get("name", "Connection"), payload.get("user_id"), load_data_fn, save_data_fn, generate_vpn_link_fn)
+ elif action == "toggle_client":
+ await _admin_toggle_client(api, chat_id, message_id, sid, proto, payload.get("client_id"), bool(payload.get("enable")), load_data_fn)
+ elif action == "remove_client":
+ await _admin_remove_client(api, chat_id, message_id, sid, proto, payload.get("client_id"), load_data_fn, save_data_fn)
diff --git a/templates/base.html b/templates/base.html
new file mode 100644
index 0000000..695c326
--- /dev/null
+++ b/templates/base.html
@@ -0,0 +1,269 @@
+
+
+
+
+
+
+
+ {{ _('api_docs_hint') }}
+{{ request.url.scheme }}://{{ request.url.netloc }}
+
+ {{ _('tunnels_hint') }}
++ {{ _('restore_confirm') }} +
++ {{ _('api_tokens_hint') }} +
+{{ _('user_label_share') }}: {{ + share_user.username }} +
+{{ _('share_protected_desc') }} +
+ + +{{ _('loading_share_conns') }}
+{{ _('no_active_conns') }}
+{{ _('loading_users') }}
+