Merge pull request #361 from DocGrid/fix/360

fix: 열린 STOMP 세션에 인증 상태 변경 반영

Author: Gimini-3

.env.example

@@ -93,3 +93,6 @@ MINIO_SECRET_KEY=minioadmin1234
 # REDIS_PASSWORD=
 # REDIS_TIMEOUT=1s
 # REDIS_CONNECT_TIMEOUT=1s
+# 열린 STOMP 세션의 token 만료·blacklist·역할 변경을 다시 확인하는 주기와 일괄 조회 크기입니다.
+# STOMP_SESSION_REVALIDATION_INTERVAL=5s
+# STOMP_SESSION_REVALIDATION_BATCH_SIZE=500

backend/src/main/java/com/opensource/docgrid/domain/auth/config/StompSessionSchedulingConfig.java

@@ -0,0 +1,15 @@
+package com.opensource.docgrid.domain.auth.config;
+
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+/**
+ * 열린 STOMP 세션 재검증 스케줄러를 다른 도메인의 Worker·Dashboard 설정과 독립적으로 활성화한다.
+ *
+ * <p>{@code @EnableScheduling}을 여러 설정에서 선언해도 Spring은 하나의 scheduling infrastructure로
+ * 처리한다. 인증 수명 검증이 다른 기능의 활성화 여부에 따라 조용히 멈추지 않도록 별도 경계를 둔다.
+ */
+@Configuration
+@EnableScheduling
+public class StompSessionSchedulingConfig {
+}

backend/src/main/java/com/opensource/docgrid/domain/auth/jwt/StompAuthChannelInterceptor.java

@@ -1,6 +1,9 @@
 package com.opensource.docgrid.domain.auth.jwt;
 
+import java.time.Instant;
+import java.util.Date;
 import java.util.List;
+import java.util.Set;
 
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
@@ -14,6 +17,9 @@
 import org.springframework.stereotype.Component;
 import org.springframework.util.StringUtils;
 
+import com.opensource.docgrid.domain.auth.websocket.StompSessionAuthorization;
+import com.opensource.docgrid.domain.auth.websocket.StompSessionRegistry;
+
 import io.jsonwebtoken.Claims;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -41,8 +47,8 @@
  *
  * <p>HTTP는 요청 하나로 끝나 매번 {@code SecurityContextHolder}를 새로 채우지만, WebSocket은 연결이
  * 오래 유지되는 세션이라 {@code accessor.setUser()}로 세션 자체에 Principal을 붙여 이후 프레임에서
- * 재사용한다. 이 검증은 신규 연결에만 적용되며 이미 열린 세션의 로그아웃·만료 반영은 별도 세션 수명
- * 주기에서 다룬다.
+ * 재사용한다. 연결 당시의 jti·만료 시각·role snapshot은 {@code StompSessionRegistry}에도 등록하고,
+ * 별도 재검증 작업이 열린 세션의 로그아웃·만료·역할 변경을 확인한다.
  *
  * <p>이때 Accessor는 반드시 {@link MessageHeaderAccessor#getAccessor}로 가져와야 한다.
  * {@code StompHeaderAccessor.wrap(message)}는 검증 전용 복사본이라 그 위에 {@code setUser()}를
@@ -61,6 +67,7 @@ public class StompAuthChannelInterceptor implements ChannelInterceptor {
     private final JwtProvider jwtProvider;
     private final TokenBlacklistService tokenBlacklistService;
     private final RoleAuthorityService roleAuthorityService;
+    private final StompSessionRegistry stompSessionRegistry;
 
     @Override
     public Message<?> preSend(Message<?> message, MessageChannel channel) {
@@ -81,9 +88,16 @@ public Message<?> preSend(Message<?> message, MessageChannel channel) {
                 throw invalidAuthentication();
             }
 
-            // 4. 거부되지 않은 토큰에 대해서만 현재 권한을 조회하고 세션 Principal을 만든다.
+            // 4. 수명 검증에 필요한 식별자가 없으면 추적할 수 없는 연결이므로 거부한다.
             String email = claims.getSubject();
             Long userId = claims.get("userId", Long.class);
+            Date expiration = claims.getExpiration();
+            String sessionId = accessor.getSessionId();
+            if (userId == null || expiration == null || !StringUtils.hasText(sessionId)) {
+                throw invalidAuthentication();
+            }
+
+            // 5. 거부되지 않은 토큰에 대해서만 현재 권한을 조회하고 세션 Principal을 만든다.
             List<String> roles = roleAuthorityService.getRoles(userId);
             List<SimpleGrantedAuthority> authorities = roles.stream()
                 .map(role -> new SimpleGrantedAuthority("ROLE_" + role))
@@ -92,6 +106,17 @@ public Message<?> preSend(Message<?> message, MessageChannel channel) {
             UsernamePasswordAuthenticationToken authentication =
                 new UsernamePasswordAuthenticationToken(email, null, authorities);
             authentication.setDetails(userId);
+
+            // 6. 물리 연결이 이미 추적 중일 때만 인증을 완료해 검사에서 빠지는 세션을 만들지 않는다.
+            StompSessionAuthorization authorization = new StompSessionAuthorization(
+                userId,
+                jti,
+                Instant.ofEpochMilli(expiration.getTime()),
+                Set.copyOf(roles)
+            );
+            if (!stompSessionRegistry.authenticate(sessionId, authorization)) {
+                throw invalidAuthentication();
+            }
             accessor.setUser(authentication);
         }
 

backend/src/main/java/com/opensource/docgrid/domain/auth/jwt/TokenBlacklistService.java

@@ -1,12 +1,23 @@
 package com.opensource.docgrid.domain.auth.jwt;
 
 import java.time.Duration;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
 
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import lombok.RequiredArgsConstructor;
 
+/**
+ * 로그아웃한 access token의 jti를 Redis에 보관하고 단건·일괄 폐기 여부 조회를 제공한다.
+ *
+ * <p>HTTP 인증은 단건 조회를 사용하고, 열린 STOMP 세션 재검증은 네트워크 왕복이 세션 수만큼
+ * 늘지 않도록 MGET 기반 일괄 조회를 사용한다. Redis 장애에 대한 fail-open·fail-closed 결정은 각
+ * 호출 경로가 자신의 가용성 요구에 맞게 담당한다.
+ */
 @Component
 @RequiredArgsConstructor
 public class TokenBlacklistService {
@@ -22,4 +33,34 @@ public void blacklist(String jti, long ttlSeconds) {
     public boolean isBlacklisted(String jti) {
         return Boolean.TRUE.equals(redisTemplate.hasKey(KEY_PREFIX + jti));
     }
+
+    /**
+     * 여러 STOMP 세션의 token 폐기 상태를 Redis MGET 한 번으로 확인한다.
+     *
+     * <p>응답 누락을 정상 token으로 오인하면 기존 연결이 계속 살아남으므로, Redis가 요청 key와 같은
+     * 개수의 결과를 주지 않으면 검증 실패로 처리한다. 호출자는 WebSocket fail-closed 정책에 따라
+     * 검사 대상 세션을 종료한다.
+     */
+    public Set<String> findBlacklistedJtis(Collection<String> jtis) {
+        List<String> distinctJtis = jtis.stream().distinct().toList();
+        if (distinctJtis.isEmpty()) {
+            return Set.of();
+        }
+
+        List<String> keys = distinctJtis.stream()
+            .map(jti -> KEY_PREFIX + jti)
+            .toList();
+        List<String> values = redisTemplate.opsForValue().multiGet(keys);
+        if (values == null || values.size() != keys.size()) {
+            throw new IllegalStateException("Redis blacklist 일괄 조회 결과가 완전하지 않습니다.");
+        }
+
+        Set<String> blacklisted = new HashSet<>();
+        for (int index = 0; index < distinctJtis.size(); index++) {
+            if (values.get(index) != null) {
+                blacklisted.add(distinctJtis.get(index));
+            }
+        }
+        return Set.copyOf(blacklisted);
+    }
 }

backend/src/main/java/com/opensource/docgrid/domain/auth/websocket/StompSessionAuthorization.java

@@ -0,0 +1,30 @@
+package com.opensource.docgrid.domain.auth.websocket;
+
+import java.time.Instant;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * STOMP CONNECT가 성공한 시점의 token 식별자·만료 시각·역할 snapshot을 세션 수명 검증에 전달한다.
+ *
+ * <p>JWT 원문과 이메일은 보관하지 않는다. 역할은 순서와 중복에 영향을 받지 않도록 불변 Set으로
+ * 정규화하며, 이후 검사에서 Redis blacklist와 DB의 현재 역할을 이 snapshot과 비교한다.
+ */
+public record StompSessionAuthorization(
+    Long userId,
+    String jti,
+    Instant expiresAt,
+    Set<String> roles
+) {
+
+    public StompSessionAuthorization {
+        Objects.requireNonNull(userId, "userId는 필수입니다.");
+        Objects.requireNonNull(jti, "jti는 필수입니다.");
+        Objects.requireNonNull(expiresAt, "expiresAt은 필수입니다.");
+        roles = Set.copyOf(Objects.requireNonNull(roles, "roles는 필수입니다."));
+    }
+
+    public boolean isExpired(Instant now) {
+        return !now.isBefore(expiresAt);
+    }
+}

backend/src/main/java/com/opensource/docgrid/domain/auth/websocket/StompSessionRegistry.java

@@ -0,0 +1,119 @@
+package com.opensource.docgrid.domain.auth.websocket;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.socket.CloseStatus;
+import org.springframework.web.socket.WebSocketSession;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * 현재 Backend 인스턴스가 소유한 물리 WebSocket 연결과 STOMP 인증 snapshot을 함께 관리한다.
+ *
+ * <p>물리 연결은 WebSocket decorator가 먼저 등록하고, CONNECT 인증이 끝난 뒤 같은 sessionId에
+ * 인증 snapshot을 결합한다. 주기 검사는 인증 완료 세션만 읽으며, 종료와 인증이 경합해도 하나의
+ * ConcurrentMap entry를 기준으로 정리해 닫힌 연결이 다시 등록되는 것을 막는다.
+ *
+ * <p>이 registry는 로컬 전송 자원만 관리한다. 여러 Backend 인스턴스는 각자 자신의 registry를
+ * 검사하고 공통 Redis·DB 상태를 읽으므로 분산 session registry나 lock이 필요하지 않다.
+ */
+@Component
+@Slf4j
+public class StompSessionRegistry {
+
+    private static final CloseStatus AUTHORIZATION_INVALID = CloseStatus.POLICY_VIOLATION;
+
+    private final ConcurrentMap<String, SessionState> sessions = new ConcurrentHashMap<>();
+
+    public void registerTransport(WebSocketSession session) {
+        SessionState previous = sessions.putIfAbsent(session.getId(), new SessionState(session));
+        if (previous != null) {
+            throw new IllegalStateException("이미 등록된 WebSocket sessionId입니다.");
+        }
+    }
+
+    public boolean authenticate(String sessionId, StompSessionAuthorization authorization) {
+        return sessions.computeIfPresent(sessionId, (ignored, state) -> {
+            state.authenticate(authorization);
+            return state;
+        }) != null;
+    }
+
+    public void remove(String sessionId) {
+        sessions.remove(sessionId);
+    }
+
+    public List<SessionSnapshot> authenticatedSessions() {
+        List<SessionSnapshot> snapshots = new ArrayList<>();
+        sessions.forEach((sessionId, state) -> {
+            if (!state.session().isOpen()) {
+                sessions.remove(sessionId, state);
+                return;
+            }
+            StompSessionAuthorization authorization = state.authorization();
+            if (authorization != null) {
+                snapshots.add(new SessionSnapshot(sessionId, authorization));
+            }
+        });
+        return List.copyOf(snapshots);
+    }
+
+    public int authenticatedSessionCount() {
+        return (int) sessions.values().stream()
+            .filter(state -> state.session().isOpen() && state.authorization() != null)
+            .count();
+    }
+
+    public boolean close(String sessionId) {
+        SessionState state = sessions.get(sessionId);
+        if (state == null) {
+            return false;
+        }
+        if (!state.session().isOpen()) {
+            sessions.remove(sessionId, state);
+            return false;
+        }
+
+        try {
+            state.session().close(AUTHORIZATION_INVALID);
+            sessions.remove(sessionId, state);
+            return true;
+        } catch (IOException | RuntimeException exception) {
+            // 추적 정보를 남겨 다음 검사에서 다시 닫을 수 있게 한다. 식별 정보는 로그에 노출하지 않는다.
+            log.warn("유효하지 않은 STOMP WebSocket 세션 종료에 실패했습니다: {}", exception.getMessage());
+            return false;
+        }
+    }
+
+    /** 주기 검사에 필요한 sessionId와 불변 인증 snapshot만 노출하는 조회 경계다. */
+    public record SessionSnapshot(String sessionId, StompSessionAuthorization authorization) {
+    }
+
+    /** 물리 연결과 CONNECT 이후 추가되는 인증 snapshot을 하나의 map entry에 보관한다. */
+    private static final class SessionState {
+
+        private final WebSocketSession session;
+        private volatile StompSessionAuthorization authorization;
+
+        private SessionState(WebSocketSession session) {
+            this.session = session;
+        }
+
+        private WebSocketSession session() {
+            return session;
+        }
+
+        private StompSessionAuthorization authorization() {
+            return authorization;
+        }
+
+        private void authenticate(StompSessionAuthorization authorization) {
+            this.authorization = authorization;
+        }
+    }
+}