admin: standardize --admin-access-token usage

closes #489

reana_server/decorators.py

@@ -12,6 +12,7 @@
 import logging
 import os
 import traceback
+import sys
 
 import click
 from flask import jsonify, request
@@ -20,6 +21,7 @@
 
 from reana_server.utils import (
     _get_user_from_invenio_user,
+    _validate_admin_access_token,
     get_user_from_token,
     get_quota_excess_message,
 )
@@ -36,6 +38,14 @@ def admin_access_token_option(func):
     )
     @functools.wraps(func)
     def wrapper(*args, **kwargs):
+        try:
+            _validate_admin_access_token(kwargs.get("admin_access_token"))
+        except ValueError as e:
+            click.echo(
+                click.style(str(e), fg="red"),
+                err=True,
+            )
+            sys.exit(1)
         return func(*args, **kwargs)
 
     return wrapper

reana_server/reana_admin/cli.py

@@ -233,8 +233,6 @@ def token_grant(admin_access_token, id_, email):
     """Grant a token to the selected user."""
     try:
         admin = User.query.filter_by(id_=ADMIN_USER_ID).one_or_none()
-        if admin_access_token != admin.access_token:
-            raise ValueError("Admin access token invalid.")
         user = _get_user_by_criteria(id_, email)
         error_msg = None
         if not user:
@@ -298,10 +296,7 @@ def token_revoke(admin_access_token, id_, email):
     """Revoke selected user's token."""
     try:
         admin = User.query.filter_by(id_=ADMIN_USER_ID).one_or_none()
-        if admin_access_token != admin.access_token:
-            raise ValueError("Admin access token invalid.")
         user = _get_user_by_criteria(id_, email)
-
         error_msg = None
         if not user:
             error_msg = f"User {id_ or email} does not exist."
@@ -559,8 +554,11 @@ def list_quota_resources(ctx):
 @click.option(
     "--limit", "-l", help="New limit in canonical unit.", required=True, type=int
 )
+@admin_access_token_option
 @click.pass_context
-def set_quota_limit(ctx, emails, resource_type, resource_name, limit):
+def set_quota_limit(
+    ctx, emails, resource_type, resource_name, limit, admin_access_token
+):
     """Set quota limits to the given users per resource."""
     try:
         for email in emails:
@@ -638,8 +636,9 @@ def set_quota_limit(ctx, emails, resource_type, resource_name, limit):
     "quota-set-default-limits",
     help="Set default quota limits to users that do not have any.",
 )
+@admin_access_token_option
 @click.pass_context
-def set_default_quota_limit(ctx):
+def set_default_quota_limit(ctx, admin_access_token: str):
     """Set default quota limits to users that do not have any."""
     users_without_quota_limits = User.query.filter(~User.resources.any()).all()
     if not users_without_quota_limits:
@@ -682,11 +681,13 @@ def set_default_quota_limit(ctx):
     default=False,
     help="Manually decide which messages to remove from the queue.",
 )
+@admin_access_token_option
 def queue_consume(
     queue_name: str,
     key: Optional[str],
     values_to_delete: List[str],
     interactive: bool,
+    admin_access_token: str,
 ):
     """Start consuming specified queue and remove selected messages.
 
@@ -750,12 +751,14 @@ def queue_consume(
 )
 @add_user_options
 @add_workflow_option()
+@admin_access_token_option
 def retention_rules_apply(
     dry_run: bool,
     force_date: Optional[datetime.datetime],
     yes_i_am_sure: bool,
     user: Optional[User],
     workflow: Optional[Workflow],
+    admin_access_token: str,
 ) -> None:
     """Apply pending retentions rules."""
     if user and workflow and user.id_ != workflow.owner_id:
@@ -838,7 +841,10 @@ def retention_rules_apply(
     required=True,
     type=click.IntRange(min=0),
 )
-def retention_rules_extend(workflow: Optional[Workflow], days: int) -> None:
+@admin_access_token_option
+def retention_rules_extend(
+    workflow: Optional[Workflow], days: int, admin_access_token: str
+) -> None:
     """Extend active retentions rules."""
     click.echo("Fetching all the active rules")
     active_rules = WorkspaceRetentionRule.query.filter(
@@ -875,8 +881,11 @@ def retention_rules_extend(workflow: Optional[Workflow], days: int) -> None:
     default=None,
     help="Default value is now.",
 )
+@admin_access_token_option
 def check_workflows(
-    date_start: datetime.datetime, date_end: Optional[datetime.datetime]
+    date_start: datetime.datetime,
+    date_end: Optional[datetime.datetime],
+    admin_access_token: str,
 ) -> None:
     """Check consistency of selected workflow run statuses between database, message queue and Kubernetes."""
     from .check_workflows import (
@@ -968,7 +977,10 @@ def check_workflows(
     default=False,
     help="Show which interactive sessions would be closed, without closing them. [default=False]",
 )
-def interactive_session_cleanup(days: int, dry_run: bool) -> None:
+@admin_access_token_option
+def interactive_session_cleanup(
+    days: int, dry_run: bool, admin_access_token: str
+) -> None:
     """Close inactive interactive sessions."""
     click.echo(
         f"Starting to close interactive sessions running longer than {days} days.."

reana_server/utils.py

@@ -289,11 +289,16 @@ def _load_and_save_yadage_spec(workflow: Workflow, operational_options: Dict):
     Session.commit()
 
 
-def _get_users(_id, email, user_access_token, admin_access_token):
-    """Return all users matching search criteria."""
+def _validate_admin_access_token(admin_access_token: str):
+    """Validate admin access token."""
     admin = Session.query(User).filter_by(id_=ADMIN_USER_ID).one_or_none()
     if admin_access_token != admin.access_token:
         raise ValueError("Admin access token invalid.")
+
+
+def _get_users(_id, email, user_access_token, admin_access_token):
+    """Return all users matching search criteria."""
+    _validate_admin_access_token(admin_access_token)
     search_criteria = dict()
     if _id:
         search_criteria["id_"] = _id
@@ -310,9 +315,7 @@ def _get_users(_id, email, user_access_token, admin_access_token):
 def _create_user(email, user_access_token, admin_access_token):
     """Create user with provided credentials."""
     try:
-        admin = Session.query(User).filter_by(id_=ADMIN_USER_ID).one_or_none()
-        if admin_access_token != admin.access_token:
-            raise ValueError("Admin access token invalid.")
+        _validate_admin_access_token(admin_access_token)
         if not user_access_token:
             user_access_token = secrets.token_urlsafe(16)
         user_parameters = dict(access_token=user_access_token)
@@ -332,9 +335,7 @@ def _export_users(admin_access_token):
     :param admin_access_token: Admin access token.
     :type admin_access_token: str
     """
-    admin = User.query.filter_by(id_=ADMIN_USER_ID).one_or_none()
-    if admin_access_token != admin.access_token:
-        raise ValueError("Admin access token invalid.")
+    _validate_admin_access_token(admin_access_token)
     csv_file_obj = io.StringIO()
     csv_writer = csv.writer(csv_file_obj, dialect="unix")
     for user in User.query.all():
@@ -352,9 +353,7 @@ def _import_users(admin_access_token, users_csv_file):
     :param users_csv_file: CSV file object containing a list of users.
     :type users_csv_file: _io.TextIOWrapper
     """
-    admin = User.query.filter_by(id_=ADMIN_USER_ID).one_or_none()
-    if admin_access_token != admin.access_token:
-        raise ValueError("Admin access token invalid.")
+    _validate_admin_access_token(admin_access_token)
     csv_reader = csv.reader(users_csv_file)
     for row in csv_reader:
         user = User(

tests/test_cli.py

@@ -394,7 +394,12 @@ def test_is_input_or_output(file_or_dir, expected_result):
     ],
 )
 def test_retention_rules_apply(
-    workflow_with_retention_rules, session, time_delta, to_be_kept, to_be_deleted
+    default_user,
+    workflow_with_retention_rules,
+    session,
+    time_delta,
+    to_be_kept,
+    to_be_deleted,
 ):
     """Test the deletion of files when applying retention rules."""
 
@@ -425,7 +430,11 @@ def init_workspace(workspace, files):
     session.add(other_workflow)
     session.commit()
 
-    command = ["retention-rules-apply"]
+    command = [
+        "retention-rules-apply",
+        "--admin-access-token",
+        default_user.access_token,
+    ]
     if time_delta is not None:
         forced_date = datetime.datetime.now() + time_delta
         command += ["--force-date", forced_date.strftime("%Y-%m-%dT%H:%M:%S")]
@@ -459,14 +468,21 @@ def init_workspace(workspace, files):
 
 @patch("reana_server.reana_admin.cli.RetentionRuleDeleter.apply_rule")
 def test_retention_rules_apply_error(
-    apply_rule_mock: Mock, workflow_with_retention_rules
+    apply_rule_mock: Mock, workflow_with_retention_rules, default_user
 ):
     """Test that rules are reset to `active` if there are errors."""
     workflow = workflow_with_retention_rules
     apply_rule_mock.side_effect = Exception()
 
     runner = CliRunner()
-    result = runner.invoke(reana_admin, "retention-rules-apply")
+    result = runner.invoke(
+        reana_admin,
+        [
+            "retention-rules-apply",
+            "--admin-access-token",
+            default_user.access_token,
+        ],
+    )
 
     assert result.exit_code == 0
     assert "Error while applying rule" in result.output
@@ -475,20 +491,37 @@ def test_retention_rules_apply_error(
         assert rule.status == WorkspaceRetentionRuleStatus.active
 
 
-def test_retention_rules_extend(workflow_with_retention_rules):
+def test_retention_rules_extend(workflow_with_retention_rules, default_user):
     """Test extending of retention rules."""
     workflow = workflow_with_retention_rules
     runner = CliRunner()
     extend_days = 5
 
     result = runner.invoke(
-        reana_admin, ["retention-rules-extend", "-w non-valid-id", "-d", extend_days]
+        reana_admin,
+        [
+            "retention-rules-extend",
+            "-w non-valid-id",
+            "-d",
+            extend_days,
+            "--admin-access-token",
+            default_user.access_token,
+        ],
     )
     assert result.output == "Invalid workflow UUID.\n"
     assert result.exit_code == 1
 
     result = runner.invoke(
-        reana_admin, ["retention-rules-extend", "-w", workflow.id_, "-d", extend_days]
+        reana_admin,
+        [
+            "retention-rules-extend",
+            "-w",
+            workflow.id_,
+            "-d",
+            extend_days,
+            "--admin-access-token",
+            default_user.access_token,
+        ],
     )
     assert "Extending rule" in result.output
     assert result.exit_code == 0
@@ -522,7 +555,7 @@ def test_retention_rule_deleter_file_outside_workspace(tmp_path):
 )
 @patch("reana_server.reana_admin.cli.requests.get")
 def test_interactive_session_cleanup(
-    mock_requests, sample_serial_workflow_in_db, days, output
+    mock_requests, sample_serial_workflow_in_db, days, output, default_user
 ):
     """Test closure of long running interactive sessions."""
     runner = CliRunner()
@@ -559,7 +592,14 @@ def test_interactive_session_cleanup(
             ),
         ):
             result = runner.invoke(
-                reana_admin, ["interactive-session-cleanup", "-d", days]
+                reana_admin,
+                [
+                    "interactive-session-cleanup",
+                    "-d",
+                    days,
+                    "--admin-access-token",
+                    default_user.access_token,
+                ],
             )
             assert output in result.output