366 lines
13 KiB
Python
Executable File
366 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""3x-ui 命令行封装：登录（含 CSRF token 处理）、建入站、列入站与客户端。
|
||
|
||
3x-ui 的登录接口缺 X-CSRF-Token 会返回 403，而 token 不在 cookie 里，
|
||
而在登录页 HTML 的 <meta name="csrf-token" content="..."> 中。本脚本自动处理。
|
||
|
||
用法：
|
||
python3 3xui-api.py --base http://127.0.0.1:2053 --user admin --password 'pw' list
|
||
python3 3xui-api.py --base http://127.0.0.1:2053 --user admin --password 'pw' \
|
||
add --remark <入站名> --port <端口> --path /<路径> --uuid <UUID>
|
||
python3 3xui-api.py --base ... --user ... --password ... ping
|
||
"""
|
||
|
||
import argparse
|
||
import http.cookiejar
|
||
import json
|
||
import re
|
||
import sys
|
||
import urllib.parse
|
||
import urllib.request
|
||
import uuid as uuidlib
|
||
|
||
|
||
def opener_with_cookies():
|
||
jar = http.cookiejar.CookieJar()
|
||
return urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
|
||
|
||
|
||
def http_get(op, url):
|
||
req = urllib.request.Request(url, method="GET")
|
||
with op.open(req, timeout=20) as r:
|
||
return r.read().decode("utf-8", "replace")
|
||
|
||
|
||
def http_post(op, url, form=None, js=None, headers=None):
|
||
h = dict(headers or {})
|
||
if js is not None:
|
||
body = json.dumps(js).encode("utf-8")
|
||
h["Content-Type"] = "application/json"
|
||
else:
|
||
body = urllib.parse.urlencode(form or {}).encode("utf-8")
|
||
h["Content-Type"] = "application/x-www-form-urlencoded"
|
||
req = urllib.request.Request(url, data=body, headers=h, method="POST")
|
||
with op.open(req, timeout=20) as r:
|
||
return r.read().decode("utf-8", "replace")
|
||
|
||
|
||
def extract_csrf(html):
|
||
for pat in (
|
||
r'name="csrf-token"[^>]*content="([^"]+)"',
|
||
r'content="([^"]+)"[^>]*name="csrf-token"',
|
||
r'csrf-token"\s+content="([^"]+)"',
|
||
):
|
||
m = re.search(pat, html)
|
||
if m:
|
||
return m.group(1)
|
||
return ""
|
||
|
||
|
||
def login(base, user, password):
|
||
op = opener_with_cookies()
|
||
html = http_get(op, base + "/")
|
||
token = extract_csrf(html)
|
||
if not token:
|
||
sys.exit("登录页未找到 csrf-token，检查面板路径与版本")
|
||
raw = http_post(
|
||
op,
|
||
base + "/login",
|
||
form={"username": user, "password": password},
|
||
headers={"Origin": base, "Referer": base + "/", "X-CSRF-Token": token},
|
||
)
|
||
try:
|
||
data = json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
sys.exit("登录返回非 JSON：%s" % raw[:200])
|
||
if not data.get("success"):
|
||
sys.exit("登录失败：%s" % data.get("msg"))
|
||
return op, token
|
||
|
||
|
||
def api_headers(token):
|
||
return {"X-CSRF-Token": token, "Origin": "", "Referer": ""}
|
||
|
||
|
||
def build_inbound(remark, port, path, uuid, listen="127.0.0.1", email=None):
|
||
email = email or remark
|
||
uuid = uuid or str(uuidlib.uuid4())
|
||
return {
|
||
"up": 0,
|
||
"down": 0,
|
||
"total": 0,
|
||
"remark": remark,
|
||
"enable": True,
|
||
"expiryTime": 0,
|
||
"listen": listen,
|
||
"port": int(port),
|
||
"protocol": "vless",
|
||
"settings": json.dumps(
|
||
{
|
||
"clients": [
|
||
{"id": uuid, "flow": "", "email": email, "enable": True}
|
||
],
|
||
"decryption": "none",
|
||
"fallbacks": [],
|
||
}
|
||
),
|
||
"streamSettings": json.dumps(
|
||
{"network": "ws", "security": "none", "wsSettings": {"path": path}}
|
||
),
|
||
"sniffing": json.dumps({"enabled": True, "destOverride": ["http", "tls"]}),
|
||
"tag": remark,
|
||
}
|
||
|
||
|
||
def as_obj(v):
|
||
if isinstance(v, (dict, list)):
|
||
return v
|
||
try:
|
||
return json.loads(v or "{}")
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
def cmd_ping(op, token, base):
|
||
raw = http_get(op, base + "/panel/api/inbounds/list")
|
||
data = json.loads(raw)
|
||
print("登录有效，入站数 = %d" % len(data.get("obj") or []))
|
||
|
||
|
||
def cmd_list(op, token, base):
|
||
raw = http_get(op, base + "/panel/api/inbounds/list")
|
||
data = json.loads(raw)
|
||
items = data.get("obj") or []
|
||
if not items:
|
||
print("当前没有入站")
|
||
return
|
||
for ib in items:
|
||
st = as_obj(ib.get("streamSettings"))
|
||
se = as_obj(ib.get("settings"))
|
||
path = (st.get("wsSettings") or {}).get("path")
|
||
print(
|
||
"入站 id=%s 备注=%s 协议=%s 监听=%s:%s WS=%s"
|
||
% (ib.get("id"), ib.get("remark"), ib.get("protocol"),
|
||
ib.get("listen"), ib.get("port"), path)
|
||
)
|
||
for c in se.get("clients") or []:
|
||
print(" 客户端 标识=%s UUID=%s" % (c.get("email"), c.get("id")))
|
||
for cs in ib.get("clientStats") or []:
|
||
print(
|
||
" 统计 标识=%s 上行=%s 下行=%s 限额=%s 启用=%s"
|
||
% (cs.get("email"), cs.get("up"), cs.get("down"),
|
||
cs.get("total"), cs.get("enable"))
|
||
)
|
||
|
||
|
||
def cmd_add(op, token, base, args):
|
||
payload = build_inbound(
|
||
args.remark, args.port, args.path, args.uuid, args.listen, args.email
|
||
)
|
||
raw = http_post(
|
||
op,
|
||
base + "/panel/api/inbounds/add",
|
||
js=payload,
|
||
headers={"X-CSRF-Token": token, "Origin": base,
|
||
"Referer": base + "/panel/inbounds"},
|
||
)
|
||
data = json.loads(raw)
|
||
if not data.get("success"):
|
||
sys.exit("建入站失败：%s" % data.get("msg"))
|
||
obj = data.get("obj") or {}
|
||
# 服务端可能自行生成 UUID，回读一次保证打印的就是实际的
|
||
ib, uid = find_inbound_uuid(fetch_inbounds(op, base), args.remark, args.email or args.remark)
|
||
print("已创建 id=%s 备注=%s 端口=%s 路径=%s UUID=%s"
|
||
% (obj.get("id") or (ib or {}).get("id"), args.remark, args.port, args.path, uid or "(回读失败)"))
|
||
|
||
|
||
def fetch_inbounds(op, base):
|
||
raw = http_get(op, base + "/panel/api/inbounds/list")
|
||
return json.loads(raw).get("obj") or []
|
||
|
||
|
||
def save_inbound(op, token, base, ib):
|
||
raw = http_post(
|
||
op,
|
||
base + "/panel/api/inbounds/update/%s" % ib.get("id"),
|
||
js=ib,
|
||
headers={"X-CSRF-Token": token, "Origin": base,
|
||
"Referer": base + "/panel/inbounds"},
|
||
)
|
||
data = json.loads(raw)
|
||
if not data.get("success"):
|
||
sys.exit("更新入站 %s 失败：%s" % (ib.get("remark"), data.get("msg")))
|
||
|
||
|
||
def api_post(op, token, base, path, js, referer="/panel/clients"):
|
||
raw = http_post(
|
||
op, base + path, js=js,
|
||
headers={"X-CSRF-Token": token, "Origin": base, "Referer": base + referer},
|
||
)
|
||
try:
|
||
return json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
sys.exit("接口 %s 返回非 JSON：%s" % (path, raw[:200]))
|
||
|
||
|
||
def target_inbound_ids(items, only):
|
||
ids = [ib.get("id") for ib in items
|
||
if not only or ib.get("remark") == only]
|
||
if not ids:
|
||
sys.exit("没有匹配的入站（检查 --inbound 的备注名）")
|
||
return ids
|
||
|
||
|
||
def cmd_add_user(op, token, base, args):
|
||
items = fetch_inbounds(op, base)
|
||
if not items:
|
||
sys.exit("没有可用的入站")
|
||
ids = target_inbound_ids(items, args.inbound)
|
||
new_uuid = args.uuid or str(uuidlib.uuid4())
|
||
payload = {
|
||
"client": {"email": args.email, "uuid": new_uuid, "enable": True},
|
||
"inboundIds": ids,
|
||
}
|
||
data = api_post(op, token, base, "/panel/api/clients/add", payload)
|
||
if not data.get("success"):
|
||
sys.exit("建号失败：%s" % data.get("msg"))
|
||
# v3 的服务端会自行生成 UUID，这里回读真实值，避免打印出来的和实际不一致
|
||
actual = new_uuid
|
||
for ib in fetch_inbounds(op, base):
|
||
for c in (as_obj(ib.get("settings")).get("clients") or []):
|
||
if c.get("email") == args.email:
|
||
actual = c.get("id") or actual
|
||
break
|
||
names = [ib.get("remark") for ib in items if ib.get("id") in ids]
|
||
print("账号 %s 已创建，UUID = %s" % (args.email, actual))
|
||
print("已挂到入站：" + "、".join(names))
|
||
|
||
|
||
def find_inbound_uuid(items, remark, email):
|
||
for ib in items:
|
||
if remark and ib.get("remark") != remark:
|
||
continue
|
||
for c in (as_obj(ib.get("settings")).get("clients") or []):
|
||
if c.get("email") == email:
|
||
return ib, (c.get("id") or "")
|
||
return None, ""
|
||
|
||
|
||
def cmd_uuid(op, token, base, args):
|
||
ib, uid = find_inbound_uuid(fetch_inbounds(op, base), args.inbound, args.email)
|
||
if not ib or not uid:
|
||
sys.exit("没找到 %s 在入站 %s 里的客户端" % (args.email, args.inbound))
|
||
print(uid)
|
||
|
||
|
||
def cmd_link(op, token, base, args):
|
||
ib, uid = find_inbound_uuid(fetch_inbounds(op, base), args.inbound, args.email)
|
||
if not ib or not uid:
|
||
sys.exit("没找到 %s 在入站 %s 里的客户端" % (args.email, args.inbound))
|
||
path = (as_obj(ib.get("streamSettings")).get("wsSettings") or {}).get("path") or "/"
|
||
query = urllib.parse.urlencode(
|
||
[
|
||
("encryption", "none"),
|
||
("security", "tls"),
|
||
("sni", args.domain),
|
||
("alpn", "http/1.1"),
|
||
("type", "ws"),
|
||
("host", args.domain),
|
||
("path", path),
|
||
("fp", "chrome"),
|
||
],
|
||
quote_via=urllib.parse.quote,
|
||
)
|
||
frag = args.fragment or ib.get("remark") or "link"
|
||
print("vless://%s@%s:%s?%s#%s" % (uid, args.domain, args.port, query, frag))
|
||
|
||
|
||
def cmd_del_inbound(op, token, base, args):
|
||
items = [ib for ib in fetch_inbounds(op, base) if ib.get("remark") == args.inbound]
|
||
if not items:
|
||
sys.exit("没有备注为 %s 的入站" % args.inbound)
|
||
for ib in items:
|
||
data = api_post(op, token, base, "/panel/api/inbounds/del/%s" % ib.get("id"), {},
|
||
referer="/panel/inbounds")
|
||
if not data.get("success"):
|
||
sys.exit("删除入站 %s 失败：%s" % (ib.get("remark"), data.get("msg")))
|
||
print("已删除入站 id=%s 备注=%s 端口=%s" % (ib.get("id"), ib.get("remark"), ib.get("port")))
|
||
|
||
|
||
def cmd_del_user(op, token, base, args):
|
||
data = api_post(
|
||
op, token, base,
|
||
"/panel/api/clients/del/" + urllib.parse.quote(args.email),
|
||
{},
|
||
)
|
||
if not data.get("success"):
|
||
sys.exit("销号失败：%s" % data.get("msg"))
|
||
print("账号 %s 已删除（从所有入站一并摘除）" % args.email)
|
||
|
||
|
||
def main():
|
||
p = argparse.ArgumentParser(description="3x-ui 命令行封装")
|
||
p.add_argument("--base", default="http://127.0.0.1:2053")
|
||
p.add_argument("--user", default="admin")
|
||
p.add_argument("--password", required=True)
|
||
sub = p.add_subparsers(dest="cmd")
|
||
|
||
sub.add_parser("ping", help="验证登录是否可用")
|
||
sub.add_parser("list", help="列出入站与客户端")
|
||
|
||
a = sub.add_parser("add", help="创建 VLESS+WS 入站")
|
||
a.add_argument("--remark", required=True)
|
||
a.add_argument("--port", required=True, type=int)
|
||
a.add_argument("--path", required=True)
|
||
a.add_argument("--uuid", default=None, help="不填则自动生成")
|
||
a.add_argument("--listen", default="127.0.0.1")
|
||
a.add_argument("--email", default=None)
|
||
|
||
u = sub.add_parser("del-inbound", help="按备注删掉整条入站")
|
||
u.add_argument("--inbound", required=True)
|
||
|
||
u = sub.add_parser("add-user", help="给所有入站加同一个用户，一人两条链路")
|
||
u.add_argument("--email", required=True)
|
||
u.add_argument("--uuid", default=None)
|
||
u.add_argument("--inbound", default=None, help="只加到指定备注的入站")
|
||
|
||
d = sub.add_parser("del-user", help="从所有入站移除该用户")
|
||
d.add_argument("--email", required=True)
|
||
d.add_argument("--inbound", default=None)
|
||
|
||
g = sub.add_parser("uuid", help="打印某入站某账号的真实 UUID")
|
||
g.add_argument("--inbound", required=True)
|
||
g.add_argument("--email", required=True)
|
||
|
||
k = sub.add_parser("link", help="直接拼出可导入客户端的 vless 链接")
|
||
k.add_argument("--inbound", required=True, help="入站备注名")
|
||
k.add_argument("--email", required=True, help="账号标识")
|
||
k.add_argument("--domain", required=True)
|
||
k.add_argument("--port", default="443")
|
||
k.add_argument("--fragment", default=None, help="客户端里显示的名字")
|
||
|
||
args = p.parse_args()
|
||
if not args.cmd:
|
||
p.error("缺少子命令：ping / list / add")
|
||
op, token = login(args.base, args.user, args.password)
|
||
if args.cmd == "ping":
|
||
cmd_ping(op, token, args.base)
|
||
elif args.cmd == "list":
|
||
cmd_list(op, token, args.base)
|
||
elif args.cmd == "add":
|
||
cmd_add(op, token, args.base, args)
|
||
elif args.cmd == "add-user":
|
||
cmd_add_user(op, token, args.base, args)
|
||
elif args.cmd == "del-user":
|
||
cmd_del_user(op, token, args.base, args)
|
||
elif args.cmd == "del-inbound":
|
||
cmd_del_inbound(op, token, args.base, args)
|
||
elif args.cmd == "uuid":
|
||
cmd_uuid(op, token, args.base, args)
|
||
elif args.cmd == "link":
|
||
cmd_link(op, token, args.base, args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|