First pass at implementing email update confirmation workflow

Author: chriscarrollsmith

main.py

@@ -33,7 +33,7 @@ async def lifespan(app: FastAPI):
     # Optional shutdown logic
 
 
-app = FastAPI(lifespan=lifespan)
+app: FastAPI = FastAPI(lifespan=lifespan)
 
 # Mount static files (e.g., CSS, JS)
 app.mount("/static", StaticFiles(directory="static"), name="static")

routers/authentication.py

@@ -8,6 +8,7 @@
 from sqlmodel import Session, select
 from utils.models import User, UserPassword
 from utils.auth import (
+    AuthenticationError,
     get_session,
     get_user_from_reset_token,
     create_password_validator,
@@ -18,7 +19,10 @@
     create_access_token,
     create_refresh_token,
     validate_token,
-    send_reset_email
+    send_reset_email,
+    send_email_update_confirmation,
+    get_user_from_email_update_token,
+    get_authenticated_user
 )
 
 logger = getLogger("uvicorn.error")
@@ -102,6 +106,17 @@ async def as_form(
                    new_password=new_password, confirm_new_password=confirm_new_password)
 
 
+class UpdateEmail(BaseModel):
+    new_email: EmailStr
+
+    @classmethod
+    async def as_form(
+        cls,
+        new_email: EmailStr = Form(...)
+    ):
+        return cls(new_email=new_email)
+
+
 # --- DB Request and Response Models ---
 
 
@@ -289,3 +304,72 @@ def logout():
     response.delete_cookie("access_token")
     response.delete_cookie("refresh_token")
     return response
+
+
+@router.post("/update_email")
+async def request_email_update(
+    update: UpdateEmail = Depends(UpdateEmail.as_form),
+    user: User = Depends(get_authenticated_user),
+    session: Session = Depends(get_session)
+):
+    # Check if the new email is already registered
+    existing_user = session.exec(
+        select(User).where(User.email == update.new_email)
+    ).first()
+
+    if existing_user:
+        raise HTTPException(
+            status_code=400,
+            detail="This email is already registered"
+        )
+
+    if not user or not user.id:
+        raise HTTPException(
+            status_code=400,
+            detail="User not found"
+        )
+
+    # Send confirmation email
+    send_email_update_confirmation(
+        current_email=user.email,
+        new_email=update.new_email,
+        user_id=user.id,
+        session=session
+    )
+
+    return RedirectResponse(
+        url="/settings?email_update_requested=true",
+        status_code=303
+    )
+
+
+@router.get("/confirm_email_update")
+async def confirm_email_update(
+    user_id: int,
+    token: str,
+    new_email: str,
+    session: Session = Depends(get_session)
+):
+    user, update_token = get_user_from_email_update_token(
+        user_id, token, session
+    )
+
+    if not user or not update_token:
+        raise AuthenticationError()
+
+    # Get the new email from the most recent unconfirmed token
+    if update_token.is_expired():
+        raise HTTPException(
+            status_code=400,
+            detail="Token has expired"
+        )
+
+    # Update email and mark token as used
+    user.email = new_email
+    update_token.used = True
+    session.commit()
+
+    return RedirectResponse(
+        url="/settings?email_updated=true",
+        status_code=303
+    )

templates/emails/update_email_email.html

@@ -0,0 +1,12 @@
+{% extends "emails/base_email.html" %}
+
+{% block email_title %}Email Update Request{% endblock %}
+
+{% block email_content %}
+    <h1>Confirm Your New Email Address</h1>
+    <p>You have requested to change your email address from {{ current_email }} to {{ new_email }}.</p>
+    <p>Click the link below to confirm this change:</p>
+    <p><a href="{{ confirmation_url }}">Confirm Email Update</a></p>
+    <p>If you did not request this change, please ignore this email.</p>
+    <p>This link will expire in 1 hour.</p>
+{% endblock %}

tests/test_auth.py

@@ -17,7 +17,7 @@
 )
 
 
-def test_convert_python_regex_to_html():
+def test_convert_python_regex_to_html() -> None:
     PYTHON_SPECIAL_CHARS = r"(?=.*[\[\]\\@$!%*?&{}<>.,'#\-_=+\(\):;|~/\^])"
     HTML_EQUIVALENT = r"(?=.*[\[\]\\@$!%*?&\{\}\<\>\.\,\\'#\-_=\+\(\):;\|~\/\^])"
 
@@ -26,14 +26,14 @@ def test_convert_python_regex_to_html():
     assert PYTHON_SPECIAL_CHARS == HTML_EQUIVALENT
 
 
-def test_password_hashing():
+def test_password_hashing() -> None:
     password = "Test123!@#"
     hashed = get_password_hash(password)
     assert verify_password(password, hashed)
     assert not verify_password("wrong_password", hashed)
 
 
-def test_token_creation_and_validation():
+def test_token_creation_and_validation() -> None:
     data = {"sub": "[email protected]"}
 
     # Test access token
@@ -51,21 +51,21 @@ def test_token_creation_and_validation():
     assert decoded["type"] == "refresh"
 
 
-def test_expired_token():
+def test_expired_token() -> None:
     data = {"sub": "[email protected]"}
     expired_delta = timedelta(minutes=-10)
     expired_token = create_access_token(data, expired_delta)
     decoded = validate_token(expired_token, "access")
     assert decoded is None
 
 
-def test_invalid_token_type():
+def test_invalid_token_type() -> None:
     data = {"sub": "[email protected]"}
     access_token = create_access_token(data)
     decoded = validate_token(access_token, "refresh")
     assert decoded is None
 
-def test_password_reset_url_generation():
+def test_password_reset_url_generation() -> None:
     """
     Tests that the password reset URL is correctly formatted and contains
     the required query parameters.
@@ -91,7 +91,7 @@ def test_password_reset_url_generation():
     assert query_params["email"][0] == test_email
     assert query_params["token"][0] == test_token
 
-def test_password_pattern():
+def test_password_pattern() -> None:
     """
     Tests that the password pattern is correctly defined. to require at least
     one uppercase letter, one lowercase letter, one digit, and one special

utils/auth.py

@@ -16,7 +16,7 @@
 from fastapi.templating import Jinja2Templates
 from fastapi import Depends, Cookie, HTTPException, status
 from utils.db import get_session
-from utils.models import User, Role, PasswordResetToken
+from utils.models import User, Role, PasswordResetToken, EmailUpdateToken
 
 load_dotenv()
 resend.api_key = os.environ["RESEND_API_KEY"]
@@ -434,3 +434,86 @@ def get_user_with_relations(
     ).one()
 
     return eager_user
+
+
+def generate_email_update_url(user_id: int, token: str, new_email: str) -> str:
+    """
+    Generates the email update confirmation URL with proper query parameters.
+    """
+    base_url = os.getenv('BASE_URL')
+    return f"{base_url}/auth/confirm_email_update?user_id={user_id}&token={token}&new_email={new_email}"
+
+
+def send_email_update_confirmation(
+    current_email: str,
+    new_email: str,
+    user_id: int,
+    session: Session
+) -> None:
+    # Check for an existing unexpired token
+    existing_token = session.exec(
+        select(EmailUpdateToken)
+        .where(
+            EmailUpdateToken.user_id == user_id,
+            EmailUpdateToken.expires_at > datetime.now(UTC),
+            EmailUpdateToken.used == False
+        )
+    ).first()
+
+    if existing_token:
+        logger.debug("An unexpired email update token already exists for this user.")
+        return
+
+    # Generate a new token
+    token = EmailUpdateToken(user_id=user_id)
+    session.add(token)
+
+    try:
+        confirmation_url = generate_email_update_url(
+            user_id, token.token, new_email)
+
+        # Render the email template
+        template = templates.get_template("emails/update_email.html")
+        html_content = template.render({
+            "confirmation_url": confirmation_url,
+            "current_email": current_email,
+            "new_email": new_email
+        })
+
+        params: resend.Emails.SendParams = {
+            "from": "[email protected]",
+            "to": [current_email],
+            "subject": "Confirm Email Update",
+            "html": html_content,
+        }
+
+        sent_email: resend.Email = resend.Emails.send(params)
+        logger.debug(f"Email update confirmation sent: {sent_email.get('id')}")
+
+        session.commit()
+    except Exception as e:
+        logger.error(f"Failed to send email update confirmation: {e}")
+        session.rollback()
+
+
+def get_user_from_email_update_token(
+    user_id: int,
+    token: str,
+    session: Session
+) -> tuple[Optional[User], Optional[EmailUpdateToken]]:
+    result = session.exec(
+        select(User, EmailUpdateToken)
+        .where(
+            User.id == user_id,
+            EmailUpdateToken.token == token,
+            EmailUpdateToken.expires_at > datetime.now(UTC),
+            EmailUpdateToken.used == False,
+            EmailUpdateToken.user_id == User.id
+        )
+    ).first()
+
+    if not result:
+        return None, None
+
+    user, update_token = result
+    return user, update_token

utils/models.py

@@ -161,6 +161,25 @@ def is_expired(self) -> bool:
         return datetime.now(UTC) > self.expires_at.replace(tzinfo=UTC)
 
 
+class EmailUpdateToken(SQLModel, table=True):
+    id: Optional[int] = Field(default=None, primary_key=True)
+    user_id: Optional[int] = Field(foreign_key="user.id")
+    token: str = Field(default_factory=lambda: str(
+        uuid4()), index=True, unique=True)
+    expires_at: datetime = Field(
+        default_factory=lambda: datetime.now(UTC) + timedelta(hours=1))
+    used: bool = Field(default=False)
+
+    user: Mapped[Optional["User"]] = Relationship(
+        back_populates="email_update_tokens")
+
+    def is_expired(self) -> bool:
+        """
+        Check if the token has expired
+        """
+        return datetime.now(UTC) > self.expires_at.replace(tzinfo=UTC)
+
+
 class UserPassword(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
     user_id: Optional[int] = Field(foreign_key="user.id", unique=True)