From 103b0dfe8d6ed710d4f3dad56c6c441b71d434d1 Mon Sep 17 00:00:00 2001 From: Sanaei Date: Mon, 24 Aug 2026 13:27:40 +0200 Subject: [PATCH] fix(job): expire stored client IPs of offline clients ipStaleAfterSeconds was only applied while a row was being rewritten, and rows are only rewritten for clients present in the current online scan. A client that stopped connecting therefore kept its last addresses forever in inbound_client_ips, and node_client_ips rows (including those of deleted clients) were never revisited at all. Sweep both tables every five minutes, dropping entries past the cutoff and deleting rows that end up empty. The sweep runs ahead of the fail2ban and api-mode gates so retention holds even on panels that collect nothing. Closes #6286 --- internal/web/job/check_client_ip_job.go | 18 +++++ internal/web/service/client_ips_prune_test.go | 71 +++++++++++++++++++ internal/web/service/inbound_client_ips.go | 36 ++++++++++ internal/web/service/inbound_node_ips.go | 33 +++++++++ 4 files changed, 158 insertions(+) create mode 100644 internal/web/service/client_ips_prune_test.go diff --git a/internal/web/job/check_client_ip_job.go b/internal/web/job/check_client_ip_job.go index 93a8cef12..146db0b70 100644 --- a/internal/web/job/check_client_ip_job.go +++ b/internal/web/job/check_client_ip_job.go @@ -36,6 +36,7 @@ type CheckClientIpJob struct { bannedSeen map[string]int64 xrayService service.XrayService allowlist ipLimitAllowlist + lastIpPrune int64 } var job *CheckClientIpJob @@ -44,6 +45,9 @@ const defaultXrayAPIPort = 62789 const ipStaleAfterSeconds = int64(30 * 60) +// pruneStaleIpRows cadence; the scan itself cannot prune offline clients' rows. +const ipPruneIntervalSeconds = int64(5 * 60) + // NewCheckClientIpJob creates a new client IP monitoring job instance. func NewCheckClientIpJob() *CheckClientIpJob { job = new(CheckClientIpJob) @@ -51,6 +55,7 @@ func NewCheckClientIpJob() *CheckClientIpJob { } func (j *CheckClientIpJob) Run() { + j.pruneStaleIpRows() observed, apiMode := j.collectFromOnlineAPI() if !apiMode { // xray is down or predates the online-stats API. There is no access-log @@ -769,3 +774,16 @@ func (j *CheckClientIpJob) getInboundByEmail(clientEmail string) (*model.Inbound return nil, err } + +// Runs before the fail2ban/apiMode gates: retention must hold for stored rows +// even while nothing is being collected. +func (j *CheckClientIpJob) pruneStaleIpRows() { + now := time.Now().Unix() + if now-j.lastIpPrune < ipPruneIntervalSeconds { + return + } + j.lastIpPrune = now + if err := (&service.InboundService{}).PruneStaleClientIps(); err != nil { + logger.Warning("prune stale client ip rows failed:", err) + } +} diff --git a/internal/web/service/client_ips_prune_test.go b/internal/web/service/client_ips_prune_test.go new file mode 100644 index 000000000..d6a2c33e1 --- /dev/null +++ b/internal/web/service/client_ips_prune_test.go @@ -0,0 +1,71 @@ +package service + +import ( + "encoding/json" + "testing" + "time" + + "github.com/mhsanaei/3x-ui/v3/internal/database" + "github.com/mhsanaei/3x-ui/v3/internal/database/model" +) + +// Rows for clients absent from the online scan are never rewritten, so the +// sweep is the only thing standing between them and indefinite retention. +func TestPruneStaleClientIpsExpiresUnobservedRows(t *testing.T) { + setupClientIpTestDB(t) + db := database.GetDB() + + now := time.Now().Unix() + stale := now - clientIpStaleAfterSeconds - 300 + fresh := now - 60 + + mkNode := func(entries ...model.ClientIpEntry) string { + t.Helper() + b, err := json.Marshal(entries) + if err != nil { + t.Fatalf("marshal node ips: %v", err) + } + return string(b) + } + + seed := []any{ + &model.InboundClientIps{ClientEmail: "offline", Ips: marshalIps(t, clientIpEntry{IP: "198.51.100.7", Timestamp: stale})}, + &model.InboundClientIps{ClientEmail: "mixed", Ips: marshalIps(t, + clientIpEntry{IP: "198.51.100.8", Timestamp: stale}, + clientIpEntry{IP: "203.0.113.9", Timestamp: fresh})}, + &model.NodeClientIp{NodeGuid: "g1", Email: "node-offline", Ips: mkNode(model.ClientIpEntry{IP: "198.51.100.9", Timestamp: stale})}, + &model.NodeClientIp{NodeGuid: "g1", Email: "node-fresh", Ips: mkNode(model.ClientIpEntry{IP: "203.0.113.10", Timestamp: fresh})}, + } + for _, row := range seed { + if err := db.Create(row).Error; err != nil { + t.Fatalf("seed %T: %v", row, err) + } + } + + if err := (&InboundService{}).PruneStaleClientIps(); err != nil { + t.Fatalf("PruneStaleClientIps: %v", err) + } + + if _, exists := readClientIps(t, "offline"); exists { + t.Fatal("fully stale inbound_client_ips row must be deleted") + } + got, exists := readClientIps(t, "mixed") + if !exists { + t.Fatal("row with a fresh entry must survive") + } + if len(got) != 1 || got["203.0.113.9"] != fresh { + t.Fatalf("mixed row = %v, want only 203.0.113.9@%d", got, fresh) + } + + var nodeRows []model.NodeClientIp + if err := db.Where("node_guid = ?", "g1").Find(&nodeRows).Error; err != nil { + t.Fatalf("read node rows: %v", err) + } + if len(nodeRows) != 1 || nodeRows[0].Email != "node-fresh" { + t.Fatalf("node rows after prune = %+v, want only node-fresh", nodeRows) + } + var kept []model.ClientIpEntry + if err := json.Unmarshal([]byte(nodeRows[0].Ips), &kept); err != nil || len(kept) != 1 || kept[0].IP != "203.0.113.10" { + t.Fatalf("node-fresh ips = %q (err %v), want 203.0.113.10 kept", nodeRows[0].Ips, err) + } +} diff --git a/internal/web/service/inbound_client_ips.go b/internal/web/service/inbound_client_ips.go index dabd049ac..80c0d75bd 100644 --- a/internal/web/service/inbound_client_ips.go +++ b/internal/web/service/inbound_client_ips.go @@ -236,3 +236,39 @@ func (s *InboundService) ClearClientIps(clientEmail string) error { } return nil } + +// PruneStaleClientIps enforces clientIpStaleAfterSeconds for rows the online +// scan no longer rewrites: an offline client's addresses must still expire. +func (s *InboundService) PruneStaleClientIps() error { + db := database.GetDB() + cutoff := time.Now().Unix() - clientIpStaleAfterSeconds + + var rows []model.InboundClientIps + if err := db.Find(&rows).Error; err != nil { + return err + } + for _, row := range rows { + var entries []clientIpEntry + if row.Ips != "" { + // Legacy blobs without timestamps stay untouched; the next scan rewrites them. + if err := json.Unmarshal([]byte(row.Ips), &entries); err != nil { + continue + } + } + kept := mergeClientIpEntries(nil, entries, cutoff) + if len(kept) == 0 { + if err := db.Delete(&model.InboundClientIps{}, row.Id).Error; err != nil { + return err + } + continue + } + if len(kept) == len(entries) { + continue + } + b, _ := json.Marshal(kept) + if err := db.Model(&model.InboundClientIps{}).Where("id = ?", row.Id).Update("ips", string(b)).Error; err != nil { + return err + } + } + return pruneStaleNodeClientIps(cutoff) +} diff --git a/internal/web/service/inbound_node_ips.go b/internal/web/service/inbound_node_ips.go index 2064a14e8..2384046c1 100644 --- a/internal/web/service/inbound_node_ips.go +++ b/internal/web/service/inbound_node_ips.go @@ -299,3 +299,36 @@ func (s *InboundService) DeleteNodeClientIpsByGuid(guid string) error { db := database.GetDB() return db.Where("node_guid = ?", guid).Delete(&model.NodeClientIp{}).Error } + +// pruneStaleNodeClientIps sweeps every attribution row: upsertNodeClientIps +// only revisits emails present in a scan, so unobserved rows never expire there. +func pruneStaleNodeClientIps(cutoff int64) error { + db := database.GetDB() + var rows []model.NodeClientIp + if err := db.Find(&rows).Error; err != nil { + return err + } + for _, row := range rows { + var entries []model.ClientIpEntry + if row.Ips != "" { + if err := json.Unmarshal([]byte(row.Ips), &entries); err != nil { + continue + } + } + kept := mergeModelClientIpEntries(nil, entries, cutoff) + if len(kept) == 0 { + if err := db.Delete(&model.NodeClientIp{}, row.Id).Error; err != nil { + return err + } + continue + } + if len(kept) == len(entries) { + continue + } + b, _ := json.Marshal(kept) + if err := db.Model(&model.NodeClientIp{}).Where("id = ?", row.Id).Update("ips", string(b)).Error; err != nil { + return err + } + } + return nil +}