test2/Amnezia-Web-Panel-main
Template
1
2
Fork 1
Code Issues Pull Requests Packages Projects Releases 18 Wiki Activity

Load inbounds from 3x-ui and use panel share links only.

Browse Source 
Restore inbound selection from the panel API and send a minimal addClient payload so locally invented fields do not break configs.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
orohi
2026-07-26 03:47:06 +03:00
co-authored by Cursor
parent 8dc090bd67
commit 2707e0af18
7 changed files with 267 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
managers/xui_api.py
+108 -79
View File
@@ -158,79 +158,113 @@ class XuiApi:
})
return result
async def get_inbound(self, inbound_id: int) -> dict:
inbound_id = int(inbound_id)
for inbound in await self.list_inbounds():
if isinstance(inbound, dict) and int(inbound.get('id') or 0) == inbound_id:
return inbound
raise XuiApiError(f'Inbound #{inbound_id} not found on 3x-ui')
@staticmethod
def _parse_inbound_settings(inbound: dict) -> dict:
settings_raw = inbound.get('settings') or '{}'
if isinstance(settings_raw, str):
try:
settings_obj = json.loads(settings_raw)
except Exception:
settings_obj = {}
else:
settings_obj = settings_raw if isinstance(settings_raw, dict) else {}
return settings_obj if isinstance(settings_obj, dict) else {}
def _client_template_from_inbound(self, inbound: dict) -> dict:
"""Copy only fields 3x-ui already uses on this inbound — do not invent extras."""
settings_obj = self._parse_inbound_settings(inbound)
clients = settings_obj.get('clients') or []
template = next((c for c in clients if isinstance(c, dict)), None) or {}
flow = template.get('flow')
if not isinstance(flow, str):
flow = ''
# If no existing clients, detect Reality → vision (required for those inbounds)
if not flow:
stream_raw = inbound.get('streamSettings') or '{}'
if isinstance(stream_raw, str):
try:
stream = json.loads(stream_raw)
except Exception:
stream = {}
else:
stream = stream_raw if isinstance(stream_raw, dict) else {}
security = (stream.get('security') or '').lower()
network = (stream.get('network') or '').lower()
if security == 'reality' and network in ('tcp', 'raw', ''):
flow = 'xtls-rprx-vision'
return {
'flow': flow,
'limitIp': int(template.get('limitIp') or 0),
'totalGB': 0,
'tgId': '' if template.get('tgId') in (None, 0) else str(template.get('tgId') or ''),
}
async def add_vless_client(
self,
*,
email: str,
inbound_id: int,
comment: str = '',
enable: bool = True,
total_gb: int = 0,
expiry_time: int = 0,
limit_ip: int = 0,
flow: str = '',
) -> dict:
"""Add client via 3x-ui addClient only. Minimal payload; flow taken from inbound."""
email = (email or '').strip()
if not email:
raise XuiApiError('Client email is required')
if not inbound_id:
raise XuiApiError('VLESS inbound id is required')
inbound = await self.get_inbound(int(inbound_id))
if (inbound.get('protocol') or '').lower() != 'vless':
raise XuiApiError(f'Inbound #{inbound_id} is not VLESS')
tmpl = self._client_template_from_inbound(inbound)
client_uuid = str(uuid.uuid4())
sub_id = _new_sub_id()
# Minimal client object — nothing beyond what 3x-ui expects for this inbound
client = {
'id': client_uuid,
'email': email,
'enable': enable,
'flow': flow or '',
'limitIp': limit_ip,
'totalGB': total_gb,
'expiryTime': expiry_time,
'tgId': 0,
'enable': bool(enable),
'flow': tmpl['flow'],
'limitIp': tmpl['limitIp'],
'totalGB': 0,
'expiryTime': int(expiry_time or 0),
'tgId': tmpl['tgId'],
'subId': sub_id,
'comment': comment or '',
}
# Modern API
try:
await self._request(
'POST',
'/panel/api/clients/add',
json={'client': client, 'inboundIds': [int(inbound_id)]},
)
except XuiApiError as modern_err:
logger.info('Modern clients/add failed (%s), trying legacy addClient', modern_err)
# Legacy: settings must be a JSON-encoded string on many builds
settings_obj = {'clients': [client]}
try:
await self._request(
'POST',
'/panel/api/inbounds/addClient',
json={
'id': int(inbound_id),
'settings': json.dumps(settings_obj),
},
)
except XuiApiError:
# Some builds accept nested object
await self._request(
'POST',
'/panel/api/inbounds/addClient',
json={
'id': int(inbound_id),
'settings': settings_obj,
},
)
# Official path: addClient with settings as JSON string (object form breaks some builds)
await self._request(
'POST',
'/panel/api/inbounds/addClient',
json={
'id': int(inbound_id),
'settings': json.dumps({'clients': [client]}),
},
)
# Share links ONLY from the panel — never build vless:// here
links = await self.get_client_links(email)
if not links:
links = await self.get_sub_links(sub_id)
# Share URL is resolved by the caller via resolve_share_link (panel assigns link)
return {
'client_id': email,
'uuid': client_uuid,
'sub_id': sub_id,
'email': email,
'config': '',
'links': [],
'expiry_time': expiry_time,
'links': links,
'expiry_time': int(expiry_time or 0),
'inbound_id': int(inbound_id),
}
@@ -301,33 +335,38 @@ class XuiApi:
return ''
async def resolve_share_link(self, *, email: str, sub_id: str = '', configured_sub_base: str = '') -> dict:
"""Ask the panel for the shareable subscription / protocol link (do not invent config)."""
"""Use only links returned by 3x-ui. Do not synthesize protocol configs."""
email = (email or '').strip()
sub_id = (sub_id or '').strip()
links: list = []
if email:
links = await self.get_client_links(email)
if not links and sub_id:
links = await self.get_sub_links(sub_id)
# Prefer protocol share from panel; then HTTP(S) subscription URL from panel
panel_proto = next(
(u for u in links if isinstance(u, str) and u.startswith(('vless://', 'vmess://', 'trojan://', 'ss://'))),
'',
)
panel_http = next(
(u for u in links if isinstance(u, str) and u.startswith(('http://', 'https://'))),
'',
)
# Subscription URL: panel HTTP link first; optional configured base only as last resort
base = (configured_sub_base or '').strip().rstrip('/')
if not base:
if not panel_http and not base:
try:
base = (await self.get_panel_sub_base() or '').rstrip('/')
except Exception as e:
logger.warning('get_panel_sub_base failed: %s', e)
base = ''
links = []
if sub_id:
links = await self.get_sub_links(sub_id)
if not links and email:
links = await self.get_client_links(email)
# Prefer HTTP(S) link returned by 3x-ui; else build from panel/settings sub base + subId
panel_http = next(
(u for u in links if isinstance(u, str) and u.startswith(('http://', 'https://'))),
'',
)
subscription_url = panel_http or (f'{base}/{sub_id}' if base and sub_id else '')
share = subscription_url or next(
(u for u in links if isinstance(u, str) and u.startswith(('vless://', 'vmess://', 'trojan://', 'ss://'))),
'',
)
# Prefer panel-generated protocol link (real config); else subscription URL
share = panel_proto or subscription_url or (links[0] if links else '')
return {
'subscription_url': subscription_url,
'links': links,
@@ -335,6 +374,7 @@ class XuiApi:
'sub_base': base,
}
async def delete_client(self, email: str) -> None:
email = (email or '').strip()
if not email:
@@ -408,7 +448,7 @@ async def xui_create_vless_config(
expiry_time: int = 0,
panel_id: Optional[str] = None,
) -> dict:
"""Create a client via 3x-ui API; inbound + share link come from the panel response."""
"""Create client on selected 3x-ui inbound; return only links from the panel."""
scoped = _resolve_settings(settings, panel_id)
creds = _settings_creds(scoped)
inbound = inbound_id if inbound_id not in (None, 0, '0') else creds.get('inbound_id')
@@ -416,15 +456,10 @@ async def xui_create_vless_config(
inbound = int(inbound or 0)
except (TypeError, ValueError):
inbound = 0
if not inbound:
raise XuiApiError('Select a VLESS inbound from 3x-ui')
async with XuiApi.from_panel_settings(scoped) as api:
# Panel assigns inbound: use configured default, else first VLESS on that panel
if not inbound:
vless_inbounds = await api.list_vless_inbounds()
if not vless_inbounds:
raise XuiApiError('No VLESS inbound found on 3x-ui — create one in the panel first')
inbound = int(vless_inbounds[0]['id'])
base = ''.join(ch if ch.isalnum() or ch in '._-+@' else '_' for ch in (name or 'user').strip())
base = base[:48] or f'user_{secrets.token_hex(4)}'
email = base
@@ -434,7 +469,6 @@ async def xui_create_vless_config(
created = await api.add_vless_client(
email=email,
inbound_id=inbound,
comment=name or email,
expiry_time=int(expiry_time or 0),
)
break
@@ -447,24 +481,19 @@ async def xui_create_vless_config(
created = await api.add_vless_client(
email=email,
inbound_id=inbound,
comment=name or email,
expiry_time=int(expiry_time or 0),
)
# Ask the panel for the subscription / protocol link (do not invent config here)
share = await api.resolve_share_link(
email=created.get('email') or email,
sub_id=created.get('sub_id') or '',
configured_sub_base=creds.get('sub_url') or '',
)
subscription_url = share.get('subscription_url') or ''
panel_share = share.get('share') or ''
panel_links = share.get('links') or created.get('links') or []
created['subscription_url'] = subscription_url
created['links'] = panel_links
# What we show the user: panel subscription URL, else panel protocol link
created['config'] = subscription_url or panel_share or created.get('config') or ''
created['subscription_url'] = share.get('subscription_url') or ''
# Prefer real vless:// (etc.) from 3x-ui; subscription URL only as fallback
created['config'] = share.get('share') or ''
if not created['config'] and panel_links:
created['config'] = panel_links[0]
created['inbound_id'] = int(created.get('inbound_id') or inbound)