pangolin/server/internal/agentd/singbox.go

package agentd

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/fs"
	"os"
	"path/filepath"
	"sort"
	"sync"
	"time"

	agentv1 "github.com/wangjia/pangolin/server/internal/pb/agentv1"
)

// Restarter applies a freshly-rendered sing-box config. sing-box 支持 SIGHUP
// 热重载(进程内 validate→close 旧→start 新,坏配置保留旧实例不断网),故配置
// 变更走 Reload(不整进程重启、不全断流),冷启动走 Restart。Tests inject a fake.
type Restarter interface {
	Restart(ctx context.Context) error // 冷启动:systemctl restart(~1-2s blip)
	Reload(ctx context.Context) error  // 热重载:SIGHUP 给 sing-box MainPID
}

// Cred is the agent's in-memory view of one data-plane credential. It is the
// authoritative local state together with the rendered sing-box config.
type Cred struct {
	DpUUID    string
	Protocol  agentv1.Protocol
	Flow      string
	ExpiresAt int64 // unix seconds; 0 = no expiry
}

// persistedCred is the on-disk shape. Per the no-state invariant we persist ONLY
// dp_uuid + expires_at — protocol/flow are reconstructed (defaults) on load and
// then corrected by the next Register reconciliation.
type persistedCred struct {
	DpUUID    string `json:"dp_uuid"`
	ExpiresAt int64  `json:"expires_at"`
}

type persistedState struct {
	Creds []persistedCred `json:"creds"`
}

// SingBox owns the credential table, state.json persistence, sing-box config
// rendering and the debounced restart loop.
type SingBox struct {
	cfg       Config
	restarter Restarter
	clock     func() time.Time

	// OnExpire, if set, is invoked (outside the lock) when the TTL sweep removes
	// a credential, so the agent can surface the removal upstream.
	OnExpire func(dpUUID string)

	mu            sync.Mutex
	creds         map[string]*Cred
	reality       *agentv1.RealityInbound
	hy2           *agentv1.Hy2Inbound
	configVersion int64

	// lastGoodACL 是最近一次成功加载的私有目的地 ACL。acl.json 读坏时回退到它,
	// 而不是像 WARP 那样降级为「不启用」—— 对访问控制,降级即敞开。
	lastGoodACL *ACLConfig

	// started 标记 sing-box 是否已被本 agent 冷启动过:首次走 Restart(冷启动),
	// 之后配置变更走 Reload(SIGHUP 热重载)。agent 进程重启后复位为 false,
	// 下次渲染做一次冷启动以确保与渲染配置一致。
	started bool

	// debounce machinery
	dirty   chan struct{}
	done    chan struct{}
	closed  bool
	startWG sync.WaitGroup
}

// NewSingBox builds a manager. restarter may be nil (renders only, no restart).
func NewSingBox(cfg Config, restarter Restarter) *SingBox {
	if restarter == nil {
		restarter = noopRestarter{}
	}
	return &SingBox{
		cfg:       cfg.withDefaults(),
		restarter: restarter,
		clock:     time.Now,
		creds:     make(map[string]*Cred),
		dirty:     make(chan struct{}, 1),
		done:      make(chan struct{}),
	}
}

// LoadState restores the credential table from state.json after a restart. Only
// dp_uuid + expires_at survive; protocol defaults to BOTH and flow to the
// configured default until the next Register overwrites them. Expired entries are
// dropped on load.
func (s *SingBox) LoadState() error {
	data, err := os.ReadFile(s.cfg.StatePath())
	if errors.Is(err, fs.ErrNotExist) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("agentd: read state: %w", err)
	}
	var ps persistedState
	if err := json.Unmarshal(data, &ps); err != nil {
		return fmt.Errorf("agentd: parse state: %w", err)
	}
	now := s.clock().Unix()
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, pc := range ps.Creds {
		if pc.ExpiresAt != 0 && pc.ExpiresAt <= now {
			continue
		}
		s.creds[pc.DpUUID] = &Cred{
			DpUUID:    pc.DpUUID,
			Protocol:  agentv1.ProtocolBoth,
			Flow:      DefaultFlow,
			ExpiresAt: pc.ExpiresAt,
		}
	}
	return nil
}

// saveStateLocked writes state.json atomically. Caller holds s.mu.
func (s *SingBox) saveStateLocked() error {
	ps := persistedState{Creds: make([]persistedCred, 0, len(s.creds))}
	for _, c := range s.creds {
		ps.Creds = append(ps.Creds, persistedCred{DpUUID: c.DpUUID, ExpiresAt: c.ExpiresAt})
	}
	sort.Slice(ps.Creds, func(i, j int) bool { return ps.Creds[i].DpUUID < ps.Creds[j].DpUUID })
	data, err := json.MarshalIndent(ps, "", "  ")
	if err != nil {
		return err
	}
	return atomicWrite(s.cfg.StatePath(), data, 0o600)
}

// ─── mutations (all idempotent, all schedule a render) ───────────────────────────

// Upsert adds or replaces a credential.
func (s *SingBox) Upsert(c *Cred) {
	s.mu.Lock()
	cp := *c
	if cp.Flow == "" {
		cp.Flow = DefaultFlow
	}
	s.creds[cp.DpUUID] = &cp
	_ = s.saveStateLocked()
	s.mu.Unlock()
	s.markDirty()
}

// DpUUIDs returns a snapshot of the currently provisioned dp_uuids.
// Used by the usage source to attribute node-total traffic to active users.
func (s *SingBox) DpUUIDs() []string {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]string, 0, len(s.creds))
	for u := range s.creds {
		out = append(out, u)
	}
	return out
}

// Revoke removes a credential. No-op if absent (idempotent).
func (s *SingBox) Revoke(dpUUID string) {
	s.mu.Lock()
	_, existed := s.creds[dpUUID]
	delete(s.creds, dpUUID)
	if existed {
		_ = s.saveStateLocked()
	}
	s.mu.Unlock()
	if existed {
		s.markDirty()
	}
}

// Rotate installs new alongside old, keeping old alive until graceUntil. The TTL
// sweep removes old when its grace expires, giving "宽限期内新旧并存".
func (s *SingBox) Rotate(oldDpUUID string, newCred *Cred, graceUntil int64) {
	s.mu.Lock()
	if old, ok := s.creds[oldDpUUID]; ok && oldDpUUID != newCred.DpUUID {
		// Clamp old credential's expiry to the grace deadline.
		if graceUntil > 0 && (old.ExpiresAt == 0 || graceUntil < old.ExpiresAt) {
			old.ExpiresAt = graceUntil
		}
	}
	nc := *newCred
	if nc.Flow == "" {
		nc.Flow = DefaultFlow
	}
	s.creds[nc.DpUUID] = &nc
	_ = s.saveStateLocked()
	s.mu.Unlock()
	s.markDirty()
}

// ApplyConfig replaces the node-wide inbound parameters and (if the snapshot
// carries credentials) the full credential table. Used by APPLY_CONFIG commands
// and full resync.
func (s *SingBox) ApplyConfig(snap *agentv1.ConfigSnapshot, replaceCreds bool) {
	s.mu.Lock()
	if snap.Reality != nil {
		s.reality = snap.Reality
	}
	if snap.Hy2 != nil {
		s.hy2 = snap.Hy2
	}
	if snap.ConfigVersion != 0 {
		s.configVersion = snap.ConfigVersion
	}
	if replaceCreds {
		s.creds = make(map[string]*Cred, len(snap.Credentials))
		for _, c := range snap.Credentials {
			flow := c.Flow
			if flow == "" {
				flow = DefaultFlow
			}
			s.creds[c.DpUUID] = &Cred{
				DpUUID:    c.DpUUID,
				Protocol:  c.Protocol,
				Flow:      flow,
				ExpiresAt: c.ExpiresAtUnix,
			}
		}
	}
	_ = s.saveStateLocked()
	s.mu.Unlock()
	s.markDirty()
}

// SetConfigVersion records the authoritative config version (e.g. from Register).
func (s *SingBox) SetConfigVersion(v int64) {
	s.mu.Lock()
	s.configVersion = v
	s.mu.Unlock()
}

// ─── read accessors ──────────────────────────────────────────────────────────

func (s *SingBox) ConfigVersion() int64 {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.configVersion
}

func (s *SingBox) OnlinePeers() int {
	s.mu.Lock()
	defer s.mu.Unlock()
	return len(s.creds)
}

// Has reports whether a dp_uuid is currently provisioned.
func (s *SingBox) Has(dpUUID string) bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	_, ok := s.creds[dpUUID]
	return ok
}

// Snapshot returns a copy of the current credential set (sorted by dp_uuid).
func (s *SingBox) Snapshot() []Cred {
	s.mu.Lock()
	defer s.mu.Unlock()
	out := make([]Cred, 0, len(s.creds))
	for _, c := range s.creds {
		out = append(out, *c)
	}
	sort.Slice(out, func(i, j int) bool { return out[i].DpUUID < out[j].DpUUID })
	return out
}

// ─── TTL sweep ───────────────────────────────────────────────────────────────

// sweepExpired removes every credential whose expires_at has passed. Returns the
// removed dp_uuids and whether anything changed.
func (s *SingBox) sweepExpired() []string {
	now := s.clock().Unix()
	s.mu.Lock()
	var removed []string
	for id, c := range s.creds {
		if c.ExpiresAt != 0 && c.ExpiresAt <= now {
			delete(s.creds, id)
			removed = append(removed, id)
		}
	}
	if len(removed) > 0 {
		_ = s.saveStateLocked()
	}
	s.mu.Unlock()
	if len(removed) > 0 {
		s.markDirty()
		if s.OnExpire != nil {
			for _, id := range removed {
				s.OnExpire(id)
			}
		}
	}
	sort.Strings(removed)
	return removed
}

// ─── rendering ───────────────────────────────────────────────────────────────

// RenderConfig builds the sing-box server config JSON from the current state.
// Exported (and pure w.r.t. the snapshot) so tests can assert its contents.
func (s *SingBox) RenderConfig() ([]byte, error) {
	s.mu.Lock()
	creds := make([]Cred, 0, len(s.creds))
	for _, c := range s.creds {
		creds = append(creds, *c)
	}
	reality := s.reality
	hy2 := s.hy2
	s.mu.Unlock()
	sort.Slice(creds, func(i, j int) bool { return creds[i].DpUUID < creds[j].DpUUID })

	// WARP 分流配置每次渲染读一次:编辑 warp.json 后任一渲染(或 agent 重启)即生效(#29)。
	// 读失败(坏 JSON)仅记日志、按未启用处理,绝不因坏配置产出无法启动的 sing-box 配置。
	warp, err := LoadWarpConfig(s.cfg.WarpConfigPath)
	if err != nil {
		logf("[warp] load %s failed, WARP routing disabled: %v", s.cfg.WarpConfigPath, err)
		warp = nil
	}
	return renderSingboxConfig(creds, reality, hy2, s.cfg.DeriveKey, warp, s.loadACL())
}

// loadACL 读取私有目的地 ACL,并维护 last-good 兜底。
//
// 语义刻意与 WARP 相反:WARP 读失败静默降级为「不分流」是安全的,ACL 读失败若也
// 降级为「不启用」,等于把私有服务对全体 pangolin 用户敞开,而且是静默的。故:
//   - 成功  → 更新内存 last-good 并落盘,供本进程后续与下次冷启动使用
//   - 失败/文件消失 → 回退内存 last-good,再回退磁盘 last-good,规则不消失
//   - 两级 last-good 都没有 → 只能不产出规则(白名单与目的地清单同在一个文件,
//     文件全丢时连「该拒绝哪些目的地」都无从得知),此时必须大声告警
func (s *SingBox) loadACL() *ACLConfig {
	acl, err := LoadACLConfig(s.cfg.ACLConfigPath)
	switch {
	case err == nil && acl != nil:
		if acl.Enabled && len(acl.cleanTargets()) == 0 {
			// enabled 却零 target = 几乎必然是字段名手抖(如 targets 打成 targetz)。
			// encoding/json 对未知字段静默无视,这份配置"看似合法"但毫无内容——当作
			// 加载失败,走下面的 last-good 兜底,绝不静默敞开、更不能拿它覆盖兜底快照。
			logf("[acl] ERROR %s parsed but enabled=true yields zero targets — treating as broken, falling back", s.cfg.ACLConfigPath)
			break // 落到下方 last-good 兜底(Go 的 switch 内 break 只退出 switch)
		}
		if acl.active() {
			// 只有 active 的配置才配当 last-good:否则一次手抖既关闸又冲掉兜底。
			s.mu.Lock()
			s.lastGoodACL = acl
			s.mu.Unlock()
			if perr := persistACL(s.cfg.ACLLastGoodPath(), acl); perr != nil {
				logf("[acl] persist last-good to %s failed: %v", s.cfg.ACLLastGoodPath(), perr)
			}
		} else {
			// active() 为 false 但走到了这里,只可能是显式 enabled:false —— 合法关闭。
			// 直接返回这份(空规则的)配置,不 persist,保留此前的恢复能力。
			logf("[acl] gate explicitly disabled (enabled=false)")
		}
		return acl
	case err != nil:
		logf("[acl] ERROR load %s failed: %v", s.cfg.ACLConfigPath, err)
	}

	s.mu.Lock()
	lg := s.lastGoodACL
	s.mu.Unlock()
	if lg != nil {
		logf("[acl] WARN falling back to in-memory last-good ACL")
		return lg
	}

	disk, derr := LoadACLConfig(s.cfg.ACLLastGoodPath())
	if derr != nil {
		logf("[acl] ERROR reading last-good %s: %v", s.cfg.ACLLastGoodPath(), derr)
	}
	if derr == nil && disk != nil {
		logf("[acl] WARN falling back to on-disk last-good %s", s.cfg.ACLLastGoodPath())
		s.mu.Lock()
		s.lastGoodACL = disk
		s.mu.Unlock()
		return disk
	}

	// 两级 last-good 都没有可用配置。区分两种终态:
	//   - 这台节点从未配置过 ACL(acl.json 与 last-good 均从未存在过)→ 安静返回,
	//     不刷屏告警。
	//   - 除此之外的任何情况(acl.json 存在但损坏/last-good 存在但损坏等)→ 私有
	//     服务的访问闸实质已消失,必须大声告警(§5 "两者都失败才不产出 ACL 规则,
	//     同时打 ERROR 并告警")。
	_, aclStatErr := os.Stat(s.cfg.ACLConfigPath)
	_, lgStatErr := os.Stat(s.cfg.ACLLastGoodPath())
	neverConfigured := os.IsNotExist(aclStatErr) && os.IsNotExist(lgStatErr)
	if !neverConfigured {
		logf("[acl] ALERT acl.json is broken and no last-good snapshot exists — "+
			"private destinations are UNPROTECTED (path=%s)", s.cfg.ACLConfigPath)
	}
	return nil
}

// writeAndRestart renders, writes the config file and restarts sing-box.
func (s *SingBox) writeAndRestart(ctx context.Context) error {
	if err := os.MkdirAll(filepath.Dir(s.cfg.SingboxConfigPath), 0o755); err != nil {
		return fmt.Errorf("agentd: mkdir singbox cfg: %w", err)
	}
	// Hy2 启用时:渲染前确保自签 TLS 证书就位,并把路径写回 hy2 配置(方案①)。
	if err := s.ensureHy2Cert(); err != nil {
		return err
	}
	data, err := s.RenderConfig()
	if err != nil {
		return err
	}
	if err := atomicWrite(s.cfg.SingboxConfigPath, data, 0o644); err != nil {
		return err
	}
	// 首次/进程未被本 agent 启动过 → 冷启动;已在跑 + 配置变更 → SIGHUP 热重载
	// (Reload 内部在取不到 PID/发信号失败时已自动回退 Restart)。
	s.mu.Lock()
	started := s.started
	s.mu.Unlock()
	if !started {
		if err := s.restarter.Restart(ctx); err != nil {
			return err
		}
		s.mu.Lock()
		s.started = true
		s.mu.Unlock()
		return nil
	}
	return s.restarter.Reload(ctx)
}

// ensureHy2Cert 为启用 hy2 的节点准备自签证书(幂等),并把 CertPath/KeyPath 写回当前
// hy2 配置,供 RenderConfig 渲染 hysteria2 入站的 certificate_path/key_path。
func (s *SingBox) ensureHy2Cert() error {
	s.mu.Lock()
	enabled := s.hy2 != nil
	var sni string
	if s.reality != nil {
		sni = s.reality.ServerName
	}
	s.mu.Unlock()
	if !enabled {
		return nil
	}
	dir := filepath.Dir(s.cfg.SingboxConfigPath)
	certPath := filepath.Join(dir, "hy2.crt")
	keyPath := filepath.Join(dir, "hy2.key")
	if err := ensureSelfSignedCert(certPath, keyPath, sni); err != nil {
		return err
	}
	s.mu.Lock()
	if s.hy2 != nil {
		s.hy2.CertPath = certPath
		s.hy2.KeyPath = keyPath
	}
	s.mu.Unlock()
	return nil
}

// markDirty signals the debounce loop that a render is pending (non-blocking).
func (s *SingBox) markDirty() {
	select {
	case s.dirty <- struct{}{}:
	default:
	}
}

// Refresh 请求一次重渲染。供 agent 收到 SIGHUP 时调用,使编辑节点本地配置文件
// (acl.json / warp.json)后无需重启进程即可生效 —— 重启 agent 会让 sing-box 走
// 冷启动(Restart),把全部在线用户踢下线。
func (s *SingBox) Refresh() { s.markDirty() }

// Run drives the debounce loop: it coalesces bursts of changes within
// DebounceWindow into a single write+restart. It blocks until ctx is cancelled.
func (s *SingBox) Run(ctx context.Context) {
	timer := time.NewTimer(time.Hour)
	timer.Stop()
	pending := false
	for {
		select {
		case <-ctx.Done():
			timer.Stop()
			return
		case <-s.dirty:
			if !pending {
				pending = true
				timer.Reset(s.cfg.DebounceWindow)
			}
		case <-timer.C:
			pending = false
			if err := s.writeAndRestart(ctx); err != nil {
				logf("singbox render/restart failed: %v", err)
			}
		}
	}
}

// noopRestarter renders without restarting (dev / config-check only).
type noopRestarter struct{}

func (noopRestarter) Restart(context.Context) error { return nil }
func (noopRestarter) Reload(context.Context) error  { return nil }