feat: Add multiple profiles support (#168)

* feat: Add multiple profiles support

* chore: Better requirement docs, bug fix with test

* chore: Reset singleton and deepcopy defaults

* chore: Remove side effects from processing

* chore: Only warn if AWS profile is set

Author: ruhulio

docs/README.md

@@ -91,6 +91,9 @@ options:
                         password to log in to Okta. You can also use the TOKENDITO_OKTA_PASSWORD environment variable.
   --profile USER_CONFIG_PROFILE
                         Tokendito configuration profile to use.
+  --multi-profiles USER_CONFIG_PROFILE
+                        Similar to --profile, but can be specified multiple times.
+                        Note: Using this will override --profile and cause --aws-profile to be ignored and replaced with this value.
   --config-file USER_CONFIG_FILE
                         Use an alternative configuration file. Defaults to tokendito.ini with location depending on the OS.
   --loglevel {DEBUG,INFO,WARN,ERROR}, -l {DEBUG,INFO,WARN,ERROR}

tests/unit/test_user.py

@@ -727,8 +727,7 @@ def test_process_interactive_input(mocker):
     pytest_config.user["quiet"] = True
     pytest_config.okta["username"] = ""
     ret = user.process_interactive_input(pytest_config)
-    pytest_config.update(ret)
-    assert pytest_config.okta["username"] == ""
+    assert ret is None
 
     # Check that a bad object raises an exception
     with pytest.raises(AttributeError) as error:
@@ -955,3 +954,51 @@ def test_extract_arns(saml, expected):
     from tokendito import user
 
     assert user.extract_arns(saml) == expected
+
+
+def test_single_profile(mocker):
+    """Test single profile support."""
+    from tokendito import user
+
+    patched = mocker.patch("tokendito.user.process_args", return_value=None)
+
+    args = [
+        "--profile",
+        "profile-1",
+    ]
+    user.cmd_interface(args)
+
+    assert patched.call_count == 1
+
+    assert patched.call_args_list[0].args[0].multi_profiles is None
+
+    # skip_auth parameter should only be called with False
+    assert patched.call_args_list[0].args[1] is False
+
+
+def test_multiple_profiles(mocker):
+    """Test multiple profiles support."""
+    from tokendito import user
+
+    patched = mocker.patch("tokendito.user.process_args", return_value=None)
+
+    profile_1 = "profile-1"
+    profile_2 = "profile-2"
+    profile_3 = "profile-3"
+
+    args = [
+        "--multi-profiles",
+        profile_1,
+        "--multi-profiles",
+        profile_2,
+        "--multi-profiles",
+        profile_3,
+    ]
+    user.cmd_interface(args)
+
+    assert patched.call_count == 3
+
+    # skip_auth parameter should only be called with False the first time
+    assert patched.call_args_list[0].args[1] is False
+    assert patched.call_args_list[1].args[1] is True
+    assert patched.call_args_list[2].args[1] is True

tokendito/config.py

@@ -1,6 +1,7 @@
 # vim: set filetype=python ts=4 sw=4
 # -*- coding: utf-8 -*-
 """Tokendito configuration class."""
+import copy
 import json
 import os
 from os.path import expanduser
@@ -119,7 +120,7 @@ def _check_constraints(self, **kwargs):
     def set_defaults(self):
         """Update the object to default settings."""
         for key in self._defaults.keys():
-            setattr(self, key, self._defaults[key])
+            setattr(self, key, copy.deepcopy(self._defaults[key]))
 
     def get_defaults(self):
         """Retrieve default settings."""

tokendito/user.py

@@ -46,6 +46,32 @@ def cmd_interface(args):
     # Early logging, in case the user requests debugging via env/CLI
     setup_early_logging(args)
 
+    if args.multi_profiles:
+        if args.aws_profile or ("TOKENDITO_AWS_PROFILE" in os.environ):
+            logger.warning(
+                "Multiple profiles have been specified so the AWS profile value "
+                "will be overridden by each profile."
+            )
+
+        skip_auth = False
+        for profile in args.multi_profiles:
+            args.user_config_profile = profile
+            args.aws_profile = profile
+
+            process_args(args, skip_auth)
+
+            # Reset config singleton but retain auth
+            auth = config.okta["password"]
+            config.set_defaults()
+            config.okta["password"] = auth
+
+            skip_auth = True
+    else:
+        process_args(args, False)
+
+
+def process_args(args, skip_auth=False):
+    """Process the args and allow for skipping auth with multiple profiles."""
     # Set some required initial values
     process_options(args)
 
@@ -78,7 +104,7 @@ def cmd_interface(args):
             )
 
     # get authentication and authorization cookies from okta
-    okta.access_control(config)
+    _ = skip_auth or okta.access_control(config)
 
     if config.okta["tile"]:
         tile_label = ""
@@ -166,6 +192,13 @@ def parse_cli_args(args):
         default=config.user["config_profile"],
         help="Tokendito configuration profile to use.",
     )
+    parser.add_argument(
+        "--multi-profiles",
+        action="append",
+        help="Tokendito configuration profiles to use. Can be specified multiple times. "
+        "Using this will override --profile and cause --aws-profile to be ignored and "
+        "replaced with this value.",
+    )
     parser.add_argument(
         "--config-file",
         dest="user_config_file",
@@ -669,17 +702,11 @@ def add_sensitive_value_to_be_masked(value, key=None):
         mask_items.append(value)
 
 
-def process_ini_file(file, profile):
-    """Process options from a ConfigParser ini file.
-
-    :param file: filename
-    :param profile: profile to read
-    :return: Config object with configuration values
-    """
+def _read_ini(file, profile, default_section):
     res = dict()
     pattern = re.compile(r"^(.*?)_(.*)")
 
-    ini = configparser.RawConfigParser(default_section=config.user["config_profile"])
+    ini = configparser.RawConfigParser(default_section=default_section)
     # Here, group(1) is the dictionary key, and group(2) the configuration element
     try:
         ini.read(file)
@@ -693,26 +720,33 @@ def process_ini_file(file, profile):
     except configparser.Error as err:
         logger.error(f"Could not load profile '{profile}': {str(err)}")
         sys.exit(2)
+
+    return res
+
+
+def process_ini_file(file, profile):
+    """Process options from a ConfigParser ini file.
+
+    :param file: filename
+    :param profile: profile to read
+    :return: Config object with configuration values
+    """
+    res = _read_ini(file, profile, config.user["config_profile"])
     logger.debug(f"Found ini directives: {res}")
+    if not res:
+        return None
 
     try:
-        config_ini = Config(**res)
-
+        return Config(**res)
     except (AttributeError, KeyError, ValueError) as err:
         logger.error(
             f"The configuration file {file} in [{profile}] is incorrect: {err}"
             ". Please check your settings and try again."
         )
         sys.exit(1)
-    return config_ini
 
 
-def process_arguments(args):
-    """Process command-line arguments.
-
-    :param args: argparse object
-    :return: Config object with configuration values
-    """
+def _read_arguments(args):
     res = dict()
     pattern = re.compile(r"^(.*?)_(.*)")
 
@@ -726,18 +760,29 @@ def process_arguments(args):
             if val:
                 res[match.group(1)][match.group(2)] = val
                 add_sensitive_value_to_be_masked(val, match.group(2))
+
+    return res
+
+
+def process_arguments(args):
+    """Process command-line arguments.
+
+    :param args: argparse object
+    :return: Config object with configuration values
+    """
+    res = _read_arguments(args)
     logger.debug(f"Found arguments: {res}")
+    if not res:
+        return None
 
     try:
-        config_args = Config(**res)
-
+        return Config(**res)
     except (AttributeError, KeyError, ValueError) as err:
         logger.error(
             f"Command line arguments not correct: {err}"
             ". This should not happen, please contact the package maintainers."
         )
         sys.exit(1)
-    return config_args
 
 
 def process_environment(prefix="tokendito"):
@@ -759,21 +804,42 @@ def process_environment(prefix="tokendito"):
                 add_sensitive_value_to_be_masked(val, match.group(3))
     logger.debug(f"Found environment variables: {res}")
 
-    try:
-        config_env = Config(**res)
+    if not res:
+        return None
 
+    try:
+        return Config(**res)
     except (AttributeError, KeyError, ValueError) as err:
         logger.error(
             f"The environment variables are incorrectly set: {err}"
             ". Please check your settings and try again."
         )
         sys.exit(1)
-    return config_env
+
+
+def _build_interactive_config(details, skip_password):
+    res = dict(okta=dict())
+    # Copy the values set by get_interactive_config
+    if "okta_tile" in details:
+        res["okta"]["tile"] = details["okta_tile"]
+    if "okta_org" in details:
+        res["okta"]["org"] = details["okta_org"]
+    if "okta_username" in details:
+        res["okta"]["username"] = details["okta_username"]
+
+    if ("password" not in config.okta or config.okta["password"] == "") and not skip_password:
+        logger.debug("No password set, will try to get one interactively")
+        res["okta"]["password"] = get_secret_input()
+        add_sensitive_value_to_be_masked(res["okta"]["password"])
+
+    logger.debug(f"Interactive configuration is: {res}")
+
+    return Config(**res)
 
 
 def process_interactive_input(config, skip_password=False):
     """
-    Request input interactively interactively for elements that are not proesent.
+    Request input interactively interactively for elements that are not present.
 
     :param config: Config object with some values set.
     :param skip_password: Whether or not ask the user for a password.
@@ -782,7 +848,7 @@ def process_interactive_input(config, skip_password=False):
     # Return quickly if the user attempts to run in quiet (non-interactive) mode.
     if config.user["quiet"] is True:
         logger.debug(f"Skipping interactive config: quiet mode is {config.user['quiet']}")
-        return config
+        return None
 
     # Reuse interactive config. It will only request the portions needed.
     try:
@@ -795,25 +861,10 @@ def process_interactive_input(config, skip_password=False):
         logger.error(f"Interactive arguments are not correct: {err}")
         sys.exit(1)
 
-    # Create a dict that can be passed to Config later
-    res = dict(okta=dict())
-    # Copy the values set by get_interactive_config
-    if "okta_tile" in details:
-        res["okta"]["tile"] = details["okta_tile"]
-    if "okta_org" in details:
-        res["okta"]["org"] = details["okta_org"]
-    if "okta_username" in details:
-        res["okta"]["username"] = details["okta_username"]
-
-    if ("password" not in config.okta or config.okta["password"] == "") and not skip_password:
-        logger.debug("No password set, will try to get one interactively")
-        res["okta"]["password"] = get_secret_input()
-        add_sensitive_value_to_be_masked(res["okta"]["password"])
+    if not details:
+        return None
 
-    config_int = Config(**res)
-    logger.debug(f"Interactive configuration is: {config_int}")
-    config.update(config_int)
-    return config_int
+    return _build_interactive_config(details, skip_password)
 
 
 def get_interactive_config(tile=None, org=None, username=""):
@@ -1190,23 +1241,25 @@ def process_options(args):
         sys.exit(0)
 
     # 1: read ini file (if it exists)
-    config_ini = Config()
     if not args.configure:
         config_ini = process_ini_file(args.user_config_file, args.user_config_profile)
+        if config_ini:
+            config.update(config_ini)
 
     # 2: override with ENV
     config_env = process_environment()
+    if config_env:
+        config.update(config_env)
 
     # 3: override with args
     config_args = process_arguments(args)
-
-    config.update(config_ini)
-    config.update(config_env)
-    config.update(config_args)
+    if config_args:
+        config.update(config_args)
 
     # 4: Get missing data from the user, if necessary
     config_int = process_interactive_input(config, args.configure)
-    config.update(config_int)
+    if config_int:
+        config.update(config_int)
 
     sanitize_config_values(config)
     logger.debug(f"Final configuration is {config}")