Preserve clients' IP history

This PR preserves historical client IP records in `inbound_client_ips` while keeping IP Limit enforcement based only on IPs observed in the current access-log scan.

The previous stale IP eviction was introduced to fix the continuous Fail2Ban loop described in https://github.com/MHSanaei/3x-ui/issues/4077. A related regression was later tracked in https://github.com/MHSanaei/3x-ui/issues/4091, where db-only IPs could still affect ban decisions before `partitionLiveIps()` separated live IPs from historical ones.

This PR keeps that protection intact:

- `partitionLiveIps()` still ensures that only IPs observed in the current scan are counted toward `LimitIP`
- historical db-only IPs remain excluded from Fail2Ban decisions
- all live IPs, including IPs queued for Fail2Ban, are preserved in the database for visibility/history

The only behavior changed here is retention: client IP history is no longer removed from `inbound_client_ips` after 30 minutes.
This commit is contained in:
Vadim Reznichenko 2026-05-12 20:54:36 +03:00 committed by GitHub
parent 07bc74a521
commit 89f904b976
No known key found for this signature in database
GPG key ID: B5690EEEBB952194

View file

@ -35,25 +35,6 @@ var job *CheckClientIpJob
const defaultXrayAPIPort = 62789 const defaultXrayAPIPort = 62789
// ipStaleAfterSeconds controls how long a client IP kept in the
// per-client tracking table (model.InboundClientIps.Ips) is considered
// still "active" before it's evicted during the next scan.
//
// Without this eviction, an IP that connected once and then went away
// keeps sitting in the table with its old timestamp. Because the
// excess-IP selector sorts ascending ("oldest wins, newest loses") to
// protect the original/current connections, that stale entry keeps
// occupying a slot and the IP that is *actually* currently using the
// config gets classified as "new excess" and banned by fail2ban on
// every single run — producing the continuous ban loop from #4077.
//
// 30 minutes is chosen so an actively-streaming client (where xray
// emits a fresh `accepted` log line whenever it opens a new TCP) will
// always refresh its timestamp well within the window, but a client
// that has really stopped using the config will drop out of the table
// in a bounded time and free its slot.
const ipStaleAfterSeconds = int64(30 * 60)
// NewCheckClientIpJob creates a new client IP monitoring job instance. // NewCheckClientIpJob creates a new client IP monitoring job instance.
func NewCheckClientIpJob() *CheckClientIpJob { func NewCheckClientIpJob() *CheckClientIpJob {
job = new(CheckClientIpJob) job = new(CheckClientIpJob)
@ -130,8 +111,8 @@ func (j *CheckClientIpJob) hasLimitIp() bool {
settings := map[string][]model.Client{} settings := map[string][]model.Client{}
json.Unmarshal([]byte(inbound.Settings), &settings) json.Unmarshal([]byte(inbound.Settings), &settings)
clients := settings["clients"]
clients := settings["clients"]
for _, client := range clients { for _, client := range clients {
limitIp := client.LimitIP limitIp := client.LimitIP
if limitIp > 0 { if limitIp > 0 {
@ -144,7 +125,6 @@ func (j *CheckClientIpJob) hasLimitIp() bool {
} }
func (j *CheckClientIpJob) processLogFile() bool { func (j *CheckClientIpJob) processLogFile() bool {
ipRegex := regexp.MustCompile(`from (?:tcp:|udp:)?\[?([0-9a-fA-F\.:]+)\]?:\d+ accepted`) ipRegex := regexp.MustCompile(`from (?:tcp:|udp:)?\[?([0-9a-fA-F\.:]+)\]?:\d+ accepted`)
emailRegex := regexp.MustCompile(`email: (.+)$`) emailRegex := regexp.MustCompile(`email: (.+)$`)
timestampRegex := regexp.MustCompile(`^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2})`) timestampRegex := regexp.MustCompile(`^(\d{4}/\d{2}/\d{2} \d{2}:\d{2}:\d{2})`)
@ -166,7 +146,6 @@ func (j *CheckClientIpJob) processLogFile() bool {
} }
ip := ipMatches[1] ip := ipMatches[1]
if ip == "127.0.0.1" || ip == "::1" { if ip == "127.0.0.1" || ip == "::1" {
continue continue
} }
@ -175,6 +154,7 @@ func (j *CheckClientIpJob) processLogFile() bool {
if len(emailMatches) < 2 { if len(emailMatches) < 2 {
continue continue
} }
email := emailMatches[1] email := emailMatches[1]
// Extract timestamp from log line // Extract timestamp from log line
@ -194,18 +174,20 @@ func (j *CheckClientIpJob) processLogFile() bool {
if _, exists := inboundClientIps[email]; !exists { if _, exists := inboundClientIps[email]; !exists {
inboundClientIps[email] = make(map[string]int64) inboundClientIps[email] = make(map[string]int64)
} }
// Update timestamp - keep the latest // Update timestamp - keep the latest
if existingTime, ok := inboundClientIps[email][ip]; !ok || timestamp > existingTime { if existingTime, ok := inboundClientIps[email][ip]; !ok || timestamp > existingTime {
inboundClientIps[email][ip] = timestamp inboundClientIps[email][ip] = timestamp
} }
} }
if err := scanner.Err(); err != nil { if err := scanner.Err(); err != nil {
j.checkError(err) j.checkError(err)
} }
shouldCleanLog := false shouldCleanLog := false
for email, ipTimestamps := range inboundClientIps {
for email, ipTimestamps := range inboundClientIps {
// Convert to IPWithTimestamp slice // Convert to IPWithTimestamp slice
ipsWithTime := make([]IPWithTimestamp, 0, len(ipTimestamps)) ipsWithTime := make([]IPWithTimestamp, 0, len(ipTimestamps))
for ip, timestamp := range ipTimestamps { for ip, timestamp := range ipTimestamps {
@ -232,41 +214,41 @@ func (j *CheckClientIpJob) processLogFile() bool {
// and the merge policy can be exercised by a unit test. // and the merge policy can be exercised by a unit test.
func mergeClientIps(old, new []IPWithTimestamp, staleCutoff int64) map[string]int64 { func mergeClientIps(old, new []IPWithTimestamp, staleCutoff int64) map[string]int64 {
ipMap := make(map[string]int64, len(old)+len(new)) ipMap := make(map[string]int64, len(old)+len(new))
for _, ipTime := range old { for _, ipTime := range old {
if ipTime.Timestamp < staleCutoff { if ipTime.Timestamp < staleCutoff {
continue continue
} }
ipMap[ipTime.IP] = ipTime.Timestamp ipMap[ipTime.IP] = ipTime.Timestamp
} }
for _, ipTime := range new { for _, ipTime := range new {
if ipTime.Timestamp < staleCutoff { if ipTime.Timestamp < staleCutoff {
continue continue
} }
if existingTime, ok := ipMap[ipTime.IP]; !ok || ipTime.Timestamp > existingTime { if existingTime, ok := ipMap[ipTime.IP]; !ok || ipTime.Timestamp > existingTime {
ipMap[ipTime.IP] = ipTime.Timestamp ipMap[ipTime.IP] = ipTime.Timestamp
} }
} }
return ipMap return ipMap
} }
// partitionLiveIps splits the merged ip map into live (seen in the // partitionLiveIps splits the merged ip map into live (seen in the
// current scan) and historical (only in the db blob, still inside the // current scan) and historical (only in the db blob).
// staleness window).
// //
// only live ips count toward the per-client limit. historical ones stay // only live ips count toward the per-client limit. historical ones stay
// in the db so the panel keeps showing them, but they must not take a // in the db so the panel keeps showing them, but they must not take a
// protected slot. the 30min cutoff alone isn't tight enough: an ip that // protected slot.
// stopped connecting a few minutes ago still looks fresh to
// mergeClientIps, and since the over-limit picker sorts ascending and
// keeps the oldest, those idle entries used to win the slot while the
// ip actually connecting got classified as excess and sent to fail2ban
// every tick. see #4077 / #4091.
// //
// live is sorted ascending so the "protect original, ban newcomer" // live is sorted ascending so the "protect original, ban newcomer"
// rule still holds when several ips are really connecting at once. // rule still holds when several ips are really connecting at once.
func partitionLiveIps(ipMap map[string]int64, observedThisScan map[string]bool) (live, historical []IPWithTimestamp) { func partitionLiveIps(ipMap map[string]int64, observedThisScan map[string]bool) (live, historical []IPWithTimestamp) {
live = make([]IPWithTimestamp, 0, len(observedThisScan)) live = make([]IPWithTimestamp, 0, len(observedThisScan))
historical = make([]IPWithTimestamp, 0, len(ipMap)) historical = make([]IPWithTimestamp, 0, len(ipMap))
for ip, ts := range ipMap { for ip, ts := range ipMap {
entry := IPWithTimestamp{IP: ip, Timestamp: ts} entry := IPWithTimestamp{IP: ip, Timestamp: ts}
if observedThisScan[ip] { if observedThisScan[ip] {
@ -275,8 +257,10 @@ func partitionLiveIps(ipMap map[string]int64, observedThisScan map[string]bool)
historical = append(historical, entry) historical = append(historical, entry)
} }
} }
sort.Slice(live, func(i, j int) bool { return live[i].Timestamp < live[j].Timestamp }) 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(historical, func(i, j int) bool { return historical[i].Timestamp < historical[j].Timestamp })
return live, historical return live, historical
} }
@ -284,6 +268,7 @@ func (j *CheckClientIpJob) checkFail2BanInstalled() bool {
cmd := "fail2ban-client" cmd := "fail2ban-client"
args := []string{"-h"} args := []string{"-h"}
err := exec.Command(cmd, args...).Run() err := exec.Command(cmd, args...).Run()
return err == nil return err == nil
} }
@ -297,6 +282,7 @@ func (j *CheckClientIpJob) checkAccessLogAvailable(iplimitActive bool) bool {
if iplimitActive { if iplimitActive {
logger.Warning("[LimitIP] Access log path is not set, Please configure the access log path in Xray configs.") logger.Warning("[LimitIP] Access log path is not set, Please configure the access log path in Xray configs.")
} }
return false return false
} }
@ -312,15 +298,18 @@ func (j *CheckClientIpJob) checkError(e error) {
func (j *CheckClientIpJob) getInboundClientIps(clientEmail string) (*model.InboundClientIps, error) { func (j *CheckClientIpJob) getInboundClientIps(clientEmail string) (*model.InboundClientIps, error) {
db := database.GetDB() db := database.GetDB()
InboundClientIps := &model.InboundClientIps{} InboundClientIps := &model.InboundClientIps{}
err := db.Model(model.InboundClientIps{}).Where("client_email = ?", clientEmail).First(InboundClientIps).Error err := db.Model(model.InboundClientIps{}).Where("client_email = ?", clientEmail).First(InboundClientIps).Error
if err != nil { if err != nil {
return nil, err return nil, err
} }
return InboundClientIps, nil return InboundClientIps, nil
} }
func (j *CheckClientIpJob) addInboundClientIps(clientEmail string, ipsWithTime []IPWithTimestamp) error { func (j *CheckClientIpJob) addInboundClientIps(clientEmail string, ipsWithTime []IPWithTimestamp) error {
inboundClientIps := &model.InboundClientIps{} inboundClientIps := &model.InboundClientIps{}
jsonIps, err := json.Marshal(ipsWithTime) jsonIps, err := json.Marshal(ipsWithTime)
j.checkError(err) j.checkError(err)
@ -342,6 +331,7 @@ func (j *CheckClientIpJob) addInboundClientIps(clientEmail string, ipsWithTime [
if err != nil { if err != nil {
return err return err
} }
return nil return nil
} }
@ -360,11 +350,13 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
settings := map[string][]model.Client{} settings := map[string][]model.Client{}
json.Unmarshal([]byte(inbound.Settings), &settings) json.Unmarshal([]byte(inbound.Settings), &settings)
clients := settings["clients"] clients := settings["clients"]
// Find the client's IP limit // Find the client's IP limit
var limitIp int var limitIp int
var clientFound bool var clientFound bool
for _, client := range clients { for _, client := range clients {
if client.Email == clientEmail { if client.Email == clientEmail {
limitIp = client.LimitIP limitIp = client.LimitIP
@ -377,8 +369,10 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
// No limit or inbound disabled, just update and return // No limit or inbound disabled, just update and return
jsonIps, _ := json.Marshal(newIpsWithTime) jsonIps, _ := json.Marshal(newIpsWithTime)
inboundClientIps.Ips = string(jsonIps) inboundClientIps.Ips = string(jsonIps)
db := database.GetDB() db := database.GetDB()
db.Save(inboundClientIps) db.Save(inboundClientIps)
return false return false
} }
@ -388,9 +382,8 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
json.Unmarshal([]byte(inboundClientIps.Ips), &oldIpsWithTime) json.Unmarshal([]byte(inboundClientIps.Ips), &oldIpsWithTime)
} }
// Merge old and new IPs, evicting entries that haven't been // Keep historical IPs in the db; partitionLiveIps keeps Fail2Ban checks limited to this scan.
// re-observed in a while. See mergeClientIps / #4077 for why. ipMap := mergeClientIps(oldIpsWithTime, newIpsWithTime, 0)
ipMap := mergeClientIps(oldIpsWithTime, newIpsWithTime, time.Now().Unix()-ipStaleAfterSeconds)
// only ips seen in this scan count toward the limit. see // only ips seen in this scan count toward the limit. see
// partitionLiveIps. // partitionLiveIps.
@ -398,6 +391,7 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
for _, ipTime := range newIpsWithTime { for _, ipTime := range newIpsWithTime {
observedThisScan[ipTime.IP] = true observedThisScan[ipTime.IP] = true
} }
liveIps, historicalIps := partitionLiveIps(ipMap, observedThisScan) liveIps, historicalIps := partitionLiveIps(ipMap, observedThisScan)
shouldCleanLog := false shouldCleanLog := false
@ -405,6 +399,7 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
// historical db-only ips are excluded from this count on purpose. // historical db-only ips are excluded from this count on purpose.
var keptLive []IPWithTimestamp var keptLive []IPWithTimestamp
if len(liveIps) > limitIp { if len(liveIps) > limitIp {
shouldCleanLog = true shouldCleanLog = true
@ -422,11 +417,12 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
return false return false
} }
defer logIpFile.Close() defer logIpFile.Close()
ipLogger := log.New(logIpFile, "", log.LstdFlags) ipLogger := log.New(logIpFile, "", log.LstdFlags)
// log format is load-bearing: x-ui.sh create_iplimit_jails builds // log format is load-bearing: x-ui.sh create_iplimit_jails builds
// filter.d/3x-ipl.conf with // filter.d/3x-ipl.conf with
// failregex = \[LIMIT_IP\]\s*Email\s*=\s*<F-USER>.+</F-USER>\s*\|\|\s*Disconnecting OLD IP\s*=\s*<ADDR>\s*\|\|\s*Timestamp\s*=\s*\d+ // failregex = \[LIMIT_IP\]\s*Email\s*=\s*.+\s*\|\|\s*Disconnecting OLD IP\s*=\s*\s*\|\|\s*Timestamp\s*=\s*\d+
// don't change the wording. // don't change the wording.
for _, ipTime := range bannedLive { for _, ipTime := range bannedLive {
j.disAllowedIps = append(j.disAllowedIps, ipTime.IP) j.disAllowedIps = append(j.disAllowedIps, ipTime.IP)
@ -439,12 +435,13 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
keptLive = liveIps keptLive = liveIps
} }
// keep kept-live + historical in the blob so the panel keeps showing // keep all live + historical ips in the blob so the panel preserves
// recently seen ips. banned live ips are already in the fail2ban log // client ip history. fail2ban decisions above are still based only
// and will reappear in the next scan if they reconnect. // on live ips from the current scan.
dbIps := make([]IPWithTimestamp, 0, len(keptLive)+len(historicalIps)) dbIps := make([]IPWithTimestamp, 0, len(liveIps)+len(historicalIps))
dbIps = append(dbIps, keptLive...) dbIps = append(dbIps, liveIps...)
dbIps = append(dbIps, historicalIps...) dbIps = append(dbIps, historicalIps...)
jsonIps, _ := json.Marshal(dbIps) jsonIps, _ := json.Marshal(dbIps)
inboundClientIps.Ips = string(jsonIps) inboundClientIps.Ips = string(jsonIps)
@ -465,8 +462,8 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
// disconnectClientTemporarily removes and re-adds a client to force disconnect banned connections // disconnectClientTemporarily removes and re-adds a client to force disconnect banned connections
func (j *CheckClientIpJob) disconnectClientTemporarily(inbound *model.Inbound, clientEmail string, clients []model.Client) { func (j *CheckClientIpJob) disconnectClientTemporarily(inbound *model.Inbound, clientEmail string, clients []model.Client) {
var xrayAPI xray.XrayAPI var xrayAPI xray.XrayAPI
apiPort := j.resolveXrayAPIPort()
apiPort := j.resolveXrayAPIPort()
err := xrayAPI.Init(apiPort) err := xrayAPI.Init(apiPort)
if err != nil { if err != nil {
logger.Warningf("[LIMIT_IP] Failed to init Xray API for disconnection: %v", err) logger.Warningf("[LIMIT_IP] Failed to init Xray API for disconnection: %v", err)
@ -503,6 +500,7 @@ func (j *CheckClientIpJob) disconnectClientTemporarily(inbound *model.Inbound, c
// reading it from the inbound settings (e.g., settings["method"]). // reading it from the inbound settings (e.g., settings["method"]).
if string(inbound.Protocol) == "shadowsocks" { if string(inbound.Protocol) == "shadowsocks" {
var inboundSettings map[string]any var inboundSettings map[string]any
if err := json.Unmarshal([]byte(inbound.Settings), &inboundSettings); err != nil { if err := json.Unmarshal([]byte(inbound.Settings), &inboundSettings); err != nil {
logger.Warningf("[LIMIT_IP] Failed to parse inbound settings for shadowsocks cipher: %v", err) logger.Warningf("[LIMIT_IP] Failed to parse inbound settings for shadowsocks cipher: %v", err)
} else { } else {
@ -542,6 +540,7 @@ func (j *CheckClientIpJob) resolveXrayAPIPort() int {
db := database.GetDB() db := database.GetDB()
var template model.Setting var template model.Setting
if err := db.Where("key = ?", "xrayTemplateConfig").First(&template).Error; err == nil { if err := db.Where("key = ?", "xrayTemplateConfig").First(&template).Error; err == nil {
if port, parseErr := getAPIPortFromConfigData([]byte(template.Value)); parseErr == nil { if port, parseErr := getAPIPortFromConfigData([]byte(template.Value)); parseErr == nil {
return port return port
@ -573,6 +572,7 @@ func getAPIPortFromConfigPath(configPath string) (int, error) {
func getAPIPortFromConfigData(configData []byte) (int, error) { func getAPIPortFromConfigData(configData []byte) (int, error) {
xrayConfig := &xray.Config{} xrayConfig := &xray.Config{}
if err := json.Unmarshal(configData, xrayConfig); err != nil { if err := json.Unmarshal(configData, xrayConfig); err != nil {
return 0, err return 0, err
} }