from src.services.auth import AuthService, EmailConfirmationClaim, JWT from email.headerregistry import Address from jinja2 import Environment, PackageLoader, select_autoescape from src.services.email import EmailService, EmailDTO from .data import UserDTO, User, UserProfile, SignupError, LoginError from .repo import UserRepo def is_valid_email(email: str) -> bool: try: parsed = Address(addr_spec=email) return bool(parsed.username and parsed.domain and '.' in parsed.domain) except (ValueError, TypeError): return False def is_valid_password(password: str) -> bool: # TODO: Enforce other or saner rules? long_enough = len(password) > 8 short_enough = len(password) < 32 has_symbol = any([s in password for s in list('@#$%^&*!?/')]) return long_enough and short_enough and has_symbol class UserService: def __init__(self, repo: UserRepo, email_service: EmailService, auth_service: AuthService): self._repo = repo self._email_service = email_service self._auth_service = auth_service def login(self, user: UserDTO) -> UserProfile: full_user = self._repo.auth_as_user(user) if full_user: return full_user.to_profile() else: raise LoginError('No user found for that email or password') def confirm_email_for_user(self, user_id: int, confirmation_token: JWT) -> bool: """ Attempt to confirm that the user at the given ID has confirmed their email by returning the JWT that was minted for this purpose. Returns whether or not the confirmation was performed. """ claim = self._auth_service.validate_token(EmailConfirmationClaim, confirmation_token) # First check that the claim could be parsed and that it refers to the expected user if claim and claim.id == user_id: user = self._repo.get_user_by_id(user_id) # Double check that: # 1. The user exists in the database # 2. The claim refers to the email address on file # 3. The email was not already confirmed if user and user.email == claim.email and not user.email_confirmed: user.email_confirmed = True self._repo.update_user(user) return True # In all other cases, the confirmation was not possible so return False return False def _send_confirmation_email(self, user: User): env = Environment(loader=PackageLoader('src'), autoescape=select_autoescape()) text_template = env.get_template('mail/confirmation_email.txt') html_template = env.get_template('mail/confirmation_email.html') token = self._auth_service.mint_claim_from_user(EmailConfirmationClaim, user) # TODO: Parameterize this with configuration that also drives the API url = f'/confirm?token={token}' text_content = text_template.render(confirmation_link=url) html_content = html_template.render(confirmation_link=url) email = EmailDTO(to=user.email, subject='Confirm Your Email Address', text=text_content, html=html_content) self._email_service.send_email(email) def signup(self, user: UserDTO) -> UserProfile: if not is_valid_email(user.email): raise SignupError(f'{user.email} was not an acceptable email address') if not is_valid_password(user.raw_password.expose_secret()): raise SignupError('The given password was not acceptable') # Create a user in persistence created_user = self._repo.create_user(user) # send a confirmation email self._send_confirmation_email(created_user) return created_user.to_profile() def get_user_by_id(self, user_id: int) -> UserProfile | None: user = self._repo.get_user_by_id(user_id) if user: return user.to_profile()