add support for mfa_token env and mfa-token-command

Author: Mark Heiges

README.md

@@ -252,8 +252,12 @@ aws-adfs integrates with:
                                       of a shell command (expected JSON format:
                                       `{"username": "myusername", "password":
                                       "mypassword"}`)
-      --env                           Read username, password from environment
-                                      variables (username and password).
+      --mfa-token-command TEXT        Read MFA token from the output of a shell
+                                      command (expected JSON format:
+                                      `{"mfa_token": "123654"}`)
+      --env                           Read username, password and optionally an
+                                      MFA token from environment variables
+                                      (username, password and mfa_token).
       --stdin                         Read username, password from standard input
                                       separated by a newline.
       --authfile TEXT                 Read username, password from a local file

aws_adfs/_rsa_authenticator.py

@@ -4,6 +4,8 @@
 import logging
 import re
 
+from . import run_command
+
 try:
     # Python 3
     from urllib.parse import urlparse, parse_qs
@@ -15,7 +17,7 @@
 from .helpers import trace_http_request
 
 
-def extract(html_response, ssl_verification_enabled, session):
+def extract(html_response, ssl_verification_enabled, mfa_token_command, mfa_token, session):
     """
     :param response: raw http response
     :param html_response: html result of parsing http response
@@ -24,7 +26,15 @@ def extract(html_response, ssl_verification_enabled, session):
 
     roles_page_url = _action_url_on_validation_success(html_response)
 
-    rsa_securid_code = click.prompt(text='Enter your RSA SecurID token', type=str, hide_input=True)
+    if mfa_token_command:
+      data = run_command.run_command(mfa_token_command)
+      rsa_securid_code = data['mfa_token']
+      logging.debug(f"using RSA SecurID token from command: {rsa_securid_code}")
+    elif mfa_token:
+      rsa_securid_code = mfa_token
+      logging.debug(f"using RSA SecurID token from env: {rsa_securid_code}")
+    else:
+      rsa_securid_code = click.prompt(text='Enter your RSA SecurID token', type=str, hide_input=True)
 
     click.echo('Going for aws roles', err=True)
     return _retrieve_roles_page(

aws_adfs/_symantec_vip_access.py

@@ -4,6 +4,8 @@
 import logging
 import re
 
+from . import run_command
+
 try:
     # Python 3
     from urllib.parse import urlparse, parse_qs
@@ -15,7 +17,7 @@
 from .helpers import trace_http_request
 
 
-def extract(html_response, ssl_verification_enabled, session):
+def extract(html_response, ssl_verification_enabled, mfa_token_command, mfa_token, session):
     """
     :param response: raw http response
     :param html_response: html result of parsing http response
@@ -24,7 +26,15 @@ def extract(html_response, ssl_verification_enabled, session):
 
     roles_page_url = _action_url_on_validation_success(html_response)
 
-    vip_security_code = click.prompt(text='Enter your Symantec VIP Access code', type=str)
+    if mfa_token_command:
+      data = run_command.run_command(mfa_token_command)
+      vip_security_code = data['mfa_token']
+      logging.debug(f"using VIP Access Code from command: {vip_security_code}")
+    elif mfa_token:
+      vip_security_code = mfa_token
+      logging.debug(f"using VIP Access Code from env: {vip_security_code}")
+    else:
+      vip_security_code = click.prompt(text='Enter your Symantec VIP Access code', type=str)
 
     click.echo('Going for aws roles', err=True)
     return _retrieve_roles_page(

aws_adfs/authenticator.py

@@ -101,7 +101,7 @@ def extract():
 
     def _symantec_vip_extractor():
         def extract():
-            return symantec_vip_access.extract(html_response, config.ssl_verification, session)
+            return symantec_vip_access.extract(html_response, config.ssl_verification, config.mfa_token_command, config.mfa_token, session)
         return extract
 
     def _file_extractor():
@@ -111,7 +111,7 @@ def extract():
 
     def _rsa_auth_extractor():
         def extract():
-            return rsa_auth.extract(html_response, config.ssl_verification, session)
+            return rsa_auth.extract(html_response, config.ssl_verification, config.mfa_token_command, config.mfa_token, session)
         return extract
 
     def _azure_mfa_extractor():

aws_adfs/login.py

@@ -19,6 +19,7 @@
 from botocore import client
 
 from . import authenticator, helpers, prepare, role_chooser
+from . import run_command
 from .consts import (
     DUO_UNIVERSAL_PROMPT_FACTOR_DUO_PUSH,
     DUO_UNIVERSAL_PROMPT_FACTOR_PASSCODE,
@@ -75,10 +76,14 @@
     "--username-password-command",
     help='Read username and password from the output of a shell command (expected JSON format: `{"username": "myusername", "password": "mypassword"}`)',
 )
+@click.option(
+    "--mfa-token-command",
+    help='Read MFA token from the output of a shell command (expected JSON format: `{"mfa_token": "123654"}`)',
+)
 @click.option(
     '--env',
     is_flag=True,
-    help='Read username, password from environment variables (username and password).',
+    help='Read username, password and optionally an MFA token from environment variables (username, password and mfa_token).',
 )
 @click.option(
     '--stdin',
@@ -161,6 +166,7 @@ def login(
     provider_id,
     s3_signature_version,
     username_password_command,
+    mfa_token_command,
     env,
     stdin,
     authfile,
@@ -194,6 +200,7 @@ def login(
         session_duration,
         sspi,
         username_password_command,
+        mfa_token_command,
         duo_factor,
         duo_device,
         aad_verification_code,
@@ -220,11 +227,9 @@ def login(
         # If we fail to get an assertion, prompt for credentials and try again
         if assertion is None:
             password = None
-
             if config.username_password_command:
-                config.adfs_user, password = _username_password_command_credentials(
-                    username_password_command
-                )
+                data = run_command.run_command(username_password_command)
+                config.adfs_user, password = data['username'], data['password']
             if stdin:
                 config.adfs_user, password = _stdin_user_credentials()
             elif env:
@@ -438,40 +443,6 @@ def _emit_summary(config, session_duration):
         err=True
     )
 
-
-def _username_password_command_credentials(username_password_command):
-    try:
-        logging.debug("Executing `{}`".format(username_password_command))
-        proc = subprocess.run(
-            username_password_command,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            check=True,
-            shell=True,
-        )
-        data = json.loads(proc.stdout)
-        username = data["username"]
-        password = data["password"]
-    except subprocess.CalledProcessError as e:
-        logging.error(
-            "Failed to execute the `{}` command to retrieve username and password: \n\n{}".format(
-                username_password_command, e.output
-            )
-        )
-        username = None
-        password = None
-    except json.JSONDecodeError as e:
-        logging.error(
-            "Failed to decode the output of the `{}` command as JSON to retrieve username and password: \n\n{}".format(
-                username_password_command, e
-            )
-        )
-        username = None
-        password = None
-
-    return username, password
-
-
 def _file_user_credentials(profile, authfile):
     config = configparser.ConfigParser()
 

aws_adfs/prepare.py

@@ -21,6 +21,7 @@ def get_prepared_config(
     session_duration,
     sspi,
     username_password_command,
+    mfa_token_command,
     duo_factor,
     duo_device,
     aad_verification_code=None,
@@ -47,6 +48,7 @@ def get_prepared_config(
     :param session_duration: AWS STS session duration (default 1 hour)
     :param sspi: Whether SSPI is enabled
     :param username_password_command: The command used to retrieve username and password information
+    :param mfa_token_command: The command used to retrieve MFA token
     :param duo_factor: The specific Duo factor to use
     :param duo_device: The specific Duo device to use
     :param aad_verification_code: If verification code is in config use that for multi-factor authentication
@@ -86,6 +88,7 @@ def default_if_none(value, default):
     adfs_config.session_duration = default_if_none(session_duration, adfs_config.session_duration)
     adfs_config.sspi = default_if_none(sspi, adfs_config.sspi)
     adfs_config.username_password_command = default_if_none(username_password_command, adfs_config.username_password_command)
+    adfs_config.mfa_token_command = default_if_none(mfa_token_command, adfs_config.mfa_token_command)
     adfs_config.duo_factor = default_if_none(duo_factor, adfs_config.duo_factor)
     adfs_config.duo_device = default_if_none(duo_device, adfs_config.duo_device)
     adfs_config.aad_verification_code = aad_verification_code
@@ -149,6 +152,11 @@ def create_adfs_default_config(profile):
     # The command used to retrieve username and password information
     config.username_password_command = None
 
+    # The command used to retrieve MFA token information
+    config.mfa_token_command = None
+
+    config.mfa_token = os.environ.get('mfa_token')
+
     # The specific Duo factor and device to use
     config.duo_factor = None
     config.duo_device = None
@@ -219,6 +227,7 @@ def load_config(config, profile):
             profile, 'adfs_config.sspi',
             str(adfs_config.sspi)))
         adfs_config.username_password_command = config.get_or(profile, 'adfs_config.username_password_command', adfs_config.username_password_command)
+        adfs_config.mfa_token_command = config.get_or(profile, 'adfs_config.mfa_token_command', adfs_config.mfa_token_command)
 
         adfs_config.duo_factor = config.get_or(profile, "adfs_config.duo_factor", adfs_config.duo_factor)
         if adfs_config.duo_factor == "None":

aws_adfs/run_command.py

@@ -0,0 +1,31 @@
+import json
+import logging
+import subprocess
+
+def run_command(command):
+    try:
+        logging.debug("Executing `{}`".format(command))
+        proc = subprocess.run(
+            command,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            check=True,
+            shell=True,
+        )
+        data = json.loads(proc.stdout)
+    except subprocess.CalledProcessError as e:
+        logging.error(
+            "Failed to execute the `{}` command: \n\n{}".format(
+                command, e.output
+            )
+        )
+        data = None
+    except json.JSONDecodeError as e:
+        logging.error(
+            "Failed to decode the output of the `{}` command as JSON: \n\n{}".format(
+                command, e
+            )
+        )
+        data = None
+
+    return data

test/test_config_preparation.py

@@ -24,6 +24,7 @@ def test_when_there_is_no_profile_use_default_values(self):
         default_session_duration = 3600
         default_sspi = False
         default_username_password_command = None
+        default_mfa_token_command = None
         default_duo_factor = None
         default_duo_device = None
         default_enforce_role_arn = False
@@ -41,6 +42,7 @@ def test_when_there_is_no_profile_use_default_values(self):
             default_session_duration,
             default_sspi,
             default_username_password_command,
+            default_mfa_token_command,
             default_duo_factor,
             default_duo_device,
         )
@@ -80,6 +82,7 @@ def test_when_the_profile_exists_but_lacks_ssl_verification_use_default_value(se
         irrelevant_s3_signature_version = "irrelevant_s3_signature_version"
         irrelevant_session_duration = "irrelevant_session_duration"
         irrelevant_username_password_command = "irrelevant_username_password_command"
+        irrelevant_mfa_token_command = "irrelevant_mfa_token_command"
         irrelevant_duo_factor = "irrelevant_duo_factor"
         irrelevant_duo_device = "irrelevant_duo_device"
 
@@ -96,6 +99,7 @@ def test_when_the_profile_exists_but_lacks_ssl_verification_use_default_value(se
             irrelevant_session_duration,
             default_sspi,
             irrelevant_username_password_command,
+            irrelevant_mfa_token_command,
             irrelevant_duo_factor,
             irrelevant_duo_device,
         )