feat(backend): 로그아웃 + stateful 토큰 검증 (단일 세션 즉시 폐기)

- POST /v1/auth/logout: 저장된 access/refresh 토큰 행 삭제 → 즉시 로그아웃
- /me, refresh 에서 제시된 토큰을 저장된 토큰과 대조(불일치 시 TOKEN_REVOKED=1203)
  → 로그아웃/타기기 재로그인으로 교체된 토큰을 만료 전이라도 차단
- crud get_token 추가, ErrorType.TOKEN_REVOKED(1203) 추가, Res_Logout 프로토콜
- 로그아웃/단일세션 무효화 e2e 테스트 추가 (test_auth.py)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/common/enums.py

@@ -35,6 +35,7 @@ class ErrorType(Enum):
     ACCOUNT_INVALID_INFO = 1200
     ACCOUNT_ALREADY_EXIST = auto()
     ACCOUNT_BLOCKED_USER = auto()
+    TOKEN_REVOKED = auto()  # 제시된 토큰이 저장된 토큰과 불일치(로그아웃/타기기 로그인으로 교체됨)
 
 
 # ErrorType 의 HTTP_* 값과 status_code 를 맞춰 router 단에서 raise 한다.

backend/crud/user_crud.py

@@ -40,6 +40,10 @@ class IUserCRUD(ABC):
     async def add_token(self, cdb: AsyncSession, token: supplier_user_tokens) -> ErrorType:
         pass
 
+    @abstractmethod
+    async def get_token(self, cdb: AsyncSession, su_id, token_type: int) -> Tuple[ErrorType, str]:
+        pass
+
     @abstractmethod
     async def delete_tokens_by_su_id(self, cdb: AsyncSession, su_id) -> ErrorType:
         pass
@@ -136,6 +140,28 @@ class UserCRUD(IUserCRUD):
             LOG.e_no_callstack(ex)
             return ErrorType.DB_RUN_FAILED
 
+    async def get_token(self, cdb: AsyncSession, su_id, token_type: int) -> Tuple[ErrorType, str]:
+        # 저장된 토큰(jwt 문자열)을 반환한다. stateful 검증(제시 토큰 ↔ 저장 토큰 대조)용.
+        try:
+            query = (
+                select(supplier_user_tokens.token["jwt"].astext)
+                .where(
+                    supplier_user_tokens.su_id == su_id,
+                    supplier_user_tokens.type == token_type,
+                    supplier_user_tokens.deleted == False,  # noqa: E712
+                )
+                .limit(1)
+            )
+            err_type, row_list = await DB_SESSION_MNG.execute(cdb, query)
+            if err_type != ErrorType.SUCCESS:
+                return err_type, None
+            if len(row_list) != 1:
+                return ErrorType.DB_INVALID_KEY, None
+            return ErrorType.SUCCESS, row_list[0]
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, None
+
     async def delete_tokens_by_su_id(self, cdb: AsyncSession, su_id) -> ErrorType:
         # 단일 세션: 로그인/로그아웃 시 해당 유저의 토큰 행을 모두 제거한다(하드 삭제, 누적 방지).
         try:

backend/router/v1/auth/account.py

@@ -1,9 +1,15 @@
 from fastapi import APIRouter, Depends, Request
+from fastapi.security import HTTPAuthorizationCredentials
 
 from common.models.gmodel import UserInfo
-from router.v1.validator.dependencies import IsValidAccessToken, IsValidRefreshToken, RemoveNoneResponse
+from router.v1.validator.dependencies import (
+    IsValidAccessToken,
+    IsValidRefreshToken,
+    RemoveNoneResponse,
+    security,
+)
 from services.auth_service import AuthService
-from .protocol import Req_CreateAccount, Req_Login, Res_CreateAccount, Res_Login, Res_Me, Res_RefreshToken
+from .protocol import Req_CreateAccount, Req_Login, Res_CreateAccount, Res_Login, Res_Logout, Res_Me, Res_RefreshToken
 
 # 라우터(MVC 의 컨트롤러). 요청 검증 -> service 호출 -> RemoveNoneResponse 반환만 담당.
 router = APIRouter(prefix="/v1/auth", tags=["Auth"], responses={404: {"description": "Not found"}})
@@ -28,17 +34,35 @@ async def create_account(request: Request, req: Req_CreateAccount, service: Auth
     path="/refresh_token",
     response_model=Res_RefreshToken,
     summary="액세스 토큰 갱신",
-    description="refresh 토큰으로 access 토큰을 재발급한다. su_id DB 존재/활성은 service 에서 확인한다.",
+    description="refresh 토큰으로 access 토큰을 재발급한다. su_id DB 존재/활성 + 저장 토큰 대조를 service 에서 확인한다.",
 )
-async def refresh_token(user_info: UserInfo = Depends(IsValidRefreshToken), service: AuthService = Depends()):
+async def refresh_token(
-    return RemoveNoneResponse(await service.refresh_token(user_info))
+    user_info: UserInfo = Depends(IsValidRefreshToken),
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+    service: AuthService = Depends(),
+):
+    return RemoveNoneResponse(await service.refresh_token(user_info, credentials.credentials))
+
+
+@router.post(
+    path="/logout",
+    response_model=Res_Logout,
+    summary="로그아웃",
+    description="저장된 access/refresh 토큰을 폐기한다. 이후 보호 요청·재발급이 차단된다(단일 세션).",
+)
+async def logout(user_info: UserInfo = Depends(IsValidAccessToken), service: AuthService = Depends()):
+    return RemoveNoneResponse(await service.logout(user_info))
 
 
 @router.get(
     path="/me",
     response_model=Res_Me,
     summary="내 정보 (보호된 엔드포인트)",
-    description="access 토큰 검증(validator) 후 su_id DB 존재/활성을 service 에서 확인해 반환한다.",
+    description="access 토큰 검증(validator) 후 su_id DB 존재/활성 + 저장 토큰 대조를 service 에서 확인해 반환한다.",
 )
-async def me(user_info: UserInfo = Depends(IsValidAccessToken), service: AuthService = Depends()):
+async def me(
-    return RemoveNoneResponse(await service.get_me(user_info))
+    user_info: UserInfo = Depends(IsValidAccessToken),
+    credentials: HTTPAuthorizationCredentials = Depends(security),
+    service: AuthService = Depends(),
+):
+    return RemoveNoneResponse(await service.get_me(user_info, credentials.credentials))

backend/router/v1/auth/protocol.py

@@ -46,3 +46,7 @@ class Res_Me(Res_WebPacketProtocol):
     supplier_id: str = ""
     supplier_name: str = ""
     role: int = 0
+
+
+class Res_Logout(Res_WebPacketProtocol):
+    pass

backend/services/auth_service.py

@@ -10,7 +10,7 @@ from common.models.gmodel import UserInfo
 from common.utils.gtime import GTime
 from config.server_configs import jwt_token_config
 from crud.user_crud import IUserCRUD, UserCRUD
-from router.v1.auth.protocol import Res_CreateAccount, Res_Login, Res_Me, Res_RefreshToken
+from router.v1.auth.protocol import Res_CreateAccount, Res_Login, Res_Logout, Res_Me, Res_RefreshToken
 from router.v1.validator.dependencies import CreateAccessToken, CreateRefreshToken, GetHashedPW, VerifyPW
 
 
@@ -207,13 +207,27 @@ class AuthService:
             role=account.role,
         )
 
-    async def get_me(self, user_info: UserInfo) -> Res_Me:
+    async def __verify_stored_token(self, su_id_str: str, token_type: int, presented: str) -> bool:
-        # 토큰 디코드는 라우터 Depends(IsValidAccessToken) 에서 수행됨. 여기선 su_id DB 검증.
+        """제시된 토큰이 저장된 토큰과 일치하는지 확인한다(stateful 단일 세션).
+        로그아웃·타기기 로그인으로 교체되면 저장 토큰이 없거나 달라져 False 가 된다.
+        """
+        err_type, stored = await DB_SESSION_MNG.execute_lambda(
+            supplier_users.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.user_crud.get_token(s, uuid.UUID(su_id_str), token_type),
+        )
+        return err_type == ErrorType.SUCCESS and stored == presented
+
+    async def get_me(self, user_info: UserInfo, access_token: str) -> Res_Me:
+        # 토큰 디코드는 라우터 Depends(IsValidAccessToken) 에서 수행됨. 여기선 su_id DB 검증 + 저장 토큰 대조.
         res = Res_Me()
         err_type, info = await self.__load_active_account(user_info.su_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
+        if not await self.__verify_stored_token(info.su_id, TokenType.ACCESS.value, access_token):
+            res.result.SetResult(ErrorType.TOKEN_REVOKED)  # 로그아웃/타기기 로그인으로 무효화됨
+            return res
         res.su_id = info.su_id
         res.id = info.id
         res.name = info.name
@@ -222,13 +236,27 @@ class AuthService:
         res.role = info.role
         return res
 
-    async def refresh_token(self, user_info: UserInfo) -> Res_RefreshToken:
+    async def logout(self, user_info: UserInfo) -> Res_Logout:
-        # 토큰 디코드는 라우터 Depends(IsValidRefreshToken) 에서 수행됨. 여기선 su_id DB 검증 후 재발급.
+        # 해당 유저의 저장 토큰(access/refresh)을 모두 삭제 → 이후 보호 요청·재발급이 차단된다.
+        res = Res_Logout()
+        err_type = await DB_SESSION_MNG.execute_lambda_run(
+            [supplier_users.DBType()],
+            [lambda s: self.user_crud.delete_tokens_by_su_id(s, uuid.UUID(user_info.su_id))],
+        )
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+        return res
+
+    async def refresh_token(self, user_info: UserInfo, refresh_token: str) -> Res_RefreshToken:
+        # 토큰 디코드는 라우터 Depends(IsValidRefreshToken) 에서 수행됨. 여기선 su_id DB 검증 + 저장 토큰 대조 후 재발급.
         res = Res_RefreshToken()
         err_type, info = await self.__load_active_account(user_info.su_id)
         if err_type != ErrorType.SUCCESS:
             res.result.SetResult(err_type)
             return res
+        if not await self.__verify_stored_token(info.su_id, TokenType.REFRESH.value, refresh_token):
+            res.result.SetResult(ErrorType.TOKEN_REVOKED)  # 로그아웃/타기기 로그인으로 무효화됨
+            return res
 
         new_access = CreateAccessToken(info)  # DB 최신값으로 재구성한 토큰
         # 단일 세션: 저장된 access 행을 새 토큰으로 갱신한다(refresh 행은 유지).

backend/tests/test_auth.py

@@ -270,3 +270,55 @@ async def test_refresh_inactive_after_token(client, account_seed, db_engine):
     assert body["result"]["success"] is False
     assert body["result"]["code"] == 1202
     assert body.get("access_token", "") == ""
+
+
+# ---- 로그아웃 / stateful 토큰 검증 -------------------------------------------
+async def test_logout_revokes_tokens(client, account_seed, db_engine):
+    body = (await _login(client)).json()
+    su_id, access, refresh = body["su_id"], body["access_token"], body["refresh_token"]
+
+    # 로그아웃 성공
+    r = await client.post("/v1/auth/logout", headers={"Authorization": f"Bearer {access}"})
+    assert r.status_code == 200
+    assert r.json()["result"]["success"] is True
+
+    # 저장 토큰이 모두 삭제됨
+    async with db_engine.begin() as conn:
+        count = (
+            await conn.execute(
+                text("SELECT count(*) FROM supplier.supplier_user_tokens WHERE su_id = :sid AND deleted = false"),
+                {"sid": uuid.UUID(su_id)},
+            )
+        ).scalar()
+    assert count == 0
+
+    # 로그아웃 후 같은 access 로 /me → TOKEN_REVOKED(1203)
+    r2 = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access}"})
+    assert r2.status_code == 200
+    assert r2.json()["result"]["code"] == 1203
+
+    # 로그아웃 후 같은 refresh 로 재발급 → TOKEN_REVOKED(1203)
+    r3 = await client.post("/v1/auth/refresh_token", headers={"Authorization": f"Bearer {refresh}"})
+    assert r3.json()["result"]["code"] == 1203
+
+
+async def test_logout_without_token(client):
+    r = await client.post("/v1/auth/logout")
+    assert r.status_code in (401, 403)
+
+
+async def test_relogin_invalidates_previous_access(client, account_seed):
+    # 단일 세션: 재로그인하면 이전 세션의 access 가 무효화된다(저장 토큰이 교체됨).
+    import asyncio
+
+    first = (await _login(client)).json()
+    await asyncio.sleep(1.1)  # exp(초 단위)가 달라져 토큰이 실제로 바뀌도록
+    second = (await _login(client)).json()
+    assert first["access_token"] != second["access_token"]
+
+    # 이전 access → 무효(TOKEN_REVOKED)
+    r_old = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {first['access_token']}"})
+    assert r_old.json()["result"]["code"] == 1203
+    # 새 access → 정상
+    r_new = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {second['access_token']}"})
+    assert r_new.json()["result"]["success"] is True