dkirillov/s3-tests
1
0
Fork 0
forked from TrueCloudLab/s3-tests
Code Pull requests Activity

Merge pull request #428 from TRYTOBE8TME/wip-sts-error

Browse source 
s3tests_boto3/test_sts: Changing code for proper cleanup
This commit is contained in:
Casey Bodley 2022-01-18 09:18:50 -05:00 committed by GitHub
commit 13a477d096
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 10 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
s3tests_boto3/functional/test_sts.py
View file

@ -85,7 +85,7 @@ def put_user_policy(iam_client,username,policyname,policy_document):
role_response = iam_client.put_user_policy(UserName=username,PolicyName=policyname,PolicyDocument=policy_document) role_response = iam_client.put_user_policy(UserName=username,PolicyName=policyname,PolicyDocument=policy_document)
except ClientError as e: except ClientError as e:
role_err = e.response['Code'] role_err = e.response['Code']
return (role_err,role_response) return (role_err,role_response,policyname)
def get_s3_client_using_iam_creds(): def get_s3_client_using_iam_creds():
iam_access_key = get_iam_access_key() iam_access_key = get_iam_access_key()
@ -161,7 +161,7 @@ def test_get_session_token():
default_endpoint=get_config_endpoint() default_endpoint=get_config_endpoint()
user_policy = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Deny\",\"Action\":\"s3:*\",\"Resource\":[\"*\"],\"Condition\":{\"BoolIfExists\":{\"sts:authentication\":\"false\"}}},{\"Effect\":\"Allow\",\"Action\":\"sts:GetSessionToken\",\"Resource\":\"*\",\"Condition\":{\"BoolIfExists\":{\"sts:authentication\":\"false\"}}}]}" user_policy = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Deny\",\"Action\":\"s3:*\",\"Resource\":[\"*\"],\"Condition\":{\"BoolIfExists\":{\"sts:authentication\":\"false\"}}},{\"Effect\":\"Allow\",\"Action\":\"sts:GetSessionToken\",\"Resource\":\"*\",\"Condition\":{\"BoolIfExists\":{\"sts:authentication\":\"false\"}}}]}"
(resp_err,resp)=put_user_policy(iam_client,sts_user_id,None,user_policy) (resp_err,resp,policy_name)=put_user_policy(iam_client,sts_user_id,None,user_policy)
eq(resp['ResponseMetadata']['HTTPStatusCode'],200) eq(resp['ResponseMetadata']['HTTPStatusCode'],200)
response=sts_client.get_session_token() response=sts_client.get_session_token()
@ -175,9 +175,12 @@ def test_get_session_token():
region_name='', region_name='',
) )
bucket_name = get_new_bucket_name() bucket_name = get_new_bucket_name()
s3bucket = s3_client.create_bucket(Bucket=bucket_name) try:
eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200) s3bucket = s3_client.create_bucket(Bucket=bucket_name)
finish=s3_client.delete_bucket(Bucket=bucket_name) eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)
finish=s3_client.delete_bucket(Bucket=bucket_name)
finally: # clean up user policy even if create_bucket/delete_bucket fails
iam_client.delete_user_policy(UserName=sts_user_id,PolicyName=policy_name)
@attr(resource='get session token') @attr(resource='get session token')
@attr(method='get') @attr(method='get')
@ -194,7 +197,7 @@ def test_get_session_token_permanent_creds_denied():
s3_main_secret_key=get_main_aws_secret_key() s3_main_secret_key=get_main_aws_secret_key()
user_policy = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Deny\",\"Action\":\"s3:*\",\"Resource\":[\"*\"],\"Condition\":{\"BoolIfExists\":{\"sts:authentication\":\"false\"}}},{\"Effect\":\"Allow\",\"Action\":\"sts:GetSessionToken\",\"Resource\":\"*\",\"Condition\":{\"BoolIfExists\":{\"sts:authentication\":\"false\"}}}]}" user_policy = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Deny\",\"Action\":\"s3:*\",\"Resource\":[\"*\"],\"Condition\":{\"BoolIfExists\":{\"sts:authentication\":\"false\"}}},{\"Effect\":\"Allow\",\"Action\":\"sts:GetSessionToken\",\"Resource\":\"*\",\"Condition\":{\"BoolIfExists\":{\"sts:authentication\":\"false\"}}}]}"
(resp_err,resp)=put_user_policy(iam_client,sts_user_id,None,user_policy) (resp_err,resp,policy_name)=put_user_policy(iam_client,sts_user_id,None,user_policy)
eq(resp['ResponseMetadata']['HTTPStatusCode'],200) eq(resp['ResponseMetadata']['HTTPStatusCode'],200)
response=sts_client.get_session_token() response=sts_client.get_session_token()
@ -213,6 +216,7 @@ def test_get_session_token_permanent_creds_denied():
except ClientError as e: except ClientError as e:
s3bucket_error = e.response.get("Error", {}).get("Code") s3bucket_error = e.response.get("Error", {}).get("Code")
eq(s3bucket_error,'AccessDenied') eq(s3bucket_error,'AccessDenied')
iam_client.delete_user_policy(UserName=sts_user_id,PolicyName=policy_name)
@attr(resource='assume role') @attr(resource='assume role')
@attr(method='get') @attr(method='get')