utils: check token status before authenticating user

Before this commit, REANA was only checking that the user had at least
one valid token during the authentication process, without checking
whether the provided token was valid or revoked. This meant that users
were able to use revoked tokens, as long as they had at least one other
active token.

This commit fixes this issue by checking that the provided token is
active before proceeding with the authentication process.

CHANGES.rst

@@ -9,6 +9,7 @@ Version 0.9.1 (UNRELEASED)
 - Adds ``interactive-session-cleanup`` command that can be used by REANA administrators to close interactive sessions that are inactive for more than the specified number of days.
 - Changes the system status report to simplify and clarify the disk usage summary.
 - Fixes GitLab integration to automatically redirect the user to the correct URL when the access request is accepted.
+- Fixes authentication flow to correctly deny access to past revoked tokens in case the same user has also other new active tokens.
 - Adds logic to support SSO with third-party Keycloak authentication services to ``config.py``.
 
 Version 0.9.0 (2023-01-19)

reana_server/utils.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 #
 # This file is part of REANA.
-# Copyright (C) 2018, 2019, 2020, 2021, 2022 CERN.
+# Copyright (C) 2018, 2019, 2020, 2021, 2022, 2023 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -42,6 +42,7 @@
     RunStatus,
     User,
     UserResource,
+    UserToken,
     UserTokenStatus,
     UserTokenType,
     Workflow,
@@ -226,16 +227,14 @@ def filter_input_files(workspace: Union[str, pathlib.Path], reana_spec: Dict) ->
 
 def get_user_from_token(access_token):
     """Validate that the token provided is valid."""
-    user = (
-        Session.query(User)
-        .join(User.tokens)
-        .filter_by(token=access_token, type_=UserTokenType.reana)
+    user_token = UserToken.query.filter_by(
+        token=access_token, type_=UserTokenType.reana
     ).one_or_none()
-    if not user:
+    if not user_token:
         raise ValueError("Token not valid.")
-    if user.access_token_status == UserTokenStatus.revoked.name:
+    if user_token.status == UserTokenStatus.revoked:
         raise ValueError("User access token revoked.")
-    return user
+    return user_token.user_
 
 
 def publish_workflow_submission(workflow, user_id, parameters):

tests/test_utils.py

@@ -1,5 +1,5 @@
 # This file is part of REANA.
-# Copyright (C) 2021, 2022 CERN.
+# Copyright (C) 2021, 2022, 2023 CERN.
 #
 # REANA is free software; you can redistribute it and/or modify it
 # under the terms of the MIT License; see LICENSE file for more details.
@@ -10,7 +10,8 @@
 import pytest
 
 from reana_commons.errors import REANAValidationError
-from reana_server.utils import is_valid_email, filter_input_files
+from reana_db.models import UserToken, UserTokenStatus, UserTokenType
+from reana_server.utils import is_valid_email, filter_input_files, get_user_from_token
 
 
 @pytest.mark.parametrize(
@@ -46,3 +47,37 @@ def test_filter_input_files(tmp_path: pathlib.Path):
     assert (tmp_path / "x/w/c.txt").exists()
     assert not (tmp_path / "x/y/b.txt").exists()
     assert len(list(tmp_path.iterdir())) == 1
+
+
+def test_get_user_from_token(default_user):
+    """Test getting user from his own token."""
+    assert default_user.id_ == get_user_from_token(default_user.access_token).id_
+
+
+def test_get_user_from_token_after_revocation(default_user, session):
+    """Test getting user from revoked token."""
+    token = default_user.active_token
+    token.status = UserTokenStatus.revoked
+    session.commit()
+    with pytest.raises(ValueError, match="revoked"):
+        get_user_from_token(token.token)
+
+
+def test_get_user_from_token_two_tokens(default_user, session):
+    """Test getting user with multiple tokens."""
+    old_token = default_user.active_token
+    old_token.status = UserTokenStatus.revoked
+    new_token = UserToken(
+        token="new_token",
+        user_id=default_user.id_,
+        type_=UserTokenType.reana,
+        status=UserTokenStatus.active,
+    )
+    session.add(new_token)
+    session.commit()
+
+    # Check that new token works
+    assert default_user.id_ == get_user_from_token(new_token.token).id_
+    # Check that old revoked token does not work
+    with pytest.raises(ValueError, match="revoked"):
+        get_user_from_token(old_token.token)