Moved password hash to a separate database model, resolved some mypy type linter errors

Author: chriscarrollsmith

docs/customization.qmd

@@ -41,7 +41,7 @@ To run the tests, use these commands:
 The project uses type annotations and mypy for static type checking. To run mypy, use this command from the root directory:
 
 ```bash
-mypy
+mypy .
 ```
 
 We find that mypy is an enormous time-saver, catching many errors early and greatly reducing time spent debugging unit tests. However, note that mypy requires you type annotate every variable, function, and method in your code base, so taking advantage of it requires a lifestyle change!

main.py

@@ -255,14 +255,14 @@ async def read_organization(
     params: dict = Depends(common_authenticated_parameters)
 ):
     # Get the organization only if the user is a member of it
-    organization: Organization = params["user"].organizations.get(org_id)
-    if not organization:
+    org: Organization = params["user"].organizations.get(org_id)
+    if not org:
         raise organization.OrganizationNotFoundError()
 
     # Eagerly load roles and users
-    organization.roles
-    organization.users
-    params["organization"] = organization
+    org.roles
+    org.users
+    params["organization"] = org
 
     return templates.TemplateResponse(params["request"], "users/organization.html", params)
 

routers/authentication.py

@@ -6,7 +6,7 @@
 from fastapi.responses import RedirectResponse
 from pydantic import BaseModel, EmailStr, ConfigDict
 from sqlmodel import Session, select
-from utils.models import User
+from utils.models import User, UserPassword
 from utils.auth import (
     get_session,
     get_user_from_reset_token,
@@ -125,15 +125,19 @@ async def register(
     user: UserRegister = Depends(UserRegister.as_form),
     session: Session = Depends(get_session),
 ) -> RedirectResponse:
+    # Check if the email is already registered
     db_user = session.exec(select(User).where(
         User.email == user.email)).first()
 
     if db_user:
         raise HTTPException(status_code=400, detail="Email already registered")
 
+    # Hash the password
     hashed_password = get_password_hash(user.password)
+
+    # Create the user
     db_user = User(name=user.name, email=user.email,
-                   hashed_password=hashed_password)
+                   password=UserPassword(hashed_password=hashed_password))
     session.add(db_user)
     session.commit()
     session.refresh(db_user)
@@ -155,9 +159,11 @@ async def login(
     user: UserLogin = Depends(UserLogin.as_form),
     session: Session = Depends(get_session),
 ) -> RedirectResponse:
+    # Check if the email is registered
     db_user = session.exec(select(User).where(
         User.email == user.email)).first()
-    if not db_user or not verify_password(user.password, db_user.hashed_password):
+
+    if not db_user or not db_user.password or not verify_password(user.password, db_user.password.hashed_password):
         raise HTTPException(status_code=400, detail="Invalid credentials")
 
     # Create access token
@@ -259,7 +265,17 @@ async def reset_password(
         raise HTTPException(status_code=400, detail="Invalid or expired token")
 
     # Update password and mark token as used
-    authorized_user.hashed_password = get_password_hash(user.new_password)
+    if authorized_user.password:
+        authorized_user.password.hashed_password = get_password_hash(
+            user.new_password
+        )
+    else:
+        logger.warning(
+            "User password not found during password reset; creating new password for user")
+        authorized_user.password = UserPassword(
+            hashed_password=get_password_hash(user.new_password)
+        )
+
     reset_token.used = True
     session.commit()
     session.refresh(authorized_user)

routers/organization.py

@@ -1,11 +1,11 @@
 from logging import getLogger
 from fastapi import APIRouter, Depends, HTTPException, Form
-from fastapi.responses import RedirectResponse, HTMLResponse
+from fastapi.responses import RedirectResponse
 from pydantic import BaseModel, ConfigDict, field_validator
 from sqlmodel import Session, select
 from utils.db import get_session
 from utils.auth import get_authenticated_user, get_user_with_relations
-from utils.models import Organization, User, Role, utc_time, default_roles
+from utils.models import Organization, User, Role, utc_time, default_roles, ValidPermissions
 from datetime import datetime
 
 logger = getLogger("uvicorn.error")
@@ -139,9 +139,11 @@ def update_organization(
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
     # This will raise appropriate exceptions if org doesn't exist or user lacks access
-    organization: Organization = user.organizations.get(org.id)
+    organization: Organization | None = next(
+        (org for org in user.organizations if org.id == org.id), None)
 
-    if not organization or not any(role.permissions.EDIT_ORGANIZATION for role in organization.roles):
+    # Check if user has permission to edit organization
+    if not organization or not user.has_permission(ValidPermissions.EDIT_ORGANIZATION, organization):
         raise InsufficientPermissionsError()
 
     # Check if new name already exists for another organization
@@ -169,7 +171,8 @@ def delete_organization(
     session: Session = Depends(get_session)
 ) -> RedirectResponse:
     # Check if user has permission to delete organization
-    organization: Organization = user.organizations.get(org_id)
+    organization: Organization | None = next(
+        (org for org in user.organizations if org.id == org_id), None)
     if not organization or not any(role.permissions.DELETE_ORGANIZATION for role in organization.roles):
         raise InsufficientPermissionsError()
 

routers/user.py

@@ -63,9 +63,15 @@ async def delete_account(
     current_user: User = Depends(get_authenticated_user),
     session: Session = Depends(get_session)
 ):
+    if not current_user.password:
+        raise HTTPException(
+            status_code=500,
+            detail="User password not found in database; please contact a system administrator"
+        )
+
     if not verify_password(
         user_delete_account.confirm_delete_password,
-        current_user.hashed_password
+        current_user.password.hashed_password
     ):
         raise HTTPException(
             status_code=400,

tests/conftest.py

@@ -1,17 +1,24 @@
 import pytest
 from dotenv import load_dotenv
 from sqlmodel import create_engine, Session, select
+from sqlalchemy import Engine
 from fastapi.testclient import TestClient
 from utils.db import get_connection_url, set_up_db, tear_down_db, get_session
-from utils.models import User, PasswordResetToken, Organization, Role
+from utils.models import User, PasswordResetToken, Organization, Role, UserPassword
 from utils.auth import get_password_hash, create_access_token, create_refresh_token
 from main import app
 
 load_dotenv()
 
 
+# Define a custom exception for test setup errors
+class SetupError(Exception):
+    """Exception raised for errors in the test setup process."""
+    pass
+
+
 @pytest.fixture(scope="session")
-def engine():
+def engine() -> Engine:
     """
     Create a new SQLModel engine for the test database.
     Use an in-memory SQLite database for testing.
@@ -63,7 +70,7 @@ def test_user(session: Session):
     user = User(
         name="Test User",
         email="[email protected]",
-        hashed_password=get_password_hash("Test123!@#")
+        password=UserPassword(hashed_password=get_password_hash("Test123!@#"))
     )
     session.add(user)
     session.commit()

tests/test_authentication.py

@@ -17,7 +17,7 @@
     validate_token,
     generate_password_reset_url
 )
-
+from .conftest import SetupError
 
 # --- Fixture setup ---
 
@@ -28,7 +28,7 @@ def mock_email_response():
     """
     Returns a mock Email response object
     """
-    return resend.Email(id="6229f547-f3f6-4eb8-b0dc-82c1b09121b6")
+    return resend.Email(id="mock_resend_id")
 
 
 @pytest.fixture
@@ -104,7 +104,12 @@ def test_register_endpoint(unauth_client: TestClient, session: Session):
         User.email == "[email protected]")).first()
     assert user is not None
     assert user.name == "New User"
-    assert verify_password("NewPass123!@#", user.hashed_password)
+
+    # Verify password was hashed and matches
+    if not user.password:
+        raise SetupError(
+            "Test setup failed; user.password is None")
+    assert verify_password("NewPass123!@#", user.password.hashed_password)
 
 
 def test_login_endpoint(unauth_client: TestClient, test_user: User):
@@ -200,10 +205,14 @@ def test_password_reset_flow(unauth_client: TestClient, session: Session, test_u
     )
     assert response.status_code == 303
 
+    if not test_user.password:
+        raise SetupError(
+            "Test setup failed; test_user.password is None")
+
     # Verify password was updated and token was marked as used
     session.refresh(test_user)
     session.refresh(reset_token)
-    assert verify_password("NewPass123!@#", test_user.hashed_password)
+    assert verify_password("NewPass123!@#", test_user.password.hashed_password)
     assert reset_token.used
 
 

tests/test_db.py

@@ -1,5 +1,6 @@
 import warnings
 from sqlmodel import Session, select, inspect
+from sqlalchemy import Engine
 from utils.db import (
     get_connection_url,
     assign_permissions_to_role,
@@ -9,7 +10,7 @@
     set_up_db,
 )
 from utils.models import Role, Permission, Organization, RolePermissionLink, ValidPermissions
-from sqlalchemy import Engine
+from .conftest import SetupError
 
 
 def test_get_connection_url():
@@ -43,8 +44,12 @@ def test_create_default_roles(session: Session, test_organization: Organization)
     session.commit()
 
     # Create roles for test organization
-    roles = create_default_roles(session, test_organization.id)
-    session.commit()
+    if test_organization.id is not None:
+        roles = create_default_roles(session, test_organization.id)
+        session.commit()
+    else:
+        raise SetupError(
+            "Test setup failed; test_organization.id is None")
 
     # Verify roles were created
     assert len(roles) == 3  # Owner, Administrator, Member

tests/test_models.py

@@ -1,4 +1,5 @@
-import pytest
+from datetime import timedelta, datetime, UTC
+from typing import Optional
 from sqlmodel import select, Session
 from utils.models import (
     Permission,
@@ -8,9 +9,9 @@
     ValidPermissions,
     User,
     UserRoleLink,
-    PasswordResetToken,
+    PasswordResetToken
 )
-from datetime import timedelta, datetime, UTC
+from .conftest import SetupError
 
 
 def test_permissions_persist_after_role_deletion(session: Session):
@@ -102,10 +103,9 @@ def test_organization_users_property(session: Session, test_user: User, test_org
     session.refresh(test_organization)
 
     # Test the users property
-    users_list = test_organization.users
+    users_list: list[User] = test_organization.users
     assert len(users_list) == 1
-    # users_list is a list of lists due to the property implementation
-    assert test_user in users_list[0]
+    assert test_user in users_list
 
 
 def test_cascade_delete_organization(session: Session, test_user: User, test_organization: Organization):
@@ -185,3 +185,46 @@ def test_password_reset_token_is_expired(session: Session, test_user: User):
     # Verify expiration states
     assert expired_token.is_expired()
     assert not valid_token.is_expired()
+
+
+def test_user_has_permission(session: Session, test_user: User, test_organization: Organization):
+    """
+    Test that User.has_permission method correctly checks if a user has a specific
+    permission for a given organization.
+    """
+    # Create a role with specific permissions in the test organization
+    role = Role(name="Test Role", organization_id=test_organization.id)
+    session.add(role)
+    session.commit()
+    session.refresh(role)
+
+    # Assign permissions to the role
+    delete_org_permission: Optional[Permission] = session.exec(
+        select(Permission).where(Permission.name ==
+                                 ValidPermissions.DELETE_ORGANIZATION)
+    ).first()
+    edit_org_permission: Optional[Permission] = session.exec(
+        select(Permission).where(Permission.name ==
+                                 ValidPermissions.EDIT_ORGANIZATION)
+    ).first()
+
+    if delete_org_permission is not None and edit_org_permission is not None:
+        role.permissions.append(delete_org_permission)
+        role.permissions.append(edit_org_permission)
+    else:
+        raise SetupError(
+            "Test setup failed; permission not found in database")
+    session.commit()
+
+    # Link the user to the role
+    test_user.roles.append(role)
+    session.commit()
+    session.refresh(test_user)
+
+    # Test the has_permission method
+    assert test_user.has_permission(
+        ValidPermissions.DELETE_ORGANIZATION, test_organization) is True
+    assert test_user.has_permission(
+        ValidPermissions.EDIT_ORGANIZATION, test_organization) is True
+    assert test_user.has_permission(
+        ValidPermissions.INVITE_USER, test_organization) is False

utils/db.py

@@ -1,6 +1,6 @@
 import os
 import logging
-from typing import Generator
+from typing import Generator, Union, Sequence
 from dotenv import load_dotenv
 from sqlalchemy.engine import URL
 from sqlmodel import create_engine, Session, SQLModel, select
@@ -57,7 +57,7 @@ def get_session() -> Generator[Session, None, None]:
         yield session
 
 
-def assign_permissions_to_role(session: Session, role: Role, permissions: list[Permission], check_first: bool = False) -> None:
+def assign_permissions_to_role(session: Session, role: Role, permissions: Union[list[Permission], Sequence[Permission]], check_first: bool = False) -> None:
     """
     Assigns permissions to a role in the database.