Possible to trigger TOTP activation and login using TOTP

dds_cli/__init__.py

@@ -73,6 +73,7 @@ class DDSEndpoint:
     REVOKE_PROJECT_ACCESS = BASE_ENDPOINT + "/user/access/revoke"
     DISPLAY_USER_INFO = BASE_ENDPOINT + "/user/info"
     USER_ACTIVATION = BASE_ENDPOINT + "/user/activation"
+    USER_ACTIVATE_TOTP = BASE_ENDPOINT + "/user/totp/activate"
 
     # Authentication - user and project
     ENCRYPTED_TOKEN = BASE_ENDPOINT + "/user/encrypted_token"

dds_cli/__main__.py

@@ -340,8 +340,14 @@ def auth_group_command(_):
 
 # -- dds auth login -- #
 @auth_group_command.command(name="login")
+@click.option(
+    "--totp",
+    type=str,
+    default=None,
+    help="2FA authentication via authentication app. Default is to use one-time authentication code via mail.",
+)
 @click.pass_obj
-def login(click_ctx):
+def login(click_ctx, totp):
     """Start or renew an authenticated session.
 
     Creates or renews the authentication token stored in the '.dds_cli_token' file.
@@ -353,7 +359,7 @@ def login(click_ctx):
     if no_prompt:
         LOG.warning("The --no-prompt flag is ignored for `dds auth login`")
     try:
-        with dds_cli.auth.Auth(token_path=click_ctx.get("TOKEN_PATH")):
+        with dds_cli.auth.Auth(token_path=click_ctx.get("TOKEN_PATH"), totp=totp):
             # Authentication token renewed in the init method.
             LOG.info("[green] :white_check_mark: Authentication token created![/green]")
     except (
@@ -407,6 +413,30 @@ def info(click_ctx):
         sys.exit(1)
 
 
+@auth_group_command.command(name="twofactor")
+def twofactor():
+    """Configure your preferred method of two-factor authentication."""
+    try:
+        LOG.info("Starting configuration of one-time authentication code method.")
+        auth_method_choice = questionary.select(
+            "Which method would you like to use?", choices=["Email", "Authenticator App", "Cancel"]
+        ).ask()
+
+        if auth_method_choice == "Cancel":
+            LOG.info("Two-factor authentication method not configured.")
+            sys.exit(0)
+        elif auth_method_choice == "Authenticator App":
+            auth_method = "totp"
+        elif auth_method_choice == "Email":
+            auth_method = "hotp"
+
+        with dds_cli.auth.Auth(authenticate=True, force_renew_token=False) as authenticator:
+            authenticator.twofactor(auth_method=auth_method)
+    except (dds_cli.exceptions.DDSCLIException, dds_cli.exceptions.ApiResponseError) as err:
+        LOG.error(err)
+        sys.exit(1)
+
+
 ####################################################################################################
 ####################################################################################################
 ## USER #################################################################################### USER ##

dds_cli/auth.py

@@ -1,7 +1,10 @@
 """Data Delivery System saved authentication token manager."""
 import logging
+import requests
+import simplejson
 
 # Own modules
+import dds_cli
 from dds_cli import base
 from dds_cli import user
 
@@ -23,15 +26,18 @@ class Auth(base.DDSBaseClass):
     def __init__(
         self,
         authenticate: bool = True,
+        force_renew_token: bool = True,  # Only used if authenticate is True
         token_path: str = None,
+        totp: str = None,
     ):
         """Handle actions regarding session management in DDS."""
         # Initiate DDSBaseClass to authenticate user
         super().__init__(
             authenticate=authenticate,
             method_check=False,
-            force_renew_token=True,  # Only used if authenticate is True
+            force_renew_token=force_renew_token,
             token_path=token_path,
+            totp=totp,
         )
 
     def check(self):
@@ -51,3 +57,21 @@ def logout(self):
             LOG.info("[green] :white_check_mark: Successfully logged out![/green]")
         else:
             LOG.info("[green]Already logged out![/green]")
+
+    def twofactor(self, auth_method: str = None):
+        try:
+            response = requests.put(
+                dds_cli.DDSEndpoint.USER_ACTIVATE_TOTP,
+                headers=self.token,
+                json={"activate_totp": True if auth_method == "totp" else False},
+            )
+            response_json = response.json()
+        except requests.exceptions.RequestException as err:
+            raise dds_cli.exceptions.ApiRequestError(message=str(err))
+        except simplejson.JSONDecodeError as err:
+            raise dds_cli.exceptions.ApiResponseError(message=str(err))
+
+        if not response.ok:
+            raise dds_cli.exceptions.ApiResponseError(message=response.reason)
+
+        LOG.info(response_json.get("message"))

dds_cli/base.py

@@ -52,6 +52,7 @@ def __init__(
         authenticate: bool = True,
         method_check: bool = True,
         force_renew_token: bool = False,
+        totp: str = None,
         no_prompt: bool = False,
         token_path: str = None,
     ):
@@ -96,6 +97,7 @@ def __init__(
                 force_renew_token=force_renew_token,
                 no_prompt=no_prompt,
                 token_path=token_path,
+                totp=totp,
             )
             self.token = dds_user.token_dict
 

dds_cli/user.py

@@ -46,22 +46,23 @@ def __init__(
         force_renew_token: bool = False,
         no_prompt: bool = False,
         token_path: str = None,
+        totp: str = None,
     ):
         self.force_renew_token = force_renew_token
         self.no_prompt = no_prompt
         self.token = None
         self.token_path = token_path
 
         # Fetch encrypted JWT token or authenticate against API
-        self.__retrieve_token()
+        self.__retrieve_token(totp=totp)
 
     @property
     def token_dict(self):
         """Get token as authorization dict for requests."""
         return {"Authorization": f"Bearer {self.token}"}
 
     # Private methods ######################### Private methods #
-    def __retrieve_token(self):
+    def __retrieve_token(self, totp: str = None):
         """Fetch saved token from file otherwise authenticate user and saves the new token."""
         token_file = TokenFile(token_path=self.token_path)
 
@@ -83,10 +84,10 @@ def __retrieve_token(self):
                 )
             else:
                 LOG.info("Attempting to create the session token")
-            self.token = self.__authenticate_user()
+            self.token = self.__authenticate_user(totp=totp)
             token_file.save_token(self.token)
 
-    def __authenticate_user(self):
+    def __authenticate_user(self, totp: str = None):
         """Authenticates the username and password via a call to the API."""
         LOG.debug("Starting authentication on the API.")
 
@@ -116,35 +117,67 @@ def __authenticate_user(self):
 
         # Token received from API needs to be completed with a mfa timestamp
         partial_auth_token = response_json.get("token")
+        secondfactor_method = response_json.get("secondfactor_method")
 
-        # Verify 2fa email token
-        LOG.info(
-            "Please enter the one-time authentication code sent "
-            "to your email address (leave empty to exit):"
-        )
-        done = False
-        while not done:
-            entered_one_time_code = rich.prompt.Prompt.ask("Authentication one-time code")
-            if entered_one_time_code == "":
+        totp_enabled = secondfactor_method == "TOTP"
+
+        # Verify Second Factor
+        if totp:
+            if not totp_enabled:
                 raise exceptions.AuthenticationError(
-                    message="Exited due to no one-time authentication code entered."
+                    "Authentication failed, you have not yet activated one-time authentication codes from authenticator app."
                 )
+            response_json = dds_cli.utils.perform_request(
+                dds_cli.DDSEndpoint.SECOND_FACTOR,
+                method="get",
+                headers={"Authorization": f"Bearer {partial_auth_token}"},
+                json={"TOTP": totp},
+                error_message="Failed to authenticate with one-time authentication code",
+            )
 
-            if not entered_one_time_code.isdigit():
-                LOG.info("Please enter a valid one-time code. It should consist of only digits.")
-                continue
-            if len(entered_one_time_code) != 8:
+        else:
+            if totp_enabled:
                 LOG.info(
-                    "Please enter a valid one-time code. It should consist of 8 digits "
-                    f"(you entered {len(entered_one_time_code)} digits)."
+                    "Please enter the one-time authentication code from your authenticator app."
                 )
-                continue
+                nr_digits = 6
+            else:
+                LOG.info(
+                    "Please enter the one-time authentication code sent "
+                    "to your email address (leave empty to exit):"
+                )
+                nr_digits = 8
+
+            done = False
+            while not done:
+                entered_one_time_code = rich.prompt.Prompt.ask("Authentication one-time code")
+                if entered_one_time_code == "":
+                    raise exceptions.AuthenticationError(
+                        message="Exited due to no one-time authentication code entered."
+                    )
+
+                if not entered_one_time_code.isdigit():
+                    LOG.info(
+                        "Please enter a valid one-time code. It should consist of only digits."
+                    )
+                    continue
+                if len(entered_one_time_code) != nr_digits:
+                    LOG.info(
+                        f"Please enter a valid one-time code. It should consist of {nr_digits} digits "
+                        f"(you entered {len(entered_one_time_code)} digits)."
+                    )
+                    continue
+
+                if totp_enabled:
+                    json_request = {"TOTP": entered_one_time_code}
+                else:
+                    json_request = {"HOTP": entered_one_time_code}
 
             response_json = dds_cli.utils.perform_request(
                 dds_cli.DDSEndpoint.SECOND_FACTOR,
                 method="get",
                 headers={"Authorization": f"Bearer {partial_auth_token}"},
-                json={"HOTP": entered_one_time_code},
+                json=json_request,
                 error_message="Failed to authenticate with second factor",
             )