Add tests to exercise duplicate session ID handling

psiphon/server/server_test.go

@@ -740,7 +740,7 @@ func runServer(t *testing.T, runConfig *runServerConfig) {
 
 	// TODO: monitor logs for more robust wait-until-loaded. For example,
 	// especially with the race detector on, QUIC-OSSH tests can fail as the
-	// client sends its initial pacjet before the server is ready.
+	// client sends its initial packet before the server is ready.
 	time.Sleep(1 * time.Second)
 
 	// Test: hot reload (of psinet and traffic rules)

psiphon/server/sessionID_test.go

@@ -0,0 +1,279 @@
+/*
+ * Copyright (c) 2016, Psiphon Inc.
+ * All rights reserved.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU 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 General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package server
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/Psiphon-Labs/goarista/monotime"
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon"
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common"
+	"github.com/Psiphon-Labs/psiphon-tunnel-core/psiphon/common/protocol"
+)
+
+func TestDuplicateSessionID(t *testing.T) {
+
+	testDataDirName, err := ioutil.TempDir("", "psiphond-duplicate-session-id-test")
+	if err != nil {
+		t.Fatalf("TempDir failed: %s", err)
+	}
+	defer os.RemoveAll(testDataDirName)
+
+	psiphon.SetEmitDiagnosticNotices(true, true)
+
+	// Configure server
+
+	generateConfigParams := &GenerateConfigParams{
+		ServerIPAddress:      "127.0.0.1",
+		EnableSSHAPIRequests: true,
+		WebServerPort:        8000,
+		TunnelProtocolPorts:  map[string]int{"OSSH": 4000},
+	}
+
+	serverConfigJSON, _, _, _, encodedServerEntry, err := GenerateConfig(generateConfigParams)
+	if err != nil {
+		t.Fatalf("error generating server config: %s", err)
+	}
+
+	var serverConfig map[string]interface{}
+	json.Unmarshal(serverConfigJSON, &serverConfig)
+
+	serverConfig["LogFilename"] = filepath.Join(testDataDirName, "psiphond.log")
+	serverConfig["LogLevel"] = "debug"
+
+	serverConfigJSON, _ = json.Marshal(serverConfig)
+
+	numConcurrentClients := 100
+
+	stoppingEvent := "stopping existing client with duplicate session ID"
+	abortingEvent := "aborting new client with duplicate session ID"
+
+	// Sufficiently buffer channel so log callback handler doesn't cause server
+	// operations to block while handling concurrent clients.
+	duplicateSessionIDEvents := make(chan string, numConcurrentClients)
+
+	setLogCallback(func(log []byte) {
+		strLog := string(log)
+		var event string
+		if strings.Contains(strLog, stoppingEvent) {
+			event = stoppingEvent
+		} else if strings.Contains(strLog, abortingEvent) {
+			event = abortingEvent
+		}
+		if event != "" {
+			select {
+			case duplicateSessionIDEvents <- event:
+			default:
+			}
+		}
+
+	})
+
+	// Run server
+
+	serverWaitGroup := new(sync.WaitGroup)
+	serverWaitGroup.Add(1)
+	go func() {
+		defer serverWaitGroup.Done()
+		err := RunServices(serverConfigJSON)
+		if err != nil {
+			t.Fatalf("error running server: %s", err)
+		}
+	}()
+
+	defer func() {
+		p, _ := os.FindProcess(os.Getpid())
+		p.Signal(os.Interrupt)
+		serverWaitGroup.Wait()
+	}()
+
+	// TODO: monitor logs for more robust wait-until-loaded.
+	time.Sleep(1 * time.Second)
+
+	// Initialize tunnel clients. Bypassing Controller and using Tunnel directly
+	// to permit multiple concurrent clients.
+	//
+	// Limitation: all tunnels still use one singleton datastore and notice
+	// handler.
+
+	psiphon.SetNoticeWriter(ioutil.Discard)
+
+	clientConfigJSONTemplate := `
+    {
+        "DataStoreDirectory" : "%s",
+        "SponsorId" : "0",
+        "PropagationChannelId" : "0",
+        "SessionID" : "00000000000000000000000000000000"
+    }`
+
+	clientConfigJSON := fmt.Sprintf(
+		clientConfigJSONTemplate,
+		testDataDirName)
+
+	clientConfig, err := psiphon.LoadConfig([]byte(clientConfigJSON))
+	if err != nil {
+		t.Fatalf("LoadConfig failed: %s", err)
+	}
+	err = clientConfig.Commit()
+	if err != nil {
+		t.Fatalf("Commit failed: %s", err)
+	}
+
+	err = psiphon.OpenDataStore(clientConfig)
+	if err != nil {
+		t.Fatalf("OpenDataStore failed: %s", err)
+	}
+	defer psiphon.CloseDataStore()
+
+	serverEntry, err := protocol.DecodeServerEntry(
+		string(encodedServerEntry),
+		common.GetCurrentTimestamp(),
+		protocol.SERVER_ENTRY_SOURCE_EMBEDDED)
+	if err != nil {
+		t.Fatalf("DecodeServerEntry failed: %s", err)
+	}
+
+	dialTunnel := func(ctx context.Context) *psiphon.Tunnel {
+
+		dialParams, err := psiphon.MakeDialParameters(
+			clientConfig,
+			func(_ *protocol.ServerEntry, _ string) bool { return false },
+			func(_ *protocol.ServerEntry) (string, bool) { return "OSSH", true },
+			serverEntry,
+			false,
+			1)
+		if err != nil {
+			t.Fatalf("MakeDialParameters failed: %s", err)
+		}
+
+		tunnel, err := psiphon.ConnectTunnel(
+			ctx,
+			clientConfig,
+			monotime.Now(),
+			dialParams)
+		if err != nil {
+			t.Fatalf("ConnectTunnel failed: %s", err)
+		}
+
+		return tunnel
+	}
+
+	handshakeTunnel := func(tunnel *psiphon.Tunnel, expectSuccess bool) {
+
+		_, err = psiphon.NewServerContext(tunnel)
+
+		if expectSuccess && err != nil || (!expectSuccess && err == nil) {
+			t.Fatalf("Unexpected handshake result: %s", err)
+		}
+	}
+
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	defer cancelFunc()
+
+	// Test: normal case
+	//
+	// First tunnel, t1, fully establishes and then is superceded by new tunnel, t2.
+
+	t1 := dialTunnel(ctx)
+
+	handshakeTunnel(t1, true)
+
+	t2 := dialTunnel(ctx)
+
+	expectEvent := <-duplicateSessionIDEvents
+
+	if expectEvent != stoppingEvent {
+		t.Fatalf("Unexpected duplicate session ID event")
+	}
+
+	handshakeTunnel(t2, true)
+
+	t1.Close(true)
+	t2.Close(true)
+
+	// Test: simultaneous/interleaved case
+	//
+	// First tunnel connects but then tries to handshake after second tunnel has
+	// connected.
+
+	t1 = dialTunnel(ctx)
+
+	// TODO: await log confirmation that t1 completed registerEstablishedClient?
+	// Otherwise, there's some small chance that t2 is the "first" tunnel and the
+	// test could fail (false negative).
+
+	t2 = dialTunnel(ctx)
+
+	expectEvent = <-duplicateSessionIDEvents
+
+	if expectEvent != stoppingEvent {
+		t.Fatalf("Unexpected duplicate session ID event")
+	}
+
+	handshakeTunnel(t1, false)
+
+	handshakeTunnel(t2, true)
+
+	t1.Close(true)
+	t2.Close(true)
+
+	// Test: 100 concurrent clients, all with the same session ID.
+	//
+	// This should be enough concurrent clients to trigger both the "stopping"
+	// and "aborting" duplicate session ID cases.
+
+	tunnels := make([]*psiphon.Tunnel, numConcurrentClients)
+
+	waitGroup := new(sync.WaitGroup)
+	for i := 0; i < numConcurrentClients; i++ {
+		waitGroup.Add(1)
+		go func(i int) {
+			defer waitGroup.Done()
+			tunnels[i] = dialTunnel(ctx)
+		}(i)
+	}
+	waitGroup.Wait()
+
+	for _, t := range tunnels {
+		t.Close(true)
+	}
+
+	receivedEvents := make(map[string]int)
+	for i := 0; i < numConcurrentClients-1; i++ {
+		receivedEvents[<-duplicateSessionIDEvents] += 1
+	}
+
+	if receivedEvents[stoppingEvent] < 1 {
+		t.Fatalf("No stopping events received")
+	}
+
+	if receivedEvents[abortingEvent] < 1 {
+		t.Fatalf("No aborting events received")
+	}
+}