From bbf4bb42b2cd85c07edcb3bf736a83298fa7b62c Mon Sep 17 00:00:00 2001 From: ULookup Date: Wed, 22 Jul 2026 04:52:50 +0000 Subject: [PATCH] fix(push): reject revoked websocket tokens --- .../references/core-flows.md | 8 ++-- common/auth/jwt_store.hpp | 30 +++++++++---- common/infra/metrics.hpp | 3 ++ push/source/push_server.h | 26 +++++++++++- tests/func/ws_notify_test.go | 42 +++++++++++++++++++ tests/pkg/client/ws.go | 34 ++++++++++++--- 6 files changed, 122 insertions(+), 21 deletions(-) diff --git a/.agents/skills/chatnow-orienting/references/core-flows.md b/.agents/skills/chatnow-orienting/references/core-flows.md index 13a0e62..4fe8f1c 100644 --- a/.agents/skills/chatnow-orienting/references/core-flows.md +++ b/.agents/skills/chatnow-orienting/references/core-flows.md @@ -42,10 +42,10 @@ These are current-state flows for the `3.0-dev` line. Re-verify affected symbols - Entry/contracts: `proto/identity/identity_service.proto`; `identity/source/identity_server.h`; `common/auth/jwt_codec.hpp`; `common/auth/jwt_store.hpp`; `gateway/source/gateway_auth.hpp`; `push/source/push_server.h`. - Stores: Identity uses MySQL for users/devices and Redis for active refresh tokens, rotation/reuse detection, and revocation state. -- Trust: Gateway validates Bearer access tokens and revocation before deriving metadata. Push verifies WS `CLIENT_AUTH` and binds claim identity to the connection. Downstream handlers use `common/auth/auth_context.hpp`; service-to-service forwarding uses `common/auth/forward_auth.hpp` where required. -- Sync/retry: Login and refresh are synchronous; refresh rotation detects reuse. Cache/store failure behavior must be inspected before changing fail-open/fail-closed semantics. -- Tests: `tests/bvt/auth_test.go`, `tests/func/identity_test.go`, `tests/func/auth_middleware_test.go`, `tests/func/security_test.go`, `tests/func/scenarios_test.go`. -- Invariants: only Identity issues/refreshes tokens; access and refresh token purposes remain distinct; downstream identity comes from verified claims and forwarded metadata, not request bodies. +- Trust: Gateway validates Bearer access tokens and revocation before deriving metadata. Push verifies WS `CLIENT_AUTH`, queries revocation once at admission, and binds claim identity only after a `kNotRevoked` result. Revoked tokens and unavailable revocation state are rejected before connection, route, presence, or resend side effects. Downstream handlers use `common/auth/auth_context.hpp`; service-to-service forwarding uses `common/auth/forward_auth.hpp` where required. +- Sync/retry: Login and refresh are synchronous; refresh rotation detects reuse. Push performs no revocation lookup per message or heartbeat. Its admission boundary fails closed on Redis errors, while the legacy `JwtStore::is_revoked` bool API retains fail-open compatibility for unchanged callers. +- Tests: `tests/bvt/auth_test.go`, `tests/func/identity_test.go`, `tests/func/auth_middleware_test.go`, `tests/func/security_test.go`, `tests/func/scenarios_test.go`, and `tests/func/ws_notify_test.go` (`FN-WS-09`). +- Invariants: only Identity issues/refreshes tokens; access and refresh token purposes remain distinct; downstream identity comes from verified claims and forwarded metadata, not request bodies; Push admission must resolve revocation before publishing any authenticated-session side effect. ## Media upload and download diff --git a/common/auth/jwt_store.hpp b/common/auth/jwt_store.hpp index 35098c4..2943287 100644 --- a/common/auth/jwt_store.hpp +++ b/common/auth/jwt_store.hpp @@ -7,9 +7,11 @@ * im:jwt:rt:{user_id}:{device_id} -> refresh_jti TTL = refresh 寿命 * im:jwt:rt_chain:{old_jti} -> "rotated" TTL = 24h * - * 失败模式：底层 redis 抛异常时函数自身吞掉 + LOG_ERROR + 返回保守值 - * - is_revoked 失败 → false（不阻断业务，避免雪崩） - * - 写失败 → 仅日志，调用方按业务决定 + * Failure modes: + * - query_revocation_status exposes Redis failures as kUnavailable so + * security boundaries can fail closed. + * - is_revoked preserves its legacy fail-open bool contract. + * - write failures are logged and left to the caller's policy. * * 重放检测：rotate_refresh_or_detect_reuse 用 SET NX 原子保护链节点。 */ @@ -34,8 +36,11 @@ class JwtStore { using ptr = std::shared_ptr; explicit JwtStore(chatnow::RedisClient::ptr c) : _c(std::move(c)) {} + enum class RevocationStatus { kNotRevoked, kRevoked, kUnavailable }; + void revoke(const std::string& jti, int ttl_sec); bool is_revoked(const std::string& jti); + RevocationStatus query_revocation_status(const std::string& jti); void put_active_refresh(const std::string& user_id, const std::string& device_id, @@ -74,18 +79,26 @@ inline void JwtStore::revoke(const std::string& jti, int ttl_sec) { _c->set(std::string(jwt_key::kRevokedPrefix) + jti, "1", std::chrono::seconds(ttl_sec)); } catch (const std::exception& e) { - LOG_ERROR("JwtStore.revoke 失败 jti={}: {}", jti, e.what()); + LOG_ERROR("JwtStore revoke failed: {}", e.what()); } } inline bool JwtStore::is_revoked(const std::string& jti) { if (jti.empty()) return false; + // Preserve existing callers' fail-open bool semantics. + return query_revocation_status(jti) == RevocationStatus::kRevoked; +} + +inline JwtStore::RevocationStatus JwtStore::query_revocation_status( + const std::string& jti) { + if (jti.empty()) return RevocationStatus::kUnavailable; try { auto v = _c->get(std::string(jwt_key::kRevokedPrefix) + jti); - return v.has_value(); + return v.has_value() ? RevocationStatus::kRevoked + : RevocationStatus::kNotRevoked; } catch (const std::exception& e) { - LOG_ERROR("JwtStore.is_revoked 失败 jti={}: {}", jti, e.what()); - return false; + LOG_ERROR("JwtStore revocation query failed: {}", e.what()); + return RevocationStatus::kUnavailable; } } @@ -141,8 +154,7 @@ inline JwtStore::RotateResult JwtStore::rotate_refresh_or_detect_reuse( return RotateResult::kReuseDetected; } } catch (const std::exception& e) { - LOG_ERROR("JwtStore.rotate chain SET 失败 old_jti={}: {}", - old_refresh_jti, e.what()); + LOG_ERROR("JwtStore refresh rotation failed: {}", e.what()); // 链路写失败：保守按"未被重放"放行，下次会再尝试 } put_active_refresh(user_id, device_id, new_refresh_jti, new_refresh_ttl_sec); diff --git a/common/infra/metrics.hpp b/common/infra/metrics.hpp index 7a56d8b..d0d4823 100644 --- a/common/infra/metrics.hpp +++ b/common/infra/metrics.hpp @@ -32,6 +32,9 @@ inline bvar::Adder g_rate_limit_local_fallback_total("rate_limit_local_fal inline bvar::Adder g_rate_limit_local_rejected_total("rate_limit_local_rejected_total"); inline bvar::Adder g_push_unacked_persist_failure_total("push_unacked_persist_failure_total"); inline bvar::Adder g_push_message_requeue_total("push_message_requeue_total"); +inline bvar::Adder g_push_auth_revoked_total("push_auth_revoked_total"); +inline bvar::Adder g_push_auth_revocation_unavailable_total( + "push_auth_revocation_unavailable_total"); template inline typename ::chatnow::LocalCache::MetricsSink local_cache_metrics_sink() { diff --git a/push/source/push_server.h b/push/source/push_server.h index 7c46c12..e00d257 100644 --- a/push/source/push_server.h +++ b/push/source/push_server.h @@ -13,6 +13,7 @@ #include "dao/data_redis.hpp" #include "auth/auth_context.hpp" #include "auth/forward_auth.hpp" +#include "auth/jwt_store.hpp" #include "common/auth/metadata.pb.h" #include "auth/auth_config_loader.hpp" #include "error/error_codes.hpp" @@ -57,6 +58,7 @@ class PushServiceImpl : public PushService public: PushServiceImpl(const Connection::ptr &connections, const std::shared_ptr &jwt_codec, + const chatnow::auth::JwtStore::ptr &jwt_store, const RedisClient::ptr &redis, const OnlineRoute::ptr &online_route, const UnackedPush::ptr &unacked, @@ -70,6 +72,7 @@ class PushServiceImpl : public PushService std::chrono::seconds route_l1_ttl = std::chrono::seconds(2)) : _connections(connections), _jwt_codec(jwt_codec), + _jwt_store(jwt_store), _redis(redis), _online_route(online_route), _unacked(unacked), @@ -527,6 +530,20 @@ class PushServiceImpl : public PushService std::string did = claims.did; std::string jti = claims.jti; + const auto revocation_status = _jwt_store->query_revocation_status(jti); + if (revocation_status != chatnow::auth::JwtStore::RevocationStatus::kNotRevoked) { + if (revocation_status == chatnow::auth::JwtStore::RevocationStatus::kRevoked) { + metrics::g_push_auth_revoked_total << 1; + LOG_WARN("WS authentication rejected: token revoked"); + } else { + metrics::g_push_auth_revocation_unavailable_total << 1; + LOG_WARN("WS authentication rejected: revocation state unavailable"); + } + try { conn->close(websocketpp::close::status::policy_violation, + "auth failed"); } catch (std::exception &e) { LOG_WARN("WS close failed: {}", e.what()); } + return; + } + _connections->insert(conn, uid, did, jti); if (_online_route) _online_route->bind(uid, did, _instance_id); @@ -975,6 +992,7 @@ class PushServiceImpl : public PushService Connection::ptr _connections; std::shared_ptr _jwt_codec; + chatnow::auth::JwtStore::ptr _jwt_store; RedisClient::ptr _redis; OnlineRoute::ptr _online_route; UnackedPush::ptr _unacked; @@ -1079,6 +1097,7 @@ class PushServerBuilder auto redis = RedisClientFactory::create(host, port, db, keep_alive, pool_size); _redis_client = std::make_shared(redis); } + _jwt_store = std::make_shared(_redis_client); _online_route = std::make_shared(_redis_client); _unacked = std::make_shared(_redis_client); _cross_outbox = std::make_shared(_redis_client); @@ -1229,6 +1248,7 @@ class PushServerBuilder void make_rpc_object(uint16_t port, uint32_t timeout, uint8_t num_threads, uint16_t ws_port) { if (!_redis_client) { LOG_ERROR("Push: Redis 未初始化"); abort(); } + if (!_jwt_store) { LOG_ERROR("Push: JWT store 未初始化"); abort(); } if (!_mm_channels) { LOG_ERROR("Push: 信道管理未初始化"); abort(); } if (port == ws_port) { LOG_WARN("Push: rpc_port and ws_port are both {}, may conflict", port); @@ -1236,7 +1256,8 @@ class PushServerBuilder _connections = std::make_shared(); _rpc_server = std::make_shared(); _push_service = new PushServiceImpl( - _connections, _jwt_codec, _redis_client, _online_route, _unacked, _cross_outbox, + _connections, _jwt_codec, _jwt_store, _redis_client, _online_route, + _unacked, _cross_outbox, _instance_id, _message_service_name, _mm_channels, _cross_reaper_election, _local_route_cache, _inflight_registry, _route_l1_ttl); _push_service->set_resend_params(_resend_batch, _resend_max_age_sec); @@ -1296,7 +1317,7 @@ class PushServerBuilder std::move(_mq_client), std::move(_push_subscriber), _push_service, - std::move(_stale_reaper_thread), + &_stale_reaper_thread, _stale_reaper_running); } @@ -1304,6 +1325,7 @@ class PushServerBuilder std::string _redis_seeds; RedisClient::ptr _redis_client; std::shared_ptr _jwt_codec; + chatnow::auth::JwtStore::ptr _jwt_store; OnlineRoute::ptr _online_route; UnackedPush::ptr _unacked; CrossInstanceOutbox::ptr _cross_outbox; diff --git a/tests/func/ws_notify_test.go b/tests/func/ws_notify_test.go index 1113120..39f6c16 100644 --- a/tests/func/ws_notify_test.go +++ b/tests/func/ws_notify_test.go @@ -12,6 +12,7 @@ import ( "chatnow-tests/pkg/client" "chatnow-tests/pkg/fixture" + identity "chatnow-tests/proto/chatnow/identity" msg "chatnow-tests/proto/chatnow/message" presence "chatnow-tests/proto/chatnow/presence" push "chatnow-tests/proto/chatnow/push" @@ -237,3 +238,44 @@ func TestFN_WS_MQTracePropagation(t *testing.T) { require.NoError(t, err) assert.Equal(t, traceID, notify.GetTraceId()) } + +// FN-WS-09 | P0 | authentication | A revoked access token cannot establish a WebSocket session. +func TestFN_WS_RevokedTokenRejected(t *testing.T) { + revoked, _, _ := fixture.RegisterAndLogin(t, HTTP) + observer, _, _ := fixture.RegisterAndLogin(t, HTTP) + + logoutRsp := &identity.LogoutRsp{} + require.NoError(t, revoked.DoAuth("/service/identity/logout", &identity.LogoutReq{ + RequestId: client.NewRequestID(), + }, logoutRsp)) + require.True(t, logoutRsp.GetHeader().GetSuccess()) + + ws, err := client.NewWSClient( + revoked.Config(), revoked.AccessToken, revoked.UserID, revoked.DeviceID) + require.NoError(t, err) + t.Cleanup(func() { _ = ws.Close() }) + + closeCtx, cancelClose := context.WithTimeout(context.Background(), 3*time.Second) + defer cancelClose() + require.NoError(t, ws.WaitForClose(closeCtx), "Push must close a revoked WebSocket admission") + + presenceRsp := &presence.GetPresenceRsp{} + require.NoError(t, observer.DoAuth("/service/presence/get", &presence.GetPresenceReq{ + RequestId: client.NewRequestID(), + UserId: revoked.UserID, + }, presenceRsp)) + require.True(t, presenceRsp.GetHeader().GetSuccess()) + assert.Equal(t, presence.PresenceState_OFFLINE, + presenceRsp.GetPresence().GetAggregatedState(), + "rejected admission must not publish online presence") + + validWS, err := client.NewWSClient( + observer.Config(), observer.AccessToken, observer.UserID, observer.DeviceID) + require.NoError(t, err) + t.Cleanup(func() { _ = validWS.Close() }) + + validCtx, cancelValid := context.WithTimeout(context.Background(), time.Second) + defer cancelValid() + assert.ErrorIs(t, validWS.WaitForClose(validCtx), context.DeadlineExceeded, + "a valid access token must keep its WebSocket session open") +} diff --git a/tests/pkg/client/ws.go b/tests/pkg/client/ws.go index c1a9658..5ec2df7 100644 --- a/tests/pkg/client/ws.go +++ b/tests/pkg/client/ws.go @@ -19,10 +19,12 @@ type WSClient struct { userID string deviceID string - mu sync.Mutex - notifies []*push.NotifyMessage - notifyCh chan *push.NotifyMessage - closed bool + mu sync.Mutex + notifies []*push.NotifyMessage + notifyCh chan *push.NotifyMessage + closed bool + closedCh chan struct{} + closeOnce sync.Once } // NewWSClient 连接 gateway WS，发送 CLIENT_AUTH 鉴权帧，启动 readLoop。 @@ -39,6 +41,7 @@ func NewWSClient(cfg *Config, accessToken, userID, deviceID string) (*WSClient, userID: userID, deviceID: deviceID, notifyCh: make(chan *push.NotifyMessage, 100), + closedCh: make(chan struct{}), } // 发送 CLIENT_AUTH 鉴权帧 @@ -129,20 +132,39 @@ func (w *WSClient) WaitForNotifyCount(ctx context.Context, notifyType int32, n i // Close 关闭 WS 连接。 func (w *WSClient) Close() error { + if !w.markClosed() { + return nil + } + return w.conn.Close() +} + +// WaitForClose waits until the peer closes the WebSocket or the context expires. +func (w *WSClient) WaitForClose(ctx context.Context) error { + select { + case <-ctx.Done(): + return ctx.Err() + case <-w.closedCh: + return nil + } +} + +func (w *WSClient) markClosed() bool { w.mu.Lock() if w.closed { w.mu.Unlock() - return nil + return false } w.closed = true w.mu.Unlock() - return w.conn.Close() + w.closeOnce.Do(func() { close(w.closedCh) }) + return true } func (w *WSClient) readLoop() { for { _, data, err := w.conn.ReadMessage() if err != nil { + w.markClosed() return } notify := &push.NotifyMessage{}