Coverage for users\services\auth

1import hashlib

2import logging

3import re

4import secrets

5import string

6from datetime import timedelta

8from decouple import config

9from django.conf import settings

10from django.contrib.auth.password_validation import validate_password

11from django.db.models import F

12from django.utils import timezone

13from google.auth.transport import requests as google_requests

14from google.oauth2 import id_token

15from rest_framework_simplejwt.tokens import RefreshToken

17from services.email.email_service import Email

18from users.models import USERNAME_REGEX, Users, UserType, normalize_email

19from users.services.exceptions import (EmailAlreadyRegisteredError, EmailDeliveryError, EmailDomainNotAllowedError,

20 GoogleEmailMissingError, IncorrectPasswordError, InvalidGoogleTokenError, InvalidTokenError, InvalidUsernameError,

21 InvalidUserTypeError, TokenExpiredError, UnverifiedEmailError, UsernameAlreadyTakenError, UserNotFoundError)

23# Logger para ir almacenando los logs

24logger = logging.getLogger(__name__)

26# Longitud del codigo de verificacion que se manda por correo.

28TOKEN_LENGTH = 6

30# Claim donde viaja la generacion de sesiones de la cuenta

31SESSION_EPOCH_CLAIM = 'sv'

33# Propositos de los codigos de verificacion

34PURPOSE_VERIFY_EMAIL = 'verify_email'

35PURPOSE_RESET_PASSWORD = 'reset_password'

36PURPOSE_CHANGE_EMAIL = 'change_email'

37PURPOSE_LOGIN_2FA = 'login_2fa'

38PURPOSE_CHANGE_2FA = 'change_2fa'

40# Propositos que se consideran equivalentes entre si

41INTERCHANGEABLE_PURPOSES = {PURPOSE_VERIFY_EMAIL, PURPOSE_RESET_PASSWORD}

43# Texto que se devuelve cuando el correo no es de un dominio admitido

44EMAIL_DOMAIN_ERROR_MESSAGE = "El correo no pertenece al dominio permitido, utiliza otro"

46# Servicio con la logica de autenticacion, registro y verificacion por correo

47class AuthService:

49 # Genera un codigo alfanumerico aleatorio y devuelve el codigo y su hash

50 @staticmethod

51 def generate_token():

52 token = ''.join(secrets.choice(string.ascii_letters + string.digits) for _ in range(TOKEN_LENGTH))

53 token_hash = hashlib.sha256(token.encode()).hexdigest()

54 return token, token_hash

56 # Comprueba que el codigo aportado coincida con el hash almacenado

57 @staticmethod

58 def verify_token(user_input, stored_hash):

59 input_hash = hashlib.sha256(user_input.encode()).hexdigest()

60 return input_hash == stored_hash

62 # Deduce para que se esta pidiendo el codigo

63 @staticmethod

64 def purpose_for_request(user, to_email=None):

65 """

66 El cliente no manda el proposito, se deduce:

67 hay una direccion de destino distinta -> cambio de correo

68 la cuenta no esta verificada todavia -> verificacion de la cuenta

69 en cualquier otro caso -> restablecer la contraseña

70 """

71 if to_email:

72 return PURPOSE_CHANGE_EMAIL

73 if not user.is_verify:

74 return PURPOSE_VERIFY_EMAIL

75 return PURPOSE_RESET_PASSWORD

77 # Genera un codigo, lo manda por correo y lo deja guardado en la cuenta

78 @staticmethod

79 def send_verification_code(user, to_email=None, purpose=None):

80 """

81 to_email solo llega con valor en el flujo de cambio de correo desde el perfil

82 Solo se guarda el hash del codigo, nunca el codigo en claro

83 """

84 token, hashed_token = AuthService.generate_token()

86 try:

87 email_service = Email()

88 email_service.send_verification_email(to=to_email or user.email, token=token)

89 except Exception as e:

90 logger.error(f"Error sending verification email: {str(e)}")

91 raise EmailDeliveryError(user.id)

93 user.token = hashed_token

94 user.token_purpose = purpose or AuthService.purpose_for_request(user, to_email)

95 user.token_expiration = timezone.now() + timedelta(

96 minutes=config('TOKEN_EXPIRATION_TIME', cast=int, default=5)

97 )

98 user.save()

100 logger.info(f"Token generated for user ID: {user.id} with purpose: {user.token_purpose}")

101 return user

102

103 # Comprueba que el codigo pendiente sirva para lo que se quiere hacer con el

104 @staticmethod

105 def assert_purpose(user, required):

106 """

107 Un codigo emitido para el segundo factor del login no vale para cambiar la contraseña, ni al reves

108 """

109 stored = user.token_purpose

110 if stored is None:

111 return

112

113 if stored == required:

114 return

115

116 if {stored, required} <= INTERCHANGEABLE_PURPOSES:

117 return

118

119 logger.warning(

120 f"Token purpose mismatch for user ID: {user.id}: "

121 f"issued for {stored}, used for {required}"

122 )

123 raise InvalidTokenError(user.id)

124

125 # Consume el codigo de verificacion y aplica los cambios que se pidan

126 @staticmethod

127 def consume_verification_code(user, token, email=None, password=None, validate=False,

128 required_purpose=None):

129 """

130 Comprueba el codigo y, si es correcto, aplica los cambios solicitados

131 """

132 # Si nunca se pidio un codigo, o ya se consumio, no hay nada que verificar

133 if not user.token or not user.token_expiration:

134 logger.warning(f"Verification attempt without a pending token for user ID: {user.id}")

135 raise InvalidTokenError(user.id)

136

137 if user.token_expiration < timezone.now():

138 logger.warning(f"Expired verification token for user ID: {user.id}")

139 raise TokenExpiredError(user.id)

140

141 if not AuthService.verify_token(token, user.token):

142 logger.warning(f"Invalid verification token for user ID: {user.id}")

143 raise InvalidTokenError(user.id)

144

145 if required_purpose is None:

146 if email:

147 required_purpose = PURPOSE_CHANGE_EMAIL

148 elif password:

149 required_purpose = PURPOSE_RESET_PASSWORD

150 elif validate:

151 required_purpose = PURPOSE_VERIFY_EMAIL

152 if required_purpose is not None:

153 AuthService.assert_purpose(user, required_purpose)

154

155 # La politica de contraseñas se comprueba antes, para no consumir el codigo

156 if password:

157 AuthService.assert_password_is_acceptable(password, user=user)

158

159 if email:

160 # El correo nuevo tiene que ser de un dominio admitido

161 AuthService.assert_email_domain_is_allowed(email)

162 if AuthService.email_is_taken(email, excluding=user):

163 logger.warning(f"Rejected email change for user ID: {user.id}: address already in use")

164 raise EmailAlreadyRegisteredError(email)

165 user.email = email

166 if password:

167 user.set_password(password)

168 if validate:

169 user.is_verify = True

170

171 # El codigo es de un solo uso

172 user.token = None

173 user.token_expiration = None

174 user.token_purpose = None

175 user.save()

176

177 # Si se cambia la contraseña, se cierra la sesion en todos los dispositivos y se invalidan los refresh tokens emitidos

178 if password:

179 AuthService.revoke_all_sessions(user)

180

181 logger.info(f"Post_verify_code successful. User data updated successfully for user ID: {user.id}")

182 return user

183

184 # Arranca el segundo factor y devuelve el metodo empleado

185 @staticmethod

186 def start_second_factor(user):

187 """

188 Se devuelve el metodo en lugar de no devolver nada precisamente para que sepa que metodo se ha usado

189 """

190 AuthService.send_verification_code(user, purpose=PURPOSE_LOGIN_2FA)

191 # En un futuro se pueden añadir mas metodos, como TOTP

192 return 'email'

193

194 # Indica si la cuenta tiene un codigo pendiente de canjear

195 @staticmethod

196 def has_pending_code(user):

197 return bool(user.token and user.token_expiration)

198

199 # Comprueba el codigo del segundo factor y lo consume

200 @staticmethod

201 def check_second_factor(user, code):

202 """

203 Delega en consume_verification_code, que es el mismo mecanismo que usa la recuperacion de cuenta

204 Exige que el codigo se emitiera para el login

205 """

206 AuthService.consume_verification_code(user, code, required_purpose=PURPOSE_LOGIN_2FA)

207

208 # Indica si la cuenta puede acreditarse con una contraseña

209 @staticmethod

210 def can_use_password(user):

211 """

212 Las cuentas que solo entran por Google no tienen contraseña que pedir, dejan el campo a NULL en ese caso

213 """

214 return bool(user.password)

215

216 # Arranca el cambio del ajuste y devuelve como hay que acreditarse

217 @staticmethod

218 def start_second_factor_change(user):

219 """

220 Quitar o poner el segundo factor exige volver a demostrar quien eres, se demuestra dependiendo de la cuenta,

221 si tiene contraseña, con la contraseña, si solo entra por Google, con un codigo enviado a su correo

222 """

223 if AuthService.can_use_password(user):

224 return 'password'

225

226 AuthService.send_verification_code(user, purpose=PURPOSE_CHANGE_2FA)

227 return 'email'

228

229 # Activa o desactiva el segundo factor, acreditando antes la identidad

230 @staticmethod

231 def set_second_factor(user, enabled, password=None, code=None):

232 """

233 Se pide credencial tanto para activarlo como para desactivarlo

234 Lanza IncorrectPasswordError, InvalidTokenError o TokenExpiredError.

235 """

236 if AuthService.can_use_password(user):

237 if not password or not user.check_password(password):

238 logger.warning(f"Rejected second factor change for user ID: {user.id}: wrong password")

239 raise IncorrectPasswordError(user.id)

240 else:

241 # Sin esto, verify_token intentaria codificar un None

242 if not code:

243 logger.warning(f"Rejected second factor change for user ID: {user.id}: no code provided")

244 raise InvalidTokenError(user.id)

245

246 # Comprueba el codigo, exige que se emitiera para esto y lo consume

247 AuthService.consume_verification_code(user, code, required_purpose=PURPOSE_CHANGE_2FA)

248

249 user.has_2FA = enabled

250 user.save()

251

252 # Tocar el segundo factor revoca las sesiones abiertas

253 AuthService.revoke_all_sessions(user)

254

255 logger.info(f"Second factor set to {enabled} for user ID: {user.id}")

256 return user

257

258 # Indica si un correo ya esta en uso por una cuenta activa

259 @staticmethod

260 def email_is_taken(email, excluding=None):

261 """

262 excluding sirve para no chocar contra uno mismo al actualizar el perfil

263 """

264 qs = Users.objects.filter(email=normalize_email(email), is_deleted=False)

265 if excluding is not None:

266 qs = qs.exclude(pk=excluding.pk)

267 return qs.exists()

268

269 # Indica si un nombre de usuario ya esta en uso por una cuenta activa

270 @staticmethod

271 def username_is_taken(username, excluding=None):

272 qs = Users.objects.filter(username=username, is_deleted=False)

273 if excluding is not None:

274 qs = qs.exclude(pk=excluding.pk)

275 return qs.exists()

276

277 # Invalida todos los tokens emitidos hasta ahora para esa cuenta

278 @staticmethod

279 def revoke_all_sessions(user):

280 """

281 Sube sessions_epoch, con lo que todo token que lleve un `sv` menor deja

282 de valer. Cubre el access y el refresh de una vez, porque el claim se

283 copia del refresh al access al emitirlo

284 """

285 # F() y no user.sessions_epoch + 1, dos revocaciones simultaneas leyendo el mismo valor dejarian el

286 # contador una generacion por detras, y con el los tokens de la primera seguirian valiendo

287 user.sessions_epoch = F('sessions_epoch') + 1

288 user.save(update_fields=['sessions_epoch'])

289 # save() deja el atributo como una expresion F sin resolver, hay que recargarlo para que quien

290 # llame pueda emitir tokens con el valor nuevo

291 user.refresh_from_db(fields=['sessions_epoch'])

292

293 logger.info(f"All sessions revoked for user ID: {user.id} (epoch {user.sessions_epoch})")

294

295 # Indica si un token esta revocado, comparando su generacion con la de la cuenta

296 @staticmethod

297 def token_is_revoked(user, token):

298 """

299 Un token sin el claim `sv` cuenta como generacion 0, que es donde empiezan todas las cuentas

300 """

301 return token.get(SESSION_EPOCH_CLAIM, 0) < user.sessions_epoch

302

303 # Comprueba que una contraseña cumple AUTH_PASSWORD_VALIDATORS

304 @staticmethod

305 def assert_password_is_acceptable(raw_password, user=None):

306 validate_password(raw_password, user=user)

307

308 # Comprueba que el nombre de usuario cumpla el formato admitido

309 @staticmethod

310 def assert_username_is_acceptable(username):

311 """

312 Se impiden los nombres de usuario que no cumplan el formato

313 """

314 if not re.match(USERNAME_REGEX, username or ''):

315 logger.warning("Rejected username with a not allowed format: %s", username)

316 raise InvalidUsernameError(username)

317

318 # Devuelve los dominios admitidos, o lista vacia si se admite cualquiera

319 @staticmethod

320 def allowed_email_domains():

321 """

322 ALLOWED_DOMAINS vale '*' por defecto, que permite cualquier dominio

323 """

324 return [d for d in settings.ALLOWED_DOMAINS if d and d != '*']

325

326 # Comprueba que el correo pertenezca a un dominio admitido

327 @staticmethod

328 def assert_email_domain_is_allowed(email):

329 """

330 Se compara el dominio (lo que va despues de la ultima arroba) con la lista de dominios admitidos

331 """

332 allowed = AuthService.allowed_email_domains()

333 if not allowed:

334 return

335

336 # rsplit y no split, ya que la parte local puede llevar arrobas entrecomilladas

337 # segun el RFC, y el dominio es siempre lo que va detras de la ultima

338 domain = normalize_email(email).rsplit('@', 1)[-1]

339

340 if domain not in allowed:

341 logger.warning("Rejected email outside the allowed domains: %s", email)

342 raise EmailDomainNotAllowedError(email)

343

344 # Comprueba que el nombre de usuario y el correo esten libres para un alta

345 @staticmethod

346 def prepare_registration(username, email):

347 """

348 Si el nombre de usuario lo tiene uns cuenta sin verificar, se descarta para que un registro abandonado

349 no bloquee el nombre para siempre. Si lo tiene una cuenta verificada, se rechaza

350 Lanza InvalidUsernameError, EmailDomainNotAllowedError, UsernameAlreadyTakenError o EmailAlreadyRegisteredError.

351 """

352 # Se comprueba el formato del nombre, antes que nada, porque un nombre con

353 # arroba dejaria la cuenta inaccesible en cuanto se creara

354 AuthService.assert_username_is_acceptable(username)

355

356 # Se comprueba el dominio

357 AuthService.assert_email_domain_is_allowed(email)

358

359 active = Users.objects.filter(username=username, is_deleted=False)

360 unverified = active.filter(is_verify=False)

361

362 if active.exists():

363 if unverified.exists():

364 discarded = unverified.update(

365 is_deleted=True, deleted_at=timezone.now()

366 )

367 logger.info(

368 "Discarded %s unverified registration(s) for username: %s",

369 discarded, username,

370 )

371 else:

372 logger.warning("User with same username: %s, already registered", username)

373 raise UsernameAlreadyTakenError(username)

374

375 if AuthService.email_is_taken(email):

376 logger.warning("User with same email: %s, already registered", email)

377 raise EmailAlreadyRegisteredError(email)

378

379 # Arma los tokens JWT y los datos de usuario que espera la app al entrar

380 @staticmethod

381 def build_auth_payload(user):

382 refresh = RefreshToken.for_user(user)

383

384 # La generacion de sesiones viaja dentro del token, se pone en el refresh porque refresh.access_token

385 # copia los claims del refresh, asi que con ponerlo una vez lo llevan los dos y los que emita el refresh despues

386 refresh[SESSION_EPOCH_CLAIM] = user.sessions_epoch

387

388 return {

389 'refresh': str(refresh),

390 'access': str(refresh.access_token),

391 'user': {

392 'id': str(user.id),

393 'username': user.username,

394 'email': user.email,

395 'name': user.name,

396 'surname1': user.surname1,

397 'surname2': user.surname2,

398 },

399 }

400

401 # Valida el id_token contra Google y devuelve su payload

402 @staticmethod

403 def verify_google_token(id_token_str):

404 try:

405 return id_token.verify_oauth2_token(

406 id_token_str,

407 google_requests.Request(),

408 config('GOOGLE_CLIENT_ID'),

409 )

410 except ValueError as e:

411 logger.warning("Google id_token invalid: %s", str(e))

412 raise InvalidGoogleTokenError(str(e))

413

414 # Saca el correo del payload exigiendo que el proveedor lo de por verificado

415 @staticmethod

416 def verified_email_from_payload(payload):

417 """

418 En este flujo el correo es lo unico que identifica la cuenta, asi que tiene que venir verificado

419 """

420 email = payload.get('email')

421 if not email:

422 raise GoogleEmailMissingError()

423

424 if not payload.get('email_verified', False):

425 logger.warning("Rejected external login: unverified email claim for %s", email)

426 raise UnverifiedEmailError(email)

427 return normalize_email(email)

428

429 # Da por verificada la cuenta a medias que reclama el dueño real del correo

430 @staticmethod

431 def claim_unverified_account(user):

432 """

433 Si la cuenta sin verificar tenia una contraseña, y se verifica con un proveedor externo,

434 se debe anular la contraseña, ya que si no se puede hacer un account pre-hijacking,

435 el dueño del correo entra por Google y se queda con la cuenta, pero la contraseña que habia puesto

436 otro usuario sigue valiendo y puede entrar tambien, por lo que es obligatorio eliminar la contraseña anterior

437 """

438 if user.is_verify:

439 return user

440

441 user.is_verify = True

442

443 # Se debe eliminar la contraseña si la tiene

444 if user.password:

445 logger.warning(

446 "Password cleared for user ID: %s: the account was claimed by the owner "

447 "of its email before it had ever been verified",

448 user.id,

449 )

450 user.set_password(None)

451

452 user.save()

453 return user

454

455 # Localiza la cuenta asociada al correo que devuelve Google

456 @staticmethod

457 def find_google_user(payload):

458 email = AuthService.verified_email_from_payload(payload)

459

460 try:

461 user = Users.objects.get(email=email, is_deleted=False)

462 except Users.DoesNotExist:

463 logger.warning("Google login: user not found with email:%s", email)

464 raise UserNotFoundError(email)

465

466 AuthService.claim_unverified_account(user)

467

468 logger.info("Google login successful: %s", email)

469 return user

470

471 # Da de alta una cuenta a partir del payload de Google

472 @staticmethod

473 def register_google_user(payload, username, user_type_code, surname2=None):

474 """

475 La cuenta nace ya verificada, ya que Google es quien acredita el correo, asi que

476 no tiene sentido volver a pedir un codigo. Y sin contraseña, porque solo se entra por Google

477 """

478 # El alta con Google tampoco pasa por prepare_registration, asi que el

479 # formato del nombre hay que exigirlo tambien aqui: el nombre lo escribe

480 # el usuario en el formulario, Google solo pone el correo

481 AuthService.assert_username_is_acceptable(username)

482

483 email = AuthService.verified_email_from_payload(payload)

484

485 # El alta con Google no pasa por prepare_registration, asi que la regla

486 # del dominio hay que exigirla tambien aqui, es la otra puerta de entrada

487 # y el correo lo pone el proveedor, no el formulario

488 AuthService.assert_email_domain_is_allowed(email)

489

490 if AuthService.email_is_taken(email):

491 raise EmailAlreadyRegisteredError(email)

492

493 if AuthService.username_is_taken(username):

494 raise UsernameAlreadyTakenError(username)

495

496 try:

497 user_type = UserType.objects.get(code=user_type_code, is_deleted=False)

498 except UserType.DoesNotExist:

499 raise InvalidUserTypeError(user_type_code)

500

501 given_name = payload.get('given_name', '')

502 family_name = payload.get('family_name', given_name)

503

504 user = Users(

505 username=username,

506 email=email,

507 name=given_name,

508 surname1=family_name,

509 surname2=surname2,

510 user_type=user_type,

511 is_verify=True,

512 has_2FA=False,

513 )

514 # Como se ha registrado con Google, no se necesita contraseña

515 user.set_password(None)

516 user.save()

517

518 logger.info("Google registration successful: %s", email)

519 return user

Coverage for users\services\auth_service.py: 100.0%

262 statements