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

1from urllib.parse import unquote 

2 

3from fastapi import APIRouter, HTTPException, Query, Request, Response, status 

4 

5from ...core.logging import get_structured_logger 

6from ..authenticate import authenticator 

7from ..exceptions import AuthenticationFailed 

8from ..oauth_manager import oauth_manager 

9from ..schemas.auth import LoginResponse, UserInfo 

10from ..security.jwt import get_jwt_manager 

11from ..user_manager import UserManager 

12 

13user_manager = UserManager() 

14jwt_manager = get_jwt_manager() 

15logger = get_structured_logger(__name__) 

16 

17 

18def get_oauth2_router() -> APIRouter: 

19 """OAuth2 인증을 위한 라우터 생성""" 

20 

21 router = APIRouter() 

22 

23 @router.get( 

24 "/{provider}/authorize", 

25 response_model=str, 

26 ) 

27 async def authorize( 

28 provider: str, 

29 redirect_url: str | None = None, 

30 state: str | None = Query(None), 

31 ) -> str: 

32 """ 

33 OAuth2 인증 프로세스를 시작합니다. 

34 

35 Args: 

36 provider: OAuth 제공자 (google, kakao, naver 등) 

37 redirect_url: 인증 후 리다이렉트할 URL 

38 state: CSRF 방지를 위한 state 파라미터 

39 

40 Returns: 

41 str: OAuth 제공자의 인증 URL 

42 """ 

43 try: 

44 authorization_url = await oauth_manager.generate_auth_url( 

45 provider, state, redirect_url 

46 ) 

47 return authorization_url 

48 except Exception as e: 

49 error_msg = str(e) 

50 if error_msg == "Unknown OAuth provider": 

51 logger.error(f"Unknown OAuth provider: {provider}") 

52 raise HTTPException( 

53 status_code=status.HTTP_400_BAD_REQUEST, 

54 detail=f"Unknown OAuth provider: {provider}", 

55 ) 

56 logger.error(f"{provider} authorize error: {e}") 

57 raise HTTPException( 

58 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, 

59 detail="Failed to generate authorization URL", 

60 ) 

61 

62 @router.get( 

63 "/{provider}/callback", 

64 response_model=LoginResponse, 

65 description="OAuth2 콜백 엔드포인트. 인증 백엔드에 따라 응답이 달라집니다.", 

66 ) 

67 async def callback( 

68 request: Request, 

69 response: Response, 

70 provider: str, 

71 code: str = Query(...), 

72 redirect_url: str | None = None, 

73 ) -> LoginResponse: 

74 """ 

75 OAuth2 콜백을 처리하고 사용자를 인증합니다. 

76 

77 Args: 

78 request: FastAPI Request 객체 

79 response: FastAPI Response 객체 

80 provider: OAuth 제공자 

81 code: 인증 코드 

82 redirect_url: 리다이렉트 URL 

83 

84 Returns: 

85 LoginResponse: 액세스 토큰과 사용자 정보 

86 """ 

87 decoded_code = unquote(code) 

88 

89 # (1) 액세스 토큰 및 프로필 정보 가져오기 

90 token_data, profile_data = await oauth_manager.get_access_token_and_profile( 

91 provider, 

92 decoded_code, 

93 redirect_url or oauth_manager.get_redirect_uri(provider), 

94 ) 

95 

96 # (2) 프로필 파서 정의 

97 def parse_google_profile(profile_data): 

98 return { 

99 "profile_email": profile_data.email, 

100 "profile_id": profile_data.id, 

101 "profile_image": getattr(profile_data, "picture", None), 

102 "fullname": getattr(profile_data, "name", None), 

103 } 

104 

105 def parse_kakao_profile(profile_data): 

106 return { 

107 "profile_email": profile_data.kakao_account.email, 

108 "profile_id": str(profile_data.id), 

109 "profile_image": profile_data.kakao_account.profile.profile_image_url, 

110 "fullname": profile_data.kakao_account.profile.nickname, 

111 } 

112 

113 def parse_naver_profile(profile_data): 

114 return { 

115 "profile_email": profile_data.email, 

116 "profile_id": profile_data.id, 

117 "profile_image": profile_data.profile_image, 

118 "fullname": profile_data.name, 

119 } 

120 

121 profile_parsers = { 

122 "google": parse_google_profile, 

123 "kakao": parse_kakao_profile, 

124 "naver": parse_naver_profile, 

125 } 

126 

127 if provider not in profile_parsers: 

128 raise HTTPException( 

129 status_code=status.HTTP_400_BAD_REQUEST, 

130 detail=f"Unsupported OAuth provider: {provider}", 

131 ) 

132 

133 # (3) 프로필 데이터 파싱 

134 try: 

135 profile_kwargs = profile_parsers[provider](profile_data) 

136 except Exception as e: 

137 logger.error(f"Failed to parse profile data for provider {provider}: {e}") 

138 raise HTTPException( 

139 status_code=status.HTTP_400_BAD_REQUEST, 

140 detail=f"Failed to parse profile data: {e}", 

141 ) 

142 

143 # (4) 사용자 생성 또는 업데이트 

144 user = await user_manager.oauth_callback( 

145 oauth_name=provider, 

146 token_data=token_data, # type: ignore[arg-type] 

147 **profile_kwargs, 

148 request=request, 

149 ) 

150 if not user: 

151 raise AuthenticationFailed("Failed to authenticate with OAuth provider") 

152 

153 # (5) 로그인 토큰 생성 

154 auth_token = authenticator.login(user=user, response=response) 

155 

156 return LoginResponse( 

157 access_token=auth_token.access_token if auth_token else None, 

158 refresh_token=auth_token.refresh_token if auth_token else None, 

159 token_type="bearer", 

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

161 ) 

162 

163 return router