nextcloud-spreed-signaling/hub_test.go

6112 lines
166 KiB
Go

</
/**
* Standalone signaling server for the Nextcloud Spreed app.
* Copyright (C) 2017 struktur AG
*
* @author Joachim Bauch <bauch@struktur.de>
*
* @license GNU AGPL version 3 or any later version
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
package signaling
import (
"bytes"
"context"
"crypto/ecdsa"
"crypto/ed25519"
"crypto/elliptic"
"crypto/rand"
"crypto/rsa"
"crypto/x509"
"encoding/base64"
"encoding/json"
"encoding/pem"
"errors"
"io"
"net/http"
"net/http/httptest"
"net/url"
"os"
"reflect"
"strings"
"sync"
"testing"
"time"
"github.com/dlintw/goconf"
"github.com/golang-jwt/jwt/v4"
"github.com/gorilla/mux"
"github.com/gorilla/websocket"
)
const (
testDefaultUserId = "test-userid"
authAnonymousUserId = "anonymous-userid"
testTimeout = 10 * time.Second
)
var (
testRoomProperties = []byte("{\"prop1\":\"value1\"}")
)
var (
clusteredTests = []string{
"local",
"clustered",
}
testHelloV2Algorithms = []string{
"RSA",
"ECDSA",
"Ed25519",
"Ed25519_Nextcloud",
}
)
// Only used for testing.
func (h *Hub) getRoom(id string) *Room {
h.ru.RLock()
defer h.ru.RUnlock()
// TODO: The same room might exist on different backends.
for _, room := range h.rooms {
if room.Id() == id {
return room
}
}
return nil
}
func isLocalTest(t *testing.T) bool {
return strings.HasSuffix(t.Name(), "/local")
}
func getTestConfig(server *httptest.Server) (*goconf.ConfigFile, error) {
config := goconf.NewConfigFile()
u, err := url.Parse(server.URL)
if err != nil {
return nil, err
}
config.AddOption("backend", "allowed", u.Host)
if u.Scheme == "http" {
config.AddOption("backend", "allowhttp", "true")
}
config.AddOption("backend", "secret", string(testBackendSecret))
config.AddOption("sessions", "hashkey", "12345678901234567890123456789012")
config.AddOption("sessions", "blockkey", "09876543210987654321098765432109")
config.AddOption("clients", "internalsecret", string(testInternalSecret))
config.AddOption("geoip", "url", "none")
return config, nil
}
func getTestConfigWithMultipleBackends(server *httptest.Server) (*goconf.ConfigFile, error) {
config, err := getTestConfig(server)
if err != nil {
return nil, err
}
config.RemoveOption("backend", "allowed")
config.RemoveOption("backend", "secret")
config.AddOption("backend", "backends", "backend1, backend2")
config.AddOption("backend1", "url", server.URL+"/one")
config.AddOption("backend1", "secret", string(testBackendSecret))
config.AddOption("backend2", "url", server.URL+"/two/")
config.AddOption("backend2", "secret", string(testBackendSecret))
return config, nil
}
func CreateHubForTestWithConfig(t *testing.T, getConfigFunc func(*httptest.Server) (*goconf.ConfigFile, error)) (*Hub, AsyncEvents, *mux.Router, *httptest.Server) {
r := mux.NewRouter()
registerBackendHandler(t, r)
server := httptest.NewServer(r)
t.Cleanup(func() {
server.Close()
})
events := getAsyncEventsForTest(t)
config, err := getConfigFunc(server)
if err != nil {
t.Fatal(err)
}
h, err := NewHub(config, events, nil, nil, nil, r, "no-version")
if err != nil {
t.Fatal(err)
}
b, err := NewBackendServer(config, h, "no-version")
if err != nil {
t.Fatal(err)
}
if err := b.Start(r); err != nil {
t.Fatal(err)
}
go h.Run()
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
WaitForHub(ctx, t, h)
})
return h, events, r, server
}
func CreateHubForTest(t *testing.T) (*Hub, AsyncEvents, *mux.Router, *httptest.Server) {
return CreateHubForTestWithConfig(t, getTestConfig)
}
func CreateHubWithMultipleBackendsForTest(t *testing.T) (*Hub, AsyncEvents, *mux.Router, *httptest.Server) {
h, events, r, server := CreateHubForTestWithConfig(t, getTestConfigWithMultipleBackends)
registerBackendHandlerUrl(t, r, "/one")
registerBackendHandlerUrl(t, r, "/two")
return h, events, r, server
}
func CreateClusteredHubsForTestWithConfig(t *testing.T, getConfigFunc func(*httptest.Server) (*goconf.ConfigFile, error)) (*Hub, *Hub, *mux.Router, *mux.Router, *httptest.Server, *httptest.Server) {
r1 := mux.NewRouter()
registerBackendHandler(t, r1)
server1 := httptest.NewServer(r1)
t.Cleanup(func() {
server1.Close()
})
r2 := mux.NewRouter()
registerBackendHandler(t, r2)
server2 := httptest.NewServer(r2)
t.Cleanup(func() {
server2.Close()
})
nats := startLocalNatsServer(t)
grpcServer1, addr1 := NewGrpcServerForTest(t)
grpcServer2, addr2 := NewGrpcServerForTest(t)
events1, err := NewAsyncEvents(nats)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
events1.Close()
})
config1, err := getConfigFunc(server1)
if err != nil {
t.Fatal(err)
}
client1, _ := NewGrpcClientsForTest(t, addr2)
h1, err := NewHub(config1, events1, grpcServer1, client1, nil, r1, "no-version")
if err != nil {
t.Fatal(err)
}
b1, err := NewBackendServer(config1, h1, "no-version")
if err != nil {
t.Fatal(err)
}
events2, err := NewAsyncEvents(nats)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() {
events2.Close()
})
config2, err := getConfigFunc(server2)
if err != nil {
t.Fatal(err)
}
client2, _ := NewGrpcClientsForTest(t, addr1)
h2, err := NewHub(config2, events2, grpcServer2, client2, nil, r2, "no-version")
if err != nil {
t.Fatal(err)
}
b2, err := NewBackendServer(config2, h2, "no-version")
if err != nil {
t.Fatal(err)
}
if err := b1.Start(r1); err != nil {
t.Fatal(err)
}
if err := b2.Start(r2); err != nil {
t.Fatal(err)
}
go h1.Run()
go h2.Run()
t.Cleanup(func() {
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
WaitForHub(ctx, t, h1)
WaitForHub(ctx, t, h2)
})
return h1, h2, r1, r2, server1, server2
}
func CreateClusteredHubsForTest(t *testing.T) (*Hub, *Hub, *httptest.Server, *httptest.Server) {
h1, h2, _, _, server1, server2 := CreateClusteredHubsForTestWithConfig(t, getTestConfig)
return h1, h2, server1, server2
}
func WaitForHub(ctx context.Context, t *testing.T, h *Hub) {
// Wait for any channel messages to be processed.
time.Sleep(10 * time.Millisecond)
h.Stop()
for {
h.mu.Lock()
clients := len(h.clients)
sessions := len(h.sessions)
remoteSessions := len(h.remoteSessions)
h.mu.Unlock()
h.ru.Lock()
rooms := len(h.rooms)
h.ru.Unlock()
readActive := h.readPumpActive.Load()
writeActive := h.writePumpActive.Load()
if clients == 0 &&
rooms == 0 &&
sessions == 0 &&
remoteSessions == 0 &&
readActive == 0 &&
writeActive == 0 {
break
}
select {
case <-ctx.Done():
h.mu.Lock()
h.ru.Lock()
dumpGoroutines("", os.Stderr)
t.Errorf("Error waiting for clients %+v / rooms %+v / sessions %+v / remoteSessions %v / %d read / %d write to terminate: %s", h.clients, h.rooms, h.sessions, h.remoteSessions, readActive, writeActive, ctx.Err())
h.ru.Unlock()
h.mu.Unlock()
return
default:
time.Sleep(time.Millisecond)
}
}
}
func validateBackendChecksum(t *testing.T, f func(http.ResponseWriter, *http.Request, *BackendClientRequest) *BackendClientResponse) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
body, err := io.ReadAll(r.Body)
if err != nil {
t.Fatal("Error reading body: ", err)
}
rnd := r.Header.Get(HeaderBackendSignalingRandom)
checksum := r.Header.Get(HeaderBackendSignalingChecksum)
if rnd == "" || checksum == "" {
t.Fatalf("No checksum headers found in request to %s", r.URL)
}
if verify := CalculateBackendChecksum(rnd, body, testBackendSecret); verify != checksum {
t.Fatalf("Backend checksum verification failed for request to %s", r.URL)
}
var request BackendClientRequest
if err := json.Unmarshal(body, &request); err != nil {
t.Fatal(err)
}
response := f(w, r, &request)
if response == nil {
// Function already returned a response.
return
}
data, err := json.Marshal(response)
if err != nil {
t.Fatal(err)
}
if r.Header.Get("OCS-APIRequest") != "" {
var ocs OcsResponse
ocs.Ocs = &OcsBody{
Meta: OcsMeta{
Status: "ok",
StatusCode: http.StatusOK,
Message: http.StatusText(http.StatusOK),
},
Data: (*json.RawMessage)(&data),
}
if data, err = json.Marshal(ocs); err != nil {
t.Fatal(err)
}
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
w.Write(data) // nolint
}
}
func processAuthRequest(t *testing.T, w http.ResponseWriter, r *http.Request, request *BackendClientRequest) *BackendClientResponse {
if request.Type != "auth" || request.Auth == nil {
t.Fatalf("Expected an auth backend request, got %+v", request)
}
var params TestBackendClientAuthParams
if request.Auth.Params != nil && len(*request.Auth.Params) > 0 {
if err := json.Unmarshal(*request.Auth.Params, &params); err != nil {
t.Fatal(err)
}
}
if params.UserId == "" {
params.UserId = testDefaultUserId
} else if params.UserId == authAnonymousUserId {
params.UserId = ""
}
response := &BackendClientResponse{
Type: "auth",
Auth: &BackendClientAuthResponse{
Version: BackendVersion,
UserId: params.UserId,
},
}
userdata := map[string]string{
"displayname": "Displayname " + params.UserId,
}
if data, err := json.Marshal(userdata); err != nil {
t.Fatal(err)
} else {
response.Auth.User = (*json.RawMessage)(&data)
}
return response
}
func processRoomRequest(t *testing.T, w http.ResponseWriter, r *http.Request, request *BackendClientRequest) *BackendClientResponse {
if request.Type != "room" || request.Room == nil {
t.Fatalf("Expected an room backend request, got %+v", request)
}
switch request.Room.RoomId {
case "test-room-slow":
time.Sleep(100 * time.Millisecond)
case "test-room-takeover-room-session":
// Additional checks for testcase "TestClientTakeoverRoomSession"
if request.Room.Action == "leave" && request.Room.UserId == "test-userid1" {
t.Errorf("Should not receive \"leave\" event for first user, received %+v", request.Room)
}
}
// Allow joining any room.
response := &BackendClientResponse{
Type: "room",
Room: &BackendClientRoomResponse{
Version: BackendVersion,
RoomId: request.Room.RoomId,
Properties: (*json.RawMessage)(&testRoomProperties),
},
}
switch request.Room.RoomId {
case "test-room-with-sessiondata":
data := map[string]string{
"userid": "userid-from-sessiondata",
}
tmp, err := json.Marshal(data)
if err != nil {
t.Fatalf("Could not marshal %+v: %s", data, err)
}
response.Room.Session = (*json.RawMessage)(&tmp)
case "test-room-initial-permissions":
permissions := []Permission{PERMISSION_MAY_PUBLISH_AUDIO}
response.Room.Permissions = &permissions
}
return response
}
var (
sessionRequestHander struct {
sync.Mutex
handlers map[*testing.T]func(*BackendClientSessionRequest)
}
)
func setSessionRequestHandler(t *testing.T, f func(*BackendClientSessionRequest)) {
sessionRequestHander.Lock()
defer sessionRequestHander.Unlock()
if sessionRequestHander.handlers == nil {
sessionRequestHander.handlers = make(map[*testing.T]func(*BackendClientSessionRequest))
}
if _, found := sessionRequestHander.handlers[t]; !found {
t.Cleanup(func() {
sessionRequestHander.Lock()
defer sessionRequestHander.Unlock()
delete(sessionRequestHander.handlers, t)
})
}
sessionRequestHander.handlers[t] = f
}
func clearSessionRequestHandler(t *testing.T) { // nolint
sessionRequestHander.Lock()
defer sessionRequestHander.Unlock()
delete(sessionRequestHander.handlers, t)
}
func processSessionRequest(t *testing.T, w http.ResponseWriter, r *http.Request, request *BackendClientRequest) *BackendClientResponse {
if request.Type != "session" || request.Session == nil {
t.Fatalf("Expected an session backend request, got %+v", request)
}
sessionRequestHander.Lock()
defer sessionRequestHander.Unlock()
if f, found := sessionRequestHander.handlers[t]; found {
f(request.Session)
}
response := &BackendClientResponse{
Type: "session",
Session: &BackendClientSessionResponse{
Version: BackendVersion,
RoomId: request.Session.RoomId,
},
}
return response
}
var pingRequests map[*testing.T][]*BackendClientRequest
func getPingRequests(t *testing.T) []*BackendClientRequest {
return pingRequests[t]
}
func clearPingRequests(t *testing.T) {
delete(pingRequests, t)
}
func storePingRequest(t *testing.T, request *BackendClientRequest) {
if entries, found := pingRequests[t]; !found {
if pingRequests == nil {
pingRequests = make(map[*testing.T][]*BackendClientRequest)
}
pingRequests[t] = []*BackendClientRequest{
request,
}
t.Cleanup(func() {
clearPingRequests(t)
})
} else {
pingRequests[t] = append(entries, request)
}
}
func processPingRequest(t *testing.T, w http.ResponseWriter, r *http.Request, request *BackendClientRequest) *BackendClientResponse {
if request.Type != "ping" || request.Ping == nil {
t.Fatalf("Expected an ping backend request, got %+v", request)
}
if request.Ping.RoomId == "test-room-with-sessiondata" {
if entries := request.Ping.Entries; len(entries) != 1 {
t.Errorf("Expected one entry, got %+v", entries)
} else {
if entries[0].UserId != "" {
t.Errorf("Expected empty userid, got %+v", entries[0])
}
}
}
storePingRequest(t, request)
response := &BackendClientResponse{
Type: "ping",
Ping: &BackendClientRingResponse{
Version: BackendVersion,
RoomId: request.Ping.RoomId,
},
}
return response
}
func ensureAuthTokens(t *testing.T) (string, string) {
if privateKey := os.Getenv("PRIVATE_AUTH_TOKEN_" + t.Name()); privateKey != "" {
publicKey := os.Getenv("PUBLIC_AUTH_TOKEN_" + t.Name())
if publicKey == "" {
// should not happen, always both keys are created
t.Fatal("public key is empty")
}
return privateKey, publicKey
}
var private []byte
var public []byte
if strings.Contains(t.Name(), "ECDSA") {
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
if err != nil {
t.Fatal(err)
}
private, err = x509.MarshalECPrivateKey(key)
if err != nil {
t.Fatal(err)
}
private = pem.EncodeToMemory(&pem.Block{
Type: "ECDSA PRIVATE KEY",
Bytes: private,
})
public, err = x509.MarshalPKIXPublicKey(&key.PublicKey)
if err != nil {
t.Fatal(err)
}
public = pem.EncodeToMemory(&pem.Block{
Type: "ECDSA PUBLIC KEY",
Bytes: public,
})
} else if strings.Contains(t.Name(), "Ed25519") {
publicKey, privateKey, err := ed25519.GenerateKey(rand.Reader)
if err != nil {
t.Fatal(err)
}
private, err = x509.MarshalPKCS8PrivateKey(privateKey)
if err != nil {
t.Fatal(err)
}
private = pem.EncodeToMemory(&pem.Block{
Type: "Ed25519 PRIVATE KEY",
Bytes: private,
})
public, err = x509.MarshalPKIXPublicKey(publicKey)
if err != nil {
t.Fatal(err)
}
public = pem.EncodeToMemory(&pem.Block{
Type: "Ed25519 PUBLIC KEY",
Bytes: public,
})
} else {
key, err := rsa.GenerateKey(rand.Reader, 1024)
if err != nil {
t.Fatal(err)
}
private = pem.EncodeToMemory(&pem.Block{
Type: "RSA PRIVATE KEY",
Bytes: x509.MarshalPKCS1PrivateKey(key),
})
public, err = x509.MarshalPKIXPublicKey(&key.PublicKey)
if err != nil {
t.Fatal(err)
}
public = pem.EncodeToMemory(&pem.Block{
Type: "RSA PUBLIC KEY",
Bytes: public,
})
}
privateKey := base64.StdEncoding.EncodeToString(private)
t.Setenv("PRIVATE_AUTH_TOKEN_"+t.Name(), privateKey)
publicKey := base64.StdEncoding.EncodeToString(public)
t.Setenv("PUBLIC_AUTH_TOKEN_"+t.Name(), publicKey)
return privateKey, publicKey
}
func getPrivateAuthToken(t *testing.T) (key interface{}) {
private, _ := ensureAuthTokens(t)
data, err := base64.StdEncoding.DecodeString(private)
if err != nil {
t.Fatal(err)
}
if strings.Contains(t.Name(), "ECDSA") {
key, err = jwt.ParseECPrivateKeyFromPEM(data)
} else if strings.Contains(t.Name(), "Ed25519") {
key, err = jwt.ParseEdPrivateKeyFromPEM(data)
} else {
key, err = jwt.ParseRSAPrivateKeyFromPEM(data)
}
if err != nil {
t.Fatal(err)
}
return key
}
func getPublicAuthToken(t *testing.T) (key interface{}) {
_, public := ensureAuthTokens(t)
data, err := base64.StdEncoding.DecodeString(public)
if err != nil {
t.Fatal(err)
}
if strings.Contains(t.Name(), "ECDSA") {
key, err = jwt.ParseECPublicKeyFromPEM(data)
} else if strings.Contains(t.Name(), "Ed25519") {
key, err = jwt.ParseEdPublicKeyFromPEM(data)
} else {
key, err = jwt.ParseRSAPublicKeyFromPEM(data)
}
if err != nil {
t.Fatal(err)
}
return key
}
func registerBackendHandler(t *testing.T, router *mux.Router) {
registerBackendHandlerUrl(t, router, "/")
}
func registerBackendHandlerUrl(t *testing.T, router *mux.Router, url string) {
handleFunc := validateBackendChecksum(t, func(w http.ResponseWriter, r *http.Request, request *BackendClientRequest) *BackendClientResponse {
switch request.Type {
case "auth":
return processAuthRequest(t, w, r, request)
case "room":
return processRoomRequest(t, w, r, request)
case "session":
return processSessionRequest(t, w, r, request)
case "ping":
return processPingRequest(t, w, r, request)
default:
t.Fatalf("Unsupported request received: %+v", request)
return nil
}
})
router.HandleFunc(url, handleFunc)
if !strings.HasSuffix(url, "/") {
url += "/"
}
handleCapabilitiesFunc := func(w http.ResponseWriter, r *http.Request) {
features := []string{
"foo",
"bar",
}
if strings.Contains(t.Name(), "V3Api") {
features = append(features, "signaling-v3")
}
signaling := map[string]interface{}{
"foo": "bar",
"baz": 42,
}
config := map[string]interface{}{
"signaling": signaling,
}
if strings.Contains(t.Name(), "MultiRoom") {
signaling[ConfigKeySessionPingLimit] = 2
}
useV2 := true
if os.Getenv("SKIP_V2_CAPABILITIES") != "" {
useV2 = false
}
if strings.Contains(t.Name(), "V2") && useV2 {
key := getPublicAuthToken(t)
public, err := x509.MarshalPKIXPublicKey(key)
if err != nil {
t.Fatal(err)
}
var pemType string
if strings.Contains(t.Name(), "ECDSA") {
pemType = "ECDSA PUBLIC KEY"
} else if strings.Contains(t.Name(), "Ed25519") {
pemType = "Ed25519 PUBLIC KEY"
} else {
pemType = "RSA PUBLIC KEY"
}
public = pem.EncodeToMemory(&pem.Block{
Type: pemType,
Bytes: public,
})
if strings.Contains(t.Name(), "Ed25519_Nextcloud") {
// Simulate Nextcloud which returns the Ed25519 key as base64-encoded data.
encoded := base64.StdEncoding.EncodeToString(key.(ed25519.PublicKey))
signaling[ConfigKeyHelloV2TokenKey] = encoded
} else {
signaling[ConfigKeyHelloV2TokenKey] = string(public)
}
}
spreedCapa, _ := json.Marshal(map[string]interface{}{
"features": features,
"config": config,
})
response := &CapabilitiesResponse{
Version: CapabilitiesVersion{
Major: 20,
},
Capabilities: map[string]*json.RawMessage{
"spreed": (*json.RawMessage)(&spreedCapa),
},
}
data, err := json.Marshal(response)
if err != nil {
t.Errorf("Could not marshal %+v: %s", response, err)
}
var ocs OcsResponse
ocs.Ocs = &OcsBody{
Meta: OcsMeta{
Status: "ok",
StatusCode: http.StatusOK,
Message: http.StatusText(http.StatusOK),
},
Data: (*json.RawMessage)(&data),
}
if data, err = json.Marshal(ocs); err != nil {
t.Fatal(err)
}
w.Header().Add("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
w.Write(data) // nolint
}
router.HandleFunc(url+"ocs/v2.php/cloud/capabilities", handleCapabilitiesFunc)
if strings.Contains(t.Name(), "V3Api") {
router.HandleFunc(url+"ocs/v2.php/apps/spreed/api/v3/signaling/backend", handleFunc)
} else {
router.HandleFunc(url+"ocs/v2.php/apps/spreed/api/v1/signaling/backend", handleFunc)
}
}
func performHousekeeping(hub *Hub, now time.Time) *sync.WaitGroup {
var wg sync.WaitGroup
wg.Add(1)
go func() {
hub.performHousekeeping(now)
wg.Done()
}()
return &wg
}
func TestInitialWelcome(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
client := NewTestClientContext(ctx, t, server, hub)
defer client.CloseWithBye()
msg, err := client.RunUntilMessage(ctx)
if err != nil {
t.Fatal(err)
}
if msg.Type != "welcome" {
t.Errorf("Expected \"welcome\" message, got %+v", msg)
} else if msg.Welcome.Version == "" {
t.Errorf("Expected welcome version, got %+v", msg)
} else if len(msg.Welcome.Features) == 0 {
t.Errorf("Expected welcome features, got %+v", msg)
}
}
func TestExpectClientHello(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
// The server will send an error and close the connection if no "Hello"
// is sent.
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
// Perform housekeeping in the future, this will cause the connection to
// be terminated due to the missing "Hello" request.
performHousekeeping(hub, time.Now().Add(initialHelloTimeout+time.Second))
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
message, err := client.RunUntilMessage(ctx)
if err := checkUnexpectedClose(err); err != nil {
t.Fatal(err)
}
message2, err := client.RunUntilMessage(ctx)
if message2 != nil {
t.Fatalf("Received multiple messages, already have %+v, also got %+v", message, message2)
}
if err := checkUnexpectedClose(err); err != nil {
t.Fatal(err)
}
if err := checkMessageType(message, "bye"); err != nil {
t.Error(err)
} else if message.Bye.Reason != "hello_timeout" {
t.Errorf("Expected \"hello_timeout\" reason, got %+v", message.Bye)
}
}
func TestExpectClientHelloUnsupportedVersion(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
params := TestBackendClientAuthParams{
UserId: testDefaultUserId,
}
if err := client.SendHelloParams(server.URL, "0.0", "", nil, params); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
message, err := client.RunUntilMessage(ctx)
if err := checkUnexpectedClose(err); err != nil {
t.Fatal(err)
}
if err := checkMessageType(message, "error"); err != nil {
t.Error(err)
} else if message.Error.Code != "invalid_hello_version" {
t.Errorf("Expected \"invalid_hello_version\" reason, got %+v", message.Error)
}
}
func TestClientHelloV1(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
if hello, err := client.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
}
}
func TestClientHelloV2(t *testing.T) {
CatchLogForTest(t)
for _, algo := range testHelloV2Algorithms {
t.Run(algo, func(t *testing.T) {
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHelloV2(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
data := hub.decodeSessionId(hello.Hello.SessionId, publicSessionName)
if data == nil {
t.Fatalf("Could not decode session id: %s", hello.Hello.SessionId)
}
hub.mu.RLock()
session := hub.sessions[data.Sid]
hub.mu.RUnlock()
if session == nil {
t.Fatalf("Could not get session for id %+v", data)
}
var userdata map[string]string
if err := json.Unmarshal(*session.UserData(), &userdata); err != nil {
t.Fatal(err)
}
if expected := "Displayname " + testDefaultUserId; userdata["displayname"] != expected {
t.Errorf("Expected displayname %s, got %s", expected, userdata["displayname"])
}
})
}
}
func TestClientHelloV2_IssuedInFuture(t *testing.T) {
CatchLogForTest(t)
for _, algo := range testHelloV2Algorithms {
t.Run(algo, func(t *testing.T) {
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
issuedAt := time.Now().Add(time.Minute)
expiresAt := issuedAt.Add(time.Second)
if err := client.SendHelloV2WithTimes(testDefaultUserId, issuedAt, expiresAt); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
message, err := client.RunUntilMessage(ctx)
if err := checkUnexpectedClose(err); err != nil {
t.Fatal(err)
}
if err := checkMessageType(message, "error"); err != nil {
t.Error(err)
} else if message.Error.Code != "token_not_valid_yet" {
t.Errorf("Expected \"token_not_valid_yet\" reason, got %+v", message.Error)
}
})
}
}
func TestClientHelloV2_Expired(t *testing.T) {
CatchLogForTest(t)
for _, algo := range testHelloV2Algorithms {
t.Run(algo, func(t *testing.T) {
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
issuedAt := time.Now().Add(-time.Minute)
if err := client.SendHelloV2WithTimes(testDefaultUserId, issuedAt, issuedAt.Add(time.Second)); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
message, err := client.RunUntilMessage(ctx)
if err := checkUnexpectedClose(err); err != nil {
t.Fatal(err)
}
if err := checkMessageType(message, "error"); err != nil {
t.Error(err)
} else if message.Error.Code != "token_expired" {
t.Errorf("Expected \"token_expired\" reason, got %+v", message.Error)
}
})
}
}
func TestClientHelloV2_IssuedAtMissing(t *testing.T) {
CatchLogForTest(t)
for _, algo := range testHelloV2Algorithms {
t.Run(algo, func(t *testing.T) {
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
var issuedAt time.Time
expiresAt := time.Now().Add(time.Minute)
if err := client.SendHelloV2WithTimes(testDefaultUserId, issuedAt, expiresAt); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
message, err := client.RunUntilMessage(ctx)
if err := checkUnexpectedClose(err); err != nil {
t.Fatal(err)
}
if err := checkMessageType(message, "error"); err != nil {
t.Error(err)
} else if message.Error.Code != "token_not_valid_yet" {
t.Errorf("Expected \"token_not_valid_yet\" reason, got %+v", message.Error)
}
})
}
}
func TestClientHelloV2_ExpiresAtMissing(t *testing.T) {
CatchLogForTest(t)
for _, algo := range testHelloV2Algorithms {
t.Run(algo, func(t *testing.T) {
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
issuedAt := time.Now().Add(-time.Minute)
var expiresAt time.Time
if err := client.SendHelloV2WithTimes(testDefaultUserId, issuedAt, expiresAt); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
message, err := client.RunUntilMessage(ctx)
if err := checkUnexpectedClose(err); err != nil {
t.Fatal(err)
}
if err := checkMessageType(message, "error"); err != nil {
t.Error(err)
} else if message.Error.Code != "token_expired" {
t.Errorf("Expected \"token_expired\" reason, got %+v", message.Error)
}
})
}
}
func TestClientHelloV2_CachedCapabilities(t *testing.T) {
CatchLogForTest(t)
for _, algo := range testHelloV2Algorithms {
t.Run(algo, func(t *testing.T) {
hub, _, _, server := CreateHubForTest(t)
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
// Simulate old-style Nextcloud without capabilities for Hello V2.
t.Setenv("SKIP_V2_CAPABILITIES", "1")
client1 := NewTestClient(t, server, hub)
defer client1.CloseWithBye()
if err := client1.SendHelloV1(testDefaultUserId + "1"); err != nil {
t.Fatal(err)
}
hello1, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
if hello1.Hello.UserId != testDefaultUserId+"1" {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId+"1", hello1.Hello)
}
if hello1.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello1.Hello)
}
// Simulate updated Nextcloud with capabilities for Hello V2.
t.Setenv("SKIP_V2_CAPABILITIES", "")
client2 := NewTestClient(t, server, hub)
defer client2.CloseWithBye()
if err := client2.SendHelloV2(testDefaultUserId + "2"); err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
if hello2.Hello.UserId != testDefaultUserId+"2" {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId+"2", hello2.Hello)
}
if hello2.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello2.Hello)
}
})
}
}
func TestClientHelloWithSpaces(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
userId := "test user with spaces"
if err := client.SendHello(userId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
if hello, err := client.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != userId {
t.Errorf("Expected \"%s\", got %+v", userId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
}
}
func TestClientHelloAllowAll(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTestWithConfig(t, func(server *httptest.Server) (*goconf.ConfigFile, error) {
config, err := getTestConfig(server)
if err != nil {
return nil, err
}
config.RemoveOption("backend", "allowed")
config.AddOption("backend", "allowall", "true")
return config, nil
})
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
if hello, err := client.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
}
}
func TestClientHelloSessionLimit(t *testing.T) {
CatchLogForTest(t)
for _, subtest := range clusteredTests {
t.Run(subtest, func(t *testing.T) {
t.Parallel()
var hub1 *Hub
var hub2 *Hub
var server1 *httptest.Server
var server2 *httptest.Server
if isLocalTest(t) {
var router1 *mux.Router
hub1, _, router1, server1 = CreateHubForTestWithConfig(t, func(server *httptest.Server) (*goconf.ConfigFile, error) {
config, err := getTestConfig(server)
if err != nil {
return nil, err
}
config.RemoveOption("backend", "allowed")
config.RemoveOption("backend", "secret")
config.AddOption("backend", "backends", "backend1, backend2")
config.AddOption("backend1", "url", server.URL+"/one")
config.AddOption("backend1", "secret", string(testBackendSecret))
config.AddOption("backend1", "sessionlimit", "1")
config.AddOption("backend2", "url", server.URL+"/two")
config.AddOption("backend2", "secret", string(testBackendSecret))
return config, nil
})
registerBackendHandlerUrl(t, router1, "/one")
registerBackendHandlerUrl(t, router1, "/two")
hub2 = hub1
server2 = server1
} else {
var router1 *mux.Router
var router2 *mux.Router
hub1, hub2, router1, router2, server1, server2 = CreateClusteredHubsForTestWithConfig(t, func(server *httptest.Server) (*goconf.ConfigFile, error) {
// Make sure all backends use the same server
if server1 == nil {
server1 = server
} else {
server = server1
}
config, err := getTestConfig(server)
if err != nil {
return nil, err
}
config.RemoveOption("backend", "allowed")
config.RemoveOption("backend", "secret")
config.AddOption("backend", "backends", "backend1, backend2")
config.AddOption("backend1", "url", server.URL+"/one")
config.AddOption("backend1", "secret", string(testBackendSecret))
config.AddOption("backend1", "sessionlimit", "1")
config.AddOption("backend2", "url", server.URL+"/two")
config.AddOption("backend2", "secret", string(testBackendSecret))
return config, nil
})
registerBackendHandlerUrl(t, router1, "/one")
registerBackendHandlerUrl(t, router1, "/two")
registerBackendHandlerUrl(t, router2, "/one")
registerBackendHandlerUrl(t, router2, "/two")
}
client := NewTestClient(t, server1, hub1)
defer client.CloseWithBye()
params1 := TestBackendClientAuthParams{
UserId: testDefaultUserId,
}
if err := client.SendHelloParams(server1.URL+"/one", HelloVersionV1, "client", nil, params1); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
if hello, err := client.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
}
// The second client can't connect as it would exceed the session limit.
client2 := NewTestClient(t, server2, hub2)
defer client2.CloseWithBye()
params2 := TestBackendClientAuthParams{
UserId: testDefaultUserId + "2",
}
if err := client2.SendHelloParams(server1.URL+"/one", HelloVersionV1, "client", nil, params2); err != nil {
t.Fatal(err)
}
msg, err := client2.RunUntilMessage(ctx)
if err != nil {
t.Error(err)
} else {
if msg.Type != "error" || msg.Error == nil {
t.Errorf("Expected error message, got %+v", msg)
} else if msg.Error.Code != "session_limit_exceeded" {
t.Errorf("Expected error \"session_limit_exceeded\", got %+v", msg.Error.Code)
}
}
// The client can connect to a different backend.
if err := client2.SendHelloParams(server1.URL+"/two", HelloVersionV1, "client", nil, params2); err != nil {
t.Fatal(err)
}
if hello, err := client2.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId+"2" {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId+"2", hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
}
// If the first client disconnects (and releases the session), a new one can connect.
client.CloseWithBye()
if err := client.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
client3 := NewTestClient(t, server2, hub2)
defer client3.CloseWithBye()
params3 := TestBackendClientAuthParams{
UserId: testDefaultUserId + "3",
}
if err := client3.SendHelloParams(server1.URL+"/one", HelloVersionV1, "client", nil, params3); err != nil {
t.Fatal(err)
}
if hello, err := client3.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId+"3" {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId+"3", hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
}
})
}
}
func TestSessionIdsUnordered(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
publicSessionIds := make([]string, 0)
for i := 0; i < 20; i++ {
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
if hello, err := client.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
break
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
break
}
data := hub.decodeSessionId(hello.Hello.SessionId, publicSessionName)
if data == nil {
t.Errorf("Could not decode session id: %s", hello.Hello.SessionId)
break
}
hub.mu.RLock()
session := hub.sessions[data.Sid]
hub.mu.RUnlock()
if session == nil {
t.Errorf("Could not get session for id %+v", data)
break
}
publicSessionIds = append(publicSessionIds, session.PublicId())
}
}
if len(publicSessionIds) == 0 {
t.Fatal("no session ids decoded")
}
larger := 0
smaller := 0
prevSid := ""
for i, sid := range publicSessionIds {
if i > 0 {
if sid > prevSid {
larger++
} else if sid < prevSid {
smaller--
} else {
t.Error("should not have received the same session id twice")
}
}
prevSid = sid
}
// Public session ids should not be ordered.
if len(publicSessionIds) == larger {
t.Error("the session ids are all larger than the previous ones")
} else if len(publicSessionIds) == smaller {
t.Error("the session ids are all smaller than the previous ones")
}
}
func TestClientHelloResume(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
client.Close()
if err := client.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
client = NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
hello2, err := client.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello2.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello2.Hello)
}
if hello2.Hello.SessionId != hello.Hello.SessionId {
t.Errorf("Expected session id %s, got %+v", hello.Hello.SessionId, hello2.Hello)
}
if hello2.Hello.ResumeId != hello.Hello.ResumeId {
t.Errorf("Expected resume id %s, got %+v", hello.Hello.ResumeId, hello2.Hello)
}
}
}
func TestClientHelloResumeExpired(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
client.Close()
if err := client.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
// Perform housekeeping in the future, this will cause the session to be
// cleaned up after it is expired.
performHousekeeping(hub, time.Now().Add(sessionExpireDuration+time.Second)).Wait()
client = NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
msg, err := client.RunUntilMessage(ctx)
if err != nil {
t.Error(err)
} else {
if msg.Type != "error" || msg.Error == nil {
t.Errorf("Expected error message, got %+v", msg)
} else if msg.Error.Code != "no_such_session" {
t.Errorf("Expected error \"no_such_session\", got %+v", msg.Error.Code)
}
}
}
func TestClientHelloResumeTakeover(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client1 := NewTestClient(t, server, hub)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client1.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
client2 := NewTestClient(t, server, hub)
defer client2.CloseWithBye()
if err := client2.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello2.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello2.Hello)
}
if hello2.Hello.SessionId != hello.Hello.SessionId {
t.Errorf("Expected session id %s, got %+v", hello.Hello.SessionId, hello2.Hello)
}
if hello2.Hello.ResumeId != hello.Hello.ResumeId {
t.Errorf("Expected resume id %s, got %+v", hello.Hello.ResumeId, hello2.Hello)
}
}
// The first client got disconnected with a reason in a "Bye" message.
msg, err := client1.RunUntilMessage(ctx)
if err != nil {
t.Error(err)
} else {
if msg.Type != "bye" || msg.Bye == nil {
t.Errorf("Expected bye message, got %+v", msg)
} else if msg.Bye.Reason != "session_resumed" {
t.Errorf("Expected reason \"session_resumed\", got %+v", msg.Bye.Reason)
}
}
if msg, err := client1.RunUntilMessage(ctx); err == nil {
t.Errorf("Expected error but received %+v", msg)
} else if !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
t.Errorf("Expected close error but received %+v", err)
}
}
func TestClientHelloResumeOtherHub(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
client.Close()
if err := client.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
// Simulate a restart of the hub.
hub.sid.Store(0)
sessions := make([]Session, 0)
hub.mu.Lock()
for _, session := range hub.sessions {
sessions = append(sessions, session)
}
hub.mu.Unlock()
for _, session := range sessions {
session.Close()
}
hub.mu.Lock()
count := len(hub.sessions)
hub.mu.Unlock()
if count > 0 {
t.Errorf("Should have removed all sessions (still has %d)", count)
}
// The new client will get the same (internal) sid for his session.
newClient := NewTestClient(t, server, hub)
defer newClient.CloseWithBye()
if err := newClient.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
if hello, err := newClient.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
// The previous session (which had the same internal sid) can't be resumed.
client = NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
msg, err := client.RunUntilMessage(ctx)
if err != nil {
t.Error(err)
} else {
if msg.Type != "error" || msg.Error == nil {
t.Errorf("Expected error message, got %+v", msg)
} else if msg.Error.Code != "no_such_session" {
t.Errorf("Expected error \"no_such_session\", got %+v", msg.Error.Code)
}
}
// Expire old sessions
hub.performHousekeeping(time.Now().Add(2 * sessionExpireDuration))
}
func TestClientHelloResumePublicId(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
// Test that a client can't resume a "public" session of another user.
hub, _, _, server := CreateHubForTest(t)
client1 := NewTestClient(t, server, hub)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId + "1"); err != nil {
t.Fatal(err)
}
client2 := NewTestClient(t, server, hub)
defer client2.CloseWithBye()
if err := client2.SendHello(testDefaultUserId + "2"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello1, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
if hello1.Hello.SessionId == hello2.Hello.SessionId {
t.Fatalf("Expected different session ids, got %s twice", hello1.Hello.SessionId)
}
recipient2 := MessageClientMessageRecipient{
Type: "session",
SessionId: hello2.Hello.SessionId,
}
data := "from-1-to-2"
client1.SendMessage(recipient2, data) // nolint
var payload string
var sender *MessageServerMessageSender
if err := checkReceiveClientMessageWithSender(ctx, client2, "session", hello1.Hello, &payload, &sender); err != nil {
t.Error(err)
} else if payload != data {
t.Errorf("Expected payload %s, got %s", data, payload)
}
client1.Close()
if err := client1.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
client1 = NewTestClient(t, server, hub)
defer client1.CloseWithBye()
// Can't resume a session with the id received from messages of a client.
if err := client1.SendHelloResume(sender.SessionId); err != nil {
t.Fatal(err)
}
msg, err := client1.RunUntilMessage(ctx)
if err != nil {
t.Error(err)
} else {
if msg.Type != "error" || msg.Error == nil {
t.Errorf("Expected error message, got %+v", msg)
} else if msg.Error.Code != "no_such_session" {
t.Errorf("Expected error \"no_such_session\", got %+v", msg.Error.Code)
}
}
// Expire old sessions
hub.performHousekeeping(time.Now().Add(2 * sessionExpireDuration))
}
func TestClientHelloByeResume(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
if err := client.SendBye(); err != nil {
t.Fatal(err)
}
if message, err := client.RunUntilMessage(ctx); err != nil {
t.Error(err)
} else {
if err := checkMessageType(message, "bye"); err != nil {
t.Error(err)
}
}
client.Close()
if err := client.WaitForSessionRemoved(ctx, hello.Hello.SessionId); err != nil {
t.Error(err)
}
if err := client.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
client = NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
msg, err := client.RunUntilMessage(ctx)
if err != nil {
t.Error(err)
} else {
if msg.Type != "error" || msg.Error == nil {
t.Errorf("Expected \"error\", got %+v", *msg)
} else if msg.Error.Code != "no_such_session" {
t.Errorf("Expected error \"no_such_session\", got %+v", *msg)
}
}
}
func TestClientHelloResumeAndJoin(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
client.Close()
if err := client.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
client = NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
hello2, err := client.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello2.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello2.Hello)
}
if hello2.Hello.SessionId != hello.Hello.SessionId {
t.Errorf("Expected session id %s, got %+v", hello.Hello.SessionId, hello2.Hello)
}
if hello2.Hello.ResumeId != hello.Hello.ResumeId {
t.Errorf("Expected resume id %s, got %+v", hello.Hello.ResumeId, hello2.Hello)
}
}
// Join room by id.
roomId := "test-room"
if room, err := client.JoinRoom(ctx, roomId); err != nil {
t.Fatal(err)
} else if room.Room.RoomId != roomId {
t.Fatalf("Expected room %s, got %s", roomId, room.Room.RoomId)
}
}
func runGrpcProxyTest(t *testing.T, f func(hub1, hub2 *Hub, server1, server2 *httptest.Server)) {
t.Helper()
var hub1 *Hub
var hub2 *Hub
var server1 *httptest.Server
var server2 *httptest.Server
var router1 *mux.Router
var router2 *mux.Router
hub1, hub2, router1, router2, server1, server2 = CreateClusteredHubsForTestWithConfig(t, func(server *httptest.Server) (*goconf.ConfigFile, error) {
// Make sure all backends use the same server
if server1 == nil {
server1 = server
} else {
server = server1
}
config, err := getTestConfig(server)
if err != nil {
return nil, err
}
config.RemoveOption("backend", "allowed")
config.RemoveOption("backend", "secret")
config.AddOption("backend", "backends", "backend1")
config.AddOption("backend1", "url", server.URL)
config.AddOption("backend1", "secret", string(testBackendSecret))
config.AddOption("backend1", "sessionlimit", "1")
return config, nil
})
registerBackendHandlerUrl(t, router1, "/")
registerBackendHandlerUrl(t, router2, "/")
f(hub1, hub2, server1, server2)
}
func TestClientHelloResumeProxy(t *testing.T) {
CatchLogForTest(t)
ensureNoGoroutinesLeak(t, func(t *testing.T) {
runGrpcProxyTest(t, func(hub1, hub2 *Hub, server1, server2 *httptest.Server) {
client1 := NewTestClient(t, server1, hub1)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
client1.Close()
if err := client1.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
client2 := NewTestClient(t, server2, hub2)
defer client2.CloseWithBye()
if err := client2.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello2.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello2.Hello)
}
if hello2.Hello.SessionId != hello.Hello.SessionId {
t.Errorf("Expected session id %s, got %+v", hello.Hello.SessionId, hello2.Hello)
}
if hello2.Hello.ResumeId != hello.Hello.ResumeId {
t.Errorf("Expected resume id %s, got %+v", hello.Hello.ResumeId, hello2.Hello)
}
}
// Join room by id.
roomId := "test-room"
if room, err := client2.JoinRoom(ctx, roomId); err != nil {
t.Fatal(err)
} else if room.Room.RoomId != roomId {
t.Fatalf("Expected room %s, got %s", roomId, room.Room.RoomId)
}
// We will receive a "joined" event.
if err := client2.RunUntilJoined(ctx, hello.Hello); err != nil {
t.Error(err)
}
if room := hub1.getRoom(roomId); room == nil {
t.Fatalf("Could not find room %s", roomId)
}
if room := hub2.getRoom(roomId); room != nil {
t.Fatalf("Should not have gotten room %s, got %+v", roomId, room)
}
users := []map[string]interface{}{
{
"sessionId": "the-session-id",
"inCall": 1,
},
}
room := hub1.getRoom(roomId)
if room == nil {
t.Fatalf("Could not find room %s", roomId)
}
room.PublishUsersInCallChanged(users, users)
if err := checkReceiveClientEvent(ctx, client2, "update", nil); err != nil {
t.Error(err)
}
})
})
}
func TestClientHelloResumeProxy_Takeover(t *testing.T) {
CatchLogForTest(t)
ensureNoGoroutinesLeak(t, func(t *testing.T) {
runGrpcProxyTest(t, func(hub1, hub2 *Hub, server1, server2 *httptest.Server) {
client1 := NewTestClient(t, server1, hub1)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
client2 := NewTestClient(t, server2, hub2)
defer client2.CloseWithBye()
if err := client2.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello2.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello2.Hello)
}
if hello2.Hello.SessionId != hello.Hello.SessionId {
t.Errorf("Expected session id %s, got %+v", hello.Hello.SessionId, hello2.Hello)
}
if hello2.Hello.ResumeId != hello.Hello.ResumeId {
t.Errorf("Expected resume id %s, got %+v", hello.Hello.ResumeId, hello2.Hello)
}
}
// The first client got disconnected with a reason in a "Bye" message.
if msg, err := client1.RunUntilMessage(ctx); err != nil {
t.Error(err)
} else {
if msg.Type != "bye" || msg.Bye == nil {
t.Errorf("Expected bye message, got %+v", msg)
} else if msg.Bye.Reason != "session_resumed" {
t.Errorf("Expected reason \"session_resumed\", got %+v", msg.Bye.Reason)
}
}
if msg, err := client1.RunUntilMessage(ctx); err == nil {
t.Errorf("Expected error but received %+v", msg)
} else if !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
t.Errorf("Expected close error but received %+v", err)
}
client3 := NewTestClient(t, server1, hub1)
defer client3.CloseWithBye()
if err := client3.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
hello3, err := client3.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello3.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello2.Hello)
}
if hello3.Hello.SessionId != hello.Hello.SessionId {
t.Errorf("Expected session id %s, got %+v", hello.Hello.SessionId, hello2.Hello)
}
if hello3.Hello.ResumeId != hello.Hello.ResumeId {
t.Errorf("Expected resume id %s, got %+v", hello.Hello.ResumeId, hello2.Hello)
}
}
// The second client got disconnected with a reason in a "Bye" message.
if msg, err := client2.RunUntilMessage(ctx); err != nil {
t.Error(err)
} else {
if msg.Type != "bye" || msg.Bye == nil {
t.Errorf("Expected bye message, got %+v", msg)
} else if msg.Bye.Reason != "session_resumed" {
t.Errorf("Expected reason \"session_resumed\", got %+v", msg.Bye.Reason)
}
}
if msg, err := client2.RunUntilMessage(ctx); err == nil {
t.Errorf("Expected error but received %+v", msg)
} else if !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseNoStatusReceived) {
t.Errorf("Expected close error but received %+v", err)
}
})
})
}
func TestClientHelloResumeProxy_Disconnect(t *testing.T) {
CatchLogForTest(t)
ensureNoGoroutinesLeak(t, func(t *testing.T) {
runGrpcProxyTest(t, func(hub1, hub2 *Hub, server1, server2 *httptest.Server) {
client1 := NewTestClient(t, server1, hub1)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
client1.Close()
if err := client1.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
client2 := NewTestClient(t, server2, hub2)
defer client2.CloseWithBye()
if err := client2.SendHelloResume(hello.Hello.ResumeId); err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Error(err)
} else {
if hello2.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello2.Hello)
}
if hello2.Hello.SessionId != hello.Hello.SessionId {
t.Errorf("Expected session id %s, got %+v", hello.Hello.SessionId, hello2.Hello)
}
if hello2.Hello.ResumeId != hello.Hello.ResumeId {
t.Errorf("Expected resume id %s, got %+v", hello.Hello.ResumeId, hello2.Hello)
}
}
// Simulate unclean shutdown of second instance.
hub2.rpcServer.conn.Stop()
if err := client2.WaitForClientRemoved(ctx); err != nil {
t.Error(err)
}
})
})
}
func TestClientHelloClient(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHelloClient(testDefaultUserId); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
if hello, err := client.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
}
func TestClientHelloClient_V3Api(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
params := TestBackendClientAuthParams{
UserId: testDefaultUserId,
}
// The "/api/v1/signaling/" URL will be changed to use "v3" as the "signaling-v3"
// feature is returned by the capabilities endpoint.
if err := client.SendHelloParams(server.URL+"/ocs/v2.php/apps/spreed/api/v1/signaling/backend", HelloVersionV1, "client", nil, params); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
if hello, err := client.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != testDefaultUserId {
t.Errorf("Expected \"%s\", got %+v", testDefaultUserId, hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
}
func TestClientHelloInternal(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client := NewTestClient(t, server, hub)
defer client.CloseWithBye()
if err := client.SendHelloInternal(); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
if hello, err := client.RunUntilHello(ctx); err != nil {
t.Error(err)
} else {
if hello.Hello.UserId != "" {
t.Errorf("Expected empty user id, got %+v", hello.Hello)
}
if hello.Hello.SessionId == "" {
t.Errorf("Expected session id, got %+v", hello.Hello)
}
if hello.Hello.ResumeId == "" {
t.Errorf("Expected resume id, got %+v", hello.Hello)
}
}
}
func TestClientMessageToSessionId(t *testing.T) {
CatchLogForTest(t)
for _, subtest := range clusteredTests {
t.Run(subtest, func(t *testing.T) {
t.Parallel()
var hub1 *Hub
var hub2 *Hub
var server1 *httptest.Server
var server2 *httptest.Server
if isLocalTest(t) {
hub1, _, _, server1 = CreateHubForTest(t)
hub2 = hub1
server2 = server1
} else {
hub1, hub2, server1, server2 = CreateClusteredHubsForTest(t)
}
client1 := NewTestClient(t, server1, hub1)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId + "1"); err != nil {
t.Fatal(err)
}
client2 := NewTestClient(t, server2, hub2)
defer client2.CloseWithBye()
if err := client2.SendHello(testDefaultUserId + "2"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello1, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
if hello1.Hello.SessionId == hello2.Hello.SessionId {
t.Fatalf("Expected different session ids, got %s twice", hello1.Hello.SessionId)
}
recipient1 := MessageClientMessageRecipient{
Type: "session",
SessionId: hello1.Hello.SessionId,
}
recipient2 := MessageClientMessageRecipient{
Type: "session",
SessionId: hello2.Hello.SessionId,
}
data1 := "from-1-to-2"
client1.SendMessage(recipient2, data1) // nolint
data2 := "from-2-to-1"
client2.SendMessage(recipient1, data2) // nolint
var payload string
if err := checkReceiveClientMessage(ctx, client1, "session", hello2.Hello, &payload); err != nil {
t.Error(err)
} else if payload != data2 {
t.Errorf("Expected payload %s, got %s", data2, payload)
}
if err := checkReceiveClientMessage(ctx, client2, "session", hello1.Hello, &payload); err != nil {
t.Error(err)
} else if payload != data1 {
t.Errorf("Expected payload %s, got %s", data1, payload)
}
})
}
}
func TestClientControlToSessionId(t *testing.T) {
CatchLogForTest(t)
for _, subtest := range clusteredTests {
t.Run(subtest, func(t *testing.T) {
t.Parallel()
var hub1 *Hub
var hub2 *Hub
var server1 *httptest.Server
var server2 *httptest.Server
if isLocalTest(t) {
hub1, _, _, server1 = CreateHubForTest(t)
hub2 = hub1
server2 = server1
} else {
hub1, hub2, server1, server2 = CreateClusteredHubsForTest(t)
}
client1 := NewTestClient(t, server1, hub1)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId + "1"); err != nil {
t.Fatal(err)
}
client2 := NewTestClient(t, server2, hub2)
defer client2.CloseWithBye()
if err := client2.SendHello(testDefaultUserId + "2"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello1, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
if hello1.Hello.SessionId == hello2.Hello.SessionId {
t.Fatalf("Expected different session ids, got %s twice", hello1.Hello.SessionId)
}
recipient1 := MessageClientMessageRecipient{
Type: "session",
SessionId: hello1.Hello.SessionId,
}
recipient2 := MessageClientMessageRecipient{
Type: "session",
SessionId: hello2.Hello.SessionId,
}
data1 := "from-1-to-2"
client1.SendControl(recipient2, data1) // nolint
data2 := "from-2-to-1"
client2.SendControl(recipient1, data2) // nolint
var payload string
if err := checkReceiveClientControl(ctx, client1, "session", hello2.Hello, &payload); err != nil {
t.Error(err)
} else if payload != data2 {
t.Errorf("Expected payload %s, got %s", data2, payload)
}
if err := checkReceiveClientControl(ctx, client2, "session", hello1.Hello, &payload); err != nil {
t.Error(err)
} else if payload != data1 {
t.Errorf("Expected payload %s, got %s", data1, payload)
}
})
}
}
func TestClientControlMissingPermissions(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client1 := NewTestClient(t, server, hub)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId + "1"); err != nil {
t.Fatal(err)
}
client2 := NewTestClient(t, server, hub)
defer client2.CloseWithBye()
if err := client2.SendHello(testDefaultUserId + "2"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello1, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
if hello1.Hello.SessionId == hello2.Hello.SessionId {
t.Fatalf("Expected different session ids, got %s twice", hello1.Hello.SessionId)
}
session1 := hub.GetSessionByPublicId(hello1.Hello.SessionId).(*ClientSession)
if session1 == nil {
t.Fatalf("Session %s does not exist", hello1.Hello.SessionId)
}
session2 := hub.GetSessionByPublicId(hello2.Hello.SessionId).(*ClientSession)
if session2 == nil {
t.Fatalf("Session %s does not exist", hello2.Hello.SessionId)
}
// Client 1 may not send control messages (will be ignored).
session1.SetPermissions([]Permission{
PERMISSION_MAY_PUBLISH_AUDIO,
PERMISSION_MAY_PUBLISH_VIDEO,
})
// Client 2 may send control messages.
session2.SetPermissions([]Permission{
PERMISSION_MAY_PUBLISH_AUDIO,
PERMISSION_MAY_PUBLISH_VIDEO,
PERMISSION_MAY_CONTROL,
})
recipient1 := MessageClientMessageRecipient{
Type: "session",
SessionId: hello1.Hello.SessionId,
}
recipient2 := MessageClientMessageRecipient{
Type: "session",
SessionId: hello2.Hello.SessionId,
}
data1 := "from-1-to-2"
client1.SendControl(recipient2, data1) // nolint
data2 := "from-2-to-1"
client2.SendControl(recipient1, data2) // nolint
var payload string
if err := checkReceiveClientControl(ctx, client1, "session", hello2.Hello, &payload); err != nil {
t.Error(err)
} else if payload != data2 {
t.Errorf("Expected payload %s, got %s", data2, payload)
}
ctx2, cancel2 := context.WithTimeout(context.Background(), 100*time.Millisecond)
defer cancel2()
if err := checkReceiveClientMessage(ctx2, client2, "session", hello1.Hello, &payload); err != nil {
if err != ErrNoMessageReceived {
t.Error(err)
}
} else {
t.Errorf("Expected no payload, got %+v", payload)
}
}
func TestClientMessageToUserId(t *testing.T) {
t.Parallel()
CatchLogForTest(t)
hub, _, _, server := CreateHubForTest(t)
client1 := NewTestClient(t, server, hub)
defer client1.CloseWithBye()
if err := client1.SendHello(testDefaultUserId + "1"); err != nil {
t.Fatal(err)
}
client2 := NewTestClient(t, server, hub)
defer client2.CloseWithBye()
if err := client2.SendHello(testDefaultUserId + "2"); err != nil {
t.Fatal(err)
}
ctx, cancel := context.WithTimeout(context.Background(), testTimeout)
defer cancel()
hello1, err := client1.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
hello2, err := client2.RunUntilHello(ctx)
if err != nil {
t.Fatal(err)
}
if hello1.Hello.SessionId == hello2.Hello.SessionId {
t.Fatalf("Expected different session ids, got %s twice", hello1.Hello.SessionId)
} else if hello1.Hello.UserId == hello2.Hello.UserId {
t.Fatalf("Expected different user ids, got %s twice", hello1.Hello.UserId)
}
recipient1 := MessageClientMessageRecipient{
Type: "user",
UserId: hello1.Hello.UserId,
}
recipient2 := MessageClientMessageRecipient{
Type: "user",
UserId: hello2.Hello.UserId,
}
data1 := "from-1-to-2"
client1.SendMessage(recipient2, data1) // nolint
data2 := "from-2-to-1"
client2.SendMessage(recipient1, data2) // nolint
var payload string
if err := checkReceiveClientMessage(ctx, client1, "user", hello2.Hello, &payload); err != nil {