From f9874454f0e011b15dc926ca2405bca30185c280 Mon Sep 17 00:00:00 2001 From: typist <1003659191@qq.com> Date: Sun, 1 Nov 2026 06:58:42 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=20Clash/mihomo=20=E8=8A=82?= =?UTF-8?q?=E7=82=B9=E8=BE=93=E5=87=BA:=20sbp=20user=20show=20--clash?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 20 +++++- src/sbp/clash.py | 59 ++++++++++++++++++ src/sbp/commands/user.py | 5 +- src/sbp/ops/show.py | 7 ++- src/sbp/protocols/anytls.py | 14 +++++ src/sbp/protocols/base.py | 11 ++++ src/sbp/protocols/hysteria2.py | 17 ++++++ src/sbp/protocols/shadowsocks.py | 12 ++++ src/sbp/protocols/trojan.py | 13 ++++ src/sbp/protocols/tuic.py | 16 +++++ src/sbp/protocols/vless_reality.py | 19 ++++++ src/sbp/share.py | 27 ++++++++- src/sbp/tui/screens/users.py | 8 ++- tests/test_clash.py | 97 ++++++++++++++++++++++++++++++ tests/test_config.py | 4 +- tests/test_state.py | 3 +- 16 files changed, 318 insertions(+), 14 deletions(-) create mode 100644 src/sbp/clash.py create mode 100644 tests/test_clash.py diff --git a/README.md b/README.md index 9c44f65..45b1ceb 100644 --- a/README.md +++ b/README.md @@ -101,6 +101,19 @@ sing-box run -c client.json 客户端配置包含 `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 @@ -123,7 +136,7 @@ TUI 提供这些页面: - Dashboard: 状态总览, 启动, 停止, 重启. - Protocols: 启用, 停用, 改端口, 改选项. -- Users: 增加, 删除, 查看分享链接和客户端配置. +- Users: 增加, 删除, 查看分享链接, Clash 配置和客户端配置. - Settings: 修改 host 和 SNI, 更换证书, 升级, 重装, 卸载. - Logs: 查看 journal. @@ -178,13 +191,14 @@ src/sbp/ commands/ 每个命令组一个模块 state.py 状态模型, JSON 读写, 迁移 config.py 生成 sing-box 服务端和客户端配置 - share.py 生成分享链接 + share.py 生成分享链接和 Clash 节点 + clash.py Clash (mihomo) YAML 输出 protocols/ 每个协议一个模块 ops/ 业务操作, CLI 和 TUI 共用 tui/ curses 界面 ``` -协议模块只做三件事: 生成 inbound, 生成分享链接, 生成客户端 outbound. +协议模块只做四件事: 生成 inbound, 分享链接, 客户端 outbound, Clash 节点. 它不读写文件, 不操作服务, 不访问网络. 新增协议只需要在 `protocols/` 增加一个模块, 并在 `protocols/__init__.py` 注册. diff --git a/src/sbp/clash.py b/src/sbp/clash.py new file mode 100644 index 0000000..0b60020 --- /dev/null +++ b/src/sbp/clash.py @@ -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" diff --git a/src/sbp/commands/user.py b/src/sbp/commands/user.py index 17d9239..d356298 100644 --- a/src/sbp/commands/user.py +++ b/src/sbp/commands/user.py @@ -22,6 +22,7 @@ def register(sub) -> None: show.add_argument("--protocol", choices=protocols.keys()) 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("--clash", action="store_true", help="print a Clash proxies block") 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) print(f"deleted user {args.name}") 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: from ..errors import SbpError diff --git a/src/sbp/ops/show.py b/src/sbp/ops/show.py index 0a70206..ebe263a 100644 --- a/src/sbp/ops/show.py +++ b/src/sbp/ops/show.py @@ -8,7 +8,7 @@ from pathlib import Path from ..config import build_client_config from ..errors import SbpError from ..paths import Paths -from ..share import all_uris, share_uri +from ..share import all_uris, clash_yaml, share_uri 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)) +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: return json.dumps(client_config(paths, name, protocol_key), indent=2) diff --git a/src/sbp/protocols/anytls.py b/src/sbp/protocols/anytls.py index 1ed7286..1ba0994 100644 --- a/src/sbp/protocols/anytls.py +++ b/src/sbp/protocols/anytls.py @@ -10,6 +10,7 @@ from .base import ( node_name, quote, server_tls, + skip_cert_verify, uri_host, ) @@ -45,3 +46,16 @@ class AnyTls(Protocol): "password": user.password, "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), + } diff --git a/src/sbp/protocols/base.py b/src/sbp/protocols/base.py index d9c08bd..60b6149 100644 --- a/src/sbp/protocols/base.py +++ b/src/sbp/protocols/base.py @@ -84,6 +84,11 @@ def insecure_flag(ctx: ServerContext) -> str: 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): info: ProtocolInfo @@ -123,5 +128,11 @@ class Protocol(ABC): ) -> 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]]: return [] diff --git a/src/sbp/protocols/hysteria2.py b/src/sbp/protocols/hysteria2.py index f9af0c3..104dd92 100644 --- a/src/sbp/protocols/hysteria2.py +++ b/src/sbp/protocols/hysteria2.py @@ -5,6 +5,7 @@ from __future__ import annotations import urllib.parse from typing import Any +from ..clash import cert_fingerprint from ..credentials import token from ..hostinfo import is_ip from .base import ( @@ -93,6 +94,22 @@ class Hysteria2(Protocol): } 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): return [ ("obfs", "salamander" if settings.get("obfs") else "off"), diff --git a/src/sbp/protocols/shadowsocks.py b/src/sbp/protocols/shadowsocks.py index 634037a..9a03e34 100644 --- a/src/sbp/protocols/shadowsocks.py +++ b/src/sbp/protocols/shadowsocks.py @@ -55,5 +55,17 @@ class Shadowsocks(Protocol): "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): return [("method", settings.get("method", METHOD))] diff --git a/src/sbp/protocols/trojan.py b/src/sbp/protocols/trojan.py index 3799be6..28da659 100644 --- a/src/sbp/protocols/trojan.py +++ b/src/sbp/protocols/trojan.py @@ -11,6 +11,7 @@ from .base import ( node_name, quote, server_tls, + skip_cert_verify, uri_host, ) @@ -51,3 +52,15 @@ class Trojan(Protocol): "password": user.password, "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), + } diff --git a/src/sbp/protocols/tuic.py b/src/sbp/protocols/tuic.py index a172009..25c9105 100644 --- a/src/sbp/protocols/tuic.py +++ b/src/sbp/protocols/tuic.py @@ -12,6 +12,7 @@ from .base import ( node_name, quote, server_tls, + skip_cert_verify, uri_host, ) @@ -68,5 +69,20 @@ class Tuic(Protocol): "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): return [("congestion_control", settings.get("congestion_control", "bbr"))] diff --git a/src/sbp/protocols/vless_reality.py b/src/sbp/protocols/vless_reality.py index b8c1288..2cd3174 100644 --- a/src/sbp/protocols/vless_reality.py +++ b/src/sbp/protocols/vless_reality.py @@ -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): return [ ("server_name", settings.get("server_name") or "auto"), diff --git a/src/sbp/share.py b/src/sbp/share.py index b8eb2fb..e7af6f8 100644 --- a/src/sbp/share.py +++ b/src/sbp/share.py @@ -1,14 +1,17 @@ -"""Share URI generation for enabled protocols.""" +"""Client-side output: share URIs and Clash proxies.""" from __future__ import annotations +from typing import Any + from . import protocols +from .clash import proxies_yaml from .config import server_context from .errors import SbpError 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) if protocol_state is None or not protocol_state.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) settings = protocol.normalize(protocol_state.options, 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) def all_uris(state: State, user: User) -> list[tuple[str, str]]: return [ (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)) diff --git a/src/sbp/tui/screens/users.py b/src/sbp/tui/screens/users.py index c4c5074..4e4fed5 100644 --- a/src/sbp/tui/screens/users.py +++ b/src/sbp/tui/screens/users.py @@ -33,15 +33,17 @@ def _detail(app: "App", name: str) -> None: while True: choice = app.term.menu( 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 if choice == 0: 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" app.term.scroll_text(f"Share URIs: {name}", text) 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] if not enabled: 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]}", 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."): app.term.run_external("Delete user", lambda progress: users_op.remove(app.paths, name)) return diff --git a/tests/test_clash.py b/tests/test_clash.py new file mode 100644 index 0000000..3fb3d9d --- /dev/null +++ b/tests/test_clash.py @@ -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() diff --git a/tests/test_config.py b/tests/test_config.py index 43b92cd..384f253 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -4,7 +4,7 @@ import unittest from sbp.config import build_client_config, build_server_config from sbp.errors import SbpError 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): @@ -114,7 +114,7 @@ class ShareUriTest(unittest.TestCase): def test_hysteria2_pin_and_obfs(self) -> None: 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) def test_vless_reality_parameters(self) -> None: diff --git a/tests/test_state.py b/tests/test_state.py index 01f3c08..79eeb36 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -6,6 +6,7 @@ from sbp.state import ProtocolState, RealityState, State, TlsState, User, migrat USER_UUID = "11111111-2222-3333-4444-555555555555" USER_SS_KEY = base64.b64encode(b"0123456789abcdef").decode() SERVER_SS_KEY = base64.b64encode(b"abcdef0123456789").decode() +CERT_SHA256 = "ab" * 32 def sample_state() -> State: @@ -21,7 +22,7 @@ def sample_state() -> State: sni="www.bing.com", self_signed=True, spki_pin="SPKI", - cert_sha256="SHA256", + cert_sha256=CERT_SHA256, ), reality=RealityState(private_key="PRIVATE", public_key="PUBLIC"), users=[User("u1", "PASSWORD", USER_UUID, USER_SS_KEY, "2026-01-01T00:00:00Z")],