Muse
db6f666ff7
CI / go-test (push) Waiting to run
CI / postgres-durable-first (push) Waiting to run
CI / codegen (push) Waiting to run
CI / govulncheck (push) Waiting to run
CI / race (push) Waiting to run
CI / fuzz-smoke (push) Waiting to run
CI / golangci (push) Waiting to run
CI / frontend (push) Waiting to run
Release 3X-UI / build (386) (push) Waiting to run
Release 3X-UI / build (amd64) (push) Waiting to run
Release 3X-UI / build (arm64) (push) Waiting to run
Release 3X-UI / build (armv5) (push) Waiting to run
Release 3X-UI / build (armv6) (push) Waiting to run
Release 3X-UI / build (armv7) (push) Waiting to run
Release 3X-UI / build (s390x) (push) Waiting to run
Release 3X-UI / Build for Windows (amd64) (push) Waiting to run
Release 3X-UI / Publish rolling dev release (push) Blocked by required conditions
CodeQL Advanced / Analyze (go) (push) Waiting to run
CodeQL Advanced / Analyze (actions) (push) Waiting to run
CodeQL Advanced / Analyze (javascript-typescript) (push) Waiting to run
990 lines
33 KiB
Go
990 lines
33 KiB
Go
package controller
|
||
|
||
import (
|
||
"crypto/rand"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"encoding/json"
|
||
"errors"
|
||
"net"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
"github.com/mhsanaei/3x-ui/v3/internal/config"
|
||
"github.com/mhsanaei/3x-ui/v3/internal/database"
|
||
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
|
||
"github.com/mhsanaei/3x-ui/v3/internal/logger"
|
||
"github.com/mhsanaei/3x-ui/v3/internal/selfupdate"
|
||
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
|
||
"github.com/mhsanaei/3x-ui/v3/internal/web/service/probe"
|
||
"github.com/mhsanaei/3x-ui/v3/internal/xray"
|
||
|
||
"github.com/gin-gonic/gin"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// AgentController serves the lightweight agent API.
|
||
//
|
||
// Agent endpoints (register/heartbeat/config/traffic) authenticate with the
|
||
// agent's own Bearer token, which is NOT a panel API token — they are mounted
|
||
// on a public route group and must never go through the panel auth middleware.
|
||
// The panel-authed administration endpoints (token issuance, egress
|
||
// management) live under the authenticated /panel/api group instead.
|
||
type AgentController struct {
|
||
xrayService service.XrayService
|
||
singboxService service.SingboxService
|
||
inboundService service.InboundService
|
||
}
|
||
|
||
func NewAgentController(g *gin.RouterGroup) *AgentController {
|
||
a := &AgentController{}
|
||
a.initRouter(g)
|
||
return a
|
||
}
|
||
|
||
func (a *AgentController) initRouter(g *gin.RouterGroup) {
|
||
g.POST("/register", a.register)
|
||
g.POST("/heartbeat", a.heartbeat)
|
||
g.GET("/config", a.getConfig)
|
||
g.POST("/traffic", a.reportTraffic)
|
||
// V2.0.1 probe batch: the lightweight fast report (runtime
|
||
// metrics only) that replaces the metrics half of the heartbeat
|
||
// at the operator-configured cadence.
|
||
g.POST("/probe-report", a.probeReport)
|
||
}
|
||
|
||
// RegisterAdminRoutes mounts the panel-authenticated administration endpoints
|
||
// on an already-authed route group.
|
||
func (a *AgentController) RegisterAdminRoutes(g *gin.RouterGroup) {
|
||
g.POST("/tokens", a.createToken)
|
||
g.GET("/tokens", a.listTokens)
|
||
g.DELETE("/tokens/:id", a.deleteToken)
|
||
g.GET("/egress/:token", a.getEgress)
|
||
g.PUT("/egress/:token", a.setEgress)
|
||
g.GET("/install/:nodeId", a.getAgentInstall)
|
||
}
|
||
|
||
func hashAgentToken(token string) string {
|
||
sum := sha256.Sum256([]byte(token))
|
||
return hex.EncodeToString(sum[:])
|
||
}
|
||
|
||
var errRequest = errors.New("bad request")
|
||
|
||
func bearerToken(c *gin.Context) string {
|
||
auth := c.GetHeader("Authorization")
|
||
if len(auth) > 7 && auth[:7] == "Bearer " {
|
||
return auth[7:]
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func (a *AgentController) findAgent(token string) (*model.AgentNode, bool) {
|
||
if token == "" {
|
||
return nil, false
|
||
}
|
||
var agent model.AgentNode
|
||
if err := database.GetDB().Where("token_hash = ?", hashAgentToken(token)).First(&agent).Error; err != nil {
|
||
return nil, false
|
||
}
|
||
// Item 19 backfill: the token has just passed the hash check, so for
|
||
// rows created before plaintext storage existed, adopt the presented
|
||
// token as the recoverable copy. The guarded update only fires while
|
||
// token_plain is unset — an authenticated request can never overwrite
|
||
// an already-stored token (only issuance writes a fresh one). "Unset"
|
||
// must cover SQL NULL as well as '': AutoMigrate added the column to
|
||
// pre-existing rows as NULL, and NULL = '' is never true, so an
|
||
// equality-only guard silently matched zero legacy rows in
|
||
// production (the GORM-scanned value reads as "" either way, which is
|
||
// why the in-memory check above cannot distinguish them).
|
||
if agent.TokenPlain == "" {
|
||
if res := database.GetDB().Model(&model.AgentNode{}).
|
||
Where("id = ? AND (token_plain IS NULL OR token_plain = '')", agent.Id).
|
||
Update("token_plain", token); res.Error == nil && res.RowsAffected > 0 {
|
||
agent.TokenPlain = token
|
||
}
|
||
}
|
||
return &agent, true
|
||
}
|
||
|
||
// requestSourceIP extracts the real source IP of an agent request.
|
||
// Agents reach the master through Cloudflare, so the TCP peer is a CF
|
||
// edge and the agent's own address travels in CF-Connecting-IP; the
|
||
// first X-Forwarded-For hop is the next best signal, and the transport
|
||
// RemoteAddr is the direct-connection fallback. Only literal IPs are
|
||
// accepted — anything unparsable falls through to the next source.
|
||
func requestSourceIP(c *gin.Context) net.IP {
|
||
var candidates []string
|
||
if v := c.GetHeader("CF-Connecting-IP"); v != "" {
|
||
candidates = append(candidates, v)
|
||
}
|
||
if v := c.GetHeader("X-Forwarded-For"); v != "" {
|
||
candidates = append(candidates, strings.Split(v, ",")[0])
|
||
}
|
||
if host, _, err := net.SplitHostPort(c.Request.RemoteAddr); err == nil {
|
||
candidates = append(candidates, host)
|
||
} else {
|
||
candidates = append(candidates, c.Request.RemoteAddr)
|
||
}
|
||
for _, cand := range candidates {
|
||
if ip := net.ParseIP(strings.TrimSpace(cand)); ip != nil {
|
||
return ip
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// captureSeenIP records the source address of an authenticated agent
|
||
// request on the agent row (SeenIPv4/SeenIPv6). This is the master's
|
||
// view of the agent's public side: for a NAT'd agent it is the only
|
||
// place the public address ever appears. Writes are change-only —
|
||
// traffic reports land every ~10s and the address almost never moves,
|
||
// so an unchanged address must not turn into a DB write per request.
|
||
func (a *AgentController) captureSeenIP(c *gin.Context, agent *model.AgentNode) {
|
||
ip := requestSourceIP(c)
|
||
if ip == nil || agent == nil {
|
||
return
|
||
}
|
||
updates := map[string]any{"seen_ip_at": time.Now().UnixMilli()}
|
||
if v4 := ip.To4(); v4 != nil {
|
||
if agent.SeenIPv4 == v4.String() {
|
||
return
|
||
}
|
||
updates["seen_ipv4"] = v4.String()
|
||
} else {
|
||
if agent.SeenIPv6 == ip.String() {
|
||
return
|
||
}
|
||
updates["seen_ipv6"] = ip.String()
|
||
}
|
||
if err := database.GetDB().Model(&model.AgentNode{}).
|
||
Where("id = ?", agent.Id).Updates(updates).Error; err != nil {
|
||
return
|
||
}
|
||
if v, ok := updates["seen_ipv4"]; ok {
|
||
agent.SeenIPv4 = v.(string)
|
||
}
|
||
if v, ok := updates["seen_ipv6"]; ok {
|
||
agent.SeenIPv6 = v.(string)
|
||
}
|
||
agent.SeenIPAt = updates["seen_ip_at"].(int64)
|
||
}
|
||
|
||
func (a *AgentController) touchAgent(id uint, name string) {
|
||
updates := map[string]any{"last_seen": time.Now().UnixMilli()}
|
||
if name != "" {
|
||
updates["name"] = name
|
||
}
|
||
db := database.GetDB()
|
||
_ = db.Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error
|
||
// If bound to a node, refresh the node's heartbeat so the legacy node
|
||
// list shows it online (the legacy probe can't reach agent-only nodes).
|
||
var nodeID *int
|
||
_ = db.Model(&model.AgentNode{}).Select("node_id").Where("id = ?", id).Scan(&nodeID).Error
|
||
if nodeID != nil {
|
||
now := time.Now().UnixMilli()
|
||
_ = db.Exec("UPDATE nodes SET last_heartbeat = ?, status = 'online' WHERE id = ?", now, *nodeID).Error
|
||
}
|
||
}
|
||
|
||
// register activates a pre-issued token. Tokens are issued bound to a node via
|
||
// POST /panel/api/agent/tokens; the agent only claims its token here.
|
||
func (a *AgentController) register(c *gin.Context) {
|
||
var req struct {
|
||
Token string `json:"token"`
|
||
Name string `json:"name"`
|
||
ProbeFacts *heartbeatProbeFacts `json:"probeFacts"`
|
||
}
|
||
if err := c.ShouldBindJSON(&req); err != nil || req.Token == "" {
|
||
jsonMsg(c, "Token required", errRequest)
|
||
return
|
||
}
|
||
agent, ok := a.findAgent(req.Token)
|
||
if !ok {
|
||
c.JSON(401, gin.H{"message": "unknown token"})
|
||
return
|
||
}
|
||
a.captureSeenIP(c, agent)
|
||
a.touchAgent(uint(agent.Id), req.Name)
|
||
if req.ProbeFacts != nil {
|
||
a.updateAgentProbe(uint(agent.Id), req.ProbeFacts, nil)
|
||
}
|
||
logger.Infof("Agent registered: %s (token %s...)", req.Name, agent.TokenPrefix)
|
||
jsonObj(c, gin.H{"message": "registered"}, nil)
|
||
}
|
||
|
||
// heartbeatTelemetry is the optional telemetry object an agent piggybacks
|
||
// on its heartbeat (P13). Every field is a pointer: a missing field means
|
||
// "not reported this round" and must not overwrite the stored value. Older
|
||
// agents send no telemetry object at all and are handled exactly as before.
|
||
type heartbeatTelemetry struct {
|
||
AgentVersion *string `json:"agentVersion"`
|
||
XrayVersion *string `json:"xrayVersion"`
|
||
SingboxVersion *string `json:"singboxVersion"`
|
||
UptimeSeconds *uint64 `json:"uptimeSeconds"`
|
||
LatencyMs *int `json:"latencyMs"`
|
||
CpuPercent *float64 `json:"cpuPercent"`
|
||
MemPercent *float64 `json:"memPercent"`
|
||
IPv4 *string `json:"ipv4"`
|
||
IPv6 *string `json:"ipv6"`
|
||
}
|
||
|
||
// updateAgentTelemetry persists the reported telemetry fields onto the
|
||
// agent row. Only fields actually present in the report are written.
|
||
func (a *AgentController) updateAgentTelemetry(id uint, t *heartbeatTelemetry) {
|
||
updates := map[string]any{"telemetry_at": time.Now().UnixMilli()}
|
||
if t.AgentVersion != nil {
|
||
updates["agent_version"] = *t.AgentVersion
|
||
}
|
||
if t.XrayVersion != nil {
|
||
updates["xray_version"] = *t.XrayVersion
|
||
}
|
||
if t.SingboxVersion != nil {
|
||
updates["singbox_version"] = *t.SingboxVersion
|
||
}
|
||
if t.UptimeSeconds != nil {
|
||
updates["uptime_secs"] = *t.UptimeSeconds
|
||
}
|
||
if t.LatencyMs != nil {
|
||
updates["latency_ms"] = *t.LatencyMs
|
||
}
|
||
if t.CpuPercent != nil {
|
||
updates["cpu_pct"] = *t.CpuPercent
|
||
}
|
||
if t.MemPercent != nil {
|
||
updates["mem_pct"] = *t.MemPercent
|
||
}
|
||
if t.IPv4 != nil {
|
||
updates["ipv4"] = *t.IPv4
|
||
}
|
||
if t.IPv6 != nil {
|
||
updates["ipv6"] = *t.IPv6
|
||
}
|
||
_ = database.GetDB().Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error
|
||
}
|
||
|
||
// heartbeatProbeFacts / heartbeatProbeMetrics are the monitor-probe-style
|
||
// host snapshot an agent piggybacks on register and heartbeat (item 7 B1).
|
||
// Every field is a pointer for the same reason as telemetry: an absent
|
||
// field means "not reported" and must not overwrite the stored value, so
|
||
// old agents (no probe block at all) and partial reports both merge
|
||
// cleanly. JSON names must match internal/agent's ProbeFacts/ProbeMetrics.
|
||
type heartbeatProbeFacts struct {
|
||
Hostname *string `json:"hostname"`
|
||
OS *string `json:"os"`
|
||
Kernel *string `json:"kernel"`
|
||
Arch *string `json:"arch"`
|
||
Virt *string `json:"virt"`
|
||
CPUName *string `json:"cpuName"`
|
||
CPUCores *uint32 `json:"cpuCores"`
|
||
MemTotal *uint64 `json:"memTotal"`
|
||
SwapTotal *uint64 `json:"swapTotal"`
|
||
DiskTotal *uint64 `json:"diskTotal"`
|
||
}
|
||
|
||
type heartbeatProbeMetrics struct {
|
||
BootID *string `json:"bootId"`
|
||
Iface *string `json:"iface"`
|
||
Uptime *uint64 `json:"uptime"`
|
||
Load *[3]float32 `json:"load"`
|
||
MemTotal *uint64 `json:"memTotal"`
|
||
MemUsed *uint64 `json:"memUsed"`
|
||
SwapTotal *uint64 `json:"swapTotal"`
|
||
SwapUsed *uint64 `json:"swapUsed"`
|
||
DiskTotal *uint64 `json:"diskTotal"`
|
||
DiskUsed *uint64 `json:"diskUsed"`
|
||
NetRxTotal *uint64 `json:"netRxTotal"`
|
||
NetTxTotal *uint64 `json:"netTxTotal"`
|
||
NetRx *uint64 `json:"netRx"`
|
||
NetTx *uint64 `json:"netTx"`
|
||
TCP *uint32 `json:"tcp"`
|
||
UDP *uint32 `json:"udp"`
|
||
Procs *uint32 `json:"procs"`
|
||
CPU *float32 `json:"cpu"`
|
||
}
|
||
|
||
// updateAgentProbe persists the reported probe facts/metrics onto the agent
|
||
// row. Only fields actually present in the report are written. Facts and
|
||
// metrics share one update so a heartbeat carries a single write; register
|
||
// calls it with metrics == nil.
|
||
func (a *AgentController) updateAgentProbe(id uint, f *heartbeatProbeFacts, m *heartbeatProbeMetrics) {
|
||
updates := map[string]any{}
|
||
if f != nil {
|
||
if f.Hostname != nil {
|
||
updates["probe_hostname"] = *f.Hostname
|
||
}
|
||
if f.OS != nil {
|
||
updates["probe_os"] = *f.OS
|
||
}
|
||
if f.Kernel != nil {
|
||
updates["probe_kernel"] = *f.Kernel
|
||
}
|
||
if f.Arch != nil {
|
||
updates["probe_arch"] = *f.Arch
|
||
}
|
||
if f.Virt != nil {
|
||
updates["probe_virt"] = *f.Virt
|
||
}
|
||
if f.CPUName != nil {
|
||
updates["probe_cpu_name"] = *f.CPUName
|
||
}
|
||
if f.CPUCores != nil {
|
||
updates["probe_cpu_cores"] = *f.CPUCores
|
||
}
|
||
if f.MemTotal != nil {
|
||
updates["probe_mem_total"] = *f.MemTotal
|
||
}
|
||
if f.SwapTotal != nil {
|
||
updates["probe_swap_total"] = *f.SwapTotal
|
||
}
|
||
if f.DiskTotal != nil {
|
||
updates["probe_disk_total"] = *f.DiskTotal
|
||
}
|
||
}
|
||
if m != nil {
|
||
updates["probe_at"] = time.Now().UnixMilli()
|
||
if m.BootID != nil {
|
||
updates["probe_boot_id"] = *m.BootID
|
||
}
|
||
if m.Iface != nil {
|
||
updates["probe_iface"] = *m.Iface
|
||
}
|
||
if m.Load != nil {
|
||
updates["probe_load1"] = (*m.Load)[0]
|
||
updates["probe_load5"] = (*m.Load)[1]
|
||
updates["probe_load15"] = (*m.Load)[2]
|
||
}
|
||
if m.MemTotal != nil {
|
||
updates["probe_mem_total"] = *m.MemTotal
|
||
}
|
||
if m.MemUsed != nil {
|
||
updates["probe_mem_used"] = *m.MemUsed
|
||
}
|
||
if m.SwapTotal != nil {
|
||
updates["probe_swap_total"] = *m.SwapTotal
|
||
}
|
||
if m.SwapUsed != nil {
|
||
updates["probe_swap_used"] = *m.SwapUsed
|
||
}
|
||
if m.DiskTotal != nil {
|
||
updates["probe_disk_total"] = *m.DiskTotal
|
||
}
|
||
if m.DiskUsed != nil {
|
||
updates["probe_disk_used"] = *m.DiskUsed
|
||
}
|
||
if m.NetRxTotal != nil {
|
||
updates["probe_net_rx_tot"] = *m.NetRxTotal
|
||
}
|
||
if m.NetTxTotal != nil {
|
||
updates["probe_net_tx_tot"] = *m.NetTxTotal
|
||
}
|
||
if m.NetRx != nil {
|
||
updates["probe_net_rx"] = *m.NetRx
|
||
}
|
||
if m.NetTx != nil {
|
||
updates["probe_net_tx"] = *m.NetTx
|
||
}
|
||
if m.TCP != nil {
|
||
updates["probe_tcp"] = *m.TCP
|
||
}
|
||
if m.UDP != nil {
|
||
updates["probe_udp"] = *m.UDP
|
||
}
|
||
if m.Procs != nil {
|
||
updates["probe_procs"] = *m.Procs
|
||
}
|
||
if m.CPU != nil {
|
||
updates["probe_cpu"] = *m.CPU
|
||
}
|
||
}
|
||
if len(updates) == 0 {
|
||
return
|
||
}
|
||
_ = database.GetDB().Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error
|
||
}
|
||
|
||
// probeSample flattens a heartbeat metrics block into the probe
|
||
// package's Sample. Absent fields become zero values — the agent sends
|
||
// a full block whenever it sends one, and the probe store treats the
|
||
// block's presence (not per-field presence) as the signal, mirroring
|
||
// the hub's report frames.
|
||
func probeSample(m *heartbeatProbeMetrics) probe.Sample {
|
||
var s probe.Sample
|
||
if m.BootID != nil {
|
||
s.BootID = *m.BootID
|
||
}
|
||
if m.Uptime != nil {
|
||
s.Uptime = *m.Uptime
|
||
}
|
||
if m.CPU != nil {
|
||
s.CPU = *m.CPU
|
||
}
|
||
if m.MemUsed != nil {
|
||
s.MemUsed = *m.MemUsed
|
||
}
|
||
if m.SwapUsed != nil {
|
||
s.SwapUsed = *m.SwapUsed
|
||
}
|
||
if m.DiskUsed != nil {
|
||
s.DiskUsed = *m.DiskUsed
|
||
}
|
||
if m.NetRxTotal != nil {
|
||
s.NetRxTotal = *m.NetRxTotal
|
||
}
|
||
if m.NetTxTotal != nil {
|
||
s.NetTxTotal = *m.NetTxTotal
|
||
}
|
||
if m.NetRx != nil {
|
||
s.NetRx = *m.NetRx
|
||
}
|
||
if m.NetTx != nil {
|
||
s.NetTx = *m.NetTx
|
||
}
|
||
if m.TCP != nil {
|
||
s.TCP = *m.TCP
|
||
}
|
||
if m.UDP != nil {
|
||
s.UDP = *m.UDP
|
||
}
|
||
if m.Procs != nil {
|
||
s.Procs = *m.Procs
|
||
}
|
||
return s
|
||
}
|
||
|
||
// heartbeatUpdateStatus is the agent's report about a one-click update
|
||
// intent (item 20), piggybacked on the heartbeat like telemetry.
|
||
type heartbeatUpdateStatus struct {
|
||
RequestedAt int64 `json:"requestedAt"`
|
||
State string `json:"state"`
|
||
Error string `json:"error"`
|
||
}
|
||
|
||
// applyAgentUpdateStatus folds an agent's update report into the stored
|
||
// intent. Reports for a superseded intent (requested_at mismatch while an
|
||
// intent is pending) are ignored. Terminal reports (failed / current)
|
||
// clear the intent so the config poll stops re-offering it; the recorded
|
||
// state stays visible on the node row.
|
||
func (a *AgentController) applyAgentUpdateStatus(id uint, st *heartbeatUpdateStatus) {
|
||
db := database.GetDB()
|
||
var row model.AgentNode
|
||
if err := db.First(&row, id).Error; err != nil {
|
||
return
|
||
}
|
||
if row.UpdateRequestedAt == 0 || st.RequestedAt != row.UpdateRequestedAt {
|
||
return
|
||
}
|
||
updates := map[string]any{"update_state_at": time.Now().UnixMilli()}
|
||
switch st.State {
|
||
case "downloading", "swapped":
|
||
updates["update_state"] = st.State
|
||
updates["update_error"] = ""
|
||
case "failed":
|
||
updates["update_state"] = "failed"
|
||
updates["update_error"] = st.Error
|
||
updates["update_requested_at"] = int64(0)
|
||
case "current":
|
||
updates["update_state"] = "current"
|
||
updates["update_error"] = ""
|
||
updates["update_requested_at"] = int64(0)
|
||
default:
|
||
return
|
||
}
|
||
_ = db.Model(&model.AgentNode{}).Where("id = ?", id).Updates(updates).Error
|
||
}
|
||
|
||
// confirmAgentUpdate marks a pending intent successful once the agent
|
||
// reports an A-series version at or beyond the intent's target — the new
|
||
// version appearing is the end-to-end success signal of the update.
|
||
func (a *AgentController) confirmAgentUpdate(id uint, reportedVersion string) {
|
||
if reportedVersion == "" {
|
||
return
|
||
}
|
||
db := database.GetDB()
|
||
var row model.AgentNode
|
||
if err := db.First(&row, id).Error; err != nil {
|
||
return
|
||
}
|
||
if row.UpdateRequestedAt == 0 || row.UpdateTarget == "" {
|
||
return
|
||
}
|
||
if cmp, ok := config.CompareDisplayVersion(reportedVersion, row.UpdateTarget); !ok || cmp < 0 {
|
||
return
|
||
}
|
||
_ = db.Model(&model.AgentNode{}).Where("id = ?", id).Updates(map[string]any{
|
||
"update_state": "success",
|
||
"update_error": "",
|
||
"update_requested_at": int64(0),
|
||
"update_state_at": time.Now().UnixMilli(),
|
||
}).Error
|
||
}
|
||
|
||
// heartbeatProbeResult is one latency sample an agent piggybacks on
|
||
// its heartbeat (item 7 B4). LatencyMs is a pointer: an absent field
|
||
// means "no sample" (a resolution overrun produces none) and the row
|
||
// is skipped, while a present negative value is a genuine recorded
|
||
// miss — the two must never collapse into each other.
|
||
type heartbeatProbeResult struct {
|
||
TaskID int64 `json:"taskId"`
|
||
LatencyMs *int64 `json:"latencyMs"`
|
||
}
|
||
|
||
// maxProbeResultsPerHeartbeat caps one heartbeat's probe batch. The
|
||
// agent's buffer holds far less than this in practice (64 tasks × 60 s
|
||
// at the fastest legal interval is under a thousand samples); the cap
|
||
// exists so a buggy or hostile reporter cannot make one request file
|
||
// an unbounded batch.
|
||
const maxProbeResultsPerHeartbeat = 2048
|
||
|
||
func (a *AgentController) heartbeat(c *gin.Context) {
|
||
agent, ok := a.findAgent(bearerToken(c))
|
||
if !ok {
|
||
c.JSON(401, gin.H{"message": "unauthorized"})
|
||
return
|
||
}
|
||
a.captureSeenIP(c, agent)
|
||
// Telemetry is optional and best-effort: a body that fails to parse
|
||
// (or no body at all) must never fail the heartbeat itself.
|
||
var req struct {
|
||
Telemetry *heartbeatTelemetry `json:"telemetry"`
|
||
UpdateStatus *heartbeatUpdateStatus `json:"updateStatus"`
|
||
ProbeFacts *heartbeatProbeFacts `json:"probeFacts"`
|
||
ProbeMetrics *heartbeatProbeMetrics `json:"probeMetrics"`
|
||
ProbeResults []heartbeatProbeResult `json:"probeResults"`
|
||
}
|
||
_ = c.ShouldBindJSON(&req)
|
||
// The last-seen stamp as it stood before this heartbeat: the probe
|
||
// ledger's unseen-boot rule measures a reboot against it, so it
|
||
// must be captured before touchAgent overwrites it.
|
||
prevLastSeen := agent.LastSeen
|
||
a.touchAgent(uint(agent.Id), "")
|
||
reportedVersion := agent.AgentVersion
|
||
if req.ProbeFacts != nil || req.ProbeMetrics != nil {
|
||
a.updateAgentProbe(uint(agent.Id), req.ProbeFacts, req.ProbeMetrics)
|
||
}
|
||
if req.ProbeMetrics != nil {
|
||
// Item 7 B2: file the sample into probe.db history + the
|
||
// whole-machine traffic ledger. Best-effort by design — the
|
||
// probe package logs and swallows its own failures.
|
||
probe.Record(int64(agent.Id), probeSample(req.ProbeMetrics), prevLastSeen)
|
||
}
|
||
if len(req.ProbeResults) > 0 {
|
||
// Item 7 B4: file the latency batch, stamped at arrival (hub
|
||
// semantics). Results ride the heartbeat rather than the
|
||
// traffic report on purpose: probing must keep working while
|
||
// xray is down, and the traffic report early-returns then.
|
||
results := make([]probe.PingResult, 0, len(req.ProbeResults))
|
||
for _, r := range req.ProbeResults {
|
||
if r.TaskID <= 0 || r.LatencyMs == nil {
|
||
continue
|
||
}
|
||
results = append(results, probe.PingResult{TaskID: r.TaskID, LatencyMs: *r.LatencyMs})
|
||
if len(results) >= maxProbeResultsPerHeartbeat {
|
||
break
|
||
}
|
||
}
|
||
probe.InsertPings(int64(agent.Id), results)
|
||
}
|
||
if req.Telemetry != nil {
|
||
a.updateAgentTelemetry(uint(agent.Id), req.Telemetry)
|
||
if req.Telemetry.AgentVersion != nil && *req.Telemetry.AgentVersion != "" {
|
||
reportedVersion = *req.Telemetry.AgentVersion
|
||
}
|
||
}
|
||
if req.UpdateStatus != nil {
|
||
a.applyAgentUpdateStatus(uint(agent.Id), req.UpdateStatus)
|
||
}
|
||
a.confirmAgentUpdate(uint(agent.Id), reportedVersion)
|
||
jsonObj(c, gin.H{"message": "ok"}, nil)
|
||
}
|
||
|
||
// getConfig returns the full sing-box config for the agent's bound node.
|
||
//
|
||
// ITEM6 B5 (switchover): the sing-box config is the only runtime
|
||
// payload this master serves. The xray payload (config_hash /
|
||
// xray_config) is retired — generation reads only the sing-box data
|
||
// model since the B1 migration. Dual-capability agents (A2.0.0) take
|
||
// singbox_config as the switch signal: they flush their xray tail
|
||
// traffic, stop the xray core and run sing-box alone. Older agents
|
||
// ignore the unknown keys and keep running their last applied xray
|
||
// config until they are updated, which is why the rollout order is
|
||
// agents first, master second (ITEM6-SCOPE §6). A build failure is a
|
||
// hard 500: the agent keeps its current runtime state and retries on
|
||
// the next poll, and the failure stays visible in both logs instead
|
||
// of silently freezing the node's config.
|
||
// probeStatIface reads the agent's counted-interface pin from its
|
||
// probe metadata ("" = automatic selection, and the answer for any
|
||
// read failure — a config poll must not fail over a display-side
|
||
// setting).
|
||
func probeStatIface(agentID int64) string {
|
||
meta, found, err := probe.MetaFor(agentID)
|
||
if err != nil || !found {
|
||
return ""
|
||
}
|
||
return meta.StatIface
|
||
}
|
||
|
||
// probeReport ingests one fast report (V2.0.1): a probe metrics
|
||
// block and nothing else. Since V2.0.3 it also touches the agent's
|
||
// last-seen stamp, like every other report channel: the hub's
|
||
// themes gate each batch on the node's last_seen being newer than
|
||
// the last one they applied, so a metrics-only channel that never
|
||
// advanced the stamp froze the public cards between the 10 s
|
||
// traffic reports. Liveness now means "any report from the agent"
|
||
// — all channels come from the same process — and the traffic
|
||
// ledger's unseen-boot rule is unaffected: Record still receives
|
||
// the pre-touch stamp, so gaps measure exactly as before. Agents
|
||
// on older masters never see this endpoint (404) and simply keep
|
||
// reporting metrics on the heartbeat; masters accept metrics on
|
||
// either channel.
|
||
func (a *AgentController) probeReport(c *gin.Context) {
|
||
agent, ok := a.findAgent(bearerToken(c))
|
||
if !ok {
|
||
c.JSON(401, gin.H{"message": "unauthorized"})
|
||
return
|
||
}
|
||
a.captureSeenIP(c, agent)
|
||
var req struct {
|
||
ProbeMetrics *heartbeatProbeMetrics `json:"probeMetrics"`
|
||
}
|
||
_ = c.ShouldBindJSON(&req)
|
||
// The last-seen stamp as it stood before this report (see the
|
||
// heartbeat handler): captured before touchAgent overwrites it.
|
||
prevLastSeen := agent.LastSeen
|
||
a.touchAgent(uint(agent.Id), "")
|
||
if req.ProbeMetrics != nil {
|
||
a.updateAgentProbe(uint(agent.Id), nil, req.ProbeMetrics)
|
||
probe.Record(int64(agent.Id), probeSample(req.ProbeMetrics), prevLastSeen)
|
||
}
|
||
jsonObj(c, gin.H{"message": "ok"}, nil)
|
||
}
|
||
|
||
func (a *AgentController) getConfig(c *gin.Context) {
|
||
agent, ok := a.findAgent(bearerToken(c))
|
||
if !ok {
|
||
c.JSON(401, gin.H{"message": "unauthorized"})
|
||
return
|
||
}
|
||
a.captureSeenIP(c, agent)
|
||
if agent.NodeID == nil {
|
||
c.JSON(409, gin.H{"message": "agent token is not bound to a node"})
|
||
return
|
||
}
|
||
sbCfg, sbErr := a.singboxService.GetSingboxConfigForAgent(*agent.NodeID, agent.SingboxOutbounds, agent.SingboxRoutingRules)
|
||
if sbErr != nil {
|
||
logger.Warningf("agent config: singbox config build failed for node %d: %v", *agent.NodeID, sbErr)
|
||
c.JSON(500, gin.H{"message": "failed to build sing-box config"})
|
||
return
|
||
}
|
||
sbRaw, mErr := json.Marshal(sbCfg)
|
||
if mErr != nil {
|
||
logger.Warningf("agent config: singbox config encode failed for node %d: %v", *agent.NodeID, mErr)
|
||
c.JSON(500, gin.H{"message": "failed to encode sing-box config"})
|
||
return
|
||
}
|
||
sbSum := sha256.Sum256(sbRaw)
|
||
payload := gin.H{
|
||
"singbox_config": json.RawMessage(sbRaw),
|
||
"singbox_config_hash": hex.EncodeToString(sbSum[:]),
|
||
// Item 7 B4: the agent's probe task set rides the existing
|
||
// config poll (no new channel). Always present from this
|
||
// master version — an empty list means "run no probes", while
|
||
// an absent field (older master) leaves the agent's current
|
||
// set untouched.
|
||
"probe_tasks": probe.PingTasksFor(int64(agent.Id)),
|
||
// V2.0.1 probe batch: the fast-report cadence and this node's
|
||
// counted-interface pin ride the same poll. Older agents
|
||
// ignore both keys and keep the pre-V2.0.1 behaviour
|
||
// (metrics on the 60 s heartbeat, automatic interface
|
||
// selection).
|
||
"probe_fast_interval": service.ProbeFastIntervalSec(),
|
||
"probe_stat_iface": probeStatIface(int64(agent.Id)),
|
||
}
|
||
// Item 20: a pending one-click update intent rides along on the config
|
||
// poll (no new channel — the agent already polls this endpoint).
|
||
if agent.UpdateRequestedAt > 0 &&
|
||
(agent.UpdateState == "pending" || agent.UpdateState == "downloading") {
|
||
payload["agent_update"] = gin.H{
|
||
"requested_at": agent.UpdateRequestedAt,
|
||
"target": agent.UpdateTarget,
|
||
"tag": agent.UpdateTag,
|
||
"base_url": selfupdate.BaseURL(),
|
||
}
|
||
}
|
||
c.JSON(200, payload)
|
||
}
|
||
|
||
type agentTrafficReport struct {
|
||
Inbounds []struct {
|
||
Tag string `json:"tag"`
|
||
Up int64 `json:"up"`
|
||
Down int64 `json:"down"`
|
||
} `json:"inbounds"`
|
||
Users []struct {
|
||
Email string `json:"email"`
|
||
Up int64 `json:"up"`
|
||
Down int64 `json:"down"`
|
||
} `json:"users"`
|
||
// Onlines lists client emails with at least one live connection.
|
||
Onlines []string `json:"onlines"`
|
||
}
|
||
|
||
// reportTraffic ingests per-poll traffic deltas from the agent. Inbound deltas
|
||
// land on the agent's node inbound rows; client deltas reuse the shared
|
||
// email-keyed write path (plus its lifecycle maintenance, which also drives
|
||
// config changes the agent picks up on its next poll).
|
||
func (a *AgentController) reportTraffic(c *gin.Context) {
|
||
agent, ok := a.findAgent(bearerToken(c))
|
||
if !ok {
|
||
c.JSON(401, gin.H{"message": "unauthorized"})
|
||
return
|
||
}
|
||
a.captureSeenIP(c, agent)
|
||
var req agentTrafficReport
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
jsonMsg(c, "Invalid request", err)
|
||
return
|
||
}
|
||
// ITEM6 B5 recording gate: a sing-box core registers counters
|
||
// only when every stats family is listed in its config, so the
|
||
// generated config can no longer omit a switched-off category
|
||
// (see the stats wiring in SingboxService). The whole-network
|
||
// accounting switches (settings.trafficStatsSwitches, the
|
||
// master switch on the Settings page) are therefore enforced
|
||
// here, at recording time: with the inbound category off
|
||
// (either direction), inbound totals and per-user traffic are
|
||
// accepted but not recorded — the off period is never counted,
|
||
// matching the switches' documented no-backfill semantics.
|
||
// Online presence is not traffic accounting and stays ungated.
|
||
sw := a.singboxService.GetTrafficStatsSwitches()
|
||
recordInbound := sw.InboundUplink && sw.InboundDownlink
|
||
db := database.GetDB()
|
||
if recordInbound && agent.NodeID != nil {
|
||
for _, ib := range req.Inbounds {
|
||
if ib.Tag == "" || ib.Up+ib.Down <= 0 {
|
||
continue
|
||
}
|
||
_ = db.Model(&model.Inbound{}).
|
||
Where("tag = ? AND node_id = ?", ib.Tag, *agent.NodeID).
|
||
Updates(map[string]any{
|
||
"up": gorm.Expr(database.ClampedAddExpr("up"), ib.Up),
|
||
"down": gorm.Expr(database.ClampedAddExpr("down"), ib.Down),
|
||
}).Error
|
||
}
|
||
}
|
||
var clientTraffics []*xray.ClientTraffic
|
||
if recordInbound {
|
||
for _, u := range req.Users {
|
||
if u.Email == "" || u.Up+u.Down <= 0 {
|
||
continue
|
||
}
|
||
clientTraffics = append(clientTraffics, &xray.ClientTraffic{
|
||
Email: u.Email,
|
||
Up: u.Up,
|
||
Down: u.Down,
|
||
})
|
||
}
|
||
}
|
||
if len(clientTraffics) > 0 {
|
||
if _, _, err := a.inboundService.AddTraffic(nil, clientTraffics); err != nil {
|
||
jsonMsg(c, "Failed to store traffic", err)
|
||
return
|
||
}
|
||
}
|
||
if agent.NodeID != nil {
|
||
a.inboundService.SetAgentOnlineClients(*agent.NodeID, req.Onlines)
|
||
}
|
||
a.touchAgent(uint(agent.Id), "")
|
||
jsonObj(c, gin.H{"message": "ok"}, nil)
|
||
}
|
||
|
||
// issueAgentToken generates a new agent token bound to nodeID. The hash is
|
||
// what authenticates; the raw token is also kept recoverably (TokenPlain,
|
||
// item 19) so the node edit page can re-display the install command.
|
||
func issueAgentToken(nodeID *int, name string) (string, int, error) {
|
||
raw := make([]byte, 32)
|
||
if _, err := rand.Read(raw); err != nil {
|
||
return "", 0, err
|
||
}
|
||
token := hex.EncodeToString(raw)
|
||
agent := model.AgentNode{
|
||
TokenHash: hashAgentToken(token),
|
||
TokenPrefix: token[:12],
|
||
TokenPlain: token,
|
||
Name: name,
|
||
NodeID: nodeID,
|
||
}
|
||
if err := database.GetDB().Create(&agent).Error; err != nil {
|
||
return "", 0, err
|
||
}
|
||
return token, agent.Id, nil
|
||
}
|
||
|
||
// createToken issues a new agent token bound to a node. The raw token is
|
||
// returned in this response and kept recoverably for the install command.
|
||
func (a *AgentController) createToken(c *gin.Context) {
|
||
var req struct {
|
||
NodeID *int `json:"node_id"`
|
||
Name string `json:"name"`
|
||
}
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
jsonMsg(c, "Invalid request", err)
|
||
return
|
||
}
|
||
if req.NodeID != nil {
|
||
var node model.Node
|
||
if err := database.GetDB().First(&node, *req.NodeID).Error; err != nil {
|
||
jsonMsg(c, "Node not found", errRequest)
|
||
return
|
||
}
|
||
}
|
||
token, _, err := issueAgentToken(req.NodeID, req.Name)
|
||
if err != nil {
|
||
jsonMsg(c, "Failed to generate token", err)
|
||
return
|
||
}
|
||
var agent model.AgentNode
|
||
_ = database.GetDB().Where("token_hash = ?", hashAgentToken(token)).First(&agent).Error
|
||
jsonObj(c, gin.H{
|
||
"id": agent.Id,
|
||
"token": token,
|
||
"token_prefix": agent.TokenPrefix,
|
||
"node_id": agent.NodeID,
|
||
"name": agent.Name,
|
||
}, nil)
|
||
}
|
||
|
||
// getAgentInstall returns what the node edit page needs to re-display the
|
||
// full install command for an agent node (item 19): the agent's name and,
|
||
// when the master holds a recoverable copy, its raw token. Legacy rows
|
||
// whose token has not been backfilled yet answer hasToken=false with an
|
||
// empty token — the UI then shows the command without the token and points
|
||
// at /etc/x-ui/agent.json on the node itself.
|
||
func (a *AgentController) getAgentInstall(c *gin.Context) {
|
||
nodeID, err := strconv.Atoi(c.Param("nodeId"))
|
||
if err != nil {
|
||
jsonMsg(c, "Invalid node id", errRequest)
|
||
return
|
||
}
|
||
var agent model.AgentNode
|
||
if err := database.GetDB().Where("node_id = ?", nodeID).
|
||
Order("last_seen DESC, id DESC").First(&agent).Error; err != nil {
|
||
jsonMsg(c, "Agent not found for node", errRequest)
|
||
return
|
||
}
|
||
jsonObj(c, gin.H{
|
||
"nodeId": nodeID,
|
||
"name": agent.Name,
|
||
"token": agent.TokenPlain,
|
||
"hasToken": agent.TokenPlain != "",
|
||
}, nil)
|
||
}
|
||
|
||
func (a *AgentController) listTokens(c *gin.Context) {
|
||
var agents []model.AgentNode
|
||
_ = database.GetDB().Order("id DESC").Find(&agents).Error
|
||
out := make([]gin.H, 0, len(agents))
|
||
for _, ag := range agents {
|
||
out = append(out, gin.H{
|
||
"id": ag.Id,
|
||
"token_prefix": ag.TokenPrefix,
|
||
"name": ag.Name,
|
||
"node_id": ag.NodeID,
|
||
"last_seen": ag.LastSeen,
|
||
"has_egress": ag.Outbounds != "" || ag.RoutingRules != "",
|
||
"created_at": ag.CreatedAt,
|
||
})
|
||
}
|
||
jsonObj(c, out, nil)
|
||
}
|
||
|
||
func (a *AgentController) deleteToken(c *gin.Context) {
|
||
_ = database.GetDB().Delete(&model.AgentNode{}, "id = ?", c.Param("id")).Error
|
||
// Item 7 B3: the agent's probe data dies with its row (hub
|
||
// delete_node cascades the same set, metrics and traffic
|
||
// included) and the shared snapshot drops it at once.
|
||
if id, err := strconv.Atoi(c.Param("id")); err == nil {
|
||
probe.DeleteAgentData(int64(id))
|
||
service.InvalidateProbeSnapshot()
|
||
}
|
||
jsonObj(c, gin.H{"message": "deleted"}, nil)
|
||
}
|
||
|
||
// getEgress returns the panel-managed egress (outbounds + routing rules) for
|
||
// the agent identified by its token.
|
||
func (a *AgentController) getEgress(c *gin.Context) {
|
||
agent, ok := a.findAgent(c.Param("token"))
|
||
if !ok {
|
||
c.JSON(404, gin.H{"message": "agent not found"})
|
||
return
|
||
}
|
||
jsonObj(c, gin.H{
|
||
"id": agent.Id,
|
||
"token_prefix": agent.TokenPrefix,
|
||
"name": agent.Name,
|
||
"node_id": agent.NodeID,
|
||
"outbounds": json.RawMessage(orEmptyArray(agent.Outbounds)),
|
||
"routing_rules": json.RawMessage(orEmptyArray(agent.RoutingRules)),
|
||
}, nil)
|
||
}
|
||
|
||
// setEgress replaces the panel-managed egress for the agent. The panel is the
|
||
// only source of truth: agents never report local egress, they only apply
|
||
// what the generated config carries.
|
||
func (a *AgentController) setEgress(c *gin.Context) {
|
||
agent, ok := a.findAgent(c.Param("token"))
|
||
if !ok {
|
||
c.JSON(404, gin.H{"message": "agent not found"})
|
||
return
|
||
}
|
||
var req struct {
|
||
Outbounds json.RawMessage `json:"outbounds"`
|
||
RoutingRules json.RawMessage `json:"routing_rules"`
|
||
}
|
||
if err := c.ShouldBindJSON(&req); err != nil {
|
||
jsonMsg(c, "Invalid request", err)
|
||
return
|
||
}
|
||
outbounds, err := canonicalJSONArray(req.Outbounds)
|
||
if err != nil {
|
||
jsonMsg(c, "outbounds must be a JSON array", err)
|
||
return
|
||
}
|
||
rules, err := canonicalJSONArray(req.RoutingRules)
|
||
if err != nil {
|
||
jsonMsg(c, "routing_rules must be a JSON array", err)
|
||
return
|
||
}
|
||
if err := database.GetDB().Model(&model.AgentNode{}).Where("id = ?", agent.Id).
|
||
Updates(map[string]any{"outbounds": outbounds, "routing_rules": rules}).Error; err != nil {
|
||
jsonMsg(c, "Failed to store egress", err)
|
||
return
|
||
}
|
||
jsonObj(c, gin.H{"message": "saved"}, nil)
|
||
}
|
||
|
||
func orEmptyArray(s string) string {
|
||
if s == "" {
|
||
return "[]"
|
||
}
|
||
return s
|
||
}
|
||
|
||
// canonicalJSONArray validates that raw is a JSON array and returns it in
|
||
// canonical form. Empty input yields an empty string (template defaults).
|
||
func canonicalJSONArray(raw json.RawMessage) (string, error) {
|
||
if len(raw) == 0 {
|
||
return "", nil
|
||
}
|
||
var arr []any
|
||
if err := json.Unmarshal(raw, &arr); err != nil {
|
||
return "", err
|
||
}
|
||
out, err := json.Marshal(arr)
|
||
if err != nil {
|
||
return "", err
|
||
}
|
||
return string(out), nil
|
||
}
|