diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index df63162..bf0b07e 100644 --- a/CONTRIBUTORS.md +++ b/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) diff --git a/cli/src/aws_sso_util/cfn.py b/cli/src/aws_sso_util/cfn.py index b671376..2a932af 100644 --- a/cli/src/aws_sso_util/cfn.py +++ b/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 diff --git a/cli/src/aws_sso_util/check.py b/cli/src/aws_sso_util/check.py index 62d4a64..de59483 100644 --- a/cli/src/aws_sso_util/check.py +++ b/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), }) diff --git a/cli/src/aws_sso_util/configure_profile.py b/cli/src/aws_sso_util/configure_profile.py index 3ec63cf..94b2e87 100644 --- a/cli/src/aws_sso_util/configure_profile.py +++ b/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 diff --git a/cli/src/aws_sso_util/credential_process.py b/cli/src/aws_sso_util/credential_process.py index 43cc191..892fc8e 100644 --- a/cli/src/aws_sso_util/credential_process.py +++ b/cli/src/aws_sso_util/credential_process.py @@ -15,12 +15,12 @@ # 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 @@ -28,6 +28,7 @@ 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, ) diff --git a/cli/src/aws_sso_util/deploy_macro.py b/cli/src/aws_sso_util/deploy_macro.py index 5414929..081f170 100644 --- a/cli/src/aws_sso_util/deploy_macro.py +++ b/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 diff --git a/cli/src/aws_sso_util/lookup.py b/cli/src/aws_sso_util/lookup.py index 0030e1c..0ecfc7c 100644 --- a/cli/src/aws_sso_util/lookup.py +++ b/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 diff --git a/cli/src/aws_sso_util/populate_profiles.py b/cli/src/aws_sso_util/populate_profiles.py index bcd28aa..1824cfb 100644 --- a/cli/src/aws_sso_util/populate_profiles.py +++ b/cli/src/aws_sso_util/populate_profiles.py @@ -13,7 +13,6 @@ import sys import os -import argparse import logging import json import subprocess diff --git a/lib/aws_sso_lib/config.py b/lib/aws_sso_lib/config.py index 0788521..ee1ca5a 100644 --- a/lib/aws_sso_lib/config.py +++ b/lib/aws_sso_lib/config.py @@ -11,13 +11,15 @@ # ANY KIND, either express or implied. See the License for the specific # language governing permissions and limitations under the License. +import json import os import logging import re from collections import namedtuple -from typing import Optional +from typing import Callable, Dict, List, NoReturn, Optional -import botocore +from aws_sso_lib.exceptions import InvalidSSOConfigError +from botocore.session import Session from botocore.exceptions import ProfileNotFound LOGGER = logging.getLogger(__name__) @@ -55,9 +57,86 @@ def __bool__(self): def to_strs(cls, instances, region=None): return ", ".join(i.to_str(region=region) for i in instances) -def _get_instance_from_profile(profile_name, scoped_config: dict, missing_ok=False) -> Optional[SSOInstance]: - start_url = scoped_config.get("sso_start_url") - region = scoped_config.get("sso_region") +class SSOValidatedConfig(namedtuple("SSOValidatedConfig", ["sso_start_url", "sso_region", "sso_account_id", "sso_role_name"])): + sso_start_url: str + sso_region: str + sso_account_id: str + sso_role_name: str + + REQUIRED_VARS = ["sso_start_url", "sso_region", "sso_account_id", "sso_role_name"] + + def __str__(self) -> str: + return json.dumps(self._asdict()) + +class SSOConfig(namedtuple("SSOConfig", ["sso_start_url", "sso_region", "sso_account_id", "sso_role_name"])): + """ + The SSO configuration for a profile. It resolves the sso_session indirection. + """ + sso_start_url: Optional[str] + sso_region: Optional[str] + sso_account_id: Optional[str] + sso_role_name: Optional[str] + + def missing_vars(self): + d = self._asdict() + return [var for var in SSOValidatedConfig.REQUIRED_VARS if d[var] is None] + + def validate(self, raise_missing_vars_error: Optional[Callable[[List[str]], NoReturn]]) -> SSOValidatedConfig: + missing_vars = self.missing_vars() + if missing_vars: + if raise_missing_vars_error: + raise_missing_vars_error(missing_vars) + else: + raise InvalidSSOConfigError(f"Missing {', '.join(missing_vars)}") + return SSOValidatedConfig( + sso_start_url=self.sso_start_url, + sso_region=self.sso_region, + sso_account_id=self.sso_account_id, + sso_role_name=self.sso_role_name, + ) + + def __str__(self) -> str: + return json.dumps(self._asdict()) + +def get_sso_config( + profile_config: dict, + sso_sessions: dict, + arg_config: Optional[Dict[str, str]] = None) -> SSOConfig: + """ + Get the SSO configuration for a profile. + It resolves the sso_session indirection. + + :param profile_config: The profile configuration + :param sso_sessions: The sso_session configurations + :param arg_config: The configuration from CLI, which overrides the profile configuration + """ + sso_session_name = profile_config.get("sso_session") + configs: List[dict] = [] + # in order of precedence + if arg_config is not None: + configs.append(arg_config) + configs.append(profile_config) + if sso_session_name is not None: + sso_session = sso_sessions.get(sso_session_name) + if isinstance(sso_session, dict): + configs.append(sso_session) + def get_value(name: str) -> Optional[str]: + return next((config[name] for config in configs if isinstance(config.get(name), str)), None) + return SSOConfig( + sso_start_url=get_value("sso_start_url"), + sso_region=get_value("sso_region"), + sso_account_id=get_value("sso_account_id"), + sso_role_name=get_value("sso_role_name"), + ) + +def _get_instance_from_profile( + profile_name: str, + scoped_config: dict, + sso_sessions: dict, + missing_ok=False) -> Optional[SSOInstance]: + sso_config = get_sso_config(scoped_config, sso_sessions) + start_url = sso_config.sso_start_url + region = sso_config.sso_region if not (start_url and region): if not missing_ok: LOGGER.debug(f"Did not find config in profile {profile_name}") @@ -69,7 +148,12 @@ def _get_instance_from_profile(profile_name, scoped_config: dict, missing_ok=Fal def _get_all_instances_from_config(full_config: dict): instances: dict = {} for profile_name, scoped_config in full_config.get("profiles", {}).items(): - instance = _get_instance_from_profile(profile_name, scoped_config, missing_ok=True) + instance = _get_instance_from_profile( + profile_name, + scoped_config, + full_config.get("sso_sessions", {}), + missing_ok=True + ) if not instance: continue if instance.start_url in instances and instance.region != instances[instance.start_url].region: @@ -96,13 +180,16 @@ def _find_instance_from_profile( region=None, region_source=None): try: - session = botocore.session.Session(profile=profile_name) - instance = _get_instance_from_profile(profile_name, session.get_scoped_config()) + session = Session(profile=profile_name) + instance = _get_instance_from_profile( + profile_name, + session.get_scoped_config(), + session.full_config.get("sso_sessions", {}), + ) except ProfileNotFound: return None - if not instance: - return None - _validate_instance(instance, profile_source, start_url, start_url_source, region, region_source) + if instance: + _validate_instance(instance, profile_source, start_url, start_url_source, region, region_source) return instance def _get_specifier( @@ -178,7 +265,7 @@ def find_all_instances( unique_instances.append(specifier) all_instances.append(specifier) - session = botocore.session.Session(session_vars={ + session = Session(session_vars={ 'profile': (None, None, None, None), 'region': (None, None, None, None), }) @@ -252,7 +339,7 @@ def find_instances( LOGGER.debug("Specifier has literal start URL and region, not searching for instances") return [specifier], specifier, [specifier] - session = botocore.session.Session(session_vars={ + session = Session(session_vars={ 'profile': (None, None, None, None), 'region': (None, None, None, None), })