Test the scope of a transaction IDs after using refresh token

internal/client/client.go

@@ -264,8 +264,14 @@ func (c *CSAPI) SetPushRule(t *testing.T, scope string, kind string, ruleID stri
 // SendEventUnsynced sends `e` into the room.
 // Returns the event ID of the sent event.
 func (c *CSAPI) SendEventUnsynced(t *testing.T, roomID string, e b.Event) string {
-	t.Helper()
 	txnID := int(atomic.AddInt64(&c.txnID, 1))
+	return c.SendEventUnsyncedWithTxnID(t, roomID, e, txnID)
+}
+
+// SendEventUnsynced sends `e` into the room.
+// Returns the event ID of the sent event.
+func (c *CSAPI) SendEventUnsyncedWithTxnID(t *testing.T, roomID string, e b.Event, txnID int) string {
+	t.Helper()
 	paths := []string{"_matrix", "client", "v3", "rooms", roomID, "send", e.Type, strconv.Itoa(txnID)}
 	if e.StateKey != nil {
 		paths = []string{"_matrix", "client", "v3", "rooms", roomID, "state", e.Type, *e.StateKey}
@@ -438,7 +444,81 @@ func (c *CSAPI) LoginUser(t *testing.T, localpart, password string) (userID, acc
 	return userID, accessToken, deviceID
 }
 
-//RegisterUser will register the user with given parameters and
+// LoginUserWithDeviceID will log in to a homeserver on an existing device
+func (c *CSAPI) LoginUserWithDeviceID(t *testing.T, localpart, password, deviceID string) (userID, accessToken string) {
+	t.Helper()
+	reqBody := map[string]interface{}{
+		"identifier": map[string]interface{}{
+			"type": "m.id.user",
+			"user": localpart,
+		},
+		"device_id": deviceID,
+		"password":  password,
+		"type":      "m.login.password",
+	}
+	res := c.MustDoFunc(t, "POST", []string{"_matrix", "client", "v3", "login"}, WithJSONBody(t, reqBody))
+
+	body, err := ioutil.ReadAll(res.Body)
+	if err != nil {
+		t.Fatalf("unable to read response body: %v", err)
+	}
+
+	userID = gjson.GetBytes(body, "user_id").Str
+	accessToken = gjson.GetBytes(body, "access_token").Str
+	if gjson.GetBytes(body, "device_id").Str != deviceID {
+		t.Fatalf("device_id returned by login does not match the one requested")
+	}
+	return userID, accessToken
+}
+
+// LoginUserWithRefreshToken will log in to a homeserver, with refresh token enabled,
+// and create a new device on an existing user.
+func (c *CSAPI) LoginUserWithRefreshToken(t *testing.T, localpart, password string) (userID, accessToken, refreshToken, deviceID string, expiresInMs int64) {
+	t.Helper()
+	reqBody := map[string]interface{}{
+		"identifier": map[string]interface{}{
+			"type": "m.id.user",
+			"user": localpart,
+		},
+		"password":      password,
+		"type":          "m.login.password",
+		"refresh_token": true,
+	}
+	res := c.MustDoFunc(t, "POST", []string{"_matrix", "client", "v3", "login"}, WithJSONBody(t, reqBody))
+
+	body, err := ioutil.ReadAll(res.Body)
+	if err != nil {
+		t.Fatalf("unable to read response body: %v", err)
+	}
+
+	userID = gjson.GetBytes(body, "user_id").Str
+	accessToken = gjson.GetBytes(body, "access_token").Str
+	deviceID = gjson.GetBytes(body, "device_id").Str
+	refreshToken = gjson.GetBytes(body, "refresh_token").Str
+	expiresInMs = gjson.GetBytes(body, "expires_in_ms").Int()
+	return userID, accessToken, refreshToken, deviceID, expiresInMs
+}
+
+// RefreshToken will consume a refresh token and return a new access token and refresh token.
+func (c *CSAPI) ConsumeRefreshToken(t *testing.T, refreshToken string) (newAccessToken, newRefreshToken string, expiresInMs int64) {
+	t.Helper()
+	reqBody := map[string]interface{}{
+		"refresh_token": refreshToken,
+	}
+	res := c.MustDoFunc(t, "POST", []string{"_matrix", "client", "v3", "refresh"}, WithJSONBody(t, reqBody))
+
+	body, err := ioutil.ReadAll(res.Body)
+	if err != nil {
+		t.Fatalf("unable to read response body: %v", err)
+	}
+
+	newAccessToken = gjson.GetBytes(body, "access_token").Str
+	newRefreshToken = gjson.GetBytes(body, "refresh_token").Str
+	expiresInMs = gjson.GetBytes(body, "expires_in_ms").Int()
+	return newAccessToken, newRefreshToken, expiresInMs
+}
+
+// RegisterUser will register the user with given parameters and
 // return user ID & access token, and fail the test on network error
 func (c *CSAPI) RegisterUser(t *testing.T, localpart, password string) (userID, accessToken, deviceID string) {
 	t.Helper()

tests/csapi/txnid_test.go

@@ -0,0 +1,226 @@
+package csapi_tests
+
+import (
+	"testing"
+
+	"github.com/matrix-org/complement/internal/b"
+	"github.com/matrix-org/complement/internal/client"
+	"github.com/matrix-org/complement/runtime"
+	"github.com/tidwall/gjson"
+)
+
+func mustHaveTransactionID(t *testing.T, roomID, eventID string) client.SyncCheckOpt {
+	return client.SyncTimelineHas(roomID, func(r gjson.Result) bool {
+		if r.Get("event_id").Str == eventID {
+			if !r.Get("unsigned.transaction_id").Exists() {
+				t.Fatalf("Event %s in room %s should have a 'transaction_id', but it did not", eventID, roomID)
+			}
+
+			return true
+		}
+
+		return false
+	})
+}
+
+func mustNotHaveTransactionID(t *testing.T, roomID, eventID string) client.SyncCheckOpt {
+	return client.SyncTimelineHas(roomID, func(r gjson.Result) bool {
+		if r.Get("event_id").Str == eventID {
+			res := r.Get("unsigned.transaction_id")
+			if res.Exists() {
+				t.Fatalf("Event %s in room %s should NOT have a 'transaction_id', but it did (%s)", eventID, roomID, res.Str)
+			}
+
+			return true
+		}
+
+		return false
+	})
+}
+
+// TestTxnScopeOnLocalEcho tests that transaction IDs are scoped to the access token, not the device
+// on the sync response
+func TestTxnScopeOnLocalEcho(t *testing.T) {
+	// Conduit scope transaction IDs to the device ID, not the access token.
+	runtime.SkipIf(t, runtime.Conduit)
+
+	deployment := Deploy(t, b.BlueprintCleanHS)
+	defer deployment.Destroy(t)
+
+	deployment.RegisterUser(t, "hs1", "alice", "password", false)
+
+	// Create a first client, which allocates a device ID.
+	c1 := deployment.Client(t, "hs1", "")
+	c1.UserID, c1.AccessToken, c1.DeviceID = c1.LoginUser(t, "alice", "password")
+
+	// Create a room where we can send events.
+	roomID := c1.CreateRoom(t, map[string]interface{}{})
+
+	// Let's send an event, and wait for it to appear in the timeline.
+	eventID := c1.SendEventUnsynced(t, roomID, b.Event{
+		Type: "m.room.message",
+		Content: map[string]interface{}{
+			"msgtype": "m.text",
+			"body":    "first",
+		},
+	})
+
+	// When syncing, we should find the event and it should have a transaction ID on the first client.
+	c1.MustSyncUntil(t, client.SyncReq{}, mustHaveTransactionID(t, roomID, eventID))
+
+	// Create a second client, inheriting the first device ID.
+	c2 := deployment.Client(t, "hs1", "")
+	c2.UserID, c2.AccessToken = c2.LoginUserWithDeviceID(t, "alice", "password", c1.DeviceID)
+	c2.DeviceID = c1.DeviceID
+
+	// When syncing, we should find the event and it should *not* have a transaction ID on the second client.
+	c2.MustSyncUntil(t, client.SyncReq{}, mustNotHaveTransactionID(t, roomID, eventID))
+}
+
+// TestTxnIdempotencyNotScopedToDevicet tests that transaction IDs are not scoped to a device ID which is an
+// anti-pattern
+func TestTxnIdempotencyNotScopedToDevice(t *testing.T) {
+	// Conduit scope transaction IDs to the device ID, not the access token.
+	runtime.SkipIf(t, runtime.Conduit)
+
+	deployment := Deploy(t, b.BlueprintCleanHS)
+	defer deployment.Destroy(t)
+
+	deployment.RegisterUser(t, "hs1", "alice", "password", false)
+
+	// Create a first client, which allocates a device ID.
+	c1 := deployment.Client(t, "hs1", "")
+	c1.UserID, c1.AccessToken, c1.DeviceID = c1.LoginUser(t, "alice", "password")
+
+	// Create a room where we can send events.
+	roomID := c1.CreateRoom(t, map[string]interface{}{})
+
+	txnId := 1
+	event := b.Event{
+		Type: "m.room.message",
+		Content: map[string]interface{}{
+			"msgtype": "m.text",
+			"body":    "foo",
+		},
+	}
+	// send an event with set txnId
+	eventID1 := c1.SendEventUnsyncedWithTxnID(t, roomID, event, txnId)
+
+	// Create a second client, inheriting the first device ID.
+	c2 := deployment.Client(t, "hs1", "")
+	c2.UserID, c2.AccessToken = c2.LoginUserWithDeviceID(t, "alice", "password", c1.DeviceID)
+	c2.DeviceID = c1.DeviceID
+
+	// send another event with the same txnId
+	eventID2 := c2.SendEventUnsyncedWithTxnID(t, roomID, event, txnId)
+
+	// the two events should have different event IDs as they came from different clients
+	if (eventID1 == eventID2) {
+		t.Fatalf("Expected event IDs to be different from two clients sharing the same device ID")
+	}
+}
+
+// TestTxnIdempotency tests that PUT requests idempotency follows required semantics
+func TestTxnIdempotency(t *testing.T) {
+	deployment := Deploy(t, b.BlueprintCleanHS)
+	defer deployment.Destroy(t)
+
+	deployment.RegisterUser(t, "hs1", "alice", "password", false)
+
+	// Create a first client, which allocates a device ID.
+	c1 := deployment.Client(t, "hs1", "")
+	c1.UserID, c1.AccessToken, c1.DeviceID = c1.LoginUser(t, "alice", "password")
+
+	// Create a room where we can send events.
+	roomID1 := c1.CreateRoom(t, map[string]interface{}{})
+	roomID2 := c1.CreateRoom(t, map[string]interface{}{})
+
+	// choose a transaction ID
+	txnId := 1
+	event1 := b.Event{
+		Type: "m.room.message",
+		Content: map[string]interface{}{
+			"msgtype": "m.text",
+			"body":    "first",
+		},
+	}
+	event2 := b.Event{
+		Type: "m.room.message",
+		Content: map[string]interface{}{
+			"msgtype": "m.text",
+			"body":    "second",
+		},
+	}
+
+	// we send the event and get an event ID back
+	eventID1 := c1.SendEventUnsyncedWithTxnID(t, roomID1, event1, txnId)
+
+	// we send the identical event again and should get back the same event ID
+	eventID2 := c1.SendEventUnsyncedWithTxnID(t, roomID1, event1, txnId)
+
+	if eventID1 != eventID2 {
+		t.Fatalf("Expected event IDs to be the same, but they were not")
+	}
+
+	// even if we change the content we should still get back the same event ID as transaction ID is the same
+	eventID3 := c1.SendEventUnsyncedWithTxnID(t, roomID1, event2, txnId)
+
+	if eventID1 != eventID3 {
+		t.Fatalf("Expected event IDs to be the same even with different content, but they were not")
+	}
+
+	// if we change the room ID we should be able to use the same transaction ID
+	eventID4 := c1.SendEventUnsyncedWithTxnID(t, roomID2, event1, txnId)
+
+	if eventID4 == eventID3 {
+		t.Fatalf("Expected event IDs to be the different, but they were not")
+	}
+}
+
+// TestTxnAfterRefresh tests that when a client refreshes its access token,
+// it still gets back a transaction ID in the sync response.
+func TestTxnAfterRefresh(t *testing.T) {
+	// Dendrite and Conduit don't support refresh tokens yet.
+	// Synapse has a broken implementation of refresh tokens: https://github.com/matrix-org/synapse/issues/15141
+	runtime.SkipIf(t, runtime.Dendrite, runtime.Conduit, runtime.Synapse)
+
+	deployment := Deploy(t, b.BlueprintCleanHS)
+	defer deployment.Destroy(t)
+
+	deployment.RegisterUser(t, "hs1", "alice", "password", false)
+
+	c := deployment.Client(t, "hs1", "")
+
+	var refreshToken string
+	c.UserID, c.AccessToken, refreshToken, c.DeviceID, _ = c.LoginUserWithRefreshToken(t, "alice", "password")
+
+	// Create a room where we can send events.
+	roomID := c.CreateRoom(t, map[string]interface{}{})
+
+	// Let's send an event, and wait for it to appear in the sync.
+	eventID := c.SendEventUnsynced(t, roomID, b.Event{
+		Type: "m.room.message",
+		Content: map[string]interface{}{
+			"msgtype": "m.text",
+			"body":    "first",
+		},
+	})
+
+	// When syncing, we should find the event and it should have a transaction ID.
+	token := c.MustSyncUntil(t, client.SyncReq{}, mustHaveTransactionID(t, roomID, eventID))
+
+	// Now do the same, but refresh the token before syncing.
+	eventID = c.SendEventUnsynced(t, roomID, b.Event{
+		Type: "m.room.message",
+		Content: map[string]interface{}{
+			"msgtype": "m.text",
+			"body":    "second",
+		},
+	})
+
+	// Use the refresh token to get a new access token.
+	c.AccessToken, refreshToken, _ = c.ConsumeRefreshToken(t, refreshToken)
+
+	// When syncing, we should find the event and it should also have a transaction ID.
+	c.MustSyncUntil(t, client.SyncReq{Since: token}, mustHaveTransactionID(t, roomID, eventID))
+}