Don't return users with deleted column set to True

Author: chriscarrollsmith

routers/authentication.py

@@ -1,5 +1,5 @@
 # auth.py
-from logging import getLogger
+from logging import getLogger, DEBUG
 from typing import Optional
 from datetime import datetime
 from fastapi import APIRouter, Depends, HTTPException, BackgroundTasks, Form
@@ -22,6 +22,7 @@
 )
 
 logger = getLogger("uvicorn.error")
+logger.setLevel(DEBUG)
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
@@ -126,7 +127,9 @@ async def register(
     session: Session = Depends(get_session),
 ) -> RedirectResponse:
     db_user = session.exec(select(User).where(
-        User.email == user.email)).first()
+        User.email == user.email,
+        User.deleted == False
+    )).first()
 
     if db_user:
         raise HTTPException(status_code=400, detail="Email already registered")
@@ -156,7 +159,9 @@ async def login(
     session: Session = Depends(get_session),
 ) -> RedirectResponse:
     db_user = session.exec(select(User).where(
-        User.email == user.email)).first()
+        User.email == user.email,
+        User.deleted == False
+    )).first()
     if not db_user or not verify_password(user.password, db_user.hashed_password):
         raise HTTPException(status_code=400, detail="Invalid credentials")
 
@@ -205,7 +210,9 @@ async def refresh_token(
 
     user_email = decoded_token.get("sub")
     db_user = session.exec(select(User).where(
-        User.email == user_email)).first()
+        User.email == user_email,
+        User.deleted == False
+    )).first()
     if not db_user:
         return RedirectResponse(url="/login", status_code=303)
 
@@ -239,7 +246,9 @@ async def forgot_password(
     session: Session = Depends(get_session)
 ):
     db_user = session.exec(select(User).where(
-        User.email == user.email)).first()
+        User.email == user.email,
+        User.deleted == False
+    )).first()
 
     if db_user:
         background_tasks.add_task(send_reset_email, user.email, session)
@@ -255,8 +264,13 @@ async def reset_password(
     authorized_user, reset_token = get_user_from_reset_token(
         user.email, user.token, session)
 
-    if not authorized_user or not reset_token:
+    logger.debug(f"authorized_user: {authorized_user}")
+    logger.debug(f"reset_token: {reset_token}")
+
+    if not reset_token:
         raise HTTPException(status_code=400, detail="Invalid or expired token")
+    elif not authorized_user:
+        raise HTTPException(status_code=400, detail="User not found")
 
     # Update password and mark token as used
     authorized_user.hashed_password = get_password_hash(user.new_password)

routers/user.py

@@ -37,6 +37,22 @@ async def as_form(
         return cls(confirm_delete_password=confirm_delete_password)
 
 
+class UpdateProfile(BaseModel):
+    """Request model for updating user profile information"""
+    name: str
+    email: EmailStr
+    avatar_url: str
+
+    @classmethod
+    async def as_form(
+        cls,
+        name: str = Form(...),
+        email: EmailStr = Form(...),
+        avatar_url: str = Form(...),
+    ):
+        return cls(name=name, email=email, avatar_url=avatar_url)
+
+
 # -- Routes --
 
 
@@ -48,37 +64,38 @@ async def view_profile(
     return {"user": current_user}
 
 
-@router.post("/edit_profile", response_class=RedirectResponse)
-async def edit_profile(
-    name: str = Form(...),
-    email: str = Form(...),
-    avatar_url: str = Form(...),
+@router.post("/update_profile", response_class=RedirectResponse)
+async def update_profile(
+    profile_update: UpdateProfile = Depends(UpdateProfile.as_form),
     current_user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ):
     # Update user details
-    current_user.name = name
-    current_user.email = email
-    current_user.avatar_url = avatar_url
+    current_user.name = profile_update.name
+    current_user.email = profile_update.email
+    current_user.avatar_url = profile_update.avatar_url
     session.commit()
     session.refresh(current_user)
     return RedirectResponse(url="/profile", status_code=303)
 
 
 @router.post("/delete_account", response_class=RedirectResponse)
 async def delete_account(
-    confirm_delete_password: str = Form(...),
+    user_delete_account: UserDeleteAccount = Depends(
+        UserDeleteAccount.as_form),
     current_user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ):
-    if not verify_password(confirm_delete_password, current_user.hashed_password):
+    if not verify_password(user_delete_account.confirm_delete_password, current_user.hashed_password):
         raise HTTPException(status_code=400, detail="Password is incorrect")
 
     # Mark the user as deleted
     current_user.deleted = True
     session.commit()
-    #Logs Out
-    router.get("/logout", response_class=RedirectResponse)
-    # Deletes user
-    session.delete(current_user)
-    return RedirectResponse(url="/", status_code=303)
+
+    # Delete the user's access and refresh tokens to force a logout
+    response = RedirectResponse(url="/", status_code=303)
+    response.delete_cookie("access_token")
+    response.delete_cookie("refresh_token")
+
+    return response

templates/users/profile.html

@@ -34,7 +34,7 @@ <h1 class="mb-4">User Profile</h1>
             Edit Profile
         </div>
         <div class="card-body">
-            <form action="{{ url_for('edit_profile') }}" method="post">
+            <form action="{{ url_for('update_profile') }}" method="post">
                 <div class="mb-3">
                     <label for="name" class="form-label">Name</label>
                     <input type="text" class="form-control" id="name" name="name" value="{{ user.name }}">

tests/test_authentication.py

@@ -261,6 +261,10 @@ def test_logout_endpoint(client: TestClient):
 
 
 def test_register_with_existing_email(client: TestClient, test_user: User):
+    """Test that registration fails with an existing non-deleted user's email"""
+    # Ensure test user is not deleted
+    assert not test_user.deleted
+
     response = client.post(
         "/auth/register",
         data={
@@ -273,6 +277,34 @@ def test_register_with_existing_email(client: TestClient, test_user: User):
     assert response.status_code == 400
 
 
+def test_register_with_deleted_user_email(client: TestClient, test_user: User, session: Session):
+    """Test that registration succeeds with a deleted user's email"""
+    # Mark test user as deleted
+    test_user.deleted = True
+    session.add(test_user)
+    session.commit()
+
+    response = client.post(
+        "/auth/register",
+        data={
+            "name": "New User",
+            "email": test_user.email,
+            "password": "Test123!@#",
+            "confirm_password": "Test123!@#"
+        },
+        follow_redirects=False
+    )
+    assert response.status_code == 303
+
+    # Verify new user was created
+    new_user = session.exec(select(User).where(
+        User.email == test_user.email,
+        User.deleted == False
+    )).first()
+    assert new_user is not None
+    assert new_user.id != test_user.id
+
+
 def test_login_with_invalid_credentials(client: TestClient, test_user: User):
     response = client.post(
         "/auth/login",
@@ -361,3 +393,74 @@ def test_password_reset_email_url(client: TestClient, session: Session, test_use
     assert parsed.path == str(reset_password_path)
     assert query_params["email"][0] == test_user.email
     assert query_params["token"][0] == reset_token.token
+
+
+def test_deleted_user_cannot_login(client: TestClient, test_user: User, session: Session):
+    """Test that a deleted user cannot log in"""
+    # First mark the user as deleted
+    test_user.deleted = True
+    session.add(test_user)
+    session.commit()
+
+    response = client.post(
+        "/auth/login",
+        data={
+            "email": test_user.email,
+            "password": "Test123!@#"
+        }
+    )
+    assert response.status_code == 400
+
+
+def test_deleted_user_cannot_use_tokens(client: TestClient, test_user: User, session: Session):
+    """Test that a deleted user's tokens become invalid"""
+    # Create tokens before marking user as deleted
+    access_token = create_access_token({"sub": test_user.email})
+    refresh_token = create_refresh_token({"sub": test_user.email})
+
+    # Mark user as deleted
+    test_user.deleted = True
+    session.add(test_user)
+    session.commit()
+
+    # Set tokens in cookies
+    client.cookies.set("access_token", access_token)
+    client.cookies.set("refresh_token", refresh_token)
+
+    # Try to refresh tokens
+    response = client.post("/auth/refresh", follow_redirects=False)
+    assert response.status_code == 303  # user is redirected to login
+
+
+def test_deleted_user_cannot_use_reset_token(client: TestClient, session: Session, test_user: User):
+    """Test that a deleted user cannot use a previously issued reset token"""
+    # First create a reset token
+    response = client.post(
+        "/auth/forgot_password",
+        data={"email": test_user.email},
+        follow_redirects=False
+    )
+    assert response.status_code == 303
+
+    # Get the reset token
+    reset_token = session.exec(select(PasswordResetToken)
+                               .where(PasswordResetToken.user_id == test_user.id)).first()
+    assert reset_token is not None
+
+    # Now mark user as deleted
+    test_user.deleted = True
+    session.add(test_user)
+    session.commit()
+
+    # Try to use the reset token
+    response = client.post(
+        "/auth/reset_password",
+        data={
+            "email": test_user.email,
+            "token": reset_token.token,
+            "new_password": "NewPass123!@#",
+            "confirm_new_password": "NewPass123!@#"
+        },
+        follow_redirects=False
+    )
+    assert response.status_code == 400

utils/auth.py

@@ -180,7 +180,9 @@ def validate_token_and_get_user(
     if decoded_token:
         user_email = decoded_token.get("sub")
         user = session.exec(select(User).where(
-            User.email == user_email)).first()
+            User.email == user_email,
+            User.deleted == False
+        )).first()
         if user:
             if token_type == "refresh":
                 new_access_token = create_access_token(
@@ -275,7 +277,10 @@ def generate_password_reset_url(email: str, token: str) -> str:
 
 def send_reset_email(email: str, session: Session):
     # Check for an existing unexpired token
-    user = session.exec(select(User).where(User.email == email)).first()
+    user = session.exec(select(User).where(
+        User.email == email,
+        User.deleted == False
+    )).first()
     if user:
         existing_token = session.exec(
             select(PasswordResetToken)
@@ -327,7 +332,11 @@ def get_user_from_reset_token(email: str, token: str, session: Session) -> tuple
 
     user = session.exec(select(User).where(
         User.email == email,
-        User.id == reset_token.user_id
+        User.id == reset_token.user_id,
+        User.deleted == False
     )).first()
 
+    if not user:
+        return None, None
+
     return user, reset_token

utils/models.py

@@ -3,7 +3,7 @@
 from datetime import datetime, UTC, timedelta
 from typing import Optional, List
 from sqlmodel import SQLModel, Field, Relationship
-from sqlalchemy import Column, Enum as SQLAlchemyEnum
+from sqlalchemy import Column, Enum as SQLAlchemyEnum, Index
 
 
 def utc_time():
@@ -89,7 +89,7 @@ class PasswordResetToken(SQLModel, table=True):
 class User(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
     name: str
-    email: str = Field(index=True, unique=True)
+    email: str = Field(index=True)
     hashed_password: str
     avatar_url: Optional[str] = None
     organization_id: Optional[int] = Field(
@@ -105,6 +105,15 @@ class User(SQLModel, table=True):
     password_reset_tokens: List["PasswordResetToken"] = Relationship(
         back_populates="user")
 
+    __table_args__ = (
+        Index(
+            'ix_user_email_unique_active',
+            'email',
+            unique=True,
+            postgresql_where=(deleted == False)
+        ),
+    )
+
 
 class UserOrganizationLink(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)