diff --git a/PROTOCOL.md b/PROTOCOL.md index 50d24c2..8950da3 100644 --- a/PROTOCOL.md +++ b/PROTOCOL.md @@ -144,11 +144,21 @@ Flags : VarInt # feature flags; bit 0x01 (STREAM_FC) MUST be set RecvWindow: VarInt # client's per-stream receive window, bytes (§7.3) ``` -`Flags` is a bitfield of features. Bit `0x01` (STREAM_FC) declares -**per-stream flow control** (§7.3) and is mandatory: `RecvWindow` advertises -the client's per-stream receive window in bytes and must be positive. The hub -closes the connection if the flag is missing, `RecvWindow` is absent or -non-positive, or the fields are malformed. +`Flags` is a bitfield of features. + +| Bit | Name | Meaning | +|-----|------|---------| +| `0x01` | STREAM_FC | **Per-stream flow control** (§7.3). Mandatory. | +| `0x02` | WORKER_HEARTBEAT | Mux-level `PING`/`PONG` on worker conns (§7.4). Optional. | + +STREAM_FC is mandatory: `RecvWindow` advertises the client's per-stream receive +window in bytes and must be positive. The hub closes the connection if the flag +is missing, `RecvWindow` is absent or non-positive, or the fields are malformed. + +Optional bits are **negotiated**: the hub echoes in `SessionReady` only those it +accepts, and the client enables a feature only when its bit comes back. A hub +that does not know WORKER_HEARTBEAT simply omits the bit and the client falls +back to TCP keepalive alone. The hub: @@ -269,6 +279,8 @@ Data : Bytes[...] # remainder of the frame payload | `0x02` | FIN | both | *(empty)* — graceful close of the stream (both directions). This is the "disconnect" the hub sends when the player leaves. | | `0x03` | RST | both | *(optional 1 byte reason)* — abnormal close (e.g. CID unknown/expired, destination dial failed). | | `0x04` | WND | both | `Delta: VarInt` — flow-control credit grant (§7.3). | +| `0x05` | PING | both | `Nonce: I64` — liveness probe on reserved StreamID `0` (§7.4). | +| `0x06` | PONG | both | `Nonce: I64` — echoes the probe's nonce (§7.4). | There is no explicit SYN-ACK: success is implied by the hub forwarding the buffered Handshake as the stream's first `DATA`; failure is an `RST`. @@ -276,12 +288,22 @@ buffered Handshake as the stream's first `DATA`; failure is an `RST`. ### 7.1 Stream allocation (client side) The client keeps a pool of `1 ≤ N ≤ max_conn` worker conns (`max_conn` -configurable, `1..8`). To place a new stream: +configurable, `1..8`). The pool grows **breadth-first**: spreading streams over +several connections keeps any single TCP connection from becoming the shared +point of failure for every player on the tunnel. To place a new stream: -1. Pick the worker conn with the **fewest active streams**. -2. If that minimum conn is **saturated** (active streams `> 8`) **and** - `poolSize < max_conn`, dial a new worker conn and use it instead. -3. Otherwise use the least-loaded conn (even if it exceeds 8 at `max_conn`). +1. Pick the worker conn with the **fewest active streams**, and use it. +2. If that conn already carries at least one stream and + `poolSize + dialsInFlight < max_conn`, dial another worker conn **in the + background**. The stream just placed is not delayed by that dial; the new + conn becomes the least-loaded one and picks up subsequent streams. +3. Once the pool is at `max_conn`, streams stack on the least-loaded conn. + Exceeding `8` active streams there is logged as pool saturation. + +Only when the pool is *empty* does a caller dial synchronously, and then exactly +one caller dials while the others wait for its result. A dial is never performed +while holding the pool lock: session establishment is network I/O, and one +unresponsive hub must not be able to block unrelated players. ### 7.2 End-to-end player flow @@ -333,6 +355,37 @@ Every stream carries an independent credit window per direction: Both windows may differ (each side enforces the one its peer advertised). `Delta` must be positive; a `WND` for an unknown stream id is ignored. +### 7.4 Liveness + +TCP alone cannot tell a healthy idle connection from a dead one. When a stateful +middlebox forgets an established flow — conntrack expiry, a firewall reload, a +cloud load balancer's idle timeout — it sends neither `FIN` nor `RST`. Both ends +keep a socket that will never again carry a byte, and a reader parked on it waits +forever. Without an application-level probe the client cannot notice: its worker +conn stays in the pool, the hub keeps routing players to a control session nobody +reads, and service does not return until the client process is restarted. + +Every established session is therefore covered by a heartbeat: + +* **Control session** — the client sends `Ping` every `pingIntervalMs` and the + hub answers `Pong`. If no `Pong` arrives for `3 × pingIntervalMs`, the client + closes the session, which triggers its normal reconnect with backoff. +* **Worker conns** — when WORKER_HEARTBEAT was negotiated, the same exchange + runs as mux `PING`/`PONG` frames on the reserved StreamID `0`. On timeout the + client closes the conn; its streams are reset and it is dropped from the pool, + so the next player gets a freshly dialed connection. +* **Hub side** — an established redapricot session that receives no frame for + `sessionIdleTimeoutMs` (default 90000, `0` disables) is closed. Player + connections are never subject to this. + +Both ends also enable TCP keepalive, which catches the narrower case of a peer +that has become unreachable at the IP layer. + +Session establishment (§3.2) is bounded by a single deadline covering the dial, +the `Rekey` write and the `SessionReady` read, and every frame write is bounded +too — a peer that stops reading must not be able to park a whole multiplexed +connection inside one write. + ## 8. HAProxy protocol v2 (optional) When a mapping has `proxyProtocol: true`, the client prepends a PROXY v2 header @@ -361,13 +414,18 @@ big-endian. "psk": "change-me", "timestampWindowMs": 30000, "pendingTimeoutMs": 10000, - "streamWindowBytes": 262144 + "streamWindowBytes": 262144, + "sessionIdleTimeoutMs": 90000 } ``` `streamWindowBytes` (optional, default 262144, clamped to [32768, 8388608]) is the hub's advertised per-stream receive window (§7.3). +`sessionIdleTimeoutMs` (optional, default 90000) closes an established control +session or worker conn that has gone silent for that long (§7.4). It must stay +comfortably above the client's `pingIntervalMs`; `0` disables the watchdog. + ### 9.2 Client — JSON ```json @@ -392,6 +450,15 @@ which is `mc\\.example\\.com` in JSON); an unescaped `.` matches any character. Use ordinary regex to route wildcards, e.g. `.*\.example\.com` for every subdomain or `(alpha|beta)\.mc\.net` for a fixed set. +`velocitySecret` (optional, per mapping) makes the client speak Velocity +"modern forwarding" towards that destination: during the Minecraft login phase +it swallows the backend's `velocity:player_info` Login Plugin Request and +answers with an HMAC-SHA256-signed payload carrying the player's real IP, +username and UUID (the UUID claimed in Login Start, or the offline-mode UUID +for protocols that carry none). Set it to the backend's +`proxies.velocity.secret`. This is purely client↔destination behavior — it does +not appear on the tunnel wire, and the exchange is invisible to the player. + ## 10. Constants summary | Name | Value | @@ -408,9 +475,13 @@ subdomain or `(alpha|beta)\.mc\.net` for a fixed set. | pattern matching | case-insensitive, whole-string regex; first match wins | | CID length | 16 bytes | | max frame payload | 1 MiB | -| saturation threshold | active streams `> 8` | +| pool growth | breadth-first: grow to `max_conn` before stacking | +| saturation threshold (logged) | active streams `> 8` at `max_conn` | | max worker conns | `max_conn ∈ [1,8]` | -| feature flag: per-stream flow control | `0x01` | +| feature flag: per-stream flow control | `0x01` (mandatory) | +| feature flag: worker heartbeat | `0x02` (negotiated) | +| heartbeat timeout | `3 × pingIntervalMs` | +| hub session idle timeout | 90000 ms (`0` disables) | | stream window default / bounds | 256 KiB, clamped to [32 KiB, 8 MiB] | | WND grant batching (reference) | one grant per window/2 consumed | | DATA chunk cap (reference) | 32 KiB | diff --git a/README.md b/README.md index aecee12..b5150b8 100644 --- a/README.md +++ b/README.md @@ -45,7 +45,8 @@ that carries many players as lightweight *streams* — opens a stream for that C dials the real destination (optionally announcing the player's real IP with the **HAProxy v2** protocol), and bridges the two ends. Worker connections are pooled: the client uses up to `maxConn` of them and always places a new stream -on the least-loaded one. +on the least-loaded one, growing the pool to `maxConn` before stacking streams +so no single TCP connection carries every player. ## Repository layout @@ -126,6 +127,10 @@ just add the hub's IP with that hostname), then join `mc.example.com` in Minecraft. The hub matches the hostname against the registered regex patterns and tunnels you to `127.0.0.1:25566` behind the client. With `proxyProtocol: true`, the real server sees your true IP (enable `proxy-protocol` / a compatible front-end on that server to consume it). +For a Paper backend, setting `velocitySecret` instead is usually nicer: the +client answers the backend's Velocity modern-forwarding login query, so the +server sees your real IP, username and UUID without any front-end — configure +the backend with `proxies.velocity.enabled: true` and the same secret. ## Container image (client) @@ -163,6 +168,7 @@ secrets. The base image and build flags live in `.ko.yaml`. | `psk` | *(required)* | Shared secret; must match every client. | | `timestampWindowMs` | `30000` | Allowed clock skew for a client's rekey timestamp. | | `pendingTimeoutMs` | `10000` | How long a matched player waits for a worker to take over. | +| `sessionIdleTimeoutMs` | `90000` | Close an established control/worker session that receives no frame for this long. Must exceed the client's `pingIntervalMs`; `0` disables. Player connections are unaffected. | ### Client (`client/config.example.json`) @@ -171,11 +177,12 @@ secrets. The base image and build flags live in `.ko.yaml`. | `server` | *(required)* | Hub `host:port`. | | `psk` | *(required)* | Shared secret; must match the hub. | | `maxConn` | `1` (clamped 1–8) | Max worker connections in the pool. | -| `pingIntervalMs` | `20000` | Control-session keepalive interval. | +| `pingIntervalMs` | `20000` (min 1000) | Heartbeat interval for the control session and every worker conn. A session with no reply for `3×` this is dropped and re-established. | | `mappings[]` | *(≥1 required)* | Route table (below). | | `mappings[].pattern` | — | Regex matched against the whole player hostname, case-insensitively. Escape dots (`mc\.example\.com`); `.` is a wildcard. | | `mappings[].destination` | — | Real server `host:port` to forward to. | | `mappings[].proxyProtocol` | `false` | Prepend a HAProxy v2 header carrying the player's IP. | +| `mappings[].velocitySecret` | *(off)* | Answer the destination's [Velocity modern forwarding](https://docs.papermc.io/velocity/player-information-forwarding/) login query with this secret, forwarding the player's real IP, username and UUID. Match it to the backend's `proxies.velocity.secret` (Paper). The forwarded profile carries no skin properties — the tunnel performs no Mojang authentication. | ## Testing @@ -201,7 +208,8 @@ The e2e suite covers: a full player round-trip with verbatim handshake forwarding and case-insensitive matching, regex wildcard pattern routing, multi-megabyte transfers, concurrent streams spreading across multiple worker connections, HAProxy v2 source-address -propagation, player- and destination-initiated disconnect propagation, wrong-PSK +propagation, Velocity modern-forwarding interception (signed player-info +handoff to a mock Paper backend), player- and destination-initiated disconnect propagation, wrong-PSK rejection, dropping of unmatched hostnames, stream isolation under a slow player and under a slow destination (no head-of-line blocking), and rejection of pre-flow-control peers. The Go and Java crypto layers are @@ -220,6 +228,12 @@ drift apart. application-level head-of-line blocking between streams. What remains is TCP-level HOL (packet loss stalls a whole worker connection briefly); raising `maxConn` spreads that. +* **Liveness is explicit.** Every session heartbeats, every socket write is + bounded, and session establishment has a deadline. A path that dies silently — + no `FIN`, no `RST`, as when a NAT or firewall forgets an established flow — is + detected within `3 × pingIntervalMs`, the dead connection is dropped from the + pool, and service is restored without operator action. TCP keepalive is on as + a second line of defence. * **Single hub event loop.** The hub deploys one Vert.x verticle, so all state is confined to one event loop (no locking). Throughput is bounded by one core; ample for hundreds of players, not designed for tens of thousands. diff --git a/client/client.go b/client/client.go index 2165e0f..d0ee552 100644 --- a/client/client.go +++ b/client/client.go @@ -8,6 +8,7 @@ import ( "log" "net" "sync" + "sync/atomic" "time" "github.com/iceBear67/redapricot/client/wire" @@ -64,17 +65,30 @@ func clampWindow(w int) int { return w } +// session is an established redapricot session: the frame transport plus what +// was negotiated during establishment. +type session struct { + fc *wire.FramedConn + peerWnd int // hub's advertised per-stream receive window + heartbeat bool // hub accepted mux-level PING/PONG on worker conns +} + // dialSession opens a TCP connection, performs the Intent-17 handshake, the -// Phase-A rekey, and reads SessionReady, returning an established frame conn -// and the hub's advertised per-stream receive window. Per-stream flow control -// is mandatory: a hub that does not echo the STREAM_FC flag is rejected. -func (c *Client) dialSession(magic byte) (fc *wire.FramedConn, peerWnd int, err error) { - conn, err := net.DialTimeout("tcp", c.cfg.Server, 10*time.Second) +// Phase-A rekey, and reads SessionReady. Per-stream flow control is mandatory: +// a hub that does not echo the STREAM_FC flag is rejected. +// +// The whole exchange is bounded by HandshakeTimeout. A hub that accepts the +// socket but never answers (wedged event loop, a load balancer accepting on its +// behalf) must fail fast rather than park the caller forever. +func (c *Client) dialSession(magic byte) (sess *session, err error) { + conn, err := net.DialTimeout("tcp", c.cfg.Server, HandshakeTimeout) if err != nil { - return nil, 0, err + return nil, err } if tcp, ok := conn.(*net.TCPConn); ok { _ = tcp.SetNoDelay(true) + _ = tcp.SetKeepAlive(true) + _ = tcp.SetKeepAlivePeriod(TCPKeepAlivePeriod) } ok := false defer func() { @@ -82,15 +96,18 @@ func (c *Client) dialSession(magic byte) (fc *wire.FramedConn, peerWnd int, err _ = conn.Close() } }() + if err := conn.SetDeadline(time.Now().Add(HandshakeTimeout)); err != nil { + return nil, err + } // 1. plaintext Minecraft Handshake, Intent 17, address = hex(SHA3-224(PSK)). hs := wire.BuildHandshake(ProtocolVersion, c.pskAddr, c.serverPort, IntentRedapricot) if _, err := conn.Write(hs); err != nil { - return nil, 0, err + return nil, err } // 2. Phase-A ciphers derived from the PSK. - fc = wire.NewFramedConn(conn, + fc := wire.NewFramedConn(conn, wire.CipherFor(c.pskBytes, wire.DirS2C), // in: server -> client wire.CipherFor(c.pskBytes, wire.DirC2S), // out: client -> server ) @@ -99,13 +116,14 @@ func (c *Client) dialSession(magic byte) (fc *wire.FramedConn, peerWnd int, err // per-stream receive window. rnd := make([]byte, 16) if _, err := crand.Read(rnd); err != nil { - return nil, 0, err + return nil, err } ts := time.Now().UnixMilli() + offered := FlagStreamFC | FlagWorkerHeartbeat rekeyMsg := wire.NewWriter().U8(magic).VarInt(len(rnd)).Bytes(rnd).I64(ts). - VarInt(FlagStreamFC).VarInt(c.streamWnd).Out() + VarInt(offered).VarInt(c.streamWnd).Out() if err := fc.WriteFrame(rekeyMsg); err != nil { - return nil, 0, err + return nil, err } // 4. Switch to Phase-B ciphers: REKEY = Rand || Timestamp(I64 BE). @@ -123,22 +141,28 @@ func (c *Client) dialSession(magic byte) (fc *wire.FramedConn, peerWnd int, err // its per-stream receive window. Both are required. payload, err := fc.ReadFrame() if err != nil { - return nil, 0, err + return nil, err } if len(payload) < 1 || payload[0] != CtlSessionReady { - return nil, 0, fmt.Errorf("expected SessionReady, got %v", payload) + return nil, fmt.Errorf("expected SessionReady, got %v", payload) } r := wire.NewReader(payload[1:]) flags, ferr := r.VarInt() hubWnd, werr := r.VarInt() if ferr != nil || werr != nil || flags&FlagStreamFC == 0 || hubWnd <= 0 { - return nil, 0, fmt.Errorf("hub did not accept per-stream flow control (unsupported hub version?)") + return nil, fmt.Errorf("hub did not accept per-stream flow control (unsupported hub version?)") } if hubWnd > MaxStreamWindow { hubWnd = MaxStreamWindow } + + // The session is live: drop the establishment deadline. From here on + // liveness is the heartbeat's job (and WriteFrame bounds each write). + if err := conn.SetDeadline(time.Time{}); err != nil { + return nil, err + } ok = true - return fc, hubWnd, nil + return &session{fc: fc, peerWnd: hubWnd, heartbeat: flags&FlagWorkerHeartbeat != 0}, nil } // Start establishes the control session and registers all patterns. It returns @@ -149,20 +173,30 @@ func (c *Client) Start(ctx context.Context) error { } func (c *Client) connectControl(ctx context.Context) error { - fc, _, err := c.dialSession(MagicControl) + sess, err := c.dialSession(MagicControl) if err != nil { return fmt.Errorf("control connect: %w", err) } - c.registerAll(fc) + ctrl := &ctrlSession{fc: sess.fc} + ctrl.lastPong.Store(time.Now().UnixMilli()) + c.registerAll(sess.fc) c.mu.Lock() - c.ctrl = fc + c.ctrl = sess.fc c.mu.Unlock() log.Printf("control session established with %s", c.cfg.Server) - go c.serveControl(ctx, fc) - go c.pingLoop(ctx, fc) + go c.serveControl(ctx, ctrl) + go c.pingLoop(ctx, ctrl) return nil } +// ctrlSession tracks liveness for one control connection. A control session +// whose path dies silently must be detected, otherwise the hub keeps routing +// players to a session the client will never read from and nobody can connect. +type ctrlSession struct { + fc *wire.FramedConn + lastPong atomic.Int64 // unix ms of the most recent Pong +} + func (c *Client) registerAll(fc *wire.FramedConn) { for pattern := range c.mappings { msg := wire.NewWriter().U8(CtlRegister).String(pattern).Out() @@ -174,15 +208,15 @@ func (c *Client) registerAll(fc *wire.FramedConn) { } } -func (c *Client) serveControl(ctx context.Context, fc *wire.FramedConn) { +func (c *Client) serveControl(ctx context.Context, ctrl *ctrlSession) { for { - payload, err := fc.ReadFrame() + payload, err := ctrl.fc.ReadFrame() if err != nil { break } - c.dispatchControl(payload) + c.dispatchControl(ctrl, payload) } - _ = fc.Close() + _ = ctrl.fc.Close() if ctx.Err() != nil { return } @@ -200,7 +234,7 @@ func (c *Client) serveControl(ctx context.Context, fc *wire.FramedConn) { } } -func (c *Client) dispatchControl(payload []byte) { +func (c *Client) dispatchControl(ctrl *ctrlSession, payload []byte) { r := wire.NewReader(payload) t, err := r.U8() if err != nil { @@ -223,22 +257,33 @@ func (c *Client) dispatchControl(payload []byte) { port, _ := r.U16() go c.handleControlRequest(cid, pattern, ip, int(port)) case CtlPong: - // ignore + ctrl.lastPong.Store(time.Now().UnixMilli()) default: log.Printf("control: unknown message type %d", t) } } -func (c *Client) pingLoop(ctx context.Context, fc *wire.FramedConn) { - ticker := time.NewTicker(time.Duration(c.cfg.PingIntervalMs) * time.Millisecond) +// pingLoop keeps the control session alive and, crucially, verifies that the +// hub is still answering. A path that dies silently (no FIN/RST) would +// otherwise leave the read loop parked forever: the client would believe it is +// still registered while the hub routes players into the void. +func (c *Client) pingLoop(ctx context.Context, ctrl *ctrlSession) { + ticker := time.NewTicker(c.cfg.pingInterval()) defer ticker.Stop() + timeout := c.cfg.heartbeatTimeout() for { select { case <-ctx.Done(): return case <-ticker.C: + last := time.UnixMilli(ctrl.lastPong.Load()) + if time.Since(last) > timeout { + log.Printf("control session silent for %s; dropping it to force a reconnect", time.Since(last).Round(time.Second)) + _ = ctrl.fc.Close() // unblocks serveControl, which reconnects + return + } msg := wire.NewWriter().U8(CtlPing).I64(time.Now().UnixMilli()).Out() - if err := fc.WriteFrame(msg); err != nil { + if err := ctrl.fc.WriteFrame(msg); err != nil { return } } @@ -260,10 +305,16 @@ func (c *Client) handleControlRequest(cid []byte, pattern, ip string, port int) return } st := newStream(wc, sid, cid, mapping, ip, port) + // Register before SYN so inbound DATA can never race ahead of the table, + // and start the pumps before the (bounded) SYN write so a failed or slow + // SYN cannot strand a stream that nothing would ever tear down. wc.registerStream(sid, st) - wc.sendSyn(sid, cid) go st.writeLoop() go st.run() + if err := wc.sendSyn(sid, cid); err != nil { + log.Printf("stream %d: SYN failed: %v", sid, err) + st.teardown(false) + } } // WorkerConnCount reports the current number of open worker connections diff --git a/client/config.go b/client/config.go index 4f43b3e..0fb75be 100644 --- a/client/config.go +++ b/client/config.go @@ -5,6 +5,7 @@ import ( "fmt" "os" "strings" + "time" ) // Protocol constants (mirror of the Java Protocol class; see PROTOCOL.md). @@ -30,13 +31,27 @@ const ( MuxFin = 0x02 MuxRst = 0x03 MuxWnd = 0x04 + MuxPing = 0x05 + MuxPong = 0x06 + + // MuxCtlSid is the reserved stream id carrying connection-scoped mux frames + // (PING/PONG). Real streams are numbered from 1. + MuxCtlSid = 0 FrameError = 0x7F + // SaturationThreshold caps how many streams share one worker conn once the + // pool has grown to maxConn. Below maxConn the pool grows first (§7.1), so a + // single connection is never a shared point of failure for every player. SaturationThreshold = 8 // Session-establishment feature flags (trailing VarInt on the Rekey message). FlagStreamFC = 0x01 + // FlagWorkerHeartbeat enables mux-level PING/PONG on worker conns. Without + // it a worker conn whose path is silently blackholed (NAT/conntrack drop, + // firewall) is never detected: the read loop parks forever, the dead conn + // stays in the pool, and no player can be served until the client restarts. + FlagWorkerHeartbeat = 0x02 // Per-stream flow-control window bounds (bytes). The advertised window is the // receiver's promise of how much un-credited DATA it will buffer per stream. @@ -49,11 +64,49 @@ const ( DataChunkSize = 32 * 1024 ) +// Timeouts. Every tunnel socket is covered by one of these: without them a +// silently dropped path (no FIN/RST) leaves the client parked forever. +const ( + // HandshakeTimeout bounds session establishment end to end — the TCP dial, + // the Rekey write and the SessionReady read. A hub that accepts the socket + // but never answers must not park the caller (and, for the pool, every other + // player behind it) indefinitely. + HandshakeTimeout = 15 * time.Second + + // TCPKeepAlivePeriod asks the kernel to probe idle tunnel sockets, so a peer + // that becomes unreachable is detected even when no frames are in flight. + TCPKeepAlivePeriod = 30 * time.Second + + // MissedHeartbeats is how many ping intervals may pass with no reply before + // a session is declared dead and dropped. + MissedHeartbeats = 3 + + // MinPingIntervalMs floors the configured ping interval so the derived + // heartbeat timeout can never be short enough to cause spurious drops. + // Applied in LoadConfig, i.e. to configs that come from disk. + MinPingIntervalMs = 1000 +) + +// heartbeatTimeout is how long a session may go without a reply before it is +// considered dead, derived from the configured ping interval. +func (c *Config) heartbeatTimeout() time.Duration { + return c.pingInterval() * MissedHeartbeats +} + +// pingInterval is the configured heartbeat period. +func (c *Config) pingInterval() time.Duration { + return time.Duration(c.PingIntervalMs) * time.Millisecond +} + // Mapping routes a registered pattern to a real destination. type Mapping struct { Pattern string `json:"pattern"` Destination string `json:"destination"` ProxyProtocol bool `json:"proxyProtocol"` + // VelocitySecret, when non-empty, answers the destination's Velocity + // modern-forwarding login query (velocity:player_info) with this secret, + // forwarding the player's real IP, username and UUID (see velocity.go). + VelocitySecret string `json:"velocitySecret"` } // Config is the client configuration (PROTOCOL.md §9.2). @@ -91,6 +144,9 @@ func LoadConfig(path string) (*Config, error) { if c.PingIntervalMs <= 0 { c.PingIntervalMs = 20000 } + if c.PingIntervalMs < MinPingIntervalMs { + c.PingIntervalMs = MinPingIntervalMs + } if len(c.Mappings) == 0 { return nil, fmt.Errorf("at least one mapping is required") } diff --git a/client/pool_test.go b/client/pool_test.go new file mode 100644 index 0000000..a4a5152 --- /dev/null +++ b/client/pool_test.go @@ -0,0 +1,112 @@ +package client + +import ( + "net" + "sync" + "testing" + "time" +) + +// stalledHub accepts connections and then says nothing: it never answers the +// Rekey frame with SessionReady, and never closes. This models a hub with a +// wedged event loop, or a load balancer accepting on behalf of a dead backend. +func stalledHub(t *testing.T) string { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + var mu sync.Mutex + var held []net.Conn + t.Cleanup(func() { + _ = ln.Close() + mu.Lock() + for _, c := range held { + _ = c.Close() + } + mu.Unlock() + }) + go func() { + for { + c, err := ln.Accept() + if err != nil { + return + } + mu.Lock() + held = append(held, c) + mu.Unlock() + } + }() + return ln.Addr().String() +} + +// TestAllocateDoesNotWedgePoolOnStalledHub is the regression guard for the +// worst failure mode found in the stability audit: Allocate used to dial while +// holding the pool mutex, and the handshake read had no deadline. One +// unresponsive hub therefore parked every present and future allocation +// forever, so no player could be served again until the process restarted. +func TestAllocateDoesNotWedgePoolOnStalledHub(t *testing.T) { + c := New(&Config{ + Server: stalledHub(t), + PSK: "pool-test", + MaxConn: 8, + PingIntervalMs: 20000, + Mappings: []Mapping{{Pattern: "mc.local", Destination: "127.0.0.1:1"}}, + }) + + done := make(chan error, 2) + go func() { _, _, err := c.pool.Allocate(); done <- err }() + time.Sleep(200 * time.Millisecond) // let the first caller get into the dial + go func() { _, _, err := c.pool.Allocate(); done <- err }() + + // Both must give up on their own; neither may be stuck behind the other. + limit := time.After(HandshakeTimeout + 15*time.Second) + for i := 0; i < 2; i++ { + select { + case err := <-done: + if err == nil { + t.Fatal("Allocate succeeded against a hub that never answers") + } + case <-limit: + t.Fatalf("Allocate #%d never returned: the pool is wedged again", i+1) + } + } +} + +// TestAllocateSpreadsAcrossConns guards the allocation rule: the pool must fan +// out to maxConn before stacking streams, so a single worker connection is +// never the shared point of failure for every player. Seven players used to all +// land on one conn, which meant one dead TCP connection dropped everybody. +func TestAllocateSpreadsAcrossConns(t *testing.T) { + p := &WorkerPool{maxConn: 4} + p.cond = sync.NewCond(&p.mu) + + newConn := func() *WorkerConn { + return &WorkerConn{pool: p, streams: make(map[int]*Stream), nextSid: 1, done: make(chan struct{})} + } + p.conns = []*WorkerConn{newConn()} + p.conns[0].registerStream(1, &Stream{sid: 1}) + + // One conn holding a stream, pool below maxConn: growth is warranted. + _, bestCount := p.leastLoadedLocked() + if bestCount < StreamsBeforeGrowing { + t.Fatalf("a conn with %d stream(s) should trigger growth", bestCount) + } + + // Once the pool is at maxConn, growth stops and streams stack on the + // least-loaded conn instead. + for len(p.conns) < p.maxConn { + p.conns = append(p.conns, newConn()) + } + best, bestCount := p.leastLoadedLocked() + if bestCount != 0 { + t.Fatalf("expected an empty conn to be least-loaded, got %d streams", bestCount) + } + p.maybeGrowLocked(bestCount) + if p.dialing != 0 { + t.Fatalf("pool dialed past maxConn=%d", p.maxConn) + } + if best == nil { + t.Fatal("no conn selected") + } +} diff --git a/client/velocity.go b/client/velocity.go new file mode 100644 index 0000000..44e939c --- /dev/null +++ b/client/velocity.go @@ -0,0 +1,372 @@ +package client + +import ( + "crypto/hmac" + "crypto/md5" + "crypto/sha256" + "errors" + "sync" + "sync/atomic" + + "github.com/iceBear67/redapricot/client/wire" +) + +// Velocity "modern forwarding" support. +// +// A Paper backend configured with Velocity modern player-info forwarding runs +// in offline mode and instead trusts a signed login payload from its proxy: +// during the login phase it sends a Login Plugin Request on channel +// "velocity:player_info" and expects a Login Plugin Response whose data is an +// HMAC-SHA256 signature followed by the player's real address, UUID, username +// and profile properties. +// +// redapricot is a transparent tunnel, so that request would reach the vanilla +// player, who cannot answer it and gets kicked. When a mapping sets +// "velocitySecret", the stream answers on the player's behalf: it observes the +// player's Handshake and Login Start to learn the protocol version, username +// and UUID, swallows the backend's velocity:player_info request instead of +// forwarding it, and injects the signed response. Everything else — and +// everything after the exchange — is forwarded verbatim. On any traffic that +// does not look like a vanilla login (status pings, parse errors, oversized +// packets) the stream fails open into pure passthrough. +// +// The forwarded profile carries no properties (skin/cape textures): the tunnel +// never talks to Mojang, exactly like an offline-mode proxy. +const ( + velocityChannel = "velocity:player_info" + + // Forwarding payload versions (Velocity's VelocityConstants). We never use + // versions 2/3 (WITH_KEY): they exist only for 1.19–1.19.2 chat signing, + // and version 1 remains acceptable to every backend. + velocityVersionDefault = 1 + velocityVersionLazySession = 4 + + // Minecraft protocol versions at which the Login Start layout changes. + protocol1_19 = 759 // + optional signature key + protocol1_19_1 = 760 // + optional profile UUID (after the key) + protocol1_19_3 = 761 // key removed, optional UUID stays + protocol1_20_2 = 764 // UUID mandatory + + // Handshake intents that enter the login phase. + intentLogin = 2 + intentTransfer = 3 + + // Login-phase packet ids (stable across protocol versions). + loginC2SPluginResponse = 0x02 + loginS2CDisconnect = 0x00 + loginS2CEncryptionRequest = 0x01 + loginS2CSuccess = 0x02 + loginS2CSetCompression = 0x03 + loginS2CPluginRequest = 0x04 + + // Sniff-buffer caps. Login-phase packets are small; anything larger means + // this is not the exchange we are looking for. + maxC2SSniff = 8 << 10 + maxS2CSniff = 64 << 10 +) + +var errVelocitySniff = errors.New("velocity: connection does not follow the vanilla login flow") + +// velocityForwarder is the per-stream login interceptor. ObserveC2S is called +// from the worker read loop, ProcessS2C from the stream's destination-read +// goroutine; the mutex orders them, and passthrough short-circuits both once +// interception is over. +type velocityForwarder struct { + passthrough atomic.Bool // fully transparent, buffers empty: skip the mutex + + mu sync.Mutex + secret []byte + srcIP string + + // player -> server observation + c2sBuf []byte + c2sDone bool + handshakeParsed bool + protocol int + loginStartSeen bool + username string + uuid [16]byte + + // server -> player interception; done means the s2c side (and with it the + // whole interceptor) is finished. + s2cBuf []byte + done bool +} + +func newVelocityForwarder(secret, srcIP string) *velocityForwarder { + return &velocityForwarder{secret: []byte(secret), srcIP: srcIP} +} + +// Passthrough reports that interception is over and both directions may skip +// the forwarder entirely. +func (v *velocityForwarder) Passthrough() bool { return v.passthrough.Load() } + +// abortLocked gives up on interception: the stream becomes pure passthrough. +// s2cBuf is deliberately kept — ProcessS2C flushes it to the player. +func (v *velocityForwarder) abortLocked() { + v.done = true + v.c2sDone = true + v.c2sBuf = nil + if len(v.s2cBuf) == 0 { + v.passthrough.Store(true) + } +} + +// ObserveC2S watches player->server bytes (already being forwarded verbatim by +// the caller) until the Handshake and Login Start have been parsed. +func (v *velocityForwarder) ObserveC2S(data []byte) { + if v.passthrough.Load() { + return + } + v.mu.Lock() + defer v.mu.Unlock() + if v.c2sDone { + return + } + v.c2sBuf = append(v.c2sBuf, data...) + for !v.c2sDone { + _, body, rest, ok, err := nextPacket(v.c2sBuf, maxC2SSniff) + if err != nil { + v.abortLocked() + return + } + if !ok { + if len(v.c2sBuf) > maxC2SSniff { + v.abortLocked() + } + return + } + v.c2sBuf = rest + if err := v.observeC2SPacket(body); err != nil { + v.abortLocked() + return + } + } + v.c2sBuf = nil +} + +// observeC2SPacket handles one player packet: first the Handshake, then Login +// Start. Any deviation from the vanilla login flow is an error (→ fail open). +func (v *velocityForwarder) observeC2SPacket(body []byte) error { + r := wire.NewReader(body) + id, err := r.VarInt() + if err != nil || id != 0x00 { // Handshake and Login Start are both 0x00 + return errVelocitySniff + } + if !v.handshakeParsed { + proto, perr := r.VarInt() + _, aerr := r.String() // address + _, poerr := r.U16() // port + intent, ierr := r.VarInt() + if perr != nil || aerr != nil || poerr != nil || ierr != nil { + return errVelocitySniff + } + if intent != intentLogin && intent != intentTransfer { + return errVelocitySniff // status ping etc.: nothing to intercept + } + v.protocol = proto + v.handshakeParsed = true + return nil + } + return v.parseLoginStart(r) +} + +func (v *velocityForwarder) parseLoginStart(r *wire.Reader) error { + name, err := r.String() + if err != nil || len(name) == 0 || len(name) > 16 { + return errVelocitySniff + } + if v.protocol >= protocol1_19 && v.protocol < protocol1_19_3 { + // Optional chat-signing key: expiry + public key + signature. + hasKey, err := r.U8() + if err != nil { + return errVelocitySniff + } + if hasKey != 0 { + if _, err := r.I64(); err != nil { + return errVelocitySniff + } + for i := 0; i < 2; i++ { + n, err := r.VarInt() + if err != nil { + return errVelocitySniff + } + if _, err := r.Bytes(n); err != nil { + return errVelocitySniff + } + } + } + } + haveUUID := false + switch { + case v.protocol >= protocol1_20_2: + haveUUID = true + case v.protocol >= protocol1_19_1: + flag, err := r.U8() + if err != nil { + return errVelocitySniff + } + haveUUID = flag != 0 + } + if haveUUID { + b, err := r.Bytes(16) + if err != nil { + return errVelocitySniff + } + copy(v.uuid[:], b) + } else { + v.uuid = offlineUUID(name) + } + v.username = name + v.loginStartSeen = true + v.c2sDone = true + return nil +} + +// ProcessS2C consumes one chunk of server->player bytes. It returns the bytes +// to forward to the player and, once the velocity query has been answered, the +// Login Plugin Response to inject towards the server. Complete packets are +// forwarded as they parse; a trailing partial packet stays buffered until the +// next chunk. +func (v *velocityForwarder) ProcessS2C(data []byte) (forward, inject []byte) { + v.mu.Lock() + defer v.mu.Unlock() + if v.done { + // Interception ended from the c2s side while bytes sat buffered here. + if len(v.s2cBuf) > 0 { + forward = append(v.s2cBuf, data...) + v.s2cBuf = nil + v.passthrough.Store(true) + return forward, nil + } + v.passthrough.Store(true) + return data, nil + } + v.s2cBuf = append(v.s2cBuf, data...) +loop: + for { + raw, body, rest, ok, err := nextPacket(v.s2cBuf, maxS2CSniff) + if err != nil || (!ok && len(v.s2cBuf) > maxS2CSniff) { + v.abortLocked() // unconsumed bytes are flushed below + break + } + if !ok { + break // partial packet: wait for the next chunk + } + r := wire.NewReader(body) + id, err := r.VarInt() + if err != nil { + v.abortLocked() + break + } + switch id { + case loginS2CPluginRequest: + msgID, merr := r.VarInt() + channel, cerr := r.String() + if merr != nil || cerr != nil { + v.abortLocked() + break loop + } + if channel == velocityChannel { + if !v.loginStartSeen { + // Cannot answer without a parsed Login Start; let the + // request through — the backend will kick the player with + // its own clear message. + v.abortLocked() + break loop + } + inject = v.buildResponseLocked(msgID, r.Remaining()) + v.s2cBuf = rest // swallow the request: the player never sees it + v.done = true + break loop + } + // Another plugin channel (e.g. a mod handshake): the player + // answers it itself; forward and keep watching. + case loginS2CDisconnect, loginS2CEncryptionRequest, loginS2CSuccess, loginS2CSetCompression: + // Login phase is over (or turning encrypted/compressed) and no + // velocity query showed up: stop watching. + v.done = true + default: + // Cookie Request (0x05, 1.20.5+) or future packets: forward. + } + v.s2cBuf = rest + forward = append(forward, raw...) + if v.done { + break + } + } + if v.done { + forward = append(forward, v.s2cBuf...) + v.s2cBuf = nil + v.c2sBuf = nil + v.c2sDone = true + v.passthrough.Store(true) + } + return forward, inject +} + +// buildResponseLocked crafts the serverbound Login Plugin Response carrying the +// signed forwarding payload (mirrors Velocity's createForwardingData). +func (v *velocityForwarder) buildResponseLocked(msgID int, reqData []byte) []byte { + // The request data is the backend's maximum supported forwarding version + // (absent on very old backends → 1). + requested := velocityVersionDefault + if len(reqData) > 0 { + if n, err := wire.NewReader(reqData).VarInt(); err == nil { + requested = n + } + } + version := velocityVersionDefault + if requested >= velocityVersionLazySession && v.protocol >= protocol1_19_3 { + version = velocityVersionLazySession + } + payload := wire.NewWriter(). + VarInt(version). + String(v.srcIP). + Bytes(v.uuid[:]). // UUID = 16 raw bytes (two big-endian longs) + String(v.username). + VarInt(0). // profile properties + Out() + mac := hmac.New(sha256.New, v.secret) + mac.Write(payload) + body := wire.NewWriter(). + VarInt(loginC2SPluginResponse). + VarInt(msgID). + U8(1). // successful + Bytes(mac.Sum(nil)). + Bytes(payload). + Out() + return append(wire.AppendVarInt(nil, len(body)), body...) +} + +// nextPacket splits one length-prefixed Minecraft packet off buf. raw includes +// the length header, body is the packet payload, rest what follows. ok is +// false while the packet is still incomplete; err reports a malformed or +// oversized length header. +func nextPacket(buf []byte, max int) (raw, body, rest []byte, ok bool, err error) { + r := wire.NewReader(buf) + n, verr := r.VarInt() + if verr != nil { + if len(buf) >= wire.VarIntMaxBytes { + return nil, nil, buf, false, verr + } + return nil, nil, buf, false, nil // header not complete yet + } + if n <= 0 || n > max { + return nil, nil, buf, false, errVelocitySniff + } + hdr := len(buf) - len(r.Remaining()) + if len(buf) < hdr+n { + return nil, nil, buf, false, nil + } + return buf[:hdr+n], buf[hdr : hdr+n], buf[hdr+n:], true, nil +} + +// offlineUUID derives the offline-mode UUID for a username, identical to +// Java's UUID.nameUUIDFromBytes("OfflinePlayer:" + name): a v3 (MD5) UUID. +func offlineUUID(name string) [16]byte { + sum := md5.Sum([]byte("OfflinePlayer:" + name)) + sum[6] = sum[6]&0x0f | 0x30 // version 3 + sum[8] = sum[8]&0x3f | 0x80 // IETF variant + return sum +} diff --git a/client/velocity_test.go b/client/velocity_test.go new file mode 100644 index 0000000..268f0bc --- /dev/null +++ b/client/velocity_test.go @@ -0,0 +1,348 @@ +package client + +import ( + "bytes" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "testing" + + "github.com/iceBear67/redapricot/client/wire" +) + +// ---- packet builders (player/backend side) ---- + +func mkPacket(body []byte) []byte { + return append(wire.AppendVarInt(nil, len(body)), body...) +} + +func mkLoginStart(proto int, name string, uuid []byte, keyData bool) []byte { + w := wire.NewWriter().VarInt(0x00).String(name) + if proto >= protocol1_19 && proto < protocol1_19_3 { + if keyData { + w.U8(1).I64(1234567890) + pub := bytes.Repeat([]byte{0xAA}, 33) + sig := bytes.Repeat([]byte{0xBB}, 17) + w.VarInt(len(pub)).Bytes(pub).VarInt(len(sig)).Bytes(sig) + } else { + w.U8(0) + } + } + switch { + case proto >= protocol1_20_2: + w.Bytes(uuid) + case proto >= protocol1_19_1: + if uuid != nil { + w.U8(1).Bytes(uuid) + } else { + w.U8(0) + } + } + return mkPacket(w.Out()) +} + +func mkPluginRequest(msgID int, channel string, data []byte) []byte { + body := wire.NewWriter().VarInt(loginS2CPluginRequest).VarInt(msgID).String(channel).Bytes(data).Out() + return mkPacket(body) +} + +// feedLogin drives a full player login prologue through ObserveC2S in n-byte +// chunks. +func feedLogin(v *velocityForwarder, proto int, intent int, loginStart []byte, chunk int) { + stream := wire.BuildHandshake(proto, "mc.example.com", 25565, intent) + if loginStart != nil { + stream = append(stream, loginStart...) + } + for len(stream) > 0 { + n := chunk + if n > len(stream) { + n = len(stream) + } + v.ObserveC2S(stream[:n]) + stream = stream[n:] + } +} + +// parseResponse validates the injected Login Plugin Response and returns the +// echoed message id and the signed forwarding payload. +func parseResponse(t *testing.T, secret string, inject []byte) (msgID int, payload *wire.Reader) { + t.Helper() + r := wire.NewReader(inject) + plen, err := r.VarInt() + if err != nil || plen != len(r.Remaining()) { + t.Fatalf("bad response length prefix: %v (declared %d, have %d)", err, plen, len(r.Remaining())) + } + id, _ := r.VarInt() + if id != loginC2SPluginResponse { + t.Fatalf("response packet id = %#x, want 0x02", id) + } + msgID, _ = r.VarInt() + ok, _ := r.U8() + if ok != 1 { + t.Fatalf("response not marked successful") + } + sig, err := r.Bytes(32) + if err != nil { + t.Fatalf("response missing signature: %v", err) + } + data := r.Remaining() + mac := hmac.New(sha256.New, []byte(secret)) + mac.Write(data) + if !hmac.Equal(sig, mac.Sum(nil)) { + t.Fatalf("forwarding payload signature does not verify") + } + return msgID, wire.NewReader(data) +} + +func assertPayload(t *testing.T, r *wire.Reader, version int, ip, name, uuidHex string) { + t.Helper() + gotVer, _ := r.VarInt() + if gotVer != version { + t.Fatalf("forwarding version = %d, want %d", gotVer, version) + } + gotIP, _ := r.String() + if gotIP != ip { + t.Fatalf("forwarded address = %q, want %q", gotIP, ip) + } + gotUUID, err := r.Bytes(16) + if err != nil { + t.Fatalf("payload missing uuid: %v", err) + } + if hex.EncodeToString(gotUUID) != uuidHex { + t.Fatalf("forwarded uuid = %x, want %s", gotUUID, uuidHex) + } + gotName, _ := r.String() + if gotName != name { + t.Fatalf("forwarded username = %q, want %q", gotName, name) + } + props, err := r.VarInt() + if err != nil || props != 0 { + t.Fatalf("forwarded properties = %d (%v), want 0", props, err) + } + if len(r.Remaining()) != 0 { + t.Fatalf("trailing bytes in forwarding payload: %x", r.Remaining()) + } +} + +// ---- tests ---- + +const testSecret = "unit-secret" + +func TestVelocityInterceptModern(t *testing.T) { + uuid, _ := hex.DecodeString("00112233445566778899aabbccddeeff") + v := newVelocityForwarder(testSecret, "203.0.113.7") + feedLogin(v, 767, intentLogin, mkLoginStart(767, "icybear", uuid, false), 1) + + // Backend query, requesting up to forwarding version 4, fed byte by byte: + // nothing may reach the player, and the response appears with the last byte. + req := mkPluginRequest(99, velocityChannel, []byte{0x04}) + var inject []byte + for i, b := range req { + fwd, inj := v.ProcessS2C([]byte{b}) + if len(fwd) != 0 { + t.Fatalf("byte %d: request leaked to the player: %x", i, fwd) + } + if inj != nil { + inject = inj + } + } + if inject == nil { + t.Fatalf("no response was injected") + } + msgID, payload := parseResponse(t, testSecret, inject) + if msgID != 99 { + t.Fatalf("echoed message id = %d, want 99", msgID) + } + assertPayload(t, payload, velocityVersionLazySession, "203.0.113.7", "icybear", + "00112233445566778899aabbccddeeff") + if !v.Passthrough() { + t.Fatalf("interceptor should be passthrough after answering") + } + garbage := []byte{0xde, 0xad, 0xbe, 0xef} + if fwd, inj := v.ProcessS2C(garbage); !bytes.Equal(fwd, garbage) || inj != nil { + t.Fatalf("post-login bytes not passed through verbatim") + } +} + +func TestVelocityNegativeMessageID(t *testing.T) { + // Paper picks the message id with ThreadLocalRandom.nextInt(): it is + // negative half the time and must be echoed bit-exactly. + uuid, _ := hex.DecodeString("00112233445566778899aabbccddeeff") + v := newVelocityForwarder(testSecret, "198.51.100.1") + feedLogin(v, 767, intentLogin, mkLoginStart(767, "neg", uuid, false), 64) + _, inject := v.ProcessS2C(mkPluginRequest(-123456, velocityChannel, []byte{0x04})) + if inject == nil { + t.Fatalf("no response was injected") + } + msgID, _ := parseResponse(t, testSecret, inject) + if !bytes.Equal(wire.AppendVarInt(nil, msgID), wire.AppendVarInt(nil, -123456)) { + t.Fatalf("negative message id not echoed bit-exactly (got %d)", msgID) + } +} + +func TestVelocityOfflineUUIDAndV1(t *testing.T) { + // 1.18.2 player: no UUID in Login Start -> Java's offline UUID; an old + // backend requesting version 1 gets version 1. + v := newVelocityForwarder(testSecret, "192.0.2.9") + feedLogin(v, 758, intentLogin, mkLoginStart(758, "Notch", nil, false), 3) + _, inject := v.ProcessS2C(mkPluginRequest(7, velocityChannel, []byte{0x01})) + if inject == nil { + t.Fatalf("no response was injected") + } + _, payload := parseResponse(t, testSecret, inject) + // UUID.nameUUIDFromBytes("OfflinePlayer:Notch".getBytes(UTF_8)). + assertPayload(t, payload, velocityVersionDefault, "192.0.2.9", "Notch", + "b50ad385829d3141a2167e7d7539ba7f") +} + +func TestVelocityVersionGating(t *testing.T) { + // A modern backend (requests 4) behind a pre-1.19.3 player must get v1. + v := newVelocityForwarder(testSecret, "192.0.2.9") + feedLogin(v, 758, intentLogin, mkLoginStart(758, "Old", nil, false), 5) + _, inject := v.ProcessS2C(mkPluginRequest(1, velocityChannel, []byte{0x04})) + _, payload := parseResponse(t, testSecret, inject) + ver, _ := payload.VarInt() + if ver != velocityVersionDefault { + t.Fatalf("version = %d, want 1 for a pre-1.19.3 player", ver) + } + // An empty request (very old backend) also means v1. + v2 := newVelocityForwarder(testSecret, "192.0.2.9") + feedLogin(v2, 767, intentLogin, mkLoginStart(767, "New", make([]byte, 16), false), 5) + _, inject2 := v2.ProcessS2C(mkPluginRequest(1, velocityChannel, nil)) + _, payload2 := parseResponse(t, testSecret, inject2) + ver2, _ := payload2.VarInt() + if ver2 != velocityVersionDefault { + t.Fatalf("version = %d, want 1 for an empty version request", ver2) + } +} + +func TestVelocityLoginStartVariants(t *testing.T) { + uuid, _ := hex.DecodeString("ffeeddccbbaa99887766554433221100") + cases := []struct { + name string + proto int + start []byte + uuidHex string + }{ + {"1.19 with key, no uuid", 759, mkLoginStart(759, "Notch", nil, true), + "b50ad385829d3141a2167e7d7539ba7f"}, + {"1.19.1 with key and uuid", 760, mkLoginStart(760, "Notch", uuid, true), + "ffeeddccbbaa99887766554433221100"}, + {"1.19.3 optional uuid present", 761, mkLoginStart(761, "Notch", uuid, false), + "ffeeddccbbaa99887766554433221100"}, + {"1.19.3 optional uuid absent", 761, mkLoginStart(761, "Notch", nil, false), + "b50ad385829d3141a2167e7d7539ba7f"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + v := newVelocityForwarder(testSecret, "192.0.2.1") + feedLogin(v, tc.proto, intentLogin, tc.start, 2) + _, inject := v.ProcessS2C(mkPluginRequest(3, velocityChannel, []byte{0x01})) + if inject == nil { + t.Fatalf("no response was injected") + } + _, payload := parseResponse(t, testSecret, inject) + assertPayload(t, payload, velocityVersionDefault, "192.0.2.1", "Notch", tc.uuidHex) + }) + } +} + +func TestVelocityStatusPassthrough(t *testing.T) { + v := newVelocityForwarder(testSecret, "192.0.2.1") + feedLogin(v, 767, 1 /* status */, nil, 100) + if !v.Passthrough() { + t.Fatalf("status intent should turn the stream transparent") + } + data := []byte("not a minecraft packet at all") + if fwd, inj := v.ProcessS2C(data); !bytes.Equal(fwd, data) || inj != nil { + t.Fatalf("status traffic must pass through untouched") + } +} + +func TestVelocityOtherChannelForwarded(t *testing.T) { + uuid := make([]byte, 16) + v := newVelocityForwarder(testSecret, "192.0.2.1") + feedLogin(v, 767, intentLogin, mkLoginStart(767, "modded", uuid, false), 50) + + // A modded-handshake query and the velocity query coalesced in one chunk: + // the first must reach the player, the second must not. + other := mkPluginRequest(1, "fml:loginwrapper", []byte{0x00, 0x01}) + velo := mkPluginRequest(2, velocityChannel, []byte{0x04}) + fwd, inject := v.ProcessS2C(append(append([]byte{}, other...), velo...)) + if !bytes.Equal(fwd, other) { + t.Fatalf("non-velocity query not forwarded verbatim:\n got %x\nwant %x", fwd, other) + } + if inject == nil { + t.Fatalf("velocity query in the same chunk was not answered") + } +} + +func TestVelocityNoQueryLoginSuccess(t *testing.T) { + // Backend without velocity forwarding: the first login packet ends + // interception and everything flows verbatim. + uuid := make([]byte, 16) + v := newVelocityForwarder(testSecret, "192.0.2.1") + feedLogin(v, 767, intentLogin, mkLoginStart(767, "plain", uuid, false), 50) + + success := mkPacket(wire.NewWriter().VarInt(loginS2CSuccess).Bytes(uuid).String("plain").VarInt(0).Out()) + tail := []byte("compressed gibberish after login") + fwd, inject := v.ProcessS2C(append(append([]byte{}, success...), tail...)) + if inject != nil { + t.Fatalf("nothing should be injected without a velocity query") + } + want := append(append([]byte{}, success...), tail...) + if !bytes.Equal(fwd, want) { + t.Fatalf("login success not flushed verbatim") + } + if !v.Passthrough() { + t.Fatalf("interceptor should be passthrough after Login Success") + } +} + +func TestVelocityQueryBeforeLoginStartFailsOpen(t *testing.T) { + // A query arriving before the Login Start was observed cannot be answered: + // it must reach the player unmodified (who will then be kicked by the + // backend with its own message). + v := newVelocityForwarder(testSecret, "192.0.2.1") + feedLogin(v, 767, intentLogin, nil, 50) // handshake only + req := mkPluginRequest(5, velocityChannel, []byte{0x04}) + fwd, inject := v.ProcessS2C(req) + if inject != nil { + t.Fatalf("must not answer without a Login Start") + } + if !bytes.Equal(fwd, req) { + t.Fatalf("unanswerable query not passed through") + } + if !v.Passthrough() { + t.Fatalf("interceptor should fail open") + } +} + +func TestVelocityOversizedC2SFailsOpen(t *testing.T) { + v := newVelocityForwarder(testSecret, "192.0.2.1") + // A declared c2s packet length beyond the sniff cap aborts interception. + v.ObserveC2S(wire.AppendVarInt(nil, maxC2SSniff+1)) + if !v.Passthrough() { + t.Fatalf("oversized login packet should turn the stream transparent") + } +} + +func TestVelocityC2SAbortFlushesBufferedS2C(t *testing.T) { + v := newVelocityForwarder(testSecret, "192.0.2.1") + feedLogin(v, 767, intentLogin, nil, 50) // handshake only; login pending + req := mkPluginRequest(5, velocityChannel, []byte{0x04}) + half := len(req) / 2 + if fwd, _ := v.ProcessS2C(req[:half]); len(fwd) != 0 { + t.Fatalf("partial packet must stay buffered") + } + // The c2s side now aborts (e.g. unparseable player bytes) while s2c bytes + // sit buffered: they must not be lost. + v.ObserveC2S(wire.AppendVarInt(nil, maxC2SSniff+1)) + fwd, inject := v.ProcessS2C(req[half:]) + if inject != nil { + t.Fatalf("aborted interceptor must not inject") + } + if !bytes.Equal(fwd, req) { + t.Fatalf("buffered s2c bytes lost on abort:\n got %x\nwant %x", fwd, req) + } +} diff --git a/client/wire/framedconn.go b/client/wire/framedconn.go index d309cfa..6c5182f 100644 --- a/client/wire/framedconn.go +++ b/client/wire/framedconn.go @@ -6,6 +6,7 @@ import ( "io" "net" "sync" + "time" "golang.org/x/crypto/chacha20" ) @@ -13,7 +14,19 @@ import ( // MaxFrame is the maximum decrypted frame payload size (1 MiB). const MaxFrame = 1 << 20 -var errFrameTooBig = errors.New("wire: frame exceeds max size") +// WriteTimeout bounds a single frame write. A peer that stops reading must not +// be able to park every stream on the connection inside WriteFrame forever: the +// write mutex is held for the whole socket write, so one stalled write would +// otherwise wedge the entire multiplexed connection. +const WriteTimeout = 30 * time.Second + +var ( + errFrameTooBig = errors.New("wire: frame exceeds max size") + // ErrBroken is returned once a write has failed. The ChaCha20 keystream has + // already advanced (and the socket may hold a partial frame), so the + // connection can never be resynchronized and is closed for good. + ErrBroken = errors.New("wire: connection is broken") +) // FramedConn is the encrypted, length-prefixed frame transport (PROTOCOL.md §3.1). // The VarInt length prefix is plaintext; the payload is ChaCha20-encrypted with a @@ -24,7 +37,9 @@ type FramedConn struct { r *bufio.Reader in *chacha20.Cipher out *chacha20.Cipher - wmu sync.Mutex + + wmu sync.Mutex + broken bool } func NewFramedConn(conn net.Conn, in, out *chacha20.Cipher) *FramedConn { @@ -62,15 +77,29 @@ func (f *FramedConn) ReadFrame() ([]byte, error) { } // WriteFrame encrypts and sends one frame payload. Safe for concurrent callers. +// +// The write is bounded by WriteTimeout. On any write error the connection is +// marked broken and closed, which unblocks the reader so the owner can tear the +// session down instead of leaving every stream parked on the write mutex. func (f *FramedConn) WriteFrame(payload []byte) error { f.wmu.Lock() defer f.wmu.Unlock() + if f.broken { + return ErrBroken + } ct := make([]byte, len(payload)) f.out.XORKeyStream(ct, payload) out := AppendVarInt(make([]byte, 0, VarIntMaxBytes+len(ct)), len(ct)) out = append(out, ct...) + _ = f.conn.SetWriteDeadline(time.Now().Add(WriteTimeout)) _, err := f.conn.Write(out) - return err + if err != nil { + f.broken = true + _ = f.conn.Close() + return err + } + _ = f.conn.SetWriteDeadline(time.Time{}) + return nil } func (f *FramedConn) Close() error { return f.conn.Close() } diff --git a/client/worker.go b/client/worker.go index 66e4754..1464d8a 100644 --- a/client/worker.go +++ b/client/worker.go @@ -4,30 +4,87 @@ import ( "log" "net" "sync" + "sync/atomic" "time" "github.com/iceBear67/redapricot/client/wire" ) +// StreamsBeforeGrowing is how many streams a worker conn may carry before the +// pool starts opening another one. Set to 1 so the pool fans out to maxConn +// under load *before* stacking streams: concentrating every player on a single +// TCP connection makes that connection a shared point of failure, which is +// exactly how a whole server's worth of players used to drop at once. +const StreamsBeforeGrowing = 1 + // WorkerPool manages up to maxConn worker connections and allocates streams // using the least-loaded strategy (PROTOCOL.md §7.1). type WorkerPool struct { client *Client maxConn int - mu sync.Mutex - conns []*WorkerConn + mu sync.Mutex + cond *sync.Cond + conns []*WorkerConn + dialing int // dials currently in flight (foreground + background) + dialGen uint64 + dialErr error // most recent dial failure } func newWorkerPool(c *Client, maxConn int) *WorkerPool { - return &WorkerPool{client: c, maxConn: maxConn} + p := &WorkerPool{client: c, maxConn: maxConn} + p.cond = sync.NewCond(&p.mu) + return p } // Allocate returns a worker conn and a fresh stream id to place a new stream on. +// +// A dial is never performed while holding p.mu: session establishment involves +// network I/O, and holding the pool lock across it would park every other +// player behind one unresponsive hub. When the pool is empty exactly one caller +// dials and the rest wait on the condition variable; when the pool is merely +// below maxConn, growth happens in the background and the caller is served +// immediately by an existing conn. func (p *WorkerPool) Allocate() (*WorkerConn, int, error) { p.mu.Lock() - defer p.mu.Unlock() + for { + best, bestCount := p.leastLoadedLocked() + if best != nil { + p.maybeGrowLocked(bestCount) + p.mu.Unlock() + return best, best.newSid(), nil + } + if p.dialing > 0 { + // Someone is already dialing the first conn; wait for it rather + // than piling up redundant connections. + gen := p.dialGen + p.cond.Wait() + if len(p.conns) == 0 && p.dialGen != gen && p.dialErr != nil { + err := p.dialErr + p.mu.Unlock() + return nil, 0, err + } + continue + } + p.dialing++ + p.mu.Unlock() + wc, err := p.dialWorker() + p.mu.Lock() + p.dialing-- + p.dialGen++ + p.dialErr = err + if err != nil { + p.cond.Broadcast() + p.mu.Unlock() + return nil, 0, err + } + p.conns = append(p.conns, wc) + p.cond.Broadcast() + } +} +// leastLoadedLocked returns the worker conn carrying the fewest streams. +func (p *WorkerPool) leastLoadedLocked() (*WorkerConn, int) { var best *WorkerConn bestCount := 0 for _, wc := range p.conns { @@ -37,39 +94,74 @@ func (p *WorkerPool) Allocate() (*WorkerConn, int, error) { bestCount = n } } - - needNew := best == nil || (bestCount > SaturationThreshold && len(p.conns) < p.maxConn) - if needNew { - wc, err := p.dialWorker() - if err != nil { - if best == nil { - return nil, 0, err - } - log.Printf("worker dial failed, reusing existing conn: %v", err) - } else { - p.conns = append(p.conns, wc) - best = wc - } - } - return best, best.newSid(), nil + return best, bestCount } +// maybeGrowLocked opens one more worker conn in the background when the pool is +// below maxConn and the least-loaded conn is already carrying streams. The +// caller does not wait for it: it keeps using the conn it already has, and the +// new one picks up subsequent players. +func (p *WorkerPool) maybeGrowLocked(bestCount int) { + if bestCount < StreamsBeforeGrowing { + return + } + if len(p.conns)+p.dialing >= p.maxConn { + if bestCount > SaturationThreshold { + log.Printf("worker pool at maxConn=%d with %d streams on the least-loaded conn", p.maxConn, bestCount) + } + return + } + p.dialing++ + go func() { + wc, err := p.dialWorker() + var surplus *WorkerConn + p.mu.Lock() + p.dialing-- + p.dialGen++ + p.dialErr = err + switch { + case err != nil: + case len(p.conns) < p.maxConn: + p.conns = append(p.conns, wc) + default: + surplus = wc // raced with another dial + } + p.cond.Broadcast() + p.mu.Unlock() + if err != nil { + log.Printf("worker pool: background dial failed: %v", err) + } + if surplus != nil { + _ = surplus.fc.Close() + } + }() +} + +// dialWorker establishes one worker conn. It must be called without p.mu held. func (p *WorkerPool) dialWorker() (*WorkerConn, error) { - fc, peerWnd, err := p.client.dialSession(MagicWorker) + sess, err := p.client.dialSession(MagicWorker) if err != nil { return nil, err } wc := &WorkerConn{ pool: p, - fc: fc, - sendWndInit: peerWnd, + fc: sess.fc, + sendWndInit: sess.peerWnd, recvWndInit: p.client.streamWnd, streams: make(map[int]*Stream), nextSid: 1, + done: make(chan struct{}), } + wc.lastPong.Store(time.Now().UnixMilli()) go wc.readLoop() - log.Printf("opened worker conn (#%d in pool, send window %d, recv window %d)", - len(p.conns)+1, wc.sendWndInit, wc.recvWndInit) + if sess.heartbeat { + go wc.heartbeatLoop(p.client.cfg.pingInterval(), p.client.cfg.heartbeatTimeout()) + } else { + log.Printf("worker conn: hub does not support the mux heartbeat; " + + "a silently dropped path will only be caught by TCP keepalive") + } + log.Printf("opened worker conn (send window %d, recv window %d, heartbeat %v)", + wc.sendWndInit, wc.recvWndInit, sess.heartbeat) return wc, nil } @@ -85,6 +177,9 @@ func (p *WorkerPool) remove(wc *WorkerConn) { for i, c := range p.conns { if c == wc { p.conns = append(p.conns[:i], p.conns[i+1:]...) + // A waiter parked on an empty pool must re-evaluate: it may now + // need to dial rather than keep waiting for this conn. + p.cond.Broadcast() return } } @@ -107,11 +202,41 @@ type WorkerConn struct { sendWndInit int // hub's advertised per-stream receive window (our send budget) recvWndInit int // our advertised per-stream receive window (bounds each recv queue) + done chan struct{} // closed when readLoop exits + lastPong atomic.Int64 // unix ms of the most recent PONG + mu sync.Mutex streams map[int]*Stream nextSid int } +// heartbeatLoop proves the worker conn is still carrying frames end to end. TCP +// alone cannot tell us: a middlebox that drops an established flow (conntrack +// expiry, firewall state loss) sends no FIN or RST, so the read loop would park +// forever, the dead conn would stay in the pool, and every player routed to it +// would silently fail until the process restarted. +func (wc *WorkerConn) heartbeatLoop(interval, timeout time.Duration) { + ticker := time.NewTicker(interval) + defer ticker.Stop() + for { + select { + case <-wc.done: + return + case <-ticker.C: + if silent := time.Since(time.UnixMilli(wc.lastPong.Load())); silent > timeout { + log.Printf("worker conn silent for %s; dropping it and its %d stream(s)", + silent.Round(time.Second), wc.streamCount()) + _ = wc.fc.Close() // readLoop unblocks and tears everything down + return + } + msg := wire.NewWriter().U8(MuxPing).VarInt(MuxCtlSid).I64(time.Now().UnixMilli()).Out() + if err := wc.fc.WriteFrame(msg); err != nil { + return + } + } + } +} + func (wc *WorkerConn) streamCount() int { wc.mu.Lock() defer wc.mu.Unlock() @@ -184,11 +309,17 @@ func (wc *WorkerConn) readLoop() { if st := wc.removeStream(sid); st != nil { st.teardown(false) } + case MuxPing: + nonce, _ := r.I64() + _ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxPong).VarInt(MuxCtlSid).I64(nonce).Out()) + case MuxPong: + wc.lastPong.Store(time.Now().UnixMilli()) default: log.Printf("worker: unknown mux type %d", ftype) } } // Connection lost: tear down all streams and drop from pool. + close(wc.done) wc.pool.remove(wc) wc.mu.Lock() streams := make([]*Stream, 0, len(wc.streams)) @@ -202,8 +333,8 @@ func (wc *WorkerConn) readLoop() { } } -func (wc *WorkerConn) sendSyn(sid int, cid []byte) { - _ = wc.fc.WriteFrame(wire.NewWriter().U8(MuxSyn).VarInt(sid).Bytes(cid).Out()) +func (wc *WorkerConn) sendSyn(sid int, cid []byte) error { + return wc.fc.WriteFrame(wire.NewWriter().U8(MuxSyn).VarInt(sid).Bytes(cid).Out()) } func (wc *WorkerConn) sendData(sid int, data []byte) error { @@ -235,6 +366,7 @@ type Stream struct { mapping Mapping srcIP string srcPort int + vel *velocityForwarder // non-nil when the mapping sets velocitySecret mu sync.Mutex cond *sync.Cond @@ -242,14 +374,26 @@ type Stream struct { connected bool closed bool finPending bool // hub sent FIN; close the destination once the queue drains - q [][]byte // hub -> destination, waiting for writeLoop - qBytes int - sendWnd int // flow control: budget for destination -> hub DATA - consumed int // flow control: drained bytes not yet credited back to the hub + q []qentry // hub/local -> destination, waiting for writeLoop + qBytes int // hub bytes only: bounds the peer against its window + sendWnd int // flow control: budget for destination -> hub DATA + consumed int // flow control: drained bytes not yet credited back to the hub +} + +// qentry is one queued write towards the destination. Only hub-originated +// entries take part in flow control; locally injected bytes (the velocity +// login response) are neither counted against the hub's window nor credited +// back when drained. +type qentry struct { + data []byte + fromHub bool } func newStream(wc *WorkerConn, sid int, cid []byte, m Mapping, ip string, port int) *Stream { s := &Stream{wc: wc, sid: sid, cid: cid, mapping: m, srcIP: ip, srcPort: port, sendWnd: wc.sendWndInit} + if m.VelocitySecret != "" { + s.vel = newVelocityForwarder(m.VelocitySecret, ip) + } s.cond = sync.NewCond(&s.mu) return s } @@ -285,6 +429,12 @@ func (s *Stream) run() { } s.dest = dest s.connected = true + if s.finPending { + // The hub FIN'd while we were still dialing, so gracefulFin could not + // arm the drain deadline (there was no destination yet). Arm it now, + // otherwise writeLoop can block on an unresponsive destination forever. + _ = dest.SetWriteDeadline(time.Now().Add(finDrainTimeout)) + } s.cond.Broadcast() // wake writeLoop: queued hub bytes can flow now s.mu.Unlock() @@ -293,10 +443,15 @@ func (s *Stream) run() { for { n, err := dest.Read(buf) if n > 0 { - if !s.acquireSendWnd(n) { - break + chunk := buf[:n] + if s.vel != nil && !s.vel.Passthrough() { + fwd, inject := s.vel.ProcessS2C(chunk) + if len(inject) > 0 { + s.injectToDest(inject) + } + chunk = fwd } - if werr := s.wc.sendData(s.sid, buf[:n]); werr != nil { + if !s.sendToHub(chunk) { break } } @@ -307,6 +462,26 @@ func (s *Stream) run() { s.teardown(true) } +// sendToHub forwards destination bytes to the hub in DATA frames of at most +// DataChunkSize, honoring the stream send window. Returns false once the +// stream closed or the worker conn failed. +func (s *Stream) sendToHub(data []byte) bool { + for len(data) > 0 { + n := len(data) + if n > DataChunkSize { + n = DataChunkSize + } + if !s.acquireSendWnd(n) { + return false + } + if err := s.wc.sendData(s.sid, data[:n]); err != nil { + return false + } + data = data[n:] + } + return true +} + // writeLoop is the only writer to the destination. It drains the receive queue, // credits the hub as bytes land on the destination socket, and performs the // deferred graceful close when a FIN arrived with data still queued. @@ -325,17 +500,21 @@ func (s *Stream) writeLoop() { s.teardown(false) return } - data := s.q[0] + e := s.q[0] s.q = s.q[1:] - s.qBytes -= len(data) + if e.fromHub { + s.qBytes -= len(e.data) + } dest := s.dest s.mu.Unlock() - if _, err := dest.Write(data); err != nil { + if _, err := dest.Write(e.data); err != nil { s.teardown(true) return } - s.credit(len(data)) + if e.fromHub { + s.credit(len(e.data)) + } } } @@ -343,6 +522,9 @@ func (s *Stream) writeLoop() { // readLoop; it never blocks — a peer that exceeds the advertised window is a // protocol violator and gets the stream reset. func (s *Stream) deliverFromHub(data []byte) { + if s.vel != nil { + s.vel.ObserveC2S(data) // observation only; bytes still forwarded verbatim + } s.mu.Lock() if s.closed || s.finPending { s.mu.Unlock() @@ -356,12 +538,25 @@ func (s *Stream) deliverFromHub(data []byte) { s.teardown(false) return } - s.q = append(s.q, data) + s.q = append(s.q, qentry{data: data, fromHub: true}) s.qBytes += len(data) s.cond.Broadcast() s.mu.Unlock() } +// injectToDest queues locally generated bytes (the velocity login response) +// for the destination, outside flow-control accounting. +func (s *Stream) injectToDest(data []byte) { + s.mu.Lock() + if s.closed || s.finPending { + s.mu.Unlock() + return + } + s.q = append(s.q, qentry{data: data}) + s.cond.Broadcast() + s.mu.Unlock() +} + // acquireSendWnd blocks until the stream may send n more bytes to the hub. // Returns false if the stream closed while waiting. func (s *Stream) acquireSendWnd(n int) bool { diff --git a/e2e/e2e_test.go b/e2e/e2e_test.go index 1fd3142..f70e91a 100644 --- a/e2e/e2e_test.go +++ b/e2e/e2e_test.go @@ -14,12 +14,19 @@ import ( // startClient builds and starts an in-process client against the hub, with the // given mappings, returning the running client. func startClient(t *testing.T, hubAddr, psk string, maxConn int, mappings []client.Mapping) *client.Client { + t.Helper() + return startClientWithPing(t, hubAddr, psk, maxConn, 20000, mappings) +} + +// startClientWithPing is startClient with an explicit heartbeat interval, for +// tests that need liveness detection to trigger quickly. +func startClientWithPing(t *testing.T, hubAddr, psk string, maxConn, pingMs int, mappings []client.Mapping) *client.Client { t.Helper() cfg := &client.Config{ Server: hubAddr, PSK: psk, MaxConn: maxConn, - PingIntervalMs: 20000, + PingIntervalMs: pingMs, Mappings: mappings, } c := client.New(cfg) diff --git a/e2e/resilience_test.go b/e2e/resilience_test.go new file mode 100644 index 0000000..98ead77 --- /dev/null +++ b/e2e/resilience_test.go @@ -0,0 +1,154 @@ +package e2e + +import ( + "fmt" + "io" + "net" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/iceBear67/redapricot/client" + "github.com/iceBear67/redapricot/client/wire" +) + +// blackholeRelay forwards TCP to the hub until it is switched off, after which +// already-established pairs silently stop carrying bytes while their sockets +// stay open. That is what a stateful middlebox looks like when it forgets a +// flow: conntrack expiry, a firewall reload, or a cloud LB idle timeout. No FIN +// and no RST ever reach either end, so nothing below the application layer can +// notice. +// Only flows that already existed when the switch is thrown go dark; new +// connections are carried normally, exactly as when a middlebox forgets +// established state but keeps forwarding fresh traffic. +type blackholeRelay struct { + addr string + backend string + gen atomic.Uint64 + mu sync.Mutex + held []net.Conn +} + +func newBlackholeRelay(t *testing.T, backend string) *blackholeRelay { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("relay listen: %v", err) + } + r := &blackholeRelay{addr: ln.Addr().String(), backend: backend} + t.Cleanup(func() { + _ = ln.Close() + r.mu.Lock() + for _, c := range r.held { + _ = c.Close() + } + r.mu.Unlock() + }) + go func() { + for { + cli, err := ln.Accept() + if err != nil { + return + } + go r.handle(cli) + } + }() + return r +} + +func (r *blackholeRelay) handle(cli net.Conn) { + born := r.gen.Load() + up, err := net.Dial("tcp", r.backend) + if err != nil { + _ = cli.Close() + return + } + r.mu.Lock() + r.held = append(r.held, cli, up) + r.mu.Unlock() + pipe := func(dst, src net.Conn) { + buf := make([]byte, 32*1024) + for { + n, err := src.Read(buf) + if n > 0 && r.gen.Load() == born { + if _, werr := dst.Write(buf[:n]); werr != nil { + return + } + } + if err != nil { + return + } + } + } + go pipe(up, cli) + go pipe(cli, up) +} + +// blackhole strands every currently-established pair. Later connections are +// unaffected. +func (r *blackholeRelay) blackhole() { r.gen.Add(1) } + +// TestBlackholedPathRecovers is the end-to-end regression guard for the +// stability bug this hardening was written for: with the tunnel's path silently +// dropped, the client used to notice nothing at all. Its read loops parked +// forever, the dead worker conn stayed in the pool, the hub kept routing players +// to a control session nobody was reading, and no player could connect again +// until the client process was restarted. +// +// The heartbeats must now detect the silence, drop both sessions, and let the +// existing reconnect path restore service on its own. +func TestBlackholedPathRecovers(t *testing.T) { + const psk = "e2e-blackhole" + hubPort := freePort(t) + hubAddr := fmt.Sprintf("127.0.0.1:%d", hubPort) + startHub(t, hubPort, psk) + dest := newMockDest(t, modeEcho) + + // The client reaches the hub only through the relay; players connect to the + // hub directly, as they would from the internet. + relay := newBlackholeRelay(t, hubAddr) + const pingMs = 400 // heartbeat timeout is 3x this + c := startClientWithPing(t, relay.addr, psk, 4, pingMs, []client.Mapping{ + {Pattern: "mc.local", Destination: dest.addr}, + }) + + play := func(what string) error { + pc, err := net.DialTimeout("tcp", hubAddr, 5*time.Second) + if err != nil { + return err + } + defer pc.Close() + if _, err := pc.Write(wire.BuildHandshake(767, "mc.local", 25565, 2)); err != nil { + return err + } + msg := []byte(what) + if _, err := pc.Write(msg); err != nil { + return err + } + got := make([]byte, len(msg)) + _ = pc.SetReadDeadline(time.Now().Add(15 * time.Second)) + _, err = io.ReadFull(pc, got) + return err + } + + if err := play("before"); err != nil { + t.Fatalf("baseline round-trip failed: %v", err) + } + + relay.blackhole() + t.Log("path blackholed: no FIN, no RST, sockets held open") + + // Wait for the heartbeats to fire, the sessions to be dropped, and the + // control session to reconnect through a fresh relay pair. + deadline := time.Now().Add(45 * time.Second) + var lastErr error + for time.Now().Before(deadline) { + if lastErr = play("after"); lastErr == nil { + t.Logf("recovered on its own; worker conns now %d", c.WorkerConnCount()) + return + } + time.Sleep(500 * time.Millisecond) + } + t.Fatalf("client never recovered from the blackholed path (last error: %v)", lastErr) +} diff --git a/e2e/velocity_test.go b/e2e/velocity_test.go new file mode 100644 index 0000000..d33395b --- /dev/null +++ b/e2e/velocity_test.go @@ -0,0 +1,219 @@ +package e2e + +import ( + "bufio" + "bytes" + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "net" + "testing" + "time" + + "github.com/iceBear67/redapricot/client" + "github.com/iceBear67/redapricot/client/wire" +) + +// ---- mock velocity-forwarding destination ---- + +// veloEvent is what the mock backend saw in the (verified) forwarding payload. +type veloEvent struct { + version int + ip string + uuid []byte + name string + err error +} + +type veloDest struct { + addr string + secret string + success []byte // the Login Success packet the backend sends after the exchange + events chan veloEvent +} + +// newVeloDest starts a mock backend that requires Velocity modern forwarding: +// it reads the handshake and Login Start, sends the velocity:player_info +// query (with a negative message id, as Paper's random ids often are), verifies +// the HMAC-signed response, and finally sends a recognizable Login Success. +func newVeloDest(t *testing.T, secret string) *veloDest { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("velo dest listen: %v", err) + } + d := &veloDest{ + addr: ln.Addr().String(), + secret: secret, + success: mcPacket(wire.NewWriter(). + VarInt(0x02). // Login Success + Bytes(bytes.Repeat([]byte{0x42}, 16)). + String("e2ePlayer"). + VarInt(0). + Out()), + events: make(chan veloEvent, 16), + } + t.Cleanup(func() { _ = ln.Close() }) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go d.handle(conn) + } + }() + return d +} + +const veloMsgID = -777 + +func (d *veloDest) handle(conn net.Conn) { + defer conn.Close() + ev := d.exchange(conn) + d.events <- ev + if ev.err == nil { + _, _ = conn.Write(d.success) + } + _, _ = io.Copy(io.Discard, conn) // hold the connection until the peer closes +} + +func (d *veloDest) exchange(conn net.Conn) veloEvent { + br := bufio.NewReader(conn) + if _, err := readMCPacket(br); err != nil { // handshake + return veloEvent{err: fmt.Errorf("read handshake: %w", err)} + } + if _, err := readMCPacket(br); err != nil { // login start + return veloEvent{err: fmt.Errorf("read login start: %w", err)} + } + + query := mcPacket(wire.NewWriter(). + VarInt(0x04). // Login Plugin Request + VarInt(veloMsgID). + String("velocity:player_info"). + U8(0x04). // max supported forwarding version + Out()) + if _, err := conn.Write(query); err != nil { + return veloEvent{err: err} + } + + resp, err := readMCPacket(br) + if err != nil { + return veloEvent{err: fmt.Errorf("read plugin response: %w", err)} + } + r := wire.NewReader(resp) + id, _ := r.VarInt() + if id != 0x02 { + return veloEvent{err: fmt.Errorf("expected Login Plugin Response, got packet %#x", id)} + } + msgID, _ := r.VarInt() + if !bytes.Equal(wire.AppendVarInt(nil, msgID), wire.AppendVarInt(nil, veloMsgID)) { + return veloEvent{err: fmt.Errorf("message id not echoed: got %d", msgID)} + } + okFlag, _ := r.U8() + if okFlag != 1 { + return veloEvent{err: fmt.Errorf("response marked unsuccessful")} + } + sig, err := r.Bytes(32) + if err != nil { + return veloEvent{err: fmt.Errorf("missing signature: %w", err)} + } + payload := r.Remaining() + mac := hmac.New(sha256.New, []byte(d.secret)) + mac.Write(payload) + if !hmac.Equal(sig, mac.Sum(nil)) { + return veloEvent{err: fmt.Errorf("forwarding signature does not verify")} + } + + pr := wire.NewReader(payload) + var ev veloEvent + ev.version, _ = pr.VarInt() + ev.ip, _ = pr.String() + ev.uuid, _ = pr.Bytes(16) + ev.name, err = pr.String() + if err != nil { + return veloEvent{err: fmt.Errorf("truncated payload: %w", err)} + } + if props, err := pr.VarInt(); err != nil || props != 0 || len(pr.Remaining()) != 0 { + return veloEvent{err: fmt.Errorf("unexpected properties/trailer in payload")} + } + return ev +} + +func mcPacket(body []byte) []byte { + return append(wire.AppendVarInt(nil, len(body)), body...) +} + +func readMCPacket(br *bufio.Reader) ([]byte, error) { + n, err := wire.ReadVarInt(br) + if err != nil { + return nil, err + } + if n <= 0 || n > 1<<20 { + return nil, fmt.Errorf("bad packet length %d", n) + } + pkt := make([]byte, n) + if _, err := io.ReadFull(br, pkt); err != nil { + return nil, err + } + return pkt, nil +} + +// ---- test ---- + +// TestVelocityForwarding drives a full player login through the hub and a +// velocity-enabled mapping: the backend's velocity:player_info query must be +// answered by the client (never reaching the player), carrying the player's +// real IP, username and UUID, and the player's first bytes must be the +// backend's Login Success. +func TestVelocityForwarding(t *testing.T) { + const psk = "e2e-velocity" + const secret = "velo-forwarding-secret" + port := freePort(t) + hubAddr := fmt.Sprintf("127.0.0.1:%d", port) + startHub(t, port, psk) + dest := newVeloDest(t, secret) + startClient(t, hubAddr, psk, 2, []client.Mapping{ + {Pattern: `velo\.local`, Destination: dest.addr, VelocitySecret: secret}, + }) + + pc := dialPlayer(t, hubAddr, "velo.local") // protocol 767, login intent + defer pc.Close() + uuid, _ := hex.DecodeString("00112233445566778899aabbccddeeff") + loginStart := mcPacket(wire.NewWriter().VarInt(0x00).String("e2ePlayer").Bytes(uuid).Out()) + if _, err := pc.Write(loginStart); err != nil { + t.Fatalf("player login start: %v", err) + } + + var ev veloEvent + select { + case ev = <-dest.events: + case <-time.After(10 * time.Second): + t.Fatalf("backend never completed the forwarding exchange") + } + if ev.err != nil { + t.Fatalf("backend rejected the forwarding exchange: %v", ev.err) + } + if ev.version != 4 { + t.Fatalf("forwarding version = %d, want 4 (lazy session)", ev.version) + } + if ev.ip != "127.0.0.1" { + t.Fatalf("forwarded IP = %q, want the player's real 127.0.0.1", ev.ip) + } + if ev.name != "e2ePlayer" || !bytes.Equal(ev.uuid, uuid) { + t.Fatalf("forwarded profile = %s/%x, want e2ePlayer/%x", ev.name, ev.uuid, uuid) + } + + // The player must see the Login Success as its very first bytes — the + // velocity query must have been swallowed by the client. + got := make([]byte, len(dest.success)) + _ = pc.SetReadDeadline(time.Now().Add(10 * time.Second)) + if _, err := io.ReadFull(pc, got); err != nil { + t.Fatalf("player read login success: %v", err) + } + if !bytes.Equal(got, dest.success) { + t.Fatalf("player's first bytes are not the Login Success:\n got %x\nwant %x", got, dest.success) + } +} diff --git a/server/config.example.json b/server/config.example.json index cf3c22b..62fe45a 100644 --- a/server/config.example.json +++ b/server/config.example.json @@ -3,5 +3,6 @@ "psk": "change-me-to-a-long-random-passphrase", "timestampWindowMs": 30000, "pendingTimeoutMs": 10000, - "streamWindowBytes": 262144 + "streamWindowBytes": 262144, + "sessionIdleTimeoutMs": 90000 } diff --git a/server/src/main/java/io/icybear/redapricot/Config.java b/server/src/main/java/io/icybear/redapricot/Config.java index 3894145..93a67ae 100644 --- a/server/src/main/java/io/icybear/redapricot/Config.java +++ b/server/src/main/java/io/icybear/redapricot/Config.java @@ -12,7 +12,8 @@ public record Config( String psk, long timestampWindowMs, long pendingTimeoutMs, - int streamWindowBytes + int streamWindowBytes, + long sessionIdleTimeoutMs ) { public static Config load(Path file) throws Exception { JsonObject json = new JsonObject(Files.readString(file)); @@ -35,6 +36,9 @@ public record Config( psk, json.getLong("timestampWindowMs", 30_000L), json.getLong("pendingTimeoutMs", 10_000L), - window); + window, + // Comfortably above the client's default 20s ping interval; + // 0 disables the watchdog. + json.getLong("sessionIdleTimeoutMs", 90_000L)); } } diff --git a/server/src/main/java/io/icybear/redapricot/HubConnection.java b/server/src/main/java/io/icybear/redapricot/HubConnection.java index 145c6b1..d26200c 100644 --- a/server/src/main/java/io/icybear/redapricot/HubConnection.java +++ b/server/src/main/java/io/icybear/redapricot/HubConnection.java @@ -165,6 +165,9 @@ public final class HubConnection { return; } peerWindow = Math.min(peerWindow, Protocol.MAX_STREAM_WINDOW); + // Heartbeat is optional: accept it only when the client offered it, so + // older clients keep working (they just lose silent-path detection). + boolean heartbeat = (flags & Protocol.FLAG_WORKER_HEARTBEAT) != 0; // REKEY = Rand || Timestamp(I64 big-endian). Magic is excluded. byte[] rekey = new byte[randLen + 8]; @@ -179,9 +182,10 @@ public final class HubConnection { Crypto.decryptCipher(rekey, Crypto.DIR_C2S), Crypto.encryptCipher(rekey, Crypto.DIR_S2C)); // Echo the accepted flags plus our own receive window. + int accepted = Protocol.FLAG_STREAM_FC | (heartbeat ? Protocol.FLAG_WORKER_HEARTBEAT : 0); frames.send(new ProtoWriter() .u8(Protocol.CTL_SESSION_READY) - .varInt(Protocol.FLAG_STREAM_FC) + .varInt(accepted) .varInt(hub.config.streamWindowBytes()) .toBytes()); @@ -189,16 +193,48 @@ public final class HubConnection { ControlSession session = new ControlSession(hub, frames, id); frames.setHandler(session::onFrame); closeCleanup = session::onClose; - LOG.info("{} control session established", id); + LOG.info("{} control session established (heartbeat {})", id, heartbeat); } else if (magic == Protocol.MAGIC_WORKER) { WorkerConn worker = new WorkerConn(hub, frames, id, peerWindow, hub.config.streamWindowBytes()); frames.setHandler(worker::onFrame); closeCleanup = worker::onClose; - LOG.info("{} worker conn established (peer window {})", id, peerWindow); + LOG.info("{} worker conn established (peer window {}, heartbeat {})", id, peerWindow, heartbeat); } else { LOG.warn("{} bad magic {}; closing", id, magic); frames.close(); + return; } + armIdleWatchdog(); + } + + /** + * Drop an established redapricot session that has gone silent. Clients ping + * both their control session and every worker conn, so silence means the + * path is dead — without this the hub would keep a zombie control session + * registered and keep routing players into it, and zombie worker conns would + * hold player sockets open forever. Player connections are never subject to + * this; only authenticated sessions are. + */ + private void armIdleWatchdog() { + long idleMs = hub.config.sessionIdleTimeoutMs(); + if (idleMs <= 0) return; + long timer = hub.vertx.setPeriodic(idleMs / 2, tid -> { + if (frames.isClosed()) { + hub.vertx.cancelTimer(tid); + return; + } + long silent = System.currentTimeMillis() - frames.lastFrameAt(); + if (silent > idleMs) { + LOG.warn("{} session silent for {}ms; closing", id, silent); + hub.vertx.cancelTimer(tid); + frames.close(); + } + }); + Runnable inner = closeCleanup; + closeCleanup = () -> { + hub.vertx.cancelTimer(timer); + inner.run(); + }; } // ---- player connection ---- diff --git a/server/src/main/java/io/icybear/redapricot/HubServer.java b/server/src/main/java/io/icybear/redapricot/HubServer.java index ff3591d..ac6010e 100644 --- a/server/src/main/java/io/icybear/redapricot/HubServer.java +++ b/server/src/main/java/io/icybear/redapricot/HubServer.java @@ -24,6 +24,9 @@ public final class HubServer extends AbstractVerticle { .setHost(config.host()) .setPort(config.port()) .setTcpNoDelay(true) + // Probe idle sockets so a peer that becomes unreachable is + // eventually detected even when no frames are in flight. + .setTcpKeepAlive(true) .setReuseAddress(true); NetServer server = vertx.createNetServer(opts); diff --git a/server/src/main/java/io/icybear/redapricot/Protocol.java b/server/src/main/java/io/icybear/redapricot/Protocol.java index 7374e92..77519f8 100644 --- a/server/src/main/java/io/icybear/redapricot/Protocol.java +++ b/server/src/main/java/io/icybear/redapricot/Protocol.java @@ -32,10 +32,17 @@ public final class Protocol { public static final int MUX_FIN = 0x02; public static final int MUX_RST = 0x03; public static final int MUX_WND = 0x04; // per-stream flow-control credit grant + public static final int MUX_PING = 0x05; // liveness probe, StreamID 0 + public static final int MUX_PONG = 0x06; // liveness reply, echoes the nonce + + /** Reserved stream id for connection-scoped mux frames (PING/PONG). Streams start at 1. */ + public static final int MUX_CTL_SID = 0; // Session-establishment feature flags (trailing VarInt on the Rekey message, // echoed after the SessionReady type byte when accepted). public static final int FLAG_STREAM_FC = 0x01; + /** Mux-level PING/PONG on worker conns, so a silently dropped path is detected. */ + public static final int FLAG_WORKER_HEARTBEAT = 0x02; // Per-stream flow-control window bounds (bytes). public static final int DEFAULT_STREAM_WINDOW = 256 * 1024; diff --git a/server/src/main/java/io/icybear/redapricot/WorkerConn.java b/server/src/main/java/io/icybear/redapricot/WorkerConn.java index 67bd191..8010b13 100644 --- a/server/src/main/java/io/icybear/redapricot/WorkerConn.java +++ b/server/src/main/java/io/icybear/redapricot/WorkerConn.java @@ -65,6 +65,8 @@ public final class WorkerConn { case Protocol.MUX_DATA -> handleData(sid, r.readBuffer(r.remaining())); case Protocol.MUX_WND -> handleWnd(sid, r.readVarInt()); case Protocol.MUX_FIN, Protocol.MUX_RST -> closeStream(sid); + case Protocol.MUX_PING -> sendPong(r.readI64()); + case Protocol.MUX_PONG -> { /* liveness only; arrival is what matters */ } case Protocol.FRAME_ERROR -> LOG.warn("worker {} error frame", id); default -> LOG.warn("worker {} unknown mux type {}", id, type); } @@ -217,6 +219,11 @@ public final class WorkerConn { frames.send(new ProtoWriter().u8(Protocol.MUX_WND).varInt(sid).varInt(delta).toBytes()); } + /** Answer the client's liveness probe, echoing its nonce. */ + private void sendPong(long nonce) { + frames.send(new ProtoWriter().u8(Protocol.MUX_PONG).varInt(Protocol.MUX_CTL_SID).i64(nonce).toBytes()); + } + public void onClose() { for (StreamState st : streams.values()) st.player.close(); streams.clear(); diff --git a/server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java b/server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java index 4bdc135..5e899e5 100644 --- a/server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java +++ b/server/src/main/java/io/icybear/redapricot/net/EncryptedFrames.java @@ -23,6 +23,7 @@ public final class EncryptedFrames { private FrameHandler handler; private Buffer buf = Buffer.buffer(); private boolean closed = false; + private long lastFrameAt = System.currentTimeMillis(); public EncryptedFrames(NetSocket socket, Cipher in, Cipher out, FrameHandler handler) { this.socket = socket; @@ -71,6 +72,7 @@ public final class EncryptedFrames { byte[] pt = in.update(ct); if (pt == null) pt = new byte[0]; buf = buf.getBuffer(hdr + payloadLen, buf.length()); + lastFrameAt = System.currentTimeMillis(); FrameHandler h = handler; if (h != null) { try { @@ -96,6 +98,9 @@ public final class EncryptedFrames { public boolean writeQueueFull() { return socket.writeQueueFull(); } + /** Wall-clock millis when the last complete frame was decoded; basis for idle detection. */ + public long lastFrameAt() { return lastFrameAt; } + public void close() { if (closed) return; closed = true; diff --git a/server/src/test/java/io/icybear/redapricot/CryptoCodecTest.java b/server/src/test/java/io/icybear/redapricot/CryptoCodecTest.java index c29f74c..11c5632 100644 --- a/server/src/test/java/io/icybear/redapricot/CryptoCodecTest.java +++ b/server/src/test/java/io/icybear/redapricot/CryptoCodecTest.java @@ -77,7 +77,8 @@ class CryptoCodecTest { /** A Hub whose event loop is never touched (register/match/normalize use no Vert.x state). */ private static Hub testHub() { - return new Hub(null, new Config("0.0.0.0", 25565, "test-psk", 30_000L, 10_000L, Protocol.DEFAULT_STREAM_WINDOW)); + return new Hub(null, new Config("0.0.0.0", 25565, "test-psk", 30_000L, 10_000L, + Protocol.DEFAULT_STREAM_WINDOW, 90_000L)); } private static ControlSession testSession(Hub hub, String id) {