From f4e4d2383129d3f3840933d750d08ebd942e845e Mon Sep 17 00:00:00 2001 From: DiheChen Date: Mon, 9 Nov 2026 21:24:32 +0800 Subject: [PATCH] feat: bind room tickets to member sessions and disconnect revoked sessions --- db/migrations/00003_member_accounts.sql | 9 + db/queries/members.sql | 5 +- db/queries/tickets.sql | 6 +- internal/adapter/in/api/api.go | 4 +- internal/adapter/in/api/api_test.go | 2 +- internal/adapter/in/api/events.go | 4 +- internal/adapter/in/api/events_test.go | 2 +- internal/adapter/in/api/handlers.go | 2 +- internal/adapter/in/api/room_test.go | 4 +- internal/adapter/in/api/ws_test.go | 2 +- internal/adapter/out/postgres/apikeys_test.go | 5 +- .../postgres/internal/query/members.sql.go | 24 +- .../postgres/internal/query/tickets.sql.go | 32 ++- internal/adapter/out/postgres/members.go | 27 ++- internal/adapter/out/postgres/members_test.go | 22 +- internal/adapter/out/postgres/migrate_test.go | 46 ++++ internal/adapter/out/postgres/tickets.go | 22 +- internal/adapter/out/postgres/tickets_test.go | 87 +++++++- internal/adapter/out/redis/bus.go | 13 ++ internal/adapter/out/redis/bus_test.go | 19 ++ internal/adapter/out/redis/codec.go | 11 +- internal/app/admin/admin_test.go | 7 +- internal/app/identity/account.go | 39 +++- internal/app/identity/account_test.go | 207 ++++++++++++++++-- internal/app/identity/identity.go | 20 +- internal/app/identity/identity_test.go | 84 ++++++- internal/app/lobby/lobby_test.go | 2 +- internal/app/port/bus.go | 20 +- internal/app/port/member.go | 7 +- internal/app/port/porttest/redis.go | 5 + internal/app/port/porttest/store.go | 55 +++-- internal/app/port/ticket.go | 6 +- internal/app/realtime/cluster_test.go | 2 +- internal/app/realtime/gateway.go | 47 +++- internal/app/realtime/gateway_test.go | 14 +- internal/app/realtime/observer.go | 20 +- internal/app/realtime/observer_test.go | 65 +++++- internal/app/realtime/session.go | 6 +- internal/bootstrap/public.go | 2 +- internal/bootstrap/testdata/http/events.txtar | 2 +- .../testdata/ws/account-sessions.txtar | 142 ++++++++++++ protocol/idl.json | 2 +- protocol/openapi.gen.json | 2 +- protocol/types.gen.go | 2 +- web/src/core/api/client.test.ts | 24 ++ web/src/core/api/client.ts | 8 +- web/src/core/realtime/protocol.gen.ts | 2 +- 47 files changed, 975 insertions(+), 165 deletions(-) create mode 100644 internal/bootstrap/testdata/ws/account-sessions.txtar diff --git a/db/migrations/00003_member_accounts.sql b/db/migrations/00003_member_accounts.sql index d7447b7..b217e79 100644 --- a/db/migrations/00003_member_accounts.sql +++ b/db/migrations/00003_member_accounts.sql @@ -28,3 +28,12 @@ ALTER TABLE members ALTER TABLE site ALTER COLUMN brand_name DROP NOT NULL, ADD COLUMN registration_open boolean NOT NULL DEFAULT true; + +-- 票据属于签发它的会话，会话删除时票据随之失效。bot 的 API Key 票据没有会话。 +ALTER TABLE room_tickets ADD COLUMN session_token_hash bytea REFERENCES member_sessions ON DELETE CASCADE; + +CREATE INDEX ON room_tickets (session_token_hash); + +UPDATE room_tickets t SET session_token_hash = s.token_hash +FROM member_sessions s +WHERE s.member_id = t.member_id; diff --git a/db/queries/members.sql b/db/queries/members.sql index 4e56d2f..4ee166f 100644 --- a/db/queries/members.sql +++ b/db/queries/members.sql @@ -45,6 +45,7 @@ SELECT coalesce(password_hash, '')::text AS password_hash FROM members WHERE id -- name: DeleteSession :execrows DELETE FROM member_sessions WHERE token_hash = @token_hash; --- name: DeleteMemberSessions :exec +-- name: DeleteMemberSessions :many -- keep 为 NULL 时删除这个成员的全部会话。 -DELETE FROM member_sessions WHERE member_id = @member_id AND token_hash IS DISTINCT FROM sqlc.narg(keep)::bytea; +DELETE FROM member_sessions WHERE member_id = @member_id AND token_hash IS DISTINCT FROM sqlc.narg(keep)::bytea +RETURNING token_hash; diff --git a/db/queries/tickets.sql b/db/queries/tickets.sql index bbbd356..d568cd3 100644 --- a/db/queries/tickets.sql +++ b/db/queries/tickets.sql @@ -3,10 +3,12 @@ SELECT coalesce(password_hash, '')::text AS password_hash FROM rooms WHERE id = @id FOR SHARE; -- name: CreateTicket :exec -INSERT INTO room_tickets (token_hash, room_id, member_id) VALUES (@token_hash, @room_id, @member_id); +-- bot 的 API Key 没有会话，session_token_hash 为 NULL。 +INSERT INTO room_tickets (token_hash, room_id, member_id, session_token_hash) +VALUES (@token_hash, @room_id, @member_id, sqlc.narg(session_token_hash)); -- name: TicketHolder :one -SELECT t.room_id, m.id, m.name, m.avatar_hash, r.bot_member_id IS NOT DISTINCT FROM m.id AS bot +SELECT t.room_id, m.id, m.name, m.avatar_hash, r.bot_member_id IS NOT DISTINCT FROM m.id AS bot, t.session_token_hash FROM room_tickets t JOIN members m ON m.id = t.member_id JOIN rooms r ON r.id = t.room_id diff --git a/internal/adapter/in/api/api.go b/internal/adapter/in/api/api.go index f9a43ee..4cf3c04 100644 --- a/internal/adapter/in/api/api.go +++ b/internal/adapter/in/api/api.go @@ -68,6 +68,8 @@ type call struct { member member.Member roomID string ticket string + // 票据签发给的成员会话的 token 哈希，API Key 没有会话。 + memberSession []byte // 票据是常驻房间的 API Key。 bot bool // 发出请求的客户端地址，按来源限流时使用。 @@ -211,7 +213,7 @@ func (s *server) serve(ctx context.Context, w http.ResponseWriter, r *http.Reque if err != nil { return ctx, nil, err } - c.roomID, c.member, c.bot = holder.RoomID, holder.Member, holder.Bot + c.roomID, c.member, c.memberSession, c.bot = holder.RoomID, holder.Member, holder.SessionTokenHash, holder.Bot ctx = logctx.WithMember(logctx.WithRoom(ctx, holder.RoomID), holder.Member.ID) case protocol.AuthAdmin: if err := s.Admin.Authorize(bearer(r)); err != nil { diff --git a/internal/adapter/in/api/api_test.go b/internal/adapter/in/api/api_test.go index 595f820..b3f60c4 100644 --- a/internal/adapter/in/api/api_test.go +++ b/internal/adapter/in/api/api_test.go @@ -42,7 +42,7 @@ func services(store *porttest.Store, rooms port.RoomDirectory, adminToken string l := lobby.New(rooms, node) return Services{ Identity: identity.New(store, rooms, store, gateway), - Accounts: identity.NewAccounts(store, store, nil), + Accounts: identity.NewAccounts(store, store, nil, gateway, slog.New(slog.DiscardHandler)), Lobby: l, Admin: admin.New(adminToken, l, rooms, store, store, porttest.NewNetEaseQRLogin(), porttest.NewQQQRLogin(), porttest.NewBilibiliQRLogin(), porttest.NewYouTubeCookies(), gateway), Gateway: gateway, diff --git a/internal/adapter/in/api/events.go b/internal/adapter/in/api/events.go index 81b9114..b8cd29f 100644 --- a/internal/adapter/in/api/events.go +++ b/internal/adapter/in/api/events.go @@ -23,7 +23,7 @@ const ( var streamCloses = map[realtime.CloseReason]streamClose{ realtime.CloseRestart: {Code: protocol.CloseServiceRestart, Message: "服务端实例退出或订阅中断，请重新连接"}, - realtime.CloseRevoked: {Code: protocol.CloseUnauthorized, Message: "凭证已失效：房间设置或修改了密码，或者 API Key 已重置"}, + realtime.CloseRevoked: {Code: protocol.CloseUnauthorized, Message: "凭证已失效：房间设置或修改了密码，API Key 已重置，或者账号在这台设备上退出了登录、密码被修改或重置"}, realtime.CloseRoomGone: {Code: protocol.CloseNotFound, Message: "房间已关闭或被删除"}, realtime.CloseOverflow: {Code: protocol.CloseSendQueueFull, Message: "积压的事件过多，请重新连接"}, // 不是 1012：共用一把 Key 的两个 bot 看到它就该停下，而不是重连之后互相顶掉。 @@ -83,7 +83,7 @@ func (s *server) roomEvents(w http.ResponseWriter, r *http.Request) { func (s *server) observe(ctx context.Context, c call) (any, error) { // 内存中只留凭证的哈希。 credential := sha256.Sum256([]byte(c.ticket)) - o, err := s.Gateway.Observe(ctx, c.roomID, c.member, c.bot, string(credential[:])) + o, err := s.Gateway.Observe(ctx, c.roomID, c.member, c.memberSession, c.bot, string(credential[:])) if err != nil { return nil, err } diff --git a/internal/adapter/in/api/events_test.go b/internal/adapter/in/api/events_test.go index 9e0edb5..3cbb4a3 100644 --- a/internal/adapter/in/api/events_test.go +++ b/internal/adapter/in/api/events_test.go @@ -145,7 +145,7 @@ func TestRoomEvents(t *testing.T) { if _, err := f.admin.ResetRoomAPIKey(t.Context(), hall); err != nil { t.Fatal(err) } - checkEvent(t, r, "event: close", `data: {"code":4401,"message":"凭证已失效：房间设置或修改了密码，或者 API Key 已重置"}`) + checkEvent(t, r, "event: close", `data: {"code":4401,"message":"凭证已失效：房间设置或修改了密码，API Key 已重置，或者账号在这台设备上退出了登录、密码被修改或重置"}`) checkEnded(t, r) }) } diff --git a/internal/adapter/in/api/handlers.go b/internal/adapter/in/api/handlers.go index 520241a..2124593 100644 --- a/internal/adapter/in/api/handlers.go +++ b/internal/adapter/in/api/handlers.go @@ -98,7 +98,7 @@ func (s *server) createTicket(ctx context.Context, c call) (any, error) { if err != nil { return nil, err } - ticket, err := s.Identity.IssueTicket(ctx, c.member.ID, id, deref(req.Password)) + ticket, err := s.Identity.IssueTicket(ctx, c.member.ID, bearer(c.r), id, deref(req.Password)) if err != nil { return nil, err } diff --git a/internal/adapter/in/api/room_test.go b/internal/adapter/in/api/room_test.go index 8990c43..da049e3 100644 --- a/internal/adapter/in/api/room_test.go +++ b/internal/adapter/in/api/room_test.go @@ -232,7 +232,7 @@ func TestRoomRequestOfDisconnectedClientIsNotCounted(t *testing.T) { func TestRoomListingShowsOnlineCount(t *testing.T) { f := newRoomFixture(t) id, m, _ := f.enter(t) - session, err := f.gateway.Join(t.Context(), id, m) + session, err := f.gateway.Join(t.Context(), id, m, nil) if err != nil { t.Fatal(err) } @@ -252,7 +252,7 @@ func TestProfileChangeReachesRooms(t *testing.T) { session := decode[protocol.CreateMemberResponse](t, send(t, f.server, http.MethodPost, "/api/members", "{}")) created := decode[protocol.CreateRoomResponse](t, send(t, f.server, http.MethodPost, "/api/rooms", `{"name":"临时","description":""}`)) m := member.Member{ID: session.Member.ID, Name: session.Member.Name} - s, err := f.gateway.Join(t.Context(), created.ID, m) + s, err := f.gateway.Join(t.Context(), created.ID, m, nil) if err != nil { t.Fatal(err) } diff --git a/internal/adapter/in/api/ws_test.go b/internal/adapter/in/api/ws_test.go index 6eb93a2..5c49158 100644 --- a/internal/adapter/in/api/ws_test.go +++ b/internal/adapter/in/api/ws_test.go @@ -191,7 +191,7 @@ func (f *wsFixture) enter(t *testing.T, roomID string) (string, member.Member, s t.Fatal(err) } } - ticket, err := f.identity.IssueTicket(t.Context(), session.Member.ID, roomID, "") + ticket, err := f.identity.IssueTicket(t.Context(), session.Member.ID, session.MemberToken, roomID, "") if err != nil { t.Fatal(err) } diff --git a/internal/adapter/out/postgres/apikeys_test.go b/internal/adapter/out/postgres/apikeys_test.go index e593704..7eadcaf 100644 --- a/internal/adapter/out/postgres/apikeys_test.go +++ b/internal/adapter/out/postgres/apikeys_test.go @@ -2,6 +2,7 @@ package postgres import ( "errors" + "reflect" "testing" "time" @@ -58,7 +59,7 @@ func TestResetRoomAPIKey(t *testing.T) { if _, err := store.TicketHolder(t.Context(), tokenHash("k1")); !errors.Is(err, port.ErrTicketNotFound) { t.Errorf("重置之后旧 Key 的错误 = %v，期望 ErrTicketNotFound", err) } - if holder, err := store.TicketHolder(t.Context(), tokenHash("k2")); err != nil || holder != bot { + if holder, err := store.TicketHolder(t.Context(), tokenHash("k2")); err != nil || !reflect.DeepEqual(holder, bot) { t.Errorf("重置之后新 Key 的持有者 = %#v, %v，期望 bot 成员不变", holder, err) } if n := memberTickets(t, pool, bot.Member.ID); n != 1 { @@ -84,7 +85,7 @@ func TestSettingPasswordKeepsAPIKey(t *testing.T) { if err := store.UpdateRoomAccess(t.Context(), id, port.AccessChange{PasswordHash: ptr("h1")}); err != nil { t.Fatal(err) } - if holder, err := store.TicketHolder(t.Context(), tokenHash("k1")); err != nil || holder != bot { + if holder, err := store.TicketHolder(t.Context(), tokenHash("k1")); err != nil || !reflect.DeepEqual(holder, bot) { t.Errorf("设置密码之后 Key 的持有者 = %#v, %v，期望 Key 仍然有效", holder, err) } if n := ticketCount(t, pool, id); n != 1 { diff --git a/internal/adapter/out/postgres/internal/query/members.sql.go b/internal/adapter/out/postgres/internal/query/members.sql.go index 697bd53..5d2ddce 100644 --- a/internal/adapter/out/postgres/internal/query/members.sql.go +++ b/internal/adapter/out/postgres/internal/query/members.sql.go @@ -62,8 +62,9 @@ func (q *Queries) CreateSession(ctx context.Context, arg CreateSessionParams) er return err } -const deleteMemberSessions = `-- name: DeleteMemberSessions :exec +const deleteMemberSessions = `-- name: DeleteMemberSessions :many DELETE FROM member_sessions WHERE member_id = $1 AND token_hash IS DISTINCT FROM $2::bytea +RETURNING token_hash ` type DeleteMemberSessionsParams struct { @@ -72,9 +73,24 @@ type DeleteMemberSessionsParams struct { } // keep 为 NULL 时删除这个成员的全部会话。 -func (q *Queries) DeleteMemberSessions(ctx context.Context, arg DeleteMemberSessionsParams) error { - _, err := q.db.Exec(ctx, deleteMemberSessions, arg.MemberID, arg.Keep) - return err +func (q *Queries) DeleteMemberSessions(ctx context.Context, arg DeleteMemberSessionsParams) ([][]byte, error) { + rows, err := q.db.Query(ctx, deleteMemberSessions, arg.MemberID, arg.Keep) + if err != nil { + return nil, err + } + defer rows.Close() + var items [][]byte + for rows.Next() { + var token_hash []byte + if err := rows.Scan(&token_hash); err != nil { + return nil, err + } + items = append(items, token_hash) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil } const deleteSession = `-- name: DeleteSession :execrows diff --git a/internal/adapter/out/postgres/internal/query/tickets.sql.go b/internal/adapter/out/postgres/internal/query/tickets.sql.go index 6666c0d..5a51fa0 100644 --- a/internal/adapter/out/postgres/internal/query/tickets.sql.go +++ b/internal/adapter/out/postgres/internal/query/tickets.sql.go @@ -10,17 +10,25 @@ import ( ) const createTicket = `-- name: CreateTicket :exec -INSERT INTO room_tickets (token_hash, room_id, member_id) VALUES ($1, $2, $3) +INSERT INTO room_tickets (token_hash, room_id, member_id, session_token_hash) +VALUES ($1, $2, $3, $4) ` type CreateTicketParams struct { - TokenHash []byte - RoomID string - MemberID string + TokenHash []byte + RoomID string + MemberID string + SessionTokenHash []byte } +// bot 的 API Key 没有会话，session_token_hash 为 NULL。 func (q *Queries) CreateTicket(ctx context.Context, arg CreateTicketParams) error { - _, err := q.db.Exec(ctx, createTicket, arg.TokenHash, arg.RoomID, arg.MemberID) + _, err := q.db.Exec(ctx, createTicket, + arg.TokenHash, + arg.RoomID, + arg.MemberID, + arg.SessionTokenHash, + ) return err } @@ -37,7 +45,7 @@ func (q *Queries) LockRoomPassword(ctx context.Context, id string) (string, erro } const ticketHolder = `-- name: TicketHolder :one -SELECT t.room_id, m.id, m.name, m.avatar_hash, r.bot_member_id IS NOT DISTINCT FROM m.id AS bot +SELECT t.room_id, m.id, m.name, m.avatar_hash, r.bot_member_id IS NOT DISTINCT FROM m.id AS bot, t.session_token_hash FROM room_tickets t JOIN members m ON m.id = t.member_id JOIN rooms r ON r.id = t.room_id @@ -45,11 +53,12 @@ WHERE t.token_hash = $1 ` type TicketHolderRow struct { - RoomID string - ID string - Name string - AvatarHash *string - Bot bool + RoomID string + ID string + Name string + AvatarHash *string + Bot bool + SessionTokenHash []byte } func (q *Queries) TicketHolder(ctx context.Context, tokenHash []byte) (TicketHolderRow, error) { @@ -61,6 +70,7 @@ func (q *Queries) TicketHolder(ctx context.Context, tokenHash []byte) (TicketHol &i.Name, &i.AvatarHash, &i.Bot, + &i.SessionTokenHash, ) return i, err } diff --git a/internal/adapter/out/postgres/members.go b/internal/adapter/out/postgres/members.go index c7949e7..7bb384b 100644 --- a/internal/adapter/out/postgres/members.go +++ b/internal/adapter/out/postgres/members.go @@ -131,17 +131,21 @@ func (s *Store) DeleteSession(ctx context.Context, tokenHash []byte) error { return nil } -func (s *Store) SetPassword(ctx context.Context, memberID, passwordHash string) error { +func (s *Store) SetPassword(ctx context.Context, memberID, passwordHash string) ([][]byte, error) { + var deleted [][]byte err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error { - return setPassword(ctx, s.q.WithTx(tx), memberID, passwordHash, nil) + var err error + deleted, err = setPassword(ctx, s.q.WithTx(tx), memberID, passwordHash, nil) + return err }) if err != nil { - return fmt.Errorf("修改成员 %s 的密码：%w", memberID, err) + return nil, fmt.Errorf("修改成员 %s 的密码：%w", memberID, err) } - return nil + return deleted, nil } -func (s *Store) ChangePassword(ctx context.Context, memberID, currentPasswordHash, passwordHash string, keepTokenHash []byte) error { +func (s *Store) ChangePassword(ctx context.Context, memberID, currentPasswordHash, passwordHash string, keepTokenHash []byte) ([][]byte, error) { + var deleted [][]byte err := pgx.BeginFunc(ctx, s.pool, func(tx pgx.Tx) error { q := s.q.WithTx(tx) // 先锁住成员行，再用新的语句查会话：并发的修改或重置提交之后才开始查，才能看到它删掉的会话。 @@ -162,21 +166,22 @@ func (s *Store) ChangePassword(ctx context.Context, memberID, currentPasswordHas if current != currentPasswordHash { return port.ErrAccountPasswordChanged } - return setPassword(ctx, q, memberID, passwordHash, keepTokenHash) + deleted, err = setPassword(ctx, q, memberID, passwordHash, keepTokenHash) + return err }) if err != nil { - return fmt.Errorf("修改成员 %s 的密码：%w", memberID, err) + return nil, fmt.Errorf("修改成员 %s 的密码：%w", memberID, err) } - return nil + return deleted, nil } -func setPassword(ctx context.Context, q *query.Queries, memberID, passwordHash string, keepTokenHash []byte) error { +func setPassword(ctx context.Context, q *query.Queries, memberID, passwordHash string, keepTokenHash []byte) ([][]byte, error) { n, err := q.SetMemberPassword(ctx, query.SetMemberPasswordParams{ID: memberID, PasswordHash: &passwordHash}) if err != nil { - return err + return nil, err } if n == 0 { - return port.ErrMemberNotFound + return nil, port.ErrMemberNotFound } return q.DeleteMemberSessions(ctx, query.DeleteMemberSessionsParams{MemberID: memberID, Keep: keepTokenHash}) } diff --git a/internal/adapter/out/postgres/members_test.go b/internal/adapter/out/postgres/members_test.go index 8a15437..59b2b11 100644 --- a/internal/adapter/out/postgres/members_test.go +++ b/internal/adapter/out/postgres/members_test.go @@ -228,8 +228,9 @@ func TestMemberSessions(t *testing.T) { if err := store.Register(t.Context(), alice.ID, "alice", "old"); err != nil { t.Fatal(err) } - if err := store.ChangePassword(t.Context(), alice.ID, "old", "new", tokenHash("alice-1")); err != nil { - t.Fatal(err) + deleted, err := store.ChangePassword(t.Context(), alice.ID, "old", "new", tokenHash("alice-1")) + if want := [][]byte{tokenHash("alice-2"), tokenHash("alice-3")}; err != nil || !sameHashes(deleted, want) { + t.Errorf("ChangePassword 返回 %x, %v，期望被删除的 alice-2、alice-3", deleted, err) } if ownerOf(t, store, "alice-1") != alice.ID || ownerOf(t, store, "alice-2") != "" || ownerOf(t, store, "alice-3") != "" { t.Error("ChangePassword 应只留下发出请求的会话") @@ -245,10 +246,10 @@ func TestMemberSessions(t *testing.T) { func TestSetPassword(t *testing.T) { store, _ := newStore(t) guest := createMember(t, store, "guest", "游客-1a2b") - if err := store.SetPassword(t.Context(), guest.ID, "hash"); !errors.Is(err, port.ErrMemberNotFound) { + if _, err := store.SetPassword(t.Context(), guest.ID, "hash"); !errors.Is(err, port.ErrMemberNotFound) { t.Errorf("给游客设置密码 = %v，期望 ErrMemberNotFound", err) } - if err := store.ChangePassword(t.Context(), guest.ID, "", "hash", tokenHash("guest")); !errors.Is(err, port.ErrMemberNotFound) { + if _, err := store.ChangePassword(t.Context(), guest.ID, "", "hash", tokenHash("guest")); !errors.Is(err, port.ErrMemberNotFound) { t.Errorf("游客修改密码 = %v，期望 ErrMemberNotFound", err) } alice := createMember(t, store, "alice-1", "游客-3c4d") @@ -259,8 +260,8 @@ func TestSetPassword(t *testing.T) { if err := store.ReplaceSession(t.Context(), tokenHash("other"), alice.ID, tokenHash("alice-2"), "old"); err != nil { t.Fatal(err) } - if err := store.ChangePassword(t.Context(), alice.ID, "old", "new", tokenHash("alice-1")); err != nil { - t.Fatal(err) + if deleted, err := store.ChangePassword(t.Context(), alice.ID, "old", "new", tokenHash("alice-1")); err != nil || !sameHashes(deleted, [][]byte{tokenHash("alice-2")}) { + t.Errorf("ChangePassword 返回 %x, %v，期望被删除的 alice-2", deleted, err) } if account, err := store.AccountByUsername(t.Context(), "alice"); err != nil || account.PasswordHash != "new" { t.Errorf("修改之后的密码哈希 = %q, %v，期望 new", account.PasswordHash, err) @@ -268,8 +269,8 @@ func TestSetPassword(t *testing.T) { if ownerOf(t, store, "alice-1") != alice.ID || ownerOf(t, store, "alice-2") != "" { t.Error("修改密码应只留下指定的会话") } - if err := store.SetPassword(t.Context(), alice.ID, "newer"); err != nil { - t.Fatal(err) + if deleted, err := store.SetPassword(t.Context(), alice.ID, "newer"); err != nil || !sameHashes(deleted, [][]byte{tokenHash("alice-1")}) { + t.Errorf("SetPassword 返回 %x, %v，期望被删除的 alice-1", deleted, err) } if got := ownerOf(t, store, "alice-1"); got != "" { t.Errorf("SetPassword 应删除全部会话，alice-1 仍属于 %s", got) @@ -295,7 +296,7 @@ func TestChangePasswordChecksSessionAndPassword(t *testing.T) { {"会话属于别的成员", "h1", "bob", port.ErrSessionNotFound}, {"密码哈希已经变了", "h0", "alice-1", port.ErrAccountPasswordChanged}, } { - if err := store.ChangePassword(t.Context(), alice.ID, tc.current, "h2", tokenHash(tc.keep)); !errors.Is(err, tc.want) { + if _, err := store.ChangePassword(t.Context(), alice.ID, tc.current, "h2", tokenHash(tc.keep)); !errors.Is(err, tc.want) { t.Errorf("%s时 = %v，期望 %v", tc.name, err, tc.want) } } @@ -324,7 +325,8 @@ func changePasswordMeanwhile(t *testing.T, store *Store, pool *pgxpool.Pool, mem result := make(chan error, 1) go func() { - result <- store.ChangePassword(t.Context(), memberID, "h1", "h3", tokenHash("alice-1")) + _, err := store.ChangePassword(t.Context(), memberID, "h1", "h3", tokenHash("alice-1")) + result <- err }() waitForLockWait(t, pool, 1) if err := tx.Commit(t.Context()); err != nil { diff --git a/internal/adapter/out/postgres/migrate_test.go b/internal/adapter/out/postgres/migrate_test.go index 5f0c829..3b5842e 100644 --- a/internal/adapter/out/postgres/migrate_test.go +++ b/internal/adapter/out/postgres/migrate_test.go @@ -179,3 +179,49 @@ func TestMemberSessionsMigrationKeepsMemberTokens(t *testing.T) { t.Errorf("bot 成员原来的 token 在迁移之后 = %v，期望不再有效", err) } } + +func TestTicketSessionsMigrationBindsTicketsToSessions(t *testing.T) { + ctx := t.Context() + name := fmt.Sprintf("test_%d", databases.Add(1)) + if _, err := admin.Exec(ctx, "CREATE DATABASE "+name); err != nil { + t.Fatal(err) + } + cfg := server.Copy() + cfg.ConnConfig.Database = name + pool := connect(t, cfg) + db := stdlib.OpenDBFromPool(pool) + t.Cleanup(func() { _ = db.Close() }) + provider, err := goose.NewProvider(goose.DialectPostgres, db, migrations.FS) + if err != nil { + t.Fatal(err) + } + if _, err := provider.UpTo(ctx, 2); err != nil { + t.Fatal(err) + } + var guest, bot, roomID string + if err := pool.QueryRow(ctx, "INSERT INTO members (token_hash, name) VALUES ($1, '游客-1a2b') RETURNING id", tokenHash("guest")).Scan(&guest); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, "INSERT INTO members (token_hash, name) VALUES ($1, 'bot') RETURNING id", tokenHash("bot")).Scan(&bot); err != nil { + t.Fatal(err) + } + if err := pool.QueryRow(ctx, "INSERT INTO rooms (kind, name, description, bot_member_id) VALUES ('persistent', '大厅', '', $1) RETURNING id", bot).Scan(&roomID); err != nil { + t.Fatal(err) + } + for ticket, memberID := range map[string]string{"guest-ticket": guest, "api-key": bot} { + if _, err := pool.Exec(ctx, "INSERT INTO room_tickets (token_hash, room_id, member_id) VALUES ($1, $2, $3)", tokenHash(ticket), roomID, memberID); err != nil { + t.Fatal(err) + } + } + if _, err := provider.Up(ctx); err != nil { + t.Fatal(err) + } + + store := New(pool) + if holder, err := store.TicketHolder(ctx, tokenHash("guest-ticket")); err != nil || !bytes.Equal(holder.SessionTokenHash, tokenHash("guest")) { + t.Errorf("游客的票据 = %#v, %v，期望归到游客的会话", holder, err) + } + if holder, err := store.TicketHolder(ctx, tokenHash("api-key")); err != nil || !holder.Bot || holder.SessionTokenHash != nil { + t.Errorf("API Key 的票据 = %#v, %v，期望保留，没有会话", holder, err) + } +} diff --git a/internal/adapter/out/postgres/tickets.go b/internal/adapter/out/postgres/tickets.go index 53be3b0..386a046 100644 --- a/internal/adapter/out/postgres/tickets.go +++ b/internal/adapter/out/postgres/tickets.go @@ -6,6 +6,7 @@ import ( "fmt" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/dihemoe/alisten/internal/adapter/out/postgres/internal/query" "github.com/dihemoe/alisten/internal/app/port" @@ -27,10 +28,10 @@ func (s *Store) CreateTicket(ctx context.Context, t port.NewTicket) error { if hash != t.PasswordHash { return port.ErrPasswordChanged } - return q.CreateTicket(ctx, query.CreateTicketParams{TokenHash: t.TokenHash, RoomID: t.RoomID, MemberID: t.MemberID}) + return q.CreateTicket(ctx, query.CreateTicketParams{TokenHash: t.TokenHash, RoomID: t.RoomID, MemberID: t.MemberID, SessionTokenHash: t.SessionTokenHash}) }) if err != nil { - return fmt.Errorf("签发房间 %s 的票据：%w", t.RoomID, err) + return fmt.Errorf("签发房间 %s 的票据：%w", t.RoomID, sessionError(err)) } return nil } @@ -44,8 +45,19 @@ func (s *Store) TicketHolder(ctx context.Context, tokenHash []byte) (port.Ticket return port.TicketHolder{}, fmt.Errorf("查找票据：%w", err) } return port.TicketHolder{ - RoomID: row.RoomID, - Member: member.Member{ID: row.ID, Name: row.Name, AvatarHash: orEmpty(row.AvatarHash)}, - Bot: row.Bot, + RoomID: row.RoomID, + Member: member.Member{ID: row.ID, Name: row.Name, AvatarHash: orEmpty(row.AvatarHash)}, + Bot: row.Bot, + SessionTokenHash: row.SessionTokenHash, }, nil } + +const foreignKeyViolation = "23503" + +func sessionError(err error) error { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == foreignKeyViolation && pgErr.ConstraintName == "room_tickets_session_token_hash_fkey" { + return port.ErrSessionNotFound + } + return err +} diff --git a/internal/adapter/out/postgres/tickets_test.go b/internal/adapter/out/postgres/tickets_test.go index a6a50b2..5642e77 100644 --- a/internal/adapter/out/postgres/tickets_test.go +++ b/internal/adapter/out/postgres/tickets_test.go @@ -2,6 +2,8 @@ package postgres import ( "errors" + "reflect" + "slices" "testing" "time" @@ -16,11 +18,11 @@ func TestTicketHolder(t *testing.T) { alice := createMember(t, store, "alice", "甲") id := createRoom(t, store, port.NewRoom{Kind: room.Ephemeral, Name: "临时"}) - if err := store.CreateTicket(t.Context(), port.NewTicket{TokenHash: tokenHash("t1"), RoomID: id, MemberID: alice.ID}); err != nil { + if err := store.CreateTicket(t.Context(), port.NewTicket{TokenHash: tokenHash("t1"), RoomID: id, MemberID: alice.ID, SessionTokenHash: tokenHash("alice")}); err != nil { t.Fatal(err) } got, err := store.TicketHolder(t.Context(), tokenHash("t1")) - if want := (port.TicketHolder{RoomID: id, Member: alice}); err != nil || got != want { + if want := (port.TicketHolder{RoomID: id, Member: alice, SessionTokenHash: tokenHash("alice")}); err != nil || !reflect.DeepEqual(got, want) { t.Errorf("TicketHolder = %#v, %v，期望 %#v", got, err, want) } @@ -37,6 +39,87 @@ func TestTicketHolder(t *testing.T) { } } +func TestCreateTicketRejectsDeletedSession(t *testing.T) { + store, _ := newStore(t) + alice := createMember(t, store, "alice", "甲") + id := createRoom(t, store, port.NewRoom{Kind: room.Ephemeral, Name: "临时"}) + if err := store.DeleteSession(t.Context(), tokenHash("alice")); err != nil { + t.Fatal(err) + } + + err := store.CreateTicket(t.Context(), port.NewTicket{TokenHash: tokenHash("t1"), RoomID: id, MemberID: alice.ID, SessionTokenHash: tokenHash("alice")}) + if !errors.Is(err, port.ErrSessionNotFound) { + t.Errorf("会话已经删除时 CreateTicket = %v，期望 ErrSessionNotFound", err) + } +} + +func TestTicketsBelongToSessions(t *testing.T) { + store, _ := newStore(t) + alice := createMember(t, store, "alice-1", "甲") + id := createRoom(t, store, port.NewRoom{Kind: room.Ephemeral, Name: "临时"}) + for _, guest := range []string{"guest-2", "guest-3", "guest-4"} { + createMember(t, store, guest, "游客") + } + for _, s := range []struct{ guest, token string }{{"guest-2", "alice-2"}, {"guest-3", "alice-3"}} { + if err := store.ReplaceSession(t.Context(), tokenHash(s.guest), alice.ID, tokenHash(s.token), ""); err != nil { + t.Fatal(err) + } + } + for _, session := range []string{"alice-1", "alice-2", "alice-3", "guest-4"} { + memberID := alice.ID + if session == "guest-4" { + memberID = ownerOf(t, store, "guest-4") + } + if err := store.CreateTicket(t.Context(), port.NewTicket{TokenHash: tokenHash("ticket-" + session), RoomID: id, MemberID: memberID, SessionTokenHash: tokenHash(session)}); err != nil { + t.Fatal(err) + } + } + valid := func(session string) bool { + t.Helper() + _, err := store.TicketHolder(t.Context(), tokenHash("ticket-"+session)) + if err != nil && !errors.Is(err, port.ErrTicketNotFound) { + t.Fatal(err) + } + return err == nil + } + + if err := store.Register(t.Context(), alice.ID, "alice", "old"); err != nil { + t.Fatal(err) + } + deleted, err := store.ChangePassword(t.Context(), alice.ID, "old", "new", tokenHash("alice-1")) + if want := [][]byte{tokenHash("alice-2"), tokenHash("alice-3")}; err != nil || !sameHashes(deleted, want) { + t.Errorf("ChangePassword 返回 %x, %v，期望被删除的 alice-2、alice-3", deleted, err) + } + if !valid("alice-1") || valid("alice-2") || valid("alice-3") { + t.Error("删除会话之后，只有保留的会话的票据仍然有效") + } + if err := store.ReplaceSession(t.Context(), tokenHash("guest-4"), alice.ID, tokenHash("alice-4"), "new"); err != nil { + t.Fatal(err) + } + if valid("guest-4") { + t.Error("登录换掉游客的会话之后，游客的票据仍然有效") + } + if err := store.DeleteSession(t.Context(), tokenHash("alice-1")); err != nil { + t.Fatal(err) + } + if valid("alice-1") { + t.Error("退出登录之后，这个会话的票据仍然有效") + } +} + +// 不计顺序。 +func sameHashes(got, want [][]byte) bool { + key := func(hashes [][]byte) []string { + s := make([]string, len(hashes)) + for i, h := range hashes { + s[i] = string(h) + } + slices.Sort(s) + return s + } + return slices.Equal(key(got), key(want)) +} + func TestCreateTicketChecksPasswordHash(t *testing.T) { store, _ := newStore(t) alice := createMember(t, store, "alice", "甲") diff --git a/internal/adapter/out/redis/bus.go b/internal/adapter/out/redis/bus.go index 7b1d501..2dfeba1 100644 --- a/internal/adapter/out/redis/bus.go +++ b/internal/adapter/out/redis/bus.go @@ -198,6 +198,15 @@ func (b *Bus) PublishMember(ctx context.Context, m member.Member) error { return err } +func (b *Bus) PublishSessionsRevoked(ctx context.Context, sessionTokenHashes [][]byte) error { + data, err := json.Marshal(sessionTokenHashes) + if err != nil { + return err + } + _, err = b.publish(ctx, membersChannel, kindSessionsRevoked, data, nil) + return err +} + func (b *Bus) PublishPrepared(ctx context.Context, o port.PrepareOutcome) error { data, err := json.Marshal(o) if err != nil { @@ -362,6 +371,10 @@ func (b *Bus) delivery(channel string, e envelope) (port.Delivery, error) { var m member.Member err := json.Unmarshal(e.Data, &m) return port.MemberDelivery{Member: m}, err + case channel == membersChannel && e.Kind == kindSessionsRevoked: + var hashes [][]byte + err := json.Unmarshal(e.Data, &hashes) + return port.SessionsRevokedDelivery{SessionTokenHashes: hashes}, err case channel == preparedChannel && e.Kind == kindPrepared: var o port.PrepareOutcome err := json.Unmarshal(e.Data, &o) diff --git a/internal/adapter/out/redis/bus_test.go b/internal/adapter/out/redis/bus_test.go index 648c933..88945b2 100644 --- a/internal/adapter/out/redis/bus_test.go +++ b/internal/adapter/out/redis/bus_test.go @@ -344,6 +344,25 @@ func TestPreparedOutcomes(t *testing.T) { } } +func TestSessionsRevoked(t *testing.T) { + client := newClient(t) + a, b := newBus(t, client, "v1"), newBus(t, client, "v1") + for _, f := range []*busFixture{a, b} { + if err := f.bus.SubscribeInstance(t.Context()); err != nil { + t.Fatal(err) + } + } + hashes := [][]byte{{0x01, 0x02}, {0xff, 0x00, 0x7f}} + if err := a.bus.PublishSessionsRevoked(t.Context(), hashes); err != nil { + t.Fatal(err) + } + for _, f := range []*busFixture{a, b} { + if got := f.next(t); !reflect.DeepEqual(got, port.SessionsRevokedDelivery{SessionTokenHashes: hashes}) { + t.Errorf("收到 %#v，期望被删除的会话 %x", got, hashes) + } + } +} + func TestUnsubscribeRoom(t *testing.T) { client := newClient(t) a, b := newBus(t, client, "v1"), newBus(t, client, "v1") diff --git a/internal/adapter/out/redis/codec.go b/internal/adapter/out/redis/codec.go index d7112b0..bdc42d1 100644 --- a/internal/adapter/out/redis/codec.go +++ b/internal/adapter/out/redis/codec.go @@ -55,11 +55,12 @@ type ( ) const ( - kindEmitted = "Emitted" - kindReplied = "Replied" - kindInput = "RoomInput" - kindMember = "Member" - kindPrepared = "Prepared" + kindEmitted = "Emitted" + kindReplied = "Replied" + kindInput = "RoomInput" + kindMember = "Member" + kindSessionsRevoked = "SessionsRevoked" + kindPrepared = "Prepared" // room.Command 的 Action 是接口，不能像 inputs 中的类型那样直接编码。 inputCommand = "Command" ) diff --git a/internal/app/admin/admin_test.go b/internal/app/admin/admin_test.go index d054a69..ac630bf 100644 --- a/internal/app/admin/admin_test.go +++ b/internal/app/admin/admin_test.go @@ -5,6 +5,7 @@ import ( "crypto/sha256" "errors" "log/slog" + "reflect" "strconv" "strings" "testing" @@ -149,7 +150,7 @@ func (f fixture) enter(t *testing.T, roomID, name string) ([]byte, *realtime.Ses if err := f.store.CreateTicket(t.Context(), port.NewTicket{TokenHash: hash[:], RoomID: roomID, MemberID: m.ID, PasswordHash: r.PasswordHash}); err != nil { t.Fatal(err) } - session, err := f.gateway.Join(t.Context(), roomID, m) + session, err := f.gateway.Join(t.Context(), roomID, m, nil) if err != nil { t.Fatal(err) } @@ -322,7 +323,7 @@ func TestRoomAPIKey(t *testing.T) { if _, err := holder(key); !errors.Is(err, port.ErrTicketNotFound) { t.Errorf("重置之后旧 Key 的错误 = %v，期望 ErrTicketNotFound", err) } - if got, err := holder(reset); err != nil || got != bot { + if got, err := holder(reset); err != nil || !reflect.DeepEqual(got, bot) { t.Errorf("重置之后新 Key 的持有者 = %+v, %v，期望 bot 成员不变", got, err) } @@ -331,7 +332,7 @@ func TestRoomAPIKey(t *testing.T) { if err := f.admin.UpdateRoom(t.Context(), id, RoomChange{Password: ptr(password)}); err != nil { t.Fatal(err) } - if got, err := holder(reset); err != nil || got != bot { + if got, err := holder(reset); err != nil || !reflect.DeepEqual(got, bot) { t.Errorf("密码改为 %q 之后 Key 的持有者 = %+v, %v，期望 Key 仍然有效", password, got, err) } } diff --git a/internal/app/identity/account.go b/internal/app/identity/account.go index d414162..3f461ae 100644 --- a/internal/app/identity/account.go +++ b/internal/app/identity/account.go @@ -3,10 +3,12 @@ package identity import ( "context" "errors" + "log/slog" "net/netip" "time" "github.com/dihemoe/alisten/internal/app/port" + "github.com/dihemoe/alisten/internal/app/realtime" "github.com/dihemoe/alisten/internal/domain/member" ) @@ -37,14 +39,18 @@ type Account struct { Email string } +// 删除会话之后，用这些会话的票据建立的房间连接立即断开；数据已经修改而断开的通知没有发出时返回错误， +// 只有撤销发起请求的这台设备自己的会话（Login、Logout）例外，只记日志。 type Accounts struct { members port.Members site port.Site quota port.AccountQuota + gateway *realtime.Gateway + logger *slog.Logger } -func NewAccounts(members port.Members, site port.Site, quota port.AccountQuota) *Accounts { - return &Accounts{members: members, site: site, quota: quota} +func NewAccounts(members port.Members, site port.Site, quota port.AccountQuota, gateway *realtime.Gateway, logger *slog.Logger) *Accounts { + return &Accounts{members: members, site: site, quota: quota, gateway: gateway, logger: logger} } // 游客没有账号，ok 为 false。 @@ -97,7 +103,8 @@ func (a *Accounts) Login(ctx context.Context, current member.Member, currentToke return Session{}, Account{}, err } token, hash := newSecret() - err = a.members.ReplaceSession(ctx, hashSecret(currentToken), account.Member.ID, hash, account.PasswordHash) + guest := hashSecret(currentToken) + err = a.members.ReplaceSession(ctx, guest, account.Member.ID, hash, account.PasswordHash) switch { case errors.Is(err, port.ErrSessionNotFound): return Session{}, Account{}, ErrInvalidMemberToken @@ -106,6 +113,10 @@ func (a *Accounts) Login(ctx context.Context, current member.Member, currentToke case err != nil: return Session{}, Account{}, err } + // 页面随后会重新加载，游客的连接即使没断开也不再使用；此时返回错误会让用户拿不到已经签发的新会话。 + if err := a.gateway.RevokeSessions(ctx, [][]byte{guest}); err != nil { + a.logger.WarnContext(ctx, "登录后断开游客会话的通知没有发出", "error", err) + } return Session{MemberToken: token, Member: account.Member}, Account{Username: account.Username, Email: account.Email}, nil } @@ -135,7 +146,15 @@ func (a *Accounts) verify(ctx context.Context, username, password string) (port. } func (a *Accounts) Logout(ctx context.Context, token string) error { - return a.members.DeleteSession(ctx, hashSecret(token)) + hash := hashSecret(token) + if err := a.members.DeleteSession(ctx, hash); err != nil { + return err + } + // 页面随后会重新加载，不因通知没有发出而让退出失败。 + if err := a.gateway.RevokeSessions(ctx, [][]byte{hash}); err != nil { + a.logger.WarnContext(ctx, "退出后断开会话的通知没有发出", "error", err) + } + return nil } // 修改成功后，账号除了发出请求的这个会话，其他会话都失效。 @@ -162,14 +181,16 @@ func (a *Accounts) ChangePassword(ctx context.Context, memberID, token, currentP return err } // 校验之后密码可能被重置或被别的设备改过：发出请求的会话已经被删掉时按 token 失效处理，同一个会话先改过时当前密码已经不对。 - err = a.members.ChangePassword(ctx, memberID, account.PasswordHash, hash, hashSecret(token)) + deleted, err := a.members.ChangePassword(ctx, memberID, account.PasswordHash, hash, hashSecret(token)) switch { case errors.Is(err, port.ErrSessionNotFound): return ErrInvalidMemberToken case errors.Is(err, port.ErrAccountPasswordChanged): return ErrWrongCurrentPassword + case err != nil: + return err } - return err + return a.gateway.RevokeSessions(ctx, deleted) } // Find 和 ResetPassword 只供后台使用，调用方要先校验管理 token。 @@ -192,7 +213,11 @@ func (a *Accounts) ResetPassword(ctx context.Context, username string) (string, if err != nil { return "", err } - if err := a.members.SetPassword(ctx, account.Member.ID, hash); err != nil { + deleted, err := a.members.SetPassword(ctx, account.Member.ID, hash) + if err != nil { + return "", err + } + if err := a.gateway.RevokeSessions(ctx, deleted); err != nil { return "", err } return password, nil diff --git a/internal/app/identity/account_test.go b/internal/app/identity/account_test.go index dfaac43..1681fbe 100644 --- a/internal/app/identity/account_test.go +++ b/internal/app/identity/account_test.go @@ -1,9 +1,11 @@ package identity import ( + "bytes" "context" "errors" "fmt" + "log/slog" "regexp" "slices" "sync" @@ -12,6 +14,7 @@ import ( "github.com/dihemoe/alisten/internal/app/port" "github.com/dihemoe/alisten/internal/app/port/porttest" + "github.com/dihemoe/alisten/internal/app/realtime" "github.com/dihemoe/alisten/internal/domain/member" ) @@ -36,11 +39,65 @@ func (q *fakeQuota) Take(_ context.Context, key string, burst int, refill time.D return 0, nil } -func newAccounts(t *testing.T) (*Accounts, *Service, *porttest.Store, *fakeQuota) { +// revoked 返回此前广播过的被删除的会话。 +func newAccounts(t *testing.T) (a *Accounts, s *Service, store *porttest.Store, q *fakeQuota, revoked func() [][]byte) { t.Helper() - s, store := newService(t) - q := &fakeQuota{exhausted: map[string]bool{}} - return NewAccounts(store, store, q), s, store, q + store = porttest.New() + node := porttest.NewRedis().Node("a") + if err := node.SubscribeInstance(t.Context()); err != nil { + t.Fatal(err) + } + gateway := newGateway(store, node, node) + q = &fakeQuota{exhausted: map[string]bool{}} + revoked = func() [][]byte { + t.Helper() + done, cancel := context.WithCancel(t.Context()) + cancel() + var hashes [][]byte + for { + d, err := node.Receive(done) + if err != nil { + return hashes + } + if r, ok := d.(port.SessionsRevokedDelivery); ok { + hashes = append(hashes, r.SessionTokenHashes...) + } + } + } + return NewAccounts(store, store, q, gateway, slog.New(slog.DiscardHandler)), New(store, store, store, gateway), store, q, revoked +} + +// 不计顺序。 +func sameHashes(got [][]byte, tokens ...string) bool { + want := make([]string, len(tokens)) + for i, token := range tokens { + want[i] = string(hashSecret(token)) + } + keys := make([]string, len(got)) + for i, h := range got { + keys[i] = string(h) + } + slices.Sort(want) + slices.Sort(keys) + return slices.Equal(keys, want) +} + +func issueTicket(t *testing.T, s *Service, session Session, roomID string) string { + t.Helper() + ticket, err := s.IssueTicket(t.Context(), session.Member.ID, session.MemberToken, roomID, "") + if err != nil { + t.Fatal(err) + } + return ticket +} + +func ticketValid(t *testing.T, s *Service, ticket string) bool { + t.Helper() + _, err := s.VerifyTicket(t.Context(), ticket) + if err != nil && !errors.Is(err, ErrInvalidTicket) { + t.Fatal(err) + } + return err == nil } // 返回 token 当前所属的成员；token 已经失效时 ok 为 false。 @@ -66,7 +123,7 @@ func registerAccount(t *testing.T, a *Accounts, s *Service, username, password s } func TestRegister(t *testing.T) { - a, s, store, q := newAccounts(t) + a, s, store, q, _ := newAccounts(t) guest := createMember(t, s) account, err := a.Register(t.Context(), guest.Member.ID, clientIP, "alice", "correct horse") if err != nil || account != (Account{Username: "alice"}) { @@ -91,7 +148,7 @@ func TestRegister(t *testing.T) { } func TestAccountKeepsEmail(t *testing.T) { - a, s, _, _ := newAccounts(t) + a, s, _, _, _ := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") guest := createMember(t, s) @@ -122,14 +179,14 @@ func TestAccountKeepsEmail(t *testing.T) { } func TestGuestHasNoAccount(t *testing.T) { - a, s, _, _ := newAccounts(t) + a, s, _, _, _ := newAccounts(t) if _, ok, err := a.Account(t.Context(), createMember(t, s).Member.ID); err != nil || ok { t.Errorf("游客的 Account = %v, %v，期望没有账号", ok, err) } } func TestRegisterRejected(t *testing.T) { - a, s, store, q := newAccounts(t) + a, s, store, q, _ := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") if _, err := a.Register(t.Context(), createMember(t, s).Member.ID, clientIP, "alice", "another horse"); !errors.Is(err, port.ErrUsernameTaken) { t.Errorf("用户名已被占用时 = %v，期望 ErrUsernameTaken", err) @@ -160,14 +217,22 @@ func TestRegisterRejected(t *testing.T) { } func TestLogin(t *testing.T) { - a, s, store, q := newAccounts(t) + a, s, store, q, revoked := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") guest := createMember(t, s) + id := createRoom(t, store, "") + guestTicket, aliceTicket := issueTicket(t, s, guest, id), issueTicket(t, s, alice, id) q.taken = nil session, account, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse") if err != nil { t.Fatal(err) } + if got := revoked(); !sameHashes(got, guest.MemberToken) { + t.Errorf("广播的被删除的会话 = %x，期望游客原来的会话", got) + } + if ticketValid(t, s, guestTicket) || !ticketValid(t, s, aliceTicket) { + t.Error("登录之后只有游客原来的会话的票据失效") + } if session.Member != alice.Member || account != (Account{Username: "alice"}) { t.Errorf("Login = %#v, %#v，期望登录 alice", session.Member, account) } @@ -189,7 +254,7 @@ func TestLogin(t *testing.T) { } func TestLoginRejected(t *testing.T) { - a, s, store, q := newAccounts(t) + a, s, store, q, _ := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") guest := createMember(t, s) for _, tc := range []struct{ name, username, password string }{ @@ -217,7 +282,7 @@ func TestLoginRejected(t *testing.T) { } func TestAccountQuotaCountsIPv6ByNetwork(t *testing.T) { - a, s, _, q := newAccounts(t) + a, s, _, q, _ := newAccounts(t) guest := createMember(t, s) q.exhausted["login:ip:2001:db8:1:2::/64"] = true for _, tc := range []struct { @@ -250,24 +315,36 @@ func TestAccountQuotaCountsIPv6ByNetwork(t *testing.T) { } func TestLogout(t *testing.T) { - a, s, store, _ := newAccounts(t) + a, s, store, _, revoked := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") + id := createRoom(t, store, "") + aliceTicket := issueTicket(t, s, alice, id) + revoked() if err := a.Logout(t.Context(), alice.MemberToken); err != nil { t.Fatal(err) } if _, ok := tokenOwner(t, store, alice.MemberToken); ok { t.Error("退出登录之后 token 应失效") } + if got := revoked(); !sameHashes(got, alice.MemberToken) { + t.Errorf("退出登录之后广播的会话 = %x，期望当前会话", got) + } + if ticketValid(t, s, aliceTicket) { + t.Error("退出登录之后这个会话的票据应失效") + } } func TestChangePassword(t *testing.T) { - a, s, store, q := newAccounts(t) + a, s, store, q, revoked := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") guest := createMember(t, s) other, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse") if err != nil { t.Fatal(err) } + id := createRoom(t, store, "") + aliceTicket, otherTicket := issueTicket(t, s, alice, id), issueTicket(t, s, other, id) + revoked() q.taken = nil if err := a.ChangePassword(t.Context(), alice.Member.ID, alice.MemberToken, "wrong horse", "battery staple"); !errors.Is(err, ErrWrongCurrentPassword) { t.Errorf("当前密码错误时 = %v，期望 ErrWrongCurrentPassword", err) @@ -284,6 +361,12 @@ func TestChangePassword(t *testing.T) { if _, ok := tokenOwner(t, store, other.MemberToken); ok { t.Error("修改密码之后其他会话应失效") } + if got := revoked(); !sameHashes(got, other.MemberToken) { + t.Errorf("修改密码之后广播的会话 = %x，期望其他设备的会话", got) + } + if !ticketValid(t, s, aliceTicket) || ticketValid(t, s, otherTicket) { + t.Error("修改密码之后只有其他设备的票据失效") + } next := createMember(t, s) if _, _, err := a.Login(t.Context(), next.Member, next.MemberToken, clientIP, "alice", "correct horse"); !errors.Is(err, ErrInvalidCredentials) { t.Errorf("用旧密码登录 = %v，期望 ErrInvalidCredentials", err) @@ -298,8 +381,9 @@ func TestChangePassword(t *testing.T) { } func TestResetPassword(t *testing.T) { - a, s, store, _ := newAccounts(t) + a, s, store, _, revoked := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") + aliceTicket := issueTicket(t, s, alice, createRoom(t, store, "")) if _, err := a.Find(t.Context(), "nobody"); !errors.Is(err, ErrAccountNotFound) { t.Errorf("Find 不存在的用户名 = %v，期望 ErrAccountNotFound", err) } @@ -319,26 +403,107 @@ func TestResetPassword(t *testing.T) { if _, ok := tokenOwner(t, store, alice.MemberToken); ok { t.Error("重置密码之后账号的所有会话应失效") } + if got := revoked(); !sameHashes(got, alice.MemberToken) { + t.Errorf("重置密码之后广播的会话 = %x，期望账号的所有会话", got) + } + if ticketValid(t, s, aliceTicket) { + t.Error("重置密码之后账号的票据应失效") + } guest := createMember(t, s) if _, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", password); err != nil { t.Errorf("用重置后的密码登录 = %v", err) } } +type noRevocation struct { + port.Bus +} + +func (noRevocation) PublishSessionsRevoked(context.Context, [][]byte) error { + return porttest.ErrRedisDown +} + +// 撤销的是别的设备的会话：断开的通知没有发出时报错，让用户知道它们没有被踢出去。 +func TestOtherDeviceOperationsFailWhenRevocationFails(t *testing.T) { + store := porttest.New() + node := porttest.NewRedis().Node("a") + gateway := newGateway(store, node, noRevocation{node}) + s := New(store, store, store, gateway) + a := NewAccounts(store, store, &fakeQuota{exhausted: map[string]bool{}}, gateway, slog.New(slog.DiscardHandler)) + alice := registerAccount(t, a, s, "alice", "correct horse") + login := func() { + t.Helper() + guest := createMember(t, s) + if _, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse"); err != nil { + t.Fatal(err) + } + } + // 每一步之前都先登录一次，让这一步有会话可删。 + for _, step := range []struct { + name string + op func() error + }{ + {"ChangePassword", func() error { + return a.ChangePassword(t.Context(), alice.Member.ID, alice.MemberToken, "correct horse", "correct horse") + }}, + {"ResetPassword", func() error { + _, err := a.ResetPassword(t.Context(), "alice") + return err + }}, + } { + login() + if err := step.op(); !errors.Is(err, realtime.ErrUnavailable) { + t.Errorf("广播失败时 %s = %v，期望 realtime.ErrUnavailable", step.name, err) + } + } +} + +// 登录和退出撤销的是发起请求的这台设备的会话，页面随后会重新加载：广播失败时只记日志，照常成功。 +func TestLoginAndLogoutSucceedWhenRevocationFails(t *testing.T) { + store := porttest.New() + node := porttest.NewRedis().Node("a") + gateway := newGateway(store, node, noRevocation{node}) + s := New(store, store, store, gateway) + var logs bytes.Buffer + a := NewAccounts(store, store, &fakeQuota{exhausted: map[string]bool{}}, gateway, slog.New(slog.NewJSONHandler(&logs, nil))) + alice := registerAccount(t, a, s, "alice", "correct horse") + + guest := createMember(t, s) + session, _, err := a.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse") + if err != nil { + t.Fatalf("广播失败时 Login = %v，期望成功", err) + } + if _, ok := tokenOwner(t, store, guest.MemberToken); ok { + t.Error("游客原来的会话应已经删除") + } + if m, ok := tokenOwner(t, store, session.MemberToken); !ok || m.ID != alice.Member.ID { + t.Error("新会话应已经签发") + } + if err := a.Logout(t.Context(), session.MemberToken); err != nil { + t.Fatalf("广播失败时 Logout = %v，期望成功", err) + } + if _, ok := tokenOwner(t, store, session.MemberToken); ok { + t.Error("退出的会话应已经删除") + } + if n := bytes.Count(logs.Bytes(), []byte(`"level":"WARN"`)); n != 2 { + t.Errorf("警告日志 %d 条，期望 Login 和 Logout 各 1 条：\n%s", n, logs.String()) + } +} + // 在登录校验密码之后、签发新会话之前修改账号的密码。 type passwordChangedMembers struct { *porttest.Store } func (s passwordChangedMembers) ReplaceSession(ctx context.Context, oldTokenHash []byte, memberID string, newTokenHash []byte, passwordHash string) error { - if err := s.SetPassword(ctx, memberID, "changed"); err != nil { + if _, err := s.SetPassword(ctx, memberID, "changed"); err != nil { return err } return s.Store.ReplaceSession(ctx, oldTokenHash, memberID, newTokenHash, passwordHash) } func TestLoginRejectedWhenGuestSessionOrPasswordChangesMeanwhile(t *testing.T) { - a, s, store, _ := newAccounts(t) + a, s, store, _, _ := newAccounts(t) registerAccount(t, a, s, "alice", "correct horse") guest := createMember(t, s) if err := store.DeleteSession(t.Context(), hashSecret(guest.MemberToken)); err != nil { @@ -348,7 +513,8 @@ func TestLoginRejectedWhenGuestSessionOrPasswordChangesMeanwhile(t *testing.T) { t.Errorf("游客的会话在校验之后失效，Login = %v，期望 ErrInvalidMemberToken", err) } - changing := NewAccounts(passwordChangedMembers{store}, store, &fakeQuota{exhausted: map[string]bool{}}) + node := porttest.NewRedis().Node("a") + changing := NewAccounts(passwordChangedMembers{store}, store, &fakeQuota{exhausted: map[string]bool{}}, newGateway(store, node, node), slog.New(slog.DiscardHandler)) guest = createMember(t, s) if _, _, err := changing.Login(t.Context(), guest.Member, guest.MemberToken, clientIP, "alice", "correct horse"); !errors.Is(err, ErrInvalidCredentials) { t.Errorf("密码在校验之后被修改，Login = %v，期望 ErrInvalidCredentials", err) @@ -371,10 +537,11 @@ func (s passwordReadMembers) AccountByUsername(ctx context.Context, username str } func TestChangePasswordRejectedWhenPasswordChangesMeanwhile(t *testing.T) { - a, s, store, _ := newAccounts(t) + a, s, store, _, _ := newAccounts(t) alice := registerAccount(t, a, s, "alice", "correct horse") + node := porttest.NewRedis().Node("a") racing := func(meanwhile func()) *Accounts { - return NewAccounts(passwordReadMembers{store, meanwhile}, store, &fakeQuota{exhausted: map[string]bool{}}) + return NewAccounts(passwordReadMembers{store, meanwhile}, store, &fakeQuota{exhausted: map[string]bool{}}, newGateway(store, node, node), slog.New(slog.DiscardHandler)) } var reset string @@ -409,7 +576,7 @@ func TestChangePasswordRejectedWhenPasswordChangesMeanwhile(t *testing.T) { // 名额占满、请求已经取消时，要做密码计算的请求以 context.Canceled 结束，不做的直接返回结果。 func TestLoginDoesPasswordWorkForUnknownUsername(t *testing.T) { - a, s, _, _ := newAccounts(t) + a, s, _, _, _ := newAccounts(t) registerAccount(t, a, s, "alice", "correct horse") guest := createMember(t, s) if _, err := dummyPasswordHash(); err != nil { diff --git a/internal/app/identity/identity.go b/internal/app/identity/identity.go index 250b304..119f122 100644 --- a/internal/app/identity/identity.go +++ b/internal/app/identity/identity.go @@ -84,7 +84,8 @@ func (s *Service) UpdateProfile(ctx context.Context, memberID string, p Profile) return m, nil } -func (s *Service) IssueTicket(ctx context.Context, memberID, roomID, password string) (string, error) { +// memberToken 是发出请求的会话，票据属于这个会话：会话删除时票据随之失效。 +func (s *Service) IssueTicket(ctx context.Context, memberID, memberToken, roomID, password string) (string, error) { r, err := s.rooms.FindRoom(ctx, roomID) if err != nil { return "", err @@ -99,11 +100,15 @@ func (s *Service) IssueTicket(ctx context.Context, memberID, roomID, password st } } ticket, hash := newSecret() - err = s.tickets.CreateTicket(ctx, port.NewTicket{TokenHash: hash, RoomID: r.Info.ID, MemberID: memberID, PasswordHash: r.PasswordHash}) + err = s.tickets.CreateTicket(ctx, port.NewTicket{TokenHash: hash, RoomID: r.Info.ID, MemberID: memberID, PasswordHash: r.PasswordHash, SessionTokenHash: hashSecret(memberToken)}) if errors.Is(err, port.ErrPasswordChanged) { // 刚才是按旧密码校验的；客户端重新输入密码后，按新密码校验。 return "", ErrWrongPassword } + if errors.Is(err, port.ErrSessionNotFound) { + // 会话在鉴权之后被删除（例如另一台设备退出登录或修改了密码），此时成员 token 确实已经失效。 + return "", ErrInvalidMemberToken + } if err != nil { return "", err } @@ -137,7 +142,16 @@ func (s *Service) EnterRoom(ctx context.Context, roomID, ticket string) (*realti if holder.Bot { return nil, ErrBotSession } - return s.gateway.Join(logctx.WithMember(logctx.WithRoom(ctx, r.Info.ID), holder.Member.ID), r.Info.ID, holder.Member) + session, err := s.gateway.Join(logctx.WithMember(logctx.WithRoom(ctx, r.Info.ID), holder.Member.ID), r.Info.ID, holder.Member, holder.SessionTokenHash) + if err != nil { + return nil, err + } + // 作废时先删除票据、再通知断开：会话加入网关之前发出的通知它收不到，所以加入之后再确认一次票据。 + if _, err := s.VerifyTicket(ctx, ticket); err != nil { + s.gateway.Leave(session) + return nil, err + } + return session, nil } func NewAPIKey() (string, port.NewAPIKey) { diff --git a/internal/app/identity/identity_test.go b/internal/app/identity/identity_test.go index cd66dad..864ec2d 100644 --- a/internal/app/identity/identity_test.go +++ b/internal/app/identity/identity_test.go @@ -9,6 +9,7 @@ import ( "log/slog" "regexp" "testing" + "testing/synctest" "github.com/dihemoe/alisten/internal/app/port" "github.com/dihemoe/alisten/internal/app/port/porttest" @@ -152,14 +153,88 @@ func TestIssueTicketFailsWhenPasswordChangesMeanwhile(t *testing.T) { changed := passwordChangedStore{Store: store, newHash: hashPassword(t, "新密码")} node := porttest.NewRedis().Node("a") s := New(store, store, changed, newGateway(store, node, node)) - alice := createMember(t, s).Member + alice := createMember(t, s) id := createRoom(t, store, "旧密码") - if _, err := s.IssueTicket(t.Context(), alice.ID, id, "旧密码"); !errors.Is(err, ErrWrongPassword) { + if _, err := s.IssueTicket(t.Context(), alice.Member.ID, alice.MemberToken, id, "旧密码"); !errors.Is(err, ErrWrongPassword) { t.Errorf("密码在校验之后被修改，错误 = %v，期望 ErrWrongPassword", err) } } +func TestTicketBelongsToIssuingSession(t *testing.T) { + s, store := newService(t) + alice := createMember(t, s) + id := createRoom(t, store, "") + ticket, err := s.IssueTicket(t.Context(), alice.Member.ID, alice.MemberToken, id, "") + if err != nil { + t.Fatal(err) + } + holder, err := s.VerifyTicket(t.Context(), ticket) + if err != nil || holder.Member != alice.Member || !bytes.Equal(holder.SessionTokenHash, hashSecret(alice.MemberToken)) { + t.Errorf("票据的持有者 = %#v, %v，期望属于 alice 签发它的会话", holder, err) + } +} + +func TestIssueTicketRejectsDeletedSession(t *testing.T) { + s, store := newService(t) + alice := createMember(t, s) + id := createRoom(t, store, "") + if err := store.DeleteSession(t.Context(), hashSecret(alice.MemberToken)); err != nil { + t.Fatal(err) + } + + if _, err := s.IssueTicket(t.Context(), alice.Member.ID, alice.MemberToken, id, ""); !errors.Is(err, ErrInvalidMemberToken) { + t.Errorf("会话在签发之前被删除，错误 = %v，期望 ErrInvalidMemberToken", err) + } +} + +// 第一次查询照常，之后的查询都认为票据已经作废。 +type revokedAfterFirstLookup struct { + port.Tickets + lookups int +} + +func (r *revokedAfterFirstLookup) TicketHolder(ctx context.Context, tokenHash []byte) (port.TicketHolder, error) { + r.lookups++ + if r.lookups > 1 { + return port.TicketHolder{}, port.ErrTicketNotFound + } + return r.Tickets.TicketHolder(ctx, tokenHash) +} + +func TestEnterRoomLeavesWhenTicketIsRevokedWhileJoining(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + store := porttest.New() + node := porttest.NewRedis().Node("a") + host, gateway := realtime.New(store, store, nil, porttest.NewPreparer(), realtime.Cluster{Instance: "a", Leases: node, Liveness: node, Bus: node}, slog.New(slog.DiscardHandler)) + ctx, stop := context.WithCancel(t.Context()) + wait, err := realtime.Start(ctx, host, gateway) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = host.Shutdown(context.Background()) + _ = gateway.Shutdown(context.Background()) + stop() + wait() + }) + s := New(store, store, &revokedAfterFirstLookup{Tickets: store}, gateway) + alice := createMember(t, s) + id := createRoom(t, store, "") + ticket, err := s.IssueTicket(t.Context(), alice.Member.ID, alice.MemberToken, id, "") + if err != nil { + t.Fatal(err) + } + + if _, err := s.EnterRoom(t.Context(), id, ticket); !errors.Is(err, ErrInvalidTicket) { + t.Errorf("加入期间票据被作废，错误 = %v，期望 ErrInvalidTicket", err) + } + if n := gateway.Sessions(); n != 0 { + t.Errorf("网关上残留 %d 个会话，期望 0", n) + } + }) +} + type loggedSubscriptions struct { port.Bus logger *slog.Logger @@ -177,12 +252,13 @@ func TestEnterRoomJoinsTheStoredRoom(t *testing.T) { node := porttest.NewRedis().Node("a") var logs bytes.Buffer s := New(store, store, store, newGateway(store, node, loggedSubscriptions{node, slog.New(logctx.NewHandler(slog.NewJSONHandler(&logs, nil)))})) - alice := createMember(t, s).Member + session := createMember(t, s) + alice := session.Member id, err := store.CreateRoom(t.Context(), port.NewRoom{Kind: room.Persistent, Name: "大厅", Slug: "hall"}) if err != nil { t.Fatal(err) } - ticket, err := s.IssueTicket(t.Context(), alice.ID, id, "") + ticket, err := s.IssueTicket(t.Context(), alice.ID, session.MemberToken, id, "") if err != nil { t.Fatal(err) } diff --git a/internal/app/lobby/lobby_test.go b/internal/app/lobby/lobby_test.go index f2140a7..f9cf93c 100644 --- a/internal/app/lobby/lobby_test.go +++ b/internal/app/lobby/lobby_test.go @@ -47,7 +47,7 @@ func TestOnlineCount(t *testing.T) { // 同一个成员的两个会话算两个在线。 alice := member.Member{ID: "00000000-0000-7000-8000-00000000000a", Name: "alice"} for range 2 { - session, err := gateway.Join(t.Context(), busy, alice) + session, err := gateway.Join(t.Context(), busy, alice, nil) if err != nil { t.Fatal(err) } diff --git a/internal/app/port/bus.go b/internal/app/port/bus.go index 49bcce4..9cd4e14 100644 --- a/internal/app/port/bus.go +++ b/internal/app/port/bus.go @@ -92,19 +92,25 @@ type ( Outcome PrepareOutcome } + // 这些会话已经删除，用它们签发的票据建立的会话和事件流要以 4401 断开。 + SessionsRevokedDelivery struct { + SessionTokenHashes [][]byte + } + // 订阅连接曾经断开，断开期间发给本实例的消息都已丢失。 SubscriptionLost struct{} ) -func (RoomDelivery) delivery() {} -func (InputDelivery) delivery() {} -func (MemberDelivery) delivery() {} -func (PreparedDelivery) delivery() {} -func (SubscriptionLost) delivery() {} +func (RoomDelivery) delivery() {} +func (InputDelivery) delivery() {} +func (MemberDelivery) delivery() {} +func (PreparedDelivery) delivery() {} +func (SessionsRevokedDelivery) delivery() {} +func (SubscriptionLost) delivery() {} // 实现绑定在本实例上：订阅的是本实例的频道，发出的消息带有本实例的 ID。 type Bus interface { - // 订阅本实例的频道、成员资料的频道和准备结果的频道，返回时订阅已经生效。 + // 订阅本实例的频道、成员的频道和准备结果的频道，返回时订阅已经生效。 SubscribeInstance(ctx context.Context) error // 返回时订阅已经生效：此后发布到这个房间频道的消息都会收到。 SubscribeRoom(ctx context.Context, roomID string) error @@ -114,6 +120,8 @@ type Bus interface { SendInput(ctx context.Context, instanceID string, in RoomInput) (bool, error) PublishMember(ctx context.Context, m member.Member) error // 各实例都会收到，包括发布的实例。 + PublishSessionsRevoked(ctx context.Context, sessionTokenHashes [][]byte) error + // 各实例都会收到，包括发布的实例。 PublishPrepared(ctx context.Context, o PrepareOutcome) error // 按收到的顺序返回下一条消息，直到 ctx 结束；ctx 结束之后 Bus 不能再使用。只能由一个 goroutine 调用。 Receive(ctx context.Context) (Delivery, error) diff --git a/internal/app/port/member.go b/internal/app/port/member.go index 1804ebb..9149a9f 100644 --- a/internal/app/port/member.go +++ b/internal/app/port/member.go @@ -30,6 +30,7 @@ type AccountProfile struct { } // 成员 token 只以 SHA-256 哈希的形式出现在端口中。一个成员 token 是成员的一个会话，同一个成员可以有多个会话。 +// 删除会话时，签发给这个会话的票据一并删除。 type Members interface { // 创建游客和它的第一个会话。 CreateMember(ctx context.Context, tokenHash []byte, name string) (member.Member, error) @@ -45,9 +46,9 @@ type Members interface { // 修改时对会话的删除已经执行，此时创建的会话不会再被删除。两种情况都不做任何修改。 ReplaceSession(ctx context.Context, oldTokenHash []byte, memberID string, newTokenHash []byte, passwordHash string) error DeleteSession(ctx context.Context, tokenHash []byte) error - // 在同一个事务中修改已注册成员的密码哈希，并删除该成员的全部会话。成员是游客时返回 ErrMemberNotFound。 - SetPassword(ctx context.Context, memberID, passwordHash string) error + // 在同一个事务中修改已注册成员的密码哈希，并删除该成员的全部会话。返回被删除的会话的 token 哈希。成员是游客时返回 ErrMemberNotFound。 + SetPassword(ctx context.Context, memberID, passwordHash string) ([][]byte, error) // 与 SetPassword 相同，但保留发出请求的 keepTokenHash 的会话。这个会话已经不存在时返回 ErrSessionNotFound；密码哈希已经不是 // 校验当前密码时读到的 currentPasswordHash 时返回 ErrAccountPasswordChanged：密码在校验之后被修改或重置过。两种情况都不做任何修改。 - ChangePassword(ctx context.Context, memberID, currentPasswordHash, passwordHash string, keepTokenHash []byte) error + ChangePassword(ctx context.Context, memberID, currentPasswordHash, passwordHash string, keepTokenHash []byte) ([][]byte, error) } diff --git a/internal/app/port/porttest/redis.go b/internal/app/port/porttest/redis.go index 4569239..e8f809a 100644 --- a/internal/app/port/porttest/redis.go +++ b/internal/app/port/porttest/redis.go @@ -282,6 +282,11 @@ func (n *Node) PublishMember(_ context.Context, m member.Member) error { return err } +func (n *Node) PublishSessionsRevoked(_ context.Context, sessionTokenHashes [][]byte) error { + _, err := n.publish("members", port.SessionsRevokedDelivery{SessionTokenHashes: sessionTokenHashes}) + return err +} + func (n *Node) PublishPrepared(_ context.Context, o port.PrepareOutcome) error { _, err := n.publish("tracks.prepared", port.PreparedDelivery{Outcome: o}) return err diff --git a/internal/app/port/porttest/store.go b/internal/app/port/porttest/store.go index 0124a4a..8c9f29f 100644 --- a/internal/app/port/porttest/store.go +++ b/internal/app/port/porttest/store.go @@ -45,6 +45,7 @@ type storedMember struct { type storedTicket struct { roomID string memberID string + session string createdAt time.Time } @@ -203,7 +204,7 @@ func (s *Store) ReplaceSession(_ context.Context, oldTokenHash []byte, memberID if _, ok := s.sessions[string(newTokenHash)]; ok { return errors.New("token 哈希重复") } - delete(s.sessions, string(oldTokenHash)) + s.deleteSession(string(oldTokenHash)) s.sessions[string(newTokenHash)] = memberID return nil } @@ -211,45 +212,56 @@ func (s *Store) ReplaceSession(_ context.Context, oldTokenHash []byte, memberID func (s *Store) DeleteSession(_ context.Context, tokenHash []byte) error { s.mu.Lock() defer s.mu.Unlock() - delete(s.sessions, string(tokenHash)) + s.deleteSession(string(tokenHash)) return nil } -func (s *Store) SetPassword(_ context.Context, memberID, passwordHash string) error { +func (s *Store) SetPassword(_ context.Context, memberID, passwordHash string) ([][]byte, error) { s.mu.Lock() defer s.mu.Unlock() m := s.member(memberID) if m == nil || m.username == "" { - return port.ErrMemberNotFound + return nil, port.ErrMemberNotFound } m.passwordHash = passwordHash - s.deleteSessions(memberID, nil) - return nil + return s.deleteSessions(memberID, nil), nil } -func (s *Store) ChangePassword(_ context.Context, memberID, currentPasswordHash, passwordHash string, keepTokenHash []byte) error { +func (s *Store) ChangePassword(_ context.Context, memberID, currentPasswordHash, passwordHash string, keepTokenHash []byte) ([][]byte, error) { s.mu.Lock() defer s.mu.Unlock() m := s.member(memberID) if m == nil || m.username == "" { - return port.ErrMemberNotFound + return nil, port.ErrMemberNotFound } if owner, ok := s.sessions[string(keepTokenHash)]; !ok || owner != memberID { - return port.ErrSessionNotFound + return nil, port.ErrSessionNotFound } if m.passwordHash != currentPasswordHash { - return port.ErrAccountPasswordChanged + return nil, port.ErrAccountPasswordChanged } m.passwordHash = passwordHash - s.deleteSessions(memberID, keepTokenHash) - return nil + return s.deleteSessions(memberID, keepTokenHash), nil } // 调用方持有 s.mu。 -func (s *Store) deleteSessions(memberID string, keepTokenHash []byte) { - maps.DeleteFunc(s.sessions, func(hash, id string) bool { - return id == memberID && (keepTokenHash == nil || hash != string(keepTokenHash)) - }) +func (s *Store) deleteSessions(memberID string, keepTokenHash []byte) [][]byte { + var deleted [][]byte + for hash, id := range s.sessions { + if id == memberID && (keepTokenHash == nil || hash != string(keepTokenHash)) { + deleted = append(deleted, []byte(hash)) + } + } + for _, hash := range deleted { + s.deleteSession(string(hash)) + } + return deleted +} + +// 与 postgres 的外键一致：会话的票据随会话一起删除。调用方持有 s.mu。 +func (s *Store) deleteSession(hash string) { + delete(s.sessions, hash) + maps.DeleteFunc(s.tickets, func(_ string, t storedTicket) bool { return t.session == hash }) } func (s *Store) CreateTicket(_ context.Context, t port.NewTicket) error { @@ -265,10 +277,13 @@ func (s *Store) CreateTicket(_ context.Context, t port.NewTicket) error { if !slices.ContainsFunc(s.members, func(m storedMember) bool { return m.member.ID == t.MemberID }) { return errors.New("成员不存在") } + if _, ok := s.sessions[string(t.SessionTokenHash)]; t.SessionTokenHash != nil && !ok { + return port.ErrSessionNotFound + } if _, ok := s.tickets[string(t.TokenHash)]; ok { return errors.New("票据哈希重复") } - s.tickets[string(t.TokenHash)] = storedTicket{roomID: r.Info.ID, memberID: t.MemberID, createdAt: time.Now()} + s.tickets[string(t.TokenHash)] = storedTicket{roomID: r.Info.ID, memberID: t.MemberID, session: string(t.SessionTokenHash), createdAt: time.Now()} return nil } @@ -281,7 +296,11 @@ func (s *Store) TicketHolder(_ context.Context, tokenHash []byte) (port.TicketHo } for _, m := range s.members { if m.member.ID == t.memberID { - return port.TicketHolder{RoomID: t.roomID, Member: m.member, Bot: s.bots[t.roomID] == t.memberID}, nil + holder := port.TicketHolder{RoomID: t.roomID, Member: m.member, Bot: s.bots[t.roomID] == t.memberID} + if t.session != "" { + holder.SessionTokenHash = []byte(t.session) + } + return holder, nil } } return port.TicketHolder{}, port.ErrTicketNotFound diff --git a/internal/app/port/ticket.go b/internal/app/port/ticket.go index fce12ce..b1e1ef6 100644 --- a/internal/app/port/ticket.go +++ b/internal/app/port/ticket.go @@ -18,6 +18,8 @@ type NewTicket struct { MemberID string // 签发前校验密码时依据的密码哈希，空字符串表示房间没有密码。 PasswordHash string + // 发出请求的会话的 token 哈希：会话删除时票据随之失效。 + SessionTokenHash []byte } type TicketHolder struct { @@ -26,11 +28,13 @@ type TicketHolder struct { Member member.Member // 票据是常驻房间的 API Key，持有者是这个房间的 bot 成员。 Bot bool + // 票据所属的会话的 token 哈希，API Key 没有会话，为 nil。 + SessionTokenHash []byte } type Tickets interface { // 房间的密码哈希已经不是 t.PasswordHash 时返回 ErrPasswordChanged：密码在校验之后被修改过， - // 设置或修改密码时的作废已经执行，此时签发的票据不会再被作废。 + // 设置或修改密码时的作废已经执行，此时签发的票据不会再被作废。t.SessionTokenHash 的会话已经不存在时返回 ErrSessionNotFound。 CreateTicket(ctx context.Context, t NewTicket) error TicketHolder(ctx context.Context, tokenHash []byte) (TicketHolder, error) } diff --git a/internal/app/realtime/cluster_test.go b/internal/app/realtime/cluster_test.go index a011402..be9af77 100644 --- a/internal/app/realtime/cluster_test.go +++ b/internal/app/realtime/cluster_test.go @@ -398,7 +398,7 @@ func TestGatewayRejoinsAfterItsAliveKeyExpired(t *testing.T) { if _, err := a.gateway.Do(t.Context(), id, alice, room.Sync{}); err != nil { t.Fatal(err) } - s, err := g.Join(t.Context(), id, bob) + s, err := g.Join(t.Context(), id, bob, nil) if err != nil { t.Fatal(err) } diff --git a/internal/app/realtime/gateway.go b/internal/app/realtime/gateway.go index 8e5015a..857a812 100644 --- a/internal/app/realtime/gateway.go +++ b/internal/app/realtime/gateway.go @@ -105,12 +105,12 @@ func New(snapshots port.RoomSnapshots, directory port.RoomDirectory, music *cata } // 会话结束时必须调用 Leave。 -func (g *Gateway) Join(ctx context.Context, roomID string, m member.Member) (*Session, error) { +func (g *Gateway) Join(ctx context.Context, roomID string, m member.Member, memberSession []byte) (*Session, error) { link, err := g.hold(ctx, roomID) if err != nil { return nil, err } - s := newSession(room.SessionID(g.nextID("s")), roomID, m) + s := newSession(room.SessionID(g.nextID("s")), roomID, m, memberSession) s.link = link if !g.register(s) { g.unhold(roomID, link, false) @@ -208,6 +208,19 @@ func (g *Gateway) RevokeAPIKey(ctx context.Context, roomID string) error { return g.announce(ctx, roomID, port.APIKeyReset{}) } +// RevokeSessions 在成员会话删除之后通知各实例，以 CloseRevoked 关闭用这些会话的票据建立的会话和事件流。 +func (g *Gateway) RevokeSessions(ctx context.Context, memberSessions [][]byte) error { + if len(memberSessions) == 0 { + return nil + } + ctx, cancel := context.WithTimeout(ctx, clusterTimeout) + defer cancel() + if err := g.bus.PublishSessionsRevoked(ctx, memberSessions); err != nil { + return unavailable(err) + } + return nil +} + func (g *Gateway) announce(ctx context.Context, roomID string, m port.RoomMessage) error { ctx, cancel := context.WithTimeout(ctx, clusterTimeout) defer cancel() @@ -470,6 +483,34 @@ func (g *Gateway) memberChanged(m member.Member) { }() } +// 在接收消息的 goroutine 上调用，不能阻塞。 +func (g *Gateway) sessionsRevoked(memberSessions [][]byte) { + revoked := map[string]bool{} + for _, h := range memberSessions { + revoked[string(h)] = true + } + g.mu.Lock() + var closing []*sendQueue + for _, set := range g.sessions { + for s := range set { + if revoked[s.memberSession] { + closing = append(closing, s.sendQueue) + } + } + } + for _, set := range g.observers { + for _, o := range set { + if revoked[o.memberSession] { + closing = append(closing, o.sendQueue) + } + } + } + g.mu.Unlock() + for _, q := range closing { + q.close(CloseRevoked) + } +} + // owner 靠 Join、Leave 和 MemberUpdated 知道谁在线，这几条消息丢失后没有别的办法补回：离开的会话会一直留在在线列表中， // 临时房间也因此无法关闭；仍然在线的会话则不在列表中。所以送不到 owner 时要一直重试，直到送达。 // next 在每次尝试前调用，返回这一次要发的输入；返回 false 表示这条消息已经不必再发。 @@ -628,6 +669,8 @@ func (g *Gateway) receive(ctx context.Context, lost chan<- struct{}) { g.host.deliver(d.Context, d.Input) case port.MemberDelivery: g.memberChanged(d.Member) + case port.SessionsRevokedDelivery: + g.sessionsRevoked(d.SessionTokenHashes) case port.PreparedDelivery: g.host.preparer.Finished(d.Outcome) case port.SubscriptionLost: diff --git a/internal/app/realtime/gateway_test.go b/internal/app/realtime/gateway_test.go index 2838a86..9a500e1 100644 --- a/internal/app/realtime/gateway_test.go +++ b/internal/app/realtime/gateway_test.go @@ -57,7 +57,7 @@ func startLoggingInstance(t *testing.T, redis *porttest.Redis, id string, snapsh func (in *instance) join(t *testing.T, roomID string, m member.Member) *Session { t.Helper() - s, err := in.gateway.Join(t.Context(), roomID, m) + s, err := in.gateway.Join(t.Context(), roomID, m, nil) if err != nil { t.Fatalf("%s 加入房间：%v", m.Name, err) } @@ -366,7 +366,7 @@ func TestShutdownClosesSessions(t *testing.T) { store := porttest.New() f := newGatewayFixture(t, store, store) id := createRoom(t, store, room.Persistent) - a, err := f.gateway.Join(t.Context(), id, alice) + a, err := f.gateway.Join(t.Context(), id, alice, nil) if err != nil { t.Fatal(err) } @@ -384,7 +384,7 @@ func TestShutdownClosesSessions(t *testing.T) { if err := <-done; err != nil { t.Fatal(err) } - if _, err := f.gateway.Join(t.Context(), id, bob); !errors.Is(err, ErrUnavailable) { + if _, err := f.gateway.Join(t.Context(), id, bob, nil); !errors.Is(err, ErrUnavailable) { t.Errorf("Shutdown 之后加入的错误 = %v，期望 ErrUnavailable", err) } }) @@ -463,7 +463,7 @@ func TestUnsubscribeAfterLastUse(t *testing.T) { sessionOnly := createRoom(t, store, room.Persistent) withHTTP := createRoom(t, store, room.Persistent) - s, err := f.gateway.Join(t.Context(), sessionOnly, alice) + s, err := f.gateway.Join(t.Context(), sessionOnly, alice, nil) if err != nil { t.Fatal(err) } @@ -481,7 +481,7 @@ func TestUnsubscribeAfterLastUse(t *testing.T) { t.Fatal(err) } time.Sleep(30 * time.Second) - s, err = f.gateway.Join(t.Context(), withHTTP, alice) + s, err = f.gateway.Join(t.Context(), withHTTP, alice, nil) if err != nil { t.Fatal(err) } @@ -620,7 +620,7 @@ func TestConfirmedSubscriptionIsNotRepeated(t *testing.T) { store := porttest.New() g, node, bus := startCountingGateway(t, store) id := createRoom(t, store, room.Persistent) - s, err := g.Join(t.Context(), id, alice) + s, err := g.Join(t.Context(), id, alice, nil) if err != nil { t.Fatal(err) } @@ -657,7 +657,7 @@ func TestConfirmationAcrossDisconnectIsNotReused(t *testing.T) { store := porttest.New() g, node, bus := startCountingGateway(t, store) id := createRoom(t, store, room.Persistent) - s, err := g.Join(t.Context(), id, alice) + s, err := g.Join(t.Context(), id, alice, nil) if err != nil { t.Fatal(err) } diff --git a/internal/app/realtime/observer.go b/internal/app/realtime/observer.go index e31bee6..b0fdfc4 100644 --- a/internal/app/realtime/observer.go +++ b/internal/app/realtime/observer.go @@ -17,26 +17,28 @@ var ErrTooManyObservers = errors.New("这个房间的事件流超过 5 条，请 // Observer 是房间的事件流：先收到房间的快照，之后是房间的事件；房间开始新的一次运行时事件重新编号，再收到一份新的快照。 type Observer struct { *sendQueue - room string - link *roomLink - member member.Member - bot bool - credential string + room string + link *roomLink + member member.Member + // 凭证签发给的成员会话的 token 哈希，API Key 没有会话，为空。 + memberSession string + bot bool + credential string // 由 sendQueue 的 mu 保护：最近一次快照的任期，和收到过的 owner 变更通知中最大的任期。 term, latest uint32 } // Observe 订阅房间的事件但不加入房间：事件流不出现在在线列表中，不计入投票切歌所需的人数，也不让 AutoDJ 认为房间里有人。 -// m 是凭证的持有者，bot 表示凭证是常驻房间的 API Key，credential 是凭证的哈希。名额用完时，同一凭证打开的最早的事件流以 -// CloseReplaced 关闭，名额让给这一条；没有同一凭证的事件流时返回 ErrTooManyObservers。 +// m 是凭证的持有者，memberSession 是凭证签发给的成员会话，bot 表示凭证是常驻房间的 API Key，credential 是凭证的哈希。 +// 名额用完时，同一凭证打开的最早的事件流以 CloseReplaced 关闭，名额让给这一条；没有同一凭证的事件流时返回 ErrTooManyObservers。 // 返回时快照已经放进队列。结束时必须调用一次 Unobserve。 -func (g *Gateway) Observe(ctx context.Context, roomID string, m member.Member, bot bool, credential string) (*Observer, error) { +func (g *Gateway) Observe(ctx context.Context, roomID string, m member.Member, memberSession []byte, bot bool, credential string) (*Observer, error) { link, err := g.hold(ctx, roomID) if err != nil { return nil, err } - o := &Observer{sendQueue: newSendQueue(), room: roomID, link: link, member: m, bot: bot, credential: credential} + o := &Observer{sendQueue: newSendQueue(), room: roomID, link: link, member: m, memberSession: string(memberSession), bot: bot, credential: credential} if err := g.watch(o); err != nil { g.unhold(roomID, link, false) return nil, err diff --git a/internal/app/realtime/observer_test.go b/internal/app/realtime/observer_test.go index 39b6cc6..6d53772 100644 --- a/internal/app/realtime/observer_test.go +++ b/internal/app/realtime/observer_test.go @@ -25,7 +25,7 @@ var ( // 只能在 synctest 的 bubble 中调用。 func (in *instance) observe(t *testing.T, roomID string, m member.Member, bot bool, credential string) *Observer { t.Helper() - o, err := in.gateway.Observe(t.Context(), roomID, m, bot, credential) + o, err := in.gateway.Observe(t.Context(), roomID, m, nil, bot, credential) if err != nil { t.Fatalf("%s 打开事件流：%v", m.Name, err) } @@ -126,6 +126,59 @@ func TestAPIKeyResetClosesOnlyBotObservers(t *testing.T) { }) } +func TestRevokedSessionsCloseOnlyTheirConnections(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + store := porttest.New() + f := newClusterFixture(t, store, store) + lounge := createRoom(t, store, room.Persistent) + join := func(in *instance, roomID string, m member.Member, memberSession string) *Session { + t.Helper() + s, err := in.gateway.Join(t.Context(), roomID, m, []byte(memberSession)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { in.gateway.Leave(s) }) + return s + } + observe := func(in *instance, m member.Member, memberSession []byte, bot bool, credential string) *Observer { + t.Helper() + o, err := in.gateway.Observe(t.Context(), f.room, m, memberSession, bot, credential) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { in.gateway.Unobserve(o) }) + return o + } + revokedOnA, revokedOnB, kept := join(f.a, f.room, alice, "alice-1"), join(f.b, lounge, alice, "alice-1"), join(f.b, f.room, alice, "alice-2") + other := join(f.b, f.room, bob, "bob-1") + revokedStream, keptStream := observe(f.b, alice, []byte("alice-1"), false, "t1"), observe(f.b, alice, []byte("alice-2"), false, "t2") + bot := observe(f.b, robot, nil, true, "key") + for _, q := range []interface { + Next(context.Context) (Message, CloseReason, error) + }{revokedOnA, revokedOnB, kept, other, revokedStream, keptStream, bot} { + received(t, q) + } + + if err := f.a.gateway.RevokeSessions(t.Context(), [][]byte{[]byte("alice-1"), []byte("carol-1")}); err != nil { + t.Fatal(err) + } + revoked := fmt.Sprintf("closed %d", CloseRevoked) + checkReceived(t, revokedOnA, revoked) + checkReceived(t, revokedOnB, revoked) + checkObserved(t, revokedStream, revoked) + for _, s := range []*Session{kept, other} { + if got := received(t, s); slices.Contains(got, revoked) { + t.Errorf("%s 的会话也被关闭了：%v", s.Member().Name, got) + } + } + for _, o := range []*Observer{keptStream, bot} { + if got := received(t, o); slices.Contains(got, revoked) { + t.Errorf("%s 的事件流也被关闭了：%v", o.member.Name, got) + } + } + }) +} + func TestObserverGetsNewSnapshotWhenRunChanges(t *testing.T) { synctest.Test(t, func(t *testing.T) { store := &failingStore{Store: porttest.New()} @@ -203,7 +256,7 @@ func TestObserverSkipsSnapshotOfEarlierRun(t *testing.T) { opened := make(chan error, 1) go func() { var err error - o, err = b.gateway.Observe(context.Background(), id, robot, true, "key") + o, err = b.gateway.Observe(context.Background(), id, robot, nil, true, "key") opened <- err }() synctest.Wait() @@ -225,14 +278,14 @@ func TestObserverLimitPerRoom(t *testing.T) { store := porttest.New() f := newGatewayFixture(t, store, store) full, other := createRoom(t, store, room.Persistent), createRoom(t, store, room.Persistent) - first, err := f.gateway.Observe(t.Context(), full, robot, true, "key") + first, err := f.gateway.Observe(t.Context(), full, robot, nil, true, "key") if err != nil { t.Fatal(err) } for range observerLimit - 1 { f.observe(t, full, robot, true, "key") } - if _, err := f.gateway.Observe(t.Context(), full, alice, false, "alice"); !errors.Is(err, ErrTooManyObservers) { + if _, err := f.gateway.Observe(t.Context(), full, alice, nil, false, "alice"); !errors.Is(err, ErrTooManyObservers) { t.Errorf("第 %d 个事件流的错误 = %v，期望 ErrTooManyObservers", observerLimit+1, err) } f.observe(t, other, alice, false, "alice") @@ -258,7 +311,7 @@ func TestFullRoomReplacesOldestObserverOfSameCredential(t *testing.T) { checkObserved(t, o, "snapshot 1/0 []") } - if _, err := f.gateway.Observe(t.Context(), id, bob, false, "bob 的另一张票据"); !errors.Is(err, ErrTooManyObservers) { + if _, err := f.gateway.Observe(t.Context(), id, bob, nil, false, "bob 的另一张票据"); !errors.Is(err, ErrTooManyObservers) { t.Errorf("名额用完、没有同一凭证的事件流时的错误 = %v，期望 ErrTooManyObservers", err) } for _, o := range open { @@ -273,7 +326,7 @@ func TestFailedObserveFreesItsSlot(t *testing.T) { f := newGatewayFixture(t, store, store) missing := "01a15778-1911-7a28-8b2e-e08c2a25f388" for i := range observerLimit + 1 { - if _, err := f.gateway.Observe(t.Context(), missing, robot, true, fmt.Sprint(i)); !errors.Is(err, port.ErrRoomNotFound) { + if _, err := f.gateway.Observe(t.Context(), missing, robot, nil, true, fmt.Sprint(i)); !errors.Is(err, port.ErrRoomNotFound) { t.Fatalf("第 %d 次打开不存在的房间的事件流 = %v，期望 ErrRoomNotFound", i+1, err) } } diff --git a/internal/app/realtime/session.go b/internal/app/realtime/session.go index 1449a3d..40486aa 100644 --- a/internal/app/realtime/session.go +++ b/internal/app/realtime/session.go @@ -83,6 +83,8 @@ type Session struct { id room.SessionID room string link *roomLink + // 连接所用的票据签发给的成员会话的 token 哈希。 + memberSession string // 在发出 Leave 之前设置。重新加入的 Join 可能排在 Leave 之后，网关据此补发 Leave。 left atomic.Bool @@ -91,8 +93,8 @@ type Session struct { outstanding int } -func newSession(id room.SessionID, roomID string, m member.Member) *Session { - return &Session{sendQueue: newSendQueue(), id: id, room: roomID, member: m} +func newSession(id room.SessionID, roomID string, m member.Member, memberSession []byte) *Session { + return &Session{sendQueue: newSendQueue(), id: id, room: roomID, member: m, memberSession: string(memberSession)} } func (s *Session) ended() bool { diff --git a/internal/bootstrap/public.go b/internal/bootstrap/public.go index b1cfca2..51eab44 100644 --- a/internal/bootstrap/public.go +++ b/internal/bootstrap/public.go @@ -78,7 +78,7 @@ func newPublic(cfg serveConfig, pool *pgxpool.Pool, rdb *goredis.Client, instanc rooms := lobby.New(store, leases) services := api.Services{ Identity: identity.New(store, store, store, gateway), - Accounts: identity.NewAccounts(store, store, redis.NewAccountQuota(rdb)), + Accounts: identity.NewAccounts(store, store, redis.NewAccountQuota(rdb), gateway, base.With("component", "accounts")), Lobby: rooms, Admin: admin.New(cfg.adminToken, rooms, store, store, store, src.neteaseLogin, src.qqLogin, src.bilibiliLogin, store, gateway), Gateway: gateway, diff --git a/internal/bootstrap/testdata/http/events.txtar b/internal/bootstrap/testdata/http/events.txtar index b9c60d1..d4c61da 100644 --- a/internal/bootstrap/testdata/http/events.txtar +++ b/internal/bootstrap/testdata/http/events.txtar @@ -107,7 +107,7 @@ Location: /api/admin/rooms/$hall/api-key {"key": "$resetKey"} -- event bot -- event: close -data: {"code": 4401, "message": "凭证已失效：房间设置或修改了密码，或者 API Key 已重置"} +data: {"code": 4401, "message": "凭证已失效：房间设置或修改了密码，API Key 已重置，或者账号在这台设备上退出了登录、密码被修改或重置"} -- request -- POST /api/room/messages diff --git a/internal/bootstrap/testdata/ws/account-sessions.txtar b/internal/bootstrap/testdata/ws/account-sessions.txtar new file mode 100644 index 0000000..c54f5b7 --- /dev/null +++ b/internal/bootstrap/testdata/ws/account-sessions.txtar @@ -0,0 +1,142 @@ +票据属于签发它的会话：账号在两台设备上各自进入同一个房间，一台设备修改密码之后，另一台设备的 WebSocket 以 4401 断开， +用它的票据打开的事件流收到 4401 的 close；这台设备的旧票据再也不能用来进房或调用房间接口。发出请求的设备不受影响。 + +-- request -- +POST /api/members + +{} +-- response -- +201 +{"memberToken": "$alice", "member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}} + +-- request -- +POST /api/account +Authorization: Bearer $alice + +{"username": "alice", "password": "correct horse"} +-- response -- +201 +{"account": {"username": "alice"}} + +-- request -- +POST /api/members + +{} +-- response -- +201 +{"memberToken": "$guest", "member": {"id": "$guestId", "name": "$guestName", "avatarHash": ""}} + +-- request -- +POST /api/sessions +Authorization: Bearer $guest + +{"username": "alice", "password": "correct horse"} +-- response -- +201 +{"memberToken": "$alice2", "member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, "account": {"username": "alice"}} + +-- request -- +POST /api/rooms + +{"name": "临时", "description": ""} +-- response -- +201 +Location: /api/rooms/$room +{"id": "$room"} + +-- request -- +POST /api/rooms/$room/tickets +Authorization: Bearer $alice + +{} +-- response -- +201 +{"ticket": "$ticketA"} + +-- request -- +POST /api/rooms/$room/tickets +Authorization: Bearer $alice2 + +{} +-- response -- +201 +{"ticket": "$ticketB"} + +-- connect a -- +-- send a -- +{"kind": "request", "type": "hello", "id": "h1", "data": {"roomId": "$room", "ticket": "$ticketA", "protocolVersion": "$protocolVersion"}} +-- receive a -- +{"kind": "response", "type": "hello", "id": "h1", "data": { + "member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, + "serverTimeMs": 1, + "room": {"term": 1, "seq": 1, "room": {"id": "$room", "name": "临时", "description": "", "kind": "ephemeral"}, + "playMode": "sequential", "nowPlaying": null, "pending": null, "items": [], + "members": {"sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 1}}}} + +-- connect b -- +-- send b -- +{"kind": "request", "type": "hello", "id": "h1", "data": {"roomId": "$room", "ticket": "$ticketB", "protocolVersion": "$protocolVersion"}} +-- receive b -- +{"kind": "response", "type": "hello", "id": "h1", "data": { + "member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, + "serverTimeMs": 1, + "room": {"term": 1, "seq": 2, "room": {"id": "$room", "name": "临时", "description": "", "kind": "ephemeral"}, + "playMode": "sequential", "nowPlaying": null, "pending": null, "items": [], + "members": {"sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 2}}}} +-- receive a -- +{"kind": "event", "type": "members.updated", "term": 1, "seq": 2, "data": { + "sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 2}} + +-- stream b -- +Authorization: Bearer $ticketB +-- event b -- +event: snapshot +data: {"room": {"term": 1, "seq": 2, "room": {"id": "$room", "name": "临时", "description": "", "kind": "ephemeral"}, + "playMode": "sequential", "nowPlaying": null, "pending": null, "items": [], + "members": {"sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 2}}} + +-- request -- +PUT /api/account/password +Authorization: Bearer $alice + +{"currentPassword": "correct horse", "newPassword": "battery staple"} +-- response -- +204 +-- closed b -- +4401 +-- event b -- +event: close +data: {"code": 4401, "message": "凭证已失效：房间设置或修改了密码，API Key 已重置，或者账号在这台设备上退出了登录、密码被修改或重置"} +-- receive a -- +{"kind": "event", "type": "members.updated", "term": 1, "seq": 3, "data": { + "sessions": [{"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}], "onlineCount": 1}} + +-- send a -- +{"kind": "request", "type": "ping", "id": "p1", "data": {"clientTimeMs": 1234}} +-- receive a -- +{"kind": "response", "type": "ping", "id": "p1", "data": {"clientTimeMs": 1234, "serverTimeMs": 1}} + +-- connect again -- +-- send again -- +{"kind": "request", "type": "hello", "id": "h1", "data": {"roomId": "$room", "ticket": "$ticketB", "protocolVersion": "$protocolVersion"}} +-- receive again -- +{"kind": "response", "type": "hello", "id": "h1", "data": {}, "error": {"code": "UNAUTHORIZED", "message": "票据无效"}} +-- closed again -- +4401 + +-- request -- +GET /api/room +Authorization: Bearer $ticketB +-- response -- +401 +{"error": {"code": "UNAUTHORIZED", "message": "票据无效"}} + +-- request -- +POST /api/room/messages +Authorization: Bearer $ticketA + +{"content": "还在"} +-- response -- +204 +-- receive a -- +{"kind": "event", "type": "chat.message", "term": 1, "seq": 4, "data": {"member": {"id": "$aliceId", "name": "$aliceName", "avatarHash": ""}, "content": "还在", "sentAtMs": 1}} diff --git a/protocol/idl.json b/protocol/idl.json index 026e713..c2dee81 100644 --- a/protocol/idl.json +++ b/protocol/idl.json @@ -367,7 +367,7 @@ {"name": "internal", "code": 1011, "description": "握手时发生内部错误"}, {"name": "serviceRestart", "code": 1012, "description": "服务端实例退出，或实例与 Redis 的订阅连接中断"}, {"name": "unavailable", "code": 1013, "description": "握手时服务暂时不可用"}, - {"name": "unauthorized", "code": 4401, "description": "票据无效、用 API Key 握手，或房间设置、修改了密码"}, + {"name": "unauthorized", "code": 4401, "description": "票据无效、用 API Key 握手、房间设置或修改了密码，或者票据所属的成员会话已被删除（退出登录、修改或重置账号密码）"}, {"name": "notFound", "code": 4404, "description": "房间不存在，或房间已关闭、被删除"}, {"name": "handshakeTimeout", "code": 4408, "description": "握手超时"}, {"name": "clientOutdated", "code": 4426, "description": "前端与服务端的协议版本不一致"}, diff --git a/protocol/openapi.gen.json b/protocol/openapi.gen.json index 353fab1..855de9e 100644 --- a/protocol/openapi.gen.json +++ b/protocol/openapi.gen.json @@ -1500,7 +1500,7 @@ }, "info": { "title": "Alisten", - "version": "eae091a3e1daf18c" + "version": "88c848e7b53dc1b8" }, "openapi": "3.1.0", "paths": { diff --git a/protocol/types.gen.go b/protocol/types.gen.go index f66ecdd..4e8bc8e 100644 --- a/protocol/types.gen.go +++ b/protocol/types.gen.go @@ -4,7 +4,7 @@ package protocol import "encoding/json" -const Version = "eae091a3e1daf18c" +const Version = "88c848e7b53dc1b8" type APIKey struct { Key string `json:"key"` diff --git a/web/src/core/api/client.test.ts b/web/src/core/api/client.test.ts index b5942d9..f4e027b 100644 --- a/web/src/core/api/client.test.ts +++ b/web/src/core/api/client.test.ts @@ -11,6 +11,14 @@ type Handler = (token: string | null) => Response | Promise class FakeStorage { readonly #items = new Map() + get length(): number { + return this.#items.size + } + + key(index: number): string | null { + return [...this.#items.keys()][index] ?? null + } + getItem(key: string): string | null { return this.#items.get(key) ?? null } @@ -172,6 +180,18 @@ describe('token 失效时更换身份', () => { expect(events).toEqual([{ kind: 'rejected' }, { kind: 'renewed', member: issued }]) }) + it('票据属于被拒绝的身份，一并删除', async () => { + local.setItem(memberTokenKey, 'expired') + local.setItem('alisten.ticket.r1', 't1') + local.setItem('alisten.ticket.r2', 't2') + routes.set('PATCH /api/me', unauthorized) + const { api, IdentityRenewedError } = await load() + await expect(api.updateMe({ name: 'alice' })).rejects.toBeInstanceOf(IdentityRenewedError) + + expect(local.getItem('alisten.ticket.r1')).toBeNull() + expect(local.getItem('alisten.ticket.r2')).toBeNull() + }) + it('同时被拒绝的请求只更换一次身份', async () => { local.setItem(memberTokenKey, 'expired') routes.set('PATCH /api/me', unauthorized) @@ -202,6 +222,7 @@ describe('token 失效时更换身份', () => { it('其他标签页已经换了 token 时不删除新的 token', async () => { local.setItem(memberTokenKey, 'expired') + local.setItem('alisten.ticket.r1', 't1') routes.set('PATCH /api/me', () => { local.setItem(memberTokenKey, 'good') return unauthorized() @@ -210,6 +231,7 @@ describe('token 失效时更换身份', () => { await expect(api.updateMe({ name: 'a' })).rejects.toThrow('刚才的操作没有生效') expect(local.getItem(memberTokenKey)).toBe('good') + expect(local.getItem('alisten.ticket.r1')).toBe('t1') expect(callsTo('POST /api/members')).toHaveLength(0) expect(events).toEqual([{ kind: 'rejected' }, { kind: 'renewed', member }]) }) @@ -272,6 +294,7 @@ describe('token 失效时更换身份', () => { it('修改密码时当前密码错误返回 INVALID_CREDENTIALS，不更换身份', async () => { local.setItem(memberTokenKey, 'good') + local.setItem('alisten.ticket.r1', 't1') routes.set('PUT /api/account/password', () => errorResponse(401, 'INVALID_CREDENTIALS', '当前密码错误')) const { api, events, IdentityRenewedError } = await load() const error: unknown = await api.changePassword({ currentPassword: 'wrong horse', newPassword: 'correct horse' }).catch((reason: unknown) => reason) @@ -279,6 +302,7 @@ describe('token 失效时更换身份', () => { expect(error).toMatchObject({ code: 'INVALID_CREDENTIALS', status: 401 }) expect(events).toEqual([]) expect(local.getItem(memberTokenKey)).toBe('good') + expect(local.getItem('alisten.ticket.r1')).toBe('t1') }) it('账号密码错误不更换身份', async () => { diff --git a/web/src/core/api/client.ts b/web/src/core/api/client.ts index 35ad237..38dad83 100644 --- a/web/src/core/api/client.ts +++ b/web/src/core/api/client.ts @@ -1,7 +1,7 @@ import type { Auth, CreateTicketRequest, CreateTicketResponse, ErrorResponse, HttpRequest } from './http.gen' import type { ErrorCode, Member } from '@/core/realtime' import { Emitter } from '@/shared/lib/emitter' -import { deleteMemberToken, loadAdminToken, loadMemberToken, memberTokenKey, saveMemberToken } from './credentials' +import { deleteAllTickets, deleteMemberToken, loadAdminToken, loadMemberToken, memberTokenKey, saveMemberToken } from './credentials' import { httpClient } from './http.gen' export class ApiError extends Error { @@ -163,9 +163,11 @@ function renewIdentity(rejected: string): Promise { // 更换过程中用 unrenewed 读取成员：新 token 也被拒绝时不再更换，避免循环。 async function replaceIdentity(rejected: string): Promise { - // 其他标签页可能已经更换了身份，只删除被拒绝的那个 token。 - if (loadMemberToken() === rejected) + // 其他标签页可能已经更换了身份，只删除被拒绝的那个 token。票据属于被拒绝的身份，一并删除。 + if (loadMemberToken() === rejected) { deleteMemberToken() + deleteAllTickets() + } identityEvents.emit({ kind: 'rejected' }) let member: Member try { diff --git a/web/src/core/realtime/protocol.gen.ts b/web/src/core/realtime/protocol.gen.ts index d93e606..8b9e799 100644 --- a/web/src/core/realtime/protocol.gen.ts +++ b/web/src/core/realtime/protocol.gen.ts @@ -1,6 +1,6 @@ // Code generated by protocolgen from protocol/idl.json; DO NOT EDIT. -export const PROTOCOL_VERSION = 'eae091a3e1daf18c' +export const PROTOCOL_VERSION = '88c848e7b53dc1b8' export interface APIKey { key: string