pjandro/3x-ui
1
0
Fork 0
mirror of https://github.com/MHSanaei/3x-ui.git synced 2026-10-24 14:55:02 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(iplimit): keep recording client IPs with fail2ban disabled (#6830)

Browse Source 
With XUI_ENABLE_FAIL2BAN off, the IP job returned early, so the panel stopped
recording and showing client IPs altogether, although only enforcement needs
fail2ban. Only enforcement is now gated on it: collection-only runs go through
the merge, 30-minute window and compare-and-swap path, so the IP list stays
current. Tied timestamps sort by IP, so an unchanged client no longer rewrites
its row every scan and the ban pick among tied IPs is stable. Closes #6722.
This commit is contained in:
clearlove13
2026-10-24 12:09:01 +08:00
committed by GitHub
parent 0729fc6dbb
commit 49c0600f35
4 changed files with 191 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/architecture.md
+1 -1
View File
@@ -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 |
internal/web/job/check_client_ip_job.go
+76 -46
View File
@@ -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 {
internal/web/job/check_client_ip_job_integration_test.go
+62
View File
@@ -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.
internal/web/job/check_client_ip_job_test.go
+52
View File
@@ -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)
}
})
}
}