Coverage for src / mysingle / auth / router / auth.py: 0%

1"""Health check utilities and endpoints.""" 

2 

3from typing import Annotated 

4 

5from beanie import PydanticObjectId 

6from fastapi import ( 

7 APIRouter, 

8 Cookie, 

9 Depends, 

10 Header, 

11 HTTPException, 

12 Request, 

13 Response, 

14 status, 

15) 

16from fastapi.security import OAuth2PasswordRequestForm 

17 

18from ...core.config import settings 

19from ...core.logging import get_structured_logger 

20from ..authenticate import authenticator 

21from ..deps import get_current_user, get_current_user_optional, verified_only 

22from ..exceptions import AuthenticationFailed 

23from ..schemas.auth import LoginResponse, UserInfo, VerifyTokenResponse 

24from ..user_manager import UserManager 

25 

26logger = get_structured_logger(__name__) 

27access_token_expire_minutes = settings.ACCESS_TOKEN_EXPIRE_MINUTES 

28user_manager = UserManager() 

29authenticator = authenticator 

30 

31 

32def create_auth_router() -> APIRouter: 

33 router = APIRouter() 

34 

35 @router.post( 

36 "/login", 

37 response_model=LoginResponse, 

38 status_code=status.HTTP_200_OK, # 202 -> 200 변경 

39 ) 

40 async def login( 

41 response: Response, # Response 객체를 직접 받도록 수정 

42 form_data: Annotated[OAuth2PasswordRequestForm, Depends()], 

43 ) -> LoginResponse | None: 

44 user = await user_manager.authenticate( 

45 username=form_data.username, password=form_data.password 

46 ) 

47 

48 if not user: 

49 raise AuthenticationFailed("Invalid credentials") 

50 

51 # authenticator.login을 호출하여 토큰 생성 

52 token_data = authenticator.login(user=user, response=response) 

53 

54 if settings.TOKEN_TRANSPORT_TYPE in ["bearer", "hybrid"]: 

55 # Bearer 또는 Hybrid 방식: 응답에 토큰 포함 

56 return LoginResponse( 

57 access_token=token_data.access_token if token_data else None, 

58 refresh_token=token_data.refresh_token if token_data else None, 

59 token_type="bearer", 

60 user_info=UserInfo(**user.model_dump(by_alias=True)), 

61 ) 

62 else: 

63 # Cookie 방식: 토큰은 쿠키에만 설정, 응답에는 사용자 정보만 

64 return None 

65 

66 @router.post("/logout", status_code=status.HTTP_204_NO_CONTENT) 

67 async def logout( 

68 request: Request, 

69 response: Response, 

70 ) -> None: 

71 """ 

72 로그아웃 엔드포인트. 

73 

74 쿠키에서 토큰을 삭제하고 로그아웃 처리를 합니다. 

75 """ 

76 # 현재 사용자 가져오기 (선택적) 

77 user = get_current_user_optional(request) 

78 

79 # authenticator를 사용하여 쿠키 삭제 

80 authenticator.logout(response) 

81 

82 # 사용자가 인증된 경우에만 후처리 로직 실행 

83 if user: 

84 try: 

85 current_user = await user_manager.get(PydanticObjectId(user.id)) 

86 if current_user: 

87 await user_manager.on_after_logout(current_user, request) 

88 except Exception as e: 

89 logger.warning(f"Failed to execute logout callback: {e}") 

90 

91 # HTTP 204는 응답 본문이 없어야 하므로 None 반환 

92 return None 

93 

94 @router.post("/refresh", response_model=LoginResponse) 

95 async def refresh_token( 

96 response: Response, # Response 객체 추가 

97 refresh_token_header: str | None = Header(None, alias="X-Refresh-Token"), 

98 refresh_token_cookie: str | None = Cookie(None, alias="refresh_token"), 

99 ) -> LoginResponse | None: 

100 """JWT 토큰 갱신 엔드포인트""" 

101 

102 # 토큰 전송 방식에 따라 refresh token 소스 결정 

103 if settings.TOKEN_TRANSPORT_TYPE == "bearer": 

104 refresh_token = refresh_token_header 

105 elif settings.TOKEN_TRANSPORT_TYPE == "cookie": 

106 refresh_token = refresh_token_cookie 

107 else: # hybrid 

108 refresh_token = refresh_token_header or refresh_token_cookie 

109 

110 if not refresh_token: 

111 raise AuthenticationFailed("Refresh token not provided") 

112 

113 try: 

114 # 토큰 전송 방식에 맞게 새 토큰 생성 

115 from typing import Literal 

116 

117 transport_type: Literal["cookie", "header", "bearer", "hybrid"] 

118 if settings.TOKEN_TRANSPORT_TYPE == "bearer": 

119 transport_type = "bearer" 

120 elif settings.TOKEN_TRANSPORT_TYPE == "cookie": 

121 transport_type = "cookie" 

122 else: # hybrid 

123 transport_type = "hybrid" 

124 

125 token_data = authenticator.refresh_token( 

126 refresh_token=refresh_token, 

127 response=response, 

128 transport_type=transport_type, 

129 ) 

130 except HTTPException: 

131 raise AuthenticationFailed("Invalid refresh token") 

132 

133 # 사용자 정보 조회 

134 try: 

135 payload = authenticator.validate_token(refresh_token) 

136 user_id = payload.get("sub") 

137 user = await user_manager.get(PydanticObjectId(user_id)) 

138 if not user: 

139 raise AuthenticationFailed("User not found") 

140 except Exception: 

141 raise AuthenticationFailed("Failed to retrieve user information") 

142 

143 # 토큰 전송 방식에 따른 응답 생성 

144 if settings.TOKEN_TRANSPORT_TYPE in ["bearer", "hybrid"] and token_data: 

145 return LoginResponse( 

146 access_token=token_data.access_token, 

147 refresh_token=token_data.refresh_token, 

148 token_type=token_data.token_type, 

149 user_info=UserInfo(**user.model_dump(by_alias=True)), 

150 ) 

151 else: 

152 # Cookie 방식 

153 return None 

154 

155 @router.get("/token/verify") 

156 @verified_only 

157 async def verify_token( 

158 request: Request, 

159 ) -> VerifyTokenResponse: 

160 """토큰 검증 및 사용자 정보 반환 (디버깅용)""" 

161 current_user = get_current_user(request) 

162 return VerifyTokenResponse( 

163 valid=True, 

164 user_id=str(current_user.id), 

165 email=current_user.email, 

166 is_verified=current_user.is_verified, 

167 is_superuser=current_user.is_superuser, 

168 is_active=current_user.is_active, 

169 ) 

170 

171 return router