From 05409d0914bdbb35868efb57097ad0c597a45102 Mon Sep 17 00:00:00 2001 From: grendervill Date: Tue, 22 Sep 2026 21:58:23 +0300 Subject: [PATCH] =?UTF-8?q?fix(gateway):=20=D1=81=D0=BC=D0=B5=D0=BD=D0=B0?= =?UTF-8?q?=20=D0=BF=D0=B0=D1=80=D0=BE=D0=BB=D1=8F=20=D0=BD=D0=B5=20=D0=B7?= =?UTF-8?q?=D0=B0=D0=BA=D1=80=D1=8B=D0=B2=D0=B0=D0=B5=D1=82=20=D1=81=D0=B5?= =?UTF-8?q?=D1=81=D1=81=D0=B8=D1=8E=20=D1=82=D0=B5=D0=BA=D1=83=D1=89=D0=B5?= =?UTF-8?q?=D0=B3=D0=BE=20=D1=83=D1=81=D1=82=D1=80=D0=BE=D0=B9=D1=81=D1=82?= =?UTF-8?q?=D0=B2=D0=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ChangePassword отзывает остальные сессии, но обработчик закрывал соединения всех устройств пользователя — включая то, с которого пароль сменили: страница теряла шлюз в момент успешного ответа, и подтверждение «Пароль изменён» не показывалось (пункт 12 матрицы 11.6, меняли пароль через интерфейс). Добавлен InvalidateUserExcept: текущее соединение опознаётся по хэшу токена сессии, который теперь хранит и буфер RESUME. Для logout-all, бана и админского сброса поведение прежнее — закрываются все соединения. Тест: TestInvalidateUserExceptKeepsCurrentSession (второе устройство получает INVALID_SESSION, текущее продолжает отвечать на heartbeat). --- internal/gateway/gateway.go | 21 ++++++++++++-- internal/gateway/gateway_test.go | 50 ++++++++++++++++++++++++++++++++ internal/gateway/session.go | 21 ++++++++------ internal/server/api_users.go | 6 ++-- 4 files changed, 85 insertions(+), 13 deletions(-) diff --git a/internal/gateway/gateway.go b/internal/gateway/gateway.go index b8f8535..4ff7253 100644 --- a/internal/gateway/gateway.go +++ b/internal/gateway/gateway.go @@ -396,12 +396,29 @@ func (s *Service) SendToUser(userID uint64, event string, payload any) { // (logout-all, смена пароля, действия администратора) — иначе второе // устройство остаётся «в приложении» до следующего запроса (AGENT.md 11.6). func (s *Service) InvalidateUser(userID uint64, reason string) { + s.invalidateUser(userID, "", reason) +} + +// InvalidateUserExcept закрывает соединения пользователя, кроме текущего: +// смена пароля отзывает остальные сессии, но устройство, с которого её +// сменили, продолжает работать (AGENT.md 7.1, 11.6). +func (s *Service) InvalidateUserExcept(userID uint64, keepTokenHash, reason string) { + s.invalidateUser(userID, keepTokenHash, reason) +} + +func (s *Service) invalidateUser(userID uint64, keepTokenHash, reason string) { s.mu.Lock() targets := make([]*clientSession, 0, 2) for _, session := range s.sessions { - if session.userID == userID { - targets = append(targets, session) + if session.userID != userID { + continue } + // Соединение опознаём по хэшу токена сессии: у каждой сессии свой буфер + // RESUME, он же хранит владельца (AGENT.md 8.3). + if keepTokenHash != "" && session.buffer != nil && session.buffer.tokenHash == keepTokenHash { + continue + } + targets = append(targets, session) } s.mu.Unlock() for _, session := range targets { diff --git a/internal/gateway/gateway_test.go b/internal/gateway/gateway_test.go index 4913687..2b49091 100644 --- a/internal/gateway/gateway_test.go +++ b/internal/gateway/gateway_test.go @@ -424,6 +424,56 @@ func TestDispatchReachesConnectedClient(t *testing.T) { } } +// TestInvalidateUserExceptKeepsCurrentSession: смена пароля отзывает остальные +// сессии, но соединение устройства, с которого её сменили, остаётся живым, +// иначе пользователь терял бы интерфейс в момент успешной смены (AGENT.md 7.1). +func TestInvalidateUserExceptKeepsCurrentSession(t *testing.T) { + f := newFixture(t) + ctx := context.Background() + token := registerUser(t, f, "except_user", "except@example.com") + // Вторая сессия того же пользователя: вход тем же паролем с другого + // «устройства» (в тесте — второй токен). + user, currentSession, err := f.auth.ResolveSession(ctx, token) + if err != nil { + t.Fatalf("ResolveSession: %v", err) + } + email, err := f.auth.Email(ctx, user.ID) + if err != nil { + t.Fatalf("Email: %v", err) + } + _, otherToken, otherSession, err := f.auth.Login(ctx, auth.LoginInput{ + Email: email, Password: "correct-horse-battery", + }) + if err != nil { + t.Fatalf("Login: %v", err) + } + + currentConn, _ := f.dial(t) + send(t, currentConn, gateway.OpIdentify, map[string]any{"token": token}) + if ready := readEnvelope(t, currentConn); ready.T != "READY" { + t.Fatalf("expected READY, got %q", ready.T) + } + otherConn, _ := f.dial(t) + send(t, otherConn, gateway.OpIdentify, map[string]any{"token": otherToken}) + if ready := readEnvelope(t, otherConn); ready.T != "READY" { + t.Fatalf("expected READY for the second device, got %q", ready.T) + } + + f.service.InvalidateUserExcept(user.ID, currentSession.TokenHash, "password_changed") + + // Второе устройство получает INVALID_SESSION... + invalid := readEnvelope(t, otherConn) + if invalid.Op != gateway.OpInvalidSess { + t.Fatalf("other device op = %d, want INVALID_SESSION (%d)", invalid.Op, gateway.OpInvalidSess) + } + // ...а текущее продолжает работать: heartbeat отвечает. + send(t, currentConn, gateway.OpHeartbeat, nil) + if ack := readEnvelope(t, currentConn); ack.Op != gateway.OpHeartbeatAck { + t.Fatalf("current device op = %d, want HEARTBEAT_ACK", ack.Op) + } + _ = otherSession +} + func TestResumeReplaysMissedEvents(t *testing.T) { f := newFixture(t) token := registerUser(t, f, "resume_user", "resume@example.com") diff --git a/internal/gateway/session.go b/internal/gateway/session.go index 6066f88..f33957e 100644 --- a/internal/gateway/session.go +++ b/internal/gateway/session.go @@ -325,7 +325,7 @@ func (s *Service) register(session *clientSession, tokenHash string) { s.sessions[session.id] = session buffer, ok := s.buffers[tokenHash] if !ok { - buffer = newResumeBuffer(ResumeBufferSize) + buffer = newResumeBuffer(ResumeBufferSize, tokenHash) s.buffers[tokenHash] = buffer } // Владелец буфера нужен, чтобы доставлять адресные события (SendToUser) @@ -347,7 +347,7 @@ func (s *Service) bufferFor(tokenHash string) *resumeBuffer { if buffer, ok := s.buffers[tokenHash]; ok { return buffer } - buffer := newResumeBuffer(ResumeBufferSize) + buffer := newResumeBuffer(ResumeBufferSize, tokenHash) s.buffers[tokenHash] = buffer return buffer } @@ -405,11 +405,14 @@ func newSessionID() string { // resumeBuffer хранит последние события для RESUME (AGENT.md 8.3). type resumeBuffer struct { - mu sync.Mutex - owner uint64 - limit int - items []bufferedEvent - updated time.Time + mu sync.Mutex + // tokenHash — сессия пользователя, которой принадлежит буфер: по нему + // адресные отзывы отличают текущее соединение от остальных. + tokenHash string + owner uint64 + limit int + items []bufferedEvent + updated time.Time } type bufferedEvent struct { @@ -417,8 +420,8 @@ type bufferedEvent struct { payload []byte } -func newResumeBuffer(limit int) *resumeBuffer { - return &resumeBuffer{limit: limit, updated: time.Now()} +func newResumeBuffer(limit int, tokenHash string) *resumeBuffer { + return &resumeBuffer{limit: limit, tokenHash: tokenHash, updated: time.Now()} } func (b *resumeBuffer) append(seq int64, payload []byte) { diff --git a/internal/server/api_users.go b/internal/server/api_users.go index 917e6fa..797e1fc 100644 --- a/internal/server/api_users.go +++ b/internal/server/api_users.go @@ -270,9 +270,11 @@ func (s *Server) registerUserRoutes(api huma.API) { if err := s.auth.ChangePassword(ctx, user.ID, session.ID, input.Body.CurrentPassword, input.Body.NewPassword); err != nil { return nil, humaError(err) } - // Смена пароля отзывает остальные сессии: соединения закрываем сразу. + // Смена пароля отзывает остальные сессии: их соединения закрываем сразу, + // а устройство, с которого пароль сменили, продолжает работать + // (AGENT.md 7.1, 11.6). if s.gateway != nil { - s.gateway.InvalidateUser(user.ID, "password_changed") + s.gateway.InvalidateUserExcept(user.ID, session.TokenHash, "password_changed") } return newOKOutput(), nil })