diff --git a/docs/architecture.md b/docs/architecture.md index 39ecafe02..c0c21dccc 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -397,7 +397,7 @@ All registered in `web.go` → `startTask()`. Each is a struct with a `Run()` me | `@every 5s` | `xray_traffic_job` | Pull traffic stats from Xray (5s start delay) | | `@every 5s` | `node_heartbeat_job` | Probe child nodes (online/offline) | | `@every 5s` | `node_traffic_sync_job` | Pull + merge node traffic; push reconciliation | -| `@every 10s` | `check_client_ip_job` | Enforce per-client IP limits | +| `@every 10s` | `check_client_ip_job` | Record client IPs (30-min window) and enforce per-client IP limits | | `@every 10s` | `mtproto_job` | Reconcile `mtg` sidecars against enabled MTProto inbounds | | `@every 10s` | `amneziawg_job` | Reconcile embedded AmneziaWG interfaces against enabled local inbounds | | `@every 5m` | `outbound_subscription_job` | Refresh outbound provider configs | diff --git a/internal/web/job/check_client_ip_job.go b/internal/web/job/check_client_ip_job.go index 1e1a3720d..89d488364 100644 --- a/internal/web/job/check_client_ip_job.go +++ b/internal/web/job/check_client_ip_job.go @@ -28,19 +28,23 @@ type IPWithTimestamp struct { Timestamp int64 `json:"timestamp"` } +// onlineUsersSource is the one service.XrayService method the scan needs, so +// tests can drive Run end to end without a live xray core. +type onlineUsersSource interface { + GetOnlineUsers() ([]xray.OnlineUser, bool, error) +} + // CheckClientIpJob monitors client IP addresses and manages IP blocking based // on configured limits. The per-client IPs come from the core's online-stats // API; no access log is involved. On a core too old to expose that API the job // simply skips the run (the bundled core always supports it). type CheckClientIpJob struct { bannedSeen map[string]int64 - xrayService service.XrayService + onlineUsers onlineUsersSource allowlist ipLimitAllowlist lastIpPrune int64 } -var job *CheckClientIpJob - const defaultXrayAPIPort = 62789 const ipStaleAfterSeconds = int64(30 * 60) @@ -50,8 +54,7 @@ const ipPruneIntervalSeconds = int64(5 * 60) // NewCheckClientIpJob creates a new client IP monitoring job instance. func NewCheckClientIpJob() *CheckClientIpJob { - job = new(CheckClientIpJob) - return job + return &CheckClientIpJob{onlineUsers: &service.XrayService{}} } func (j *CheckClientIpJob) Run() { @@ -64,30 +67,31 @@ func (j *CheckClientIpJob) Run() { return } - if !isFail2BanEnabled() { - return - } - + // XUI_ENABLE_FAIL2BAN=false disables enforcement, not tracking (#6722): + // the scan still records IPs so the panel keeps showing them. hasLimit := j.hasLimitIp() f2bInstalled := false if hasLimit { f2bInstalled = j.checkFail2BanInstalled() } + // Windows enforces through disconnects and never needs the fail2ban binary. + needsBinary := runtime.GOOS != "windows" + enforce := j.resolveEnforce(isFail2BanEnabled(), hasLimit, f2bInstalled, needsBinary) // Read only when the limit is actually applied: this runs every 10s and // most panels carry no IP limit at all. - enforce := j.resolveEnforce(hasLimit, f2bInstalled) if enforce { j.allowlist = j.loadAllowlist() } j.processObserved(observed, enforce, true) } -// resolveEnforce decides whether limits can actually be enforced this run. -// Without fail2ban on a platform that needs it the limit can't be applied, so -// enforcement is skipped (the panel resets these limits to 0 on upgrade and -// disables the field, so this is normally a no-op). -func (j *CheckClientIpJob) resolveEnforce(hasLimit, f2bInstalled bool) bool { - if hasLimit && runtime.GOOS != "windows" && !f2bInstalled { +// resolveEnforce decides whether limits can actually be enforced this run: a +// disabled fail2ban gates every platform, a missing binary the ones needing it. +func (j *CheckClientIpJob) resolveEnforce(f2bEnabled, hasLimit, f2bInstalled, needsBinary bool) bool { + if !f2bEnabled { + return false + } + if hasLimit && needsBinary && !f2bInstalled { return false } return hasLimit @@ -98,7 +102,7 @@ func (j *CheckClientIpJob) resolveEnforce(hasLimit, f2bInstalled bool) bool { // API is unavailable — xray not running, an older core, or a transient gRPC // failure — and the caller skips the run (there is no access-log fallback). func (j *CheckClientIpJob) collectFromOnlineAPI() (map[string]map[string]int64, bool) { - onlineUsers, ok, err := j.xrayService.GetOnlineUsers() + onlineUsers, ok, err := j.onlineUsers.GetOnlineUsers() if err != nil { logger.Debug("[LimitIP] online-stats API unavailable this run:", err) return nil, false @@ -454,8 +458,18 @@ func partitionLiveIps(ipMap map[string]int64, observedThisScan map[string]bool) historical = append(historical, entry) } } - sort.Slice(live, func(i, j int) bool { return live[i].Timestamp < live[j].Timestamp }) - sort.Slice(historical, func(i, j int) bool { return historical[i].Timestamp < historical[j].Timestamp }) + sort.Slice(live, func(i, j int) bool { + if live[i].Timestamp != live[j].Timestamp { + return live[i].Timestamp < live[j].Timestamp + } + return live[i].IP < live[j].IP + }) + sort.Slice(historical, func(i, j int) bool { + if historical[i].Timestamp != historical[j].Timestamp { + return historical[i].Timestamp < historical[j].Timestamp + } + return historical[i].IP < historical[j].IP + }) return live, historical } @@ -499,16 +513,7 @@ func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps return nil, 0 } - if !enforce || limitIp <= 0 || !inbound.Enable { - // Nothing to enforce (collection-only run, no limit on the clients row, - // or inbound disabled): record the observed IPs for the panel and return. - jsonIps, _ := json.Marshal(newIpsWithTime) - inboundClientIps.Ips = string(jsonIps) - if err := tx.Save(inboundClientIps).Error; err != nil { - logger.Error("failed to save inboundClientIps:", err) - } - return nil, 0 - } + enforceLimit := enforce && limitIp > 0 && inbound.Enable // only ips seen in this scan count toward the limit. see // partitionLiveIps. @@ -526,24 +531,33 @@ func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps _ = json.Unmarshal([]byte(inboundClientIps.Ips), &oldIpsWithTime) } ipMap := mergeClientIps(oldIpsWithTime, newIpsWithTime, staleCutoff, observedAreLive) - liveIps, historicalIps := partitionLiveIps(ipMap, observedThisScan) - // historical db-only ips are excluded from this count on purpose. - limitedIps, allowedIps := j.allowlist.split(liveIps) - keptLive, bannedLive := selectIpsToBan(limitedIps, limitIp) - // Allowlisted addresses stay connected and out of the count: charging them - // against the limit would still cut the shared network the entry protects. - keptLive = append(keptLive, allowedIps...) - - // keep kept-live + historical in the blob so the panel keeps showing recently - // seen ips; banned live ips reappear in the next scan if they reconnect. - dbIps := make([]IPWithTimestamp, 0, len(keptLive)+len(historicalIps)) - dbIps = append(dbIps, keptLive...) - dbIps = append(dbIps, historicalIps...) + dbIps := make([]IPWithTimestamp, 0, len(ipMap)) + var bannedLive []IPWithTimestamp + keptLive := 0 + if enforceLimit { + liveIps, historicalIps := partitionLiveIps(ipMap, observedThisScan) + // historical db-only ips are excluded from this count on purpose. + limitedIps, allowedIps := j.allowlist.split(liveIps) + kept, banned := selectIpsToBan(limitedIps, limitIp) + // Allowlisted addresses stay connected and out of the count: charging them + // against the limit would still cut the shared network the entry protects. + kept = append(kept, allowedIps...) + bannedLive = banned + keptLive = len(kept) + // keep kept-live + historical in the blob so the panel keeps showing recently + // seen ips; banned live ips reappear in the next scan if they reconnect. + dbIps = append(dbIps, kept...) + dbIps = append(dbIps, historicalIps...) + } else { + // Collection-only run (#6722): no ban selection; keep every IP in + // the 30-minute window, sorted so unchanged scans skip the write. + dbIps = sortedIpsFromMap(ipMap) + } jsonIps, _ := json.Marshal(dbIps) newIps := string(jsonIps) if newIps == inboundClientIps.Ips { - return bannedLive, len(keptLive) + return bannedLive, keptLive } updated, err := service.CasUpdateInboundClientIps(tx, inboundClientIps.Id, inboundClientIps.Ips, newIps) @@ -553,7 +567,7 @@ func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps } if updated { inboundClientIps.Ips = newIps - return bannedLive, len(keptLive) + return bannedLive, keptLive } if err := tx.Where("id = ?", inboundClientIps.Id).First(inboundClientIps).Error; err != nil { logger.Error("failed to re-read inboundClientIps after a concurrent write:", err) @@ -564,6 +578,22 @@ func (j *CheckClientIpJob) updateInboundClientIps(tx *gorm.DB, inboundClientIps return nil, 0 } +// sortedIpsFromMap orders the merged IP set by timestamp with the IP as +// tie-breaker, giving a stable blob for the skip-when-unchanged compare. +func sortedIpsFromMap(ipMap map[string]int64) []IPWithTimestamp { + out := make([]IPWithTimestamp, 0, len(ipMap)) + for ip, ts := range ipMap { + out = append(out, IPWithTimestamp{IP: ip, Timestamp: ts}) + } + sort.Slice(out, func(i, j int) bool { + if out[i].Timestamp != out[j].Timestamp { + return out[i].Timestamp < out[j].Timestamp + } + return out[i].IP < out[j].IP + }) + return out +} + // pendingBan carries one client's enforcement outcome from inside the scan's // transaction to the publication that may only follow a successful commit. type pendingBan struct { @@ -835,8 +865,8 @@ 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. +// Runs before the apiMode gate: 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 { diff --git a/internal/web/job/check_client_ip_job_integration_test.go b/internal/web/job/check_client_ip_job_integration_test.go index da3997c5d..6b5c66934 100644 --- a/internal/web/job/check_client_ip_job_integration_test.go +++ b/internal/web/job/check_client_ip_job_integration_test.go @@ -15,6 +15,7 @@ import ( "github.com/mhsanaei/3x-ui/v3/internal/database/dbtest" "github.com/mhsanaei/3x-ui/v3/internal/database/model" xuilogger "github.com/mhsanaei/3x-ui/v3/internal/logger" + "github.com/mhsanaei/3x-ui/v3/internal/xray" ) // 3x-ui logger must be initialised once before any code path that can @@ -320,6 +321,67 @@ func TestProcessObserved_CollectsIpsWithoutLimit(t *testing.T) { } } +// #6722: collection-only runs keep the enforcing run's 30-minute window; the +// old branch overwrote the blob, dropping IPs seen minutes earlier. +func TestProcessObserved_CollectionOnlyKeepsRecentWindow(t *testing.T) { + setupIntegrationDB(t) + + const email = "window-user" + seedInboundWithClient(t, "inbound-window", email, 0) + fiveMinAgo := time.Now().Unix() - 5*60 + seedClientIps(t, email, []IPWithTimestamp{{IP: "198.51.100.1", Timestamp: fiveMinAgo}}) + + observed := map[string]map[string]int64{ + email: {"203.0.113.30": time.Now().Unix()}, + } + NewCheckClientIpJob().processObserved(observed, false, true) + + got := ipSet(readClientIps(t, email)) + if _, ok := got["198.51.100.1"]; !ok { + t.Fatalf("collection-only run dropped the recent IP 198.51.100.1, got %v", got) + } + if _, ok := got["203.0.113.30"]; !ok { + t.Fatalf("collection-only run missed the observed IP 203.0.113.30, got %v", got) + } +} + +// staticOnlineUsers pins the job's online-stats source so Run can be driven +// end to end without a live xray core. +type staticOnlineUsers []xray.OnlineUser + +func (s staticOnlineUsers) GetOnlineUsers() ([]xray.OnlineUser, bool, error) { + return s, true, nil +} + +// #6722: fail2ban disabled — an over-limit client is recorded but never banned; +// the setup can ban (limit 1, two IPs, fake fail2ban), so the flag stops it. +func TestRun_RecordsIpsWhenFail2banDisabled(t *testing.T) { + setupIntegrationDB(t) + t.Setenv("XUI_ENABLE_FAIL2BAN", "false") + fakeFail2BanClient(t) + + const email = "track-only-user" + seedLinkedInboundWithClient(t, "inbound-track-only", email, 1) + + j := NewCheckClientIpJob() + j.onlineUsers = staticOnlineUsers{{Email: email, IPs: []xray.OnlineIP{ + {IP: "203.0.113.20"}, {IP: "203.0.113.21"}, + }}} + // Bans need a stored row (the first scan only records), so run twice: the + // second scan is where enforcement would fire if the flag allowed it. + j.Run() + j.Run() + + got := ipSet(readClientIps(t, email)) + if len(got) != 2 { + t.Fatalf("expected both observed IPs recorded with fail2ban disabled, got %v", got) + } + if info, err := os.Stat(readIpLimitLogPath()); err == nil && info.Size() > 0 { + body, _ := os.ReadFile(readIpLimitLogPath()) + t.Fatalf("3xipl.log must stay empty with fail2ban disabled, got:\n%s", body) + } +} + // #4963: an observed IP for a renamed/deleted client (its email no longer maps // to any inbound) must not create or resurrect an inbound_client_ips row, and // must drop any orphan left behind — instead of erroring every run. diff --git a/internal/web/job/check_client_ip_job_test.go b/internal/web/job/check_client_ip_job_test.go index 211d68cd9..22d50e0af 100644 --- a/internal/web/job/check_client_ip_job_test.go +++ b/internal/web/job/check_client_ip_job_test.go @@ -289,6 +289,30 @@ func TestCheckFail2BanInstalled_EmptyEnvSkipsClientProbe(t *testing.T) { } } +// Map-iteration order must not reach the blob: an unchanged client would CAS-rewrite +// every 10s, and the ban pick among tied IPs would be random per scan. +func TestPartitionLiveIps_TiedTimestampsKeepStableIpOrder(t *testing.T) { + ipMap := map[string]int64{"203.0.113.3": 100, "203.0.113.1": 100, "203.0.113.2": 100} + observed := map[string]bool{"203.0.113.1": true, "203.0.113.2": true, "203.0.113.3": true} + var first []IPWithTimestamp + for i := range 50 { + live, _ := partitionLiveIps(ipMap, observed) + if i == 0 { + first = live + continue + } + if !reflect.DeepEqual(first, live) { + t.Fatalf("tied timestamps produced different orders across runs: %v vs %v", first, live) + } + } + want := []string{"203.0.113.1", "203.0.113.2", "203.0.113.3"} + for i, ip := range want { + if first[i].IP != ip { + t.Fatalf("tied timestamps must order by IP, got %v", first) + } + } +} + func TestIsFail2BanEnabled_DefaultsToEnabledWhenUnset(t *testing.T) { value, ok := os.LookupEnv("XUI_ENABLE_FAIL2BAN") os.Unsetenv("XUI_ENABLE_FAIL2BAN") @@ -336,3 +360,31 @@ func fakeFail2BanClient(t *testing.T) string { t.Setenv("PATH", dir+string(os.PathListSeparator)+os.Getenv("PATH")) return marker } + +// resolveEnforce is platform-pure (the caller supplies needsBinary), so this +// table pins every branch — the Windows carve-out included — statically. +func TestResolveEnforce(t *testing.T) { + cases := []struct { + name string + f2bEnabled bool + hasLimit bool + f2bInstalled bool + needsBinary bool + want bool + }{ + {"disabled env with fail2ban present never enforces", false, true, true, true, false}, + {"disabled env without fail2ban never enforces", false, true, false, true, false}, + {"enabled with fail2ban and a limit enforces", true, true, true, true, true}, + {"enabled, binary needed but absent, cannot enforce", true, true, false, true, false}, + {"enabled, no binary needed (windows), enforces without fail2ban", true, true, false, false, true}, + {"no limit never enforces even fully installed", true, false, true, true, false}, + } + j := NewCheckClientIpJob() + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := j.resolveEnforce(tc.f2bEnabled, tc.hasLimit, tc.f2bInstalled, tc.needsBinary); got != tc.want { + t.Fatalf("resolveEnforce(%v, %v, %v, %v) = %v, want %v", tc.f2bEnabled, tc.hasLimit, tc.f2bInstalled, tc.needsBinary, got, tc.want) + } + }) + } +}