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

Allow re-copying invite configs and end-user server choice (v3.0.1).

Browse Source 
Invite pages keep created configs for repeat copy; guest/invite can pick a server when enabled.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
orohimaru2
2026-08-09 18:31:46 +03:00
co-authored by Cursor
parent abbd160dd8
commit 3ea638d9c5
10 changed files with 610 additions and 163 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+260 -56
View File
@@ -103,7 +103,7 @@ else:
application_path = os.path.dirname(__file__)
DATA_FILE = os.path.join(application_path, 'data.json') # legacy JSON; used only for one-shot import / export
CURRENT_VERSION = "v3.0.0"
CURRENT_VERSION = "v3.0.1"
RELEASES_REPO_URL = repo_url()
RELEASES_API_LATEST = api_latest_url()
BIN_DIR = os.environ.get('TUNNEL_BIN_DIR', os.path.join(application_path, 'bin'))
@@ -2401,6 +2401,7 @@ class GuestSettings(BaseModel):
create_server_id: int = 0
create_inbound_id: int = 0
create_xui_panel_id: str = ''
create_allow_server_choice: bool = True
class DonateMethodSettings(BaseModel):
@@ -2474,6 +2475,7 @@ class ShareAuthRequest(BaseModel):
class GuestCreateRequest(BaseModel):
name: str = 'Guest VPN'
server_id: Optional[int] = None
class InviteCreateRequest(BaseModel):
@@ -2488,6 +2490,7 @@ class InviteCreateRequest(BaseModel):
duration_days: int = 0 # client lifetime after redeem; 0 = no expiry
note: str = ''
enabled: bool = True
allow_server_choice: bool = True
class InviteUpdateRequest(BaseModel):
@@ -2504,10 +2507,12 @@ class InviteUpdateRequest(BaseModel):
note: Optional[str] = None
enabled: Optional[bool] = None
reset_used: bool = False
allow_server_choice: Optional[bool] = None
class InviteRedeemRequest(BaseModel):
name: str = 'Invite VPN'
server_id: Optional[int] = None
class TunnelStartRequest(BaseModel):
@@ -5705,6 +5710,7 @@ def _guest_settings(data: Optional[dict] = None) -> dict:
'create_server_id': 0,
'create_inbound_id': 0,
'create_xui_panel_id': '',
'create_allow_server_choice': True,
}
defaults.update(guest)
return defaults
@@ -5780,22 +5786,26 @@ async def api_guest_connections(token: str, request: Request):
data, guest, holder, err = _resolve_guest(token, request)
if err:
return err
allow_choice = bool(guest.get('create_allow_server_choice', True))
protocol = guest.get('create_protocol') or 'xui'
servers = _pickable_servers_for_protocol(data, protocol) if (
guest.get('allow_create') and allow_choice and protocol_base(protocol) != 'xui'
) else []
meta = {
'allow_create': bool(guest.get('allow_create')),
'create_protocol': protocol,
'allow_server_choice': bool(servers),
'default_server_id': int(guest.get('create_server_id') or 0),
'servers': servers,
}
if not holder:
return {
'connections': [],
'allow_create': bool(guest.get('allow_create')),
'create_protocol': guest.get('create_protocol') or 'xui',
}
return {'connections': [], **meta}
conns = [
_enrich_guest_conn(c, data)
for c in data.get('user_connections', [])
if c['user_id'] == holder['id']
]
return {
'connections': conns,
'allow_create': bool(guest.get('allow_create')),
'create_protocol': guest.get('create_protocol') or 'xui',
}
return {'connections': conns, **meta}
@app.post('/api/guest/{token}/config/{connection_id}', tags=["Guest"])
@@ -5817,33 +5827,7 @@ async def api_guest_config(token: str, connection_id: str, request: Request):
return JSONResponse({'error': 'Subscription expired'}, status_code=403)
if maybe_start_user_expiration(data, holder['id']):
save_data(data)
if protocol_base(conn.get('protocol', '')) == 'xui':
from managers.xui_api import xui_get_config
config = await xui_get_config(
data.get('settings', {}),
conn['client_id'],
panel_id=conn.get('xui_panel_id') or None,
)
vpn_link = config if str(config).startswith('http') else (generate_vpn_link(config) if config else '')
return {
'config': config,
'vpn_link': vpn_link,
'subscription_url': config if str(config).startswith('http') else '',
'expires_at': holder.get('expiration_date'),
}
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()
manager = get_protocol_manager(ssh, conn['protocol'])
config = _manager_call(manager, 'get_client_config', conn['protocol'], conn['client_id'], get_server_connect_host(server, conn['protocol']), port)
ssh.disconnect()
vpn_link = generate_vpn_link(config) if config else ''
return {'config': config, 'vpn_link': vpn_link, 'expires_at': holder.get('expiration_date')}
return await _fetch_connection_config_payload(data, conn, expires_at=holder.get('expiration_date'))
except Exception as e:
logger.exception("Error getting guest config")
return JSONResponse({'error': str(e)}, status_code=500)
@@ -5864,18 +5848,32 @@ async def api_guest_create(token: str, req: GuestCreateRequest, request: Request
name = (req.name or 'Guest VPN').strip() or 'Guest VPN'
# Unique-ish name to avoid collisions
name = f"{name}_{secrets.token_hex(3)}"
allow_choice = bool(guest.get('create_allow_server_choice', True))
try:
from managers.user_expiration import maybe_start_user_expiration, user_is_expired
if user_is_expired(holder):
return JSONResponse({'error': 'Subscription expired'}, status_code=403)
if protocol_base(protocol) == 'xui':
sid = 0
else:
try:
sid = _resolve_chosen_server_id(
data,
protocol=protocol,
default_server_id=int(guest.get('create_server_id') or 0),
requested_server_id=req.server_id,
allow_choice=allow_choice,
)
except ValueError as ve:
return JSONResponse({'error': str(ve)}, status_code=400)
if protocol_base(protocol) == 'aivpn':
from managers.aivpn_manager import resolve_provision_protocol
sid = int(guest.get('create_server_id') or 0)
if sid >= len(data['servers']):
return JSONResponse({'error': 'Guest server not found'}, status_code=400)
protocol = resolve_provision_protocol(data['servers'][sid], 'aivpn')
elif protocol_base(protocol) != 'xui':
protocol = _match_protocol_on_server(data['servers'][sid], protocol)
if protocol_base(protocol) == 'xui':
from managers.xui_api import xui_create_vless_config
@@ -5900,9 +5898,6 @@ async def api_guest_create(token: str, req: GuestCreateRequest, request: Request
sid = 0
protocol = 'xui'
else:
sid = int(guest.get('create_server_id') or 0)
if sid >= len(data['servers']):
return JSONResponse({'error': 'Guest server not found'}, status_code=400)
server = data['servers'][sid]
proto_info = server.get('protocols', {}).get(protocol, {})
port = proto_info.get('port', '55424')
@@ -5970,12 +5965,90 @@ async def api_guest_regenerate_token(request: Request):
# ======================== Invite links (limited config creation) ========================
def _invite_public_view(link: dict) -> dict:
def _server_supports_protocol(server: dict, protocol: str) -> bool:
"""True if server has an installed client VPN matching protocol (or any for aivpn)."""
base = protocol_base(protocol)
if base == 'xui':
return False
protocols = server.get('protocols') or {}
for key, info in protocols.items():
if not isinstance(info, dict) or not info.get('installed'):
continue
kb = protocol_base(key)
if kb not in CLIENT_VPN_BASES or kb == 'xui':
continue
if base == 'aivpn' or kb == base or key == protocol:
return True
return False
def _match_protocol_on_server(server: dict, protocol: str) -> str:
"""Map requested protocol to an installed key on this server (prefer exact)."""
protocols = server.get('protocols') or {}
if protocol in protocols and isinstance(protocols.get(protocol), dict) and protocols[protocol].get('installed'):
return protocol
base = protocol_base(protocol)
if base == 'aivpn':
return protocol
candidates = [
key for key, info in protocols.items()
if isinstance(info, dict) and info.get('installed') and protocol_base(key) == base
]
if not candidates:
return protocol
candidates.sort()
return candidates[0]
def _pickable_servers_for_protocol(data: dict, protocol: str) -> list:
"""Safe server list for end-user pickers (name/host/id only)."""
out = []
for idx, server in enumerate(data.get('servers') or []):
if not isinstance(server, dict):
continue
if not _server_supports_protocol(server, protocol):
continue
out.append({
'id': idx,
'name': server.get('name') or server.get('host') or f'Server {idx + 1}',
'host': server.get('host') or '',
})
return out
def _resolve_chosen_server_id(
data: dict,
*,
protocol: str,
default_server_id: int,
requested_server_id: Optional[int],
allow_choice: bool,
) -> int:
"""Pick server_id for guest/invite create; validate against installed protocols."""
servers = data.get('servers') or []
default_sid = int(default_server_id or 0)
if not allow_choice or requested_server_id is None:
sid = default_sid
else:
sid = int(requested_server_id)
if sid < 0 or sid >= len(servers):
raise ValueError('Server not found')
if protocol_base(protocol) != 'xui' and not _server_supports_protocol(servers[sid], protocol):
raise ValueError('Selected server does not have the required protocol')
return sid
def _invite_public_view(link: dict, data: Optional[dict] = None) -> dict:
max_uses = int(link.get('max_uses') or 0)
used = int(link.get('used_count') or 0)
remaining = None if max_uses <= 0 else max(0, max_uses - used)
exhausted = remaining is not None and remaining <= 0
duration_days = int(link.get('duration_days') or 0)
protocol = link.get('protocol') or 'awg'
allow_server_choice = bool(link.get('allow_server_choice', True)) and protocol_base(protocol) != 'xui'
servers = []
if data is not None and allow_server_choice:
servers = _pickable_servers_for_protocol(data, protocol)
return {
'id': link.get('id'),
'name': link.get('name') or 'Invite',
@@ -5988,8 +6061,10 @@ def _invite_public_view(link: dict) -> dict:
'expired': False,
'exhausted': exhausted,
'has_password': bool(link.get('password_hash')),
'protocol': link.get('protocol') or 'awg',
'protocol': protocol,
'server_id': int(link.get('server_id') or 0),
'allow_server_choice': allow_server_choice,
'servers': servers,
'duration_days': duration_days,
'user_id': link.get('user_id') or '',
'note': link.get('note') or '',
@@ -6008,6 +6083,58 @@ def _invite_auth_ok(link: dict, request: Request) -> bool:
return bool(request.session.get(f"invite_auth_{link.get('token')}"))
def _enrich_invite_conn(c: dict, data: dict) -> dict:
out = dict(c)
if protocol_base(out.get('protocol', '')) == 'xui':
out['server_name'] = '3x-ui'
else:
sid = int(out.get('server_id') or 0)
if sid < len(data.get('servers') or []):
srv = data['servers'][sid]
out['server_name'] = srv.get('name') or srv.get('host') or ''
else:
out['server_name'] = 'Unknown'
return out
async def _fetch_connection_config_payload(data: dict, conn: dict, expires_at: Optional[str] = None) -> dict:
"""Shared config fetch for guest/invite/share self-service surfaces."""
if protocol_base(conn.get('protocol', '')) == 'xui':
from managers.xui_api import xui_get_config
config = await xui_get_config(
data.get('settings', {}),
conn['client_id'],
panel_id=conn.get('xui_panel_id') or None,
)
vpn_link = config if str(config).startswith('http') else (generate_vpn_link(config) if config else '')
return {
'config': config,
'vpn_link': vpn_link,
'subscription_url': config if str(config).startswith('http') else '',
'expires_at': expires_at,
}
sid = int(conn.get('server_id') or 0)
if sid >= len(data.get('servers') or []):
raise RuntimeError('Server not found')
server = data['servers'][sid]
protocol = conn['protocol']
proto_info = server.get('protocols', {}).get(protocol, {})
port = proto_info.get('port', '55424')
ssh = get_ssh(server)
await asyncio.to_thread(ssh.connect)
try:
manager = get_protocol_manager(ssh, protocol)
config = await asyncio.to_thread(
_manager_call, manager, 'get_client_config',
protocol, conn['client_id'], get_server_connect_host(server, protocol), port,
)
finally:
await asyncio.to_thread(ssh.disconnect)
vpn_link = generate_vpn_link(config) if config else ''
return {'config': config, 'vpn_link': vpn_link, 'expires_at': expires_at}
def _expiry_ms_from_duration_days(duration_days: int) -> int:
"""3x-ui expiryTime is unix ms; 0 means no expiry. Starts now (on redeem)."""
days = int(duration_days or 0)
@@ -6176,11 +6303,12 @@ async def api_create_invite(request: Request, req: InviteCreateRequest):
'expires_at': None,
'duration_days': int(req.duration_days or 0),
'note': req.note or '',
'allow_server_choice': bool(req.allow_server_choice),
'created_at': datetime.now().isoformat(),
}
data.setdefault('invite_links', []).append(link)
save_data(data)
view = _invite_public_view(link)
view = _invite_public_view(link, data)
return {'status': 'success', 'invite': view, 'url': f"/invite/{link['token']}"}
@@ -6222,6 +6350,8 @@ async def api_update_invite(request: Request, invite_id: str, req: InviteUpdateR
link['note'] = req.note
if req.enabled is not None:
link['enabled'] = bool(req.enabled)
if req.allow_server_choice is not None:
link['allow_server_choice'] = bool(req.allow_server_choice)
if req.reset_used:
link['used_count'] = 0
if protocol_base(link.get('protocol') or 'xui') == 'xui':
@@ -6237,7 +6367,7 @@ async def api_update_invite(request: Request, invite_id: str, req: InviteUpdateR
if not int(link.get('xui_inbound_id') or 0):
return JSONResponse({'error': 'Select a VLESS inbound from 3x-ui'}, status_code=400)
save_data(data)
return {'status': 'success', 'invite': _invite_public_view(link)}
return {'status': 'success', 'invite': _invite_public_view(link, data)}
@app.delete('/api/invites/{invite_id}', tags=["Invites"])
@@ -6263,7 +6393,7 @@ async def invite_public_page(token: str, request: Request):
f"<h1>{_t('invite_not_found', lang)}</h1><p>{_t('invite_not_found_desc', lang)}</p>",
status_code=404,
)
view = _invite_public_view(link)
view = _invite_public_view(link, data)
need_password = bool(link.get('password_hash')) and not _invite_auth_ok(link, request)
return tpl(
request,
@@ -6298,7 +6428,7 @@ async def api_invite_info(token: str, request: Request):
return JSONResponse({'error': 'Not found'}, status_code=404)
if not _invite_auth_ok(link, request):
return JSONResponse({'error': 'Unauthorized'}, status_code=401)
view = _invite_public_view(link)
view = _invite_public_view(link, data)
# Don't leak admin note / ids beyond what's needed
return {
'name': view['name'],
@@ -6311,9 +6441,64 @@ async def api_invite_info(token: str, request: Request):
'max_uses': view['max_uses'],
'used_count': view['used_count'],
'protocol': view['protocol'],
'allow_server_choice': view['allow_server_choice'],
'server_id': view['server_id'],
'servers': view.get('servers') or [],
'duration_days': view['duration_days'],
}
@app.get('/api/invite/{token}/connections', tags=["Invites"])
async def api_invite_connections(token: str, request: Request):
"""List configs created via this invite (re-copy without consuming another use)."""
data = load_data()
link = _find_invite(data, token)
if not link:
return JSONResponse({'error': 'Not found'}, status_code=404)
if not _invite_auth_ok(link, request):
return JSONResponse({'error': 'Unauthorized'}, status_code=401)
invite_id = link.get('id')
holder_id = link.get('user_id') or ''
conns = [
_enrich_invite_conn(c, data)
for c in data.get('user_connections', [])
if c.get('invite_id') == invite_id and (not holder_id or c.get('user_id') == holder_id)
]
conns.sort(key=lambda c: c.get('created_at') or '', reverse=True)
return {
'connections': conns,
'invite': _invite_public_view(link, data),
}
@app.post('/api/invite/{token}/config/{connection_id}', tags=["Invites"])
async def api_invite_config(token: str, connection_id: str, request: Request):
data = load_data()
link = _find_invite(data, token)
if not link:
return JSONResponse({'error': 'Not found'}, status_code=404)
if not _invite_auth_ok(link, request):
return JSONResponse({'error': 'Unauthorized'}, status_code=401)
invite_id = link.get('id')
holder_id = link.get('user_id') or ''
conn = next(
(
c for c in data.get('user_connections', [])
if c.get('id') == connection_id
and c.get('invite_id') == invite_id
and (not holder_id or c.get('user_id') == holder_id)
),
None,
)
if not conn:
return JSONResponse({'error': 'Not found'}, status_code=404)
try:
return await _fetch_connection_config_payload(data, conn, expires_at=conn.get('expires_at'))
except Exception as e:
logger.exception('Error getting invite config')
return JSONResponse({'error': str(e)}, status_code=500)
@app.post('/api/invite/{token}/create', tags=["Invites"])
async def api_invite_create_config(token: str, req: InviteRedeemRequest, request: Request):
data = load_data()
@@ -6323,7 +6508,7 @@ async def api_invite_create_config(token: str, req: InviteRedeemRequest, request
if not _invite_auth_ok(link, request):
return JSONResponse({'error': 'Unauthorized'}, status_code=401)
view = _invite_public_view(link)
view = _invite_public_view(link, data)
if not view['available']:
if view['expired']:
return JSONResponse({'error': 'Invite link expired'}, status_code=403)
@@ -6335,13 +6520,29 @@ async def api_invite_create_config(token: str, req: InviteRedeemRequest, request
if not holder_id or not any(u['id'] == holder_id for u in data['users']):
return JSONResponse({'error': 'Invite holder user is not configured'}, status_code=400)
protocol = link.get('protocol') or 'xui'
allow_choice = bool(link.get('allow_server_choice', True))
try:
if protocol_base(protocol) == 'xui':
sid = int(link.get('server_id') or 0)
else:
sid = _resolve_chosen_server_id(
data,
protocol=protocol,
default_server_id=int(link.get('server_id') or 0),
requested_server_id=req.server_id,
allow_choice=allow_choice,
)
except ValueError as ve:
return JSONResponse({'error': str(ve)}, status_code=400)
# Reserve a use slot under lock
async with DATA_LOCK:
data = load_data()
link = _find_invite(data, token)
if not link:
return JSONResponse({'error': 'Not found'}, status_code=404)
view = _invite_public_view(link)
view = _invite_public_view(link, data)
if not view['available']:
return JSONResponse({'error': 'Invite link is no longer available'}, status_code=403)
link['used_count'] = int(link.get('used_count') or 0) + 1
@@ -6352,10 +6553,11 @@ async def api_invite_create_config(token: str, req: InviteRedeemRequest, request
try:
data = load_data()
protocol = link.get('protocol') or 'xui'
sid = int(link.get('server_id') or 0)
if protocol_base(protocol) == 'aivpn' and sid < len(data.get('servers') or []):
from managers.aivpn_manager import resolve_provision_protocol
protocol = resolve_provision_protocol(data['servers'][sid], 'aivpn')
elif protocol_base(protocol) != 'xui' and sid < len(data.get('servers') or []):
protocol = _match_protocol_on_server(data['servers'][sid], protocol)
created = await _create_config_for_protocol(
data,
protocol=protocol,
@@ -6372,16 +6574,18 @@ async def api_invite_create_config(token: str, req: InviteRedeemRequest, request
'protocol': created['protocol'],
'client_id': created['client_id'],
'name': name,
'invite_id': link.get('id'),
'xui_panel_id': created.get('xui_panel_id') or link.get('xui_panel_id') or '',
'created_at': datetime.now().isoformat(),
}
if created.get('expires_at'):
conn['expires_at'] = created['expires_at']
async with DATA_LOCK:
data = load_data()
data.setdefault('user_connections', []).append(conn)
save_data(data)
config = created.get('config') or ''
subscription_url = created.get('subscription_url') or ''
# For subscription URLs, vpn_link is the same shareable string
vpn_link = subscription_url or (generate_vpn_link(config) if config else '')
data = load_data()
link = _find_invite(data, token) or link
@@ -6391,8 +6595,8 @@ async def api_invite_create_config(token: str, req: InviteRedeemRequest, request
'subscription_url': subscription_url,
'vpn_link': vpn_link,
'expires_at': created.get('expires_at'),
'connection': conn,
'invite': _invite_public_view(link),
'connection': _enrich_invite_conn(conn, data),
'invite': _invite_public_view(link, data),
}
except Exception as e:
# Roll back reserved use