pk/atst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use _get_client everywhere

Browse Source
This commit is contained in:
richard-dds 2019-10-02 11:44:44 -04:00
parent 10aa1c9bff
commit 7a8b96d2a8
1 changed files with 22 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
atst/domain/csp/cloud.py
View File

@ -594,12 +594,7 @@ class AWSCloudProvider(CloudProviderInterface):
credentials = assumed_role_object["Credentials"] credentials = assumed_role_object["Credentials"]
# Use the temporary credentials that AssumeRole returns to make a new connection to IAM # Use the temporary credentials that AssumeRole returns to make a new connection to IAM
iam_client = self.boto3.client( iam_client = self._get_client("iam", credentials=credentials)
"iam",
aws_access_key_id=credentials["AccessKeyId"],
aws_secret_access_key=credentials["SecretAccessKey"],
aws_session_token=credentials["SessionToken"],
)
# Create the user with a PermissionBoundary # Create the user with a PermissionBoundary
permission_boundary_arn = "arn:aws:iam::aws:policy/AlexaForBusinessDeviceSetup" permission_boundary_arn = "arn:aws:iam::aws:policy/AlexaForBusinessDeviceSetup"
@ -671,16 +666,29 @@ class AWSCloudProvider(CloudProviderInterface):
def _get_client(self, service: str, credentials=None): def _get_client(self, service: str, credentials=None):
""" """
A helper for creating a client of a given AWS service. A helper for creating a client of a given AWS service.
"""
credentials = credentials or { If `credentials` aren't provided, the configured root credentials will be used.
"AccessKeyId": self.access_key_id,
"SecretAccessKey": self.secret_key, `credentials` format:
{
"AccessKeyId": "access-key-id",
"SecretAccessKey": "secret-access-key",
"SessionToken": "session-token" # optional
} }
"""
credentials = credentials or {}
credential_kwargs = {
"aws_access_key_id": credentials.get("AccessKeyId", self.access_key_id),
"aws_secret_access_key": credentials.get(
"SecretAccessKey", self.secret_key
),
}
if "SessionToken" in credentials:
credential_kwargs["aws_session_token"] = credentials["SessionToken"]
return self.boto3.client( return self.boto3.client(
service, service, region_name=self.region_name, **credential_kwargs
aws_access_key_id=credentials["AccessKeyId"],
aws_secret_access_key=credentials["SecretAccessKey"],
region_name=self.region_name,
) )
def _inline_org_management_policy(self, account_id: str) -> Dict: def _inline_org_management_policy(self, account_id: str) -> Dict: