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


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
    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 {}
    print("已创建 id=%s 备注=%s 端口=%s 路径=%s"
          % (obj.get("id"), args.remark, args.port, args.path))


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", required=True)
    a.add_argument("--listen", default="127.0.0.1")
    a.add_argument("--email", default=None)

    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)


if __name__ == "__main__":
    main()