Update AWS SSO configuration handling (reading)

CONTRIBUTORS.md

@@ -8,3 +8,4 @@
 * [O'Shaughnessy Evans](https://github.com/oshaughnessy)
 * [Anthony Engelstad](https://github.com/anton0)
 * [Shaun Dewberry](https://github.com/shaundewberry)
+* [ikeyan](https://github.com/ikeyan)

cli/src/aws_sso_util/cfn.py

@@ -11,7 +11,6 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 
-import argparse
 from collections import namedtuple, OrderedDict
 from pathlib import Path
 import logging

cli/src/aws_sso_util/check.py

@@ -19,17 +19,17 @@
 import traceback
 import datetime
 import textwrap
-import json
+from typing import List
 
-import botocore
+from botocore.session import Session
 import click
 
 from aws_error_utils import catch_aws_error
 
-from aws_sso_lib.sso import get_boto3_session, list_available_accounts, list_available_roles, login, get_token_fetcher, SSO_TOKEN_DIR
-from aws_sso_lib.config import find_instances, SSOInstance
+from aws_sso_lib.sso import list_available_accounts, list_available_roles, login, get_token_fetcher, SSO_TOKEN_DIR
+from aws_sso_lib.config import find_instances, SSOInstance, get_sso_config
 
-from .utils import configure_logging, GetInstanceError
+from .utils import configure_logging
 
 from .login import LOGIN_DEFAULT_START_URL_VARS, LOGIN_DEFAULT_SSO_REGION_VARS
 from .configure_profile import CONFIGURE_DEFAULT_START_URL_VARS, CONFIGURE_DEFAULT_SSO_REGION_VARS
@@ -128,27 +128,24 @@ def check(
     if check_profile:
         if (sso_start_url or sso_region or account or role_name):
             raise click.UsageError("Cannot specify --sso-start-url, --sso-region, --account-id, or --role-name with --check-profile")
-        config_session = botocore.session.Session(profile=check_profile)
-        missing = []
-        profile_config = {}
-        for key in ["sso_start_url", "sso_region", "sso_account_id", "sso_role_name"]:
-            value = config_session.get_scoped_config().get(key)
-            if not value:
-                missing.append(key)
-            else:
-                profile_config[key] = value
-        if missing:
-            raise click.UsageError(f"Profile {check_profile} is missing config fields {', '.join(missing)}")
+        config_session = Session(profile=check_profile)
+        def raise_missing_vars_error(missing: List[str]):
+            message = f"Profile {check_profile} is missing config fields {', '.join(missing)}"
+            raise click.UsageError(message)
+        sso_config = get_sso_config(
+            profile_config=config_session.get_scoped_config(),
+            sso_sessions=config_session.full_config.get("sso_sessions", {}),
+        ).validate(raise_missing_vars_error=raise_missing_vars_error)
 
         start_url_source = f"CLI-specified profile {check_profile}"
         region_source = f"CLI-specified profile {check_profile}"
 
-        sso_start_url = profile_config["sso_start_url"]
-        sso_region = profile_config["sso_region"]
-        account = profile_config["sso_account_id"]
-        role_name = profile_config["sso_role_name"]
+        sso_start_url = sso_config.sso_start_url
+        sso_region = sso_config.sso_region
+        account = sso_config.sso_account_id
+        role_name = sso_config.sso_role_name
         if verbose:
-            LOGGER.info(f"Configuration for profile {check_profile}: {json.dumps(profile_config)}")
+            LOGGER.info("Configuration for profile %s: %s", check_profile, sso_config)
         else:
             LOGGER.info(textwrap.dedent(f"""\
             Configuration for profile {check_profile}:
@@ -242,7 +239,7 @@ def check(
             sys.exit(201)
     else:
         try:
-            session = botocore.session.Session(session_vars={
+            session = Session(session_vars={
                 'profile': (None, None, None, None),
                 'region': (None, None, None, None),
             })

cli/src/aws_sso_util/configure_profile.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import argparse
 import os
 import subprocess
 import sys

cli/src/aws_sso_util/credential_process.py

@@ -15,19 +15,20 @@
 # This code is based on the code for the AWS CLI v2"s `aws sso login` functionality
 # https://github.com/aws/aws-cli/tree/v2/awscli/customizations/sso
 
-import argparse
 import os
 import sys
 import json
 import logging
 import datetime
+from typing import Dict, List, Optional
 
 from botocore.session import Session
 from botocore.exceptions import ClientError
 
 import click
 
 from aws_sso_lib.sso import get_credentials
+from aws_sso_lib.config import get_sso_config
 from aws_sso_lib.exceptions import InvalidSSOConfigError, AuthDispatchError, AuthenticationNeededError, UnauthorizedSSOTokenError
 
 LOG_FILE = os.path.expanduser(
@@ -36,34 +37,12 @@
 
 LOGGER = logging.getLogger(__name__)
 
-CONFIG_VARS = [
-    ("start url", "sso_start_url"),
-    ("SSO region", "sso_region"),
-    ("account", "sso_account_id"),
-    ("role", "sso_role_name")
-]
-
-def get_config(arg_config, profile_config):
-    sso_config = {}
-    missing_vars = []
-    for friendly_name, config_var_name in CONFIG_VARS:
-        if arg_config.get(config_var_name):
-            sso_config[config_var_name] = arg_config[config_var_name]
-        elif config_var_name not in profile_config:
-            missing_vars.append((friendly_name, config_var_name))
-            sso_config[config_var_name] = None
-        else:
-            sso_config[config_var_name] = profile_config[config_var_name]
-
-    required_vars = ["sso_start_url", "sso_region", "sso_account_id", "sso_role_name"]
-
-    missing_requred_vars = [v[0] for v in missing_vars if v[1] in required_vars]
-    if missing_requred_vars:
-        raise InvalidSSOConfigError(
-            "Missing " + ", ".join(missing_requred_vars)
-        )
-    return sso_config
-
+REQUIRED_VAR_MAP: Dict[str, str] = {
+    "sso_start_url": "start url",
+    "sso_region": "SSO region",
+    "sso_account_id": "account",
+    "sso_role_name": "role"
+}
 
 @click.command("credential-process")
 @click.option("--profile", help="Extract settings from the given profile")
@@ -75,13 +54,13 @@ def get_config(arg_config, profile_config):
 @click.option("--force-refresh", is_flag=True, help="Do not reuse cached Identity Center token")
 @click.option( "--verbose", "-v", "--debug", count=True, help="Write to the debugging log file")
 def credential_process(
-        profile,
-        start_url,
-        region,
-        account_id,
-        role_name,
-        force_refresh,
-        verbose):
+        profile: Optional[str],
+        start_url: Optional[str],
+        region: Optional[str],
+        account_id: Optional[str],
+        role_name: Optional[str],
+        force_refresh: bool,
+        verbose: int):
     """Helper for AWS SDKs that don't yet support Identity Center.
 
     This is not a command you use directly.
@@ -125,12 +104,15 @@ def credential_process(
     if profile:
         session_kwargs["profile"] = profile
 
-    arg_config = {
-        "sso_start_url": start_url,
-        "sso_region": region,
-        "sso_role_name": role_name,
-        "sso_account_id": account_id,
-    }
+    arg_config: Dict[str, str] = {}
+    if start_url:
+        arg_config["sso_start_url"] = start_url
+    if region:
+        arg_config["sso_region"] = region
+    if role_name:
+        arg_config["sso_role_name"] = role_name
+    if account_id:
+        arg_config["sso_account_id"] = account_id
 
     LOGGER.info("CONFIG FROM ARGS: {}".format(json.dumps(arg_config)))
 
@@ -143,25 +125,28 @@ def credential_process(
         else:
             profile_config = {}
 
-        config = get_config(arg_config, profile_config)
-
-        LOGGER.info("CONFIG: {}".format(json.dumps(config)))
+        sso_sessions = session.full_config.get("sso_sessions", {})
+        def raise_missing_vars_error(missing_vars: List[str]) -> str:
+            message = f"Missing {', '.join([REQUIRED_VAR_MAP[var] for var in missing_vars])}"
+            raise InvalidSSOConfigError(message)
+        config = get_sso_config(profile_config, sso_sessions, arg_config).validate(
+            raise_missing_vars_error=raise_missing_vars_error
+        )
 
-        if (config.get("sso_interactive_auth") or "").lower() == "true":
-            raise InvalidSSOConfigError("Interactive auth has been removed. See https://github.com/benkehoe/aws-sso-credential-process/issues/4")
+        LOGGER.info("CONFIG: %s", config)
 
-        if not config["sso_account_id"]:
+        if not config.sso_account_id:
             raise InvalidSSOConfigError("Missing account id")
 
-        if not config["sso_role_name"]:
+        if not config.sso_role_name:
             raise InvalidSSOConfigError("Missing role")
 
         credentials = get_credentials(
             session=session,
-            start_url=config["sso_start_url"],
-            sso_region=config["sso_region"],
-            account_id=config["sso_account_id"],
-            role_name=config["sso_role_name"],
+            start_url=config.sso_start_url,
+            sso_region=config.sso_region,
+            account_id=config.sso_account_id,
+            role_name=config.sso_role_name,
             force_refresh=force_refresh,
         )
 

cli/src/aws_sso_util/deploy_macro.py

@@ -11,7 +11,6 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 
-import argparse
 import subprocess
 import tempfile
 import sys

cli/src/aws_sso_util/lookup.py

@@ -11,7 +11,6 @@
 # ANY KIND, either express or implied. See the License for the specific
 # language governing permissions and limitations under the License.
 
-import argparse
 import sys
 import os
 from collections import namedtuple

cli/src/aws_sso_util/populate_profiles.py

@@ -13,7 +13,6 @@
 
 import sys
 import os
-import argparse
 import logging
 import json
 import subprocess