Merge remote-tracking branch 'origin/main' into modal

Author: github-actions[bot]

exceptions/http_exceptions.py

@@ -146,16 +146,6 @@ def __init__(self):
         )
 
 
-class ActiveInvitationExistsError(HTTPException):
-    """Raised when trying to invite a user for whom an active invitation already exists."""
-
-    def __init__(self):
-        super().__init__(
-            status_code=409,
-            detail="An active invitation already exists for this email address in this organization.",
-        )
-
-
 class InvalidRoleForOrganizationError(HTTPException):
     """Raised when a role provided does not belong to the target organization.
     Note: If the role ID simply doesn't exist, a standard 404 RoleNotFoundError should be raised.
@@ -186,10 +176,31 @@ def __init__(self):
 
 
 class InvalidInvitationTokenError(HTTPException):
-    """Raised when an invitation token is invalid, expired, or not found."""
+    """Raised when an invitation token is missing, superseded, or already used."""
+
+    def __init__(self):
+        super().__init__(
+            status_code=404,
+            detail=(
+                "This invitation link is no longer valid. "
+                "If you were invited again recently, use the link from the "
+                "most recent invitation email."
+            ),
+        )
+
+
+class ExpiredInvitationTokenError(HTTPException):
+    """Raised when an invitation token exists but has passed its expiry date."""
 
     def __init__(self):
-        super().__init__(status_code=404, detail="Invitation not found or expired")
+        super().__init__(
+            status_code=404,
+            detail=(
+                "This invitation link has expired. Ask your organization "
+                "administrator to send a new invitation, then use the link in "
+                "the latest email."
+            ),
+        )
 
 
 class InvitationEmailMismatchError(HTTPException):

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "fastapi-jinja2-postgres-webapp"
-version = "0.1.25"
+version = "0.1.26"
 description = "A template webapp with a pure-Python FastAPI backend, frontend templating with Jinja2, and a Postgres database to power user auth"
 readme = "README.md"
 package-mode = false

routers/core/account.py

@@ -45,6 +45,7 @@
     get_account_from_recovery_token,
     get_account_from_credentials,
     require_unauthenticated_client,
+    require_unauthenticated_unless_invitation_warning,
     get_verified_account,
 )
 from exceptions.http_exceptions import (
@@ -54,14 +55,17 @@
     EmailNotVerifiedError,
     MaxEmailsReachedError,
     PasswordValidationError,
-    InvalidInvitationTokenError,
     InvitationEmailMismatchError,
     InvitationProcessingError,
 )
 from routers.core.dashboard import router as dashboard_router
 from routers.core.user import router as user_router
 from routers.core.organization import router as org_router
-from utils.core.invitations import process_invitation
+from utils.core.invitations import (
+    process_invitation,
+    require_active_invitation_by_token,
+    get_invitation_token_warning,
+)
 from utils.core.rate_limit import (
     check_login_ip_rate_limit,
     check_login_email_rate_limit,
@@ -147,37 +151,56 @@ def logout(
 @router.get("/login")
 async def read_login(
     request: Request,
-    _: None = Depends(require_unauthenticated_client),
+    _: None = Depends(require_unauthenticated_unless_invitation_warning),
     invitation_token: Optional[str] = Query(None),
+    user: Optional[User] = Depends(get_optional_user),
+    session: Session = Depends(get_session),
 ):
     """
     Render login page or redirect to dashboard if already logged in.
     """
+    invitation_token_warning = (
+        get_invitation_token_warning(session, invitation_token)
+        if invitation_token
+        else None
+    )
     return templates.TemplateResponse(
         request,
         "account/login.html",
-        {"user": None, "invitation_token": invitation_token},
+        {
+            "user": user,
+            "invitation_token": invitation_token,
+            "invitation_token_warning": invitation_token_warning,
+        },
     )
 
 
 @router.get("/register")
 async def read_register(
     request: Request,
-    _: None = Depends(require_unauthenticated_client),
+    _: None = Depends(require_unauthenticated_unless_invitation_warning),
     email: Optional[EmailStr] = Query(None),
     invitation_token: Optional[str] = Query(None),
+    user: Optional[User] = Depends(get_optional_user),
+    session: Session = Depends(get_session),
 ):
     """
     Render registration page or redirect to dashboard if already logged in.
     """
+    invitation_token_warning = (
+        get_invitation_token_warning(session, invitation_token)
+        if invitation_token
+        else None
+    )
     return templates.TemplateResponse(
         request,
         "account/register.html",
         {
-            "user": None,
+            "user": user,
             "password_pattern": HTML_PASSWORD_PATTERN,
             "email": email,
             "invitation_token": invitation_token,
+            "invitation_token_warning": invitation_token_warning,
         },
     )
 
@@ -270,6 +293,18 @@ async def register(
     """
     Register a new user account, optionally processing an invitation.
     """
+    pending_invitation: Optional[Invitation] = None
+    if invitation_token:
+        pending_invitation = require_active_invitation_by_token(
+            session, invitation_token
+        )
+        if email != pending_invitation.invitee_email:
+            logger.warning(
+                f"Invitation email mismatch for token {invitation_token} during registration. "
+                f"Account: {email}, Invitation: {pending_invitation.invitee_email}"
+            )
+            raise InvitationEmailMismatchError()
+
     # Check if the email is already registered
     existing_account: Optional[Account] = session.exec(
         select(Account).where(Account.email == email)
@@ -313,46 +348,27 @@ async def register(
     redirect_url = dashboard_router.url_path_for("read_dashboard")
 
     # Process invitation if token is provided (BEFORE final commit)
-    if invitation_token:
+    if pending_invitation:
         logger.info(
             f"Registration attempt with invitation token: {invitation_token} for email {email}"
         )
-        # Fetch the invitation
-        statement = select(Invitation).where(Invitation.token == invitation_token)
-        invitation = session.exec(statement).first()
-
-        if not invitation or not invitation.is_active():
-            logger.warning(
-                f"Invalid or inactive invitation token provided during registration: {invitation_token}"
-            )
-            # Consider raising a more generic error to avoid exposing token validity
-            raise InvalidInvitationTokenError()
-
-        # Verify email matches
-        if email != invitation.invitee_email:
-            logger.warning(
-                f"Invitation email mismatch for token {invitation_token} during registration. "
-                f"Account: {email}, Invitation: {invitation.invitee_email}"
-            )
-            # Consider raising a more generic error to avoid confirming email existence
-            raise InvitationEmailMismatchError()
 
         # Process the invitation (adds changes to the session)
         try:
             logger.info(
-                f"Processing invitation {invitation.id} for new user {new_user.name} ({email}) during registration."
+                f"Processing invitation {pending_invitation.id} for new user {new_user.name} ({email}) during registration."
             )
-            process_invitation(invitation, new_user, session)
+            process_invitation(pending_invitation, new_user, session)
             # Set redirect to the organization page
             redirect_url = org_router.url_path_for(
-                "read_organization", org_id=invitation.organization_id
+                "read_organization", org_id=pending_invitation.organization_id
             )
             logger.info(
-                f"Redirecting new user {new_user.name} to organization {invitation.organization_id} after accepting invitation {invitation.id}."
+                f"Redirecting new user {new_user.name} to organization {pending_invitation.organization_id} after accepting invitation {pending_invitation.id}."
             )
         except Exception as e:
             logger.error(
-                f"Error processing invitation {invitation.id} for new user {new_user.name} ({email}) during registration: {e}",
+                f"Error processing invitation {pending_invitation.id} for new user {new_user.name} ({email}) during registration: {e}",
                 exc_info=True,
             )
             session.rollback()
@@ -434,15 +450,7 @@ async def login(
         logger.info(
             f"Login attempt with invitation token: {invitation_token} for account {account.email}"
         )
-        # Fetch the invitation
-        statement = select(Invitation).where(Invitation.token == invitation_token)
-        invitation = session.exec(statement).first()
-
-        if not invitation or not invitation.is_active():
-            logger.warning(
-                f"Invalid or inactive invitation token provided during login: {invitation_token}"
-            )
-            raise InvalidInvitationTokenError()
+        invitation = require_active_invitation_by_token(session, invitation_token)
 
         # Verify email matches (check primary and any verified secondary emails)
         account_emails = session.exec(