Coverage for api\serializers\login_serializer.py: 88.3%

1from rest_framework import serializers 

2from rest_framework_simplejwt.serializers import TokenObtainPairSerializer, TokenRefreshSerializer 

3from rest_framework_simplejwt.tokens import RefreshToken 

4from rest_framework_simplejwt.settings import api_settings 

5from rest_framework_simplejwt.exceptions import InvalidToken 

6from users.models import Users 

7from users.services.auth_service import AuthService 

8from users.services.user_service import UserService 

9from users.services.exceptions import (EmailDeliveryError, InvalidTokenError, TokenExpiredError, UserNotFoundError) 

10from api.exceptions import CustomAPIException 

11from api.errors import ErrorCodes 

12import logging 

13 

14 

15logger = logging.getLogger(__name__) 

16 

17""" 

18 Clase para obtener tokens JWT personalizados. 

19 

20 Args: 

21 username (str): El nombre de usuario o email del usuario 

22 password (str): La contraseña del usuario 

23 code (str, optional): El codigo de autenticación de dos factores (2FA) si el usuario lo tiene habilitado 

24  

25 Returns: 

26 dict: Un diccionario con los tokens JWT. 

27 

28 Raises: 

29 CustomAPIException: Si las credenciales son invalidas o el usuario no esta verificado. 

30""" 

31 

32class CustomTokenObtainPairSerializer(TokenObtainPairSerializer): 

33 

34 # Campo para el nombre de usuario o email 

35 username = serializers.CharField(required=True, write_only=True) 

36 password = serializers.CharField(write_only=True) 

37 code = serializers.CharField(required=False, write_only=True) 

38 

39 """ 

40 Valida las credenciales del usuario, si el usuario tiene habilitado el 2FA, envia el codigo al email 

41 y lo verifica, si todo es correcto se generan los token JWT 

42 

43 Args: 

44 attrs (dict): Un diccionario con los campos 'username', 'password' y opcionalmente 'code'. 

45 

46 Returns: 

47 dict: Un diccionario con los tokens JWT. 

48 

49 Raises: 

50 CustomAPIException: Si las credenciales son invalidas, el usuario no esta verificado, el codigo 2FA es incorrecto o ha expirado. 

51 """ 

52 def validate(self, attrs): 

53 username = attrs.get('username') 

54 password = attrs.get('password') 

55 code = attrs.get('code') 

56 

57 logger.info("Try to login the user: %s", username) 

58 

59 if not username or not password: 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true

60 raise CustomAPIException( 

61 code=ErrorCodes.INSUFICIENT_CREDENTIALS, 

62 message="El nombre de usuario y la contraseña son obligatorios", 

63 status_code=400 

64 ) 

65 

66 try: 

67 # Se comprueba tanto el nombre de usuario como el email (siempre que no este eliminado) 

68 user = UserService.find_by_username_or_email(username) 

69 except UserNotFoundError: 

70 logger.warning("Login failed, user not found: %s", username) 

71 raise CustomAPIException( 

72 code=ErrorCodes.USER_DONT_EXIST, 

73 message="No exise un usuario registrado con ese email o nombre de usuario", 

74 status_code=400 

75 ) 

76 

77 if not user.is_verify: 

78 logger.warning("Login failed, user not verified: %s", username) 

79 raise CustomAPIException( 

80 code=ErrorCodes.USER_DONT_EXIST, # Si el usuario no esta validado, es como si no existiera 

81 message="No exise un usuario registrado con ese email o nombre de usuario", 

82 status_code=400 

83 ) 

84 

85 if user.is_deleted: 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true

86 raise CustomAPIException( 

87 code=ErrorCodes.USER_DELETED, 

88 message="No exise un usuario registrado con ese email o nombre de usuario", 

89 status_code=400 

90 ) 

91 if not user.check_password(password): 

92 logger.warning("Login failed, wrong password for user: %s", username) 

93 raise CustomAPIException( 

94 code=ErrorCodes.INVALID_CREDENTIALS, 

95 message="Nombre de usuario o contraseña incorrectos", 

96 status_code=400 

97 ) 

98 

99 if getattr(user, 'has_2FA', False): 

100 if not code: 

101 try: 

102 AuthService.start_second_factor(user) 

103 except EmailDeliveryError: 

104 raise CustomAPIException( 

105 code=ErrorCodes.EMAIL_ERROR, 

106 message="Error en el envio del email de verificación", 

107 status_code=500 

108 ) 

109 return { 

110 "requires_2fa": True, 

111 "message": "2FA code sent to email." 

112 } 

113 

114 else: 

115 if not AuthService.has_pending_code(user): 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true

116 raise CustomAPIException( 

117 code=ErrorCodes.TOKEN_EXPIRED, 

118 message="El código ha expirado", 

119 status_code=400 

120 ) 

121 try: 

122 AuthService.check_second_factor(user, code) 

123 except TokenExpiredError: 

124 raise CustomAPIException( 

125 code=ErrorCodes.TOKEN_EXPIRED, 

126 message="El código ha expirado", 

127 status_code=400 

128 ) 

129 except InvalidTokenError: 

130 raise CustomAPIException( 

131 code=ErrorCodes.INCORRECT_TOKEN, 

132 message="Código incorrecto", 

133 status_code=400 

134 ) 

135 

136 logger.info("Login successful for user: %s", username) 

137 return AuthService.build_auth_payload(user) 

138 

139class CustomTokenRefreshSerializer(TokenRefreshSerializer): 

140 """ 

141 Valida el token de refresh, si es valido se genera un nuevo token de access y opcionalmente un nuevo token de refresh 

142  

143 Args: 

144 attrs (dict): Un diccionario con el token de refresh 

145 

146 Returns: 

147 dict: Un diccionario con los tokens JWT. 

148 """ 

149 def validate(self, attrs): 

150 refresh = RefreshToken(attrs['refresh']) 

151 user_id = refresh.get(api_settings.USER_ID_CLAIM) 

152 user = Users.objects.filter(**{api_settings.USER_ID_FIELD: user_id, 'is_deleted': False}).first() if user_id else None 

153 

154 if user is None: 

155 logger.warning("Refresh rechazado: usuario del token inexistente (%s)", user_id) 

156 raise InvalidToken('El usuario del token no existe') 

157 # Se comprueba si la sesion se ha revocado (el usuario ha cerrado sesiono cambiado la contraseña) 

158 if AuthService.token_is_revoked(user, refresh): 

159 logger.warning("Refresh rechazado: token revocado para el usuario %s", user_id) 

160 raise InvalidToken('El token ha sido revocado') 

161 

162 data = {'access': str(refresh.access_token)} 

163 if api_settings.ROTATE_REFRESH_TOKENS: 163 ↛ 169line 163 didn't jump to line 169 because the condition on line 163 was always true

164 try: 

165 data['refresh'] = str(refresh) 

166 except Exception: 

167 pass 

168 

169 return data