Skip to content

Commit c8ed9bb

Browse files
authored
feat: per-connection serverbound packet rate limiter (#691)
Gate's existing quotas only limit new connections and logins per IP; an already-connected client could flood packets unchecked. Add a per-connection serverbound packet/byte rate limiter, closing the connection when a limit is exceeded. - pkg/internal/packetlimiter: a sliding-window counter (ring buffer) and a Limiter with Account(bytes) bool. A nil/zero-limit Limiter is disabled. - config: new packetLimiter section (interval, packetsPerSecond, bytesPerSecond) with defaults of a 7s window and 500 packets/s (bytes disabled). A limit <= 0 disables that dimension; both <= 0 (or interval <= 0) disables the limiter. - netmc: client connections (serverbound reads) account each packet's bytes in the read loop and are closed when over the limit. Backend connections pass a nil limiter (trusted, no limit). This adopts the rate-limiting approach Velocity added for the same purpose, fit to Gate's Go read-loop rather than its Netty pipeline. The sliding-window counter, the limiter, and the read-loop close behaviour each have tests (including a negative control that an unlimited connection is not closed).
1 parent f818164 commit c8ed9bb

10 files changed

Lines changed: 445 additions & 18 deletions

File tree

config.yml

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -134,6 +134,17 @@ config:
134134
burst: 3
135135
ops: 0.4
136136
maxEntries: 1000
137+
# Per-connection serverbound packet rate limiting. Unlike quota (which limits
138+
# new connections/logins per IP), this bounds how many packets/bytes a single
139+
# already-connected client may send, mitigating packet floods. The connection
140+
# is closed when a limit is exceeded.
141+
packetLimiter:
142+
# The sliding window the rates are measured over.
143+
interval: 7s
144+
# Max serverbound packets per second per connection. Set to 0 (or below) to disable.
145+
packetsPerSecond: 500
146+
# Max serverbound bytes per second per connection. Set to 0 (or below) to disable.
147+
bytesPerSecond: -1
137148
# Whether and how Gate should reply to GameSpy 4 (Minecraft query protocol on UDP) requests.
138149
query:
139150
enabled: false

pkg/edition/java/config/config.go

Lines changed: 23 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,11 @@ var DefaultConfig = Config{
5757
MaxEntries: 1000,
5858
},
5959
},
60+
PacketLimiter: PacketLimiter{
61+
Interval: configutil.Duration(7 * time.Second),
62+
PacketsPerSecond: 500,
63+
BytesPerSecond: -1, // disabled by default; packet-rate limiting is enough for most setups
64+
},
6065
Compression: Compression{
6166
Threshold: 256,
6267
Level: -1,
@@ -105,10 +110,11 @@ type Config struct { // TODO use https://github.com/projectdiscovery/yamldoc-go
105110
ConnectionTimeout configutil.Duration `yaml:"connectionTimeout,omitempty" json:"connectionTimeout,omitempty"` // Write timeout
106111
ReadTimeout configutil.Duration `yaml:"readTimeout,omitempty" json:"readTimeout,omitempty"` // Read timeout
107112

108-
Quota Quota `yaml:"quota,omitempty" json:"quota,omitempty"` // Rate limiting settings
109-
Compression Compression `yaml:"compression,omitempty" json:"compression,omitempty"`
110-
ProxyProtocol bool `yaml:"proxyProtocol,omitempty" json:"proxyProtocol,omitempty"` // Enable HA-Proxy protocol mode
111-
ProxyProtocolBackend bool `yaml:"proxyProtocolBackend" json:"proxyProtocolBackend,omitempty"` // Enable HA-Proxy protocol mode for backend servers
113+
Quota Quota `yaml:"quota,omitempty" json:"quota,omitempty"` // Rate limiting settings
114+
PacketLimiter PacketLimiter `yaml:"packetLimiter,omitempty" json:"packetLimiter,omitempty"` // Per-connection serverbound packet rate limiting
115+
Compression Compression `yaml:"compression,omitempty" json:"compression,omitempty"`
116+
ProxyProtocol bool `yaml:"proxyProtocol,omitempty" json:"proxyProtocol,omitempty"` // Enable HA-Proxy protocol mode
117+
ProxyProtocolBackend bool `yaml:"proxyProtocolBackend" json:"proxyProtocolBackend,omitempty"` // Enable HA-Proxy protocol mode for backend servers
112118

113119
ShouldPreventClientProxyConnections bool `yaml:"shouldPreventClientProxyConnections" json:"shouldPreventClientProxyConnections,omitempty"` // Sends player IP to Mojang on login
114120

@@ -154,7 +160,15 @@ type (
154160
Quota struct {
155161
Connections QuotaSettings `yaml:"connections"` // Limits new connections per second, per IP block.
156162
Logins QuotaSettings `yaml:"logins"` // Limits logins per second, per IP block.
157-
// Maybe add a bytes-per-sec limiter, or should be managed by a higher layer.
163+
}
164+
// PacketLimiter limits how many serverbound packets/bytes a single connection
165+
// may send over a sliding window, mitigating packet floods from already
166+
// connected clients (the Quota limits only apply at connect/login time).
167+
// A limit <= 0 disables that dimension; if both are <= 0 the limiter is off.
168+
PacketLimiter struct {
169+
Interval configutil.Duration `yaml:"interval"` // Sliding window the rates are measured over.
170+
PacketsPerSecond int `yaml:"packetsPerSecond"` // Max serverbound packets/s per connection (<=0 disables).
171+
BytesPerSecond int `yaml:"bytesPerSecond"` // Max serverbound bytes/s per connection (<=0 disables).
158172
}
159173
QuotaSettings struct {
160174
Enabled bool `yaml:"enabled"` // If false, there is no such limiting.
@@ -215,6 +229,10 @@ func (c *Config) Validate() (warns []error, errs []error) {
215229
}
216230
}
217231

232+
if pl := c.PacketLimiter; (pl.PacketsPerSecond > 0 || pl.BytesPerSecond > 0) && pl.Interval <= 0 {
233+
w("Packet limiter has a rate set but interval <= 0; the limiter is disabled. Set packetLimiter.interval > 0 to enable it.")
234+
}
235+
218236
if c.Lite.Enabled {
219237
return c.Lite.Validate()
220238
}

pkg/edition/java/netmc/connection.go

Lines changed: 24 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -27,6 +27,7 @@ import (
2727
"go.minekube.com/gate/pkg/edition/java/proto/state"
2828
"go.minekube.com/gate/pkg/edition/java/proto/version"
2929
"go.minekube.com/gate/pkg/gate/proto"
30+
"go.minekube.com/gate/pkg/internal/packetlimiter"
3031
"go.minekube.com/gate/pkg/util/errs"
3132
)
3233

@@ -141,6 +142,7 @@ func NewMinecraftConn(
141142
readTimeout time.Duration,
142143
writeTimeout time.Duration,
143144
compressionLevel int,
145+
packetLimiter *packetlimiter.Limiter,
144146
) (conn MinecraftConn, startReadLoop func()) {
145147
in := proto.ServerBound // reads from client are server bound (proxy <- client)
146148
out := proto.ClientBound // writes to client are client bound (proxy -> client)
@@ -156,17 +158,18 @@ func NewMinecraftConn(
156158

157159
ctx, cancel := context.WithCancel(ctx)
158160
c := &minecraftConn{
159-
log: log,
160-
c: base,
161-
ctx: ctx,
162-
cancelCtx: cancel,
163-
rd: NewReader(base, in, readTimeout, log),
164-
wr: NewWriter(base, out, writeTimeout, compressionLevel, log),
165-
state: state.Handshake,
166-
protocol: version.Minecraft_1_7_2.Protocol,
167-
connType: phase.Undetermined,
168-
direction: direction,
169-
autoReading: newStateControl(true),
161+
log: log,
162+
c: base,
163+
ctx: ctx,
164+
cancelCtx: cancel,
165+
rd: NewReader(base, in, readTimeout, log),
166+
wr: NewWriter(base, out, writeTimeout, compressionLevel, log),
167+
state: state.Handshake,
168+
protocol: version.Minecraft_1_7_2.Protocol,
169+
connType: phase.Undetermined,
170+
direction: direction,
171+
autoReading: newStateControl(true),
172+
packetLimiter: packetLimiter,
170173
}
171174
c.sessionHandlerMu.sessionHandlers = make(map[*state.Registry]SessionHandler)
172175
return c, c.startReadLoop
@@ -182,7 +185,8 @@ type minecraftConn struct {
182185
rd Reader
183186
wr Writer
184187

185-
autoReading *stateControl // Whether the connection should automatically read packets from the underlying connection.
188+
autoReading *stateControl // Whether the connection should automatically read packets from the underlying connection.
189+
packetLimiter *packetlimiter.Limiter // Per-connection serverbound rate limiter; nil disables it (e.g. backend connections).
186190

187191
ctx context.Context // is canceled when connection closed
188192
cancelCtx context.CancelFunc
@@ -239,6 +243,14 @@ func (c *minecraftConn) startReadLoop() {
239243
}
240244
bytesRead += int64(packetCtx.BytesRead)
241245

246+
// Enforce the per-connection serverbound packet rate limit (nil/disabled
247+
// for backend connections and when not configured).
248+
if !c.packetLimiter.Account(packetCtx.BytesRead) {
249+
c.log.Info("serverbound packet rate limit exceeded, closing connection",
250+
"remoteAddr", c.c.RemoteAddr())
251+
return false
252+
}
253+
242254
// TODO wrap packetCtx into struct with source info
243255
// (minecraftConn) and chain into packet interceptor to...
244256
// - packet interception
Lines changed: 104 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,104 @@
1+
package netmc
2+
3+
import (
4+
"context"
5+
"net"
6+
"testing"
7+
"time"
8+
9+
"github.com/go-logr/logr"
10+
11+
"go.minekube.com/gate/pkg/edition/java/proto/codec"
12+
"go.minekube.com/gate/pkg/edition/java/proto/packet"
13+
"go.minekube.com/gate/pkg/edition/java/proto/state"
14+
"go.minekube.com/gate/pkg/edition/java/proto/version"
15+
"go.minekube.com/gate/pkg/gate/proto"
16+
"go.minekube.com/gate/pkg/internal/packetlimiter"
17+
)
18+
19+
type noopSessionHandler struct{}
20+
21+
func (noopSessionHandler) HandlePacket(*proto.PacketContext) {}
22+
func (noopSessionHandler) Disconnected() {}
23+
func (noopSessionHandler) Activated() {}
24+
func (noopSessionHandler) Deactivated() {}
25+
26+
// A client that floods serverbound packets past its rate limit must have its
27+
// connection closed by the read loop.
28+
func TestReadLoopClosesConnectionOnPacketFlood(t *testing.T) {
29+
client, server := net.Pipe()
30+
defer client.Close()
31+
32+
// 1 packet/s over a 1s window: the 2nd packet in the window trips the limit.
33+
limiter := packetlimiter.New(1, -1, time.Second)
34+
conn, startReadLoop := NewMinecraftConn(
35+
context.Background(), server, proto.ServerBound,
36+
5*time.Second, 5*time.Second, 0, limiter,
37+
)
38+
conn.SetActiveSessionHandler(state.Handshake, noopSessionHandler{})
39+
40+
done := make(chan struct{})
41+
go func() { startReadLoop(); close(done) }()
42+
43+
// Flood handshake packets from the client side.
44+
go func() {
45+
enc := codec.NewEncoder(client, proto.ServerBound, logr.Discard())
46+
hs := &packet.Handshake{
47+
ProtocolVersion: int(version.Minecraft_1_21.Protocol),
48+
ServerAddress: "localhost",
49+
Port: 25565,
50+
NextStatus: 1,
51+
}
52+
for i := 0; i < 10; i++ {
53+
if _, err := enc.WritePacket(hs); err != nil {
54+
return // pipe closed once the limiter kicked in
55+
}
56+
}
57+
}()
58+
59+
select {
60+
case <-done:
61+
if !Closed(conn) {
62+
t.Fatal("read loop returned but connection is not closed")
63+
}
64+
case <-time.After(3 * time.Second):
65+
t.Fatal("connection was not closed after packet flood")
66+
}
67+
}
68+
69+
// Without a limiter, the same packet stream must NOT close the connection — this
70+
// guards against the flood test passing for an unrelated reason (e.g. a framing
71+
// or decode error).
72+
func TestReadLoopKeepsConnectionWithoutLimiter(t *testing.T) {
73+
client, server := net.Pipe()
74+
defer client.Close()
75+
76+
conn, startReadLoop := NewMinecraftConn(
77+
context.Background(), server, proto.ServerBound,
78+
5*time.Second, 5*time.Second, 0, nil, // no limiter
79+
)
80+
conn.SetActiveSessionHandler(state.Handshake, noopSessionHandler{})
81+
82+
done := make(chan struct{})
83+
go func() { startReadLoop(); close(done) }()
84+
85+
enc := codec.NewEncoder(client, proto.ServerBound, logr.Discard())
86+
hs := &packet.Handshake{
87+
ProtocolVersion: int(version.Minecraft_1_21.Protocol),
88+
ServerAddress: "localhost",
89+
Port: 25565,
90+
NextStatus: 1,
91+
}
92+
for i := 0; i < 10; i++ {
93+
if _, err := enc.WritePacket(hs); err != nil {
94+
t.Fatalf("write %d failed (connection closed unexpectedly): %v", i, err)
95+
}
96+
}
97+
98+
select {
99+
case <-done:
100+
t.Fatal("connection closed without a limiter configured")
101+
case <-time.After(200 * time.Millisecond):
102+
// Still open after processing the flood, as expected.
103+
}
104+
}

pkg/edition/java/proxy/proxy.go

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@ import (
3232
"go.minekube.com/gate/pkg/gate/proto"
3333
"go.minekube.com/gate/pkg/internal/addrquota"
3434
"go.minekube.com/gate/pkg/internal/connwrap"
35+
"go.minekube.com/gate/pkg/internal/packetlimiter"
3536
"go.minekube.com/gate/pkg/internal/reload"
3637
"go.minekube.com/gate/pkg/util/errs"
3738
"go.minekube.com/gate/pkg/util/netutil"
@@ -621,12 +622,16 @@ func (p *Proxy) HandleConn(raw net.Conn) {
621622
raw = e.Connection()
622623
}
623624

624-
// Create client connection
625+
// Create client connection. Client connections are serverbound (untrusted),
626+
// so apply the configured per-connection packet rate limiter.
627+
pl := p.cfg.PacketLimiter
628+
limiter := packetlimiter.New(pl.PacketsPerSecond, pl.BytesPerSecond, time.Duration(pl.Interval))
625629
conn, readLoop := netmc.NewMinecraftConn(
626630
ctx, raw, proto.ServerBound,
627631
time.Duration(p.cfg.ReadTimeout)*time.Millisecond,
628632
time.Duration(p.cfg.ConnectionTimeout)*time.Millisecond,
629633
p.cfg.Compression.Level,
634+
limiter,
630635
)
631636
conn.SetActiveSessionHandler(state.Handshake, newHandshakeSessionHandler(conn, &sessionHandlerDeps{
632637
proxy: p,

pkg/edition/java/proxy/server.go

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -410,6 +410,7 @@ func (s *serverConnection) connect(ctx context.Context) (result *connectionResul
410410
time.Duration(s.config().ReadTimeout)*time.Millisecond,
411411
time.Duration(s.config().ConnectionTimeout)*time.Millisecond,
412412
s.config().Compression.Level,
413+
nil, // backend connections are trusted; no serverbound rate limit
413414
)
414415
resultChan := make(chan *connResponse, 1)
415416

Lines changed: 98 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,98 @@
1+
package packetlimiter
2+
3+
import "time"
4+
5+
// counter is a sliding-window counter over a fixed interval. It keeps
6+
// (time, count) data points in a growable ring buffer, expiring entries older
7+
// than the interval and maintaining a running sum. Times are nanoseconds.
8+
//
9+
// It is a Go port of the IntervalledCounter approach used by Velocity/Paper.
10+
// Not safe for concurrent use; callers synchronize.
11+
type counter struct {
12+
interval int64 // window length in nanoseconds
13+
times []int64
14+
counts []int64
15+
head int // inclusive
16+
tail int // exclusive
17+
total int64
18+
minTime int64
19+
}
20+
21+
const initialCounterSize = 8
22+
23+
func newCounter(interval time.Duration) *counter {
24+
return &counter{
25+
interval: int64(interval),
26+
times: make([]int64, initialCounterSize),
27+
counts: make([]int64, initialCounterSize),
28+
}
29+
}
30+
31+
// updateAndAdd expires entries older than the window relative to now, then
32+
// records count at now.
33+
func (c *counter) updateAndAdd(count, now int64) {
34+
c.expire(now)
35+
c.add(now, count)
36+
}
37+
38+
// expire drops entries older than now-interval. Subtraction is used for the
39+
// comparison to stay correct across clock wraparound.
40+
func (c *counter) expire(now int64) {
41+
minTime := now - c.interval
42+
arrayLen := len(c.times)
43+
for c.head != c.tail && c.times[c.head]-minTime < 0 {
44+
c.total -= c.counts[c.head]
45+
c.counts[c.head] = 0
46+
c.head++
47+
if c.head >= arrayLen {
48+
c.head = 0
49+
}
50+
}
51+
c.minTime = minTime
52+
}
53+
54+
func (c *counter) add(now, count int64) {
55+
if now-c.minTime < 0 {
56+
return // older than the current window, ignore
57+
}
58+
nextTail := (c.tail + 1) % len(c.times)
59+
if nextTail == c.head {
60+
c.resize()
61+
nextTail = (c.tail + 1) % len(c.times)
62+
}
63+
c.times[c.tail] = now
64+
c.counts[c.tail] += count
65+
c.total += count
66+
c.tail = nextTail
67+
}
68+
69+
func (c *counter) resize() {
70+
oldTimes, oldCounts := c.times, c.counts
71+
oldLen := len(oldTimes)
72+
size := c.tail - c.head
73+
if size < 0 {
74+
size += oldLen
75+
}
76+
newTimes := make([]int64, oldLen*2)
77+
newCounts := make([]int64, oldLen*2)
78+
if c.tail >= c.head {
79+
copy(newTimes, oldTimes[c.head:c.tail])
80+
copy(newCounts, oldCounts[c.head:c.tail])
81+
} else {
82+
n := copy(newTimes, oldTimes[c.head:])
83+
copy(newTimes[n:], oldTimes[:c.tail])
84+
n = copy(newCounts, oldCounts[c.head:])
85+
copy(newCounts[n:], oldCounts[:c.tail])
86+
}
87+
c.times, c.counts = newTimes, newCounts
88+
c.head = 0
89+
c.tail = size
90+
}
91+
92+
// sum returns the total count currently within the window.
93+
func (c *counter) sum() int64 { return c.total }
94+
95+
// rate returns the per-second rate over the window.
96+
func (c *counter) rate() float64 {
97+
return float64(c.total) / (float64(c.interval) * 1e-9)
98+
}

0 commit comments

Comments
 (0)