fix(multiplayer): verify complete durable assignment rosters

2026-09-10 16:04:04 +00:00 · 2026-09-03 13:22:02 +01:00
parent bef71e1dcf
commit cee0163eac
4 changed files with 171 additions and 12 deletions
@@ -76,8 +76,20 @@ FROM assignments
 WHERE match_id = $1 AND server_id = $2 AND expires_at > $3
 ORDER BY slot, player_id`

+const AssignmentExpectedRosterSQL = `SELECT mp.player_id, i.steam_id, mp.slot, mp.team
+FROM match_participants mp
+JOIN identities i ON i.player_id = mp.player_id
+JOIN matches m ON m.match_id = mp.match_id
+JOIN allocations a ON a.allocation_id = m.allocation_id AND a.match_id = m.match_id AND a.server_id = m.server_id
+WHERE mp.match_id = $1 AND m.allocation_id = $2 AND m.server_id = $3
+  AND m.region = $4 AND m.protocol_version = $5
+  AND a.region = $4 AND a.build = $6 AND a.protocol_version = $5 AND a.transport = $7
+  AND a.state = 'ALLOCATED' AND mp.participation_active
+ORDER BY mp.player_id
+FOR UPDATE OF mp`
+
 func validateDurableAssignment(assignment DurableAssignment) error {
-	if assignment.MatchID == "" || assignment.PlayerID == "" || assignment.AllocationID == "" || assignment.ServerID == "" || assignment.Slot < 0 || assignment.Slot > 5 || (assignment.Region != "EU" && assignment.Region != "NA") || assignment.ClientBuild == "" || assignment.ProtocolVersion < 1 || (assignment.Transport != "enet" && assignment.Transport != "steam_sdr") || assignment.Endpoint == "" || assignment.JoinAuthorisation == "" || len(assignment.ManifestDigest) == 0 || assignment.ExpiresAt.IsZero() || assignment.Revision < 0 {
+	if assignment.MatchID == "" || assignment.PlayerID == "" || assignment.AllocationID == "" || assignment.ServerID == "" || assignment.Slot < 0 || assignment.Slot > 5 || (assignment.Region != "EU" && assignment.Region != "NA") || assignment.ClientBuild == "" || assignment.ProtocolVersion < 1 || (assignment.Transport != "enet" && assignment.Transport != "steam_sdr") || assignment.Endpoint == "" || assignment.JoinAuthorisation == "" || len(assignment.ManifestDigest) == 0 || assignment.ExpiresAt.IsZero() || assignment.Revision == 0 {
 		return fmt.Errorf("invalid durable assignment")
 	}
 	return nil
@@ -112,21 +124,42 @@ func SaveAssignments(ctx context.Context, db *sql.DB, assignments []DurableAssig
 	if db == nil || len(assignments) == 0 {
 		return fmt.Errorf("invalid assignment batch")
 	}
-	tx, err := db.BeginTx(ctx, nil)
-	if err != nil {
+	if err := validateAssignmentBatch(assignments); err != nil {
 		return err
 	}
-	defer tx.Rollback()
+	return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
+		return saveAssignmentsTx(ctx, tx, assignments)
+	})
+}
+
+func validateAssignmentBatch(assignments []DurableAssignment) error {
+	if len(assignments) == 0 {
+		return fmt.Errorf("invalid assignment batch")
+	}
+	first := assignments[0]
 	seen := make(map[string]struct{}, len(assignments))
+	seenSlots := make(map[int]struct{}, len(assignments))
 	for _, assignment := range assignments {
 		if err := validateDurableAssignment(assignment); err != nil {
 			return err
 		}
-		key := assignment.MatchID + "\x00" + assignment.PlayerID
-		if _, ok := seen[key]; ok {
+		if assignment.MatchID != first.MatchID || assignment.AllocationID != first.AllocationID || assignment.ServerID != first.ServerID || assignment.Region != first.Region || assignment.ClientBuild != first.ClientBuild || assignment.ProtocolVersion != first.ProtocolVersion || assignment.Transport != first.Transport || assignment.Endpoint != first.Endpoint || string(assignment.ManifestDigest) != string(first.ManifestDigest) || assignment.Revision != first.Revision {
+			return fmt.Errorf("mixed assignment batch")
+		}
+		if _, ok := seen[assignment.PlayerID]; ok {
 			return fmt.Errorf("duplicate assignment in batch")
 		}
-		seen[key] = struct{}{}
+		if _, ok := seenSlots[assignment.Slot]; ok {
+			return fmt.Errorf("duplicate assignment slot in batch")
+		}
+		seen[assignment.PlayerID] = struct{}{}
+		seenSlots[assignment.Slot] = struct{}{}
+	}
+	return nil
+}
+
+func saveAssignmentsTx(ctx context.Context, tx *sql.Tx, assignments []DurableAssignment) error {
+	for _, assignment := range assignments {
 		result, err := tx.ExecContext(ctx, AssignmentUpsertSQL, assignment.MatchID, assignment.PlayerID, assignment.AllocationID, assignment.ServerID, assignment.Slot, assignment.Region, assignment.ClientBuild, assignment.ProtocolVersion, assignment.Transport, assignment.Endpoint, assignment.JoinAuthorisation, assignment.ManifestDigest, assignment.ExpiresAt, assignment.Revision)
 		if err != nil {
 			return err
@@ -139,7 +172,7 @@ func SaveAssignments(ctx context.Context, db *sql.DB, assignments []DurableAssig
 			return fmt.Errorf("assignment persistence conflict")
 		}
 	}
-	return tx.Commit()
+	return nil
 }

 // SaveVerifiedAssignmentRoster converts the backend-verified signed roster to
@@ -179,7 +212,59 @@ func SaveVerifiedAssignmentRoster(ctx context.Context, db *sql.DB, assignment do
 			ManifestDigest: digest[:], ExpiresAt: auth.ExpiresAt, Revision: 1,
 		})
 	}
-	return SaveAssignments(ctx, db, rows)
+	if db == nil {
+		return fmt.Errorf("invalid assignment database")
+	}
+	if err := validateAssignmentBatch(rows); err != nil {
+		return err
+	}
+	return RunSerializable(ctx, db, DefaultSerializableAttempts, func(ctx context.Context, tx *sql.Tx) error {
+		if err := validateExpectedAssignmentRoster(ctx, tx, assignment, roster); err != nil {
+			return err
+		}
+		return saveAssignmentsTx(ctx, tx, rows)
+	})
+}
+
+func validateExpectedAssignmentRoster(ctx context.Context, tx *sql.Tx, assignment domain.Assignment, roster []domain.SignedJoinAuthorisation) error {
+	rows, err := tx.QueryContext(ctx, AssignmentExpectedRosterSQL,
+		assignment.Allocation.MatchID, assignment.Allocation.AllocationID, assignment.Allocation.ServerID,
+		assignment.Allocation.Region, assignment.Allocation.Protocol, assignment.Allocation.Build, assignment.Allocation.Transport)
+	if err != nil {
+		return err
+	}
+	defer rows.Close()
+	type expectedPlayer struct {
+		steamID string
+		slot    int
+		team    int
+	}
+	expected := make(map[string]expectedPlayer, len(roster))
+	for rows.Next() {
+		var playerID string
+		var player expectedPlayer
+		if err := rows.Scan(&playerID, &player.steamID, &player.slot, &player.team); err != nil {
+			return err
+		}
+		expected[playerID] = player
+	}
+	if err := rows.Err(); err != nil {
+		return err
+	}
+	if err := rows.Close(); err != nil {
+		return err
+	}
+	if len(expected) == 0 || len(expected) != len(roster) {
+		return fmt.Errorf("signed assignment roster is incomplete")
+	}
+	for _, signed := range roster {
+		auth := signed.Authorisation
+		player, ok := expected[auth.PlayerID]
+		if !ok || player.steamID != auth.SteamID || player.slot != auth.Slot || player.team != auth.Team {
+			return fmt.Errorf("signed assignment roster does not match durable participants")
+		}
+	}
+	return nil
 }

 func validateSignedRosterEntry(assignment domain.Assignment, signed domain.SignedJoinAuthorisation, verify func([]byte, []byte) bool) error {
@@ -9,8 +9,9 @@ import (

 func TestAssignmentSQLBindsPlayerAndPreservesIdenticalReplay(t *testing.T) {
 	for query, fragments := range map[string][]string{
-		AssignmentUpsertSQL: {"ON CONFLICT (match_id, player_id)", "WHERE assignments.allocation_id = EXCLUDED.allocation_id", "join_authorisation", "manifest_digest"},
-		AssignmentSelectSQL: {"a.match_id = $1", "a.player_id = $2", "a.expires_at > $3", "JOIN matches", "ASSIGNMENT_READY", "m.server_id = a.server_id"},
+		AssignmentUpsertSQL:         {"ON CONFLICT (match_id, player_id)", "WHERE assignments.allocation_id = EXCLUDED.allocation_id", "join_authorisation", "manifest_digest"},
+		AssignmentSelectSQL:         {"a.match_id = $1", "a.player_id = $2", "a.expires_at > $3", "JOIN matches", "ASSIGNMENT_READY", "m.server_id = a.server_id"},
+		AssignmentExpectedRosterSQL: {"match_participants", "identities", "allocations", "m.allocation_id = $2", "m.server_id = $3", "a.state = 'ALLOCATED'", "participation_active", "FOR UPDATE OF mp"},
 	} {
 		for _, fragment := range fragments {
 			if !contains(query, fragment) {
@@ -20,6 +21,25 @@ func TestAssignmentSQLBindsPlayerAndPreservesIdenticalReplay(t *testing.T) {
 	}
 }

+func TestAssignmentBatchRejectsMixedAuthorityAndDuplicateSlots(t *testing.T) {
+	base := DurableAssignment{MatchID: "match-1", PlayerID: "player-1", AllocationID: "allocation-1", ServerID: "server-1", Slot: 0, Region: "EU", ClientBuild: "build-1", ProtocolVersion: 1, Transport: "enet", Endpoint: "127.0.0.1:1", JoinAuthorisation: "join-1", ManifestDigest: []byte("digest"), ExpiresAt: time.Unix(1001, 0), Revision: 1}
+	other := base
+	other.PlayerID = "player-2"
+	other.JoinAuthorisation = "join-2"
+	if err := validateAssignmentBatch([]DurableAssignment{base, other}); err == nil {
+		t.Fatal("duplicate slot accepted")
+	}
+	other.Slot = 3
+	other.ServerID = "server-2"
+	if err := validateAssignmentBatch([]DurableAssignment{base, other}); err == nil {
+		t.Fatal("mixed server batch accepted")
+	}
+	other.ServerID = base.ServerID
+	if err := validateAssignmentBatch([]DurableAssignment{base, other}); err != nil {
+		t.Fatalf("valid assignment batch rejected: %v", err)
+	}
+}
+
 func TestAssignmentStoreRejectsInvalidRecoveryAndManifestInputs(t *testing.T) {
 	if _, err := GetAssignment(nil, nil, "player-1", "match-1", time.Unix(1000, 0)); err == nil {
 		t.Fatal("nil database accepted")
@@ -515,6 +515,60 @@ func TestPostgreSQLAssignmentPersistenceIsPlayerScopedAndExpiryBound(t *testing.
 	}
 }

+func TestPostgreSQLVerifiedAssignmentRosterMustMatchDurableParticipants(t *testing.T) {
+	db := openIntegrationPostgres(t)
+	applyIntegrationMigrations(t, db)
+	now := time.Now().UTC().Truncate(time.Microsecond)
+	ctx := context.Background()
+	for index, player := range []string{"roster-player-a", "roster-player-b"} {
+		if _, err := db.ExecContext(ctx, `INSERT INTO identities (player_id, steam_id) VALUES ($1, $2)`, player, "steam-"+player); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := db.ExecContext(ctx, `INSERT INTO queue_tickets (ticket_id, player_id, playlist, state, client_build, protocol_version, enqueued_at, expires_at) VALUES ($1, $2, 'casual', 'ALLOCATING', 'build-1', 1, $3, $4)`, fmt.Sprintf("roster-ticket-%d", index), player, now, now.Add(time.Minute)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := db.ExecContext(ctx, `INSERT INTO game_servers (server_id, region, build, protocol_version, transport, state, updated_at) VALUES ('roster-server', 'EU', 'build-1', 1, 'enet', 'ALLOCATED', $1)`, now); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.ExecContext(ctx, `INSERT INTO allocations (allocation_id, match_id, server_id, region, build, protocol_version, transport, request_digest, state, allocated_at) VALUES ('roster-allocation', 'roster-match', 'roster-server', 'EU', 'build-1', 1, 'enet', 'digest', 'ALLOCATED', $1)`, now); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := db.ExecContext(ctx, `INSERT INTO matches (match_id, playlist, state, region, protocol_version, server_id, allocation_id, allocation_claimed_at) VALUES ('roster-match', 'casual', 'ALLOCATING', 'EU', 1, 'roster-server', 'roster-allocation', $1)`, now); err != nil {
+		t.Fatal(err)
+	}
+	for index, player := range []string{"roster-player-a", "roster-player-b"} {
+		if _, err := db.ExecContext(ctx, `INSERT INTO match_participants (match_id, player_id, ticket_id, slot, team) VALUES ('roster-match', $1, $2, $3, $4)`, player, fmt.Sprintf("roster-ticket-%d", index), index*3, index); err != nil {
+			t.Fatal(err)
+		}
+	}
+	allocation := domain.Allocation{AllocationID: "roster-allocation", MatchID: "roster-match", ServerID: "roster-server", Region: "EU", Build: "build-1", Protocol: 1, Transport: "enet", State: domain.ServerAllocated, AllocatedAt: now}
+	assignment := domain.Assignment{Allocation: allocation, Manifest: domain.AllocationManifest{AllocationID: allocation.AllocationID, MatchID: allocation.MatchID, ServerID: allocation.ServerID, Region: allocation.Region, Build: allocation.Build, Protocol: allocation.Protocol, Transport: allocation.Transport, RosterDigest: "roster-digest"}, Endpoint: "127.0.0.1:7777"}
+	roster := []domain.SignedJoinAuthorisation{
+		{Authorisation: domain.JoinAuthorisation{MatchID: "roster-match", ServerID: "roster-server", PlayerID: "roster-player-a", SteamID: "steam-roster-player-a", Slot: 0, Team: 0, Protocol: "1", Generation: 1, ExpiresAt: now.Add(time.Minute)}, Signature: []byte("sig-a")},
+		{Authorisation: domain.JoinAuthorisation{MatchID: "roster-match", ServerID: "roster-server", PlayerID: "roster-player-b", SteamID: "steam-roster-player-b", Slot: 3, Team: 1, Protocol: "1", Generation: 1, ExpiresAt: now.Add(time.Minute)}, Signature: []byte("sig-b")},
+	}
+	verify := func([]byte, []byte) bool { return true }
+	if err := SaveVerifiedAssignmentRoster(ctx, db, assignment, roster[:1], verify); err == nil {
+		t.Fatal("partial signed roster was accepted")
+	}
+	var count int
+	if err := db.QueryRow(`SELECT count(*) FROM assignments WHERE match_id = 'roster-match'`).Scan(&count); err != nil || count != 0 {
+		t.Fatalf("partial roster persisted rows=%d err=%v", count, err)
+	}
+	if err := SaveVerifiedAssignmentRoster(ctx, db, assignment, roster, verify); err != nil {
+		t.Fatalf("complete durable roster rejected: %v", err)
+	}
+	if err := db.QueryRow(`SELECT count(*) FROM assignments WHERE match_id = 'roster-match'`).Scan(&count); err != nil || count != 2 {
+		t.Fatalf("complete roster rows=%d err=%v", count, err)
+	}
+	forged := append([]domain.SignedJoinAuthorisation(nil), roster...)
+	forged[1].Authorisation.SteamID = "steam-other"
+	if err := SaveVerifiedAssignmentRoster(ctx, db, assignment, forged, verify); err == nil {
+		t.Fatal("signed roster with wrong durable Steam identity was accepted")
+	}
+}
+
 func TestPostgreSQLConnectionReceiptsStartCompleteRelaxedCasualRoster(t *testing.T) {
 	db := openIntegrationPostgres(t)
 	applyIntegrationMigrations(t, db)