admin: standardize --admin-access-token usage

reana_server/decorators.py

@@ -10,10 +10,8 @@
 
 import functools
 import logging
-import os
 import traceback
 
-import click
 from flask import jsonify, request
 from flask_login import current_user
 from reana_commons.errors import REANAQuotaExceededError
@@ -25,22 +23,6 @@
 )
 
 
-def admin_access_token_option(func):
-    """Click option to load admin access token."""
-
-    @click.option(
-        "--admin-access-token",
-        required=True,
-        default=os.environ.get("REANA_ADMIN_ACCESS_TOKEN"),
-        help="The access token of an administrator.",
-    )
-    @functools.wraps(func)
-    def wrapper(*args, **kwargs):
-        return func(*args, **kwargs)
-
-    return wrapper
-
-
 def signin_required(include_gitlab_login=False, token_required=True):
     """Check if the user is signed in or the access token is valid and return the user."""
 

reana_server/reana_admin/cli.py

@@ -47,8 +47,11 @@
 from reana_db.utils import update_workspace_retention_rules
 
 from reana_server.config import ADMIN_EMAIL, ADMIN_USER_ID, REANA_HOSTNAME
-from reana_server.decorators import admin_access_token_option
-from reana_server.reana_admin.options import add_user_options, add_workflow_option
+from reana_server.reana_admin.options import (
+    add_user_options,
+    add_workflow_option,
+    admin_access_token_option,
+)
 from reana_server.reana_admin.retention_rule_deleter import RetentionRuleDeleter
 from reana_server.status import STATUS_OBJECT_TYPES
 from reana_server.api_client import current_rwc_api_client
@@ -126,7 +129,7 @@ def users_create_default(email, password, id_):
 def list_users(ctx, id, email, user_access_token, admin_access_token, output_format):
     """List users according to the search criteria."""
     try:
-        response = _get_users(id, email, user_access_token, admin_access_token)
+        response = _get_users(id, email, user_access_token)
         headers = ["id", "email", "access_token", "access_token_status"]
         data = []
         for user in response:
@@ -171,7 +174,7 @@ def list_users(ctx, id, email, user_access_token, admin_access_token, output_for
 def create_user(ctx, email, user_access_token, admin_access_token):
     """Create a new user. Requires the token of an administrator."""
     try:
-        response = _create_user(email, user_access_token, admin_access_token)
+        response = _create_user(email, user_access_token)
         headers = ["id", "email", "access_token"]
         data = [(str(response.id_), response.email, response.access_token)]
         click.echo(click.style("User was successfully created.", fg="green"))
@@ -192,7 +195,7 @@ def create_user(ctx, email, user_access_token, admin_access_token):
 def export_users(ctx, admin_access_token):
     """Export all users in current REANA cluster."""
     try:
-        csv_file = _export_users(admin_access_token)
+        csv_file = _export_users()
         click.echo(csv_file.getvalue(), nl=False)
     except Exception as e:
         click.secho(
@@ -215,7 +218,7 @@ def export_users(ctx, admin_access_token):
 def import_users(ctx, admin_access_token, file_):
     """Import users from file."""
     try:
-        _import_users(admin_access_token, file_)
+        _import_users(file_)
         click.secho("Users successfully imported.", fg="green")
     except Exception as e:
         click.secho(
@@ -233,8 +236,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 +299,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."
@@ -465,7 +463,7 @@ def list_quota_usage(
 ):
     """List quota usage of users."""
     try:
-        response = _get_users(id, email, user_access_token, admin_access_token)
+        response = _get_users(id, email, user_access_token)
         headers = ["id", "email", "cpu-used", "cpu-limit", "disk-used", "disk-limit"]
         health_order = {
             QuotaHealth.healthy.name: 0,
@@ -559,8 +557,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 +639,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 +684,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 +754,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 +844,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 +884,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 +980,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/reana_admin/options.py

@@ -7,14 +7,42 @@
 # under the terms of the MIT License; see LICENSE file for more details.
 """Common options for the REANA administrator command line tool."""
 
-import sys
-import click
 import functools
+import os
+import sys
 
+import click
 from reana_db.models import Workflow
-from reana_db.utils import _get_workflow_by_uuid
 
-from reana_server.utils import _get_user_by_criteria, is_uuid_v4
+from reana_server.utils import (
+    _get_user_by_criteria,
+    _validate_admin_access_token,
+    is_uuid_v4,
+)
+
+
+def admin_access_token_option(func):
+    """Click option to load admin access token."""
+
+    @click.option(
+        "--admin-access-token",
+        required=True,
+        default=os.environ.get("REANA_ADMIN_ACCESS_TOKEN"),
+        help="The access token of an administrator.",
+    )
+    @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
 
 
 def add_user_options(func):

reana_server/utils.py

@@ -288,11 +288,15 @@ 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):
+    """Return all users matching search criteria."""
     search_criteria = dict()
     if _id:
         search_criteria["id_"] = _id
@@ -306,12 +310,9 @@ def _get_users(_id, email, user_access_token, admin_access_token):
     return query.all()
 
 
-def _create_user(email, user_access_token, admin_access_token):
+def _create_user(email, user_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.")
         if not user_access_token:
             user_access_token = secrets.token_urlsafe(16)
         user_parameters = dict(access_token=user_access_token)
@@ -325,15 +326,12 @@ def _create_user(email, user_access_token, admin_access_token):
     return user
 
 
-def _export_users(admin_access_token):
+def _export_users():
     """Export all users in database as csv.
 
     :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.")
     csv_file_obj = io.StringIO()
     csv_writer = csv.writer(csv_file_obj, dialect="unix")
     for user in User.query.all():
@@ -343,17 +341,14 @@ def _export_users(admin_access_token):
     return csv_file_obj
 
 
-def _import_users(admin_access_token, users_csv_file):
+def _import_users(users_csv_file):
     """Import list of users to database.
 
     :param admin_access_token: Admin access token.
     :type admin_access_token: str
     :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.")
     csv_reader = csv.reader(users_csv_file)
     for row in csv_reader:
         user = User(