feat(multiplayer): bound control-plane websocket traffic

server/api/events.go

@@ -19,9 +19,12 @@ import (
 )
 
 const (
-	webSocketGUID      = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
-	maxWebSocketFrame  = 64 << 10
-	eventQueueCapacity = 32
+	webSocketGUID          = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
+	maxWebSocketFrame      = 64 << 10
+	eventQueueCapacity     = 32
+	webSocketIdleLimit     = 2 * time.Minute
+	webSocketMessageLimit  = 120
+	webSocketMessageWindow = time.Minute
 )
 
 // ControlPlaneEvent is the server-to-client envelope defined by the v1
@@ -139,7 +142,7 @@ func (s *Service) controlPlaneEvent(w http.ResponseWriter, r *http.Request) {
 		writeError(w, http.StatusMethodNotAllowed, "method_not_allowed")
 		return
 	}
-	if !isWebSocketUpgrade(r) || !validWebSocketKey(r.Header.Get("Sec-WebSocket-Key")) {
+	if !isWebSocketUpgrade(r) || r.Header.Get("Sec-WebSocket-Version") != "13" || !validWebSocketKey(r.Header.Get("Sec-WebSocket-Key")) {
 		writeError(w, http.StatusBadRequest, "invalid_websocket_upgrade")
 		return
 	}
@@ -250,11 +253,20 @@ func validWebSocketKey(key string) bool {
 
 func readWebSocketFrames(connection net.Conn, writeMu *sync.Mutex) {
 	reader := bufio.NewReader(connection)
+	windowStarted := time.Now()
+	messageCount := 0
 	for {
+		if err := connection.SetReadDeadline(time.Now().Add(webSocketIdleLimit)); err != nil {
+			return
+		}
 		opcode, _, err := readWebSocketFrame(reader)
 		if err != nil || opcode == 0x8 {
 			return
 		}
+		now := time.Now()
+		if !allowWebSocketMessage(now, &windowStarted, &messageCount) {
+			return
+		}
 		if opcode == 0x9 {
 			writeMu.Lock()
 			_ = writeWebSocketFrame(connection, 0xA, nil)
@@ -263,6 +275,21 @@ func readWebSocketFrames(connection net.Conn, writeMu *sync.Mutex) {
 	}
 }
 
+func allowWebSocketMessage(now time.Time, windowStarted *time.Time, count *int) bool {
+	if windowStarted == nil || count == nil || now.IsZero() {
+		return false
+	}
+	if !now.Before(windowStarted.Add(webSocketMessageWindow)) {
+		*windowStarted = now
+		*count = 0
+	}
+	if *count >= webSocketMessageLimit {
+		return false
+	}
+	*count++
+	return true
+}
+
 func readWebSocketFrame(reader *bufio.Reader) (byte, []byte, error) {
 	first, err := reader.ReadByte()
 	if err != nil {

server/api/events_test.go

@@ -0,0 +1,39 @@
+package api
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+)
+
+func TestWebSocketHandshakeRequiresRFC6455Version(t *testing.T) {
+	service := &Service{}
+	request := httptest.NewRequest(http.MethodGet, "/v1/events", nil)
+	request.Header.Set("Upgrade", "websocket")
+	request.Header.Set("Connection", "Upgrade")
+	request.Header.Set("Sec-WebSocket-Version", "12")
+	request.Header.Set("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+	recorder := httptest.NewRecorder()
+	service.controlPlaneEvent(recorder, request)
+	if recorder.Code != http.StatusBadRequest {
+		t.Fatalf("version 12 status = %d, want 400", recorder.Code)
+	}
+}
+
+func TestWebSocketMessageBudgetIsBoundedAndResets(t *testing.T) {
+	start := time.Unix(1000, 0)
+	window := start
+	count := 0
+	for i := 0; i < webSocketMessageLimit; i++ {
+		if !allowWebSocketMessage(start, &window, &count) {
+			t.Fatalf("message %d was rejected within the budget", i)
+		}
+	}
+	if allowWebSocketMessage(start, &window, &count) {
+		t.Fatal("message over the WebSocket budget was accepted")
+	}
+	if !allowWebSocketMessage(start.Add(webSocketMessageWindow), &window, &count) {
+		t.Fatal("WebSocket message budget did not reset")
+	}
+}