diff --git a/install-agent.sh b/install-agent.sh index 334a1a2..4d023c8 100755 --- a/install-agent.sh +++ b/install-agent.sh @@ -116,6 +116,22 @@ curl -fL -o geosite.dat "${RELEASE_BASE}/geosite.dat" || error "geosite 下载 [[ -s geosite.dat ]] || error "geosite 下载得到空文件: ${RELEASE_BASE}/geosite.dat" cd - >/dev/null +# 第 6 条 B3 起 agent 为双能力：除 xray 外再装 sing-box 核心（自编 +# v1.13.21，with_v2ray_api with_utls with_quic），主控下发 sing-box +# 配置时即由它承载。sing-box 资产在 NexusGate 仓的专用发行里（与 +# NexusGate 共用同一自编件）。其 version 输出为 unknown（自编件无 +# 版本戳），故另写版本记录文件供 agent 遥测读取。xray 与 geo 的 +# 下载暂保留——现行运行时仍是 xray，B5 正式切换后才在本脚本移除。 +SINGBOX_VER="1.13.21" +SINGBOX_BASE="https://git.881218.xyz/rogue/NexusGate/releases/download/sing-box-v${SINGBOX_VER}" +SINGBOX_URL="${SINGBOX_URL:-${SINGBOX_BASE}/nexusgate-sing-box-linux-${BIN_ARCH}}" +info "下载 sing-box v${SINGBOX_VER}..." +curl -fL -o "${XUI_HOME}/bin/sing-box-linux-${BIN_ARCH}.new" "$SINGBOX_URL" || error "sing-box 下载失败: $SINGBOX_URL" +[[ -s "${XUI_HOME}/bin/sing-box-linux-${BIN_ARCH}.new" ]] || error "sing-box 下载得到空文件: $SINGBOX_URL" +chmod +x "${XUI_HOME}/bin/sing-box-linux-${BIN_ARCH}.new" +mv -f "${XUI_HOME}/bin/sing-box-linux-${BIN_ARCH}.new" "${XUI_HOME}/bin/sing-box-linux-${BIN_ARCH}" +printf '%s\n' "$SINGBOX_VER" > "${XUI_HOME}/bin/sing-box.version" + # 安装收尾自检（第 17 条）：文件在不等于能用——xray 本体须存在、可执行、 # version 能真实跑通；geo 须存在且非空；agent 本体须可执行。任一不过即 # 报错退出并指出缺什么，不许带病进入写配置/起服务阶段。 @@ -131,6 +147,14 @@ self_check() { info "自检通过：$(printf '%s' "$ver_out" | head -n 1)" [[ -s "${XUI_HOME}/bin/geoip.dat" ]] || error "自检失败：geoip.dat 缺失或为空（${XUI_HOME}/bin/geoip.dat）" [[ -s "${XUI_HOME}/bin/geosite.dat" ]] || error "自检失败：geosite.dat 缺失或为空（${XUI_HOME}/bin/geosite.dat）" + local sb_bin="${XUI_HOME}/bin/sing-box-linux-${BIN_ARCH}" + [[ -x "$sb_bin" ]] || error "自检失败：sing-box 本体缺失或不可执行 ${sb_bin}" + local sb_out + if ! sb_out=$("$sb_bin" version 2>/dev/null); then + error "自检失败：sing-box 本体无法运行（${sb_bin} version 退出非零）" + fi + [[ -n "$sb_out" ]] || error "自检失败：sing-box version 无输出（${sb_bin}）" + info "sing-box 自检通过：$(printf '%s' "$sb_out" | head -n 1)（版本记录 ${SINGBOX_VER}）" [[ -x "$AGENT_BIN" ]] || error "自检失败：agent 二进制不可执行 ${AGENT_BIN}" } self_check diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 20dd4f2..2568a7b 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -14,6 +14,7 @@ import ( "net/http" "os" "path/filepath" + "strings" "sync" "sync/atomic" "time" @@ -84,6 +85,37 @@ type Agent struct { updateFailedAt time.Time updateReport *agentUpdateStatus exePathOverride string + + // sing-box runtime (ITEM6 B3, see singbox.go). sbMu serializes + // config activation, the sing-box traffic cycle and shutdown: a + // flush-before-change must never race a periodic snapshot. + sbMu sync.Mutex + sbProc *singboxProcess + sbStats *singboxStatsClient + sbOnline *singboxOnlineTracker + sbStatsAddr string + lastSingboxHash string + sbBaselineLoaded bool + + // sing-box version cache (telemetry), mirroring the xray one. + sbVerMu sync.Mutex + sbVer string + sbVerFor string + sbVerMod time.Time +} + +// ensureSingboxState lazily creates the sing-box runtime state. +// Callers hold sbMu (or are in single-threaded setup). +func (a *Agent) ensureSingboxState() { + if a.sbProc == nil { + a.sbProc = &singboxProcess{} + } + if a.sbStats == nil { + a.sbStats = newSingboxStatsClient(a.sbProc.epoch) + } + if a.sbOnline == nil { + a.sbOnline = newSingboxOnlineTracker(singboxLogPath()) + } } // New creates a new agent instance @@ -239,6 +271,7 @@ func (a *Agent) Run() error { case <-a.stopCh: logger.Info("Agent stopping...") a.stopXray() + a.stopSingbox() return nil case <-heartbeatTicker.C: if err := a.sendHeartbeat(); err != nil { @@ -324,9 +357,11 @@ func (a *Agent) fetchAndApplyConfig() error { } var result struct { - ConfigHash string `json:"config_hash"` - XrayConfig json.RawMessage `json:"xray_config"` - AgentUpdate *updateIntent `json:"agent_update"` + ConfigHash string `json:"config_hash"` + XrayConfig json.RawMessage `json:"xray_config"` + SingboxConfig json.RawMessage `json:"singbox_config"` + SingboxConfigHash string `json:"singbox_config_hash"` + AgentUpdate *updateIntent `json:"agent_update"` } if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { return err @@ -336,6 +371,18 @@ func (a *Agent) fetchAndApplyConfig() error { // when the xray config itself is unchanged. a.maybeHandleUpdate(result.AgentUpdate) + // sing-box runtime (ITEM6 B3): an independent, parallel path. Its + // failures are logged and retried on the next poll; they never fail + // the xray apply below or mark the poll failed. + if result.SingboxConfigHash != "" && result.SingboxConfigHash != a.lastSingboxHash { + if err := a.applySingboxConfig(result.SingboxConfig); err != nil { + logger.Warningf("Failed to apply sing-box config: %v", err) + } else { + a.lastSingboxHash = result.SingboxConfigHash + logger.Info("sing-box config applied successfully") + } + } + // Skip if config hasn't changed if result.ConfigHash == a.lastConfigHash && result.ConfigHash != "" { return nil @@ -613,11 +660,25 @@ func (a *Agent) reportTrafficWith(sc statsClient, post func(trafficReport) error return nil } -// collectAndReportTraffic snapshots the local xray counters and reports +// collectAndReportTraffic runs one reporting cycle per running core +// (xray and, since ITEM6 B3, sing-box). The two cores keep separate +// clients and baselines; a failure in one never blocks the other. +func (a *Agent) collectAndReportTraffic() error { + var firstErr error + if err := a.collectAndReportXrayTraffic(); err != nil { + firstErr = err + } + if err := a.collectAndReportSingboxTraffic(); err != nil && firstErr == nil { + firstErr = err + } + return firstErr +} + +// collectAndReportXrayTraffic snapshots the local xray counters and reports // the deltas since the last master-acknowledged baseline. The XrayAPI // instance is kept across polls so baselines survive; xray restarts are // handled (counters resetting to zero are treated as fresh). -func (a *Agent) collectAndReportTraffic() error { +func (a *Agent) collectAndReportXrayTraffic() error { if a.xrayProc == nil || !a.xrayProc.IsRunning() { return nil } @@ -632,6 +693,118 @@ func (a *Agent) collectAndReportTraffic() error { return a.reportTrafficWith(&a.xrayAPI, a.postTrafficReport) } +// singboxBaselineFile is the on-disk form of the sing-box tracker's +// last committed baseline. The epoch travels with the values: after +// an agent restart against a still-running (or since-restarted) core, +// a baseline from another core instance must not be restored. +type singboxBaselineFile struct { + Version int `json:"version"` + Epoch string `json:"epoch"` + Values map[string]int64 `json:"values"` +} + +// singboxBaselinePath derives the sing-box baseline location from the +// xray one (agent-traffic-baseline.json -> agent-traffic-baseline-singbox.json), +// so main's placement next to agent.json carries over. +func (a *Agent) singboxBaselinePath() string { + p := a.baselinePath() + if strings.HasSuffix(p, ".json") { + return strings.TrimSuffix(p, ".json") + "-singbox.json" + } + return p + "-singbox" +} + +func loadSingboxBaseline(path string) (*singboxBaselineFile, error) { + data, err := os.ReadFile(path) + if err != nil { + return nil, err + } + var f singboxBaselineFile + if err := json.Unmarshal(data, &f); err != nil { + return nil, fmt.Errorf("parse sing-box traffic baseline: %w", err) + } + return &f, nil +} + +func saveSingboxBaseline(path, epoch string, values map[string]int64) error { + data, err := json.Marshal(singboxBaselineFile{Version: 1, Epoch: epoch, Values: values}) + if err != nil { + return err + } + tmp := path + ".tmp" + if err := os.WriteFile(tmp, data, 0o600); err != nil { + return err + } + return os.Rename(tmp, path) +} + +// collectAndReportSingboxTraffic runs one sing-box reporting cycle. +func (a *Agent) collectAndReportSingboxTraffic() error { + a.sbMu.Lock() + defer a.sbMu.Unlock() + if a.sbProc == nil || !a.sbProc.isRunning() { + return nil + } + return a.reportSingboxTrafficLocked() +} + +// reportSingboxTrafficLocked is the sing-box counterpart of +// reportTrafficWith, with the same contract — commit only after the +// master acknowledges (HTTP 200), persist after every commit — plus +// the epoch carried through restore/persist. The online set merges +// the access-log tracker with the users whose counters just moved. +// Callers hold sbMu. +func (a *Agent) reportSingboxTrafficLocked() error { + a.ensureSingboxState() + if a.sbStatsAddr == "" { + // The applied config carries no v2ray_api stats block (the + // panel-side stats switches are off): nothing exists to read. + return nil + } + a.sbStats.SetAddr(a.sbStatsAddr) + + if !a.sbBaselineLoaded { + a.sbBaselineLoaded = true + if f, err := loadSingboxBaseline(a.singboxBaselinePath()); err != nil { + if !os.IsNotExist(err) { + logger.Warningf("sing-box traffic baseline at %s unusable (%v); starting from a fresh baseline", a.singboxBaselinePath(), err) + } + } else { + a.sbStats.RestoreTrafficBaseline(f.Values, f.Epoch) + } + } + + a.sbOnline.Poll(time.Now()) + + snap, err := a.sbStats.GetTrafficSnapshot() + if err != nil { + if errors.Is(err, errSingboxEpochChanged) { + // The core restarted around this poll (config change or + // crash): this round establishes nothing; the next poll + // rebases onto the new instance. Not a fault. + return nil + } + return err + } + + report := buildTrafficReport(snap, mergeOnlineUsers(a.sbOnline.Online(time.Now()), snap)) + if !report.empty() { + if err := a.postTrafficReport(report); err != nil { + // No commit: deltas are recomputed against the same + // baseline next poll, exactly like the xray path. + return err + } + } + + a.sbStats.CommitTrafficSnapshot(snap) + if err := saveSingboxBaseline(a.singboxBaselinePath(), a.sbProc.epoch(), snap.BaselineValues()); err != nil { + // The in-memory commit stands (see the xray path's reasoning). + logger.Warningf("Failed to persist sing-box traffic baseline: %v", err) + return err + } + return nil +} + // LoadConfig loads agent config from file func LoadConfig(path string) (*Config, error) { data, err := os.ReadFile(path) diff --git a/internal/agent/singbox.go b/internal/agent/singbox.go new file mode 100644 index 0000000..8bfdba7 --- /dev/null +++ b/internal/agent/singbox.go @@ -0,0 +1,468 @@ +package agent + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "runtime" + "strings" + "sync" + "syscall" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/config" + "github.com/mhsanaei/3x-ui/v3/internal/logger" +) + +// ITEM6 B3: the agent's sing-box runtime, living alongside the xray +// runtime until the B5 switchover. The master delivers a complete +// sing-box configuration in the same poll response as the xray one +// (singbox_config + singbox_config_hash, B2); the agent consumes it +// independently: a broken sing-box payload, a missing binary or a +// failed check never disturbs the xray path, and vice versa. +// +// Activation borrows NexusGate's discipline (agent-go activateSingBox): +// candidate file -> `sing-box check` -> flush the running core's +// counters -> swap with a .previous copy kept -> start -> on failure +// restore .previous and start that. A config whose bytes are unchanged +// and whose process is alive is a no-op. + +// errSingboxNotInstalled marks "the sing-box binary is not on this +// node yet" — an expected state during the rollout, logged without +// alarming the operator every poll (apply is only attempted when the +// delivered hash changes). +var errSingboxNotInstalled = errors.New("sing-box binary not installed on this node") + +// SingboxVersionRecord is the install-time version record file. The +// self-built sing-box binaries carry no version stamp (`sing-box +// version` prints "unknown"), so install-agent.sh records the release +// version next to the binary for telemetry, NexusGate-style. +const singboxVersionRecord = "sing-box.version" + +// singboxBinaryName mirrors the xray naming: sing-box-linux-. +func singboxBinaryName() string { + arch := runtime.GOARCH + if arch == "arm" { + arch = "arm32" + } + return fmt.Sprintf("sing-box-%s-%s", runtime.GOOS, arch) +} + +// singboxBinaryPath resolves the managed sing-box binary: the +// arch-specific name install-agent.sh installs, falling back to a +// plain "sing-box" in the same bin folder. "" when neither exists. +func singboxBinaryPath() string { + bin := config.GetBinFolderPath() + if p := filepath.Join(bin, singboxBinaryName()); fileExists(p) { + return p + } + if p := filepath.Join(bin, "sing-box"); fileExists(p) { + return p + } + return "" +} + +// singboxConfigPath is the live configuration file the managed process +// runs with; candidate/previous siblings implement the swap discipline. +func singboxConfigPath() string { + return filepath.Join(config.GetBinFolderPath(), "config-singbox.json") +} + +func singboxCandidatePath() string { return singboxConfigPath() + ".candidate" } +func singboxPreviousPath() string { return singboxConfigPath() + ".previous" } + +// singboxLogPath is where the managed process's stdout/stderr land. +// The online tracker (singbox_online.go) tails this file for the +// authenticated-connection lines sing-box logs per user. +func singboxLogPath() string { + return filepath.Join(config.GetBinFolderPath(), "singbox.log") +} + +// singboxProcess manages one sing-box child process: lifecycle only, +// no configuration semantics (mirrors the xray process discipline: +// done channel as the race-free liveness signal, SIGTERM then SIGKILL). +type singboxProcess struct { + mu sync.Mutex + cmd *exec.Cmd + done chan struct{} + logFile *os.File + exitErr error + startedAt time.Time + intentionalStop bool +} + +func (p *singboxProcess) isRunning() bool { + p.mu.Lock() + defer p.mu.Unlock() + if p.cmd == nil || p.cmd.Process == nil || p.done == nil { + return false + } + select { + case <-p.done: + return false + default: + return true + } +} + +// pid returns the running process's pid, 0 when not running. +func (p *singboxProcess) pid() int { + p.mu.Lock() + defer p.mu.Unlock() + if p.cmd == nil || p.cmd.Process == nil { + return 0 + } + select { + case <-p.done: + return 0 + default: + return p.cmd.Process.Pid + } +} + +// epoch identifies this process instance for the stats tracker: +// "sing-box::". The agent starts the +// process itself, so its recorded start time distinguishes instances +// even when the pid is recycled (NexusGate reads the kernel start +// tick for the same reason). "" while no process is running. +func (p *singboxProcess) epoch() string { + p.mu.Lock() + defer p.mu.Unlock() + if p.cmd == nil || p.cmd.Process == nil || p.done == nil { + return "" + } + select { + case <-p.done: + return "" + default: + } + return fmt.Sprintf("sing-box:%d:%d", p.cmd.Process.Pid, p.startedAt.UnixNano()) +} + +// start launches sing-box with the given config file, appending its +// output to the log file the online tracker tails. +func (p *singboxProcess) start(binPath, configPath, logPath string) error { + p.mu.Lock() + defer p.mu.Unlock() + if p.cmd != nil && p.done != nil { + select { + case <-p.done: + default: + return errors.New("sing-box is already running") + } + } + if p.logFile != nil { + _ = p.logFile.Close() + p.logFile = nil + } + logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o640) + if err != nil { + return fmt.Errorf("open sing-box log: %w", err) + } + cmd := exec.Command(binPath, "run", "-c", configPath) + cmd.Stdout = logFile + cmd.Stderr = logFile + if err := cmd.Start(); err != nil { + _ = logFile.Close() + return fmt.Errorf("start sing-box: %w", err) + } + p.cmd = cmd + p.logFile = logFile + p.done = make(chan struct{}) + p.exitErr = nil + p.startedAt = time.Now() + p.intentionalStop = false + done := p.done + go func() { + err := cmd.Wait() + p.mu.Lock() + if err != nil && !p.intentionalStop { + p.exitErr = err + } + p.mu.Unlock() + close(done) + }() + return nil +} + +// stop terminates the process (SIGTERM, then SIGKILL after a grace +// period) and waits for it to be reaped. +func (p *singboxProcess) stop() { + p.mu.Lock() + cmd, done := p.cmd, p.done + if cmd == nil || cmd.Process == nil { + p.mu.Unlock() + return + } + p.intentionalStop = true + proc := cmd.Process + logFile := p.logFile + p.mu.Unlock() + + if done != nil { + select { + case <-done: + p.finishStop(logFile) + return + default: + } + } + if runtime.GOOS == "windows" { + _ = proc.Kill() + } else if err := proc.Signal(syscall.SIGTERM); err != nil { + _ = proc.Kill() + } + if done != nil { + select { + case <-done: + case <-time.After(singboxStopGrace): + _ = proc.Kill() + select { + case <-done: + case <-time.After(singboxStopForce): + } + } + } + p.finishStop(logFile) +} + +func (p *singboxProcess) finishStop(logFile *os.File) { + p.mu.Lock() + p.cmd = nil + p.done = nil + if p.logFile != nil { + p.logFile = nil + } + p.mu.Unlock() + if logFile != nil { + _ = logFile.Close() + } +} + +// exitedErr reports the error the process died with, if it has exited +// unexpectedly. +func (p *singboxProcess) exitedErr() error { + p.mu.Lock() + defer p.mu.Unlock() + return p.exitErr +} + +var ( + singboxStopGrace = 5 * time.Second + singboxStopForce = 2 * time.Second + // singboxStartSettle is the post-start window in which an + // immediate exit means the new config cannot run (the check gate + // catches shape errors earlier; this catches runtime failures + // like a port already in use). + singboxStartSettle = 300 * time.Millisecond + singboxCheckTimeout = 30 * time.Second +) + +// checkSingboxConfig runs `sing-box check` against a candidate file. +func checkSingboxConfig(binPath, candidate string) error { + ctx, cancel := context.WithTimeout(context.Background(), singboxCheckTimeout) + defer cancel() + out, err := exec.CommandContext(ctx, binPath, "check", "-c", candidate).CombinedOutput() + if err != nil { + return fmt.Errorf("sing-box check failed: %w: %s", err, strings.TrimSpace(string(out))) + } + return nil +} + +// singboxStatsListen extracts experimental.v2ray_api.listen from a +// sing-box config, the address the stats client must query. "" when +// the config carries no v2ray_api block (no stats to read). +func singboxStatsListen(raw json.RawMessage) string { + var cfg struct { + Experimental struct { + V2RayAPI struct { + Listen string `json:"listen"` + } `json:"v2ray_api"` + } `json:"experimental"` + } + if err := json.Unmarshal(raw, &cfg); err != nil { + return "" + } + return cfg.Experimental.V2RayAPI.Listen +} + +// applySingboxConfig validates and activates a newly delivered +// sing-box configuration. Errors leave the previously running core +// (if any) completely untouched. +func (a *Agent) applySingboxConfig(raw json.RawMessage) error { + a.sbMu.Lock() + defer a.sbMu.Unlock() + a.ensureSingboxState() + return a.applySingboxConfigLocked(raw) +} + +func (a *Agent) applySingboxConfigLocked(raw json.RawMessage) error { + binPath := singboxBinaryPath() + if binPath == "" { + return errSingboxNotInstalled + } + var probe map[string]any + if err := json.Unmarshal(raw, &probe); err != nil { + return fmt.Errorf("invalid sing-box config JSON: %w", err) + } + + configPath := singboxConfigPath() + candidate := singboxCandidatePath() + if err := os.WriteFile(candidate, raw, 0o600); err != nil { + return fmt.Errorf("write sing-box candidate config: %w", err) + } + if err := checkSingboxConfig(binPath, candidate); err != nil { + _ = os.Remove(candidate) + return err + } + + current, readErr := os.ReadFile(configPath) + unchanged := readErr == nil && string(current) == string(raw) + if unchanged && a.sbProc.isRunning() { + _ = os.Remove(candidate) + return nil + } + + // Flush before change: the running core's counters die with the + // process, so report them while it still lives. A failed flush + // aborts the swap — NexusGate's discipline; the epoch guard in the + // stats tracker is the second line of defence. + if a.sbProc.isRunning() { + if err := a.reportSingboxTrafficLocked(); err != nil { + _ = os.Remove(candidate) + return fmt.Errorf("flush traffic before sing-box config change: %w", err) + } + } + + if !unchanged { + if readErr == nil { + if err := os.WriteFile(singboxPreviousPath(), current, 0o600); err != nil { + _ = os.Remove(candidate) + return fmt.Errorf("keep previous sing-box config: %w", err) + } + } + if err := os.Rename(candidate, configPath); err != nil { + _ = os.Remove(candidate) + return fmt.Errorf("activate sing-box config: %w", err) + } + } else { + _ = os.Remove(candidate) + } + + // The stats address follows the applied config; a config without + // a v2ray_api block clears it so the traffic cycle stays quiet. + a.sbStatsAddr = singboxStatsListen(raw) + + a.sbProc.stop() + if err := a.startSingboxLocked(binPath, configPath); err != nil { + // Roll back to the previous config (if one exists) so the node + // keeps serving; the apply itself reports failure either way. + if rbErr := a.rollbackSingboxLocked(binPath, configPath); rbErr != nil { + logger.Warningf("sing-box rollback failed: %v", rbErr) + } + return err + } + return nil +} + +// startSingboxLocked starts the process and verifies it survives the +// settle window. +func (a *Agent) startSingboxLocked(binPath, configPath string) error { + if err := a.sbProc.start(binPath, configPath, singboxLogPath()); err != nil { + return err + } + time.Sleep(singboxStartSettle) + if !a.sbProc.isRunning() { + if exitErr := a.sbProc.exitedErr(); exitErr != nil { + return fmt.Errorf("sing-box exited immediately after start: %w", exitErr) + } + return errors.New("sing-box exited immediately after start") + } + return nil +} + +// rollbackSingboxLocked restores the .previous config and starts it. +func (a *Agent) rollbackSingboxLocked(binPath, configPath string) error { + previous, err := os.ReadFile(singboxPreviousPath()) + if err != nil { + // Nothing to roll back to: drop the bad config so a later poll + // starts from a clean slate. + _ = os.Remove(configPath) + return fmt.Errorf("no previous sing-box config to roll back to: %w", err) + } + if err := os.WriteFile(configPath, previous, 0o600); err != nil { + return fmt.Errorf("restore previous sing-box config: %w", err) + } + a.sbProc.stop() + return a.startSingboxLocked(binPath, configPath) +} + +// stopSingbox stops the managed sing-box process, if any. +func (a *Agent) stopSingbox() { + a.sbMu.Lock() + defer a.sbMu.Unlock() + if a.sbProc != nil { + a.sbProc.stop() + } +} + +// parseSingboxVersion extracts the version from `sing-box version` +// output ("sing-box version 1.13.21"). Self-built binaries print +// "unknown"; the caller falls back to the install-time record. +func parseSingboxVersion(out string) (string, bool) { + line := out + if i := strings.IndexByte(line, '\n'); i >= 0 { + line = line[:i] + } + fields := strings.Fields(line) + if len(fields) < 3 || !strings.EqualFold(fields[0], "sing-box") || !strings.EqualFold(fields[1], "version") { + return "", false + } + v := fields[2] + if v == "" { + return "", false + } + return v, true +} + +// cachedSingboxVersion mirrors cachedXrayVersion for the sing-box +// binary: re-read only when the binary file changes; an "unknown" +// stamp falls back to the version record install-agent.sh wrote. +func (a *Agent) cachedSingboxVersion() (string, bool) { + path := singboxBinaryPath() + if path == "" { + return "", false + } + fi, err := os.Stat(path) + if err != nil { + return "", false + } + a.sbVerMu.Lock() + defer a.sbVerMu.Unlock() + if a.sbVer != "" && a.sbVerFor == path && fi.ModTime().Equal(a.sbVerMod) { + return a.sbVer, true + } + v := "" + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + out, err := exec.CommandContext(ctx, path, "version").Output() + cancel() + if err == nil { + if parsed, ok := parseSingboxVersion(string(out)); ok && parsed != "unknown" { + v = parsed + } + } + if v == "" { + if data, err := os.ReadFile(filepath.Join(config.GetBinFolderPath(), singboxVersionRecord)); err == nil { + v = strings.TrimSpace(string(data)) + } + } + if v == "" { + return "", false + } + a.sbVer, a.sbVerFor, a.sbVerMod = v, path, fi.ModTime() + return v, true +} diff --git a/internal/agent/singbox_apply_test.go b/internal/agent/singbox_apply_test.go new file mode 100644 index 0000000..289002d --- /dev/null +++ b/internal/agent/singbox_apply_test.go @@ -0,0 +1,211 @@ +package agent + +import ( + "errors" + "os" + "path/filepath" + "strings" + "testing" +) + +// fakeSingbox installs a shell-script sing-box stand-in into a temp +// bin folder: `check` rejects configs containing BADSCHEMA, `run` +// dies instantly on RUNTIMEFAIL and otherwise sleeps (standing in for +// the long-running core). The real binary's behavior is covered by +// the env-gated E2E; these tests pin the agent's activation +// discipline: check gate, no-change no-restart, rollback. +const fakeSingboxScript = `#!/bin/sh +case "$1" in + version) + echo "sing-box version 9.9.9" + exit 0 + ;; + check) + if grep -q BADSCHEMA "$3" 2>/dev/null; then + echo "check: invalid config" >&2 + exit 1 + fi + exit 0 + ;; + run) + if grep -q RUNTIMEFAIL "$3" 2>/dev/null; then + echo "fatal: cannot start" >&2 + exit 1 + fi + exec sleep 3600 + ;; +esac +exit 2 +` + +func setupFakeSingbox(t *testing.T) (binDir string) { + t.Helper() + binDir = t.TempDir() + t.Setenv("XUI_BIN_FOLDER", binDir) + path := filepath.Join(binDir, singboxBinaryName()) + if err := os.WriteFile(path, []byte(fakeSingboxScript), 0o755); err != nil { + t.Fatal(err) + } + return binDir +} + +func newApplyTestAgent() *Agent { + return New(&Config{MasterURL: "http://127.0.0.1:1", Token: "tok", NodeName: "n"}) +} + +const ( + sbCfgA = `{"log":{"level":"info"},"inbounds":[],"outbounds":[{"type":"direct","tag":"direct"}]}` + sbCfgB = `{"log":{"level":"debug"},"inbounds":[],"outbounds":[{"type":"direct","tag":"direct"}]}` +) + +func appliedConfig(t *testing.T) string { + t.Helper() + data, err := os.ReadFile(singboxConfigPath()) + if err != nil { + t.Fatalf("read applied config: %v", err) + } + return string(data) +} + +func TestApplySingboxStartsProcess(t *testing.T) { + setupFakeSingbox(t) + a := newApplyTestAgent() + t.Cleanup(a.stopSingbox) + + if err := a.applySingboxConfig([]byte(sbCfgA)); err != nil { + t.Fatalf("apply: %v", err) + } + a.sbMu.Lock() + running := a.sbProc.isRunning() + pid := a.sbProc.pid() + a.sbMu.Unlock() + if !running || pid <= 0 { + t.Fatal("sing-box process not running after apply") + } + if got := appliedConfig(t); got != sbCfgA { + t.Fatalf("applied config = %q, want cfgA", got) + } + if _, err := os.Stat(singboxCandidatePath()); !os.IsNotExist(err) { + t.Fatal("candidate file must not linger after a successful apply") + } +} + +func TestApplySingboxCheckGateKeepsRunningCore(t *testing.T) { + setupFakeSingbox(t) + a := newApplyTestAgent() + t.Cleanup(a.stopSingbox) + + if err := a.applySingboxConfig([]byte(sbCfgA)); err != nil { + t.Fatalf("apply A: %v", err) + } + a.sbMu.Lock() + pidBefore := a.sbProc.pid() + a.sbMu.Unlock() + + bad := `{"log":{"level":"info"},"inbounds":[{"type":"BADSCHEMA"}]}` + err := a.applySingboxConfig([]byte(bad)) + if err == nil || !strings.Contains(err.Error(), "check") { + t.Fatalf("apply bad config err = %v, want a check failure", err) + } + a.sbMu.Lock() + pidAfter := a.sbProc.pid() + running := a.sbProc.isRunning() + a.sbMu.Unlock() + if !running || pidAfter != pidBefore { + t.Fatalf("check failure disturbed the running core: pid %d -> %d running=%v", pidBefore, pidAfter, running) + } + if got := appliedConfig(t); got != sbCfgA { + t.Fatalf("active config changed after failed check: %q", got) + } +} + +func TestApplySingboxUnchangedNoRestart(t *testing.T) { + setupFakeSingbox(t) + a := newApplyTestAgent() + t.Cleanup(a.stopSingbox) + + if err := a.applySingboxConfig([]byte(sbCfgA)); err != nil { + t.Fatalf("apply A: %v", err) + } + a.sbMu.Lock() + pidBefore := a.sbProc.pid() + a.sbMu.Unlock() + + if err := a.applySingboxConfig([]byte(sbCfgA)); err != nil { + t.Fatalf("re-apply A: %v", err) + } + a.sbMu.Lock() + pidAfter := a.sbProc.pid() + a.sbMu.Unlock() + if pidAfter != pidBefore { + t.Fatalf("unchanged config restarted the core: pid %d -> %d", pidBefore, pidAfter) + } +} + +func TestApplySingboxRuntimeFailureRollsBack(t *testing.T) { + setupFakeSingbox(t) + a := newApplyTestAgent() + t.Cleanup(a.stopSingbox) + + if err := a.applySingboxConfig([]byte(sbCfgA)); err != nil { + t.Fatalf("apply A: %v", err) + } + + bad := `{"log":{"level":"info"},"note":"RUNTIMEFAIL","inbounds":[]}` + if err := a.applySingboxConfig([]byte(bad)); err == nil { + t.Fatal("apply of a config that dies at startup must fail") + } + a.sbMu.Lock() + running := a.sbProc.isRunning() + a.sbMu.Unlock() + if !running { + t.Fatal("rollback must leave a running core") + } + if got := appliedConfig(t); got != sbCfgA { + t.Fatalf("active config after rollback = %q, want cfgA", got) + } +} + +func TestApplySingboxMissingBinary(t *testing.T) { + binDir := t.TempDir() + t.Setenv("XUI_BIN_FOLDER", binDir) + a := newApplyTestAgent() + if err := a.applySingboxConfig([]byte(sbCfgA)); !errors.Is(err, errSingboxNotInstalled) { + t.Fatalf("err = %v, want errSingboxNotInstalled", err) + } +} + +func TestApplySingboxInvalidJSON(t *testing.T) { + setupFakeSingbox(t) + a := newApplyTestAgent() + t.Cleanup(a.stopSingbox) + if err := a.applySingboxConfig([]byte("{not json")); err == nil { + t.Fatal("invalid JSON must fail") + } + a.sbMu.Lock() + proc := a.sbProc + a.sbMu.Unlock() + if proc != nil && proc.isRunning() { + t.Fatal("no process may start from invalid JSON") + } +} + +func TestParseSingboxVersion(t *testing.T) { + cases := []struct { + out string + want string + ok bool + }{ + {"sing-box version 1.13.21\n\nEnvironment: go1.26.9 linux/amd64", "1.13.21", true}, + {"sing-box version unknown\n", "unknown", true}, + {"sing-box version \n", "", false}, + {"xray version 1.0\n", "", false}, + {"", "", false}, + } + for _, tc := range cases { + got, ok := parseSingboxVersion(tc.out) + if got != tc.want || ok != tc.ok { + t.Errorf("parseSingboxVersion(%q) = %q,%v want %q,%v", tc.out, got, ok, tc.want, tc.ok) + } + } +} diff --git a/internal/agent/singbox_e2e_test.go b/internal/agent/singbox_e2e_test.go new file mode 100644 index 0000000..6ad786d --- /dev/null +++ b/internal/agent/singbox_e2e_test.go @@ -0,0 +1,684 @@ +package agent + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/hex" + "encoding/json" + "encoding/pem" + "fmt" + "io" + "math/big" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "golang.org/x/net/proxy" +) + +// ITEM6 B3 end-to-end, env-gated: XUI_SINGBOX_BIN must point at the +// self-built sing-box v1.13.21 (with_v2ray_api with_utls with_quic). +// Skipped otherwise so the package suite stays green anywhere. +// +// Topology: a stub master serves a B2-shaped singbox_config; the real +// Agent applies it and runs the real core; a second core instance is +// the client (vless-reality / ss2022 / mixed / trojan / hysteria2 / +// a forwarding inbound), a third is the forwarding landing. +// Assertions cover: apply+serve, per-user deltas against known byte +// counts, online users, forwarding, flush-before-change + epoch +// rebase across a core restart (no double counting), the check gate. +// +// Environment limits of the build sandbox, probed at runtime: direct +// internet is unreachable (the reality handshake target is a local +// TLS server — production points it at real sites), and UDP sends +// are blocked outright (EPERM), so the hysteria2 leg only joins when +// a UDP loopback probe succeeds; its on-node verification belongs +// to the B5 pilot, like trojan+flow (N5) and wireguard (N7). + +type e2eMaster struct { + mu sync.Mutex + cfg []byte + hash string + reports []trafficReport +} + +func (m *e2eMaster) setConfig(cfg []byte) { + sum := sha256.Sum256(cfg) + m.mu.Lock() + m.cfg = cfg + m.hash = hex.EncodeToString(sum[:]) + m.mu.Unlock() +} + +func (m *e2eMaster) handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/panel/api/agent/register", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("/panel/api/agent/heartbeat", func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + mux.HandleFunc("/panel/api/agent/config", func(w http.ResponseWriter, r *http.Request) { + m.mu.Lock() + defer m.mu.Unlock() + resp := map[string]any{ + "config_hash": "xray-placeholder", + "xray_config": json.RawMessage(`{}`), + "singbox_config": json.RawMessage(m.cfg), + "singbox_config_hash": m.hash, + } + _ = json.NewEncoder(w).Encode(resp) + }) + mux.HandleFunc("/panel/api/agent/traffic", func(w http.ResponseWriter, r *http.Request) { + var rep trafficReport + if err := json.NewDecoder(r.Body).Decode(&rep); err == nil { + m.mu.Lock() + m.reports = append(m.reports, rep) + m.mu.Unlock() + } + w.WriteHeader(http.StatusOK) + }) + return mux +} + +// totals folds every report the master received into per-user and +// per-inbound cumulative deltas plus the last online set. +func (m *e2eMaster) totals() (users map[string][2]int64, inbounds map[string][2]int64, onlines []string) { + m.mu.Lock() + defer m.mu.Unlock() + users = map[string][2]int64{} + inbounds = map[string][2]int64{} + for _, rep := range m.reports { + for _, u := range rep.Users { + v := users[u.Email] + users[u.Email] = [2]int64{v[0] + u.Up, v[1] + u.Down} + } + for _, in := range rep.Inbounds { + v := inbounds[in.Tag] + inbounds[in.Tag] = [2]int64{v[0] + in.Up, v[1] + in.Down} + } + if len(rep.Onlines) > 0 { + onlines = rep.Onlines + } + } + return users, inbounds, onlines +} + +func freePort(t *testing.T) int { + t.Helper() + l, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer l.Close() + return l.Addr().(*net.TCPAddr).Port +} + +func freeUDPPort(t *testing.T) int { + t.Helper() + c, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + if err != nil { + t.Fatal(err) + } + defer c.Close() + return c.LocalAddr().(*net.UDPAddr).Port +} + +// udpLoopbackWorks probes whether UDP over loopback supports a +// full round trip in this environment. A one-way check is not +// enough: the build sandbox lets an unbound client socket send but +// refuses (EPERM) the bound server socket's reply, which no +// QUIC-based protocol can survive. +func udpLoopbackWorks() bool { + srv, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: 0}) + if err != nil { + return false + } + defer srv.Close() + go func() { + buf := make([]byte, 16) + _ = srv.SetReadDeadline(time.Now().Add(3 * time.Second)) + _, addr, err := srv.ReadFromUDP(buf) + if err == nil { + _, _ = srv.WriteToUDP([]byte("pong"), addr) + } + }() + cli, err := net.DialUDP("udp", nil, srv.LocalAddr().(*net.UDPAddr)) + if err != nil { + return false + } + defer cli.Close() + if _, err := cli.Write([]byte("ping")); err != nil { + return false + } + _ = cli.SetReadDeadline(time.Now().Add(3 * time.Second)) + buf := make([]byte, 16) + _, err = cli.Read(buf) + return err == nil +} + +func selfSignedCert(t *testing.T, dir string, dnsNames ...string) (certPath, keyPath string) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: dnsNames[0]}, + DNSNames: dnsNames, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageKeyEncipherment, + ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + certPath = filepath.Join(dir, "cert-"+dnsNames[0]+".pem") + keyPath = filepath.Join(dir, "key-"+dnsNames[0]+".pem") + certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}) + keyDER, err := x509.MarshalECPrivateKey(key) + if err != nil { + t.Fatal(err) + } + keyPEM := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}) + if err := os.WriteFile(certPath, certPEM, 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(keyPath, keyPEM, 0o600); err != nil { + t.Fatal(err) + } + return certPath, keyPath +} + +// startHandshakeServer runs the local TLS endpoint the reality +// server mirrors. +func startHandshakeServer(t *testing.T, certPath, keyPath string) (port int, stop func()) { + t.Helper() + cert, err := tls.LoadX509KeyPair(certPath, keyPath) + if err != nil { + t.Fatal(err) + } + ln, err := tls.Listen("tcp", "127.0.0.1:0", &tls.Config{Certificates: []tls.Certificate{cert}}) + if err != nil { + t.Fatal(err) + } + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func(c net.Conn) { + defer c.Close() + if tc, ok := c.(*tls.Conn); ok { + _ = tc.Handshake() + } + _, _ = io.Copy(io.Discard, c) + }(conn) + } + }() + return ln.Addr().(*net.TCPAddr).Port, func() { _ = ln.Close() } +} + +func runCore(t *testing.T, bin, name string, cfg any, dir string) *exec.Cmd { + t.Helper() + raw, err := json.Marshal(cfg) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(dir, name+".json") + if err := os.WriteFile(path, raw, 0o600); err != nil { + t.Fatal(err) + } + logFile, err := os.Create(filepath.Join(dir, name+".log")) + if err != nil { + t.Fatal(err) + } + cmd := exec.Command(bin, "run", "-c", path) + cmd.Stdout = logFile + cmd.Stderr = logFile + if err := cmd.Start(); err != nil { + t.Fatalf("start %s: %v", name, err) + } + t.Cleanup(func() { + _ = cmd.Process.Kill() + _ = cmd.Wait() + _ = logFile.Close() + }) + return cmd +} + +func waitPort(t *testing.T, port int, what string) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + conn, err := net.DialTimeout("tcp", fmt.Sprintf("127.0.0.1:%d", port), 200*time.Millisecond) + if err == nil { + _ = conn.Close() + return + } + time.Sleep(100 * time.Millisecond) + } + t.Fatalf("port %d (%s) never came up", port, what) +} + +// fetchViaSocks GETs url through the socks5 listener at port and +// returns the body length. +func fetchViaSocks(t *testing.T, port int, url string) int { + t.Helper() + dialer, err := proxy.SOCKS5("tcp", fmt.Sprintf("127.0.0.1:%d", port), nil, proxy.Direct) + if err != nil { + t.Fatal(err) + } + tr := &http.Transport{DialContext: func(_ context.Context, network, addr string) (net.Conn, error) { + return dialer.Dial(network, addr) + }} + client := &http.Client{Transport: tr, Timeout: 15 * time.Second} + resp, err := client.Get(url) + if err != nil { + t.Fatalf("GET via socks %d: %v", port, err) + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + t.Fatal(err) + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET via socks %d: status %d", port, resp.StatusCode) + } + return len(body) +} + +func TestSingboxEndToEnd(t *testing.T) { + bin := os.Getenv("XUI_SINGBOX_BIN") + if bin == "" { + t.Skip("XUI_SINGBOX_BIN not set; skipping real-core E2E") + } + if _, err := os.Stat(bin); err != nil { + t.Skipf("sing-box binary unusable: %v", err) + } + dir := t.TempDir() + udpOK := udpLoopbackWorks() + if !udpOK { + t.Log("UDP loopback unavailable in this environment; hysteria2 leg excluded (pilot-node item)") + } + + // --- fixtures --------------------------------------------------- + gen := func(args ...string) string { + out, err := exec.Command(bin, args...).Output() + if err != nil { + t.Fatalf("sing-box %v: %v", args, err) + } + return strings.TrimSpace(string(out)) + } + var privKey, pubKey string + for _, line := range strings.Split(gen("generate", "reality-keypair"), "\n") { + if strings.HasPrefix(line, "PrivateKey:") { + privKey = strings.TrimSpace(strings.TrimPrefix(line, "PrivateKey:")) + } + if strings.HasPrefix(line, "PublicKey:") { + pubKey = strings.TrimSpace(strings.TrimPrefix(line, "PublicKey:")) + } + } + aliceUUID := gen("generate", "uuid") + ssServerPass := strings.TrimSpace(gen("generate", "rand", "--base64", "16")) + ssBobPass := strings.TrimSpace(gen("generate", "rand", "--base64", "16")) + + hsCert, hsKey := selfSignedCert(t, dir, "www.microsoft.com") + hsPort, stopHS := startHandshakeServer(t, hsCert, hsKey) + defer stopHS() + tlsCert, tlsKey := selfSignedCert(t, dir, "example.com") + + targetBody := strings.Repeat("0123456789", 500) // 5000 bytes + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.WriteString(w, targetBody) + })) + defer target.Close() + targetURL := target.URL + "/" + + pVless := freePort(t) + pSS := freePort(t) + pMixed := freePort(t) + pFwd := freePort(t) + pTrojan := freePort(t) + pHy2 := freeUDPPort(t) + pStats := freePort(t) + pLanding := freePort(t) + cVless := freePort(t) + cSS := freePort(t) + cMixed := freePort(t) + cFwd := freePort(t) + cTrojan := freePort(t) + cHy2 := freePort(t) + + type leg struct { + name string + clientPort int + user string + serverTag string + } + legs := []leg{ + {"vless-reality", cVless, "alice@example.com", "vless-reality"}, + {"ss2022", cSS, "bob@example.com", "ss22"}, + {"mixed", cMixed, "carol@example.com", "mixed-in"}, + {"trojan-tls", cTrojan, "frank@example.com", "trojan-in"}, + {"forwarding", cFwd, "dave@example.com", "mixed-fwd"}, + } + if udpOK { + legs = append(legs, leg{"hysteria2", cHy2, "erin@example.com", "hy2-in"}) + } + + serverInbounds := []any{ + map[string]any{ + "type": "vless", "tag": "vless-reality", "listen": "127.0.0.1", "listen_port": pVless, + "users": []any{map[string]any{"name": "alice@example.com", "uuid": aliceUUID, "flow": "xtls-rprx-vision"}}, + "tls": map[string]any{ + "enabled": true, "server_name": "www.microsoft.com", + "reality": map[string]any{ + "enabled": true, + "handshake": map[string]any{"server": "127.0.0.1", "server_port": hsPort}, + "private_key": privKey, "short_id": []string{"abcd"}, + }, + }, + }, + map[string]any{ + "type": "shadowsocks", "tag": "ss22", "listen": "127.0.0.1", "listen_port": pSS, + "method": "2022-blake3-aes-128-gcm", "password": ssServerPass, + "users": []any{map[string]any{"name": "bob@example.com", "password": ssBobPass}}, + }, + map[string]any{ + "type": "mixed", "tag": "mixed-in", "listen": "127.0.0.1", "listen_port": pMixed, + "users": []any{map[string]any{"username": "carol@example.com", "password": "carolpass"}}, + }, + map[string]any{ + "type": "mixed", "tag": "mixed-fwd", "listen": "127.0.0.1", "listen_port": pFwd, + "users": []any{map[string]any{"username": "dave@example.com", "password": "davepass"}}, + }, + map[string]any{ + "type": "trojan", "tag": "trojan-in", "listen": "127.0.0.1", "listen_port": pTrojan, + "users": []any{map[string]any{"name": "frank@example.com", "password": "frankpass"}}, + "tls": map[string]any{"enabled": true, "server_name": "example.com", "certificate_path": tlsCert, "key_path": tlsKey}, + }, + } + statsInbounds := []string{"vless-reality", "ss22", "mixed-in", "mixed-fwd", "trojan-in"} + statsUsers := []string{"alice@example.com", "bob@example.com", "carol@example.com", "dave@example.com", "frank@example.com"} + if udpOK { + serverInbounds = append(serverInbounds, map[string]any{ + "type": "hysteria2", "tag": "hy2-in", "listen": "127.0.0.1", "listen_port": pHy2, + "users": []any{map[string]any{"name": "erin@example.com", "password": "erinpass"}}, + "tls": map[string]any{"enabled": true, "server_name": "example.com", "certificate_path": tlsCert, "key_path": tlsKey}, + }) + statsInbounds = append(statsInbounds, "hy2-in") + statsUsers = append(statsUsers, "erin@example.com") + } + + buildServerCfg := func(logLevel string) map[string]any { + return map[string]any{ + "log": map[string]any{"level": logLevel, "timestamp": true}, + "inbounds": serverInbounds, + "outbounds": []any{ + map[string]any{"type": "direct", "tag": "direct"}, + map[string]any{"type": "socks", "tag": "fwd-1", "server": "127.0.0.1", "server_port": pLanding, "version": "5"}, + }, + "route": map[string]any{ + "final": "direct", + "rules": []any{ + map[string]any{"inbound": []string{"mixed-fwd"}, "action": "route", "outbound": "fwd-1"}, + }, + }, + "experimental": map[string]any{ + "v2ray_api": map[string]any{ + "listen": fmt.Sprintf("127.0.0.1:%d", pStats), + "stats": map[string]any{ + "enabled": true, + "inbounds": statsInbounds, + "outbounds": []string{"direct", "fwd-1"}, + "users": statsUsers, + }, + }, + }, + } + } + + // Landing for the forwarding leg: a bare no-auth mixed inbound. + runCore(t, bin, "landing", map[string]any{ + "log": map[string]any{"level": "warn"}, + "inbounds": []any{map[string]any{"type": "mixed", "tag": "landing-in", "listen": "127.0.0.1", "listen_port": pLanding}}, + "outbounds": []any{map[string]any{"type": "direct", "tag": "direct"}}, + }, dir) + waitPort(t, pLanding, "landing") + + // Client core: one local socks port per protocol leg. + clientInbounds := []any{ + map[string]any{"type": "mixed", "tag": "in-vless", "listen": "127.0.0.1", "listen_port": cVless}, + map[string]any{"type": "mixed", "tag": "in-ss", "listen": "127.0.0.1", "listen_port": cSS}, + map[string]any{"type": "mixed", "tag": "in-mixed", "listen": "127.0.0.1", "listen_port": cMixed}, + map[string]any{"type": "mixed", "tag": "in-fwd", "listen": "127.0.0.1", "listen_port": cFwd}, + map[string]any{"type": "mixed", "tag": "in-trojan", "listen": "127.0.0.1", "listen_port": cTrojan}, + } + clientOutbounds := []any{ + map[string]any{ + "type": "vless", "tag": "out-vless", "server": "127.0.0.1", "server_port": pVless, + "uuid": aliceUUID, "flow": "xtls-rprx-vision", + "tls": map[string]any{ + "enabled": true, "server_name": "www.microsoft.com", + "utls": map[string]any{"enabled": true, "fingerprint": "chrome"}, + "reality": map[string]any{"enabled": true, "public_key": pubKey, "short_id": "abcd"}, + }, + }, + map[string]any{ + "type": "shadowsocks", "tag": "out-ss", "server": "127.0.0.1", "server_port": pSS, + "method": "2022-blake3-aes-128-gcm", "password": ssServerPass + ":" + ssBobPass, + }, + map[string]any{"type": "socks", "tag": "out-mixed", "server": "127.0.0.1", "server_port": pMixed, "version": "5", + "username": "carol@example.com", "password": "carolpass"}, + map[string]any{"type": "socks", "tag": "out-fwd", "server": "127.0.0.1", "server_port": pFwd, "version": "5", + "username": "dave@example.com", "password": "davepass"}, + map[string]any{ + "type": "trojan", "tag": "out-trojan", "server": "127.0.0.1", "server_port": pTrojan, "password": "frankpass", + "tls": map[string]any{"enabled": true, "server_name": "example.com", "insecure": true}, + }, + } + clientRules := []any{ + map[string]any{"inbound": []string{"in-vless"}, "action": "route", "outbound": "out-vless"}, + map[string]any{"inbound": []string{"in-ss"}, "action": "route", "outbound": "out-ss"}, + map[string]any{"inbound": []string{"in-mixed"}, "action": "route", "outbound": "out-mixed"}, + map[string]any{"inbound": []string{"in-fwd"}, "action": "route", "outbound": "out-fwd"}, + map[string]any{"inbound": []string{"in-trojan"}, "action": "route", "outbound": "out-trojan"}, + } + clientPorts := []int{cVless, cSS, cMixed, cFwd, cTrojan} + if udpOK { + clientInbounds = append(clientInbounds, + map[string]any{"type": "mixed", "tag": "in-hy2", "listen": "127.0.0.1", "listen_port": cHy2}) + clientOutbounds = append(clientOutbounds, map[string]any{ + "type": "hysteria2", "tag": "out-hy2", "server": "127.0.0.1", "server_port": pHy2, "password": "erinpass", + "tls": map[string]any{"enabled": true, "server_name": "example.com", "insecure": true, "alpn": []string{"h3"}}, + }) + clientRules = append(clientRules, + map[string]any{"inbound": []string{"in-hy2"}, "action": "route", "outbound": "out-hy2"}) + clientPorts = append(clientPorts, cHy2) + } + clientOutbounds = append(clientOutbounds, map[string]any{"type": "direct", "tag": "direct"}) + runCore(t, bin, "client", map[string]any{ + "log": map[string]any{"level": "warn"}, + "inbounds": clientInbounds, + "outbounds": clientOutbounds, + "route": map[string]any{"final": "direct", "rules": clientRules}, + }, dir) + for _, p := range clientPorts { + waitPort(t, p, "client") + } + + // --- agent under test -------------------------------------------- + binDir := filepath.Join(dir, "bin") + if err := os.MkdirAll(binDir, 0o755); err != nil { + t.Fatal(err) + } + linkPath := filepath.Join(binDir, singboxBinaryName()) + if err := os.Symlink(bin, linkPath); err != nil { + data, rerr := os.ReadFile(bin) + if rerr != nil { + t.Fatal(rerr) + } + if werr := os.WriteFile(linkPath, data, 0o755); werr != nil { + t.Fatal(werr) + } + } + t.Setenv("XUI_BIN_FOLDER", binDir) + + master := &e2eMaster{} + cfgA, _ := json.Marshal(buildServerCfg("info")) + master.setConfig(cfgA) + srv := httptest.NewServer(master.handler()) + defer srv.Close() + + agt := New(&Config{ + MasterURL: srv.URL, + Token: "e2e-token", + NodeName: "e2e", + BaselinePath: filepath.Join(dir, "agent-traffic-baseline.json"), + }) + t.Cleanup(agt.stopSingbox) + + // Apply. The xray half of the poll fails by design (no xray + // binary in the sandbox bin folder); the sing-box half must + // have succeeded before that error is returned. + _ = agt.fetchAndApplyConfig() + agt.sbMu.Lock() + running := agt.sbProc != nil && agt.sbProc.isRunning() + agt.sbMu.Unlock() + if !running { + t.Fatal("agent did not start sing-box from the delivered config") + } + for _, p := range []int{pVless, pSS, pMixed, pFwd, pTrojan, pStats} { + waitPort(t, p, "agent sing-box") + } + + cycle := func() { + t.Helper() + if err := agt.collectAndReportSingboxTraffic(); err != nil { + t.Fatalf("traffic cycle: %v", err) + } + } + + // First cycle only establishes the baseline: traffic generated + // before it must NOT show up as deltas. + if n := fetchViaSocks(t, cSS, targetURL); n != len(targetBody) { + t.Fatalf("ss leg body = %d, want %d", n, len(targetBody)) + } + cycle() + users, _, _ := master.totals() + if len(users) != 0 { + t.Fatalf("first cycle reported user deltas before baseline: %v", users) + } + + // --- per-protocol traffic, then a reporting cycle ----------------- + downloads := map[string]int{} + for _, lg := range legs { + n := fetchViaSocks(t, lg.clientPort, targetURL) + if n != len(targetBody) { + t.Fatalf("%s leg body = %d, want %d", lg.name, n, len(targetBody)) + } + downloads[lg.user] += n + } + cycle() + + users, inbounds, onlines := master.totals() + for _, lg := range legs { + got := users[lg.user] + want := int64(downloads[lg.user]) + if got[1] < want || got[1] > want*3/2+2048 { + t.Fatalf("%s (%s): reported down = %d, want within [%d, %d]", lg.name, lg.user, got[1], want, want*3/2+2048) + } + if got[0] <= 0 { + t.Fatalf("%s (%s): reported up = %d, want > 0", lg.name, lg.user, got[0]) + } + found := false + for _, e := range onlines { + if e == lg.user { + found = true + } + } + if !found { + t.Fatalf("%s (%s) missing from onlines %v", lg.name, lg.user, onlines) + } + if got := inbounds[lg.serverTag]; got[1] <= 0 { + t.Fatalf("inbound %s reported down = %d, want > 0", lg.serverTag, got[1]) + } + } + + // --- config change: flush + epoch rebase, no double counting ----- + if n := fetchViaSocks(t, cSS, targetURL); n != len(targetBody) { + t.Fatalf("pre-change ss body = %d", n) + } + downloads["bob@example.com"] += len(targetBody) + cfgB, _ := json.Marshal(buildServerCfg("debug")) + master.setConfig(cfgB) + agt.sbMu.Lock() + pidBefore := agt.sbProc.pid() + agt.sbMu.Unlock() + _ = agt.fetchAndApplyConfig() + agt.sbMu.Lock() + pidAfter := agt.sbProc.pid() + running = agt.sbProc.isRunning() + agt.sbMu.Unlock() + if !running || pidAfter == pidBefore { + t.Fatalf("config change did not restart the core: pid %d -> %d running=%v", pidBefore, pidAfter, running) + } + for _, p := range []int{pVless, pSS, pMixed, pFwd, pTrojan, pStats} { + waitPort(t, p, "agent sing-box after restart") + } + + if n := fetchViaSocks(t, cSS, targetURL); n != len(targetBody) { + t.Fatalf("post-change ss body = %d", n) + } + downloads["bob@example.com"] += len(targetBody) + cycle() + cycle() // a second cycle must add nothing new + + users, _, _ = master.totals() + bob := users["bob@example.com"] + want := int64(downloads["bob@example.com"]) + if bob[1] < want || bob[1] > want*3/2+4096 { + t.Fatalf("bob total down after restart = %d, want within [%d, %d] (double counting or loss)", bob[1], want, want*3/2+4096) + } + + // --- the check gate ------------------------------------------------- + badCfg := []byte(`{"inbounds":[{"type":"no-such-type","tag":"x"}],"outbounds":[{"type":"direct","tag":"direct"}]}`) + badSum := sha256.Sum256(badCfg) + master.setConfig(badCfg) + _ = agt.fetchAndApplyConfig() + if agt.lastSingboxHash == hex.EncodeToString(badSum[:]) { + t.Fatal("bad config hash must not be recorded as applied") + } + agt.sbMu.Lock() + pidNow := agt.sbProc.pid() + running = agt.sbProc.isRunning() + agt.sbMu.Unlock() + if !running || pidNow != pidAfter { + t.Fatalf("check-failing config disturbed the core: pid %d -> %d running=%v", pidAfter, pidNow, running) + } + applied, err := os.ReadFile(singboxConfigPath()) + if err != nil { + t.Fatal(err) + } + if string(applied) != string(cfgB) { + t.Fatal("active config file changed after a rejected apply") + } +} diff --git a/internal/agent/singbox_online.go b/internal/agent/singbox_online.go new file mode 100644 index 0000000..f3f7206 --- /dev/null +++ b/internal/agent/singbox_online.go @@ -0,0 +1,182 @@ +package agent + +import ( + "bytes" + "os" + "regexp" + "sync" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/xray" +) + +// ITEM6 B3: online users for the sing-box runtime (the §4.4 spike, +// concluded here). +// +// sing-box's v2ray_api has no online-user source. The access log +// does: for every authenticated connection sing-box logs +// +// [ ] inbound/[]: [] inbound connection to +// +// with the same user name the stats counters use (verified against +// the real v1.13.21 binary for vless, shadowsocks and mixed; +// unauthenticated attempts only ever produce the preceding +// "inbound connection from" line, never this one). The tracker tails +// the managed process's log file from a cursor — NexusGate's +// observation discipline: a truncated/rotated file resets the cursor, +// reads are window-capped, only complete lines are consumed. +// +// Semantics versus the xray core's live-connection view: a user is +// online while an authenticated connection was observed within +// singboxOnlineGrace, OR while their counters moved in the latest +// snapshot (the traffic cycle merges that in — it covers long-lived +// connections that keep transferring). A connected-but-idle user +// ages out after the grace window: sing-box logs nothing per +// connection after setup, so idle-liveness is not observable without +// opening the clash_api connection table, which the §4.4 evidence +// base rules out. The wire contract is unchanged (onlines = [email]). + +const singboxOnlineGrace = 150 * time.Second + +// maxSingboxLogWindow caps one poll's read, NexusGate-style: when the +// cursor falls further behind (long downtime), only the tail is +// scanned for recent users. +const maxSingboxLogWindow = 2 * 1024 * 1024 + +var ( + // ANSI CSI sequences (sing-box colorizes the level and conn-id + // even when redirected to a file) are stripped before matching. + ansiCSIRegex = regexp.MustCompile("\x1b\\[[0-9;?]*[ -/]*[@-~]") + // The authenticated-connection line. The user name is sing-box's + // stats identity: email for most protocols, username for + // mixed/socks/http. + singboxAuthLineRegex = regexp.MustCompile(`\[([^\]\s]+)\] inbound connection to `) +) + +// singboxOnlineTracker derives the online-user set from the managed +// process's access log. Safe for concurrent use. +type singboxOnlineTracker struct { + path string + + mu sync.Mutex + offset int64 + lastSeen map[string]int64 // user -> unix seconds of latest authenticated connection +} + +func newSingboxOnlineTracker(path string) *singboxOnlineTracker { + return &singboxOnlineTracker{path: path, lastSeen: make(map[string]int64)} +} + +// stripANSI removes terminal color/control sequences from one log line. +func stripANSI(line string) string { + if !bytes.ContainsRune([]byte(line), 0x1b) { + return line + } + return ansiCSIRegex.ReplaceAllString(line, "") +} + +// parseSingboxAuthLine extracts the user from one raw log line, or "" +// when the line is not an authenticated-connection record. +func parseSingboxAuthLine(line string) string { + m := singboxAuthLineRegex.FindStringSubmatch(stripANSI(line)) + if m == nil { + return "" + } + return m[1] +} + +// Poll consumes the log written since the previous poll and records +// the users it proves active. Poll errors are swallowed by design: +// online data is best-effort and must never fail a traffic cycle. +func (t *singboxOnlineTracker) Poll(now time.Time) { + t.mu.Lock() + defer t.mu.Unlock() + + info, err := os.Stat(t.path) + if err != nil { + return + } + size := info.Size() + if size < t.offset { + t.offset = 0 // rotated/truncated: rescan from the top + } + start := t.offset + if floor := size - maxSingboxLogWindow; floor > start { + start = floor + } + if start >= size { + t.offset = size + return + } + f, err := os.Open(t.path) + if err != nil { + return + } + defer f.Close() + buf := make([]byte, size-start) + if _, err := f.ReadAt(buf, start); err != nil { + return + } + end := bytes.LastIndex(buf, []byte("\n")) + if end < 0 { + return // no complete line yet; retry next poll + } + ts := now.Unix() + for _, line := range bytes.Split(buf[:end], []byte("\n")) { + if user := parseSingboxAuthLine(string(line)); user != "" { + t.lastSeen[user] = ts + } + } + t.offset = start + int64(end) + 1 + // Prune entries far beyond the grace window so the map stays + // bounded on long-lived nodes. + cutoff := ts - int64(singboxOnlineGrace/time.Second)*4 + for user, seen := range t.lastSeen { + if seen < cutoff { + delete(t.lastSeen, user) + } + } +} + +// Online returns the users seen within the grace window. +func (t *singboxOnlineTracker) Online(now time.Time) []string { + t.mu.Lock() + defer t.mu.Unlock() + cutoff := now.Unix() - int64(singboxOnlineGrace/time.Second) + out := make([]string, 0, len(t.lastSeen)) + for user, seen := range t.lastSeen { + if seen >= cutoff { + out = append(out, user) + } + } + return out +} + +// mergeOnlineUsers unions the log tracker's view with the users whose +// counters moved in the latest snapshot (activity within the last +// poll interval), deduplicated, in xray.OnlineUser wire shape. +func mergeOnlineUsers(logUsers []string, snap *xray.TrafficSnapshot) []xray.OnlineUser { + seen := make(map[string]struct{}) + out := make([]xray.OnlineUser, 0, len(logUsers)) + add := func(email string) { + if email == "" { + return + } + if _, dup := seen[email]; dup { + return + } + seen[email] = struct{}{} + out = append(out, xray.OnlineUser{Email: email}) + } + for _, u := range logUsers { + add(u) + } + if snap != nil { + for _, ct := range snap.ClientTraffics { + if ct != nil && ct.Up+ct.Down > 0 { + add(ct.Email) + } + } + } + return out +} diff --git a/internal/agent/singbox_online_test.go b/internal/agent/singbox_online_test.go new file mode 100644 index 0000000..253e847 --- /dev/null +++ b/internal/agent/singbox_online_test.go @@ -0,0 +1,124 @@ +package agent + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// Real line shapes captured from sing-box v1.13.21 in the B3 spike, +// ANSI color sequences included (the core colorizes the level and +// the connection id even when its output is redirected to a file). +const ( + spikeAuthSS = "+0000 2026-10-22 21:54:58 \x1b[36mINFO\x1b[0m [\x1b[38;5;51m1785752244\x1b[0m 0ms] inbound/shadowsocks[ss22]: [bob@example.com] inbound connection to 127.0.0.1:18080" + spikeAuthMixed = "+0000 2026-10-22 21:54:58 \x1b[36mINFO\x1b[0m [\x1b[38;5;231m1630640343\x1b[0m 0ms] inbound/mixed[mixed-in]: [carol@example.com] inbound connection to 127.0.0.1:18080" + spikeFromLine = "+0000 2026-10-22 21:54:58 \x1b[36mINFO\x1b[0m [\x1b[38;5;51m1785752244\x1b[0m 0ms] inbound/shadowsocks[ss22]: inbound connection from 127.0.0.1:52076" + spikeErrorLine = "+0000 2026-10-22 21:54:57 \x1b[31mERROR\x1b[0m [\x1b[38;5;158m398453065\x1b[0m 85ms] inbound/vless[vless-reality]: process connection from 127.0.0.1:42732: TLS handshake: REALITY: processed invalid connection" + spikeStartLine = "+0000 2026-10-22 21:53:31 \x1b[36mINFO\x1b[0m inbound/mixed[mixed-in]: tcp server started at 127.0.0.1:24445" +) + +func TestParseSingboxAuthLine(t *testing.T) { + cases := []struct { + line string + want string + }{ + {spikeAuthSS, "bob@example.com"}, + {spikeAuthMixed, "carol@example.com"}, + {spikeFromLine, ""}, // unauthenticated stage: never counts + {spikeErrorLine, ""}, // failed handshake: never counts + {spikeStartLine, ""}, + {"", ""}, + } + for _, tc := range cases { + if got := parseSingboxAuthLine(tc.line); got != tc.want { + t.Errorf("parse(%q) = %q, want %q", tc.line, got, tc.want) + } + } +} + +func TestSingboxOnlineTrackerPollAndGrace(t *testing.T) { + path := filepath.Join(t.TempDir(), "singbox.log") + now := time.Unix(1_700_000_000, 0) + tr := newSingboxOnlineTracker(path) + + // No file yet: quiet no-op. + tr.Poll(now) + if got := tr.Online(now); len(got) != 0 { + t.Fatalf("online = %v, want empty", got) + } + + content := spikeFromLine + "\n" + spikeAuthSS + "\n" + spikeErrorLine + "\n" + if err := os.WriteFile(path, []byte(content), 0o640); err != nil { + t.Fatal(err) + } + tr.Poll(now) + got := tr.Online(now) + if len(got) != 1 || got[0] != "bob@example.com" { + t.Fatalf("online = %v, want [bob@example.com]", got) + } + + // Re-poll without new content: no double counting, same set. + tr.Poll(now) + if got := tr.Online(now); len(got) != 1 { + t.Fatalf("online after idle poll = %v", got) + } + + // Past the grace window the user ages out. + if got := tr.Online(now.Add(singboxOnlineGrace + time.Second)); len(got) != 0 { + t.Fatalf("online after grace = %v, want empty", got) + } + + // A partial line (no trailing newline) is not consumed until it + // completes. (bob is still within his grace window here; only + // carol's appearance is being timed.) + partial := spikeAuthMixed // no trailing \n yet + if err := appendToFile(path, partial); err != nil { + t.Fatal(err) + } + later := now.Add(time.Minute) + tr.Poll(later) + if got := tr.Online(later); strings.Contains(strings.Join(got, ","), "carol@example.com") { + t.Fatalf("partial line must not count, online = %v", got) + } + if err := appendToFile(path, "\n"); err != nil { + t.Fatal(err) + } + tr.Poll(later) + got = tr.Online(later) + if !strings.Contains(strings.Join(got, ","), "carol@example.com") { + t.Fatalf("online after line completes = %v, want carol@example.com present", got) + } +} + +func TestSingboxOnlineTrackerTruncation(t *testing.T) { + path := filepath.Join(t.TempDir(), "singbox.log") + now := time.Unix(1_700_000_000, 0) + tr := newSingboxOnlineTracker(path) + if err := os.WriteFile(path, []byte(spikeAuthSS+"\n"+spikeAuthSS+"\n"), 0o640); err != nil { + t.Fatal(err) + } + tr.Poll(now) + + // File shrinks below the cursor (rotation): rescan from the top. + if err := os.WriteFile(path, []byte(spikeAuthMixed+"\n"), 0o640); err != nil { + t.Fatal(err) + } + tr.Poll(now) + got := tr.Online(now) + joined := strings.Join(got, ",") + if !strings.Contains(joined, "carol@example.com") || !strings.Contains(joined, "bob@example.com") { + t.Fatalf("online after truncation = %v, want both users", got) + } +} + +func appendToFile(path, s string) error { + f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0) + if err != nil { + return err + } + defer f.Close() + _, err = f.WriteString(s) + return err +} diff --git a/internal/agent/singbox_stats.go b/internal/agent/singbox_stats.go new file mode 100644 index 0000000..140eec9 --- /dev/null +++ b/internal/agent/singbox_stats.go @@ -0,0 +1,380 @@ +package agent + +import ( + "context" + "errors" + "fmt" + "regexp" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/xray" + + "google.golang.org/grpc" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/encoding" + "google.golang.org/protobuf/encoding/protowire" +) + +// ITEM6 B3: per-user traffic accounting for the sing-box runtime. +// +// §4.4 of the scope offered two shapes: embed sing-box's v2rayapi Go +// package, or ship NexusGate's stats-helper binary. This client takes +// the first shape without the dependency: sing-box's v2ray_api is a +// different protobuf service from xray's, but its messages are just +// as small and stable ({patterns} in, {name, value} pairs out), so — +// exactly as this build already does for xray in internal/xray/api.go +// after dropping the core dependency — the client speaks it at the +// protobuf wire level with protowire. Verified against the real +// v1.13.21 binary in the B3 end-to-end test; the helper binary remains +// the documented fallback if the wire shape ever moves. +// +// Counter names are identical to xray's +// (inbound>>>tag>>>traffic>>>uplink, user>>>name>>>traffic>>>downlink), +// and the master-side write path keys users by the same name (=email), +// so only this agent-side query implementation changed. +// +// Epoch discipline (NexusGate): counters belong to one core process +// instance. A snapshot is only meaningful when the core's epoch is +// unchanged across the query — a sample straddling a restart mixes +// two instances' counters and is discarded whole. When the epoch has +// flipped since the committed baseline, the new instance's counters +// started at zero, so they count IN FULL as fresh deltas (the panel +// side's ApplyDelta rule, and the same outcome as the xray path's +// "counter below baseline counts from zero"). Together with +// flush-before-change in the activation path — which reports the +// old instance's tail before it dies — a core restart neither +// double-reports nor loses the new instance's traffic. The one +// conservative exception: the very first snapshot an agent ever +// takes is baseline-only (it may be reading counters that predate +// the agent), exactly like the xray path's first poll. + +const singboxStatsServiceQuery = "/v2ray.core.app.stats.command.StatsService/QueryStats" + +// errSingboxEpochChanged aborts a snapshot whose core instance moved +// mid-query (or is unknown); the next poll retries against the new +// instance. It is a routine event around config changes, not a fault. +var errSingboxEpochChanged = errors.New("sing-box restarted during stats query") + +var ( + singboxTrafficRegex = regexp.MustCompile(`(inbound|outbound)>>>([^>]+)>>>traffic>>>(downlink|uplink)`) + singboxClientTrafficRegex = regexp.MustCompile(`user>>>([^>]+)>>>traffic>>>(downlink|uplink)`) +) + +// sbRawBytes / sbRawCodec mirror the xray package's raw protobuf +// pass-through codec (the standard "proto" content subtype). +type sbRawBytes []byte + +type sbRawCodec struct{} + +func (sbRawCodec) Marshal(v any) ([]byte, error) { + b, ok := v.(sbRawBytes) + if !ok { + return nil, fmt.Errorf("sing-box stats: raw codec cannot marshal %T", v) + } + return b, nil +} + +func (sbRawCodec) Unmarshal(data []byte, v any) error { + p, ok := v.(*sbRawBytes) + if !ok { + return fmt.Errorf("sing-box stats: raw codec cannot unmarshal into %T", v) + } + *p = append((*p)[:0], data...) + return nil +} + +func (sbRawCodec) Name() string { return "proto" } + +var _ encoding.Codec = sbRawCodec{} + +// singboxStatEntry is one counter from a QueryStats response +// (v2ray.core.app.stats.command.Stat: name = 1, value = 2). +type singboxStatEntry struct { + Name string + Value int64 +} + +// encodeSingboxQuery builds the QueryStatsRequest wire form: +// patterns is field 3 (repeated string). Asking for the three counter +// families keeps the response free of anything else the core counts. +func encodeSingboxQuery() []byte { + var b []byte + for _, pattern := range []string{"inbound>>>", "outbound>>>", "user>>>"} { + b = protowire.AppendTag(b, 3, protowire.BytesType) + b = protowire.AppendString(b, pattern) + } + return b +} + +// parseSingboxStatsResponse decodes a QueryStatsResponse +// (field 1: repeated Stat). +func parseSingboxStatsResponse(b []byte) ([]singboxStatEntry, error) { + var out []singboxStatEntry + for len(b) > 0 { + num, typ, n := protowire.ConsumeTag(b) + if n < 0 { + return nil, protowire.ParseError(n) + } + b = b[n:] + if num == 1 && typ == protowire.BytesType { + v, n := protowire.ConsumeBytes(b) + if n < 0 { + return nil, protowire.ParseError(n) + } + st, err := parseSingboxStat(v) + if err != nil { + return nil, err + } + out = append(out, st) + b = b[n:] + continue + } + n = protowire.ConsumeFieldValue(num, typ, b) + if n < 0 { + return nil, protowire.ParseError(n) + } + b = b[n:] + } + return out, nil +} + +func parseSingboxStat(b []byte) (singboxStatEntry, error) { + var st singboxStatEntry + for len(b) > 0 { + num, typ, n := protowire.ConsumeTag(b) + if n < 0 { + return st, protowire.ParseError(n) + } + b = b[n:] + switch { + case num == 1 && typ == protowire.BytesType: + v, n := protowire.ConsumeBytes(b) + if n < 0 { + return st, protowire.ParseError(n) + } + st.Name = string(v) + b = b[n:] + case num == 2 && typ == protowire.VarintType: + v, n := protowire.ConsumeVarint(b) + if n < 0 { + return st, protowire.ParseError(n) + } + st.Value = int64(v) + b = b[n:] + default: + n := protowire.ConsumeFieldValue(num, typ, b) + if n < 0 { + return st, protowire.ParseError(n) + } + b = b[n:] + } + } + return st, nil +} + +// singboxStatsClient is the sing-box counterpart of xray.XrayAPI for +// the agent reporting path: same snapshot/commit contract (snapshots +// never advance the baseline; only an acknowledged report commits), +// plus the epoch guard described above. Not safe for concurrent use; +// the agent serializes it under sbMu. +type singboxStatsClient struct { + addr string + epochFunc func() string + + conn *grpc.ClientConn + + committed map[string]int64 + committedReady bool + committedEpoch string + pendingEpoch string + + // queryFn replaces the gRPC query in tests. + queryFn func(ctx context.Context) ([]singboxStatEntry, error) +} + +func newSingboxStatsClient(epochFunc func() string) *singboxStatsClient { + return &singboxStatsClient{epochFunc: epochFunc} +} + +// SetAddr points the client at the v2ray_api listen address of the +// currently applied config. A changed address drops the connection. +func (c *singboxStatsClient) SetAddr(addr string) { + if addr == c.addr { + return + } + c.addr = addr + c.closeConn() +} + +func (c *singboxStatsClient) closeConn() { + if c.conn != nil { + _ = c.conn.Close() + c.conn = nil + } +} + +func (c *singboxStatsClient) queryStats(ctx context.Context) ([]singboxStatEntry, error) { + if c.queryFn != nil { + return c.queryFn(ctx) + } + if c.addr == "" { + return nil, errors.New("sing-box stats address unknown (config carries no v2ray_api listen)") + } + if c.conn == nil { + conn, err := grpc.NewClient(c.addr, grpc.WithTransportCredentials(insecure.NewCredentials())) + if err != nil { + return nil, fmt.Errorf("connect sing-box stats API: %w", err) + } + c.conn = conn + } + var resp sbRawBytes + if err := c.conn.Invoke(ctx, singboxStatsServiceQuery, sbRawBytes(encodeSingboxQuery()), &resp, grpc.ForceCodec(sbRawCodec{})); err != nil { + c.closeConn() + return nil, err + } + return parseSingboxStatsResponse(resp) +} + +// GetTrafficSnapshot reads the core counters and computes deltas +// against the committed baseline WITHOUT advancing it, under the +// epoch rules: an epoch that moves mid-query fails the round; a +// flipped epoch counts the new instance's counters in full; the +// agent's first-ever snapshot is baseline-only. +func (c *singboxStatsClient) GetTrafficSnapshot() (*xray.TrafficSnapshot, error) { + epochBefore := c.epochFunc() + if epochBefore == "" { + return nil, errSingboxEpochChanged + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + stats, err := c.queryStats(ctx) + cancel() + if err != nil { + logger.Debug("Failed to query sing-box stats:", err) + return nil, err + } + if epochAfter := c.epochFunc(); epochAfter != epochBefore { + return nil, errSingboxEpochChanged + } + c.pendingEpoch = epochBefore + + snap := &xray.TrafficSnapshot{Values: make(map[string]int64, len(stats))} + sameEpoch := c.committedReady && c.committedEpoch == epochBefore + tagTrafficMap := make(map[string]*xray.Traffic) + emailTrafficMap := make(map[string]*xray.ClientTraffic) + + for _, stat := range stats { + snap.Values[stat.Name] = stat.Value + if !c.committedReady { + continue + } + var value int64 + if sameEpoch { + lastValue, ok := c.committed[stat.Name] + if !ok || stat.Value < lastValue { + lastValue = 0 + } + value = stat.Value - lastValue + } else { + // New core instance: its counters started at zero, so + // everything they show is traffic since the restart. + value = stat.Value + } + if matches := singboxTrafficRegex.FindStringSubmatch(stat.Name); len(matches) == 4 { + processSingboxTraffic(matches, value, tagTrafficMap) + } else if matches := singboxClientTrafficRegex.FindStringSubmatch(stat.Name); len(matches) == 3 { + processSingboxClientTraffic(matches, value, emailTrafficMap) + } + } + + snap.Traffics = mapToSliceXrayTraffic(tagTrafficMap) + snap.ClientTraffics = mapToSliceXrayClientTraffic(emailTrafficMap) + return snap, nil +} + +// CommitTrafficSnapshot marks a snapshot as acknowledged by the +// master: its Values and epoch become the committed baseline. +func (c *singboxStatsClient) CommitTrafficSnapshot(s *xray.TrafficSnapshot) { + if s == nil { + return + } + c.committed = s.BaselineValues() + c.committedReady = true + c.committedEpoch = c.pendingEpoch +} + +// RestoreTrafficBaseline installs a committed baseline recovered from +// disk together with the epoch it was taken under. A baseline from a +// different core instance is dropped and the first snapshot rebases +// baseline-only: unlike a live epoch flip the agent observes itself, +// an agent restart leaves a blind gap — the running core's counters +// may include traffic the previous agent incarnation already +// reported, so counting them in full could double-report. +func (c *singboxStatsClient) RestoreTrafficBaseline(values map[string]int64, epoch string) { + if len(values) == 0 || epoch == "" || epoch != c.epochFunc() { + return + } + c.committed = make(map[string]int64, len(values)) + for k, v := range values { + c.committed[k] = v + } + c.committedReady = true + c.committedEpoch = epoch +} + +// GetOnlineUsers is unused for sing-box (the traffic cycle merges the +// log tracker's view with the snapshot's active users itself); it +// exists so the client satisfies the same shape as the xray one. +func (c *singboxStatsClient) GetOnlineUsers() ([]xray.OnlineUser, error) { + return nil, nil +} + +func processSingboxTraffic(matches []string, value int64, trafficMap map[string]*xray.Traffic) { + isInbound := matches[1] == "inbound" + tag := matches[2] + isDown := matches[3] == "downlink" + if tag == "api" { + return + } + traffic, ok := trafficMap[tag] + if !ok { + traffic = &xray.Traffic{IsInbound: isInbound, IsOutbound: !isInbound, Tag: tag} + trafficMap[tag] = traffic + } + if isDown { + traffic.Down = value + } else { + traffic.Up = value + } +} + +func processSingboxClientTraffic(matches []string, value int64, clientTrafficMap map[string]*xray.ClientTraffic) { + email := matches[1] + isDown := matches[2] == "downlink" + traffic, ok := clientTrafficMap[email] + if !ok { + traffic = &xray.ClientTraffic{Email: email} + clientTrafficMap[email] = traffic + } + if isDown { + traffic.Down = value + } else { + traffic.Up = value + } +} + +func mapToSliceXrayTraffic(m map[string]*xray.Traffic) []*xray.Traffic { + result := make([]*xray.Traffic, 0, len(m)) + for _, v := range m { + result = append(result, v) + } + return result +} + +func mapToSliceXrayClientTraffic(m map[string]*xray.ClientTraffic) []*xray.ClientTraffic { + result := make([]*xray.ClientTraffic, 0, len(m)) + for _, v := range m { + result = append(result, v) + } + return result +} diff --git a/internal/agent/singbox_stats_test.go b/internal/agent/singbox_stats_test.go new file mode 100644 index 0000000..c4438fb --- /dev/null +++ b/internal/agent/singbox_stats_test.go @@ -0,0 +1,244 @@ +package agent + +import ( + "context" + "errors" + "testing" + + "github.com/mhsanaei/3x-ui/v3/internal/xray" + + "google.golang.org/protobuf/encoding/protowire" +) + +// buildStatsResponse hand-encodes a QueryStatsResponse carrying the +// given counters, the same wire shape the real core returns. +func buildStatsResponse(counters map[string]int64) []byte { + var out []byte + for name, value := range counters { + var stat []byte + stat = protowire.AppendTag(stat, 1, protowire.BytesType) + stat = protowire.AppendString(stat, name) + stat = protowire.AppendTag(stat, 2, protowire.VarintType) + stat = protowire.AppendVarint(stat, uint64(value)) + out = protowire.AppendTag(out, 1, protowire.BytesType) + out = protowire.AppendBytes(out, stat) + } + return out +} + +func TestParseSingboxStatsResponse(t *testing.T) { + raw := buildStatsResponse(map[string]int64{ + "inbound>>>ss22>>>traffic>>>uplink": 78, + "user>>>bob@example.com>>>traffic>>>downlink": 573, + }) + entries, err := parseSingboxStatsResponse(raw) + if err != nil { + t.Fatalf("parse: %v", err) + } + if len(entries) != 2 { + t.Fatalf("entries = %d, want 2", len(entries)) + } + got := map[string]int64{} + for _, e := range entries { + got[e.Name] = e.Value + } + if got["inbound>>>ss22>>>traffic>>>uplink"] != 78 || got["user>>>bob@example.com>>>traffic>>>downlink"] != 573 { + t.Fatalf("counters = %v", got) + } + if _, err := parseSingboxStatsResponse([]byte{0xff, 0xff}); err == nil { + t.Fatal("garbage input must error") + } +} + +func TestEncodeSingboxQuery(t *testing.T) { + b := encodeSingboxQuery() + var patterns []string + for len(b) > 0 { + num, typ, n := protowire.ConsumeTag(b) + if n < 0 || num != 3 || typ != protowire.BytesType { + t.Fatalf("unexpected field num=%d typ=%d", num, typ) + } + b = b[n:] + v, n := protowire.ConsumeString(b) + if n < 0 { + t.Fatal("bad string field") + } + patterns = append(patterns, v) + b = b[n:] + } + if len(patterns) != 3 { + t.Fatalf("patterns = %v", patterns) + } +} + +// scriptedStats drives the tracker with scripted counters and a +// scripted epoch, following the fakeStatsClient pattern of the xray +// path: the delta math contract is what's under test here; the wire +// itself is covered by the parse test and the real-binary E2E. +type scriptedStats struct { + counters map[string]int64 + epoch string + err error +} + +func (s *scriptedStats) client() *singboxStatsClient { + c := newSingboxStatsClient(func() string { return s.epoch }) + c.queryFn = func(context.Context) ([]singboxStatEntry, error) { + if s.err != nil { + return nil, s.err + } + out := make([]singboxStatEntry, 0, len(s.counters)) + for name, value := range s.counters { + out = append(out, singboxStatEntry{Name: name, Value: value}) + } + return out, nil + } + return c +} + +const ( + sbUserUp = "user>>>bob@example.com>>>traffic>>>uplink" + sbUserDown = "user>>>bob@example.com>>>traffic>>>downlink" + sbInUp = "inbound>>>ss22>>>traffic>>>uplink" +) + +func snapUserDown(snap *xray.TrafficSnapshot, email string) int64 { + for _, ct := range snap.ClientTraffics { + if ct.Email == email { + return ct.Down + } + } + return 0 +} + +func TestSingboxSnapshotBaselineThenDeltas(t *testing.T) { + s := &scriptedStats{epoch: "sing-box:1:100", counters: map[string]int64{sbUserDown: 500, sbInUp: 40}} + c := s.client() + + snap, err := c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot 1: %v", err) + } + if len(snap.ClientTraffics) != 0 || len(snap.Traffics) != 0 { + t.Fatal("first snapshot of an epoch must be baseline-only") + } + c.CommitTrafficSnapshot(snap) + + s.counters[sbUserDown] = 900 + s.counters[sbInUp] = 55 + snap, err = c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot 2: %v", err) + } + if got := snapUserDown(snap, "bob@example.com"); got != 400 { + t.Fatalf("user delta = %d, want 400", got) + } + c.CommitTrafficSnapshot(snap) + + // No movement: zero deltas are still shaped but empty of content. + snap, err = c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot 3: %v", err) + } + if got := snapUserDown(snap, "bob@example.com"); got != 0 { + t.Fatalf("idle delta = %d, want 0", got) + } +} + +func TestSingboxSnapshotEpochChangeCountsInFull(t *testing.T) { + s := &scriptedStats{epoch: "sing-box:1:100", counters: map[string]int64{sbUserDown: 500}} + c := s.client() + snap, err := c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot 1: %v", err) + } + c.CommitTrafficSnapshot(snap) + + // Core restarted: counters reset and grow again. The new + // instance's counters started at zero, so they count in full + // (NexusGate's ApplyDelta rule) — never as a diff against the + // dead instance's baseline, and never dropped. + s.epoch = "sing-box:2:200" + s.counters[sbUserDown] = 120 + snap, err = c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot after restart: %v", err) + } + if got := snapUserDown(snap, "bob@example.com"); got != 120 { + t.Fatalf("post-restart delta = %d, want 120 (counted in full)", got) + } + c.CommitTrafficSnapshot(snap) + + s.counters[sbUserDown] = 150 + snap, err = c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot 3: %v", err) + } + if got := snapUserDown(snap, "bob@example.com"); got != 30 { + t.Fatalf("post-rebase delta = %d, want 30", got) + } +} + +func TestSingboxSnapshotEpochUnstableFails(t *testing.T) { + epochs := []string{"sing-box:1:100", "sing-box:2:200"} + calls := 0 + c := newSingboxStatsClient(func() string { + e := epochs[min(calls, len(epochs)-1)] + calls++ + return e + }) + c.queryFn = func(context.Context) ([]singboxStatEntry, error) { + return []singboxStatEntry{{Name: sbUserDown, Value: 10}}, nil + } + if _, err := c.GetTrafficSnapshot(); !errors.Is(err, errSingboxEpochChanged) { + t.Fatalf("err = %v, want errSingboxEpochChanged", err) + } +} + +func TestSingboxSnapshotCounterResetCountsFromZero(t *testing.T) { + s := &scriptedStats{epoch: "sing-box:1:100", counters: map[string]int64{sbUserDown: 500, sbUserUp: 10}} + c := s.client() + snap, err := c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot 1: %v", err) + } + c.CommitTrafficSnapshot(snap) + + // Same epoch but a counter moved backwards (core reset its stats + // without a process restart): counts from zero, like xray. + s.counters[sbUserDown] = 60 + s.counters[sbUserUp] = 4 // unseen-before direction appears too + snap, err = c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot 2: %v", err) + } + if got := snapUserDown(snap, "bob@example.com"); got != 60 { + t.Fatalf("reset counter delta = %d, want 60", got) + } +} + +func TestSingboxRestoreBaselineEpochGate(t *testing.T) { + s := &scriptedStats{epoch: "sing-box:7:700", counters: map[string]int64{sbUserDown: 800}} + c := s.client() + + // Baseline from a dead instance: ignored, snapshot rebases. + c.RestoreTrafficBaseline(map[string]int64{sbUserDown: 300}, "sing-box:6:600") + snap, err := c.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot: %v", err) + } + if len(snap.ClientTraffics) != 0 { + t.Fatal("stale-epoch baseline must not produce deltas") + } + + // Baseline from the live instance: deltas continue from it. + c2 := s.client() + c2.RestoreTrafficBaseline(map[string]int64{sbUserDown: 300}, "sing-box:7:700") + snap, err = c2.GetTrafficSnapshot() + if err != nil { + t.Fatalf("snapshot 2: %v", err) + } + if got := snapUserDown(snap, "bob@example.com"); got != 500 { + t.Fatalf("restored delta = %d, want 500", got) + } +} diff --git a/internal/agent/telemetry.go b/internal/agent/telemetry.go index fe16539..26c5bf3 100644 --- a/internal/agent/telemetry.go +++ b/internal/agent/telemetry.go @@ -22,14 +22,18 @@ import ( // every heartbeat (P13). Pointer fields are omitted when their collector // failed, so the master keeps the last stored value instead of a zero. type telemetryPayload struct { - AgentVersion string `json:"agentVersion"` - XrayVersion string `json:"xrayVersion,omitempty"` - UptimeSeconds *uint64 `json:"uptimeSeconds,omitempty"` - LatencyMs *int64 `json:"latencyMs,omitempty"` - CpuPercent *float64 `json:"cpuPercent,omitempty"` - MemPercent *float64 `json:"memPercent,omitempty"` - IPv4 string `json:"ipv4,omitempty"` - IPv6 string `json:"ipv6,omitempty"` + AgentVersion string `json:"agentVersion"` + XrayVersion string `json:"xrayVersion,omitempty"` + // SingboxVersion rides along since ITEM6 B3. Masters before the + // B4/B5 field rename ignore the unknown key; the xray-named + // fields stay authoritative until then. + SingboxVersion string `json:"singboxVersion,omitempty"` + UptimeSeconds *uint64 `json:"uptimeSeconds,omitempty"` + LatencyMs *int64 `json:"latencyMs,omitempty"` + CpuPercent *float64 `json:"cpuPercent,omitempty"` + MemPercent *float64 `json:"memPercent,omitempty"` + IPv4 string `json:"ipv4,omitempty"` + IPv6 string `json:"ipv6,omitempty"` } // collectTelemetry samples every source independently; a failing source @@ -41,6 +45,9 @@ func (a *Agent) collectTelemetry() *telemetryPayload { if v, ok := a.cachedXrayVersion(); ok { t.XrayVersion = v } + if v, ok := a.cachedSingboxVersion(); ok { + t.SingboxVersion = v + } if up, err := host.Uptime(); err == nil { t.UptimeSeconds = &up }