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 +}