ULookup · ULookup · Aug 7, 2026 · Jul 22, 2026
diff --git 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
@@ -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<JwtStore>;
     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
@@ -32,6 +32,9 @@ inline bvar::Adder<long> g_rate_limit_local_fallback_total("rate_limit_local_fal
 inline bvar::Adder<long> g_rate_limit_local_rejected_total("rate_limit_local_rejected_total");
 inline bvar::Adder<long> g_push_unacked_persist_failure_total("push_unacked_persist_failure_total");
 inline bvar::Adder<long> g_push_message_requeue_total("push_message_requeue_total");
+inline bvar::Adder<long> g_push_auth_revoked_total("push_auth_revoked_total");
+inline bvar::Adder<long> g_push_auth_revocation_unavailable_total(
+    "push_auth_revocation_unavailable_total");
 
 template <typename V>
 inline typename ::chatnow::LocalCache<V>::MetricsSink local_cache_metrics_sink() {

diff --git 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<chatnow::auth::JwtCodec> &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<chatnow::auth::JwtCodec> _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<RedisClient>(redis);
         }
+        _jwt_store    = std::make_shared<chatnow::auth::JwtStore>(_redis_client);
         _online_route = std::make_shared<OnlineRoute>(_redis_client);
         _unacked      = std::make_shared<UnackedPush>(_redis_client);
         _cross_outbox = std::make_shared<CrossInstanceOutbox>(_redis_client);
@@ -1229,14 +1248,16 @@ 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);
         }
         _connections = std::make_shared<Connection>();
         _rpc_server = std::make_shared<brpc::Server>();
         _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,14 +1317,15 @@ class PushServerBuilder
                                             std::move(_mq_client),
                                             std::move(_push_subscriber),
                                             _push_service,
-                                            std::move(_stale_reaper_thread),
+                                            &_stale_reaper_thread,
                                             _stale_reaper_running);
     }
 
 private:
     std::string _redis_seeds;
     RedisClient::ptr _redis_client;
     std::shared_ptr<chatnow::auth::JwtCodec> _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
@@ -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
@@ -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{}