Fixes naive vs aware datetime comparison crashes on SQLite (#3562)

* Addessed .expires_at naive datetime error

Signed-off-by: cafalchio <mcafalchio@gmail.com>

* fix(tests): improve datetime normalization test coverage

Expand test coverage for the naive vs aware datetime fix:
- Add missing None expires_at path tests for PasswordResetToken, UserRole, EmailApiToken
- Add naive not-expired and aware expired cross-case tests
- Add _lookup_api_token_sync naive datetime tests for auth.py fix
- Fix PEP 8 style: use 'is True/False' instead of '== True/False'

Signed-off-by: Mihai Criveti <crivetimihai@gmail.com>

---------

Signed-off-by: cafalchio <mcafalchio@gmail.com>
Signed-off-by: Mihai Criveti <crivetimihai@gmail.com>
Co-authored-by: Mihai Criveti <crivetimihai@gmail.com>

Author: cafalchio

mcpgateway/auth.py

@@ -453,28 +453,31 @@ def _lookup_api_token_sync(token_hash: str) -> Optional[Dict[str, Any]]:
         result = db.execute(select(EmailApiToken).where(EmailApiToken.token_hash == token_hash, EmailApiToken.is_active.is_(True)))
         api_token = result.scalar_one_or_none()
 
-        if api_token:
-            # Check expiration
-            if api_token.expires_at and api_token.expires_at < datetime.now(timezone.utc):
+        if not api_token:
+            return None
+
+        # Check expiration
+        if api_token.expires_at:
+            expires_at = api_token.expires_at.replace(tzinfo=timezone.utc) if api_token.expires_at.tzinfo is None else api_token.expires_at
+            if utc_now() > expires_at:
                 return {"expired": True}
 
-            # Check revocation
-            # First-Party
-            from mcpgateway.db import TokenRevocation  # pylint: disable=import-outside-toplevel
+        # Check revocation
+        # First-Party
+        from mcpgateway.db import TokenRevocation  # pylint: disable=import-outside-toplevel
 
-            revoke_result = db.execute(select(TokenRevocation).where(TokenRevocation.jti == api_token.jti))
-            if revoke_result.scalar_one_or_none() is not None:
-                return {"revoked": True}
+        revoke_result = db.execute(select(TokenRevocation).where(TokenRevocation.jti == api_token.jti))
+        if revoke_result.scalar_one_or_none() is not None:
+            return {"revoked": True}
 
-            # Update last_used timestamp
-            api_token.last_used = utc_now()
-            db.commit()
+        # Update last_used timestamp
+        api_token.last_used = utc_now()
+        db.commit()
 
-            return {
-                "user_email": api_token.user_email,
-                "jti": api_token.jti,
-            }
-        return None
+        return {
+            "user_email": api_token.user_email,
+            "jti": api_token.jti,
+        }
 
 
 def _get_sync_redis_client():

mcpgateway/db.py

@@ -989,9 +989,13 @@ def is_expired(self) -> bool:
         Returns:
             True if assignment has expired, False otherwise
         """
-        if not self.expires_at:
+        if self.expires_at is None:
             return False
-        return utc_now() > self.expires_at
+        expires_at = self.expires_at
+        if expires_at.tzinfo is None:
+            expires_at = expires_at.replace(tzinfo=timezone.utc)
+
+        return utc_now() > expires_at
 
 
 class PermissionAuditLog(Base):
@@ -1591,7 +1595,13 @@ def is_expired(self) -> bool:
         Returns:
             bool: True when `expires_at` is in the past.
         """
-        return self.expires_at <= utc_now()
+        if self.expires_at is None:
+            return False
+        expires_at = self.expires_at
+        if expires_at.tzinfo is None:
+            expires_at = expires_at.replace(tzinfo=timezone.utc)
+
+        return expires_at <= utc_now()
 
     def is_used(self) -> bool:
         """Return whether the reset token was already consumed.
@@ -5188,7 +5198,10 @@ def is_expired(self) -> bool:
         """
         if not self.expires_at:
             return False
-        return utc_now() > self.expires_at
+        expires_at = self.expires_at
+        if expires_at.tzinfo is None:
+            expires_at = expires_at.replace(tzinfo=timezone.utc)
+        return utc_now() > expires_at
 
     def is_valid(self) -> bool:
         """Check if token is valid (active and not expired).

tests/unit/mcpgateway/test_auth_helpers.py

@@ -272,6 +272,35 @@ def test_lookup_api_token_sync_expired(monkeypatch):
     assert auth._lookup_api_token_sync("hash") == {"expired": True}
 
 
+def test_lookup_api_token_sync_expired_naive_datetime(monkeypatch):
+    """Naive datetime from SQLite is correctly detected as expired."""
+    expired_token = SimpleNamespace(
+        expires_at=datetime(2026, 3, 8, 12, 0, 0),  # naive (no tzinfo)
+        jti="jti-1",
+        user_email="user@example.com",
+        last_used=None,
+    )
+    session = DummySession(results=[expired_token])
+    monkeypatch.setattr(auth, "fresh_db_session", lambda: _session_ctx(session))
+    monkeypatch.setattr("mcpgateway.db.utc_now", lambda: datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc))
+    assert auth._lookup_api_token_sync("hash") == {"expired": True}
+
+
+def test_lookup_api_token_sync_not_expired_naive_datetime(monkeypatch):
+    """Naive datetime from SQLite in the future is correctly detected as not expired."""
+    active_token = SimpleNamespace(
+        expires_at=datetime(2026, 3, 10, 12, 0, 0),  # naive, in the future
+        jti="jti-1",
+        user_email="user@example.com",
+        last_used=None,
+    )
+    session = DummySession(results=[active_token, None])
+    monkeypatch.setattr(auth, "fresh_db_session", lambda: _session_ctx(session))
+    monkeypatch.setattr("mcpgateway.db.utc_now", lambda: datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc))
+    result = auth._lookup_api_token_sync("hash")
+    assert result["user_email"] == "user@example.com"
+
+
 def test_lookup_api_token_sync_revoked(monkeypatch):
     api_token = SimpleNamespace(
         expires_at=None,

tests/unit/mcpgateway/test_db.py

@@ -2851,3 +2851,142 @@ def __init__(self, name: str):
     team_with_slug.slug = "workspace-alice-example-com"
     db.set_email_team_slug(None, None, team_with_slug)
     assert team_with_slug.slug == "workspace-alice-example-com"
+
+
+def test_password_reset_is_expired_naive_expired():
+    """Naive datetime in the past is detected as expired."""
+    with patch("mcpgateway.db.utc_now", return_value=datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc)):
+        token = db.PasswordResetToken(
+            user_email="user@example.com",
+            token_hash="a" * 64,
+            expires_at=datetime(2026, 3, 8, 12, 0, 0),
+        )
+        assert token.is_expired() is True
+
+
+def test_password_reset_is_expired_naive_not_expired():
+    """Naive datetime in the future is detected as not expired."""
+    with patch("mcpgateway.db.utc_now", return_value=datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc)):
+        token = db.PasswordResetToken(
+            user_email="user@example.com",
+            token_hash="a" * 64,
+            expires_at=datetime(2026, 3, 10, 12, 0, 0),
+        )
+        assert token.is_expired() is False
+
+
+def test_password_reset_is_expired_aware():
+    """Aware datetime comparison works without normalization."""
+    with patch("mcpgateway.db.utc_now", return_value=datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc)):
+        token = db.PasswordResetToken(
+            user_email="user@example.com",
+            token_hash="a" * 64,
+            expires_at=datetime(2026, 3, 8, 12, 0, 0, tzinfo=timezone.utc),
+        )
+        assert token.is_expired() is True
+
+
+def test_password_reset_is_expired_none():
+    """None expires_at is treated as not expired."""
+    token = db.PasswordResetToken(
+        user_email="user@example.com",
+        token_hash="a" * 64,
+        expires_at=None,
+    )
+    assert token.is_expired() is False
+
+
+def test_user_role_is_expired_naive_expired():
+    """Naive datetime in the past is detected as expired."""
+    with patch("mcpgateway.db.utc_now", return_value=datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc)):
+        user_role = db.UserRole(
+            user_email="user@example.com",
+            role_id="role_id",
+            scope="scope",
+            granted_by="user1@example.com",
+            granted_at=datetime(2026, 3, 7, 12, 0, 0),
+            expires_at=datetime(2026, 3, 8, 12, 0, 0),
+        )
+        assert user_role.is_expired() is True
+
+
+def test_user_role_is_expired_aware_not_expired():
+    """Aware datetime in the future is detected as not expired."""
+    with patch("mcpgateway.db.utc_now", return_value=datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc)):
+        user_role = db.UserRole(
+            user_email="user@example.com",
+            role_id="role_id",
+            scope="scope",
+            granted_by="user1@example.com",
+            granted_at=datetime(2026, 3, 7, 12, 0, 0),
+            expires_at=datetime(2026, 3, 10, 12, 0, 0, tzinfo=timezone.utc),
+        )
+        assert user_role.is_expired() is False
+
+
+def test_user_role_is_expired_none():
+    """None expires_at means role never expires."""
+    user_role = db.UserRole(
+        user_email="user@example.com",
+        role_id="role_id",
+        scope="scope",
+        granted_by="user1@example.com",
+        granted_at=datetime(2026, 3, 7, 12, 0, 0),
+        expires_at=None,
+    )
+    assert user_role.is_expired() is False
+
+
+def test_email_api_token_is_expired_naive_expired():
+    """Naive datetime in the past is detected as expired."""
+    with patch("mcpgateway.db.utc_now", return_value=datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc)):
+        token = db.EmailApiToken(
+            user_email="alice@example.com",
+            name="Production API Access",
+            server_id="prod-server-123",
+            resource_scopes=["tools.read", "resources.read"],
+            description="Read-only access to production tools",
+            expires_at=datetime(2026, 3, 8, 12, 0, 0),
+        )
+        assert token.is_expired() is True
+
+
+def test_email_api_token_is_expired_naive_not_expired():
+    """Naive datetime in the future is detected as not expired."""
+    with patch("mcpgateway.db.utc_now", return_value=datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc)):
+        token = db.EmailApiToken(
+            user_email="alice@example.com",
+            name="Production API Access",
+            server_id="prod-server-123",
+            resource_scopes=["tools.read", "resources.read"],
+            description="Read-only access to production tools",
+            expires_at=datetime(2026, 3, 13, 12, 0, 0),
+        )
+        assert token.is_expired() is False
+
+
+def test_email_api_token_is_expired_aware_not_expired():
+    """Aware datetime in the future is detected as not expired."""
+    with patch("mcpgateway.db.utc_now", return_value=datetime(2026, 3, 9, 12, 0, 0, tzinfo=timezone.utc)):
+        token = db.EmailApiToken(
+            user_email="alice@example.com",
+            name="Production API Access",
+            server_id="prod-server-123",
+            resource_scopes=["tools.read", "resources.read"],
+            description="Read-only access to production tools",
+            expires_at=datetime(2026, 3, 10, 12, 0, 0, tzinfo=timezone.utc),
+        )
+        assert token.is_expired() is False
+
+
+def test_email_api_token_is_expired_none():
+    """None expires_at means token never expires."""
+    token = db.EmailApiToken(
+        user_email="alice@example.com",
+        name="Production API Access",
+        server_id="prod-server-123",
+        resource_scopes=["tools.read", "resources.read"],
+        description="Read-only access to production tools",
+        expires_at=None,
+    )
+    assert token.is_expired() is False