in redshift_connector/iam_helper.py [0:0]
def set_iam_properties(info: RedshiftProperty) -> RedshiftProperty:
"""
Helper function to handle IAM connection properties and ensure required parameters are specified.
Parameters
"""
import pkg_resources
from packaging.version import Version
if info is None:
raise InterfaceError("Invalid connection property setting. info must be specified")
# Check for IAM keys and AuthProfile first
if info.auth_profile is not None:
if Version(pkg_resources.get_distribution("boto3").version) < Version("1.17.111"):
raise pkg_resources.VersionConflict(
"boto3 >= 1.17.111 required for authentication via Amazon Redshift authentication profile. "
"Please upgrade the installed version of boto3 to use this functionality."
)
if not all((info.access_key_id, info.secret_access_key, info.region)):
raise InterfaceError(
"Invalid connection property setting. access_key_id, secret_access_key, and region are required "
"for authentication via Redshift auth_profile"
)
else:
# info.put("region", info.region)
# info.put("endpoint_url", info.endpoint_url)
resp = IamHelper.read_auth_profile(
auth_profile=typing.cast(str, info.auth_profile),
iam_access_key_id=typing.cast(str, info.access_key_id),
iam_secret_key=typing.cast(str, info.secret_access_key),
iam_session_token=info.session_token,
info=info,
)
info.put_all(resp)
if info.is_serverless_host and info.iam:
raise ProgrammingError("This feature is not yet available")
# if Version(pkg_resources.get_distribution("boto3").version) <= Version("1.20.22"):
# raise pkg_resources.VersionConflict(
# "boto3 >= XXX required for authentication with Amazon Redshift serverless. "
# "Please upgrade the installed version of boto3 to use this functionality."
# )
if info.is_serverless_host:
info.set_account_id_from_host()
info.set_region_from_host()
# IAM requires an SSL connection to work.
# Make sure that is set to SSL level VERIFY_CA or higher.
if info.ssl is True:
if info.sslmode not in SupportedSSLMode.list():
info.put("sslmode", SupportedSSLMode.default())
_logger.debug(
"A non-supported value: {} was provides for sslmode. Falling back to default value: {}".format(
info.sslmode, SupportedSSLMode.default()
)
)
else:
info.put("sslmode", "")
if (info.ssl is False) and (info.iam is True):
raise InterfaceError("Invalid connection property setting. SSL must be enabled when using IAM")
if (info.iam is False) and (info.ssl_insecure is False):
raise InterfaceError("Invalid connection property setting. IAM must be enabled when using ssl_insecure")
elif (info.iam is False) and any(
(info.credentials_provider, info.access_key_id, info.secret_access_key, info.session_token, info.profile)
):
raise InterfaceError(
"Invalid connection property setting. IAM must be enabled when using credential_provider, "
"AWS credentials, Amazon Redshift authentication profile, or AWS profile"
)
elif info.iam is True:
_logger.debug("boto3 version: {}".format(Version(pkg_resources.get_distribution("boto3").version)))
_logger.debug("botocore version: {}".format(Version(pkg_resources.get_distribution("botocore").version)))
if info.cluster_identifier is None and not info.is_serverless_host:
raise InterfaceError(
"Invalid connection property setting. cluster_identifier must be provided when IAM is enabled"
)
if info.credentials_provider is not None:
if info.auth_profile is None and any(
(info.access_key_id, info.secret_access_key, info.session_token, info.profile)
):
raise InterfaceError(
"Invalid connection property setting. It is not valid to provide both Credentials provider and "
"AWS credentials or AWS profile"
)
elif not isinstance(info.credentials_provider, str):
raise InterfaceError(
"Invalid connection property setting. It is not valid to provide a non-string value to "
"credentials_provider."
)
elif info.profile is not None:
if info.auth_profile is None and any((info.access_key_id, info.secret_access_key, info.session_token)):
raise InterfaceError(
"Invalid connection property setting. It is not valid to provide any of access_key_id, "
"secret_access_key, or session_token when profile is provided"
)
elif info.access_key_id is not None:
if info.secret_access_key is not None:
pass
elif info.password != "":
info.put("secret_access_key", info.password)
_logger.debug("Value of password will be used for secret_access_key")
else:
raise InterfaceError(
"Invalid connection property setting. "
"secret access key must be provided in either secret_access_key or password field"
)
_logger.debug(
"AWS Credentials access_key_id: {} secret_access_key: {} session_token: {}".format(
bool(info.access_key_id), bool(info.secret_access_key), bool(info.session_token)
)
)
elif info.secret_access_key is not None:
raise InterfaceError(
"Invalid connection property setting. access_key_id is required when secret_access_key is "
"provided"
)
elif info.session_token is not None:
raise InterfaceError(
"Invalid connection property setting. access_key_id and secret_access_key are required when "
"session_token is provided"
)
if info.client_protocol_version not in ClientProtocolVersion.list():
raise InterfaceError(
"Invalid connection property setting. client_protocol_version must be in: {}".format(
ClientProtocolVersion.list()
)
)
if info.db_groups and info.force_lowercase:
info.put("db_groups", [group.lower() for group in info.db_groups])
if info.iam is True:
if info.cluster_identifier is None and not info.is_serverless_host:
raise InterfaceError(
"Invalid connection property setting. cluster_identifier must be provided when IAM is enabled"
)
IamHelper.set_iam_credentials(info)
return info