diff --git a/cmd/test-server/README.md b/cmd/test-server/README.md index 2f13f00da..00649f8fb 100644 --- a/cmd/test-server/README.md +++ b/cmd/test-server/README.md @@ -83,6 +83,80 @@ Response headers: |---|---| | `X-Lk-Mock-Region` | index of the region that served the response (blank on a failed region). Assert on this to confirm which region a failover landed on. | +## Signal connection (WebSocket) mocking + +The mock also speaks enough of the LiveKit signal protocol for SDKs to run +end-to-end signal-connection tests (connect, keepalive, reconnect, leave, and +the failure/timeout modes a client must classify). Signal behavior is selected +by a participant attribute (`lk.mock`) in the access token (see below) — the +WebSocket client can't set request headers, so it can't carry a control header. +Selecting via the token means parallel tests need no shared +state. + +Endpoints (both protocol versions are supported and behave identically): + +| Path | Purpose | +|---|---| +| `/rtc`, `/rtc/v1` | WebSocket signal connection | +| `/rtc/validate`, `/rtc/v1/validate` | HTTP validate (the client fetches this when the WS fails to open) | + +- The access token is read from the `access_token` query param (or a + `Bearer` Authorization header) and verified against the API secret. A + missing/malformed/expired/wrongly-signed token makes `validate` return + **401** (and the WS refuse the upgrade). +- Wire format is **binary protobuf**: `SignalRequest` in, `SignalResponse` out. +- The v1 embedded publisher offer (`join_request` connection param) is + **ignored** — no valid offer is required. +- Keepalive uses a short `pingTimeout=3s` / `pingInterval=1s` in the join so + timeout tests run fast. + +**Mode selection is via a participant attribute.** After the token is verified, +the server reads the `lk.mock` entry from the token's `attributes` claim +(`ClaimGrants.Attributes`, a `map[string]string`). The value of that attribute +is a stringified JSON control object whose `signal` field picks the behavior. +The `lk.mock` namespace is the attribute **key** (dot notation, matching +LiveKit's convention for internal attributes), so the value has no inner parent: + +``` +attribute key: lk.mock +attribute value: {"signal":"no_pong"} +``` + +The control object also accepts an optional `leaveAction` field — a +`LeaveRequest_Action`, given either as the number (`0`=DISCONNECT, `1`=RESUME, +`2`=RECONNECT) or the enum name (`"RECONNECT"`, case-insensitive) — that sets +the `action` on the `LeaveRequest` the leave-sending modes emit +(`leave_when_connected`, `leave_first_message`, `leave_during_reconnect`). When +absent it defaults to `0` (DISCONNECT). Examples: + +``` +attribute value: {"signal":"leave_when_connected","leaveAction":"RECONNECT"} +``` + +If the `lk.mock` attribute is absent/empty, its value is unparseable, or its +`signal` is unknown, the mode defaults to `happy`. Both the WS handlers and the +validate handlers read the mode from this same attribute. + +Behavior modes (any unknown/absent `signal` = `happy`): + +| `signal` value | Effect | +|---|---| +| `happy` | validate → 200; WS sends `JoinResponse` (or `ReconnectResponse` if `reconnect=1`), pongs pings, closes cleanly (1000) on client `LeaveRequest` | +| `validate_500` | validate → 500; WS refuses upgrade with 500 | +| `validate_service_not_found` | validate → 404 with a body *without* the room marker (client → serviceNotFound); WS refuses with 404 | +| `room_not_found` | validate → 404 with body `requested room does not exist` (client → notAllowed); WS refuses with 404 | +| `no_first_message` | WS accepted, server sends nothing (client hits connect timeout) | +| `no_pong` | WS sends the join, then never pongs (client hits ping timeout) | +| `close_before_join` | WS upgrade succeeds, then ~50ms later a clean close (code 1011, empty reason) *before* any first message — unexpected closure during connect | +| `close_when_connected` | WS sends join, then ~200ms later closes with code 1011 | +| `drop_when_connected` | WS sends join, then ~200ms later abruptly drops the TCP connection with no close handshake — client observes an abnormal closure (code 1006) | +| `leave_when_connected` | WS sends join, then ~200ms later sends a `LeaveRequest` | +| `leave_first_message` | WS sends a `LeaveRequest` as the first (and only) message | +| `leave_during_reconnect` | on a `reconnect=1` connection, sends `LeaveRequest` first; otherwise behaves like `happy` | + +`LeaveRequest`s carry `reason=SERVER_SHUTDOWN` and `action` from the control's +optional `leaveAction` (default `DISCONNECT (0)`). + ## Permission enforcement Every API method requires the same token grants the real LiveKit server checks diff --git a/cmd/test-server/main.go b/cmd/test-server/main.go index 7850a4c7c..7a2e4f52d 100644 --- a/cmd/test-server/main.go +++ b/cmd/test-server/main.go @@ -92,6 +92,10 @@ func (h *mockHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { switch { case r.URL.Path == "/settings/regions": h.handleRegions(w, r) + case isValidatePath(r.URL.Path): + h.handleValidate(w, r) + case isSignalPath(r.URL.Path): + h.handleSignal(w, r) case strings.HasPrefix(r.URL.Path, h.twirpPrefix+"/"): h.handleTwirp(w, r) case r.URL.Path == "/" || r.URL.Path == "/_test/health": diff --git a/cmd/test-server/signal.go b/cmd/test-server/signal.go new file mode 100644 index 000000000..439d669e7 --- /dev/null +++ b/cmd/test-server/signal.go @@ -0,0 +1,406 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "encoding/json" + "net/http" + "strconv" + "strings" + "time" + + "github.com/gorilla/websocket" + "google.golang.org/protobuf/proto" + + "github.com/livekit/protocol/auth" + "github.com/livekit/protocol/livekit" +) + +// Signal endpoints (/rtc, /rtc/v1 and their /validate counterparts) let SDKs +// exercise end-to-end WebSocket signal behavior. Per-connection behavior is +// selected by the `lk.mock` participant attribute (see signalControl), so tests +// need no shared state. The client fetches validate only when the WS fails to +// open, so validate-error modes refuse the upgrade with the matching status and +// let that fetch return the definitive status/body. + +const ( + // Short keepalive (seconds, sent in the JoinResponse) so timeout tests run fast. + signalPingInterval = 1 + signalPingTimeout = 3 + + // Delay after join before close_when_connected / leave_when_connected act, + // giving the client time to mark the connection established. + connectedDelay = 200 * time.Millisecond +) + +// Behavior modes, selected by the `lk.mock` attribute's `signal` field. Any +// unknown/absent signal behaves as the happy path. +const ( + // Validate-endpoint modes (the WS upgrade is refused with the same status + // so the client falls back to the validate fetch). + modeValidate500 = "validate_500" // validate → 500 + modeServiceNotFound = "validate_service_not_found" // validate → 404, generic body + modeRoomNotFound = "room_not_found" // validate → 404, "requested room does not exist" + + // WebSocket signal modes (validate → 200; behavior is on the WS). + modeHappy = "happy" // join, pong, clean close on client leave + modeNoFirstMessage = "no_first_message" // accept WS, send nothing + modeNoPong = "no_pong" // send join, never pong + modeCloseBeforeJoin = "close_before_join" // clean close 1011 before any first message + modeCloseWhenConnected = "close_when_connected" // send join, then clean close 1011 + modeDropWhenConnected = "drop_when_connected" // send join, then abrupt TCP drop (1006) + modeLeaveWhenConnected = "leave_when_connected" // send join, then LeaveRequest + modeLeaveFirstMessage = "leave_first_message" // LeaveRequest as first message + modeLeaveDuringReconnect = "leave_during_reconnect" // on reconnect=1, LeaveRequest first +) + +const signalControlAttribute = "lk.mock" + +// signalControl is the JSON value of the `lk.mock` attribute: +// {"signal":"","leaveAction":}. leaveAction is optional +// (a LeaveRequest_Action, given as the number or the enum name e.g. +// "RECONNECT"; absent/0 = DISCONNECT) and sets the action on emitted leaves. +type signalControl struct { + Signal string `json:"signal"` + LeaveAction leaveActionValue `json:"leaveAction"` +} + +// leaveActionValue is a LeaveRequest_Action that unmarshals from either a JSON +// number (2) or an enum name ("RECONNECT", case-insensitive). Anything +// unrecognized decodes to 0 (DISCONNECT) rather than failing the whole control. +type leaveActionValue livekit.LeaveRequest_Action + +func (v *leaveActionValue) UnmarshalJSON(b []byte) error { + var n int32 + if json.Unmarshal(b, &n) == nil { + *v = leaveActionValue(n) + return nil + } + var s string + if err := json.Unmarshal(b, &s); err != nil { + return err + } + *v = leaveActionValue(livekit.LeaveRequest_Action_value[strings.ToUpper(s)]) + return nil +} + +// parseSignalControl parses the `lk.mock` attribute value; absent/invalid → zero. +func parseSignalControl(grants *auth.ClaimGrants) signalControl { + if grants == nil { + return signalControl{} + } + raw := grants.Attributes[signalControlAttribute] + if raw == "" { + return signalControl{} + } + var ctrl signalControl + if err := json.Unmarshal([]byte(raw), &ctrl); err != nil { + return signalControl{} + } + return ctrl +} + +// signalMode returns the mode from the `lk.mock` `signal` field; unknown/absent → happy. +func signalMode(grants *auth.ClaimGrants) string { + switch ctrl := parseSignalControl(grants); ctrl.Signal { + case modeValidate500, modeServiceNotFound, modeRoomNotFound, + modeHappy, modeNoFirstMessage, modeNoPong, + modeCloseBeforeJoin, modeCloseWhenConnected, modeDropWhenConnected, + modeLeaveWhenConnected, modeLeaveFirstMessage, modeLeaveDuringReconnect: + return ctrl.Signal + default: + return modeHappy + } +} + +func isSignalPath(path string) bool { + return path == "/rtc" || path == "/rtc/v1" +} + +func isValidatePath(path string) bool { + return path == "/rtc/validate" || path == "/rtc/v1/validate" +} + +var signalUpgrader = websocket.Upgrader{ + EnableCompression: true, + // Auth is via the access token, so allow any origin. + CheckOrigin: func(r *http.Request) bool { return true }, +} + +// verifySignalToken reads the access token (access_token query param or Bearer +// header) and verifies it against the mock's API secret. +func (h *mockHandler) verifySignalToken(r *http.Request) (*auth.ClaimGrants, error) { + token := r.FormValue("access_token") + if token == "" { + token = strings.TrimSpace(strings.TrimPrefix(r.Header.Get("Authorization"), "Bearer ")) + } + v, err := auth.ParseAPIToken(token) + if err != nil { + return nil, err + } + _, grants, err := v.Verify(h.apiSecret) + if err != nil { + return nil, err + } + return grants, nil +} + +func grantRoom(grants *auth.ClaimGrants) string { + if grants == nil || grants.Video == nil { + return "" + } + return grants.Video.Room +} + +// handleValidate verifies the JWT (bad/expired/missing → 401), then returns the +// status the mode dictates. +func (h *mockHandler) handleValidate(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Access-Control-Allow-Origin", "*") + + grants, err := h.verifySignalToken(r) + if err != nil { + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte("invalid token: " + err.Error())) + return + } + + switch signalMode(grants) { + case modeValidate500: + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte("internal server error")) + case modeServiceNotFound: + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte("404 page not found")) + case modeRoomNotFound: + w.WriteHeader(http.StatusNotFound) + _, _ = w.Write([]byte("requested room does not exist")) + default: + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("success")) + } +} + +// handleSignal verifies the token, applies validate-error modes by refusing the +// upgrade, else upgrades and runs the selected behavior. The v1 publisher offer +// is ignored. +func (h *mockHandler) handleSignal(w http.ResponseWriter, r *http.Request) { + grants, err := h.verifySignalToken(r) + if err != nil { + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte("invalid token")) + return + } + + mode := signalMode(grants) + + switch mode { + case modeValidate500: + w.WriteHeader(http.StatusInternalServerError) + return + case modeServiceNotFound, modeRoomNotFound: + w.WriteHeader(http.StatusNotFound) + return + } + + reconnect := r.URL.Query().Get("reconnect") == "1" + + conn, err := signalUpgrader.Upgrade(w, r, nil) + if err != nil { + return + } + defer func() { _ = conn.Close() }() + + h.runSignal(conn, mode, reconnect, grants) +} + +// runSignal drives one WebSocket connection according to mode. +func (h *mockHandler) runSignal(conn *websocket.Conn, mode string, reconnect bool, grants *auth.ClaimGrants) { + writeResp := func(msg *livekit.SignalResponse) error { + payload, err := proto.Marshal(msg) + if err != nil { + return err + } + return conn.WriteMessage(websocket.BinaryMessage, payload) + } + + leaveAction := livekit.LeaveRequest_Action(parseSignalControl(grants).LeaveAction) + + // drainUntilClosed reads/discards until the peer closes, keeping the socket open. + drainUntilClosed := func() { + for { + if _, _, err := conn.ReadMessage(); err != nil { + return + } + } + } + + // Modes that decide the very first message. + switch mode { + case modeNoFirstMessage: + drainUntilClosed() + return + case modeCloseBeforeJoin: + time.Sleep(50 * time.Millisecond) + msg := websocket.FormatCloseMessage(websocket.CloseInternalServerErr, "") + _ = conn.WriteControl(websocket.CloseMessage, msg, time.Now().Add(time.Second)) + return + case modeLeaveFirstMessage: + _ = writeResp(leaveResponse(leaveAction)) + drainUntilClosed() + return + case modeLeaveDuringReconnect: + if reconnect { + _ = writeResp(leaveResponse(leaveAction)) + drainUntilClosed() + return + } + // Non-reconnect connections fall through to the happy path. + } + + // First message: reconnect response on a resume, join otherwise. + if reconnect { + if err := writeResp(reconnectResponse(h.regionIndex)); err != nil { + return + } + } else { + if err := writeResp(joinResponse(h.regionIndex, grants)); err != nil { + return + } + } + + // Post-join behaviors. + switch mode { + case modeCloseWhenConnected: + time.Sleep(connectedDelay) + msg := websocket.FormatCloseMessage(websocket.CloseInternalServerErr, "mock close_when_connected") + _ = conn.WriteControl(websocket.CloseMessage, msg, time.Now().Add(time.Second)) + return + case modeDropWhenConnected: + time.Sleep(connectedDelay) + _ = conn.UnderlyingConn().Close() + return + case modeLeaveWhenConnected: + time.Sleep(connectedDelay) + _ = writeResp(leaveResponse(leaveAction)) + } + + // Read loop: pong to pings (unless no_pong), clean close on client leave. + for { + mt, payload, err := conn.ReadMessage() + if err != nil { + return + } + if mt != websocket.BinaryMessage { + continue + } + req := &livekit.SignalRequest{} + if err := proto.Unmarshal(payload, req); err != nil { + continue + } + switch m := req.Message.(type) { + case *livekit.SignalRequest_Ping: + if mode != modeNoPong { + _ = writeResp(&livekit.SignalResponse{ + Message: &livekit.SignalResponse_Pong{Pong: time.Now().UnixMilli()}, + }) + } + case *livekit.SignalRequest_PingReq: + if mode != modeNoPong { + _ = writeResp(&livekit.SignalResponse{ + Message: &livekit.SignalResponse_PongResp{ + PongResp: &livekit.Pong{ + LastPingTimestamp: m.PingReq.Timestamp, + Timestamp: time.Now().UnixMilli(), + }, + }, + }) + } + case *livekit.SignalRequest_Leave: + msg := websocket.FormatCloseMessage(websocket.CloseNormalClosure, "") + _ = conn.WriteControl(websocket.CloseMessage, msg, time.Now().Add(time.Second)) + return + } + } +} + +func serverInfo(regionIndex int) *livekit.ServerInfo { + return &livekit.ServerInfo{ + Edition: livekit.ServerInfo_Standard, + Version: "mock", + Protocol: 15, + Region: regionName(regionIndex), + NodeId: "MOCK_NODE", + } +} + +func regionName(regionIndex int) string { + return "region-" + strconv.Itoa(regionIndex) +} + +// joinResponse builds the initial JoinResponse (non-zero ping config so the +// client arms keepalive). +func joinResponse(regionIndex int, grants *auth.ClaimGrants) *livekit.SignalResponse { + room := grantRoom(grants) + identity := "mock-participant" + name := "" + if grants != nil { + if grants.Identity != "" { + identity = grants.Identity + } + name = grants.Name + } + return &livekit.SignalResponse{ + Message: &livekit.SignalResponse_Join{ + Join: &livekit.JoinResponse{ + Room: &livekit.Room{ + Sid: "RM_MOCK", + Name: room, + }, + Participant: &livekit.ParticipantInfo{ + Sid: "PA_MOCK", + Identity: identity, + Name: name, + State: livekit.ParticipantInfo_JOINED, + }, + PingInterval: signalPingInterval, + PingTimeout: signalPingTimeout, + ServerInfo: serverInfo(regionIndex), + ServerVersion: "mock", + ServerRegion: regionName(regionIndex), + }, + }, + } +} + +func reconnectResponse(regionIndex int) *livekit.SignalResponse { + return &livekit.SignalResponse{ + Message: &livekit.SignalResponse_Reconnect{ + Reconnect: &livekit.ReconnectResponse{ + ServerInfo: serverInfo(regionIndex), + }, + }, + } +} + +func leaveResponse(action livekit.LeaveRequest_Action) *livekit.SignalResponse { + return &livekit.SignalResponse{ + Message: &livekit.SignalResponse_Leave{ + Leave: &livekit.LeaveRequest{ + Reason: livekit.DisconnectReason_SERVER_SHUTDOWN, + Action: action, + }, + }, + } +} diff --git a/cmd/test-server/signal_test.go b/cmd/test-server/signal_test.go new file mode 100644 index 000000000..3ccd97a95 --- /dev/null +++ b/cmd/test-server/signal_test.go @@ -0,0 +1,451 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/websocket" + "google.golang.org/protobuf/proto" + + "github.com/livekit/protocol/auth" + "github.com/livekit/protocol/livekit" +) + +const testSecret = "secret" + +func newTestServer() *httptest.Server { + return httptest.NewServer(&mockHandler{regionIndex: 0, apiSecret: testSecret}) +} + +// mintToken signs a token whose `lk.mock` attribute selects mode (empty → happy). +func mintToken(t *testing.T, mode string) string { + t.Helper() + if mode == "" { + return mintTokenControl(t, nil) + } + return mintTokenControl(t, &signalControl{Signal: mode}) +} + +// mintTokenControl signs a token carrying ctrl as the `lk.mock` attribute (nil → none). +func mintTokenControl(t *testing.T, ctrl *signalControl) string { + t.Helper() + at := auth.NewAccessToken("APItest", testSecret). + SetIdentity("tester"). + SetValidFor(time.Hour). + SetVideoGrant(&auth.VideoGrant{Room: "test-room", RoomJoin: true}) + if ctrl != nil { + raw, err := json.Marshal(ctrl) + if err != nil { + t.Fatalf("marshal control: %v", err) + } + at.SetAttributes(map[string]string{signalControlAttribute: string(raw)}) + } + tok, err := at.ToJWT() + if err != nil { + t.Fatalf("mint token: %v", err) + } + return tok +} + +// mintTokenAttr signs a token whose `lk.mock` attribute is the given raw value. +func mintTokenAttr(t *testing.T, attrValue string) string { + t.Helper() + at := auth.NewAccessToken("APItest", testSecret). + SetIdentity("tester"). + SetValidFor(time.Hour). + SetVideoGrant(&auth.VideoGrant{Room: "test-room", RoomJoin: true}). + SetAttributes(map[string]string{signalControlAttribute: attrValue}) + tok, err := at.ToJWT() + if err != nil { + t.Fatalf("mint token: %v", err) + } + return tok +} + +func wsURL(base, path, token string) string { + u := strings.Replace(base, "http://", "ws://", 1) + sep := "?" + if strings.Contains(path, "?") { + sep = "&" + } + return u + path + sep + "access_token=" + token +} + +func dial(t *testing.T, base, path, token string) *websocket.Conn { + t.Helper() + c, _, err := websocket.DefaultDialer.Dial(wsURL(base, path, token), nil) + if err != nil { + t.Fatalf("dial %s: %v", path, err) + } + return c +} + +func readResp(t *testing.T, c *websocket.Conn, timeout time.Duration) *livekit.SignalResponse { + t.Helper() + _ = c.SetReadDeadline(time.Now().Add(timeout)) + mt, payload, err := c.ReadMessage() + if err != nil { + t.Fatalf("read: %v", err) + } + if mt != websocket.BinaryMessage { + t.Fatalf("expected binary message, got %d", mt) + } + resp := &livekit.SignalResponse{} + if err := proto.Unmarshal(payload, resp); err != nil { + t.Fatalf("unmarshal: %v", err) + } + return resp +} + +func writeReq(t *testing.T, c *websocket.Conn, req *livekit.SignalRequest) { + t.Helper() + payload, err := proto.Marshal(req) + if err != nil { + t.Fatalf("marshal req: %v", err) + } + if err := c.WriteMessage(websocket.BinaryMessage, payload); err != nil { + t.Fatalf("write req: %v", err) + } +} + +func TestHappyJoinAndPingPong(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc", mintToken(t, "happy")) + defer c.Close() + + resp := readResp(t, c, 2*time.Second) + join := resp.GetJoin() + if join == nil { + t.Fatalf("first message not join: %T", resp.Message) + } + if join.PingTimeout == 0 || join.PingInterval == 0 { + t.Fatalf("ping timeout/interval must be non-zero: %d/%d", join.PingTimeout, join.PingInterval) + } + if join.ServerInfo == nil || join.Room == nil || join.Participant == nil { + t.Fatalf("join missing serverInfo/room/participant") + } + if join.Room.Name != "test-room" { + t.Fatalf("room name = %q, want test-room", join.Room.Name) + } + + // pingReq -> pongResp echoing timestamp + writeReq(t, c, &livekit.SignalRequest{ + Message: &livekit.SignalRequest_PingReq{PingReq: &livekit.Ping{Timestamp: 12345}}, + }) + pr := readResp(t, c, 2*time.Second) + if pr.GetPongResp() == nil || pr.GetPongResp().LastPingTimestamp != 12345 { + t.Fatalf("expected pongResp echoing 12345, got %+v", pr.Message) + } + + // legacy ping -> pong + writeReq(t, c, &livekit.SignalRequest{Message: &livekit.SignalRequest_Ping{Ping: 999}}) + pong := readResp(t, c, 2*time.Second) + if pong.GetPong() == 0 { + t.Fatalf("expected pong, got %+v", pong.Message) + } +} + +func TestReconnectResponse(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc?reconnect=1", mintToken(t, "happy")) + defer c.Close() + resp := readResp(t, c, 2*time.Second) + if resp.GetReconnect() == nil { + t.Fatalf("expected reconnect response, got %T", resp.Message) + } +} + +func TestV1PathHappy(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc/v1", mintToken(t, "happy")) + defer c.Close() + if readResp(t, c, 2*time.Second).GetJoin() == nil { + t.Fatal("v1 first message not join") + } +} + +func TestNoPong(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc", mintToken(t, "no_pong")) + defer c.Close() + if readResp(t, c, 2*time.Second).GetJoin() == nil { + t.Fatal("expected join") + } + writeReq(t, c, &livekit.SignalRequest{ + Message: &livekit.SignalRequest_PingReq{PingReq: &livekit.Ping{Timestamp: 1}}, + }) + _ = c.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + if _, _, err := c.ReadMessage(); err == nil { + t.Fatal("expected no pong (timeout), but got a message") + } +} + +func TestLeaveFirstMessage(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc", mintToken(t, "leave_first_message")) + defer c.Close() + if readResp(t, c, 2*time.Second).GetLeave() == nil { + t.Fatal("expected leave as first message") + } +} + +func TestCloseWhenConnected(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc", mintToken(t, "close_when_connected")) + defer c.Close() + if readResp(t, c, 2*time.Second).GetJoin() == nil { + t.Fatal("expected join") + } + _ = c.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, _, err := c.ReadMessage() + ce, ok := err.(*websocket.CloseError) + if !ok { + t.Fatalf("expected close error, got %v", err) + } + if ce.Code != websocket.CloseInternalServerErr { + t.Fatalf("expected close code 1011, got %d", ce.Code) + } +} + +func TestLeaveWhenConnected(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc", mintToken(t, "leave_when_connected")) + defer c.Close() + if readResp(t, c, 2*time.Second).GetJoin() == nil { + t.Fatal("expected join") + } + leave := readResp(t, c, 2*time.Second).GetLeave() + if leave == nil { + t.Fatal("expected leave after join") + } + // Default action is DISCONNECT. + if leave.Action != livekit.LeaveRequest_DISCONNECT { + t.Fatalf("default leave action = %v, want DISCONNECT", leave.Action) + } +} + +func TestLeaveActionOverride(t *testing.T) { + srv := newTestServer() + defer srv.Close() + tok := mintTokenControl(t, &signalControl{ + Signal: "leave_when_connected", + LeaveAction: leaveActionValue(livekit.LeaveRequest_RECONNECT), + }) + c := dial(t, srv.URL, "/rtc", tok) + defer c.Close() + if readResp(t, c, 2*time.Second).GetJoin() == nil { + t.Fatal("expected join") + } + leave := readResp(t, c, 2*time.Second).GetLeave() + if leave == nil { + t.Fatal("expected leave after join") + } + if leave.Action != livekit.LeaveRequest_RECONNECT { + t.Fatalf("leave action = %v, want RECONNECT", leave.Action) + } +} + +func TestLeaveActionByName(t *testing.T) { + srv := newTestServer() + defer srv.Close() + // leaveAction may be the enum name instead of the number. + tok := mintTokenAttr(t, `{"signal":"leave_when_connected","leaveAction":"RECONNECT"}`) + c := dial(t, srv.URL, "/rtc", tok) + defer c.Close() + if readResp(t, c, 2*time.Second).GetJoin() == nil { + t.Fatal("expected join") + } + leave := readResp(t, c, 2*time.Second).GetLeave() + if leave == nil { + t.Fatal("expected leave after join") + } + if leave.Action != livekit.LeaveRequest_RECONNECT { + t.Fatalf("leave action = %v, want RECONNECT", leave.Action) + } +} + +func TestNoFirstMessage(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc", mintToken(t, "no_first_message")) + defer c.Close() + _ = c.SetReadDeadline(time.Now().Add(500 * time.Millisecond)) + if _, _, err := c.ReadMessage(); err == nil { + t.Fatal("expected no first message (timeout)") + } +} + +func TestCloseBeforeJoin(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc", mintToken(t, "close_before_join")) + defer c.Close() + // First read must be the close, never a join. + _ = c.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, _, err := c.ReadMessage() + ce, ok := err.(*websocket.CloseError) + if !ok { + t.Fatalf("expected close error before any message, got %v", err) + } + if ce.Code != websocket.CloseInternalServerErr { + t.Fatalf("expected close code 1011, got %d", ce.Code) + } + if ce.Text != "" { + t.Fatalf("expected empty close reason, got %q", ce.Text) + } +} + +func TestDropWhenConnected(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc", mintToken(t, "drop_when_connected")) + defer c.Close() + if readResp(t, c, 2*time.Second).GetJoin() == nil { + t.Fatal("expected join") + } + // Abrupt TCP drop → abnormal closure (1006 to a browser): a read error that + // is not a normal (1000) close. + _ = c.SetReadDeadline(time.Now().Add(2 * time.Second)) + _, _, err := c.ReadMessage() + if err == nil { + t.Fatal("expected read error after abrupt drop") + } + if websocket.IsCloseError(err, websocket.CloseNormalClosure) { + t.Fatalf("expected abnormal (non-1000) closure, got %v", err) + } +} + +func getStatusBody(t *testing.T, url string) (int, string) { + t.Helper() + resp, err := http.Get(url) + if err != nil { + t.Fatalf("get %s: %v", url, err) + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + return resp.StatusCode, string(b) +} + +func TestValidateModes(t *testing.T) { + srv := newTestServer() + defer srv.Close() + + // happy → 200 + if st, _ := getStatusBody(t, srv.URL+"/rtc/validate?access_token="+mintToken(t, "happy")); st != 200 { + t.Fatalf("happy validate = %d, want 200", st) + } + // v1 validate happy → 200 + if st, _ := getStatusBody(t, srv.URL+"/rtc/v1/validate?access_token="+mintToken(t, "happy")); st != 200 { + t.Fatalf("v1 happy validate = %d, want 200", st) + } + // validate_500 → 500 + if st, _ := getStatusBody(t, srv.URL+"/rtc/validate?access_token="+mintToken(t, "validate_500")); st != 500 { + t.Fatalf("validate_500 = %d, want 500", st) + } + // validate_service_not_found → 404, no marker + st, body := getStatusBody(t, srv.URL+"/rtc/validate?access_token="+mintToken(t, "validate_service_not_found")) + if st != 404 || strings.Contains(body, "requested room does not exist") { + t.Fatalf("service_not_found = %d body=%q", st, body) + } + // room_not_found → 404 with marker + st, body = getStatusBody(t, srv.URL+"/rtc/validate?access_token="+mintToken(t, "room_not_found")) + if st != 404 || !strings.Contains(body, "requested room does not exist") { + t.Fatalf("room_not_found = %d body=%q", st, body) + } + // bad token → 401 + if st, _ := getStatusBody(t, srv.URL+"/rtc/validate?access_token=not-a-jwt"); st != 401 { + t.Fatalf("bad token validate = %d, want 401", st) + } + // missing token → 401 + if st, _ := getStatusBody(t, srv.URL+"/rtc/validate"); st != 401 { + t.Fatalf("missing token validate = %d, want 401", st) + } +} + +func TestValidateErrorModesRefuseWS(t *testing.T) { + srv := newTestServer() + defer srv.Close() + // Validate-error modes must refuse the upgrade so the client falls back to validate. + _, resp, err := websocket.DefaultDialer.Dial(wsURL(srv.URL, "/rtc", mintToken(t, "validate_500")), nil) + if err == nil { + t.Fatal("expected WS dial to fail for validate_500") + } + if resp == nil || resp.StatusCode != 500 { + t.Fatalf("expected 500 on WS refuse, got %v", resp) + } +} + +func TestValidateCORSHeader(t *testing.T) { + srv := newTestServer() + defer srv.Close() + // ACAO must be present on every status so a browser fetch can read it. + cases := map[string]string{ + "happy": mintToken(t, "happy"), // 200 + "bad": "bad", // 401 + "room_not_found": mintToken(t, "room_not_found"), // 404 + "validate_500": mintToken(t, "validate_500"), // 500 + } + for name, tok := range cases { + resp, err := http.Get(srv.URL + "/rtc/validate?access_token=" + tok) + if err != nil { + t.Fatalf("%s: get: %v", name, err) + } + got := resp.Header.Get("Access-Control-Allow-Origin") + resp.Body.Close() + if got != "*" { + t.Fatalf("%s (status %d): ACAO = %q, want *", name, resp.StatusCode, got) + } + } +} + +func TestWSCrossOrigin(t *testing.T) { + srv := newTestServer() + defer srv.Close() + // A mismatched browser Origin must still upgrade (CheckOrigin allows any). + hdr := http.Header{} + hdr.Set("Origin", "http://localhost:5173") + c, _, err := websocket.DefaultDialer.Dial(wsURL(srv.URL, "/rtc", mintToken(t, "happy")), hdr) + if err != nil { + t.Fatalf("cross-origin dial: %v", err) + } + defer c.Close() + if readResp(t, c, 2*time.Second).GetJoin() == nil { + t.Fatal("expected join on cross-origin WS") + } +} + +func TestLeaveDuringReconnect(t *testing.T) { + srv := newTestServer() + defer srv.Close() + c := dial(t, srv.URL, "/rtc?reconnect=1", mintToken(t, "leave_during_reconnect")) + defer c.Close() + if readResp(t, c, 2*time.Second).GetLeave() == nil { + t.Fatal("expected leave on reconnect") + } +}