Get fallback key tests working for all clients

TEST_HITLIST.md

@@ -11,7 +11,8 @@ Key backups:
 - [x] Inputting the wrong recovery key fails to decrypt the backup.
 
 One-time Keys:
-- [ ] When Alice runs out of OTKs, the fallback key is used. It is cycled when Alice becomes aware that it has been used.
+- [x] When Alice runs out of OTKs, the fallback key is used.
+- [x] Alice cycles her fallback key when she becomes aware that it has been used.
 - [ ] When a OTK is reused, Alice... (TODO: ??? rejects both, rejects latest, rejects neither?)
 
 Key Verification: (Short Authentication String)

internal/api/js/js.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -52,16 +53,18 @@ func writeToLog(s string, args ...interface{}) {
 }
 
 type JSClient struct {
-	browser    *chrome.Browser
-	listeners  map[int32]func(roomID string, ev api.Event)
-	listenerID atomic.Int32
-	userID     string
+	browser     *chrome.Browser
+	listeners   map[int32]func(roomID string, ev api.Event)
+	listenerID  atomic.Int32
+	listenersMu *sync.RWMutex
+	userID      string
 }
 
 func NewJSClient(t ct.TestLike, opts api.ClientCreationOpts) (api.Client, error) {
 	jsc := &JSClient{
-		listeners: make(map[int32]func(roomID string, ev api.Event)),
-		userID:    opts.UserID,
+		listeners:   make(map[int32]func(roomID string, ev api.Event)),
+		userID:      opts.UserID,
+		listenersMu: &sync.RWMutex{},
 	}
 	portKey := opts.UserID + opts.DeviceID
 	browser, err := chrome.RunHeadless(func(s string) {
@@ -77,7 +80,13 @@ func NewJSClient(t ct.TestLike, opts api.ClientCreationOpts) (api.Client, error)
 				writeToLog("[%s] failed to unmarshal event '%s' into Go %s\n", opts.UserID, segs[1], err)
 				return
 			}
+			jsc.listenersMu.RLock()
+			var listeners []func(roomID string, ev api.Event)
 			for _, l := range jsc.listeners {
+				listeners = append(listeners, l)
+			}
+			jsc.listenersMu.RUnlock()
+			for _, l := range listeners {
 				l(segs[0], jsToEvent(ev))
 			}
 		}
@@ -229,7 +238,9 @@ func (c *JSClient) Close(t ct.TestLike) {
 		console.log("=================== localstorage len", window.localStorage.length);
 	`)
 	c.browser.Cancel()
+	c.listenersMu.Lock()
 	c.listeners = make(map[int32]func(roomID string, ev api.Event))
+	c.listenersMu.Unlock()
 }
 
 func (c *JSClient) UserID() string {
@@ -432,9 +443,13 @@ func (c *JSClient) Type() api.ClientTypeLang {
 
 func (c *JSClient) listenForUpdates(callback func(roomID string, ev api.Event)) (cancel func()) {
 	id := c.listenerID.Add(1)
+	c.listenersMu.Lock()
 	c.listeners[id] = callback
+	c.listenersMu.Unlock()
 	return func() {
+		c.listenersMu.Lock()
 		delete(c.listeners, id)
+		c.listenersMu.Unlock()
 	}
 }
 

internal/deploy/deploy.go

@@ -254,10 +254,11 @@ func RunNewDeployment(t *testing.T, shouldTCPDump bool) *SlidingSyncDeployment {
 				Image:        "ghcr.io/matrix-org/sliding-sync:v0.99.14",
 				ExposedPorts: []string{ssExposedPort},
 				Env: map[string]string{
-					"SYNCV3_SECRET":   "secret",
-					"SYNCV3_BINDADDR": ":6789",
-					"SYNCV3_SERVER":   "http://hs1:8008",
-					"SYNCV3_DB":       "user=postgres dbname=syncv3 sslmode=disable password=postgres host=postgres",
+					"SYNCV3_SECRET":    "secret",
+					"SYNCV3_BINDADDR":  ":6789",
+					"SYNCV3_SERVER":    "http://hs1:8008",
+					"SYNCV3_LOG_LEVEL": "trace",
+					"SYNCV3_DB":        "user=postgres dbname=syncv3 sslmode=disable password=postgres host=postgres",
 				},
 				WaitingFor: wait.ForLog("listening on"),
 				Networks:   []string{networkName},

tests/one_time_keys_test.go

@@ -16,6 +16,7 @@ import (
 )
 
 func mustClaimFallbackKey(t *testing.T, claimer *client.CSAPI, target *client.CSAPI) (fallbackKeyID string, keyJSON gjson.Result) {
+	t.Helper()
 	res := claimer.MustDo(t, "POST", []string{
 		"_matrix", "client", "v3", "keys", "claim",
 	}, client.WithJSONBody(t, map[string]any{
@@ -43,6 +44,7 @@ func mustClaimFallbackKey(t *testing.T, claimer *client.CSAPI, target *client.CS
 }
 
 func mustClaimOTKs(t *testing.T, claimer *client.CSAPI, target *client.CSAPI, otkCount int) {
+	t.Helper()
 	for i := 0; i < otkCount; i++ {
 		res := claimer.MustDo(t, "POST", []string{
 			"_matrix", "client", "v3", "keys", "claim",
@@ -95,6 +97,11 @@ func TestFallbackKeyIsUsedIfOneTimeKeysRunOut(t *testing.T) {
 		aliceStopSyncing := alice.MustStartSyncing(t)
 		defer aliceStopSyncing()
 
+		// we need to send _something_ to cause /sync v2 to return a long poll response, as fallback
+		// keys don't wake up /sync v2. If we don't do this, rust SDK fails to realise it needs to upload a fallback
+		// key because SS doesn't tell it, because Synapse doesn't tell SS that the fallback key was used.
+		tc.Alice.MustCreateRoom(t, map[string]interface{}{})
+
 		// also let bob upload OTKs before we block the upload endpoint!
 		bob := LoginClientFromComplementClient(t, tc.Deployment, tc.Bob, clientTypeB)
 		defer bob.Close(t)
@@ -127,9 +134,9 @@ func TestFallbackKeyIsUsedIfOneTimeKeysRunOut(t *testing.T) {
 			// now bob tries to talk to alice, the fallback key should be used
 			roomID = tc.CreateNewEncryptedRoom(t, tc.Bob, "public_chat", []string{tc.Alice.UserID})
 			tc.Alice.MustJoinRoom(t, roomID, []string{clientTypeB.HS})
-			w := alice.WaitUntilEventInRoom(t, roomID, api.CheckEventHasMembership(alice.UserID(), "join"))
+			w := bob.WaitUntilEventInRoom(t, roomID, api.CheckEventHasMembership(alice.UserID(), "join"))
 			w.Wait(t, 5*time.Second)
-			w = bob.WaitUntilEventInRoom(t, roomID, api.CheckEventHasMembership(bob.UserID(), "join"))
+			w = alice.WaitUntilEventInRoom(t, roomID, api.CheckEventHasMembership(alice.UserID(), "join"))
 			w.Wait(t, 5*time.Second)
 			bob.SendMessage(t, roomID, "Hello world!")
 			waiter = alice.WaitUntilEventInRoom(t, roomID, api.CheckEventHasBody("Hello world!"))
@@ -146,16 +153,32 @@ func TestFallbackKeyIsUsedIfOneTimeKeysRunOut(t *testing.T) {
 		t.Logf("first fallback key %s => %s", fallbackKeyID, fallbackKey.Get("key").Str)
 
 		tc.Alice.MustSyncUntil(t, client.SyncReq{}, func(clientUserID string, topLevelSyncJSON gjson.Result) error {
-			otkCount := topLevelSyncJSON.Get("device_one_time_keys_count.signed_curve25519").Int()
+			otkCount = topLevelSyncJSON.Get("device_one_time_keys_count.signed_curve25519").Int()
 			t.Logf("Alice otk count = %d", otkCount)
 			if otkCount == 0 {
 				return fmt.Errorf("alice hasn't re-uploaded OTKs yet")
 			}
 			return nil
 		})
 
-		// TODO: now re-block /keys/upload, re-claim all otks, and check that the fallback key this time around is different
+		// now re-block /keys/upload, re-claim all otks, and check that the fallback key this time around is different
 		// to the first
+		tc.Deployment.WithMITMOptions(t, map[string]interface{}{
+			"statuscode": map[string]interface{}{
+				"return_status": http.StatusGatewayTimeout,
+				"block_request": true,
+				"filter":        "~u .*\\/keys\\/upload.*",
+			},
+		}, func() {
+			// claim all OTKs
+			mustClaimOTKs(t, otkGobbler, tc.Alice, int(otkCount))
+
+			// now claim the fallback key
+			secondFallbackKeyID, secondFallbackKey := mustClaimFallbackKey(t, otkGobbler, tc.Alice)
+			t.Logf("second fallback key %s => %s", secondFallbackKeyID, secondFallbackKey.Get("key").Str)
+			must.NotEqual(t, secondFallbackKeyID, fallbackKeyID, "fallback key id same as before, not cycled?")
+			must.NotEqual(t, fallbackKey.Get("key").Str, secondFallbackKey.Get("key").Str, "fallback key data same as before, not cycled?")
+		})
 
 	})
 }