Add 3x-ui user sync via panel API (login or Bearer token).

Co-authored-by: Cursor <cursoragent@cursor.com>

app.py

@@ -1297,6 +1297,7 @@ async def sync_users_with_remnawave(data: dict):
                         'enabled': is_active,
                         'created_at': datetime.now().isoformat(),
                         'remnawave_uuid': rw_u['uuid'],
+                        'xui_email': None,
                         'share_enabled': False,
                         'share_token': secrets.token_urlsafe(16),
                         'share_password_hash': None,
@@ -1329,6 +1330,255 @@ async def sync_users_with_remnawave(data: dict):
         return 0, f"Error: {str(e)}"
 
 
+def _xui_sanitize_username(email: str) -> str:
+    raw = (email or '').strip()
+    if not raw:
+        return ''
+    # Keep readable usernames; strip characters that break our forms
+    cleaned = re.sub(r'[^\w.\-@]+', '_', raw, flags=re.UNICODE)
+    return cleaned[:64] or raw[:64]
+
+
+def _xui_parse_clients_from_inbounds(inbounds) -> list:
+    """Extract unique clients from classic 3x-ui inbounds list payload."""
+    clients_by_email = {}
+    if not isinstance(inbounds, list):
+        return []
+    for inbound in inbounds:
+        settings = inbound.get('settings') if isinstance(inbound, dict) else None
+        if isinstance(settings, str):
+            try:
+                settings = json.loads(settings)
+            except Exception:
+                settings = {}
+        if not isinstance(settings, dict):
+            continue
+        for client in settings.get('clients') or []:
+            if not isinstance(client, dict):
+                continue
+            email = (client.get('email') or '').strip()
+            if not email:
+                continue
+            # Prefer enabled=true if the same email appears in multiple inbounds
+            prev = clients_by_email.get(email)
+            if prev and prev.get('enable') and not client.get('enable', True):
+                continue
+            clients_by_email[email] = client
+    return list(clients_by_email.values())
+
+
+async def _xui_fetch_clients(client: httpx.AsyncClient, base_url: str, headers: dict) -> list:
+    """Fetch clients from 3x-ui, supporting new and legacy API shapes."""
+    base = base_url.rstrip('/')
+
+    # Newer panels: dedicated clients list
+    for path in ('/panel/api/clients/list', '/panel/api/inbounds/list'):
+        try:
+            resp = await client.get(base + path, headers=headers)
+        except Exception:
+            continue
+        if resp.status_code != 200:
+            continue
+        try:
+            payload = resp.json()
+        except Exception:
+            continue
+        if not payload.get('success', True) and payload.get('success') is False:
+            continue
+        obj = payload.get('obj', payload.get('data'))
+        if path.endswith('/clients/list') and isinstance(obj, list):
+            # Deduplicate by email
+            by_email = {}
+            for c in obj:
+                if not isinstance(c, dict):
+                    continue
+                email = (c.get('email') or '').strip()
+                if email:
+                    by_email[email] = c
+            if by_email:
+                return list(by_email.values())
+        if path.endswith('/inbounds/list'):
+            clients = _xui_parse_clients_from_inbounds(obj if isinstance(obj, list) else [])
+            if clients:
+                return clients
+
+    # Legacy: POST /panel/api/inbounds/list (and /panel/inbound/list)
+    for path in ('/panel/api/inbounds/list', '/panel/inbound/list'):
+        try:
+            resp = await client.post(base + path, headers=headers)
+        except Exception:
+            continue
+        if resp.status_code != 200:
+            continue
+        try:
+            payload = resp.json()
+        except Exception:
+            continue
+        obj = payload.get('obj', payload.get('data'))
+        clients = _xui_parse_clients_from_inbounds(obj if isinstance(obj, list) else [])
+        if clients:
+            return clients
+
+    return []
+
+
+async def sync_users_with_xui(data: dict):
+    """Import / update panel users from an existing 3x-ui instance."""
+    settings = data.get('settings', {}).get('sync', {})
+    if not settings.get('xui_sync_users'):
+        return 0, "3x-ui synchronization is disabled in settings"
+
+    url = (settings.get('xui_url') or '').strip()
+    if not url:
+        return 0, "3x-ui URL not configured"
+
+    api_token = (settings.get('xui_api_token') or '').strip()
+    username = (settings.get('xui_username') or '').strip()
+    password = settings.get('xui_password') or ''
+
+    try:
+        async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
+            headers = {'Accept': 'application/json'}
+            if api_token:
+                headers['Authorization'] = f'Bearer {api_token}'
+            else:
+                if not username or not password:
+                    return 0, "Provide 3x-ui API token or username/password"
+                login_resp = await client.post(
+                    url.rstrip('/') + '/login',
+                    json={'username': username, 'password': password},
+                    headers=headers,
+                )
+                if login_resp.status_code != 200:
+                    return 0, f"3x-ui login failed: {login_resp.status_code} {login_resp.text[:200]}"
+                try:
+                    login_body = login_resp.json()
+                except Exception:
+                    login_body = {}
+                if login_body.get('success') is False:
+                    return 0, f"3x-ui login failed: {login_body.get('msg', 'unknown error')}"
+
+            xui_clients = await _xui_fetch_clients(client, url, headers)
+            if not xui_clients:
+                return 0, "No clients found in 3x-ui (check URL/credentials/API path)"
+
+            xui_emails = {(c.get('email') or '').strip() for c in xui_clients if (c.get('email') or '').strip()}
+
+            # 1. Delete local users that were synced from 3x-ui but no longer exist there
+            to_delete_ids = []
+            for u in data['users']:
+                email = (u.get('xui_email') or '').strip()
+                if email and email not in xui_emails:
+                    to_delete_ids.append(u['id'])
+            if to_delete_ids:
+                logger.info(f"Removing {len(to_delete_ids)} users deleted in 3x-ui")
+                await perform_mass_operations(delete_uids=to_delete_ids)
+
+            synced_count = 0
+            to_toggle = []
+            to_create_conns = []
+
+            for xc in xui_clients:
+                email = (xc.get('email') or '').strip()
+                if not email:
+                    continue
+
+                data = load_data()
+                local_u = next((u for u in data['users'] if (u.get('xui_email') or '') == email), None)
+                uname = _xui_sanitize_username(email)
+                if not local_u and uname:
+                    local_u = next((u for u in data['users'] if u['username'] == uname), None)
+
+                is_active = bool(xc.get('enable', True))
+                tg_id = xc.get('tgId') or xc.get('telegramId') or None
+                if tg_id is not None:
+                    tg_id = str(tg_id) if str(tg_id).strip() else None
+
+                # totalGB in 3x-ui is usually bytes despite the name
+                traffic_limit = int(xc.get('totalGB') or 0)
+                expiry_ms = int(xc.get('expiryTime') or 0)
+                expiration_date = None
+                if expiry_ms > 0:
+                    try:
+                        expiration_date = datetime.fromtimestamp(expiry_ms / 1000.0).isoformat()
+                    except Exception:
+                        expiration_date = None
+
+                if local_u:
+                    local_u['username'] = uname or local_u['username']
+                    local_u['telegramId'] = tg_id
+                    local_u['email'] = email if '@' in email else local_u.get('email')
+                    local_u['description'] = xc.get('comment') or local_u.get('description')
+                    local_u['xui_email'] = email
+                    local_u['traffic_limit'] = traffic_limit
+                    local_u['expiration_date'] = expiration_date
+                    if local_u.get('enabled', True) != is_active:
+                        to_toggle.append((local_u['id'], is_active))
+                    async with DATA_LOCK:
+                        current = load_data()
+                        idx = next((i for i, u in enumerate(current['users']) if u['id'] == local_u['id']), -1)
+                        if idx != -1:
+                            current['users'][idx] = local_u
+                            save_data(current)
+                    synced_count += 1
+                else:
+                    new_id = str(uuid.uuid4())
+                    # Avoid username collisions
+                    final_name = uname or f"xui_{new_id[:8]}"
+                    existing_names = {u['username'] for u in data['users']}
+                    if final_name in existing_names:
+                        final_name = f"{final_name}_{new_id[:6]}"
+                    new_user = {
+                        'id': new_id,
+                        'username': final_name,
+                        'password_hash': '',
+                        'role': 'user',
+                        'telegramId': tg_id,
+                        'email': email if '@' in email else None,
+                        'description': xc.get('comment'),
+                        'enabled': is_active,
+                        'created_at': datetime.now().isoformat(),
+                        'remnawave_uuid': None,
+                        'xui_email': email,
+                        'traffic_limit': traffic_limit,
+                        'traffic_used': 0,
+                        'traffic_total': 0,
+                        'traffic_reset_strategy': 'never',
+                        'last_reset_at': datetime.now().isoformat(),
+                        'expiration_date': expiration_date,
+                        'share_enabled': False,
+                        'share_token': secrets.token_urlsafe(16),
+                        'share_password_hash': None,
+                    }
+                    async with DATA_LOCK:
+                        current = load_data()
+                        current['users'].append(new_user)
+                        save_data(current)
+
+                    if settings.get('xui_create_conns'):
+                        sid = settings.get('xui_server_id')
+                        if sid is not None:
+                            to_create_conns.append({
+                                'user_id': new_id,
+                                'server_id': sid,
+                                'protocol': settings.get('xui_protocol', 'xray'),
+                                'name': f"{final_name}_vpn",
+                            })
+                    synced_count += 1
+
+            if to_toggle or to_create_conns:
+                logger.info(
+                    f"Executing mass ops for 3x-ui sync: toggle={len(to_toggle)}, create={len(to_create_conns)}"
+                )
+                await perform_mass_operations(toggle_uids=to_toggle, create_conns=to_create_conns)
+
+            return synced_count, "Successfully synchronized with 3x-ui"
+
+    except Exception as e:
+        logger.exception("3x-ui synchronization error")
+        return 0, f"Error: {str(e)}"
+
+
 def get_current_user(request: Request):
     user_id = request.session.get('user_id')
     if not user_id:
@@ -1513,6 +1763,15 @@ class SyncSettings(BaseModel):
     remnawave_create_conns: bool = False
     remnawave_server_id: int = 0
     remnawave_protocol: str = 'awg'
+    xui_sync: bool = False
+    xui_url: str = ''
+    xui_username: str = ''
+    xui_password: str = ''
+    xui_api_token: str = ''
+    xui_sync_users: bool = False
+    xui_create_conns: bool = False
+    xui_server_id: int = 0
+    xui_protocol: str = 'xray'
 
 class CaptchaSettings(BaseModel):
     enabled: bool = False
@@ -1646,6 +1905,9 @@ async def startup():
         if 'expiration_date' not in u:
             u['expiration_date'] = None
             migrated = True
+        if 'xui_email' not in u:
+            u['xui_email'] = None
+            migrated = True
             
         if migrated:
             changed = True
@@ -1811,6 +2073,15 @@ async def periodic_background_tasks():
                 logger.info(f"Background Remnawave sync finished: {count} users updated. {msg}")
             else:
                 logger.info("Background Remnawave sync skipped (disabled in settings)")
+
+            # --- 3. 3X-UI SYNC ---
+            logger.info("Starting background 3x-ui sync...")
+            data = load_data()
+            if data.get('settings', {}).get('sync', {}).get('xui_sync_users'):
+                count, msg = await sync_users_with_xui(data)
+                logger.info(f"Background 3x-ui sync finished: {count} users updated. {msg}")
+            else:
+                logger.info("Background 3x-ui sync skipped (disabled in settings)")
                 
         except Exception as e:
             logger.error(f"Error in periodic_background_tasks: {e}")
@@ -3150,7 +3421,10 @@ async def api_list_users(request: Request, search: str = '', page: int = 1, size
             'share_enabled': u.get('share_enabled', False),
             'share_token': u.get('share_token'),
             'has_share_password': bool(u.get('share_password_hash')),
-            'source': 'Remnawave' if u.get('remnawave_uuid') else 'Local'
+            'source': (
+                'Remnawave' if u.get('remnawave_uuid')
+                else ('3x-ui' if u.get('xui_email') else 'Local')
+            )
         })
     return {
         'users': users,
@@ -3191,6 +3465,7 @@ async def api_add_user(request: Request, req: AddUserRequest):
             'enabled': True,
             'created_at': datetime.now().isoformat(),
             'remnawave_uuid': None,
+            'xui_email': None,
             'share_enabled': False,
             'share_token': secrets.token_urlsafe(16),
             'share_password_hash': None,
@@ -3762,6 +4037,26 @@ async def api_sync_delete(request: Request):
     return {'status': 'success', 'count': len(to_delete_ids)}
 
 
+@app.post('/api/settings/xui_sync_now', tags=["Settings"])
+async def api_xui_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_xui(data)
+    return {'status': 'success', 'count': count, 'message': msg}
+
+
+@app.post('/api/settings/xui_sync_delete', tags=["Settings"])
+async def api_xui_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('xui_email')]
+    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):