diff --git a/app.py b/app.py index 5c911fb..a70fb21 100644 --- a/app.py +++ b/app.py @@ -2,6 +2,7 @@ import os import sys import json import logging +from collections import Counter from dotenv import load_dotenv load_dotenv() @@ -167,6 +168,11 @@ load_translations() DATA_LOCK = asyncio.Lock() +async def load_data_async(): + """Load panel state off the event loop (uses in-process cache).""" + return await asyncio.to_thread(load_data) + + async def save_data_async(data): """Saves panel state to PostgreSQL in a thread-safe way.""" async with DATA_LOCK: @@ -183,6 +189,37 @@ def get_ssh(server): ) +def get_current_user(request: Request, data: Optional[dict] = None): + user_id = request.session.get('user_id') + if not user_id: + return None + snapshot = data if data is not None else load_data() + for u in snapshot.get('users', []): + if u['id'] == user_id: + return u + return None + + +def tpl(request, template, **kwargs): + data = load_data() + lang = request.cookies.get('lang', 'en') + ctx = { + 'request': request, + 'current_user': get_current_user(request, data), + 'site_settings': data.get('settings', {}).get('appearance', {}), + 'captcha_settings': data.get('settings', {}).get('captcha', {}), + 'telegram_settings': data.get('settings', {}).get('telegram', {}), + 'bot_running': tg_bot.is_running(), + 'lang': lang, + '_': lambda text_id: _t(text_id, lang), + 'translations_json': json.dumps(TRANSLATIONS.get(lang, TRANSLATIONS.get('en', {}))), + # Keep for legacy JS; prefer translations_json on new pages + 'all_translations_json': json.dumps(TRANSLATIONS), + } + ctx.update(kwargs) + return templates.TemplateResponse(template, ctx) + + def get_panel_local_url(request: Optional[Request] = None): data = load_data() ssl_conf = data.get('settings', {}).get('ssl', {}) @@ -1337,36 +1374,6 @@ async def sync_users_with_remnawave(data: dict): return 0, f"Error: {str(e)}" -def get_current_user(request: Request): - user_id = request.session.get('user_id') - if not user_id: - return None - data = load_data() - for u in data.get('users', []): - if u['id'] == user_id: - return u - return None - - -def tpl(request, template, **kwargs): - data = load_data() - lang = request.cookies.get('lang', 'en') - ctx = { - 'request': request, - 'current_user': get_current_user(request), - 'site_settings': data.get('settings', {}).get('appearance', {}), - 'captcha_settings': data.get('settings', {}).get('captcha', {}), - 'telegram_settings': data.get('settings', {}).get('telegram', {}), - 'bot_running': tg_bot.is_running(), - 'lang': lang, - '_': lambda text_id: _t(text_id, lang), - 'translations_json': json.dumps(TRANSLATIONS.get(lang, TRANSLATIONS.get('en', {}))), - 'all_translations_json': json.dumps(TRANSLATIONS) - } - ctx.update(kwargs) - return templates.TemplateResponse(template, ctx) - - # ======================== Pydantic Models ======================== class LoginRequest(BaseModel): @@ -1946,22 +1953,24 @@ async def server_detail(request: Request, server_id: int): @app.get('/users', response_class=HTMLResponse, tags=["System Templates"]) async def users_page(request: Request): - user = get_current_user(request) + data = load_data() + user = get_current_user(request, data) if not user: return RedirectResponse(url='/login', status_code=302) if user['role'] not in ('admin', 'support'): return RedirectResponse(url='/my', status_code=302) - data = load_data() users_list = data.get('users', []) # Count connections per user conns = data.get('user_connections', []) + conn_counts = Counter(c.get('user_id') for c in conns if c.get('user_id')) for u in users_list: - u['connections_count'] = sum(1 for c in conns if c['user_id'] == u['id']) + u['connections_count'] = conn_counts.get(u['id'], 0) servers = data['servers'] return tpl( request, 'users.html', users=users_list, servers=servers, + current_user=user, ) @@ -2309,54 +2318,97 @@ rm -rf /opt/amnezia return JSONResponse({'error': str(e)}, status_code=500) +def _docker_container_inventory(ssh) -> dict: + """One SSH round-trip: map container name -> running bool.""" + out, _, _ = ssh.run_sudo_command( + "docker ps -a --format '{{.Names}}|{{.Status}}' 2>/dev/null" + ) + inventory = {} + for line in (out or '').splitlines(): + if '|' not in line: + continue + name, status = line.split('|', 1) + name = name.strip() + if not name: + continue + inventory[name] = status.strip().lower().startswith('up') + return inventory + + +def _docker_is_ready(ssh) -> bool: + ver, _, vcode = ssh.run_command("docker --version 2>/dev/null") + return vcode == 0 and bool((ver or '').strip()) + + @app.post('/api/servers/{server_id}/stats', tags=["Servers"]) async def api_server_stats(request: Request, server_id: int): if not _check_admin(request): return JSONResponse({'error': 'Forbidden'}, status_code=403) try: - data = load_data() + data = await load_data_async() 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() - stats = {} - out, _, _ = ssh.run_command( - "top -bn1 | grep 'Cpu(s)' | awk '{print $2}' | cut -d'%' -f1 2>/dev/null || " - "awk '{u=$2+$4; t=$2+$4+$5; if(NR==1){pu=u;pt=t} else printf \"%.1f\", (u-pu)/(t-pt)*100}' " - "<(grep 'cpu ' /proc/stat) <(sleep 0.5 && grep 'cpu ' /proc/stat) 2>/dev/null" - ) - try: - stats['cpu'] = round(float(out.strip().split('\n')[0]), 1) - except (ValueError, IndexError): - stats['cpu'] = 0 - out, _, _ = ssh.run_command("free -b | awk 'NR==2{printf \"%d %d\", $3, $2}'") - try: - parts = out.strip().split() - used, total = int(parts[0]), int(parts[1]) - stats.update(ram_used=used, ram_total=total, ram_percent=round(used / total * 100, 1) if total > 0 else 0) - except (ValueError, IndexError): - stats.update(ram_used=0, ram_total=0, ram_percent=0) - out, _, _ = ssh.run_command("df -B1 / | awk 'NR==2{printf \"%d %d\", $3, $2}'") - try: - parts = out.strip().split() - used, total = int(parts[0]), int(parts[1]) - stats.update(disk_used=used, disk_total=total, disk_percent=round(used / total * 100, 1) if total > 0 else 0) - except (ValueError, IndexError): - stats.update(disk_used=0, disk_total=0, disk_percent=0) - out, _, _ = ssh.run_command( - "DEV=$(ip route | awk '/default/ {print $5}' | head -1); " - "cat /proc/net/dev | awk -v dev=\"$DEV:\" '$1==dev{printf \"%d %d\", $2, $10}'" - ) - try: - parts = out.strip().split() - stats['net_rx'], stats['net_tx'] = int(parts[0]), int(parts[1]) - except (ValueError, IndexError): - stats['net_rx'] = stats['net_tx'] = 0 - out, _, _ = ssh.run_command("uptime -p 2>/dev/null || uptime") - stats['uptime'] = out.strip() - ssh.disconnect() - return stats + + def _fetch_stats(): + ssh = get_ssh(server) + ssh.connect() + try: + script = r""" +cpu=$(top -bn1 2>/dev/null | grep 'Cpu(s)' | awk '{print $2}' | cut -d'%' -f1) +if [ -z "$cpu" ]; then cpu=0; fi +ram=$(free -b 2>/dev/null | awk 'NR==2{printf "%d %d", $3, $2}') +disk=$(df -B1 / 2>/dev/null | awk 'NR==2{printf "%d %d", $3, $2}') +dev=$(ip route 2>/dev/null | awk '/default/ {print $5; exit}') +net=$(awk -v d="${dev}:" '$1==d{printf "%d %d", $2, $10}' /proc/net/dev 2>/dev/null) +up=$(uptime -p 2>/dev/null || uptime) +printf 'CPU|%s\nRAM|%s\nDISK|%s\nNET|%s\nUP|%s\n' "${cpu:-0}" "${ram:-0 0}" "${disk:-0 0}" "${net:-0 0}" "$up" +""" + out, _, _ = ssh.run_command(script) + finally: + ssh.disconnect() + + stats = { + 'cpu': 0, + 'ram_used': 0, 'ram_total': 0, 'ram_percent': 0, + 'disk_used': 0, 'disk_total': 0, 'disk_percent': 0, + 'net_rx': 0, 'net_tx': 0, + 'uptime': '', + } + for line in (out or '').splitlines(): + if '|' not in line: + continue + key, val = line.split('|', 1) + key, val = key.strip(), val.strip() + try: + if key == 'CPU': + stats['cpu'] = round(float(val.split()[0] if val else 0), 1) + elif key == 'RAM': + parts = val.split() + used, total = int(parts[0]), int(parts[1]) + stats.update( + ram_used=used, + ram_total=total, + ram_percent=round(used / total * 100, 1) if total > 0 else 0, + ) + elif key == 'DISK': + parts = val.split() + used, total = int(parts[0]), int(parts[1]) + stats.update( + disk_used=used, + disk_total=total, + disk_percent=round(used / total * 100, 1) if total > 0 else 0, + ) + elif key == 'NET': + parts = val.split() + stats['net_rx'], stats['net_tx'] = int(parts[0]), int(parts[1]) + elif key == 'UP': + stats['uptime'] = val + except (ValueError, IndexError): + continue + return stats + + return await asyncio.to_thread(_fetch_stats) except Exception as e: logger.exception("Error getting server stats") return JSONResponse({'error': str(e)}, status_code=500) @@ -2367,128 +2419,130 @@ async def api_check_server(request: Request, server_id: int): if not _check_admin(request): return JSONResponse({'error': 'Forbidden'}, status_code=403) try: - data = load_data() + data = await load_data_async() 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() - # Just use awg's docker checker since it uses the same command - manager = get_protocol_manager(ssh, 'awg') - status = {'connection': 'ok', 'docker_installed': manager.check_docker_installed(), 'protocols': {}} - - changed = False - if 'protocols' not in server: - server['protocols'] = {} - def merge_saved_protocol_status(proto, result=None, error=None): - """Merge live status with saved protocol metadata. - - Multi-instance protocols are source-of-truth in data.json because - they cannot be discovered from BASE_PROTOCOLS alone. A transient - check failure must not delete awg2__2/awg2__3 from the panel. - """ - db_proto = server.get('protocols', {}).get(proto, {}) or {} - merged = dict(result or {}) - merged.setdefault('protocol', proto) - if error: - merged['error'] = error - if not merged.get('port') and db_proto.get('port'): - merged['port'] = db_proto['port'] - if db_proto.get('awg_params') and not merged.get('awg_params'): - merged['awg_params'] = db_proto.get('awg_params') - merged['base_protocol'] = db_proto.get('base_protocol') or protocol_base(proto) - merged['instance'] = db_proto.get('instance') or protocol_instance(proto) - merged['display_name'] = db_proto.get('display_name') or protocol_display_name(proto) - merged['container_name'] = db_proto.get('container_name') or protocol_container_name(proto) - if protocol_base(proto) == 'adguard': - for key in ('web_port', 'mode', 'internal_ip', 'expose_web'): - if db_proto.get(key) not in (None, ''): - if key == 'web_port' and merged.get('web_exposed'): - continue - merged[key] = db_proto[key] - if 'expose_web' in db_proto and 'web_exposed' not in merged: - merged['web_exposed'] = bool(db_proto.get('expose_web')) - if protocol_base(proto) == 'nginx': - for key in ('domain', 'email', 'site_url'): - if db_proto.get(key) not in (None, ''): - merged[key] = db_proto[key] - return merged - - def should_preserve_saved_protocol(proto, result=None, err=None): - """Return True when check must not remove a saved protocol record.""" - db_proto = server.get('protocols', {}).get(proto) - if not db_proto: - return False - # Additional AWG-family instances are only known by their saved - # dynamic keys (awg__2/awg2__2/awg_legacy__2). Keep them unless - # the user explicitly uninstalls them. - if protocol_base(proto) in MULTI_INSTANCE_PROTOCOLS and protocol_instance(proto) > 1: - return True - # Do not delete any saved protocol on command/check errors; only a - # clean live result with container_exists=False may prune base apps. - if err or (result and result.get('error')): - return True - return False - - def check_proto(proto): + def _run_check(): + ssh = get_ssh(server) + ssh.connect() try: - p_manager = get_protocol_manager(ssh, proto) - result = _manager_call(p_manager, 'get_server_status', proto) - return proto, merge_saved_protocol_status(proto, result), None - except Exception as e: - return proto, merge_saved_protocol_status(proto, {}, str(e)), str(e) + status = { + 'connection': 'ok', + 'docker_installed': _docker_is_ready(ssh), + 'protocols': {}, + } + inventory = _docker_container_inventory(ssh) if status['docker_installed'] else {} - protocols_to_check = list(dict.fromkeys(BASE_PROTOCOLS + list(server.get('protocols', {}).keys()))) - # Run checks sequentially. Several managers use the same SSH connection; - # checking them in parallel through one SSH object can produce false - # negatives and previously caused dynamic AWG instances to be removed. - for proto in protocols_to_check: - proto, result, err = check_proto(proto) - status['protocols'][proto] = result - if err: - continue - if result.get('container_exists'): - if proto not in server['protocols']: - server['protocols'][proto] = { - 'installed': True, - 'port': result.get('port', '55424'), - 'awg_params': result.get('awg_params', {}), - 'base_protocol': protocol_base(proto), - 'instance': protocol_instance(proto), - 'display_name': protocol_display_name(proto), - 'container_name': protocol_container_name(proto), - } + changed = False + if 'protocols' not in server: + server['protocols'] = {} + + def merge_saved_protocol_status(proto, result=None, error=None): + db_proto = server.get('protocols', {}).get(proto, {}) or {} + merged = dict(result or {}) + merged.setdefault('protocol', proto) + if error: + merged['error'] = error + if not merged.get('port') and db_proto.get('port'): + merged['port'] = db_proto['port'] + if db_proto.get('awg_params') and not merged.get('awg_params'): + merged['awg_params'] = db_proto.get('awg_params') + merged['base_protocol'] = db_proto.get('base_protocol') or protocol_base(proto) + merged['instance'] = db_proto.get('instance') or protocol_instance(proto) + merged['display_name'] = db_proto.get('display_name') or protocol_display_name(proto) + merged['container_name'] = db_proto.get('container_name') or protocol_container_name(proto) if protocol_base(proto) == 'adguard': - server['protocols'][proto].update({ - 'mode': result.get('mode'), - 'internal_ip': result.get('internal_ip'), - 'web_port': result.get('web_port'), - 'expose_web': result.get('web_exposed'), - }) + for key in ('web_port', 'mode', 'internal_ip', 'expose_web'): + if db_proto.get(key) not in (None, ''): + if key == 'web_port' and merged.get('web_exposed'): + continue + merged[key] = db_proto[key] + if 'expose_web' in db_proto and 'web_exposed' not in merged: + merged['web_exposed'] = bool(db_proto.get('expose_web')) if protocol_base(proto) == 'nginx': - server['protocols'][proto].update({ - 'domain': result.get('domain'), - 'email': result.get('email'), - 'site_url': result.get('site_url'), - }) - changed = True - else: - if proto in server['protocols']: - if should_preserve_saved_protocol(proto, result, err): - # Keep saved dynamic instances visible as installed but stopped/unchecked. - status['protocols'][proto]['container_exists'] = True - status['protocols'][proto].setdefault('container_running', False) - status['protocols'][proto]['status_preserved'] = True + for key in ('domain', 'email', 'site_url'): + if db_proto.get(key) not in (None, ''): + merged[key] = db_proto[key] + return merged + + def should_preserve_saved_protocol(proto, result=None, err=None): + db_proto = server.get('protocols', {}).get(proto) + if not db_proto: + return False + if protocol_base(proto) in MULTI_INSTANCE_PROTOCOLS and protocol_instance(proto) > 1: + return True + if err or (result and result.get('error')): + return True + return False + + def check_proto(proto): + cname = protocol_container_name(proto) + # Fast path: skip deep manager probes when container is absent + if cname and cname not in inventory: + return proto, merge_saved_protocol_status(proto, { + 'container_exists': False, + 'container_running': False, + 'protocol': proto, + }), None + try: + p_manager = get_protocol_manager(ssh, proto) + result = _manager_call(p_manager, 'get_server_status', proto) + return proto, merge_saved_protocol_status(proto, result), None + except Exception as e: + return proto, merge_saved_protocol_status(proto, {}, str(e)), str(e) + + protocols_to_check = list(dict.fromkeys( + BASE_PROTOCOLS + list(server.get('protocols', {}).keys()) + )) + for proto in protocols_to_check: + proto, result, err = check_proto(proto) + status['protocols'][proto] = result + if err: + continue + if result.get('container_exists'): + if proto not in server['protocols']: + server['protocols'][proto] = { + 'installed': True, + 'port': result.get('port', '55424'), + 'awg_params': result.get('awg_params', {}), + 'base_protocol': protocol_base(proto), + 'instance': protocol_instance(proto), + 'display_name': protocol_display_name(proto), + 'container_name': protocol_container_name(proto), + } + if protocol_base(proto) == 'adguard': + server['protocols'][proto].update({ + 'mode': result.get('mode'), + 'internal_ip': result.get('internal_ip'), + 'web_port': result.get('web_port'), + 'expose_web': result.get('web_exposed'), + }) + if protocol_base(proto) == 'nginx': + server['protocols'][proto].update({ + 'domain': result.get('domain'), + 'email': result.get('email'), + 'site_url': result.get('site_url'), + }) + changed = True else: - del server['protocols'][proto] - changed = True - - if changed: - save_data(data) - - ssh.disconnect() - return status + if proto in server['protocols']: + if should_preserve_saved_protocol(proto, result, err): + status['protocols'][proto]['container_exists'] = True + status['protocols'][proto].setdefault('container_running', False) + status['protocols'][proto]['status_preserved'] = True + else: + del server['protocols'][proto] + changed = True + + if changed: + save_data(data) + return status + finally: + ssh.disconnect() + + return await asyncio.to_thread(_run_check) except Exception as e: logger.exception("Error checking server") return JSONResponse({'error': str(e), 'connection': 'failed'}, status_code=500) @@ -3470,9 +3524,10 @@ async def api_toggle_connection(request: Request, server_id: int, req: ToggleCon async def api_list_users(request: Request, search: str = '', page: int = 1, size: int = 10): if not _check_admin(request): return JSONResponse({'error': 'Forbidden'}, status_code=403) - data = load_data() + data = await load_data_async() all_users = data.get('users', []) conns = data.get('user_connections', []) + conn_counts = Counter(c.get('user_id') for c in conns if c.get('user_id')) # Filter filtered = [] @@ -3500,7 +3555,7 @@ async def api_list_users(request: Request, search: str = '', page: int = 1, size 'telegramId': u.get('telegramId'), 'email': u.get('email'), 'description': u.get('description'), - 'connections_count': sum(1 for c in conns if c['user_id'] == u['id']), + 'connections_count': conn_counts.get(u['id'], 0), 'traffic_used': u.get('traffic_used', 0), 'traffic_total': u.get('traffic_total', 0), 'traffic_limit': u.get('traffic_limit', 0), diff --git a/db/store.py b/db/store.py index 9edf9b3..9025dbc 100644 --- a/db/store.py +++ b/db/store.py @@ -6,10 +6,12 @@ FastAPI handlers keep working without a full rewrite. from __future__ import annotations +import copy import json import logging import os import shutil +import threading from datetime import datetime, timezone from pathlib import Path from typing import Any, Optional @@ -21,6 +23,11 @@ from .connection import get_pool, init_schema logger = logging.getLogger(__name__) +# Process-wide snapshot cache. load_data() returns a deep copy so callers can +# mutate safely; save_data() refreshes the cache after a successful write. +_DATA_CACHE: Optional[dict] = None +_DATA_CACHE_LOCK = threading.RLock() + DEFAULT_SETTINGS = { 'appearance': { 'title': 'Amnezia', @@ -208,7 +215,13 @@ def _row_to_invite(row) -> dict: } -def load_data() -> dict: +def invalidate_data_cache() -> None: + global _DATA_CACHE + with _DATA_CACHE_LOCK: + _DATA_CACHE = None + + +def _fetch_data_from_db() -> dict: init_schema() pool = get_pool() with pool.connection() as conn: @@ -266,8 +279,20 @@ def load_data() -> dict: } +def load_data() -> dict: + """Return panel state. Uses an in-process cache; always returns a deep copy.""" + global _DATA_CACHE + with _DATA_CACHE_LOCK: + if _DATA_CACHE is not None: + return copy.deepcopy(_DATA_CACHE) + data = _fetch_data_from_db() + _DATA_CACHE = data + return copy.deepcopy(data) + + def save_data(data: dict) -> None: """Replace panel state in a single transaction (same semantics as rewriting data.json).""" + global _DATA_CACHE init_schema() servers = data.get('servers') or [] users = data.get('users') or [] @@ -419,6 +444,16 @@ def save_data(data: dict) -> None: ) conn.commit() + # Keep caller dict and cache aligned with what was actually persisted + data['servers'] = servers + data['users'] = users + data['user_connections'] = connections + data['api_tokens'] = tokens + data['invite_links'] = invite_links + data['settings'] = settings + with _DATA_CACHE_LOCK: + _DATA_CACHE = copy.deepcopy(data) + def export_data_dict() -> dict: """Export current DB state as the legacy data.json document.""" diff --git a/managers/ssh_manager.py b/managers/ssh_manager.py index 437e52e..982c53e 100644 --- a/managers/ssh_manager.py +++ b/managers/ssh_manager.py @@ -66,7 +66,7 @@ class SSHManager: if not self.client: raise ConnectionError("Not connected to server") - logger.info(f"Running command: {command[:100]}...") + logger.debug(f"Running command: {command[:100]}...") stdin, stdout, stderr = self.client.exec_command(command, timeout=timeout) # Crucial: set timeout on the channel to prevent hanging indefinitely