新增 Clash/mihomo 节点输出: sbp user show --clash

This commit is contained in:
typist committed 2026-11-01 06:58:42 +08:00
1 parent 79fe594385
commit f9874454f0
16 files changed
+318 -14

No files matched your search

+17 -3
View File
@@ -101,6 +101,19 @@ sing-box run -c client.json
客户端配置包含 `127.0.0.1:1080` 的 mixed 入站. 它同时提供 SOCKS5 和 HTTP 代理. 客户端配置包含 `127.0.0.1:1080` 的 mixed 入站. 它同时提供 SOCKS5 和 HTTP 代理.
### Clash / mihomo
`--clash` 输出可合并的 `proxies:` 块:
```bash
sbp user show phone --clash
sbp user show phone --protocol trojan --clash
```
把输出粘贴到 `clash.yaml` 的 `proxies:` 下, 再在 `proxy-groups` 里引用节点名.
自签证书时, hysteria2 用 `fingerprint` 固定证书, 其余协议用 `skip-cert-verify: true`.
anytls 需要 mihomo v1.19.3 或更高版本.
### 服务与设置 ### 服务与设置
```bash ```bash
@@ -123,7 +136,7 @@ TUI 提供这些页面:
- Dashboard: 状态总览, 启动, 停止, 重启. - Dashboard: 状态总览, 启动, 停止, 重启.
- Protocols: 启用, 停用, 改端口, 改选项. - Protocols: 启用, 停用, 改端口, 改选项.
- Users: 增加, 删除, 查看分享链接和客户端配置. - Users: 增加, 删除, 查看分享链接, Clash 配置和客户端配置.
- Settings: 修改 host 和 SNI, 更换证书, 升级, 重装, 卸载. - Settings: 修改 host 和 SNI, 更换证书, 升级, 重装, 卸载.
- Logs: 查看 journal. - Logs: 查看 journal.
@@ -178,13 +191,14 @@ src/sbp/
commands/ 每个命令组一个模块 commands/ 每个命令组一个模块
state.py 状态模型, JSON 读写, 迁移 state.py 状态模型, JSON 读写, 迁移
config.py 生成 sing-box 服务端和客户端配置 config.py 生成 sing-box 服务端和客户端配置
share.py 生成分享链接 share.py 生成分享链接和 Clash 节点
clash.py Clash (mihomo) YAML 输出
protocols/ 每个协议一个模块 protocols/ 每个协议一个模块
ops/ 业务操作, CLI 和 TUI 共用 ops/ 业务操作, CLI 和 TUI 共用
tui/ curses 界面 tui/ curses 界面
``` ```
协议模块只做三件事: 生成 inbound, 生成分享链接, 生成客户端 outbound. 协议模块只做四件事: 生成 inbound, 分享链接, 客户端 outbound, Clash 节点.
它不读写文件, 不操作服务, 不访问网络. 新增协议只需要在 `protocols/` 增加一个模块, 它不读写文件, 不操作服务, 不访问网络. 新增协议只需要在 `protocols/` 增加一个模块,
并在 `protocols/__init__.py` 注册. 并在 `protocols/__init__.py` 注册.
+59
View File
@@ -0,0 +1,59 @@
"""Clash (mihomo) proxy YAML output.
The protocols own the field mapping; this module only formats the result.
It writes the small YAML subset that Clash configs use, so sbp needs no
YAML dependency.
"""
from __future__ import annotations
import json
import re
_PLAIN = re.compile(r"^[A-Za-z0-9_./-]+$")
_RESERVED = {"true", "false", "null", "yes", "no", "on", "off", "~"}
def cert_fingerprint(cert_sha256: str) -> str:
"""Format a hex SHA-256 as Clash wants it: colon-separated uppercase bytes."""
text = cert_sha256.strip().lower()
if len(text) != 64 or any(char not in "0123456789abcdef" for char in text):
raise ValueError(f"not a hex SHA-256: {cert_sha256!r}")
return ":".join(text[index:index + 2].upper() for index in range(0, 64, 2))
def _scalar(value: object) -> str:
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, int):
return str(value)
text = str(value)
if _PLAIN.match(text) and text.lower() not in _RESERVED:
return text
return json.dumps(text)
def _mapping(items: list[tuple[str, object]], indent: int) -> list[str]:
pad = " " * indent
lines: list[str] = []
for key, value in items:
if isinstance(value, dict):
lines.append(f"{pad}{key}:")
lines.extend(_mapping(list(value.items()), indent + 2))
elif isinstance(value, list):
lines.append(f"{pad}{key}:")
lines.extend(f"{pad} - {_scalar(item)}" for item in value)
else:
lines.append(f"{pad}{key}: {_scalar(value)}")
return lines
def proxies_yaml(proxies: list[dict]) -> str:
"""Render a mergeable `proxies:` block."""
lines = ["proxies:"]
for proxy in proxies:
entries = list(proxy.items())
first_key, first_value = entries[0]
lines.append(f" - {first_key}: {_scalar(first_value)}")
lines.extend(_mapping(entries[1:], 4))
return "\n".join(lines) + "\n"
+4 -1
View File
@@ -22,6 +22,7 @@ def register(sub) -> None:
show.add_argument("--protocol", choices=protocols.keys()) show.add_argument("--protocol", choices=protocols.keys())
show.add_argument("--uri", action="store_true", help="print the share URI only") show.add_argument("--uri", action="store_true", help="print the share URI only")
show.add_argument("--json", action="store_true", help="print the sing-box client config") show.add_argument("--json", action="store_true", help="print the sing-box client config")
show.add_argument("--clash", action="store_true", help="print a Clash proxies block")
def run(args: argparse.Namespace, paths: path_module.Paths) -> int: def run(args: argparse.Namespace, paths: path_module.Paths) -> int:
@@ -43,7 +44,9 @@ def run(args: argparse.Namespace, paths: path_module.Paths) -> int:
users_op.remove(paths, args.name) users_op.remove(paths, args.name)
print(f"deleted user {args.name}") print(f"deleted user {args.name}")
elif args.action == "show": elif args.action == "show":
if args.json: if args.clash:
print(show_op.user_clash(paths, args.name, args.protocol), end="")
elif args.json:
if not args.protocol: if not args.protocol:
from ..errors import SbpError from ..errors import SbpError
+6 -1
View File
@@ -8,7 +8,7 @@ from pathlib import Path
from ..config import build_client_config from ..config import build_client_config
from ..errors import SbpError from ..errors import SbpError
from ..paths import Paths from ..paths import Paths
from ..share import all_uris, share_uri from ..share import all_uris, clash_yaml, share_uri
from .common import load from .common import load
@@ -27,6 +27,11 @@ def client_config(paths: Paths, name: str, protocol_key: str) -> dict:
return build_client_config(state, protocol_key, state.user(name)) return build_client_config(state, protocol_key, state.user(name))
def user_clash(paths: Paths, name: str, protocol_key: str | None = None) -> str:
_, state = load(paths)
return clash_yaml(state, state.user(name), protocol_key)
def client_config_json(paths: Paths, name: str, protocol_key: str) -> str: def client_config_json(paths: Paths, name: str, protocol_key: str) -> str:
return json.dumps(client_config(paths, name, protocol_key), indent=2) return json.dumps(client_config(paths, name, protocol_key), indent=2)
+14
View File
@@ -10,6 +10,7 @@ from .base import (
node_name, node_name,
quote, quote,
server_tls, server_tls,
skip_cert_verify,
uri_host, uri_host,
) )
@@ -45,3 +46,16 @@ class AnyTls(Protocol):
"password": user.password, "password": user.password,
"tls": client_tls(ctx), "tls": client_tls(ctx),
} }
def clash_proxy(self, ctx, settings, user, port):
return {
"name": f"{user.name}-{self.info.key}",
"type": "anytls",
"server": ctx.host,
"port": port,
"password": user.password,
"sni": ctx.sni,
"client-fingerprint": "chrome",
"udp": True,
**skip_cert_verify(ctx),
}
+11
View File
@@ -84,6 +84,11 @@ def insecure_flag(ctx: ServerContext) -> str:
return "1" if ctx.cert_self_signed else "0" return "1" if ctx.cert_self_signed else "0"
def skip_cert_verify(ctx: ServerContext) -> dict[str, Any]:
"""Clash proxies that cannot pin a certificate need this for a self-signed one."""
return {"skip-cert-verify": True} if ctx.cert_self_signed else {}
class Protocol(ABC): class Protocol(ABC):
info: ProtocolInfo info: ProtocolInfo
@@ -123,5 +128,11 @@ class Protocol(ABC):
) -> dict[str, Any]: ) -> dict[str, Any]:
... ...
@abstractmethod
def clash_proxy(
self, ctx: ServerContext, settings: dict[str, Any], user: User, port: int
) -> dict[str, Any]:
"""One entry of a Clash `proxies:` list."""
def summary(self, settings: dict[str, Any]) -> list[tuple[str, str]]: def summary(self, settings: dict[str, Any]) -> list[tuple[str, str]]:
return [] return []
+17
View File
@@ -5,6 +5,7 @@ from __future__ import annotations
import urllib.parse import urllib.parse
from typing import Any from typing import Any
from ..clash import cert_fingerprint
from ..credentials import token from ..credentials import token
from ..hostinfo import is_ip from ..hostinfo import is_ip
from .base import ( from .base import (
@@ -93,6 +94,22 @@ class Hysteria2(Protocol):
} }
return outbound return outbound
def clash_proxy(self, ctx, settings, user, port):
proxy: dict[str, Any] = {
"name": f"{user.name}-{self.info.key}",
"type": "hysteria2",
"server": ctx.host,
"port": port,
"password": user.password,
"sni": ctx.sni,
}
if settings.get("obfs"):
proxy["obfs"] = "salamander"
proxy["obfs-password"] = settings["obfs_password"]
if ctx.cert_self_signed and ctx.cert_sha256:
proxy["fingerprint"] = cert_fingerprint(ctx.cert_sha256)
return proxy
def summary(self, settings): def summary(self, settings):
return [ return [
("obfs", "salamander" if settings.get("obfs") else "off"), ("obfs", "salamander" if settings.get("obfs") else "off"),
+12
View File
@@ -55,5 +55,17 @@ class Shadowsocks(Protocol):
"password": f"{settings['server_key']}:{user.ss_key}", "password": f"{settings['server_key']}:{user.ss_key}",
} }
def clash_proxy(self, ctx, settings, user, port):
return {
"name": f"{user.name}-{self.info.key}",
"type": "ss",
"server": ctx.host,
"port": port,
"cipher": METHOD,
# Multi-user SS2022: server identity PSK and user PSK, colon separated.
"password": f"{settings['server_key']}:{user.ss_key}",
"udp": True,
}
def summary(self, settings): def summary(self, settings):
return [("method", settings.get("method", METHOD))] return [("method", settings.get("method", METHOD))]
+13
View File
@@ -11,6 +11,7 @@ from .base import (
node_name, node_name,
quote, quote,
server_tls, server_tls,
skip_cert_verify,
uri_host, uri_host,
) )
@@ -51,3 +52,15 @@ class Trojan(Protocol):
"password": user.password, "password": user.password,
"tls": client_tls(ctx), "tls": client_tls(ctx),
} }
def clash_proxy(self, ctx, settings, user, port):
return {
"name": f"{user.name}-{self.info.key}",
"type": "trojan",
"server": ctx.host,
"port": port,
"password": user.password,
"sni": ctx.sni,
"udp": True,
**skip_cert_verify(ctx),
}
+16
View File
@@ -12,6 +12,7 @@ from .base import (
node_name, node_name,
quote, quote,
server_tls, server_tls,
skip_cert_verify,
uri_host, uri_host,
) )
@@ -68,5 +69,20 @@ class Tuic(Protocol):
"tls": client_tls(ctx, alpn=["h3"]), "tls": client_tls(ctx, alpn=["h3"]),
} }
def clash_proxy(self, ctx, settings, user, port):
return {
"name": f"{user.name}-{self.info.key}",
"type": "tuic",
"server": ctx.host,
"port": port,
"uuid": user.uuid,
"password": user.password,
"sni": ctx.sni,
"alpn": ["h3"],
"congestion-controller": settings.get("congestion_control", "bbr"),
"udp-relay-mode": "native",
**skip_cert_verify(ctx),
}
def summary(self, settings): def summary(self, settings):
return [("congestion_control", settings.get("congestion_control", "bbr"))] return [("congestion_control", settings.get("congestion_control", "bbr"))]
+19
View File
@@ -111,6 +111,25 @@ class VlessReality(Protocol):
}, },
} }
def clash_proxy(self, ctx, settings, user, port):
return {
"name": f"{user.name}-{self.info.key}",
"type": "vless",
"server": ctx.host,
"port": port,
"uuid": user.uuid,
"network": "tcp",
"udp": True,
"tls": True,
"flow": settings.get("flow", "xtls-rprx-vision"),
"servername": settings.get("server_name") or ctx.sni,
"client-fingerprint": "chrome",
"reality-opts": {
"public-key": ctx.reality_public,
"short-id": settings.get("short_id", ""),
},
}
def summary(self, settings): def summary(self, settings):
return [ return [
("server_name", settings.get("server_name") or "auto"), ("server_name", settings.get("server_name") or "auto"),
+24 -3
View File
@@ -1,14 +1,17 @@
"""Share URI generation for enabled protocols.""" """Client-side output: share URIs and Clash proxies."""
from __future__ import annotations from __future__ import annotations
from typing import Any
from . import protocols from . import protocols
from .clash import proxies_yaml
from .config import server_context from .config import server_context
from .errors import SbpError from .errors import SbpError
from .state import State, User from .state import State, User
def share_uri(state: State, protocol_key: str, user: User) -> str: def _resolve(state: State, protocol_key: str):
protocol_state = state.protocols.get(protocol_key) protocol_state = state.protocols.get(protocol_key)
if protocol_state is None or not protocol_state.enabled: if protocol_state is None or not protocol_state.enabled:
raise SbpError(f"protocol {protocol_key} is not enabled") raise SbpError(f"protocol {protocol_key} is not enabled")
@@ -16,11 +19,29 @@ def share_uri(state: State, protocol_key: str, user: User) -> str:
ctx = server_context(state) ctx = server_context(state)
settings = protocol.normalize(protocol_state.options, ctx) settings = protocol.normalize(protocol_state.options, ctx)
protocol.validate(settings, ctx) protocol.validate(settings, ctx)
return protocol, ctx, settings, protocol_state
def share_uri(state: State, protocol_key: str, user: User) -> str:
protocol, ctx, settings, protocol_state = _resolve(state, protocol_key)
return protocol.share_uri(ctx, settings, user, protocol_state.port) return protocol.share_uri(ctx, settings, user, protocol_state.port)
def all_uris(state: State, user: User) -> list[tuple[str, str]]: def all_uris(state: State, user: User) -> list[tuple[str, str]]:
return [ return [
(key, share_uri(state, key, user)) (key, share_uri(state, key, user))
for key, protocol_state in state.enabled_protocols() for key, _ in state.enabled_protocols()
] ]
def clash_proxies(state: State, user: User, protocol_key: str | None = None) -> list[dict[str, Any]]:
keys = [protocol_key] if protocol_key else [key for key, _ in state.enabled_protocols()]
result = []
for key in keys:
protocol, ctx, settings, protocol_state = _resolve(state, key)
result.append(protocol.clash_proxy(ctx, settings, user, protocol_state.port))
return result
def clash_yaml(state: State, user: User, protocol_key: str | None = None) -> str:
return proxies_yaml(clash_proxies(state, user, protocol_key))
+5 -3
View File
@@ -33,15 +33,17 @@ def _detail(app: "App", name: str) -> None:
while True: while True:
choice = app.term.menu( choice = app.term.menu(
f"User {name}", f"User {name}",
["Show share URIs", "Show client config", "Delete user", "Back"], ["Show share URIs", "Show Clash YAML", "Show client config", "Delete user", "Back"],
) )
if choice is None or choice == 3: if choice is None or choice == 4:
return return
if choice == 0: if choice == 0:
uris = show_op.user_uris(app.paths, name) uris = show_op.user_uris(app.paths, name)
text = "\n\n".join(f"{key}\n {uri}" for key, uri in uris) or "no enabled protocols" text = "\n\n".join(f"{key}\n {uri}" for key, uri in uris) or "no enabled protocols"
app.term.scroll_text(f"Share URIs: {name}", text) app.term.scroll_text(f"Share URIs: {name}", text)
elif choice == 1: elif choice == 1:
app.term.scroll_text(f"Clash: {name}", show_op.user_clash(app.paths, name))
elif choice == 2:
enabled = [row.key for row in app.status().protocols if row.enabled] enabled = [row.key for row in app.status().protocols if row.enabled]
if not enabled: if not enabled:
app.term.message("Client config", "no enabled protocols") app.term.message("Client config", "no enabled protocols")
@@ -54,7 +56,7 @@ def _detail(app: "App", name: str) -> None:
f"Client config: {name} / {enabled[picked]}", f"Client config: {name} / {enabled[picked]}",
show_op.client_config_json(app.paths, name, enabled[picked]), show_op.client_config_json(app.paths, name, enabled[picked]),
) )
elif choice == 2: elif choice == 3:
if app.term.confirm("Delete user", f"Delete {name}? Clients using this user stop working."): if app.term.confirm("Delete user", f"Delete {name}? Clients using this user stop working."):
app.term.run_external("Delete user", lambda progress: users_op.remove(app.paths, name)) app.term.run_external("Delete user", lambda progress: users_op.remove(app.paths, name))
return return
+97
View File
@@ -0,0 +1,97 @@
import unittest
from sbp.clash import cert_fingerprint, proxies_yaml
from sbp.errors import SbpError
from sbp.share import clash_proxies, clash_yaml
from tests.test_state import CERT_SHA256, SERVER_SS_KEY, USER_SS_KEY, sample_state
class FingerprintTest(unittest.TestCase):
def test_colon_uppercase(self) -> None:
self.assertEqual(cert_fingerprint("ab" * 32), ":".join(["AB"] * 32))
def test_rejects_short_value(self) -> None:
with self.assertRaises(ValueError):
cert_fingerprint("abc")
class YamlTest(unittest.TestCase):
def test_nested_mapping_and_list(self) -> None:
text = proxies_yaml([{
"name": "a",
"port": 1,
"udp": True,
"alpn": ["h3"],
"reality-opts": {"public-key": "P", "short-id": "s"},
}])
self.assertEqual(
text,
"proxies:\n"
" - name: a\n"
" port: 1\n"
" udp: true\n"
" alpn:\n"
" - h3\n"
" reality-opts:\n"
" public-key: P\n"
" short-id: s\n",
)
def test_quotes_values_that_need_it(self) -> None:
text = proxies_yaml([{"name": "a", "password": "abc:def", "server": "2602::1"}])
self.assertIn('password: "abc:def"', text)
self.assertIn('server: "2602::1"', text)
def test_quotes_reserved_words(self) -> None:
self.assertIn('name: "no"', proxies_yaml([{"name": "no"}]))
class ProxyTest(unittest.TestCase):
def setUp(self) -> None:
self.state = sample_state()
self.user = self.state.users[0]
self.proxies = {proxy["type"]: proxy for proxy in clash_proxies(self.state, self.user)}
def test_all_protocols_present(self) -> None:
self.assertEqual(
set(self.proxies), {"hysteria2", "vless", "trojan", "ss", "tuic", "anytls"}
)
def test_hysteria2_pins_certificate(self) -> None:
proxy = self.proxies["hysteria2"]
self.assertEqual(proxy["fingerprint"], cert_fingerprint(CERT_SHA256))
self.assertEqual(proxy["obfs"], "salamander")
self.assertNotIn("skip-cert-verify", proxy)
def test_vless_reality_options(self) -> None:
proxy = self.proxies["vless"]
self.assertEqual(proxy["reality-opts"]["public-key"], "PUBLIC")
self.assertEqual(proxy["servername"], "www.bing.com")
self.assertTrue(proxy["tls"])
def test_shadowsocks_sends_identity_and_user_key(self) -> None:
self.assertEqual(
self.proxies["ss"]["password"], f"{SERVER_SS_KEY}:{USER_SS_KEY}"
)
def test_self_signed_protocols_skip_verification(self) -> None:
for key in ("trojan", "tuic", "anytls"):
self.assertTrue(self.proxies[key]["skip-cert-verify"], key)
def test_disabled_protocol_is_rejected(self) -> None:
self.state.protocols["tuic"].enabled = False
with self.assertRaises(SbpError):
clash_proxies(self.state, self.user, "tuic")
def test_single_protocol(self) -> None:
proxies = clash_proxies(self.state, self.user, "trojan")
self.assertEqual([proxy["type"] for proxy in proxies], ["trojan"])
def test_yaml_covers_every_enabled_protocol(self) -> None:
text = clash_yaml(self.state, self.user)
self.assertTrue(text.startswith("proxies:\n"))
self.assertEqual(text.count("\n - name: "), 6)
if __name__ == "__main__":
unittest.main()
+2 -2
View File
@@ -4,7 +4,7 @@ import unittest
from sbp.config import build_client_config, build_server_config from sbp.config import build_client_config, build_server_config
from sbp.errors import SbpError from sbp.errors import SbpError
from sbp.share import all_uris, share_uri from sbp.share import all_uris, share_uri
from tests.test_state import SERVER_SS_KEY, USER_SS_KEY, sample_state from tests.test_state import CERT_SHA256, SERVER_SS_KEY, USER_SS_KEY, sample_state
class ServerConfigTest(unittest.TestCase): class ServerConfigTest(unittest.TestCase):
@@ -114,7 +114,7 @@ class ShareUriTest(unittest.TestCase):
def test_hysteria2_pin_and_obfs(self) -> None: def test_hysteria2_pin_and_obfs(self) -> None:
uri = share_uri(self.state, "hysteria2", self.user) uri = share_uri(self.state, "hysteria2", self.user)
self.assertIn("pinSHA256=SHA256", uri) self.assertIn(f"pinSHA256={CERT_SHA256}", uri)
self.assertIn("obfs=salamander", uri) self.assertIn("obfs=salamander", uri)
def test_vless_reality_parameters(self) -> None: def test_vless_reality_parameters(self) -> None:
+2 -1
View File
@@ -6,6 +6,7 @@ from sbp.state import ProtocolState, RealityState, State, TlsState, User, migrat
USER_UUID = "11111111-2222-3333-4444-555555555555" USER_UUID = "11111111-2222-3333-4444-555555555555"
USER_SS_KEY = base64.b64encode(b"0123456789abcdef").decode() USER_SS_KEY = base64.b64encode(b"0123456789abcdef").decode()
SERVER_SS_KEY = base64.b64encode(b"abcdef0123456789").decode() SERVER_SS_KEY = base64.b64encode(b"abcdef0123456789").decode()
CERT_SHA256 = "ab" * 32
def sample_state() -> State: def sample_state() -> State:
@@ -21,7 +22,7 @@ def sample_state() -> State:
sni="www.bing.com", sni="www.bing.com",
self_signed=True, self_signed=True,
spki_pin="SPKI", spki_pin="SPKI",
cert_sha256="SHA256", cert_sha256=CERT_SHA256,
), ),
reality=RealityState(private_key="PRIVATE", public_key="PUBLIC"), reality=RealityState(private_key="PRIVATE", public_key="PUBLIC"),
users=[User("u1", "PASSWORD", USER_UUID, USER_SS_KEY, "2026-01-01T00:00:00Z")], users=[User("u1", "PASSWORD", USER_UUID, USER_SS_KEY, "2026-01-01T00:00:00Z")],