fix(job): withhold fail2ban bans for addresses a datagram alone reports

The IP-limit job treated every address on the core's online list as one its sender proved to own and wrote the oldest over-limit ones for the 3x-ipl jail. For Shadowsocks over UDP and for mKCP the core lists a datagram's source field with no exchange behind it, so a client with an IP limit could forge sources and get any address firewalled. A client attached to an enabled inbound that accepts such datagrams, on the master or on a node, is still counted and disconnected, but its old addresses are no longer written for fail2ban; a warning says why.
2026-11-02 03:05:02 +03:00 · 2026-11-01 00:39:26 +01:00
parent 152c84efa2
commit 68e06a174b
2 changed files with 162 additions and 2 deletions
@@ -211,3 +211,102 @@ func TestProcessObserved_PublishesBanForCommittedScan(t *testing.T) {
 		t.Fatalf("a published address must be recorded so the next scan does not repeat it")
 	}
 }
+
+const (
+	rawStream      = `{"network":"tcp","security":"none"}`
+	shadowsocksUDP = `{"method":"2022-blake3-aes-256-gcm","network":"tcp,udp","clients":[{"email":"mallory","limitIp":1,"enable":true}]}`
+)
+
+// One datagram with a forged source puts any address on the core's online list,
+// so an address seen for such a client must not reach the file fail2ban follows.
+func TestPublishBans_WithholdsAddressesOnlyADatagramVouchesFor(t *testing.T) {
+	const email, forged = "mallory", "192.0.2.77"
+	node := 1
+	for _, tc := range []struct {
+		name      string
+		second    *model.Inbound
+		wantInLog bool
+	}{
+		{"only a TCP inbound", nil, true},
+		{"also on Shadowsocks with UDP", &model.Inbound{Enable: true}, false},
+		{"Shadowsocks with UDP is disabled", &model.Inbound{Enable: false}, true},
+		// Node syncs merge the node's observations into the same row (#6587).
+		{"Shadowsocks with UDP runs on a node", &model.Inbound{Enable: true, NodeID: &node}, false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			setupIntegrationDB(t)
+			db := database.GetDB()
+			// The scan resolves a client to its lowest inbound id, which is this one.
+			inbound := seedLinkedInboundWithClient(t, "inbound-vless", email, 1)
+			if tc.second != nil {
+				tc.second.Tag, tc.second.Port, tc.second.Protocol = "inbound-ss", 4322, model.Shadowsocks
+				tc.second.Settings, tc.second.StreamSettings = shadowsocksUDP, rawStream
+				if err := db.Create(tc.second).Error; err != nil {
+					t.Fatalf("seed second inbound: %v", err)
+				}
+				var client model.ClientRecord
+				if err := db.Where("email = ?", email).First(&client).Error; err != nil {
+					t.Fatalf("read client: %v", err)
+				}
+				if err := db.Create(&model.ClientInbound{ClientId: client.Id, InboundId: tc.second.Id}).Error; err != nil {
+					t.Fatalf("link second inbound: %v", err)
+				}
+			}
+
+			now := time.Now().Unix()
+			row := seedClientIps(t, email, []model.ClientIpEntry{{IP: forged, Timestamp: now - 5}})
+			live := []model.ClientIpEntry{{IP: forged, Timestamp: now - 5}, {IP: "203.0.113.4", Timestamp: now}}
+			overLimit, _ := NewCheckClientIpJob().enforceIpLimitForTest(t, row, inbound, email, 1, live, true)
+
+			if !overLimit {
+				t.Error("the client is over its limit and must still be disconnected")
+			}
+			body, _ := os.ReadFile(readIpLimitLogPath())
+			if got := contains(string(body), forged); got != tc.wantInLog {
+				t.Errorf("older address in the fail2ban log = %v, want %v:\n%s", got, tc.wantInLog, body)
+			}
+		})
+	}
+}
+
+// getInboundByEmail's settings scan is all that ties such a client to an inbound.
+func TestPublishBans_ClientOutsideTheRelationIsJudgedByItsResolvedInbound(t *testing.T) {
+	setupIntegrationDB(t)
+	const email, forged = "mallory", "192.0.2.77"
+	inbound := seedInboundOnlyWithClient(t, "inbound-ss", email, 1)
+	inbound.Protocol, inbound.Settings, inbound.StreamSettings = model.Shadowsocks, shadowsocksUDP, rawStream
+
+	now := time.Now().Unix()
+	row := seedClientIps(t, email, []model.ClientIpEntry{{IP: forged, Timestamp: now - 5}})
+	live := []model.ClientIpEntry{{IP: forged, Timestamp: now - 5}, {IP: "203.0.113.4", Timestamp: now}}
+	NewCheckClientIpJob().enforceIpLimitForTest(t, row, inbound, email, 1, live, true)
+
+	if body, _ := os.ReadFile(readIpLimitLogPath()); contains(string(body), forged) {
+		t.Errorf("an address only a datagram vouches for was written for fail2ban:\n%s", body)
+	}
+}
+
+func TestAcceptsUnverifiedSources(t *testing.T) {
+	for _, tc := range []struct {
+		name     string
+		protocol model.Protocol
+		settings string
+		stream   string
+		want     bool
+	}{
+		{"shadowsocks tcp,udp", model.Shadowsocks, shadowsocksUDP, rawStream, true},
+		{"shadowsocks network list", model.Shadowsocks, `{"network":["tcp","UDP"]}`, rawStream, true},
+		{"shadowsocks tcp", model.Shadowsocks, `{"network":"tcp"}`, rawStream, false},
+		{"shadowsocks core default", model.Shadowsocks, `{}`, rawStream, false},
+		{"vless raw", model.VLESS, `{"clients":[]}`, rawStream, false},
+		{"vless ignores settings.network", model.VLESS, `{"network":"udp"}`, rawStream, false},
+		{"vless kcp", model.VLESS, `{"clients":[]}`, `{"network":"kcp"}`, true},
+		{"vless mkcp by its newer key", model.VLESS, `{"clients":[]}`, `{"network":"tcp","method":"mKCP"}`, true},
+		{"hysteria proves the address in its QUIC handshake", model.Hysteria, `{"clients":[]}`, `{"network":"hysteria"}`, false},
+	} {
+		inbound := &model.Inbound{Protocol: tc.protocol, Settings: tc.settings, StreamSettings: tc.stream}
+		if got := acceptsUnverifiedSources(inbound); got != tc.want {
+			t.Errorf("%s: acceptsUnverifiedSources = %v, want %v", tc.name, got, tc.want)
+		}
+	}
+}
@@ -20,6 +20,8 @@ import (
 	"github.com/mhsanaei/3x-ui/v3/internal/web/service"
 	"github.com/mhsanaei/3x-ui/v3/internal/xray"

+	xnet "github.com/xtls/xray-core/common/net"
+	"github.com/xtls/xray-core/infra/conf"
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
 )
@@ -599,8 +601,8 @@ type pendingBan struct {
 	keptLive   int
 }

-// publishBans returns the clients whose lines reached the log. bannedSeen
-// advances only for those, so a failed write leaves the address retryable.
+// publishBans returns the clients whose lines reached the log or were withheld as
+// unproven. bannedSeen advances only for those, so a failed write stays retryable.
 func (j *CheckClientIpJob) publishBans(bans []pendingBan) []pendingBan {
 	published := make([]pendingBan, 0, len(bans))
 	var logIpFile *os.File
@@ -618,6 +620,14 @@ func (j *CheckClientIpJob) publishBans(bans []pendingBan) []pendingBan {
 			j.recordBannedSeen(b.email, b.candidates, nil)
 			continue
 		}
+		// The caller still disconnects the client; only a firewall ban needs an
+		// address the sender has proven to own.
+		if j.sourceCanBeForged(b.email, b.inbound) {
+			j.recordBannedSeen(b.email, b.candidates, actionable)
+			logger.Warningf("[LIMIT_IP] Client %s: %d old IPs not queued for fail2ban, an inbound of the client accepts datagrams whose source can be forged", b.email, len(actionable))
+			published = append(published, b)
+			continue
+		}
 		if logIpFile == nil {
 			f, err := os.OpenFile(xray.GetIPLimitLogPath(), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o644)
 			if err != nil {
@@ -637,6 +647,57 @@ func (j *CheckClientIpJob) publishBans(bans []pendingBan) []pendingBan {
 	return published
 }

+// sourceCanBeForged checks every enabled inbound the client is attached to: the
+// online list is per email, and node syncs merge node observations into the row.
+func (j *CheckClientIpJob) sourceCanBeForged(email string, resolved *model.Inbound) bool {
+	var inbounds []*model.Inbound
+	err := database.GetDB().Model(&model.Inbound{}).
+		Joins("JOIN client_inbounds ON client_inbounds.inbound_id = inbounds.id").
+		Joins("JOIN clients ON clients.id = client_inbounds.client_id").
+		Where("clients.email = ? AND inbounds.enable = ?", email, true).
+		Find(&inbounds).Error
+	if err != nil {
+		j.checkError(err)
+		return true
+	}
+	// A client not yet in the relation was resolved by getInboundByEmail's fallback.
+	if len(inbounds) == 0 && resolved != nil {
+		inbounds = append(inbounds, resolved)
+	}
+	for _, inbound := range inbounds {
+		if acceptsUnverifiedSources(inbound) {
+			return true
+		}
+	}
+	return false
+}
+
+// acceptsUnverifiedSources: Shadowsocks authenticates each UDP datagram on its own
+// and mKCP takes the request with the first ones, so the core never answers first.
+func acceptsUnverifiedSources(inbound *model.Inbound) bool {
+	var stream struct {
+		Method  string `json:"method"`
+		Network string `json:"network"`
+	}
+	_ = json.Unmarshal([]byte(inbound.StreamSettings), &stream)
+	// The core reads method as the newer name of network and prefers it.
+	if stream.Method != "" {
+		stream.Network = stream.Method
+	}
+	switch strings.ToLower(stream.Network) {
+	case "kcp", "mkcp":
+		return true
+	}
+	if inbound.Protocol != model.Shadowsocks {
+		return false
+	}
+	var settings struct {
+		Network *conf.NetworkList `json:"network"`
+	}
+	_ = json.Unmarshal([]byte(inbound.Settings), &settings)
+	return xnet.HasNetwork(settings.Network.Build(), xnet.Network_UDP)
+}
+
 // writeBanLines emits one line per address; the wording is load-bearing, since
 // x-ui.sh create_iplimit_jails builds filter.d/3x-ipl.conf failregex from it.
 func writeBanLines(w io.Writer, clientEmail string, actionable []model.ClientIpEntry) error {