add invalidate old invites

Author: rafizamankhan

exceptions/http_exceptions.py

@@ -178,10 +178,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="Invitation not found or expired")
+        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=(
+                "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):

routers/core/account.py

@@ -13,7 +13,6 @@
     DataIntegrityError,
     Account,
     AccountEmail,
-    Invitation,
 )
 from utils.core.dependencies import get_session
 from utils.core.models import RefreshToken
@@ -54,14 +53,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,
@@ -149,14 +151,24 @@ async def read_login(
     request: Request,
     _: None = Depends(require_unauthenticated_client),
     invitation_token: Optional[str] = Query(None),
+    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": None,
+            "invitation_token": invitation_token,
+            "invitation_token_warning": invitation_token_warning,
+        },
     )
 
 
@@ -166,10 +178,16 @@ async def read_register(
     _: None = Depends(require_unauthenticated_client),
     email: Optional[EmailStr] = Query(None),
     invitation_token: Optional[str] = Query(None),
+    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",
@@ -178,6 +196,7 @@ async def read_register(
             "password_pattern": HTML_PASSWORD_PATTERN,
             "email": email,
             "invitation_token": invitation_token,
+            "invitation_token_warning": invitation_token_warning,
         },
     )
 
@@ -317,16 +336,7 @@ async def register(
         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()
+        invitation = require_active_invitation_by_token(session, invitation_token)
 
         # Verify email matches
         if email != invitation.invitee_email:
@@ -434,15 +444,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(

routers/core/invitation.py

@@ -15,14 +15,16 @@
 )
 from utils.core.models import User, Role, Account, Invitation, Organization
 from utils.core.enums import ValidPermissions
-from utils.core.invitations import send_invitation_email, process_invitation
+from utils.core.invitations import (
+    send_invitation_email,
+    process_invitation,
+    require_active_invitation_by_token,
+)
 from exceptions.http_exceptions import (
     UserIsAlreadyMemberError,
-    ActiveInvitationExistsError,
     InvalidRoleForOrganizationError,
     OrganizationNotFoundError,
     InvitationEmailSendError,
-    InvalidInvitationTokenError,
 )
 from exceptions.exceptions import EmailSendFailedError
 from utils.core.htmx import is_htmx_request, append_toast
@@ -49,11 +51,7 @@ def get_valid_invitation(
     token: str = Query(...), session: Session = Depends(get_session)
 ) -> Invitation:
     """Dependency to retrieve a valid, active invitation based on the token."""
-    statement = select(Invitation).where(Invitation.token == token)
-    invitation = session.exec(statement).first()
-    if not invitation or not invitation.is_active():
-        raise InvalidInvitationTokenError()
-    return invitation
+    return require_active_invitation_by_token(session, token)
 
 
 @router.post("/", name="create_invitation")
@@ -108,12 +106,9 @@ async def create_invitation(
             ):
                 raise UserIsAlreadyMemberError()
 
-    # Check for active invitations with the same email
-    active_invitations = Invitation.get_active_for_org(session, organization_id)
-    if any(
-        invitation.invitee_email == invitee_email for invitation in active_invitations
-    ):
-        raise ActiveInvitationExistsError()
+    # Replace any pending invite for this email (active or expired) before creating a new one.
+    Invitation.invalidate_pending_for_email(session, organization_id, invitee_email)
+    session.flush()
 
     # Create the invitation
     token = str(uuid4())
@@ -147,7 +142,7 @@ async def create_invitation(
         logger.error(
             f"Invitation email failed for {invitee_email} in org {organization_id}: {e}"
         )
-        session.rollback()  # Rollback the invitation creation
+        session.rollback()  # Rollback invalidation and invitation creation
         raise InvitationEmailSendError()  # Raise HTTP 500
     except Exception as e:
         # Catch any other unexpected errors during flush/refresh/email/commit

templates/account/login.html

@@ -6,6 +6,15 @@
 
 {% block auth_content %}
 <div class="login-form">
+    {% if invitation_token_warning == 'expired' %}
+    <div class="alert alert-warning" role="alert">
+        This invitation link has expired. Ask your organization administrator to send a new invitation, then use the link in the latest email.
+    </div>
+    {% elif invitation_token_warning == 'invalid' %}
+    <div class="alert alert-warning" role="alert">
+        This invitation link is no longer valid. If you were invited again recently, use the link from the most recent invitation email.
+    </div>
+    {% endif %}
     <form method="POST" action="{{ url_for('login') }}"
           hx-post="{{ url_for('login') }}" hx-swap="none"
           class="needs-validation" novalidate>

templates/account/register.html

@@ -6,6 +6,15 @@
 
 {% block auth_content %}
 <div class="register-form">
+    {% if invitation_token_warning == 'expired' %}
+    <div class="alert alert-warning" role="alert">
+        This invitation link has expired. Ask your organization administrator to send a new invitation, then use the link in the latest email.
+    </div>
+    {% elif invitation_token_warning == 'invalid' %}
+    <div class="alert alert-warning" role="alert">
+        This invitation link is no longer valid. If you were invited again recently, use the link from the most recent invitation email.
+    </div>
+    {% endif %}
     <form method="POST" action="{{ url_for('register') }}"
           hx-post="{{ url_for('register') }}" hx-swap="none"
           class="needs-validation" novalidate>

tests/routers/core/test_invitation.py

@@ -1,7 +1,7 @@
 import pytest
 from datetime import datetime, timedelta, UTC
 from unittest.mock import MagicMock
-from sqlmodel import Session, select
+from sqlmodel import Session, select, col
 from tests.conftest import SetupError
 from utils.core.models import Role, Permission, User, Invitation, Organization, Account
 from utils.core.enums import ValidPermissions
@@ -498,24 +498,124 @@ def test_create_invitation_for_existing_member(
     )  # Conflict - UserIsAlreadyMemberError
 
 
-def test_create_invitation_duplicate_active(
-    auth_client, inviter_user: User, existing_invitation: Invitation
+def test_create_invitation_resend_replaces_active_invitation(
+    auth_client,
+    inviter_user: User,
+    existing_invitation: Invitation,
+    session: Session,
+    mock_resend_send,
 ):
-    """Test that creating a duplicate active invitation fails."""
+    """Re-inviting the same email replaces the pending invite and invalidates the old token."""
     assert existing_invitation.organization_id is not None
     assert existing_invitation.role_id is not None
+    old_token = existing_invitation.token
+    old_id = existing_invitation.id
+    invitee_email = existing_invitation.invitee_email
+    organization_id = existing_invitation.organization_id
+
     response = auth_client.post(
         app.url_path_for("create_invitation"),
         data={
-            "invitee_email": existing_invitation.invitee_email,  # Same email
+            "invitee_email": invitee_email,
+            "role_id": str(existing_invitation.role_id),
+            "organization_id": str(organization_id),
+        },
+        follow_redirects=False,
+    )
+
+    assert response.status_code == 303, response.text
+
+    session.expire_all()
+    assert session.get(Invitation, old_id) is None
+
+    new_invitation = session.exec(
+        select(Invitation).where(
+            Invitation.invitee_email == invitee_email,
+            Invitation.organization_id == organization_id,
+            col(Invitation.used).is_(False),
+        )
+    ).first()
+    assert new_invitation is not None
+    assert new_invitation.token != old_token
+
+    accept_response = auth_client.get(
+        app.url_path_for("accept_invitation"),
+        params={"token": old_token},
+        follow_redirects=False,
+    )
+    assert accept_response.status_code == 404
+    assert "no longer valid" in accept_response.text.lower()
+
+
+def test_create_invitation_resend_after_expired_pending_invite(
+    auth_client,
+    inviter_user: User,
+    expired_invitation: Invitation,
+    session: Session,
+    mock_resend_send,
+):
+    """Re-inviting after expiry succeeds even though the old row remains used=False in DB."""
+    assert expired_invitation.organization_id is not None
+    assert expired_invitation.role_id is not None
+    expired_id = expired_invitation.id
+    invitee_email = expired_invitation.invitee_email
+    organization_id = expired_invitation.organization_id
+    old_token = expired_invitation.token
+
+    response = auth_client.post(
+        app.url_path_for("create_invitation"),
+        data={
+            "invitee_email": invitee_email,
+            "role_id": str(expired_invitation.role_id),
+            "organization_id": str(organization_id),
+        },
+        follow_redirects=False,
+    )
+
+    assert response.status_code == 303, response.text
+    session.expire_all()
+    assert session.get(Invitation, expired_id) is None
+
+    replacement = session.exec(
+        select(Invitation).where(
+            Invitation.invitee_email == invitee_email,
+            Invitation.organization_id == organization_id,
+            col(Invitation.used).is_(False),
+        )
+    ).first()
+    assert replacement is not None
+    assert replacement.token != old_token
+
+
+def test_create_invitation_resend_email_failure_restores_old_invite(
+    auth_client,
+    inviter_user: User,
+    existing_invitation: Invitation,
+    session: Session,
+    mock_resend_send,
+):
+    """Failed resend rolls back both delete and new invite creation."""
+    assert existing_invitation.id is not None
+    old_token = existing_invitation.token
+    mock_resend_send.side_effect = Exception("Simulated email send failure")
+
+    response = auth_client.post(
+        app.url_path_for("create_invitation"),
+        data={
+            "invitee_email": existing_invitation.invitee_email,
             "role_id": str(existing_invitation.role_id),
             "organization_id": str(existing_invitation.organization_id),
         },
+        follow_redirects=False,
     )
 
-    assert response.status_code == 409, (
-        f"Expected 409 Conflict, got {response.status_code}. Response: {response.text}"
-    )  # Conflict - ActiveInvitationExistsError
+    assert response.status_code == 500
+    session.expire_all()
+    restored = session.exec(
+        select(Invitation).where(Invitation.token == old_token)
+    ).first()
+    assert restored is not None
+    assert restored.id == existing_invitation.id
 
 
 def test_create_invitation_role_not_found(