workbuddy_skills/dual-link-proxy/scripts/3xui-api.py

#!/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()