merge: item 3 agent v4 beacon

This commit is contained in:
Muse
2026-10-24 10:09:16 +00:00
5 changed files with 452 additions and 2 deletions
+13 -2
View File
@@ -66,6 +66,7 @@ type Config struct {
type Agent struct {
config *Config
httpClient *http.Client
v4httpClient *http.Client
stopCh chan struct{}
lastConfigHash string
xrayProc *xray.Process
@@ -166,8 +167,9 @@ func New(cfg *Config) *Agent {
TLSClientConfig: &tls.Config{InsecureSkipVerify: false},
},
},
stopCh: make(chan struct{}),
probes: newProbeRunner(),
v4httpClient: newV4BeaconClient(),
stopCh: make(chan struct{}),
probes: newProbeRunner(),
}
}
@@ -306,11 +308,18 @@ func (a *Agent) Run() error {
fastTimer.Stop()
defer fastTimer.Stop()
// V4-beacon ticker (item ③): its own slow cadence, not the
// heartbeat's — the proven address barely changes, and a
// failure here is silent by design (no v4 route, old panel).
beaconTicker := time.NewTicker(defaultV4BeaconInterval)
defer beaconTicker.Stop()
// Initial config fetch
if err := a.fetchAndApplyConfig(); err != nil {
logger.Warningf("Initial config fetch failed: %v", err)
}
a.armFastTimer(fastTimer)
_ = a.sendV4Beacon()
for {
select {
@@ -335,6 +344,8 @@ func (a *Agent) Run() error {
case <-fastTimer.C:
a.sendFastReport()
a.armFastTimer(fastTimer)
case <-beaconTicker.C:
_ = a.sendV4Beacon()
}
}
}
+58
View File
@@ -0,0 +1,58 @@
package agent
import (
"bytes"
"context"
"crypto/tls"
"fmt"
"io"
"net"
"net/http"
"time"
)
// Beacon cadence is deliberately slow: the proven address barely changes, so the heartbeat's 60 s would be pure noise; the first beacon fires at startup.
const defaultV4BeaconInterval = 5 * time.Minute
// v4OnlyDialContext forces every dial onto tcp4: on 1:1 NAT the local v4 is private, so only a panel-observed v4 arrival proves the public address (a v6-only panel just fails the dial).
func v4OnlyDialContext(dial func(ctx context.Context, network, address string) (net.Conn, error)) func(ctx context.Context, network, address string) (net.Conn, error) {
return func(ctx context.Context, _ string, address string) (net.Conn, error) {
return dial(ctx, "tcp4", address)
}
}
// newV4BeaconClient mirrors the agent's normal client (same TLS trust, no proxy — a proxy's egress would prove nothing about this host) except its dialer is pinned to tcp4.
func newV4BeaconClient() *http.Client {
return &http.Client{
Timeout: 10 * time.Second,
Transport: &http.Transport{
TLSClientConfig: &tls.Config{InsecureSkipVerify: false},
DialContext: v4OnlyDialContext((&net.Dialer{Timeout: 10 * time.Second}).DialContext),
},
}
}
// sendV4Beacon posts one empty beacon over a v4-only connection; the panel's observed source is the whole signal, since any self-reported address is worthless behind 1:1 NAT. Any failure is the caller's to ignore — the heartbeat keeps working over v6 meanwhile.
func (a *Agent) sendV4Beacon() error {
url := fmt.Sprintf("%s/panel/api/agent/v4-beacon", a.config.MasterURL)
req, err := http.NewRequest(http.MethodPost, url, bytes.NewReader([]byte(`{}`)))
if err != nil {
return err
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+a.config.Token)
client := a.v4httpClient
if client == nil {
client = newV4BeaconClient()
}
resp, err := client.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
_, _ = io.Copy(io.Discard, resp.Body)
if resp.StatusCode != http.StatusOK {
return fmt.Errorf("v4 beacon failed with status %d", resp.StatusCode)
}
return nil
}
+101
View File
@@ -0,0 +1,101 @@
package agent
import (
"context"
"errors"
"io"
"net"
"net/http"
"net/http/httptest"
"testing"
)
// The dial wrapper must force tcp4 no matter what network the
// transport asked for — that forcing is the whole beacon.
func TestV4OnlyDialContextForcesTCP4(t *testing.T) {
var gotNetwork, gotAddr string
sentinel := errors.New("stub dialer hit")
stub := func(_ context.Context, network, address string) (net.Conn, error) {
gotNetwork, gotAddr = network, address
return nil, sentinel
}
wrapped := v4OnlyDialContext(stub)
for _, asked := range []string{"tcp", "tcp6", "tcp4"} {
_, err := wrapped(context.Background(), asked, "example.com:443")
if !errors.Is(err, sentinel) {
t.Fatalf("asked %q: wrapped dial did not call through: %v", asked, err)
}
if gotNetwork != "tcp4" {
t.Fatalf("asked %q: dial network %q, want tcp4", asked, gotNetwork)
}
if gotAddr != "example.com:443" {
t.Fatalf("asked %q: dial address %q changed", asked, gotAddr)
}
}
}
// One beacon round trip: POST to the agent route, Bearer token,
// empty JSON body — and nothing else for the panel to read.
func TestSendV4Beacon(t *testing.T) {
var gotMethod, gotPath, gotAuth string
var gotBody []byte
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
gotMethod, gotPath, gotAuth = r.Method, r.URL.Path, r.Header.Get("Authorization")
gotBody, _ = io.ReadAll(r.Body)
w.WriteHeader(http.StatusOK)
}))
defer srv.Close()
ag := New(&Config{MasterURL: srv.URL, Token: "beacon-tok", NodeName: "n"})
if err := ag.sendV4Beacon(); err != nil {
t.Fatalf("sendV4Beacon: %v", err)
}
if gotMethod != http.MethodPost {
t.Fatalf("method %q, want POST", gotMethod)
}
if gotPath != "/panel/api/agent/v4-beacon" {
t.Fatalf("path %q, want the beacon route", gotPath)
}
if gotAuth != "Bearer beacon-tok" {
t.Fatalf("auth %q, want the agent token", gotAuth)
}
if string(gotBody) != `{}` {
t.Fatalf("body %q, want empty JSON object", gotBody)
}
}
// Beacon failures are errors to the caller (who ignores them) and must never panic; a v6-literal master must fail at the tcp4 dial rather than ever reaching the panel over v6.
func TestSendV4BeaconFailures(t *testing.T) {
t.Run("panel 500", func(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusInternalServerError)
}))
defer srv.Close()
ag := New(&Config{MasterURL: srv.URL, Token: "tok", NodeName: "n"})
if err := ag.sendV4Beacon(); err == nil {
t.Fatal("500 beacon: want an error")
}
})
t.Run("old panel 404", func(t *testing.T) {
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusNotFound)
}))
defer srv.Close()
ag := New(&Config{MasterURL: srv.URL, Token: "tok", NodeName: "n"})
if err := ag.sendV4Beacon(); err == nil {
t.Fatal("404 beacon: want an error (old panel, no beacon endpoint)")
}
})
t.Run("v6 literal master", func(t *testing.T) {
ag := New(&Config{MasterURL: "http://[::1]:1", Token: "tok", NodeName: "n"})
if err := ag.sendV4Beacon(); err == nil {
t.Fatal("v6-literal master: want a dial error, never a v6 beacon")
}
})
t.Run("unreachable master", func(t *testing.T) {
ag := New(&Config{MasterURL: "http://127.0.0.1:1", Token: "tok", NodeName: "n"})
if err := ag.sendV4Beacon(); err == nil {
t.Fatal("unreachable master: want an error")
}
})
}
+40
View File
@@ -47,6 +47,9 @@ func NewAgentController(g *gin.RouterGroup) *AgentController {
func (a *AgentController) initRouter(g *gin.RouterGroup) {
g.POST("/register", a.register)
g.POST("/heartbeat", a.heartbeat)
// Item ③: the agent's periodic IPv4 beacon (v4-forced request,
// empty body — the observed source is the whole signal).
g.POST("/v4-beacon", a.v4Beacon)
g.GET("/config", a.getConfig)
g.POST("/traffic", a.reportTraffic)
// V2.0.1 probe batch: the lightweight fast report (runtime
@@ -174,6 +177,43 @@ func (a *AgentController) captureSeenIP(c *gin.Context, agent *model.AgentNode)
agent.SeenIPAt = now
}
// v4Beacon answers one agent IPv4 beacon: authenticated, then filed via captureV4Beacon. Always ok — an unproven beacon is a silent no-op, never an agent-visible error.
func (a *AgentController) v4Beacon(c *gin.Context) {
agent, ok := a.findAgent(bearerToken(c))
if !ok {
c.JSON(401, gin.H{"message": "unauthorized"})
return
}
a.captureV4Beacon(c, agent)
jsonObj(c, gin.H{"message": "ok"}, nil)
}
// captureV4Beacon files a beacon's observed source as SeenIPv4 only when it is a public IPv4 literal; private, loopback, or v6 observations prove no public side and must not displace a good seen address.
// The reported telemetry address is never written here; resolution still prefers a public reported address over any seen one (service.ResolveAgentAddresses).
func (a *AgentController) captureV4Beacon(c *gin.Context, agent *model.AgentNode) {
ip := requestSourceIP(c)
if ip == nil || agent == nil {
return
}
v4 := ip.To4()
if v4 == nil || !service.IsPublicIPLiteral(v4.String()) {
return
}
if agent.SeenIPv4 == v4.String() {
return
}
now := time.Now().UnixMilli()
if err := execAgentColumns(uint(agent.Id), []agentColumn{
{"seen_ipv4", v4.String()},
{"seen_ip_at", now},
{"updated_at", now},
}); err != nil {
return
}
agent.SeenIPv4 = v4.String()
agent.SeenIPAt = now
}
// agentColumn is one column assignment for execAgentColumns.
type agentColumn struct {
name string
@@ -0,0 +1,240 @@
package controller
import (
"bytes"
"context"
"net"
"net/http"
"net/http/httptest"
"path/filepath"
"testing"
"github.com/gin-gonic/gin"
"github.com/mhsanaei/3x-ui/v3/internal/database"
"github.com/mhsanaei/3x-ui/v3/internal/database/model"
"github.com/mhsanaei/3x-ui/v3/internal/web/service"
)
// Item ③ IPv4 beacon: the panel files a beacon's observed source as SeenIPv4, and only as SeenIPv4 — never the reported telemetry address, never a private/loopback/v6 observation.
func TestAgentV4Beacon(t *testing.T) {
gin.SetMode(gin.TestMode)
dbDir := t.TempDir()
t.Setenv("XUI_DB_FOLDER", dbDir)
if err := database.InitDB(filepath.Join(dbDir, "x-ui.db")); err != nil {
t.Fatalf("InitDB: %v", err)
}
t.Cleanup(func() { _ = database.CloseDB() })
db := database.GetDB()
node := &model.Node{Name: "beacon-node", Address: ""}
if err := db.Create(node).Error; err != nil {
t.Fatalf("seed node: %v", err)
}
const token = "v4-beacon-test-token"
agent := &model.AgentNode{
TokenHash: hashAgentToken(token),
TokenPrefix: "v4-beacon-",
Name: "beacon-node",
NodeID: &node.Id,
}
if err := db.Create(agent).Error; err != nil {
t.Fatalf("seed agent: %v", err)
}
engine := gin.New()
NewAgentController(engine.Group("/panel/api/agent"))
beacon := func(headers map[string]string, remoteAddr, tok string) int {
req := httptest.NewRequest(http.MethodPost, "/panel/api/agent/v4-beacon", bytes.NewBufferString(`{}`))
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+tok)
for k, v := range headers {
req.Header.Set(k, v)
}
if remoteAddr != "" {
req.RemoteAddr = remoteAddr
}
rec := httptest.NewRecorder()
engine.ServeHTTP(rec, req)
return rec.Code
}
load := func() model.AgentNode {
var row model.AgentNode
if err := db.First(&row, agent.Id).Error; err != nil {
t.Fatalf("reload agent: %v", err)
}
return row
}
reset := func(ipv4, seenIPv4, seenIPv6 string) {
if err := db.Model(&model.AgentNode{}).Where("id = ?", agent.Id).
Updates(map[string]any{"ipv4": ipv4, "seen_ipv4": seenIPv4, "seen_ipv6": seenIPv6}).Error; err != nil {
t.Fatalf("reset agent row: %v", err)
}
}
// A public v4 observation lands in seen_ipv4 only.
reset("", "", "")
if code := beacon(map[string]string{"CF-Connecting-IP": "203.0.113.5"}, "", token); code != http.StatusOK {
t.Fatalf("beacon public v4: status %d", code)
}
if row := load(); row.SeenIPv4 != "203.0.113.5" || row.SeenIPAt == 0 {
t.Fatalf("public v4 beacon not stored: %+v", row)
}
// Change-only: the same beacon again must not rewrite (SeenIPAt frozen).
frozen := load().SeenIPAt
if code := beacon(map[string]string{"CF-Connecting-IP": "203.0.113.5"}, "", token); code != http.StatusOK {
t.Fatalf("repeat beacon: status %d", code)
}
if row := load(); row.SeenIPAt != frozen {
t.Fatalf("unchanged beacon rewrote the row: seen_ip_at %d -> %d", frozen, row.SeenIPAt)
}
// A public v4 via the transport RemoteAddr (direct connection, no CF).
reset("", "", "")
if code := beacon(nil, "198.51.100.23:5555", token); code != http.StatusOK {
t.Fatalf("beacon RemoteAddr: status %d", code)
}
if row := load(); row.SeenIPv4 != "198.51.100.23" {
t.Fatalf("RemoteAddr beacon not stored: %+v", row)
}
// Reported public v4 is never overwritten: the beacon moves only
// the seen column, and resolution still prefers the reported one.
reset("203.0.113.50", "203.0.113.5", "")
if code := beacon(map[string]string{"CF-Connecting-IP": "203.0.113.99"}, "", token); code != http.StatusOK {
t.Fatalf("beacon over reported public: status %d", code)
}
row := load()
if row.IPv4 != "203.0.113.50" {
t.Fatalf("beacon overwrote reported ipv4: %+v", row)
}
if row.SeenIPv4 != "203.0.113.99" {
t.Fatalf("beacon did not update seen side: %+v", row)
}
if v4, _ := service.ResolveAgentAddresses(&row); v4 != "203.0.113.50" {
t.Fatalf("resolved address followed the beacon, not the reported public v4: %q", v4)
}
// The Oracle-JP case: reported private, beacon public — the
// resolved address becomes the beacon's public one.
reset("10.0.0.206", "", "")
if code := beacon(map[string]string{"CF-Connecting-IP": "203.0.113.77"}, "", token); code != http.StatusOK {
t.Fatalf("beacon over reported private: status %d", code)
}
row = load()
if v4, _ := service.ResolveAgentAddresses(&row); v4 != "203.0.113.77" {
t.Fatalf("private reported + public beacon did not resolve to beacon: %q (row %+v)", v4, row)
}
// A private observation never becomes seen_ipv4 — neither from
// empty nor over an existing good seen address.
reset("", "", "")
if code := beacon(map[string]string{"CF-Connecting-IP": "10.0.0.206"}, "", token); code != http.StatusOK {
t.Fatalf("private beacon: status %d", code)
}
if row := load(); row.SeenIPv4 != "" {
t.Fatalf("private beacon stored as seen_ipv4: %+v", row)
}
reset("", "203.0.113.5", "")
if code := beacon(nil, "10.0.0.206:5555", token); code != http.StatusOK {
t.Fatalf("private RemoteAddr beacon: status %d", code)
}
if row := load(); row.SeenIPv4 != "203.0.113.5" {
t.Fatalf("private beacon displaced good seen_ipv4: %+v", row)
}
// A v6 observation touches neither seen column: the beacon is v4-only.
reset("", "203.0.113.5", "2001:db8::1")
if code := beacon(map[string]string{"CF-Connecting-IP": "2001:db8::2"}, "", token); code != http.StatusOK {
t.Fatalf("v6 beacon: status %d", code)
}
if row := load(); row.SeenIPv4 != "203.0.113.5" || row.SeenIPv6 != "2001:db8::1" {
t.Fatalf("v6 beacon touched a seen column: %+v", row)
}
// Loopback (a same-host panel) proves no public address either.
reset("", "", "")
if code := beacon(nil, "127.0.0.1:5555", token); code != http.StatusOK {
t.Fatalf("loopback beacon: status %d", code)
}
if row := load(); row.SeenIPv4 != "" {
t.Fatalf("loopback beacon stored as seen_ipv4: %+v", row)
}
// Unauthenticated beacons capture nothing.
reset("", "", "")
if code := beacon(map[string]string{"CF-Connecting-IP": "203.0.113.5"}, "", "wrong-token"); code != http.StatusUnauthorized {
t.Fatalf("bad token beacon: status %d, want 401", code)
}
if row := load(); row.SeenIPv4 != "" {
t.Fatalf("unauthenticated beacon captured a source: %+v", row)
}
}
// Socket-level round trip: a real v4 connection to the panel, with the test harness playing the Cloudflare edge (CF-Connecting-IP) — the agent itself never sends that header. Without it the loopback peer proves nothing.
func TestAgentV4BeaconOverSocket(t *testing.T) {
gin.SetMode(gin.TestMode)
dbDir := t.TempDir()
t.Setenv("XUI_DB_FOLDER", dbDir)
if err := database.InitDB(filepath.Join(dbDir, "x-ui.db")); err != nil {
t.Fatalf("InitDB: %v", err)
}
t.Cleanup(func() { _ = database.CloseDB() })
db := database.GetDB()
const token = "v4-beacon-socket-token"
agent := &model.AgentNode{TokenHash: hashAgentToken(token), TokenPrefix: "v4-sock-"}
if err := db.Create(agent).Error; err != nil {
t.Fatalf("seed agent: %v", err)
}
engine := gin.New()
NewAgentController(engine.Group("/panel/api/agent"))
srv := httptest.NewServer(engine)
defer srv.Close()
// Same shape as the agent's beacon client: tcp4-forced dial.
v4client := &http.Client{Transport: &http.Transport{
DialContext: func(ctx context.Context, _, addr string) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, "tcp4", addr)
},
}}
post := func(cfHeader string) int {
req, err := http.NewRequest(http.MethodPost, srv.URL+"/panel/api/agent/v4-beacon", bytes.NewBufferString(`{}`))
if err != nil {
t.Fatalf("build beacon request: %v", err)
}
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Authorization", "Bearer "+token)
if cfHeader != "" {
req.Header.Set("CF-Connecting-IP", cfHeader)
}
resp, err := v4client.Do(req)
if err != nil {
t.Fatalf("beacon over socket: %v", err)
}
defer resp.Body.Close()
return resp.StatusCode
}
load := func() model.AgentNode {
var row model.AgentNode
if err := db.First(&row, agent.Id).Error; err != nil {
t.Fatalf("reload agent: %v", err)
}
return row
}
if code := post("203.0.113.88"); code != http.StatusOK {
t.Fatalf("socket beacon with edge header: status %d", code)
}
if row := load(); row.SeenIPv4 != "203.0.113.88" {
t.Fatalf("socket beacon did not store seen_ipv4: %+v", row)
}
if code := post(""); code != http.StatusOK {
t.Fatalf("socket beacon without header: status %d", code)
}
if row := load(); row.SeenIPv4 != "203.0.113.88" {
t.Fatalf("loopback socket beacon displaced seen_ipv4: %+v", row)
}
}