TUI 改为左右分栏: 左侧栏目, 右侧内容

This commit is contained in:
typist committed 2026-11-01 07:03:40 +08:00
1 parent f9874454f0
commit e06c57b7ad
17 files changed
+815 -425

No files matched your search

+5 -5
View File
@@ -132,15 +132,15 @@ sbp cert renew
sbp tui sbp tui
``` ```
TUI 提供这些页面: 界面分左右两栏. 左栏固定显示栏目, 右栏显示当前栏目内容, 不需要一层层进菜单.
- Dashboard: 状态总览, 启动, 停止, 重启. - Dashboard: 状态总览, 启动, 停止, 重启.
- Protocols: 启用, 停用, 改端口, 改选项. - Protocols: 选中协议后, 右栏显示端口, 传输, 参数, 以及启用, 停用, 改端口, 改选项.
- Users: 增加, 删除, 查看分享链接, Clash 配置和客户端配置. - Users: 选中用户后, 可查看分享链接, Clash 配置, 客户端配置, 或删除用户.
- Settings: 修改 host 和 SNI, 更换证书, 升级, 重装, 卸载. - Settings: 修改 host 和 SNI, 更换证书, 升级, 重装, 卸载.
- Logs: 查看 journal. - Logs: 查看 journal.
操作键: `↑`/`↓` 或 `j`/`k` 移动, `Enter` 确认, `q` 返回. 操作键: `Tab` 切换左右栏, `↑`/`↓` 或 `j`/`k` 移动, `Enter` 执行, `1`-`5` 直接跳到栏目, `q` 退出.
## 证书 ## 证书
@@ -195,7 +195,7 @@ src/sbp/
clash.py Clash (mihomo) YAML 输出 clash.py Clash (mihomo) YAML 输出
protocols/ 每个协议一个模块 protocols/ 每个协议一个模块
ops/ 业务操作, CLI 和 TUI 共用 ops/ 业务操作, CLI 和 TUI 共用
tui/ curses 界面 tui/ curses 界面: frame 画两栏, views 是各栏内容
``` ```
协议模块只做四件事: 生成 inbound, 分享链接, 客户端 outbound, Clash 节点. 协议模块只做四件事: 生成 inbound, 分享链接, 客户端 outbound, Clash 节点.
+119 -59
View File
@@ -1,90 +1,150 @@
"""TUI application loop and shared helpers.""" """TUI application: sidebar navigation and the right-hand content pane."""
from __future__ import annotations from __future__ import annotations
import curses import curses
import sys import sys
from .. import constants, protocols from .. import constants
from ..errors import SbpError from ..errors import SbpError
from ..ops import install as install_op
from ..ops import status as status_op from ..ops import status as status_op
from ..paths import Paths from ..paths import Paths
from .frame import Row, draw
from .term import Term from .term import Term
from .screens import dashboard, logs, protocols as protocols_screen, settings, users from .views import dashboard, logs, protocols, settings, users
FOOTER = "Tab pane ↑/↓ move Enter select 1-5 section q quit"
class App: class App:
def __init__(self, paths: Paths) -> None: def __init__(self, paths: Paths) -> None:
self.paths = paths self.paths = paths
self.term: Term self.term: Term
self.views = [
dashboard.DashboardView(),
protocols.ProtocolsView(),
users.UsersView(),
settings.SettingsView(),
logs.LogsView(),
]
self.view_index = 0
self.focus = "content"
self.indices = [0] * len(self.views)
@property
def view(self):
return self.views[self.view_index]
def status(self) -> status_op.Status: def status(self) -> status_op.Status:
return status_op.status(self.paths) return status_op.status(self.paths)
def rows(self) -> list[Row]:
try:
return self.view.rows(self)
except SbpError as exc:
return [Row(exc.message, kind="error", selectable=False)]
def reset(self) -> None:
self.indices[self.view_index] = 0
def show_view(self, name: str) -> None:
for position, view in enumerate(self.views):
if view.name == name:
self.view_index = position
self.focus = "content"
self.reset()
return
def external(self, title: str, action) -> bool:
ok = self.term.run_external(title, action)
self.reset()
return ok
def main(self, screen) -> None: def main(self, screen) -> None:
self.term = Term(screen) self.term = Term(screen)
while True: while True:
status = self.status() rows = self.rows()
if not status.installed: index = self._clamp(rows, self.indices[self.view_index])
choice = self.term.menu("sbp", ["Install", "Quit"], subtitle="not installed") self.indices[self.view_index] = index
if choice == 0: draw(
self.install_wizard() self.term,
continue title=f"sbp {constants.VERSION}",
return subtitle=self._subtitle(),
subtitle = f"{status.host} {'active' if status.active else 'inactive'}" sidebar=[view.title for view in self.views],
choice = self.term.menu( sidebar_index=self.view_index,
f"sbp {constants.VERSION}", sidebar_focused=self.focus == "sidebar",
["Dashboard", "Protocols", "Users", "Settings", "Logs", "Quit"], content_title=self.view.title,
subtitle=subtitle, rows=rows,
index=index,
content_focused=self.focus == "content",
footer=FOOTER,
) )
if choice is None or choice == 5: key = self.term.getch()
if key == -1 or key == ord("q"):
return return
handler = ( if key == curses.KEY_RESIZE:
dashboard.show, continue
protocols_screen.show, if key in (9, curses.KEY_BTAB):
users.show, self.focus = "content" if self.focus == "sidebar" else "sidebar"
settings.show, elif key == curses.KEY_LEFT:
logs.show, self.focus = "sidebar"
)[choice] elif key == curses.KEY_RIGHT:
try: self.focus = "content"
handler(self) elif key in (curses.KEY_UP, ord("k")):
except SbpError as exc: self._move(rows, -1)
self.term.message("Error", exc.message, "error") elif key in (curses.KEY_DOWN, ord("j")):
self._move(rows, 1)
elif key in (10, 13, curses.KEY_ENTER):
if self.focus == "sidebar":
self.focus = "content"
self.reset()
else:
self._activate(rows, index)
elif ord("1") <= key <= ord("5"):
self.view_index = key - ord("1")
self.focus = "content"
self.reset()
def install_wizard(self) -> None: def _subtitle(self) -> str:
host = self.term.prompt("Install", "Client-facing host", "") status = self.status()
sni = self.term.prompt("Install", "TLS server name", constants.DEFAULT_SNI) if not status.installed:
if sni is None: return "not installed"
return state = "active" if status.active else "inactive"
keys = ", ".join(protocols.keys()) return f"{status.host} {state}"
selection = self.term.prompt("Install", f"Protocols ({keys})", "all")
if selection is None:
return
port_text = self.term.prompt("Install", "First port", str(constants.BASE_PORT))
if port_text is None:
return
try:
base_port = int(port_text)
except ValueError:
self.term.message("Error", f"invalid port: {port_text}", "error")
return
selected = None if selection.strip().lower() in ("", "all") else [
item.strip() for item in selection.split(",") if item.strip()
]
options = install_op.InstallOptions(
host=host or None,
sni=sni or None,
protocols=selected,
base_port=base_port,
)
def action(progress): def _clamp(self, rows: list[Row], index: int) -> int:
install_op.install(self.paths, options, progress=progress) if not rows:
for warning in options.warnings: return 0
print(f"warning: {warning}") index = max(0, min(index, len(rows) - 1))
if rows[index].selectable:
return index
for offset in range(1, len(rows) + 1):
for candidate in (index + offset, index - offset):
if 0 <= candidate < len(rows) and rows[candidate].selectable:
return candidate
return index
self.term.run_external("Install sbp", action) def _move(self, rows: list[Row], delta: int) -> None:
if self.focus == "sidebar":
self.view_index = max(0, min(len(self.views) - 1, self.view_index + delta))
self.reset()
return
position = self.indices[self.view_index]
candidate = position
while True:
candidate += delta
if not 0 <= candidate < len(rows):
return
if rows[candidate].selectable:
self.indices[self.view_index] = candidate
return
def _activate(self, rows: list[Row], index: int) -> None:
if not rows or not 0 <= index < len(rows):
return
action = rows[index].action
if action is not None:
action(self)
def run(paths: Paths) -> int: def run(paths: Paths) -> int:
+109
View File
@@ -0,0 +1,109 @@
"""Two-pane layout: a fixed sidebar on the left, the current view on the right."""
from __future__ import annotations
import curses
from dataclasses import dataclass
from typing import Callable, Optional
from .term import ACCENT, DIM, Term, display_width, pad, truncate
SIDEBAR_MIN = 16
SIDEBAR_MAX = 28
MIN_WIDTH = 46
MIN_HEIGHT = 10
@dataclass(frozen=True)
class Row:
text: str
action: Optional[Callable] = None
kind: str = ""
selectable: bool = True
def sidebar_width(titles: list[str]) -> int:
widest = max((display_width(title) for title in titles), default=0)
return max(SIDEBAR_MIN, min(widest + 4, SIDEBAR_MAX))
def window_offset(index: int, count: int, visible: int) -> int:
"""Scroll just enough to keep the selected row on screen."""
if visible <= 0 or count <= visible:
return 0
return max(0, min(index - visible + 1, count - visible))
def draw(
term: Term,
*,
title: str,
subtitle: str,
sidebar: list[str],
sidebar_index: int,
sidebar_focused: bool,
content_title: str,
rows: list[Row],
index: int,
content_focused: bool,
footer: str,
) -> None:
height, width = term.size()
term.clear()
term.put(0, 1, title, term.attr(ACCENT, curses.A_BOLD))
if subtitle:
term.put(0, max(2, width - display_width(subtitle) - 2), subtitle, term.attr(DIM))
term.hline(1, term.attr(DIM))
if width < MIN_WIDTH or height < MIN_HEIGHT:
term.put(height // 2, 2, f"terminal too small ({width}x{height})", term.attr(ACCENT))
term.refresh()
return
term.hline(height - 2, term.attr(DIM))
term.put(height - 1, 1, truncate(footer, width - 2), term.attr(DIM))
side = sidebar_width(sidebar)
body_top = 3
body_bottom = height - 4
for y in range(2, height - 2):
term.put(y, side, "│", term.attr(DIM))
for position, item in enumerate(sidebar):
y = body_top + position
if y > body_bottom:
break
selected = position == sidebar_index
if selected and sidebar_focused:
term.fill(y, 0, side, term.attr(ACCENT, curses.A_REVERSE))
term.put(y, 1, f"▸ {item}", term.attr(ACCENT, curses.A_REVERSE | curses.A_BOLD))
elif selected:
term.put(y, 1, f"▸ {item}", term.attr(ACCENT, curses.A_BOLD))
else:
term.put(y, 1, f" {item}")
x = side + 2
inner = max(1, width - x - 1)
term.put(2, x, truncate(content_title, inner), term.attr(ACCENT, curses.A_BOLD))
visible = max(1, body_bottom - body_top + 1)
offset = window_offset(index, len(rows), visible)
for line in range(visible):
position = offset + line
if position >= len(rows):
break
row = rows[position]
selected = position == index and row.selectable
attr = term.kind_attr(row.kind)
if selected and content_focused:
term.fill(body_top + line, x, inner, term.attr(ACCENT, curses.A_REVERSE))
term.put(body_top + line, x, pad(truncate(row.text, inner), inner),
term.attr(ACCENT, curses.A_REVERSE | curses.A_BOLD))
elif selected:
term.put(body_top + line, x, pad(truncate(row.text, inner), inner),
term.attr(ACCENT, curses.A_BOLD))
else:
term.put(body_top + line, x, truncate(row.text, inner), attr)
term.refresh()
-1
View File
@@ -1 +0,0 @@
"""TUI screens."""
-50
View File
@@ -1,50 +0,0 @@
"""Dashboard: status summary and service control."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...ops import service as service_op
if TYPE_CHECKING:
from ..app import App
def _summary(app: "App") -> str:
status = app.status()
lines = [
f"Service {'active' if status.active else 'inactive'}",
f"sing-box {status.singbox_version}",
f"Address {status.host}",
f"SNI {status.sni}",
f"TLS {status.tls_mode}",
f"Cert until {status.cert_until}",
f"Users {status.user_count}",
"",
"Protocols",
]
for row in status.protocols:
mark = "on " if row.enabled else "off"
details = ", ".join(f"{key}={value}" for key, value in row.details)
lines.append(f" [{mark}] {row.title:<18} port {row.port:<6} {row.transport} {details}")
return "\n".join(lines)
def show(app: "App") -> None:
while True:
status = app.status()
choice = app.term.menu(
"Dashboard",
["Show full status", "Start service", "Stop service", "Restart service", "Back"],
subtitle=f"{status.host} {'active' if status.active else 'inactive'}",
)
if choice is None or choice == 4:
return
if choice == 0:
app.term.scroll_text("Status", _summary(app))
elif choice == 1:
app.term.run_external("Start service", lambda progress: service_op.start(app.paths))
elif choice == 2:
app.term.run_external("Stop service", lambda progress: service_op.stop(app.paths))
elif choice == 3:
app.term.run_external("Restart service", lambda progress: service_op.restart(app.paths))
-22
View File
@@ -1,22 +0,0 @@
"""Journal viewer."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...ops import service as service_op
if TYPE_CHECKING:
from ..app import App
def show(app: "App") -> None:
while True:
choice = app.term.menu("Logs", ["Recent 100 lines", "Recent 500 lines", "Back"])
if choice is None or choice == 2:
return
lines = 100 if choice == 0 else 500
app.term.scroll_text(
"Logs",
"\n".join(service_op.logs(app.paths, lines=lines)) or "no log lines",
)
-99
View File
@@ -1,99 +0,0 @@
"""Protocol list and per-protocol actions."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ... import protocols
from ...ops import protocols as protocols_op
if TYPE_CHECKING:
from ..app import App
def _show(value) -> str:
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
def _label(row, protocol) -> str:
state = "enabled" if row[1] else "off"
details = ", ".join(f"{key}={value}" for key, value in row[3])
suffix = f" {details}" if details else ""
return f"{protocol.info.title:<18} {state:<8} {row[2]:<6} {protocol.info.transport}{suffix}"
def show(app: "App") -> None:
while True:
rows = protocols_op.summary(app.paths)
labels = [_label(row, protocols.get(row[0])) for row in rows]
labels.append("Back")
choice = app.term.menu("Protocols", labels, subtitle="select a protocol to manage")
if choice is None or choice == len(rows):
return
_detail(app, rows[choice][0])
def _detail(app: "App", key: str) -> None:
while True:
row = next(item for item in protocols_op.summary(app.paths) if item[0] == key)
protocol = protocols.get(key)
enabled = row[1]
actions = [
"Disable" if enabled else "Enable",
"Change port",
"Edit options",
"Show settings",
"Back",
]
choice = app.term.menu(
protocol.info.title,
actions,
subtitle=f"{'enabled' if enabled else 'off'} port {row[2]} {protocol.info.transport}",
)
if choice is None or choice == 4:
return
if choice == 0:
if enabled:
app.term.run_external("Disable", lambda progress: protocols_op.disable(app.paths, key))
else:
port_text = app.term.prompt("Enable", "Port (blank = auto)", "")
port = int(port_text) if port_text and port_text.isdigit() else None
app.term.run_external(
"Enable", lambda progress: protocols_op.enable(app.paths, key, port=port)
)
elif choice == 1:
current = str(row[2])
value = app.term.prompt("Change port", "Port", current)
if value and value.isdigit():
app.term.run_external(
"Change port", lambda progress: protocols_op.set_port(app.paths, key, int(value))
)
elif choice == 2:
_edit_options(app, key)
elif choice == 3:
lines = [f"{name}: {value}" for name, value in row[3]] or ["no editable options"]
app.term.scroll_text(protocol.info.title, "\n".join(lines))
def _edit_options(app: "App", key: str) -> None:
protocol = protocols.get(key)
editable = protocol.editable()
if not editable:
app.term.message("Options", "this protocol has no editable options")
return
current = protocols_op.options(app.paths, key)
labels = [f"{label}: {_show(current.get(name, ''))}" for name, label, _ in editable]
labels.append("Back")
choice = app.term.menu("Edit options", labels)
if choice is None or choice == len(editable):
return
name, label, _kind = editable[choice]
value = app.term.prompt("Edit option", label, _show(current.get(name, "")))
if value is None:
return
app.term.run_external(
"Apply option",
lambda progress: protocols_op.set_option(app.paths, key, name, value),
)
-98
View File
@@ -1,98 +0,0 @@
"""Server settings, upgrade, and uninstall."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...ops import install as install_op
from ...ops import settings as settings_op
from ...ops import uninstall as uninstall_op
from ...ops import upgrade as upgrade_op
if TYPE_CHECKING:
from ..app import App
ACTIONS = (
"Show settings",
"Set host",
"Set SNI",
"Use custom certificate",
"Renew certificate",
"Upgrade sing-box",
"Reinstall",
"Uninstall",
"Back",
)
def show(app: "App") -> None:
while True:
status = app.status()
choice = app.term.menu(
"Settings",
list(ACTIONS),
subtitle=f"{status.host} sni {status.sni}",
)
if choice is None or choice == 8:
return
if choice == 0:
app.term.scroll_text(
"Settings",
"\n".join([
f"Host {status.host}",
f"SNI {status.sni}",
f"TLS {status.tls_mode}",
f"Cert until {status.cert_until}",
f"Config {status.config_path}",
f"Created {status.created}",
]),
)
elif choice == 1:
value = app.term.prompt("Set host", "Client-facing host", status.host)
if value:
app.term.run_external(
"Set host", lambda progress: settings_op.set_host(app.paths, value)
)
elif choice == 2:
value = app.term.prompt("Set SNI", "TLS server name", status.sni)
if value:
app.term.run_external(
"Set SNI", lambda progress: settings_op.set_sni(app.paths, value)
)
elif choice == 3:
cert = app.term.prompt("Custom certificate", "Certificate path (PEM)", "")
if not cert:
continue
key = app.term.prompt("Custom certificate", "Private key path (PEM)", "")
if not key:
continue
sni = app.term.prompt("Custom certificate", "SNI", status.sni)
app.term.run_external(
"Use custom certificate",
lambda progress: settings_op.set_cert(app.paths, cert, key, sni or None),
)
elif choice == 4:
app.term.run_external(
"Renew certificate", lambda progress: settings_op.regenerate_cert(app.paths)
)
elif choice == 5:
app.term.run_external(
"Upgrade sing-box", lambda progress: upgrade_op.upgrade(app.paths, progress=progress)
)
elif choice == 6:
if app.term.confirm("Reinstall", "Reinstall sing-box and regenerate the service?"):
options = install_op.InstallOptions(force=True)
app.term.run_external(
"Reinstall",
lambda progress: install_op.install(app.paths, options, progress=progress),
)
elif choice == 7:
if app.term.confirm("Uninstall", "Stop and remove the service and binary?"):
purge = app.term.confirm(
"Uninstall",
"Also delete configuration, users, and certificates?",
)
app.term.run_external(
"Uninstall", lambda progress: uninstall_op.uninstall(app.paths, purge=purge)
)
return
-62
View File
@@ -1,62 +0,0 @@
"""User list, share links, and client configs."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ... import protocols
from ...ops import show as show_op
from ...ops import users as users_op
if TYPE_CHECKING:
from ..app import App
def show(app: "App") -> None:
while True:
users = users_op.list_users(app.paths)
labels = [f"{user.name:<20} {user.created[:10]}" for user in users]
labels.append("+ Add user")
labels.append("Back")
choice = app.term.menu("Users", labels, subtitle=f"{len(users)} users")
if choice is None or choice == len(users) + 1:
return
if choice == len(users):
name = app.term.prompt("Add user", "User name", "")
if name:
app.term.run_external("Add user", lambda progress: users_op.add(app.paths, name))
continue
_detail(app, users[choice].name)
def _detail(app: "App", name: str) -> None:
while True:
choice = app.term.menu(
f"User {name}",
["Show share URIs", "Show Clash YAML", "Show client config", "Delete user", "Back"],
)
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")
continue
labels = [protocols.get(key).info.title for key in enabled] + ["Back"]
picked = app.term.menu("Client config", labels)
if picked is None or picked == len(enabled):
continue
app.term.scroll_text(
f"Client config: {name} / {enabled[picked]}",
show_op.client_config_json(app.paths, name, enabled[picked]),
)
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
+56 -29
View File
@@ -1,7 +1,7 @@
"""Curses widgets: menus, prompts, dialogs, and text paging. """Curses primitives: colors, drawing, dialogs, and text input.
Only this module touches curses. Screens call these helpers and never draw Only this module touches curses. Views and the frame draw through these
raw escape sequences themselves. helpers and never write escape sequences themselves.
""" """
from __future__ import annotations from __future__ import annotations
@@ -17,6 +17,8 @@ WARN = 3
BAD = 4 BAD = 4
DIM = 5 DIM = 5
KINDS = {"info": ACCENT, "good": GOOD, "warn": WARN, "error": BAD, "dim": DIM}
def display_width(text: str) -> int: def display_width(text: str) -> int:
width = 0 width = 0
@@ -90,38 +92,44 @@ class Term:
return extra return extra
return curses.color_pair(pair) | extra return curses.color_pair(pair) | extra
def kind_attr(self, kind: str) -> int:
return self.attr(KINDS.get(kind, 0)) if kind in KINDS else 0
def size(self) -> tuple[int, int]: def size(self) -> tuple[int, int]:
height, width = self.screen.getmaxyx() return self.screen.getmaxyx()
return height, width
def getch(self) -> int:
return self.screen.getch()
def put(self, y: int, x: int, text: str, attr: int = 0) -> None: def put(self, y: int, x: int, text: str, attr: int = 0) -> None:
height, width = self.size() height, width = self.size()
if y < 0 or y >= height or x >= width: if y < 0 or y >= height or x < 0 or x >= width or not text:
return return
try: try:
self.screen.addstr(y, x, truncate(text, max(0, width - x - 1)), attr) self.screen.addstr(y, x, truncate(text, max(0, width - x - 1)), attr)
except curses.error: except curses.error:
pass pass
def fill(self, y: int, x: int, width: int, attr: int = 0) -> None:
height, columns = self.size()
if y < 0 or y >= height or width <= 0:
return
try:
self.screen.addstr(y, x, " " * min(width, columns - x - 1), attr)
except curses.error:
pass
def hline(self, y: int, attr: int = 0, x: int = 0) -> None:
height, width = self.size()
if 0 <= y < height:
self.fill(y, x, width - x - 1, attr)
def clear(self) -> None: def clear(self) -> None:
self.screen.erase() self.screen.erase()
def refresh(self) -> None: def refresh(self) -> None:
self.screen.refresh() self.screen.refresh()
def header(self, title: str, subtitle: str = "") -> int:
height, width = self.size()
self.put(0, 1, title, self.attr(ACCENT, curses.A_BOLD))
if subtitle:
self.put(0, max(2, width - display_width(subtitle) - 2), subtitle, self.attr(DIM))
self.put(1, 0, "─" * max(1, width - 1), self.attr(DIM))
return 3
def footer(self, text: str) -> None:
height, width = self.size()
self.put(height - 2, 0, "─" * max(1, width - 1), self.attr(DIM))
self.put(height - 1, 1, text, self.attr(DIM))
def message(self, title: str, text: str, kind: str = "info") -> None: def message(self, title: str, text: str, kind: str = "info") -> None:
pair = {"info": ACCENT, "good": GOOD, "warn": WARN, "error": BAD}.get(kind, ACCENT) pair = {"info": ACCENT, "good": GOOD, "warn": WARN, "error": BAD}.get(kind, ACCENT)
self.clear() self.clear()
@@ -135,11 +143,19 @@ class Term:
y += 1 y += 1
self.footer(" press any key ") self.footer(" press any key ")
self.refresh() self.refresh()
self.screen.getch() self.getch()
def confirm(self, title: str, text: str) -> bool: def confirm(self, title: str, text: str) -> bool:
items = ["Yes", "No"] self.clear()
return self.menu(title, items, subtitle=text, start=1) == 0 height, width = self.size()
y = self.header(title)
for line in textwrap.wrap(text, max(20, width - 6))[: max(1, height - 8)]:
self.put(y, 3, line)
y += 1
self.put(y + 1, 3, "y confirm n / Esc cancel", self.attr(ACCENT))
self.footer(" y confirm any other key cancels ")
self.refresh()
return self.getch() in (ord("y"), ord("Y"))
def prompt(self, title: str, label: str, default: str = "", secret: bool = False) -> Optional[str]: def prompt(self, title: str, label: str, default: str = "", secret: bool = False) -> Optional[str]:
self.clear() self.clear()
@@ -148,7 +164,7 @@ class Term:
self.put(3, 3, label) self.put(3, 3, label)
if default and not secret: if default and not secret:
self.put(3, 3 + display_width(label) + 1, f"[{default}]", self.attr(DIM)) self.put(3, 3 + display_width(label) + 1, f"[{default}]", self.attr(DIM))
self.put(height - 1, 1, "Enter accept Esc cancel", self.attr(DIM)) self.footer(" Enter accept Esc cancel ")
self.refresh() self.refresh()
try: try:
curses.curs_set(1) curses.curs_set(1)
@@ -192,6 +208,7 @@ class Term:
footer: str = "↑/↓ select Enter open q back", footer: str = "↑/↓ select Enter open q back",
start: int = 0, start: int = 0,
) -> Optional[int]: ) -> Optional[int]:
"""Modal picker for short action lists."""
index = min(max(start, 0), max(0, len(items) - 1)) index = min(max(start, 0), max(0, len(items) - 1))
while True: while True:
self.clear() self.clear()
@@ -207,7 +224,7 @@ class Term:
self.put(y + row, 2, marker + item, attr) self.put(y + row, 2, marker + item, attr)
self.footer(footer) self.footer(footer)
self.refresh() self.refresh()
key = self.screen.getch() key = self.getch()
if key == -1: if key == -1:
return None return None
if key in (curses.KEY_UP, ord("k")): if key in (curses.KEY_UP, ord("k")):
@@ -218,16 +235,13 @@ class Term:
return index return index
elif key in (ord("q"), 27): elif key in (ord("q"), 27):
return None return None
elif key == curses.KEY_RESIZE:
continue
elif ord("1") <= key <= ord("9"): elif ord("1") <= key <= ord("9"):
candidate = key - ord("1") candidate = key - ord("1")
if candidate < len(items): if candidate < len(items):
return candidate return candidate
def scroll_text(self, title: str, text: str, subtitle: str = "") -> None: def scroll_text(self, title: str, text: str, subtitle: str = "") -> None:
height, width = self.size() available = max(1, self.size()[1] - 4)
available = max(1, width - 4)
lines: list[str] = [] lines: list[str] = []
for source in text.splitlines() or [""]: for source in text.splitlines() or [""]:
lines.extend(wrap_display(source, available)) lines.extend(wrap_display(source, available))
@@ -244,7 +258,7 @@ class Term:
self.put(y + row, 2, lines[position]) self.put(y + row, 2, lines[position])
self.footer("↑/↓ scroll PgUp/PgDn q back") self.footer("↑/↓ scroll PgUp/PgDn q back")
self.refresh() self.refresh()
key = self.screen.getch() key = self.getch()
if key == -1 or key in (ord("q"), 27): if key == -1 or key in (ord("q"), 27):
return return
if key in (curses.KEY_UP, ord("k")): if key in (curses.KEY_UP, ord("k")):
@@ -256,6 +270,19 @@ class Term:
elif key == curses.KEY_NPAGE: elif key == curses.KEY_NPAGE:
top = min(max(0, len(lines) - body), top + body) top = min(max(0, len(lines) - body), top + body)
def header(self, title: str, subtitle: str = "") -> int:
height, width = self.size()
self.put(0, 1, title, self.attr(ACCENT, curses.A_BOLD))
if subtitle:
self.put(0, max(2, width - display_width(subtitle) - 2), subtitle, self.attr(DIM))
self.put(1, 0, "─" * max(1, width - 1), self.attr(DIM))
return 3
def footer(self, text: str) -> None:
height, width = self.size()
self.put(height - 2, 0, "─" * max(1, width - 1), self.attr(DIM))
self.put(height - 1, 1, text, self.attr(DIM))
def run_external(self, title: str, action: Callable[[Callable[[str], None]], None]) -> bool: def run_external(self, title: str, action: Callable[[Callable[[str], None]], None]) -> bool:
"""Leave curses, run action, wait for Enter, then restore curses.""" """Leave curses, run action, wait for Enter, then restore curses."""
try: try:
+1
View File
@@ -0,0 +1 @@
"""Content panes. Each view exposes a name, a title, and a list of rows."""
+82
View File
@@ -0,0 +1,82 @@
"""Status summary, service control, and the install wizard."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ... import constants, protocols
from ...ops import install as install_op
from ...ops import service as service_op
from ..frame import Row
if TYPE_CHECKING:
from ..app import App
def _install(app: "App") -> None:
term = app.term
host = term.prompt("Install", "Client-facing host (blank = detect)", "") or ""
sni = term.prompt("Install", "TLS server name", constants.DEFAULT_SNI)
if sni is None:
return
keys = ", ".join(protocols.keys())
selection = term.prompt("Install", f"Protocols ({keys})", "all")
if selection is None:
return
port_text = term.prompt("Install", "First port", str(constants.BASE_PORT))
if port_text is None:
return
try:
base_port = int(port_text)
except ValueError:
term.message("Error", f"invalid port: {port_text}", "error")
return
chosen = None if selection.strip().lower() in ("", "all") else [
item.strip() for item in selection.split(",") if item.strip()
]
options = install_op.InstallOptions(
host=host or None, sni=sni or None, protocols=chosen, base_port=base_port
)
def action(progress):
install_op.install(app.paths, options, progress=progress)
for warning in options.warnings:
print(f"warning: {warning}")
app.external("Install sbp", action)
def _service(app: "App", action: str) -> None:
call = {"start": service_op.start, "stop": service_op.stop, "restart": service_op.restart}[action]
app.external(action.capitalize(), lambda progress: call(app.paths))
class DashboardView:
name = "dashboard"
title = "Dashboard"
def rows(self, app: "App") -> list[Row]:
status = app.status()
if not status.installed:
return [
Row("sbp is not installed", kind="warn", selectable=False),
Row("", selectable=False),
Row("Install sbp", action=_install),
]
return [
Row("Service", kind="header", selectable=False),
Row(f" state {'active' if status.active else 'inactive'}",
kind="good" if status.active else "warn", selectable=False),
Row(f" sing-box {status.singbox_version}", selectable=False),
Row(f" address {status.host}", selectable=False),
Row(f" sni {status.sni}", selectable=False),
Row(f" tls {status.tls_mode}", selectable=False),
Row(f" cert until {status.cert_until}", selectable=False),
Row(f" users {status.user_count}", selectable=False),
Row("", selectable=False),
Row("Actions", kind="header", selectable=False),
Row("Start service", action=lambda a: _service(a, "start")),
Row("Stop service", action=lambda a: _service(a, "stop")),
Row("Restart service", action=lambda a: _service(a, "restart")),
Row("View logs", action=lambda a: a.show_view("logs")),
]
+34
View File
@@ -0,0 +1,34 @@
"""Service log from the systemd user journal."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...errors import SbpError
from ...service import journal
from ..frame import Row
if TYPE_CHECKING:
from ..app import App
LINES = 300
class LogsView:
name = "logs"
title = "Logs"
def rows(self, app: "App") -> list[Row]:
if not app.status().installed:
return [Row("sbp is not installed", kind="warn", selectable=False)]
try:
lines = journal(app.paths, lines=LINES)
except SbpError as exc:
return [Row(exc.message, kind="error", selectable=False)]
if not lines:
return [Row("no log lines", kind="dim", selectable=False)]
return [
Row(f"last {len(lines)} lines from journalctl --user -u {app.paths.unit_name}",
kind="dim", selectable=False),
Row("", selectable=False),
] + [Row(line, selectable=True) for line in lines]
+118
View File
@@ -0,0 +1,118 @@
"""Protocol list and per-protocol settings, shown in the content pane."""
from __future__ import annotations
from typing import TYPE_CHECKING, Optional
from ... import protocols
from ...ops import protocols as protocols_op
from ..frame import Row
if TYPE_CHECKING:
from ..app import App
def _detail_text(protocol, enabled: bool, port: int, details) -> str:
state = "enabled" if enabled else "off"
text = f"{protocol.info.title:<18} {state:<8} {port:<6} {protocol.info.transport}"
if details:
text += " " + ", ".join(f"{key}={value}" for key, value in details)
return text
class ProtocolsView:
name = "protocols"
title = "Protocols"
def __init__(self) -> None:
self.selected: Optional[str] = None
def rows(self, app: "App") -> list[Row]:
if not app.status().installed:
return [Row("sbp is not installed", kind="warn", selectable=False)]
if self.selected:
return self._detail(app, self.selected)
rows = [Row("select a protocol to manage", kind="dim", selectable=False), Row("", selectable=False)]
for key, enabled, port, details in protocols_op.summary(app.paths):
protocol = protocols.get(key)
rows.append(
Row(
_detail_text(protocol, enabled, port, details),
action=lambda a, k=key: self._open(a, k),
)
)
return rows
def _open(self, app: "App", key: str) -> None:
self.selected = key
app.reset()
def _close(self, app: "App") -> None:
self.selected = None
app.reset()
def _detail(self, app: "App", key: str) -> list[Row]:
row = next(item for item in protocols_op.summary(app.paths) if item[0] == key)
_, enabled, port, details = row
protocol = protocols.get(key)
rows = [
Row(protocol.info.title, kind="header", selectable=False),
Row(f" state {'enabled' if enabled else 'off'}",
kind="good" if enabled else "warn", selectable=False),
Row(f" port {port}", selectable=False),
Row(f" transport {protocol.info.transport}", selectable=False),
]
rows.extend(Row(f" {name:<12} {value}", selectable=False) for name, value in details)
rows.append(Row("", selectable=False))
rows.append(Row("Actions", kind="header", selectable=False))
if enabled:
rows.append(Row("Disable", action=lambda a: self._disable(a, key)))
else:
rows.append(Row("Enable", action=lambda a: self._enable(a, key)))
rows.append(Row("Change port", action=lambda a: self._port(a, key, port)))
if protocol.editable():
rows.append(Row("Edit options", action=lambda a: self._options(a, key)))
rows.append(Row("Back", action=lambda a: self._close(a)))
return rows
def _disable(self, app: "App", key: str) -> None:
app.external(f"Disable {key}", lambda progress: protocols_op.disable(app.paths, key))
def _enable(self, app: "App", key: str) -> None:
text = app.term.prompt("Enable", "Port (blank = auto)", "")
if text is None:
return
port = int(text) if text.isdigit() else None
app.external(f"Enable {key}", lambda progress: protocols_op.enable(app.paths, key, port=port))
def _port(self, app: "App", key: str, current: int) -> None:
text = app.term.prompt("Change port", "Port", str(current))
if not text or not text.isdigit():
return
app.external(
f"Change {key} port",
lambda progress: protocols_op.set_port(app.paths, key, int(text)),
)
def _options(self, app: "App", key: str) -> None:
protocol = protocols.get(key)
editable = protocol.editable()
current = protocols_op.options(app.paths, key)
labels = [f"{label}: {_show(current.get(name, ''))}" for name, label, _ in editable]
choice = app.term.menu(f"{protocol.info.title} options", labels)
if choice is None:
return
name, label, _kind = editable[choice]
value = app.term.prompt("Edit option", label, _show(current.get(name, "")))
if value is None:
return
app.external(
f"Set {key}.{name}",
lambda progress: protocols_op.set_option(app.paths, key, name, value),
)
def _show(value) -> str:
if isinstance(value, bool):
return "true" if value else "false"
return str(value)
+87
View File
@@ -0,0 +1,87 @@
"""Server settings, certificate, upgrade, and uninstall."""
from __future__ import annotations
from typing import TYPE_CHECKING
from ...ops import install as install_op
from ...ops import settings as settings_op
from ...ops import uninstall as uninstall_op
from ...ops import upgrade as upgrade_op
from ..frame import Row
if TYPE_CHECKING:
from ..app import App
class SettingsView:
name = "settings"
title = "Settings"
def rows(self, app: "App") -> list[Row]:
status = app.status()
if not status.installed:
return [Row("sbp is not installed", kind="warn", selectable=False)]
return [
Row("Server", kind="header", selectable=False),
Row(f" host {status.host}", selectable=False),
Row(f" sni {status.sni}", selectable=False),
Row(f" tls {status.tls_mode}", selectable=False),
Row(f" cert until {status.cert_until}", selectable=False),
Row(f" config {status.config_path}", selectable=False),
Row("", selectable=False),
Row("Actions", kind="header", selectable=False),
Row("Set host", action=self._host),
Row("Set SNI", action=self._sni),
Row("Use custom certificate", action=self._cert),
Row("Renew certificate", action=self._renew),
Row("Upgrade sing-box", action=self._upgrade),
Row("Reinstall", action=self._reinstall),
Row("Uninstall", action=self._uninstall),
]
def _host(self, app: "App") -> None:
value = app.term.prompt("Set host", "Client-facing host", app.status().host)
if value:
app.external("Set host", lambda progress: settings_op.set_host(app.paths, value))
def _sni(self, app: "App") -> None:
value = app.term.prompt("Set SNI", "TLS server name", app.status().sni)
if value:
app.external("Set SNI", lambda progress: settings_op.set_sni(app.paths, value))
def _cert(self, app: "App") -> None:
cert = app.term.prompt("Custom certificate", "Certificate path (PEM)", "")
if not cert:
return
key = app.term.prompt("Custom certificate", "Private key path (PEM)", "")
if not key:
return
sni = app.term.prompt("Custom certificate", "SNI", app.status().sni)
app.external(
"Use custom certificate",
lambda progress: settings_op.set_cert(app.paths, cert, key, sni or None),
)
def _renew(self, app: "App") -> None:
app.external("Renew certificate", lambda progress: settings_op.regenerate_cert(app.paths))
def _upgrade(self, app: "App") -> None:
app.external(
"Upgrade sing-box",
lambda progress: upgrade_op.upgrade(app.paths, progress=progress),
)
def _reinstall(self, app: "App") -> None:
if not app.term.confirm("Reinstall", "Reinstall sing-box and regenerate the service?"):
return
options = install_op.InstallOptions(force=True)
app.external(
"Reinstall", lambda progress: install_op.install(app.paths, options, progress=progress)
)
def _uninstall(self, app: "App") -> None:
if not app.term.confirm("Uninstall", "Stop and remove the service and binary?"):
return
purge = app.term.confirm("Uninstall", "Also delete configuration, users, and certificates?")
app.external("Uninstall", lambda progress: uninstall_op.uninstall(app.paths, purge=purge))
+89
View File
@@ -0,0 +1,89 @@
"""User list, share links, Clash YAML, and client configs."""
from __future__ import annotations
from typing import TYPE_CHECKING, Optional
from ... import protocols
from ...ops import show as show_op
from ...ops import users as users_op
from ..frame import Row
if TYPE_CHECKING:
from ..app import App
class UsersView:
name = "users"
title = "Users"
def __init__(self) -> None:
self.selected: Optional[str] = None
def rows(self, app: "App") -> list[Row]:
if not app.status().installed:
return [Row("sbp is not installed", kind="warn", selectable=False)]
if self.selected:
return self._detail(app, self.selected)
users = users_op.list_users(app.paths)
rows = [Row("select a user to see links and clients", kind="dim", selectable=False), Row("", selectable=False)]
rows.extend(
Row(f"{user.name:<20} {user.created[:10]}", action=lambda a, n=user.name: self._open(a, n))
for user in users
)
rows.append(Row("", selectable=False))
rows.append(Row("+ Add user", action=self._add))
return rows
def _open(self, app: "App", name: str) -> None:
self.selected = name
app.reset()
def _close(self, app: "App") -> None:
self.selected = None
app.reset()
def _add(self, app: "App") -> None:
name = app.term.prompt("Add user", "User name", "")
if not name:
return
app.external("Add user", lambda progress: users_op.add(app.paths, name))
def _detail(self, app: "App", name: str) -> list[Row]:
return [
Row(f"User {name}", kind="header", selectable=False),
Row("", selectable=False),
Row("Show share URIs", action=lambda a: self._uris(a, name)),
Row("Show Clash YAML", action=lambda a: self._clash(a, name)),
Row("Show client config", action=lambda a: self._client(a, name)),
Row("Delete user", action=lambda a: self._delete(a, name)),
Row("Back", action=lambda a: self._close(a)),
]
def _uris(self, app: "App", name: str) -> None:
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)
def _clash(self, app: "App", name: str) -> None:
app.term.scroll_text(f"Clash: {name}", show_op.user_clash(app.paths, name))
def _client(self, app: "App", name: str) -> None:
enabled = [row.key for row in app.status().protocols if row.enabled]
if not enabled:
app.term.message("Client config", "no enabled protocols")
return
labels = [protocols.get(key).info.title for key in enabled]
choice = app.term.menu("Client config", labels)
if choice is None:
return
app.term.scroll_text(
f"Client config: {name} / {enabled[choice]}",
show_op.client_config_json(app.paths, name, enabled[choice]),
)
def _delete(self, app: "App", name: str) -> None:
if not app.term.confirm("Delete user", f"Delete {name}? Clients using this user stop working."):
return
app.external("Delete user", lambda progress: users_op.remove(app.paths, name))
self._close(app)
+115
View File
@@ -0,0 +1,115 @@
import unittest
from unittest import mock
from sbp.tui import frame
from sbp.tui.term import Term
class FakeScreen:
def __init__(self, height: int = 24, width: int = 80) -> None:
self.height = height
self.width = width
self.writes: list[tuple[int, int, str]] = []
def getmaxyx(self):
return self.height, self.width
def addstr(self, y, x, text, attr=0):
if y >= self.height or x + len(text) > self.width:
raise AssertionError(f"write out of bounds: y={y} x={x} len={len(text)}")
self.writes.append((y, x, text))
def erase(self):
self.writes.clear()
def refresh(self):
pass
def keypad(self, flag):
pass
def getch(self):
return -1
def text(self) -> str:
return "\n".join(text for _, _, text in self.writes)
def make_term(screen: FakeScreen) -> Term:
with mock.patch.object(frame_term_curses(), "has_colors", return_value=False), \
mock.patch.object(frame_term_curses(), "curs_set", return_value=None):
return Term(screen)
def frame_term_curses():
import sbp.tui.term as module
return module.curses
class WindowOffsetTest(unittest.TestCase):
def test_no_scroll_when_everything_fits(self) -> None:
self.assertEqual(frame.window_offset(3, 5, 10), 0)
def test_keeps_selection_visible(self) -> None:
self.assertEqual(frame.window_offset(7, 20, 5), 3)
def test_clamps_to_last_page(self) -> None:
self.assertEqual(frame.window_offset(19, 20, 5), 15)
class SidebarWidthTest(unittest.TestCase):
def test_minimum(self) -> None:
self.assertEqual(frame.sidebar_width(["A"]), frame.SIDEBAR_MIN)
def test_grows_with_titles(self) -> None:
self.assertGreater(
frame.sidebar_width(["VeryLongSectionName"]), frame.sidebar_width(["A"])
)
def test_capped(self) -> None:
self.assertEqual(frame.sidebar_width(["x" * 100]), frame.SIDEBAR_MAX)
class DrawTest(unittest.TestCase):
def setUp(self) -> None:
self.screen = FakeScreen()
self.term = make_term(self.screen)
def draw(self, rows=None, index=0, width=80):
rows = rows or [frame.Row("first", action=lambda app: None), frame.Row("second")]
frame.draw(
self.term,
title="sbp 2.0.0",
subtitle="host active",
sidebar=["Dashboard", "Protocols", "Users", "Settings", "Logs"],
sidebar_index=1,
sidebar_focused=False,
content_title="Protocols",
rows=rows,
index=index,
content_focused=True,
footer="keys",
)
def test_draws_both_panes(self) -> None:
self.draw()
text = self.screen.text()
for word in ("Dashboard", "Protocols", "Users", "Settings", "Logs", "first", "second", "keys"):
self.assertIn(word, text)
def test_selected_row_is_padded_for_highlight(self) -> None:
self.draw(index=1)
row = [text for y, _, text in self.screen.writes if text.startswith("second")]
self.assertTrue(row)
self.assertGreater(len(row[0]), len("second"))
def test_small_terminal_shows_hint(self) -> None:
self.screen = FakeScreen(height=8, width=40)
self.term.screen = self.screen
self.draw()
self.assertIn("terminal too small", self.screen.text())
if __name__ == "__main__":
unittest.main()