feat(dual-link-proxy): 新增双链路代理搭建与运维技能（已脱敏）
把「在已跑 nginx 的 VPS 上从零搭起橙云 + 灰云双链路、并用 3x-ui 管多用户」 的全过程沉淀成技能：五阶段工作流、五条硬性前提、三条易踩反的判据、19 条避坑。 - references/01 前置取证与选型：泛解析不等通、端口取证、nginx 约束下的选型 - references/02 服务端落地：3x-ui 与 sing-box 两条路线，含包名与 2096 冲突 - references/03 nginx 与证书：80 → 签证书 → 443 的顺序、AEAD 套件、口令文件位置 - references/04 面板与多用户：3x-ui 三层模型、加人步骤、CSRF 登录法 - references/05 避坑清单：19 条，正向写做法、括号里附踩过的坑 - scripts/verify-links.sh 与 scripts/3xui-api.py：两个脚本均对真实环境跑通 正文已脱敏，域名/端口/口令/UUID/主机 全部改为占位符。
This commit is contained in:
admin committed
1 parent
310a964a94
commit
a9d2adaf72
9 files changed
+759
No files matched your search
@@ -0,0 +1,204 @@
|
||||
#!/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()
|
||||
Reference in new issue
Block a user