package singbox import ( "context" "encoding/json" "fmt" "log/slog" "sync" "time" box "github.com/sagernet/sing-box" "github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/include" singLog "github.com/sagernet/sing-box/log" "github.com/sagernet/sing-box/option" "github.com/sagernet/sing/common/auth" singJSON "github.com/sagernet/sing/common/json" "github.com/sagernet/sing/service" "golang.org/x/time/rate" "github.com/cedar2025/xboard-node/internal/config" "github.com/cedar2025/xboard-node/internal/kernel" "github.com/cedar2025/xboard-node/internal/panel" ) // drainTimeout is how long stop() waits for in-flight connections to finish // naturally after closing all listen sockets, before hard-killing them. const drainTimeout = 5 * time.Second // SingBox implements kernel.Kernel by embedding sing-box as a Go library. // // Connection tracking is done entirely in-process via ConnTracker, which is // registered as an adapter.ConnectionTracker with sing-box's router. This // avoids the HTTP/JSON overhead of the Clash API (saves 5-15 MB at 500+ // concurrent connections). type SingBox struct { cfg config.KernelConfig mu sync.Mutex box *box.Box ctx context.Context cancel context.CancelFunc // connTracker is our lightweight in-process byte/IP tracker. // Created fresh on every ApplyConfig (full restart). // Survives ReloadUsers (hot-swap) since live connections persist. connTracker *ConnTracker // speedLimitFunc resolves a user UUID to a *rate.Limiter. // Set once by SetSpeedLimitFunc and forwarded to every new ConnTracker. speedLimitFunc func(string) *rate.Limiter // trackerRegistered prevents duplicate AppendTracker calls on the same // Router instance during ReloadUsers. Reset to false on full restart. trackerRegistered bool } func New(cfg config.KernelConfig) *SingBox { return &SingBox{cfg: cfg} } func (s *SingBox) Name() string { return "sing-box" } func (s *SingBox) ApplyConfig(nodeConfig *panel.NodeConfig, users []panel.User, certFile, keyFile string) error { s.mu.Lock() defer s.mu.Unlock() cfgMap := buildConfig(s.cfg, nodeConfig, users, certFile, keyFile) data, err := json.Marshal(cfgMap) if err != nil { return fmt.Errorf("marshal config: %w", err) } slog.Debug("sing-box config generated", "len", len(data)) ctx, cancel := context.WithCancel(context.Background()) ctx = include.Context(ctx) opts, err := singJSON.UnmarshalExtendedContext[option.Options](ctx, data) if err != nil { cancel() return fmt.Errorf("parse sing-box options: %w", err) } s.stop() instance, err := box.New(box.Options{ Context: ctx, Options: opts, }) if err != nil { cancel() return fmt.Errorf("create sing-box instance: %w", err) } if err := instance.Start(); err != nil { instance.Close() cancel() return fmt.Errorf("start sing-box: %w", err) } s.box = instance s.ctx = ctx s.cancel = cancel // Fresh tracker on full restart (all connections were terminated by s.stop). s.connTracker = NewConnTracker(0) s.connTracker.SetUserMap(buildUserMap(users)) if s.speedLimitFunc != nil { s.connTracker.SetSpeedLimitFunc(s.speedLimitFunc) } s.trackerRegistered = false s.registerTracker(ctx) slog.Info("sing-box started (in-process)", "users", len(users)) return nil } // Reload hot-swaps the inbound users and routing rules without restarting the box. // Routes, outbounds, and the connTracker stay alive so in-flight connections // continue to be tracked correctly. func (s *SingBox) Reload(nodeConfig *panel.NodeConfig, users []panel.User, certFile, keyFile string) error { s.mu.Lock() defer s.mu.Unlock() if s.box == nil { return fmt.Errorf("not running") } cfgMap := buildConfig(s.cfg, nodeConfig, users, certFile, keyFile) data, err := json.Marshal(cfgMap) if err != nil { return fmt.Errorf("marshal config: %w", err) } opts, err := singJSON.UnmarshalExtendedContext[option.Options](s.ctx, data) if err != nil { return fmt.Errorf("parse options: %w", err) } im := service.FromContext[adapter.InboundManager](s.ctx) if im == nil { return fmt.Errorf("inbound manager not available") } router := service.FromContext[adapter.Router](s.ctx) if router == nil { return fmt.Errorf("router not available") } // Update routing rules if err := router.UpdateRules(opts.Route.Rules, opts.Route.RuleSet); err != nil { slog.Warn("routing reload failed", "error", err) } else { slog.Info("sing-box routing reloaded") } nopFactory := singLog.NewNOPFactory() for _, inb := range opts.Inbounds { tag := inb.Tag if existing, ok := im.Get(tag); ok && existing.Type() == inb.Type { var err error switch v := existing.(type) { case adapter.UpdatableInbound[option.VMessUser]: if opts, ok := inb.Options.(*option.VMessInboundOptions); ok { err = v.UpdateUsers(opts.Users) } case adapter.UpdatableInbound[option.VLESSUser]: if opts, ok := inb.Options.(*option.VLESSInboundOptions); ok { err = v.UpdateUsers(opts.Users) } case adapter.UpdatableInbound[option.TrojanUser]: if opts, ok := inb.Options.(*option.TrojanInboundOptions); ok { err = v.UpdateUsers(opts.Users) } case adapter.UpdatableInbound[option.Hysteria2User]: if opts, ok := inb.Options.(*option.Hysteria2InboundOptions); ok { err = v.UpdateUsers(opts.Users) } case adapter.UpdatableShadowsocksInbound: if opts, ok := inb.Options.(*option.ShadowsocksInboundOptions); ok { err = v.UpdateUsersByOptions(opts.Users) } case adapter.UpdatableInbound[option.TUICUser]: if opts, ok := inb.Options.(*option.TUICInboundOptions); ok { err = v.UpdateUsers(opts.Users) } case adapter.UpdatableInbound[option.AnyTLSUser]: if opts, ok := inb.Options.(*option.AnyTLSInboundOptions); ok { err = v.UpdateUsers(opts.Users) } case adapter.UpdatableInbound[auth.User]: switch opts := inb.Options.(type) { case *option.NaiveInboundOptions: err = v.UpdateUsers(opts.Users) case *option.SocksInboundOptions: err = v.UpdateUsers(opts.Users) case *option.HTTPMixedInboundOptions: err = v.UpdateUsers(opts.Users) } } if err == nil { continue } slog.Warn("incremental update failed, falling back to recreate", "tag", tag, "error", err) } // Remove the existing inbound first so im.Create() can bind the same port. // im.Create() cannot atomically swap a TCP listener — it tries to start the // new socket before the old one is closed, causing "address already in use". // The brief listen gap (< 1 ms) is far less disruptive than a full restart. _ = im.Remove(tag) // ignore error when tag doesn't exist yet logger := nopFactory.NewLogger(fmt.Sprintf("inbound/%s[%s]", inb.Type, tag)) if err := im.Create(s.ctx, router, logger, tag, inb.Type, inb.Options); err != nil { return fmt.Errorf("recreate inbound %s: %w", tag, err) } } // Trackers remain registered on the Router (which survives ReloadUsers). // Only update the user map — do NOT re-register or traffic is double-counted. if s.connTracker != nil { s.connTracker.SetUserMap(buildUserMap(users)) } slog.Info("sing-box reloaded (users and routes hot-swapped)", "users", len(users)) return nil } // registerTracker wires the ConnTracker to the current Router exactly once. // The ConnTracker handles both byte counting and optional rate limiting // in a single wrapper, so no additional trackers are needed. func (s *SingBox) registerTracker(ctx context.Context) { if s.trackerRegistered { return } router := service.FromContext[adapter.Router](ctx) if router == nil { return } if s.connTracker != nil { router.AppendTracker(s.connTracker) } s.trackerRegistered = true } func (s *SingBox) Stop() { s.mu.Lock() defer s.mu.Unlock() s.stop() } func (s *SingBox) stop() { if s.box == nil { return } // Step 1: close all listen sockets so no new connections are accepted. // im.Close() sets m.started=false so the subsequent box.Close() call is a no-op // for inbounds (avoids double-close), but box.Close() still tears down routers // and outbounds. if im := service.FromContext[adapter.InboundManager](s.ctx); im != nil { _ = im.Close() } // Step 2: wait for in-flight connections to drain naturally (best-effort). if s.connTracker != nil { deadline := time.Now().Add(drainTimeout) for time.Now().Before(deadline) { if s.connTracker.ActiveCount() == 0 { break } time.Sleep(100 * time.Millisecond) } } // Step 3: hard-close everything that is still open. s.box.Close() s.box = nil if s.cancel != nil { s.cancel() s.cancel = nil } s.ctx = nil } func (s *SingBox) IsRunning() bool { s.mu.Lock() defer s.mu.Unlock() return s.box != nil } // connTrackerLocked returns the connTracker under a read-optimised lock. func (s *SingBox) connTrackerSafe() *ConnTracker { s.mu.Lock() ct := s.connTracker s.mu.Unlock() return ct } // SetSpeedLimitFunc configures per-user bandwidth throttling. // The function is forwarded to the ConnTracker, which embeds the rate limiter // in each tracked connection wrapper. This merges byte counting and rate limiting // into a single CountFunc callback, allowing splice/zero-copy paths. func (s *SingBox) SetSpeedLimitFunc(fn func(uuid string) *rate.Limiter) { s.mu.Lock() defer s.mu.Unlock() s.speedLimitFunc = fn if s.connTracker != nil { s.connTracker.SetSpeedLimitFunc(fn) } } // GetConnections returns a snapshot of all active and recently-closed // connections directly from the in-process ConnTracker. func (s *SingBox) GetConnections(_ context.Context) ([]kernel.Connection, error) { ct := s.connTrackerSafe() if ct == nil { return nil, nil } conns := ct.Snapshot() slog.Debug("GetConnections snapshot", "count", len(conns)) return conns, nil } // CloseConnection force-closes a specific connection by its ID. func (s *SingBox) CloseConnection(_ context.Context, connID string) error { ct := s.connTrackerSafe() if ct == nil { return fmt.Errorf("not running") } if !ct.CloseByID(connID) { slog.Debug("CloseConnection: connection not found (already closed?)", "id", connID) } return nil } // buildUserMap creates a UUID→userID mapping used by ConnTracker to attribute // connections to the correct user. All sing-box protocols use the user's UUID // as the inbound name/username, so this covers every protocol. func buildUserMap(users []panel.User) map[string]int { m := make(map[string]int, len(users)) for _, u := range users { m[u.UUID] = u.ID } return m }