feat(online): use xray online-stats API for onlines and access-log-free IP limit

Adopt xray-core's statsUserOnline policy and GetUsersStats RPC so online detection is connection-based and IP limiting no longer requires an access log. Falls back to the legacy traffic-delta onlines and access-log parsing when the running core lacks the RPCs (Unimplemented), probed lazily per process so a panel-driven version switch re-evaluates automatically. Backend: - xray/api.go: GetOnlineUsers (one GetUsersStats call returns all online users and their source IPs) and IsUnimplementedErr. - xray/process.go: per-process OnlineAPISupport tri-state capability cache. - service/xray.go: ensureStatsPolicy injects statsUserOnline into every policy level of the generated config; XrayService.GetOnlineUsers probes and falls back. - job/xray_traffic_job.go: union API onlines into the delta-derived active set; bump last_online for idle-but-connected clients. - job/check_client_ip_job.go: API-first IP source with shared enforcement; live observations bypass the 30-min stale cutoff; access-log path unchanged for older cores. - service/setting.go: GetIpLimitEnable always true; new accessLogEnable default for features that genuinely read the access log. Frontend: - Client form split into Basic and Config tabs; IP Limit and IP Log no longer gated on access log; compact Auto Renew next to Start After First Use; tabBasic/tabConfig added to all 13 locales. - Xray logs button on the dashboard now gated on accessLogEnable.
2026-08-22 10:57:14 +00:00 · 2026-06-11 19:42:03 +02:00
parent 58905d81a4
commit 7bcc5830c6
33 changed files with 790 additions and 285 deletions
@@ -16,6 +16,7 @@ import (
 	"github.com/mhsanaei/3x-ui/v3/internal/database"
 	"github.com/mhsanaei/3x-ui/v3/internal/database/model"
 	"github.com/mhsanaei/3x-ui/v3/internal/logger"
+	"github.com/mhsanaei/3x-ui/v3/internal/web/service"
 	"github.com/mhsanaei/3x-ui/v3/internal/xray"

 	"gorm.io/gorm"
@@ -27,10 +28,14 @@ type IPWithTimestamp struct {
 	Timestamp int64  `json:"timestamp"`
 }

-// CheckClientIpJob monitors client IP addresses from access logs and manages IP blocking based on configured limits.
+// 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 when the running core supports it (no access log needed), falling back
+// to access-log parsing on older cores.
 type CheckClientIpJob struct {
 	lastClear     int64
 	disAllowedIps []string
+	xrayService   service.XrayService
 }

 var job *CheckClientIpJob
@@ -50,22 +55,32 @@ func (j *CheckClientIpJob) Run() {
 		j.lastClear = time.Now().Unix()
 	}

-	shouldClearAccessLog := false
 	fail2BanEnabled := isFail2BanEnabled()
 	hasLimit := fail2BanEnabled && j.hasLimitIp()
 	f2bInstalled := false
 	if hasLimit {
 		f2bInstalled = j.checkFail2BanInstalled()
 	}
+
+	if observed, apiMode := j.collectFromOnlineAPI(); apiMode {
+		if fail2BanEnabled {
+			j.processObserved(observed, j.resolveEnforce(hasLimit, f2bInstalled), true)
+		}
+		// The core tracks online IPs itself, so no access log is needed in this
+		// mode; still rotate a user-configured access log hourly so it doesn't
+		// grow unboundedly. The enforcement-triggered rotation is skipped —
+		// nothing here reads the log.
+		if j.checkAccessLogAvailable(false) && time.Now().Unix()-j.lastClear > 3600 {
+			j.clearAccessLog()
+		}
+		return
+	}
+
+	shouldClearAccessLog := false
 	isAccessLogAvailable := j.checkAccessLogAvailable(hasLimit)

 	if fail2BanEnabled && isAccessLogAvailable {
-		enforce := hasLimit
-		if hasLimit && runtime.GOOS != "windows" && !f2bInstalled {
-			logger.Warning("[LimitIP] Fail2Ban is not installed, Please install Fail2Ban from the x-ui bash menu.")
-			enforce = false
-		}
-		shouldClearAccessLog = j.processLogFile(enforce)
+		shouldClearAccessLog = j.processLogFile(j.resolveEnforce(hasLimit, f2bInstalled))
 	}

 	if shouldClearAccessLog || (isAccessLogAvailable && time.Now().Unix()-j.lastClear > 3600) {
@@ -73,6 +88,50 @@ func (j *CheckClientIpJob) Run() {
 	}
 }

+// resolveEnforce decides whether limits can actually be enforced this run,
+// warning when fail2ban is missing on a platform that needs it.
+func (j *CheckClientIpJob) resolveEnforce(hasLimit, f2bInstalled bool) bool {
+	if hasLimit && runtime.GOOS != "windows" && !f2bInstalled {
+		logger.Warning("[LimitIP] Fail2Ban is not installed, Please install Fail2Ban from the x-ui bash menu.")
+		return false
+	}
+	return hasLimit
+}
+
+// collectFromOnlineAPI builds per-email IP observations (email -> ip ->
+// last-seen unix seconds) from the core's online-stats API. ok=false means the
+// API is unavailable — xray not running, an older core, or a transient gRPC
+// failure — and the caller must fall back to access-log parsing.
+func (j *CheckClientIpJob) collectFromOnlineAPI() (map[string]map[string]int64, bool) {
+	onlineUsers, ok, err := j.xrayService.GetOnlineUsers()
+	if err != nil {
+		logger.Debug("[LimitIP] online-stats API unavailable this run:", err)
+		return nil, false
+	}
+	if !ok {
+		return nil, false
+	}
+	now := time.Now().Unix()
+	observed := make(map[string]map[string]int64, len(onlineUsers))
+	for _, user := range onlineUsers {
+		for _, entry := range user.IPs {
+			// No localhost guard needed here: the core's OnlineMap.AddIP drops
+			// 127.0.0.1/[::1] itself, so they never reach this list.
+			ts := entry.LastSeen
+			if ts <= 0 {
+				ts = now
+			}
+			if _, exists := observed[user.Email]; !exists {
+				observed[user.Email] = make(map[string]int64)
+			}
+			if existing, seen := observed[user.Email][entry.IP]; !seen || ts > existing {
+				observed[user.Email][entry.IP] = ts
+			}
+		}
+	}
+	return observed, true
+}
+
 func (j *CheckClientIpJob) clearAccessLog() {
 	logAccessP, err := os.OpenFile(xray.GetAccessPersistentLogPath(), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
 	j.checkError(err)
@@ -183,18 +242,26 @@ func (j *CheckClientIpJob) processLogFile(enforce bool) bool {
 		j.checkError(err)
 	}

-	shouldCleanLog := false
-	for email, ipTimestamps := range inboundClientIps {
+	return j.processObserved(inboundClientIps, enforce, false)
+}

-		// The access log can still reference a client that was just renamed
+// processObserved runs collection + enforcement for one scan's observations
+// (email -> ip -> last-seen unix seconds). observedAreLive marks the
+// observations as live connections (online-stats API) rather than recent log
+// lines: live entries bypass the stale cutoff, since a connection that opened
+// hours ago is still live even though its timestamp is old.
+func (j *CheckClientIpJob) processObserved(observed map[string]map[string]int64, enforce, observedAreLive bool) bool {
+	shouldCleanLog := false
+	for email, ipTimestamps := range observed {
+
+		// The observations can still reference a client that was just renamed
 		// or deleted; its email no longer matches any inbound. Skip it (and
 		// drop any orphaned tracking row) instead of recreating a row and
-		// logging an ERROR every run until the log rotates out the old email
-		// (#4963).
+		// logging an ERROR every run (#4963).
 		inbound, err := j.getInboundByEmail(email)
 		if err != nil {
 			if errors.Is(err, gorm.ErrRecordNotFound) {
-				logger.Debugf("[LimitIP] skipping stale access-log email %q (renamed or deleted)", email)
+				logger.Debugf("[LimitIP] skipping stale observed email %q (renamed or deleted)", email)
 				j.delInboundClientIps(email)
 			} else {
 				j.checkError(err)
@@ -214,13 +281,17 @@ func (j *CheckClientIpJob) processLogFile(enforce bool) bool {
 			continue
 		}

-		shouldCleanLog = j.updateInboundClientIps(clientIpsRecord, inbound, email, ipsWithTime, enforce) || shouldCleanLog
+		shouldCleanLog = j.updateInboundClientIps(clientIpsRecord, inbound, email, ipsWithTime, enforce, observedAreLive) || shouldCleanLog
 	}

 	return shouldCleanLog
 }

-func mergeClientIps(old, new []IPWithTimestamp, staleCutoff int64) map[string]int64 {
+// mergeClientIps folds this scan's observations into the persisted set,
+// dropping entries older than staleCutoff. newAlwaysLive exempts the new
+// entries from that cutoff: an API-observed IP is a live connection by
+// definition, even when its lastSeen (set at dispatch time) is hours old.
+func mergeClientIps(old, new []IPWithTimestamp, staleCutoff int64, newAlwaysLive bool) map[string]int64 {
 	ipMap := make(map[string]int64, len(old)+len(new))
 	for _, ipTime := range old {
 		if ipTime.Timestamp < staleCutoff {
@@ -229,7 +300,7 @@ func mergeClientIps(old, new []IPWithTimestamp, staleCutoff int64) map[string]in
 		ipMap[ipTime.IP] = ipTime.Timestamp
 	}
 	for _, ipTime := range new {
-		if ipTime.Timestamp < staleCutoff {
+		if !newAlwaysLive && ipTime.Timestamp < staleCutoff {
 			continue
 		}
 		if existingTime, ok := ipMap[ipTime.IP]; !ok || ipTime.Timestamp > existingTime {
@@ -239,6 +310,16 @@ func mergeClientIps(old, new []IPWithTimestamp, staleCutoff int64) map[string]in
 	return ipMap
 }

+// selectIpsToBan splits the live IPs (sorted oldest-first by partitionLiveIps)
+// into the newest `limit` entries to keep and the older remainder to ban.
+func selectIpsToBan(live []IPWithTimestamp, limit int) (kept, banned []IPWithTimestamp) {
+	if limit <= 0 || len(live) <= limit {
+		return live, nil
+	}
+	cutoff := len(live) - limit
+	return live[cutoff:], live[:cutoff]
+}
+
 func partitionLiveIps(ipMap map[string]int64, observedThisScan map[string]bool) (live, historical []IPWithTimestamp) {
 	live = make([]IPWithTimestamp, 0, len(observedThisScan))
 	historical = make([]IPWithTimestamp, 0, len(ipMap))
@@ -343,7 +424,7 @@ func (j *CheckClientIpJob) delInboundClientIps(clientEmail string) {
 	}
 }

-func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.InboundClientIps, inbound *model.Inbound, clientEmail string, newIpsWithTime []IPWithTimestamp, enforce bool) bool {
+func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.InboundClientIps, inbound *model.Inbound, clientEmail string, newIpsWithTime []IPWithTimestamp, enforce, observedAreLive bool) bool {
 	if inbound.Settings == "" {
 		logger.Debug("wrong data:", inbound)
 		return false
@@ -380,7 +461,7 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
 		json.Unmarshal([]byte(inboundClientIps.Ips), &oldIpsWithTime)
 	}

-	ipMap := mergeClientIps(oldIpsWithTime, newIpsWithTime, time.Now().Unix()-ipStaleAfterSeconds)
+	ipMap := mergeClientIps(oldIpsWithTime, newIpsWithTime, time.Now().Unix()-ipStaleAfterSeconds, observedAreLive)

 	// only ips seen in this scan count toward the limit. see
 	// partitionLiveIps.
@@ -394,15 +475,10 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun
 	j.disAllowedIps = []string{}

 	// historical db-only ips are excluded from this count on purpose.
-	var keptLive []IPWithTimestamp
-	if len(liveIps) > limitIp {
+	keptLive, bannedLive := selectIpsToBan(liveIps, limitIp)
+	if len(bannedLive) > 0 {
 		shouldCleanLog = true

-		// keep the newest live ips, ban older ones.
-		cutoff := len(liveIps) - limitIp
-		keptLive = liveIps[cutoff:]
-		bannedLive := liveIps[:cutoff]
-
 		logIpFile, err := os.OpenFile(xray.GetIPLimitLogPath(), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0644)
 		if err != nil {
 			logger.Errorf("failed to open IP limit log file: %s", err)
@@ -422,8 +498,6 @@ func (j *CheckClientIpJob) updateInboundClientIps(inboundClientIps *model.Inboun

 		// force xray to drop existing connections from banned ips
 		j.disconnectClientTemporarily(inbound, clientEmail, clients)
-	} else {
-		keptLive = liveIps
 	}

 	// keep kept-live + historical in the blob so the panel keeps showing
@@ -199,7 +199,7 @@ func TestUpdateInboundClientIps_LiveIpNotBannedByStillFreshHistoricals(t *testin
 	if err != nil {
 		t.Fatalf("getInboundByEmail: %v", err)
 	}
-	shouldCleanLog := j.updateInboundClientIps(row, inbound, email, live, true)
+	shouldCleanLog := j.updateInboundClientIps(row, inbound, email, live, true, false)

 	if shouldCleanLog {
 		t.Fatalf("shouldCleanLog must be false, nothing should have been banned with 1 live ip under limit 3")
@@ -252,7 +252,7 @@ func TestUpdateInboundClientIps_ExcessLiveIpIsStillBanned(t *testing.T) {
 	if err != nil {
 		t.Fatalf("getInboundByEmail: %v", err)
 	}
-	shouldCleanLog := j.updateInboundClientIps(row, inbound, email, live, true)
+	shouldCleanLog := j.updateInboundClientIps(row, inbound, email, live, true, false)

 	if !shouldCleanLog {
 		t.Fatalf("shouldCleanLog must be true when the live set exceeds the limit")
@@ -22,7 +22,7 @@ func TestMergeClientIps_EvictsStaleOldEntries(t *testing.T) {
 		{IP: "2.2.2.2", Timestamp: 2000}, // same IP, newer log line
 	}

-	got := mergeClientIps(old, new, 1000)
+	got := mergeClientIps(old, new, 1000, false)

 	want := map[string]int64{"2.2.2.2": 2000}
 	if !reflect.DeepEqual(got, want) {
@@ -36,7 +36,7 @@ func TestMergeClientIps_KeepsFreshOldEntriesUnchanged(t *testing.T) {
 	old := []IPWithTimestamp{
 		{IP: "1.1.1.1", Timestamp: 1500},
 	}
-	got := mergeClientIps(old, nil, 1000)
+	got := mergeClientIps(old, nil, 1000, false)

 	want := map[string]int64{"1.1.1.1": 1500}
 	if !reflect.DeepEqual(got, want) {
@@ -48,7 +48,7 @@ func TestMergeClientIps_PrefersLaterTimestampForSameIp(t *testing.T) {
 	old := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 1500}}
 	new := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 1700}}

-	got := mergeClientIps(old, new, 1000)
+	got := mergeClientIps(old, new, 1000, false)

 	if got["1.1.1.1"] != 1700 {
 		t.Fatalf("expected latest timestamp 1700, got %d", got["1.1.1.1"])
@@ -59,7 +59,7 @@ func TestMergeClientIps_DropsStaleNewEntries(t *testing.T) {
 	// A log line with a clock-skewed old timestamp must not resurrect a
 	// stale IP past the cutoff.
 	new := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 500}}
-	got := mergeClientIps(nil, new, 1000)
+	got := mergeClientIps(nil, new, 1000, false)

 	if len(got) != 0 {
 		t.Fatalf("stale new IP should have been dropped, got %v", got)
@@ -72,7 +72,7 @@ func TestMergeClientIps_NoStaleCutoffStillWorks(t *testing.T) {
 	old := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 100}}
 	new := []IPWithTimestamp{{IP: "2.2.2.2", Timestamp: 200}}

-	got := mergeClientIps(old, new, 0)
+	got := mergeClientIps(old, new, 0, false)

 	want := map[string]int64{"1.1.1.1": 100, "2.2.2.2": 200}
 	if !reflect.DeepEqual(got, want) {
@@ -80,6 +80,66 @@ func TestMergeClientIps_NoStaleCutoffStillWorks(t *testing.T) {
 	}
 }

+func TestMergeClientIps_LiveObservationsBypassStaleCutoff(t *testing.T) {
+	// online-API mode: lastSeen is set when the connection was dispatched, so
+	// a connection held open for hours has an "old" timestamp while being live
+	// by definition. It must survive the stale cutoff.
+	new := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 500}} // opened long ago, still connected
+	got := mergeClientIps(nil, new, 1000, true)
+
+	want := map[string]int64{"1.1.1.1": 500}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("live observation must bypass the stale cutoff\ngot:  %v\nwant: %v", got, want)
+	}
+}
+
+func TestMergeClientIps_LiveModeStillEvictsStaleOldEntries(t *testing.T) {
+	// the bypass applies only to this scan's observations — persisted entries
+	// from past scans still age out as before.
+	old := []IPWithTimestamp{{IP: "2.2.2.2", Timestamp: 100}}
+	new := []IPWithTimestamp{{IP: "1.1.1.1", Timestamp: 2000}}
+	got := mergeClientIps(old, new, 1000, true)
+
+	want := map[string]int64{"1.1.1.1": 2000}
+	if !reflect.DeepEqual(got, want) {
+		t.Fatalf("stale db entry must still be evicted in live mode\ngot:  %v\nwant: %v", got, want)
+	}
+}
+
+func TestSelectIpsToBan(t *testing.T) {
+	live := []IPWithTimestamp{ // sorted oldest-first, as partitionLiveIps returns
+		{IP: "A", Timestamp: 100},
+		{IP: "B", Timestamp: 200},
+		{IP: "C", Timestamp: 300},
+	}
+
+	// over the limit: oldest connections are banned, newest keep the slots
+	kept, banned := selectIpsToBan(live, 1)
+	if got := collectIps(kept); !reflect.DeepEqual(got, []string{"C"}) {
+		t.Fatalf("newest ip must keep the slot, got %v", got)
+	}
+	if got := collectIps(banned); !reflect.DeepEqual(got, []string{"A", "B"}) {
+		t.Fatalf("older ips must be banned oldest-first, got %v", got)
+	}
+
+	// at the limit: nothing banned
+	kept, banned = selectIpsToBan(live, 3)
+	if len(banned) != 0 || len(kept) != 3 {
+		t.Fatalf("at-limit set must not ban, kept=%v banned=%v", kept, banned)
+	}
+
+	// under the limit: nothing banned
+	kept, banned = selectIpsToBan(live[:1], 3)
+	if len(banned) != 0 || len(kept) != 1 {
+		t.Fatalf("under-limit set must not ban, kept=%v banned=%v", kept, banned)
+	}
+
+	// defensive: non-positive limit never reaches enforcement, but must not panic
+	if _, banned := selectIpsToBan(live, 0); banned != nil {
+		t.Fatalf("zero limit must not ban, got %v", banned)
+	}
+}
+
 func collectIps(entries []IPWithTimestamp) []string {
 	out := make([]string, 0, len(entries))
 	for _, e := range entries {
@@ -66,21 +66,37 @@ func (j *XrayTrafficJob) Run() {
 		j.xrayService.SetToNeedRestart()
 	}

-	lastOnlineMap, err := j.inboundService.GetClientsLastOnline()
-	if err != nil {
-		logger.Warning("get clients last online failed:", err)
-	}
-	if lastOnlineMap == nil {
-		lastOnlineMap = make(map[string]int64)
-	}
 	// Derive the local online set from this poll's per-email deltas rather
 	// than the shared last_online column, which remote-node syncs also bump
 	// and would otherwise make a client active only on a remote node appear
 	// online on local inbounds.
 	activeEmails := make([]string, 0, len(clientTraffics))
+	deltaActive := make(map[string]bool, len(clientTraffics))
 	for _, ct := range clientTraffics {
 		if ct != nil && ct.Up+ct.Down > 0 {
 			activeEmails = append(activeEmails, ct.Email)
+			deltaActive[ct.Email] = true
+		}
+	}
+	// When the core supports the online-stats API, union in connection-based
+	// onlines. Neither signal alone covers everything: an idle-but-connected
+	// client moves no bytes between polls (the delta heuristic's blind spot),
+	// while a short-lived connection can close before this poll yet still show
+	// in the delta. Older cores fall back to deltas alone.
+	if onlineUsers, apiMode, ouErr := j.xrayService.GetOnlineUsers(); ouErr != nil {
+		logger.Debug("get online users from xray api failed:", ouErr)
+	} else if apiMode {
+		idleOnline := make([]string, 0, len(onlineUsers))
+		for _, u := range onlineUsers {
+			if !deltaActive[u.Email] {
+				activeEmails = append(activeEmails, u.Email)
+				idleOnline = append(idleOnline, u.Email)
+			}
+		}
+		// The traffic path only bumps last_online on a non-zero delta; keep the
+		// column fresh for clients kept online purely by a live connection.
+		if err := j.inboundService.BumpClientsLastOnline(idleOnline); err != nil {
+			logger.Warning("bump last online for connected clients failed:", err)
 		}
 	}
 	// Pair the email signal with the inbound tags that moved bytes this poll.
@@ -100,6 +116,13 @@ func (j *XrayTrafficJob) Run() {
 		return
 	}

+	lastOnlineMap, err := j.inboundService.GetClientsLastOnline()
+	if err != nil {
+		logger.Warning("get clients last online failed:", err)
+	}
+	if lastOnlineMap == nil {
+		lastOnlineMap = make(map[string]int64)
+	}
 	onlineClients := j.inboundService.GetOnlineClients()
 	if onlineClients == nil {
 		onlineClients = []string{}