How to use assume_role_with_web_identity method in localstack

Best Python code snippet using localstack_python

test_sts.py

Source:test_sts.py Github

copy

Full Screen

...394@attr(assertion='assuming role through web token')395@attr('webidentity_test')396@attr('token_claims_trust_policy_test')397@attr('fails_on_dbstore')398def test_assume_role_with_web_identity():399 check_webidentity()400 iam_client=get_iam_client() 401 sts_client=get_sts_client()402 default_endpoint=get_config_endpoint()403 role_session_name=get_parameter_name()404 thumbprint=get_thumbprint()405 aud=get_aud()406 token=get_token()407 realm=get_realm_name()408 409 oidc_response = iam_client.create_open_id_connect_provider(410 Url='http://localhost:8080/auth/realms/{}'.format(realm),411 ThumbprintList=[412 thumbprint,413 ],414 )415 416 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"417 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)418 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')419 420 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\"}}"421 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)422 eq(response['ResponseMetadata']['HTTPStatusCode'],200)423 424 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token)425 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)426 427 s3_client = boto3.client('s3',428 aws_access_key_id = resp['Credentials']['AccessKeyId'],429 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],430 aws_session_token = resp['Credentials']['SessionToken'],431 endpoint_url=default_endpoint,432 region_name='',433 )434 bucket_name = get_new_bucket_name()435 s3bucket = s3_client.create_bucket(Bucket=bucket_name)436 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)437 bkt = s3_client.delete_bucket(Bucket=bucket_name)438 eq(bkt['ResponseMetadata']['HTTPStatusCode'],204)439 440 oidc_remove=iam_client.delete_open_id_connect_provider(441 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]442 )443'''444@attr(resource='assume role with web identity')445@attr(method='get')446@attr(operation='check')447@attr(assertion='assume_role_with_web_token creds expire')448@attr('webidentity_test')449def test_assume_role_with_web_identity_invalid_webtoken():450 resp_error=None451 iam_client=get_iam_client()452 sts_client=get_sts_client()453 default_endpoint=get_config_endpoint()454 role_session_name=get_parameter_name()455 thumbprint=get_thumbprint()456 aud=get_aud()457 token=get_token()458 realm=get_realm_name()459 oidc_response = iam_client.create_open_id_connect_provider(460 Url='http://localhost:8080/auth/realms/{}'.format(realm),461 ThumbprintList=[462 thumbprint,463 ],464 )465 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"466 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)467 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')468 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\"}}"469 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)470 eq(response['ResponseMetadata']['HTTPStatusCode'],200)471 resp=""472 try:473 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken='abcdef')474 except InvalidIdentityTokenException as e:475 log.debug('{}'.format(resp))476 log.debug('{}'.format(e.response.get("Error", {}).get("Code")))477 log.debug('{}'.format(e))478 resp_error = e.response.get("Error", {}).get("Code")479 eq(resp_error,'AccessDenied')480 oidc_remove=iam_client.delete_open_id_connect_provider(481 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]482 )483'''484#######################485# Session Policy Tests486#######################487@attr(resource='assume role with web identity')488@attr(method='get')489@attr(operation='check')490@attr(assertion='checking session policy working for two different buckets')491@attr('webidentity_test')492@attr('session_policy')493@attr('fails_on_dbstore')494def test_session_policy_check_on_different_buckets():495 check_webidentity()496 iam_client=get_iam_client()497 sts_client=get_sts_client()498 default_endpoint=get_config_endpoint()499 role_session_name=get_parameter_name()500 thumbprint=get_thumbprint()501 aud=get_aud()502 token=get_token()503 realm=get_realm_name()504 url = 'http://localhost:8080/auth/realms/{}'.format(realm)505 thumbprintlist = [thumbprint]506 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)507 if oidc_error is not None:508 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))509 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"510 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)511 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')512 role_policy_new = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":[\"arn:aws:s3:::test2\",\"arn:aws:s3:::test2/*\"]}}"513 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy_new)514 eq(response['ResponseMetadata']['HTTPStatusCode'],200)515 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:GetObject\",\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"516 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)517 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)518 s3_client = boto3.client('s3',519 aws_access_key_id = resp['Credentials']['AccessKeyId'],520 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],521 aws_session_token = resp['Credentials']['SessionToken'],522 endpoint_url=default_endpoint,523 region_name='',524 )525 bucket_name_1 = 'test1'526 try:527 s3bucket = s3_client.create_bucket(Bucket=bucket_name_1)528 except ClientError as e:529 s3bucket_error = e.response.get("Error", {}).get("Code")530 eq(s3bucket_error, 'AccessDenied')531 bucket_name_2 = 'test2'532 try:533 s3bucket = s3_client.create_bucket(Bucket=bucket_name_2)534 except ClientError as e:535 s3bucket_error = e.response.get("Error", {}).get("Code")536 eq(s3bucket_error, 'AccessDenied')537 bucket_body = 'please-write-something'538 #body.encode(encoding='utf_8')539 try:540 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")541 except ClientError as e:542 s3_put_obj_error = e.response.get("Error", {}).get("Code")543 eq(s3_put_obj_error,'NoSuchBucket')544 oidc_remove=iam_client.delete_open_id_connect_provider(545 OpenIDConnectProviderArn=oidc_arn546 )547@attr(resource='assume role with web identity')548@attr(method='put')549@attr(operation='check')550@attr(assertion='checking session policy working for same bucket')551@attr('webidentity_test')552@attr('session_policy')553@attr('fails_on_dbstore')554def test_session_policy_check_on_same_bucket():555 check_webidentity()556 iam_client=get_iam_client()557 sts_client=get_sts_client()558 default_endpoint=get_config_endpoint()559 role_session_name=get_parameter_name()560 thumbprint=get_thumbprint()561 aud=get_aud()562 token=get_token()563 realm=get_realm_name()564 url = 'http://localhost:8080/auth/realms/{}'.format(realm)565 thumbprintlist = [thumbprint]566 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)567 if oidc_error is not None:568 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))569 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"570 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)571 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')572 role_policy_new = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":[\"*\"]}}"573 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy_new)574 eq(response['ResponseMetadata']['HTTPStatusCode'],200)575 s3_client_iam_creds = get_s3_client_using_iam_creds()576 bucket_name_1 = 'test1'577 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name_1)578 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)579 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:GetObject\",\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"580 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)581 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)582 s3_client = boto3.client('s3',583 aws_access_key_id = resp['Credentials']['AccessKeyId'],584 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],585 aws_session_token = resp['Credentials']['SessionToken'],586 endpoint_url=default_endpoint,587 region_name='',588 )589 bucket_body = 'this is a test file'590 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")591 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)592 oidc_remove=iam_client.delete_open_id_connect_provider(593 OpenIDConnectProviderArn=oidc_arn594 )595@attr(resource='assume role with web identity')596@attr(method='get')597@attr(operation='check')598@attr(assertion='checking put_obj op denial')599@attr('webidentity_test')600@attr('session_policy')601@attr('fails_on_dbstore')602def test_session_policy_check_put_obj_denial():603 check_webidentity()604 iam_client=get_iam_client()605 iam_access_key=get_iam_access_key()606 iam_secret_key=get_iam_secret_key()607 sts_client=get_sts_client()608 default_endpoint=get_config_endpoint()609 role_session_name=get_parameter_name()610 thumbprint=get_thumbprint()611 aud=get_aud()612 token=get_token()613 realm=get_realm_name()614 url = 'http://localhost:8080/auth/realms/{}'.format(realm)615 thumbprintlist = [thumbprint]616 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)617 if oidc_error is not None:618 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))619 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"620 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)621 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')622 role_policy_new = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":[\"*\"]}}"623 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy_new)624 eq(response['ResponseMetadata']['HTTPStatusCode'],200)625 s3_client_iam_creds = get_s3_client_using_iam_creds()626 bucket_name_1 = 'test1'627 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name_1)628 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)629 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:GetObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"630 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)631 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)632 s3_client = boto3.client('s3',633 aws_access_key_id = resp['Credentials']['AccessKeyId'],634 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],635 aws_session_token = resp['Credentials']['SessionToken'],636 endpoint_url=default_endpoint,637 region_name='',638 )639 bucket_body = 'this is a test file'640 try:641 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")642 except ClientError as e:643 s3_put_obj_error = e.response.get("Error", {}).get("Code")644 eq(s3_put_obj_error, 'AccessDenied')645 oidc_remove=iam_client.delete_open_id_connect_provider(646 OpenIDConnectProviderArn=oidc_arn647 )648@attr(resource='assume role with web identity')649@attr(method='get')650@attr(operation='check')651@attr(assertion='checking put_obj working by swapping policies')652@attr('webidentity_test')653@attr('session_policy')654@attr('fails_on_dbstore')655def test_swapping_role_policy_and_session_policy():656 check_webidentity()657 iam_client=get_iam_client()658 iam_access_key=get_iam_access_key()659 iam_secret_key=get_iam_secret_key()660 sts_client=get_sts_client()661 default_endpoint=get_config_endpoint()662 role_session_name=get_parameter_name()663 thumbprint=get_thumbprint()664 aud=get_aud()665 token=get_token()666 realm=get_realm_name()667 url = 'http://localhost:8080/auth/realms/{}'.format(realm)668 thumbprintlist = [thumbprint]669 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)670 if oidc_error is not None:671 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))672 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"673 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)674 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')675 role_policy_new = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:GetObject\",\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"676 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy_new)677 eq(response['ResponseMetadata']['HTTPStatusCode'],200)678 s3_client_iam_creds = get_s3_client_using_iam_creds()679 bucket_name_1 = 'test1'680 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name_1)681 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)682 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":[\"*\"]}}"683 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)684 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)685 s3_client = boto3.client('s3',686 aws_access_key_id = resp['Credentials']['AccessKeyId'],687 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],688 aws_session_token = resp['Credentials']['SessionToken'],689 endpoint_url=default_endpoint,690 region_name='',691 )692 bucket_body = 'this is a test file'693 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")694 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)695 oidc_remove=iam_client.delete_open_id_connect_provider(696 OpenIDConnectProviderArn=oidc_arn697 )698@attr(resource='assume role with web identity')699@attr(method='put')700@attr(operation='check')701@attr(assertion='checking put_obj working by setting different permissions to role and session policy')702@attr('webidentity_test')703@attr('session_policy')704@attr('fails_on_dbstore')705def test_session_policy_check_different_op_permissions():706 check_webidentity()707 iam_client=get_iam_client()708 iam_access_key=get_iam_access_key()709 iam_secret_key=get_iam_secret_key()710 sts_client=get_sts_client()711 default_endpoint=get_config_endpoint()712 role_session_name=get_parameter_name()713 thumbprint=get_thumbprint()714 aud=get_aud()715 token=get_token()716 realm=get_realm_name()717 url = 'http://localhost:8080/auth/realms/{}'.format(realm)718 thumbprintlist = [thumbprint]719 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)720 if oidc_error is not None:721 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))722 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"723 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)724 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')725 role_policy_new = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"726 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy_new)727 eq(response['ResponseMetadata']['HTTPStatusCode'],200)728 s3_client_iam_creds = get_s3_client_using_iam_creds()729 bucket_name_1 = 'test1'730 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name_1)731 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)732 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:GetObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"733 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)734 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)735 s3_client = boto3.client('s3',736 aws_access_key_id = resp['Credentials']['AccessKeyId'],737 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],738 aws_session_token = resp['Credentials']['SessionToken'],739 endpoint_url=default_endpoint,740 region_name='',741 )742 bucket_body = 'this is a test file'743 try:744 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")745 except ClientError as e:746 s3_put_obj_error = e.response.get("Error", {}).get("Code")747 eq(s3_put_obj_error, 'AccessDenied')748 oidc_remove=iam_client.delete_open_id_connect_provider(749 OpenIDConnectProviderArn=oidc_arn750 )751@attr(resource='assume role with web identity')752@attr(method='put')753@attr(operation='check')754@attr(assertion='checking op behaviour with deny effect')755@attr('webidentity_test')756@attr('session_policy')757@attr('fails_on_dbstore')758def test_session_policy_check_with_deny_effect():759 check_webidentity()760 iam_client=get_iam_client()761 iam_access_key=get_iam_access_key()762 iam_secret_key=get_iam_secret_key()763 sts_client=get_sts_client()764 default_endpoint=get_config_endpoint()765 role_session_name=get_parameter_name()766 thumbprint=get_thumbprint()767 aud=get_aud()768 token=get_token()769 realm=get_realm_name()770 url = 'http://localhost:8080/auth/realms/{}'.format(realm)771 thumbprintlist = [thumbprint]772 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)773 if oidc_error is not None:774 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))775 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"776 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)777 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')778 role_policy_new = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Deny\",\"Action\":\"s3:*\",\"Resource\":[\"*\"]}}"779 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy_new)780 eq(response['ResponseMetadata']['HTTPStatusCode'],200)781 s3_client_iam_creds = get_s3_client_using_iam_creds()782 bucket_name_1 = 'test1'783 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name_1)784 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)785 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"786 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)787 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)788 s3_client = boto3.client('s3',789 aws_access_key_id = resp['Credentials']['AccessKeyId'],790 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],791 aws_session_token = resp['Credentials']['SessionToken'],792 endpoint_url=default_endpoint,793 region_name='',794 )795 bucket_body = 'this is a test file'796 try:797 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")798 except ClientError as e:799 s3_put_obj_error = e.response.get("Error", {}).get("Code")800 eq(s3_put_obj_error, 'AccessDenied')801 oidc_remove=iam_client.delete_open_id_connect_provider(802 OpenIDConnectProviderArn=oidc_arn803 )804@attr(resource='assume role with web identity')805@attr(method='put')806@attr(operation='check')807@attr(assertion='checking put_obj working with deny and allow on same op')808@attr('webidentity_test')809@attr('session_policy')810@attr('fails_on_dbstore')811def test_session_policy_check_with_deny_on_same_op():812 check_webidentity()813 iam_client=get_iam_client()814 iam_access_key=get_iam_access_key()815 iam_secret_key=get_iam_secret_key()816 sts_client=get_sts_client()817 default_endpoint=get_config_endpoint()818 role_session_name=get_parameter_name()819 thumbprint=get_thumbprint()820 aud=get_aud()821 token=get_token()822 realm=get_realm_name()823 url = 'http://localhost:8080/auth/realms/{}'.format(realm)824 thumbprintlist = [thumbprint]825 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)826 if oidc_error is not None:827 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))828 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"829 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)830 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')831 role_policy_new = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"832 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy_new)833 eq(response['ResponseMetadata']['HTTPStatusCode'],200)834 s3_client_iam_creds = get_s3_client_using_iam_creds()835 bucket_name_1 = 'test1'836 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name_1)837 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)838 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Deny\",\"Action\":[\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"839 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)840 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)841 s3_client = boto3.client('s3',842 aws_access_key_id = resp['Credentials']['AccessKeyId'],843 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],844 aws_session_token = resp['Credentials']['SessionToken'],845 endpoint_url=default_endpoint,846 region_name='',847 )848 bucket_body = 'this is a test file'849 try:850 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")851 except ClientError as e:852 s3_put_obj_error = e.response.get("Error", {}).get("Code")853 eq(s3_put_obj_error, 'AccessDenied')854 oidc_remove=iam_client.delete_open_id_connect_provider(855 OpenIDConnectProviderArn=oidc_arn856 )857@attr(resource='assume role with web identity')858@attr(method='put')859@attr(operation='check')860@attr(assertion='checking op when bucket policy has role arn')861@attr('webidentity_test')862@attr('session_policy')863@attr('fails_on_dbstore')864def test_session_policy_bucket_policy_role_arn():865 check_webidentity()866 iam_client=get_iam_client()867 sts_client=get_sts_client()868 default_endpoint=get_config_endpoint()869 role_session_name=get_parameter_name()870 thumbprint=get_thumbprint()871 aud=get_aud()872 token=get_token()873 realm=get_realm_name()874 url = 'http://localhost:8080/auth/realms/{}'.format(realm)875 thumbprintlist = [thumbprint]876 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)877 if oidc_error is not None:878 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))879 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"880 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)881 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')882 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":[\"*\"]}}"883 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)884 eq(response['ResponseMetadata']['HTTPStatusCode'],200)885 s3client_iamcreds = get_s3_client_using_iam_creds()886 bucket_name_1 = 'test1'887 s3bucket = s3client_iamcreds.create_bucket(Bucket=bucket_name_1)888 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)889 resource1 = "arn:aws:s3:::" + bucket_name_1890 resource2 = "arn:aws:s3:::" + bucket_name_1 + "/*"891 rolearn = "arn:aws:iam:::role/" + general_role_name892 bucket_policy = json.dumps(893 {894 "Version": "2012-10-17",895 "Statement": [{896 "Effect": "Allow",897 "Principal": {"AWS": "{}".format(rolearn)},898 "Action": ["s3:GetObject","s3:PutObject"],899 "Resource": [900 "{}".format(resource1),901 "{}".format(resource2)902 ]903 }]904 })905 s3client_iamcreds.put_bucket_policy(Bucket=bucket_name_1, Policy=bucket_policy)906 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"907 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)908 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)909 s3_client = boto3.client('s3',910 aws_access_key_id = resp['Credentials']['AccessKeyId'],911 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],912 aws_session_token = resp['Credentials']['SessionToken'],913 endpoint_url=default_endpoint,914 region_name='',915 )916 bucket_body = 'this is a test file'917 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")918 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)919 try:920 obj = s3_client.get_object(Bucket=bucket_name_1, Key="test-1.txt")921 except ClientError as e:922 s3object_error = e.response.get("Error", {}).get("Code")923 eq(s3object_error, 'AccessDenied')924 oidc_remove=iam_client.delete_open_id_connect_provider(925 OpenIDConnectProviderArn=oidc_arn926 )927@attr(resource='assume role with web identity')928@attr(method='get')929@attr(operation='check')930@attr(assertion='checking op when bucket policy has session arn')931@attr('webidentity_test')932@attr('session_policy')933@attr('fails_on_dbstore')934def test_session_policy_bucket_policy_session_arn():935 check_webidentity()936 iam_client=get_iam_client()937 sts_client=get_sts_client()938 default_endpoint=get_config_endpoint()939 role_session_name=get_parameter_name()940 thumbprint=get_thumbprint()941 aud=get_aud()942 token=get_token()943 realm=get_realm_name()944 url = 'http://localhost:8080/auth/realms/{}'.format(realm)945 thumbprintlist = [thumbprint]946 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)947 if oidc_error is not None:948 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))949 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"950 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)951 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')952 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":[\"*\"]}}"953 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)954 eq(response['ResponseMetadata']['HTTPStatusCode'],200)955 s3client_iamcreds = get_s3_client_using_iam_creds()956 bucket_name_1 = 'test1'957 s3bucket = s3client_iamcreds.create_bucket(Bucket=bucket_name_1)958 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)959 resource1 = "arn:aws:s3:::" + bucket_name_1960 resource2 = "arn:aws:s3:::" + bucket_name_1 + "/*"961 rolesessionarn = "arn:aws:iam:::assumed-role/" + general_role_name + "/" + role_session_name962 bucket_policy = json.dumps(963 {964 "Version": "2012-10-17",965 "Statement": [{966 "Effect": "Allow",967 "Principal": {"AWS": "{}".format(rolesessionarn)},968 "Action": ["s3:GetObject","s3:PutObject"],969 "Resource": [970 "{}".format(resource1),971 "{}".format(resource2)972 ]973 }]974 })975 s3client_iamcreds.put_bucket_policy(Bucket=bucket_name_1, Policy=bucket_policy)976 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"977 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)978 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)979 s3_client = boto3.client('s3',980 aws_access_key_id = resp['Credentials']['AccessKeyId'],981 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],982 aws_session_token = resp['Credentials']['SessionToken'],983 endpoint_url=default_endpoint,984 region_name='',985 )986 bucket_body = 'this is a test file'987 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")988 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)989 s3_get_obj = s3_client.get_object(Bucket=bucket_name_1, Key="test-1.txt")990 eq(s3_get_obj['ResponseMetadata']['HTTPStatusCode'],200)991 oidc_remove=iam_client.delete_open_id_connect_provider(992 OpenIDConnectProviderArn=oidc_arn993 )994@attr(resource='assume role with web identity')995@attr(method='put')996@attr(operation='check')997@attr(assertion='checking copy object op with role, session and bucket policy')998@attr('webidentity_test')999@attr('session_policy')1000@attr('fails_on_dbstore')1001def test_session_policy_copy_object():1002 check_webidentity()1003 iam_client=get_iam_client()1004 sts_client=get_sts_client()1005 default_endpoint=get_config_endpoint()1006 role_session_name=get_parameter_name()1007 thumbprint=get_thumbprint()1008 aud=get_aud()1009 token=get_token()1010 realm=get_realm_name()1011 url = 'http://localhost:8080/auth/realms/{}'.format(realm)1012 thumbprintlist = [thumbprint]1013 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)1014 if oidc_error is not None:1015 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))1016 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"1017 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1018 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1019 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":[\"*\"]}}"1020 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1021 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1022 s3client_iamcreds = get_s3_client_using_iam_creds()1023 bucket_name_1 = 'test1'1024 s3bucket = s3client_iamcreds.create_bucket(Bucket=bucket_name_1)1025 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1026 resource1 = "arn:aws:s3:::" + bucket_name_11027 resource2 = "arn:aws:s3:::" + bucket_name_1 + "/*"1028 rolesessionarn = "arn:aws:iam:::assumed-role/" + general_role_name + "/" + role_session_name1029 print (rolesessionarn)1030 bucket_policy = json.dumps(1031 {1032 "Version": "2012-10-17",1033 "Statement": [{1034 "Effect": "Allow",1035 "Principal": {"AWS": "{}".format(rolesessionarn)},1036 "Action": ["s3:GetObject","s3:PutObject"],1037 "Resource": [1038 "{}".format(resource1),1039 "{}".format(resource2)1040 ]1041 }]1042 })1043 s3client_iamcreds.put_bucket_policy(Bucket=bucket_name_1, Policy=bucket_policy)1044 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"1045 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)1046 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1047 s3_client = boto3.client('s3',1048 aws_access_key_id = resp['Credentials']['AccessKeyId'],1049 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1050 aws_session_token = resp['Credentials']['SessionToken'],1051 endpoint_url=default_endpoint,1052 region_name='',1053 )1054 bucket_body = 'this is a test file'1055 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")1056 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)1057 copy_source = {1058 'Bucket': bucket_name_1,1059 'Key': 'test-1.txt'1060 }1061 s3_client.copy(copy_source, bucket_name_1, "test-2.txt")1062 s3_get_obj = s3_client.get_object(Bucket=bucket_name_1, Key="test-2.txt")1063 eq(s3_get_obj['ResponseMetadata']['HTTPStatusCode'],200)1064 oidc_remove=iam_client.delete_open_id_connect_provider(1065 OpenIDConnectProviderArn=oidc_arn1066 )1067@attr(resource='assume role with web identity')1068@attr(method='put')1069@attr(operation='check')1070@attr(assertion='checking op is denied when no role policy')1071@attr('webidentity_test')1072@attr('session_policy')1073@attr('fails_on_dbstore')1074def test_session_policy_no_bucket_role_policy():1075 check_webidentity()1076 iam_client=get_iam_client()1077 sts_client=get_sts_client()1078 default_endpoint=get_config_endpoint()1079 role_session_name=get_parameter_name()1080 thumbprint=get_thumbprint()1081 aud=get_aud()1082 token=get_token()1083 realm=get_realm_name()1084 url = 'http://localhost:8080/auth/realms/{}'.format(realm)1085 thumbprintlist = [thumbprint]1086 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)1087 if oidc_error is not None:1088 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))1089 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"1090 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1091 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1092 s3client_iamcreds = get_s3_client_using_iam_creds()1093 bucket_name_1 = 'test1'1094 s3bucket = s3client_iamcreds.create_bucket(Bucket=bucket_name_1)1095 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1096 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:PutObject\",\"s3:GetObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"1097 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)1098 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1099 s3_client = boto3.client('s3',1100 aws_access_key_id = resp['Credentials']['AccessKeyId'],1101 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1102 aws_session_token = resp['Credentials']['SessionToken'],1103 endpoint_url=default_endpoint,1104 region_name='',1105 )1106 bucket_body = 'this is a test file'1107 try:1108 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")1109 except ClientError as e:1110 s3putobj_error = e.response.get("Error", {}).get("Code")1111 eq(s3putobj_error, 'AccessDenied')1112 oidc_remove=iam_client.delete_open_id_connect_provider(1113 OpenIDConnectProviderArn=oidc_arn1114 )1115@attr(resource='assume role with web identity')1116@attr(method='put')1117@attr(operation='check')1118@attr(assertion='checking op is denied when resource policy denies')1119@attr('webidentity_test')1120@attr('session_policy')1121@attr('fails_on_dbstore')1122def test_session_policy_bucket_policy_deny():1123 check_webidentity()1124 iam_client=get_iam_client()1125 sts_client=get_sts_client()1126 default_endpoint=get_config_endpoint()1127 role_session_name=get_parameter_name()1128 thumbprint=get_thumbprint()1129 aud=get_aud()1130 token=get_token()1131 realm=get_realm_name()1132 url = 'http://localhost:8080/auth/realms/{}'.format(realm)1133 thumbprintlist = [thumbprint]1134 (oidc_arn,oidc_error) = create_oidc_provider(iam_client, url, None, thumbprintlist)1135 if oidc_error is not None:1136 raise RuntimeError('Unable to create/get openid connect provider {}'.format(oidc_error))1137 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_arn+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":app_id\":\""+aud+"\"}}}]}"1138 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1139 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1140 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":[\"*\"]}}"1141 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1142 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1143 s3client_iamcreds = get_s3_client_using_iam_creds()1144 bucket_name_1 = 'test1'1145 s3bucket = s3client_iamcreds.create_bucket(Bucket=bucket_name_1)1146 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1147 resource1 = "arn:aws:s3:::" + bucket_name_11148 resource2 = "arn:aws:s3:::" + bucket_name_1 + "/*"1149 rolesessionarn = "arn:aws:iam:::assumed-role/" + general_role_name + "/" + role_session_name1150 bucket_policy = json.dumps(1151 {1152 "Version": "2012-10-17",1153 "Statement": [{1154 "Effect": "Deny",1155 "Principal": {"AWS": "{}".format(rolesessionarn)},1156 "Action": ["s3:GetObject","s3:PutObject"],1157 "Resource": [1158 "{}".format(resource1),1159 "{}".format(resource2)1160 ]1161 }]1162 })1163 s3client_iamcreds.put_bucket_policy(Bucket=bucket_name_1, Policy=bucket_policy)1164 session_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":[\"s3:PutObject\"],\"Resource\":[\"arn:aws:s3:::test1\",\"arn:aws:s3:::test1/*\"]}}"1165 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token,Policy=session_policy)1166 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1167 s3_client = boto3.client('s3',1168 aws_access_key_id = resp['Credentials']['AccessKeyId'],1169 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1170 aws_session_token = resp['Credentials']['SessionToken'],1171 endpoint_url=default_endpoint,1172 region_name='',1173 )1174 bucket_body = 'this is a test file'1175 try:1176 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name_1, Key="test-1.txt")1177 except ClientError as e:1178 s3putobj_error = e.response.get("Error", {}).get("Code")1179 eq(s3putobj_error, 'AccessDenied')1180 oidc_remove=iam_client.delete_open_id_connect_provider(1181 OpenIDConnectProviderArn=oidc_arn1182 )1183@attr(resource='assume role with web identity')1184@attr(method='get')1185@attr(operation='check')1186@attr(assertion='assuming role using web token using sub in trust policy')1187@attr('webidentity_test')1188@attr('token_claims_trust_policy_test')1189@attr('fails_on_dbstore')1190def test_assume_role_with_web_identity_with_sub():1191 check_webidentity()1192 iam_client=get_iam_client()1193 sts_client=get_sts_client()1194 default_endpoint=get_config_endpoint()1195 role_session_name=get_parameter_name()1196 thumbprint=get_thumbprint()1197 sub=get_sub()1198 token=get_token()1199 realm=get_realm_name()1200 oidc_response = iam_client.create_open_id_connect_provider(1201 Url='http://localhost:8080/auth/realms/{}'.format(realm),1202 ThumbprintList=[1203 thumbprint,1204 ],1205 )1206 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":sub\":\""+sub+"\"}}}]}"1207 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1208 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1209 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\"}}"1210 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1211 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1212 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token)1213 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1214 s3_client = boto3.client('s3',1215 aws_access_key_id = resp['Credentials']['AccessKeyId'],1216 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1217 aws_session_token = resp['Credentials']['SessionToken'],1218 endpoint_url=default_endpoint,1219 region_name='',1220 )1221 bucket_name = get_new_bucket_name()1222 s3bucket = s3_client.create_bucket(Bucket=bucket_name)1223 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1224 bkt = s3_client.delete_bucket(Bucket=bucket_name)1225 eq(bkt['ResponseMetadata']['HTTPStatusCode'],204)1226 oidc_remove=iam_client.delete_open_id_connect_provider(1227 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1228 )1229@attr(resource='assume role with web identity')1230@attr(method='get')1231@attr(operation='check')1232@attr(assertion='assuming role using web token using azp in trust policy')1233@attr('webidentity_test')1234@attr('token_claims_trust_policy_test')1235@attr('fails_on_dbstore')1236def test_assume_role_with_web_identity_with_azp():1237 check_webidentity()1238 iam_client=get_iam_client()1239 sts_client=get_sts_client()1240 default_endpoint=get_config_endpoint()1241 role_session_name=get_parameter_name()1242 thumbprint=get_thumbprint()1243 azp=get_azp()1244 token=get_token()1245 realm=get_realm_name()1246 oidc_response = iam_client.create_open_id_connect_provider(1247 Url='http://localhost:8080/auth/realms/{}'.format(realm),1248 ThumbprintList=[1249 thumbprint,1250 ],1251 )1252 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\"],\"Condition\":{\"StringEquals\":{\"localhost:8080/auth/realms/"+realm+":azp\":\""+azp+"\"}}}]}"1253 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1254 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1255 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\"}}"1256 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1257 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1258 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=token)1259 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1260 s3_client = boto3.client('s3',1261 aws_access_key_id = resp['Credentials']['AccessKeyId'],1262 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1263 aws_session_token = resp['Credentials']['SessionToken'],1264 endpoint_url=default_endpoint,1265 region_name='',1266 )1267 bucket_name = get_new_bucket_name()1268 s3bucket = s3_client.create_bucket(Bucket=bucket_name)1269 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1270 bkt = s3_client.delete_bucket(Bucket=bucket_name)1271 eq(bkt['ResponseMetadata']['HTTPStatusCode'],204)1272 oidc_remove=iam_client.delete_open_id_connect_provider(1273 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1274 )1275@attr(resource='assume role with web identity')1276@attr(method='get')1277@attr(operation='check')1278@attr(assertion='assuming role using web token using aws:RequestTag in trust policy')1279@attr('webidentity_test')1280@attr('abac_test')1281@attr('token_request_tag_trust_policy_test')1282@attr('fails_on_dbstore')1283def test_assume_role_with_web_identity_with_request_tag():1284 check_webidentity()1285 iam_client=get_iam_client()1286 sts_client=get_sts_client()1287 default_endpoint=get_config_endpoint()1288 role_session_name=get_parameter_name()1289 thumbprint=get_thumbprint()1290 user_token=get_user_token()1291 realm=get_realm_name()1292 oidc_response = iam_client.create_open_id_connect_provider(1293 Url='http://localhost:8080/auth/realms/{}'.format(realm),1294 ThumbprintList=[1295 thumbprint,1296 ],1297 )1298 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1299 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1300 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1301 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\"}}"1302 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1303 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1304 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1305 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1306 s3_client = boto3.client('s3',1307 aws_access_key_id = resp['Credentials']['AccessKeyId'],1308 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1309 aws_session_token = resp['Credentials']['SessionToken'],1310 endpoint_url=default_endpoint,1311 region_name='',1312 )1313 bucket_name = get_new_bucket_name()1314 s3bucket = s3_client.create_bucket(Bucket=bucket_name)1315 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1316 bkt = s3_client.delete_bucket(Bucket=bucket_name)1317 eq(bkt['ResponseMetadata']['HTTPStatusCode'],204)1318 oidc_remove=iam_client.delete_open_id_connect_provider(1319 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1320 )1321@attr(resource='assume role with web identity')1322@attr(method='get')1323@attr(operation='check')1324@attr(assertion='assuming role using web token with aws:PrincipalTag in role policy')1325@attr('webidentity_test')1326@attr('abac_test')1327@attr('token_principal_tag_role_policy_test')1328@attr('fails_on_dbstore')1329def test_assume_role_with_web_identity_with_principal_tag():1330 check_webidentity()1331 iam_client=get_iam_client()1332 sts_client=get_sts_client()1333 default_endpoint=get_config_endpoint()1334 role_session_name=get_parameter_name()1335 thumbprint=get_thumbprint()1336 user_token=get_user_token()1337 realm=get_realm_name()1338 oidc_response = iam_client.create_open_id_connect_provider(1339 Url='http://localhost:8080/auth/realms/{}'.format(realm),1340 ThumbprintList=[1341 thumbprint,1342 ],1343 )1344 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1345 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1346 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1347 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"StringEquals\":{\"aws:PrincipalTag/Department\":\"Engineering\"}}}}"1348 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1349 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1350 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1351 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1352 s3_client = boto3.client('s3',1353 aws_access_key_id = resp['Credentials']['AccessKeyId'],1354 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1355 aws_session_token = resp['Credentials']['SessionToken'],1356 endpoint_url=default_endpoint,1357 region_name='',1358 )1359 bucket_name = get_new_bucket_name()1360 s3bucket = s3_client.create_bucket(Bucket=bucket_name)1361 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1362 bkt = s3_client.delete_bucket(Bucket=bucket_name)1363 eq(bkt['ResponseMetadata']['HTTPStatusCode'],204)1364 oidc_remove=iam_client.delete_open_id_connect_provider(1365 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1366 )1367@attr(resource='assume role with web identity')1368@attr(method='get')1369@attr(operation='check')1370@attr(assertion='assuming role using web token with aws:PrincipalTag in role policy')1371@attr('webidentity_test')1372@attr('abac_test')1373@attr('token_principal_tag_role_policy_test')1374@attr('fails_on_dbstore')1375def test_assume_role_with_web_identity_for_all_values():1376 check_webidentity()1377 iam_client=get_iam_client()1378 sts_client=get_sts_client()1379 default_endpoint=get_config_endpoint()1380 role_session_name=get_parameter_name()1381 thumbprint=get_thumbprint()1382 user_token=get_user_token()1383 realm=get_realm_name()1384 oidc_response = iam_client.create_open_id_connect_provider(1385 Url='http://localhost:8080/auth/realms/{}'.format(realm),1386 ThumbprintList=[1387 thumbprint,1388 ],1389 )1390 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1391 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1392 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1393 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"ForAllValues:StringEquals\":{\"aws:PrincipalTag/Department\":[\"Engineering\",\"Marketing\"]}}}}"1394 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1395 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1396 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1397 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1398 s3_client = boto3.client('s3',1399 aws_access_key_id = resp['Credentials']['AccessKeyId'],1400 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1401 aws_session_token = resp['Credentials']['SessionToken'],1402 endpoint_url=default_endpoint,1403 region_name='',1404 )1405 bucket_name = get_new_bucket_name()1406 s3bucket = s3_client.create_bucket(Bucket=bucket_name)1407 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1408 bkt = s3_client.delete_bucket(Bucket=bucket_name)1409 eq(bkt['ResponseMetadata']['HTTPStatusCode'],204)1410 oidc_remove=iam_client.delete_open_id_connect_provider(1411 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1412 )1413@attr(resource='assume role with web identity')1414@attr(method='get')1415@attr(operation='check')1416@attr(assertion='assuming role using web token with aws:PrincipalTag in role policy')1417@attr('webidentity_test')1418@attr('abac_test')1419@attr('token_principal_tag_role_policy_test')1420@attr('fails_on_dbstore')1421def test_assume_role_with_web_identity_for_all_values_deny():1422 check_webidentity()1423 iam_client=get_iam_client()1424 sts_client=get_sts_client()1425 default_endpoint=get_config_endpoint()1426 role_session_name=get_parameter_name()1427 thumbprint=get_thumbprint()1428 user_token=get_user_token()1429 realm=get_realm_name()1430 oidc_response = iam_client.create_open_id_connect_provider(1431 Url='http://localhost:8080/auth/realms/{}'.format(realm),1432 ThumbprintList=[1433 thumbprint,1434 ],1435 )1436 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1437 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1438 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1439 #ForAllValues: The condition returns true if every key value in the request matches at least one value in the policy1440 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"ForAllValues:StringEquals\":{\"aws:PrincipalTag/Department\":[\"Engineering\"]}}}}"1441 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1442 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1443 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1444 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1445 s3_client = boto3.client('s3',1446 aws_access_key_id = resp['Credentials']['AccessKeyId'],1447 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1448 aws_session_token = resp['Credentials']['SessionToken'],1449 endpoint_url=default_endpoint,1450 region_name='',1451 )1452 bucket_name = get_new_bucket_name()1453 try:1454 s3bucket = s3_client.create_bucket(Bucket=bucket_name)1455 except ClientError as e:1456 s3bucket_error = e.response.get("Error", {}).get("Code")1457 eq(s3bucket_error,'AccessDenied')1458 oidc_remove=iam_client.delete_open_id_connect_provider(1459 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1460 )1461@attr(resource='assume role with web identity')1462@attr(method='get')1463@attr(operation='check')1464@attr(assertion='assuming role using web token with aws:TagKeys in trust policy')1465@attr('webidentity_test')1466@attr('abac_test')1467@attr('token_tag_keys_test')1468@attr('fails_on_dbstore')1469def test_assume_role_with_web_identity_tag_keys_trust_policy():1470 check_webidentity()1471 iam_client=get_iam_client()1472 sts_client=get_sts_client()1473 default_endpoint=get_config_endpoint()1474 role_session_name=get_parameter_name()1475 thumbprint=get_thumbprint()1476 user_token=get_user_token()1477 realm=get_realm_name()1478 oidc_response = iam_client.create_open_id_connect_provider(1479 Url='http://localhost:8080/auth/realms/{}'.format(realm),1480 ThumbprintList=[1481 thumbprint,1482 ],1483 )1484 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:TagKeys\":\"Department\"}}}]}"1485 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1486 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1487 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"ForAnyValue:StringEquals\":{\"aws:PrincipalTag/Department\":[\"Engineering\"]}}}}"1488 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1489 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1490 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1491 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1492 s3_client = boto3.client('s3',1493 aws_access_key_id = resp['Credentials']['AccessKeyId'],1494 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1495 aws_session_token = resp['Credentials']['SessionToken'],1496 endpoint_url=default_endpoint,1497 region_name='',1498 )1499 bucket_name = get_new_bucket_name()1500 s3bucket = s3_client.create_bucket(Bucket=bucket_name)1501 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1502 bkt = s3_client.delete_bucket(Bucket=bucket_name)1503 eq(bkt['ResponseMetadata']['HTTPStatusCode'],204)1504 oidc_remove=iam_client.delete_open_id_connect_provider(1505 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1506 )1507@attr(resource='assume role with web identity')1508@attr(method='get')1509@attr(operation='check')1510@attr(assertion='assuming role using web token with aws:TagKeys in role permission policy')1511@attr('webidentity_test')1512@attr('abac_test')1513@attr('token_tag_keys_test')1514@attr('fails_on_dbstore')1515def test_assume_role_with_web_identity_tag_keys_role_policy():1516 check_webidentity()1517 iam_client=get_iam_client()1518 sts_client=get_sts_client()1519 default_endpoint=get_config_endpoint()1520 role_session_name=get_parameter_name()1521 thumbprint=get_thumbprint()1522 user_token=get_user_token()1523 realm=get_realm_name()1524 oidc_response = iam_client.create_open_id_connect_provider(1525 Url='http://localhost:8080/auth/realms/{}'.format(realm),1526 ThumbprintList=[1527 thumbprint,1528 ],1529 )1530 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1531 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1532 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1533 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"StringEquals\":{\"aws:TagKeys\":[\"Department\"]}}}}"1534 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1535 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1536 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1537 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1538 s3_client = boto3.client('s3',1539 aws_access_key_id = resp['Credentials']['AccessKeyId'],1540 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1541 aws_session_token = resp['Credentials']['SessionToken'],1542 endpoint_url=default_endpoint,1543 region_name='',1544 )1545 bucket_name = get_new_bucket_name()1546 s3bucket = s3_client.create_bucket(Bucket=bucket_name)1547 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1548 bkt = s3_client.delete_bucket(Bucket=bucket_name)1549 eq(bkt['ResponseMetadata']['HTTPStatusCode'],204)1550 oidc_remove=iam_client.delete_open_id_connect_provider(1551 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1552 )1553@attr(resource='assume role with web identity')1554@attr(method='put')1555@attr(operation='check')1556@attr(assertion='assuming role using web token with s3:ResourceTag in role permission policy')1557@attr('webidentity_test')1558@attr('abac_test')1559@attr('token_resource_tags_test')1560@attr('fails_on_dbstore')1561def test_assume_role_with_web_identity_resource_tag():1562 check_webidentity()1563 iam_client=get_iam_client()1564 sts_client=get_sts_client()1565 default_endpoint=get_config_endpoint()1566 role_session_name=get_parameter_name()1567 thumbprint=get_thumbprint()1568 user_token=get_user_token()1569 realm=get_realm_name()1570 s3_res_iam_creds = get_s3_resource_using_iam_creds()1571 s3_client_iam_creds = s3_res_iam_creds.meta.client1572 bucket_name = get_new_bucket_name()1573 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name)1574 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1575 bucket_tagging = s3_res_iam_creds.BucketTagging(bucket_name)1576 Set_Tag = bucket_tagging.put(Tagging={'TagSet':[{'Key':'Department', 'Value': 'Engineering'},{'Key':'Department', 'Value': 'Marketing'}]})1577 oidc_response = iam_client.create_open_id_connect_provider(1578 Url='http://localhost:8080/auth/realms/{}'.format(realm),1579 ThumbprintList=[1580 thumbprint,1581 ],1582 )1583 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1584 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1585 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1586 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"StringEquals\":{\"s3:ResourceTag/Department\":[\"Engineering\"]}}}}"1587 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1588 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1589 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1590 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1591 s3_client = boto3.client('s3',1592 aws_access_key_id = resp['Credentials']['AccessKeyId'],1593 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1594 aws_session_token = resp['Credentials']['SessionToken'],1595 endpoint_url=default_endpoint,1596 region_name='',1597 )1598 bucket_body = 'this is a test file'1599 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name, Key="test-1.txt")1600 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)1601 oidc_remove=iam_client.delete_open_id_connect_provider(1602 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1603 )1604@attr(resource='assume role with web identity')1605@attr(method='put')1606@attr(operation='check')1607@attr(assertion='assuming role using web token with s3:ResourceTag with missing tags on bucket')1608@attr('webidentity_test')1609@attr('abac_test')1610@attr('token_resource_tags_test')1611@attr('fails_on_dbstore')1612def test_assume_role_with_web_identity_resource_tag_deny():1613 check_webidentity()1614 iam_client=get_iam_client()1615 sts_client=get_sts_client()1616 default_endpoint=get_config_endpoint()1617 role_session_name=get_parameter_name()1618 thumbprint=get_thumbprint()1619 user_token=get_user_token()1620 realm=get_realm_name()1621 s3_res_iam_creds = get_s3_resource_using_iam_creds()1622 s3_client_iam_creds = s3_res_iam_creds.meta.client1623 bucket_name = get_new_bucket_name()1624 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name)1625 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1626 oidc_response = iam_client.create_open_id_connect_provider(1627 Url='http://localhost:8080/auth/realms/{}'.format(realm),1628 ThumbprintList=[1629 thumbprint,1630 ],1631 )1632 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1633 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1634 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1635 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"StringEquals\":{\"s3:ResourceTag/Department\":[\"Engineering\"]}}}}"1636 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1637 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1638 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1639 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1640 s3_client = boto3.client('s3',1641 aws_access_key_id = resp['Credentials']['AccessKeyId'],1642 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1643 aws_session_token = resp['Credentials']['SessionToken'],1644 endpoint_url=default_endpoint,1645 region_name='',1646 )1647 bucket_body = 'this is a test file'1648 try:1649 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name, Key="test-1.txt")1650 except ClientError as e:1651 s3_put_obj_error = e.response.get("Error", {}).get("Code")1652 eq(s3_put_obj_error,'AccessDenied')1653 oidc_remove=iam_client.delete_open_id_connect_provider(1654 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1655 )1656@attr(resource='assume role with web identity')1657@attr(method='put')1658@attr(operation='check')1659@attr(assertion='assuming role using web token with s3:ResourceTag with wrong resource tag in policy')1660@attr('webidentity_test')1661@attr('abac_test')1662@attr('token_resource_tags_test')1663@attr('fails_on_dbstore')1664def test_assume_role_with_web_identity_wrong_resource_tag_deny():1665 check_webidentity()1666 iam_client=get_iam_client()1667 sts_client=get_sts_client()1668 default_endpoint=get_config_endpoint()1669 role_session_name=get_parameter_name()1670 thumbprint=get_thumbprint()1671 user_token=get_user_token()1672 realm=get_realm_name()1673 s3_res_iam_creds = get_s3_resource_using_iam_creds()1674 s3_client_iam_creds = s3_res_iam_creds.meta.client1675 bucket_name = get_new_bucket_name()1676 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name)1677 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1678 bucket_tagging = s3_res_iam_creds.BucketTagging(bucket_name)1679 Set_Tag = bucket_tagging.put(Tagging={'TagSet':[{'Key':'Department', 'Value': 'WrongResourcetag'}]})1680 oidc_response = iam_client.create_open_id_connect_provider(1681 Url='http://localhost:8080/auth/realms/{}'.format(realm),1682 ThumbprintList=[1683 thumbprint,1684 ],1685 )1686 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1687 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1688 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1689 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"StringEquals\":{\"s3:ResourceTag/Department\":[\"Engineering\"]}}}}"1690 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1691 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1692 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1693 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1694 s3_client = boto3.client('s3',1695 aws_access_key_id = resp['Credentials']['AccessKeyId'],1696 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1697 aws_session_token = resp['Credentials']['SessionToken'],1698 endpoint_url=default_endpoint,1699 region_name='',1700 )1701 bucket_body = 'this is a test file'1702 try:1703 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name, Key="test-1.txt")1704 except ClientError as e:1705 s3_put_obj_error = e.response.get("Error", {}).get("Code")1706 eq(s3_put_obj_error,'AccessDenied')1707 oidc_remove=iam_client.delete_open_id_connect_provider(1708 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1709 )1710@attr(resource='assume role with web identity')1711@attr(method='put')1712@attr(operation='check')1713@attr(assertion='assuming role using web token with s3:ResourceTag matching aws:PrincipalTag in role permission policy')1714@attr('webidentity_test')1715@attr('abac_test')1716@attr('token_resource_tags_test')1717@attr('fails_on_dbstore')1718def test_assume_role_with_web_identity_resource_tag_princ_tag():1719 check_webidentity()1720 iam_client=get_iam_client()1721 sts_client=get_sts_client()1722 default_endpoint=get_config_endpoint()1723 role_session_name=get_parameter_name()1724 thumbprint=get_thumbprint()1725 user_token=get_user_token()1726 realm=get_realm_name()1727 s3_res_iam_creds = get_s3_resource_using_iam_creds()1728 s3_client_iam_creds = s3_res_iam_creds.meta.client1729 bucket_name = get_new_bucket_name()1730 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name)1731 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1732 bucket_tagging = s3_res_iam_creds.BucketTagging(bucket_name)1733 Set_Tag = bucket_tagging.put(Tagging={'TagSet':[{'Key':'Department', 'Value': 'Engineering'}]})1734 oidc_response = iam_client.create_open_id_connect_provider(1735 Url='http://localhost:8080/auth/realms/{}'.format(realm),1736 ThumbprintList=[1737 thumbprint,1738 ],1739 )1740 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1741 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1742 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1743 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"StringEquals\":{\"s3:ResourceTag/Department\":[\"${aws:PrincipalTag/Department}\"]}}}}"1744 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1745 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1746 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1747 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1748 s3_client = boto3.client('s3',1749 aws_access_key_id = resp['Credentials']['AccessKeyId'],1750 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1751 aws_session_token = resp['Credentials']['SessionToken'],1752 endpoint_url=default_endpoint,1753 region_name='',1754 )1755 bucket_body = 'this is a test file'1756 tags = 'Department=Engineering&Department=Marketing'1757 key = "test-1.txt"1758 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name, Key=key, Tagging=tags)1759 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)1760 s3_get_obj = s3_client.get_object(Bucket=bucket_name, Key=key)1761 eq(s3_get_obj['ResponseMetadata']['HTTPStatusCode'],200)1762 oidc_remove=iam_client.delete_open_id_connect_provider(1763 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1764 )1765@attr(resource='assume role with web identity')1766@attr(method='put')1767@attr(operation='check')1768@attr(assertion='assuming role using web token with s3:ResourceTag used to test copy object')1769@attr('webidentity_test')1770@attr('abac_test')1771@attr('token_resource_tags_test')1772@attr('fails_on_dbstore')1773def test_assume_role_with_web_identity_resource_tag_copy_obj():1774 check_webidentity()1775 iam_client=get_iam_client()1776 sts_client=get_sts_client()1777 default_endpoint=get_config_endpoint()1778 role_session_name=get_parameter_name()1779 thumbprint=get_thumbprint()1780 user_token=get_user_token()1781 realm=get_realm_name()1782 s3_res_iam_creds = get_s3_resource_using_iam_creds()1783 s3_client_iam_creds = s3_res_iam_creds.meta.client1784 #create two buckets and add same tags to both1785 bucket_name = get_new_bucket_name()1786 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name)1787 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1788 bucket_tagging = s3_res_iam_creds.BucketTagging(bucket_name)1789 Set_Tag = bucket_tagging.put(Tagging={'TagSet':[{'Key':'Department', 'Value': 'Engineering'}]})1790 copy_bucket_name = get_new_bucket_name()1791 s3bucket = s3_client_iam_creds.create_bucket(Bucket=copy_bucket_name)1792 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1793 bucket_tagging = s3_res_iam_creds.BucketTagging(copy_bucket_name)1794 Set_Tag = bucket_tagging.put(Tagging={'TagSet':[{'Key':'Department', 'Value': 'Engineering'}]})1795 oidc_response = iam_client.create_open_id_connect_provider(1796 Url='http://localhost:8080/auth/realms/{}'.format(realm),1797 ThumbprintList=[1798 thumbprint,1799 ],1800 )1801 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"aws:RequestTag/Department\":\"Engineering\"}}}]}"1802 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None)1803 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1804 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"StringEquals\":{\"s3:ResourceTag/Department\":[\"${aws:PrincipalTag/Department}\"]}}}}"1805 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1806 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1807 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1808 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1809 s3_client = boto3.client('s3',1810 aws_access_key_id = resp['Credentials']['AccessKeyId'],1811 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1812 aws_session_token = resp['Credentials']['SessionToken'],1813 endpoint_url=default_endpoint,1814 region_name='',1815 )1816 bucket_body = 'this is a test file'1817 tags = 'Department=Engineering'1818 key = "test-1.txt"1819 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name, Key=key, Tagging=tags)1820 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)1821 #copy to same bucket1822 copy_source = {1823 'Bucket': bucket_name,1824 'Key': 'test-1.txt'1825 }1826 s3_client.copy(copy_source, bucket_name, "test-2.txt")1827 s3_get_obj = s3_client.get_object(Bucket=bucket_name, Key="test-2.txt")1828 eq(s3_get_obj['ResponseMetadata']['HTTPStatusCode'],200)1829 #copy to another bucket1830 copy_source = {1831 'Bucket': bucket_name,1832 'Key': 'test-1.txt'1833 }1834 s3_client.copy(copy_source, copy_bucket_name, "test-1.txt")1835 s3_get_obj = s3_client.get_object(Bucket=copy_bucket_name, Key="test-1.txt")1836 eq(s3_get_obj['ResponseMetadata']['HTTPStatusCode'],200)1837 oidc_remove=iam_client.delete_open_id_connect_provider(1838 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1839 )1840@attr(resource='assume role with web identity')1841@attr(method='put')1842@attr(operation='check')1843@attr(assertion='assuming role using web token with iam:ResourceTag in role trust policy')1844@attr('webidentity_test')1845@attr('abac_test')1846@attr('token_role_tags_test')1847@attr('fails_on_dbstore')1848def test_assume_role_with_web_identity_role_resource_tag():1849 check_webidentity()1850 iam_client=get_iam_client()1851 sts_client=get_sts_client()1852 default_endpoint=get_config_endpoint()1853 role_session_name=get_parameter_name()1854 thumbprint=get_thumbprint()1855 user_token=get_user_token()1856 realm=get_realm_name()1857 s3_res_iam_creds = get_s3_resource_using_iam_creds()1858 s3_client_iam_creds = s3_res_iam_creds.meta.client1859 bucket_name = get_new_bucket_name()1860 s3bucket = s3_client_iam_creds.create_bucket(Bucket=bucket_name)1861 eq(s3bucket['ResponseMetadata']['HTTPStatusCode'],200)1862 bucket_tagging = s3_res_iam_creds.BucketTagging(bucket_name)1863 Set_Tag = bucket_tagging.put(Tagging={'TagSet':[{'Key':'Department', 'Value': 'Engineering'},{'Key':'Department', 'Value': 'Marketing'}]})1864 oidc_response = iam_client.create_open_id_connect_provider(1865 Url='http://localhost:8080/auth/realms/{}'.format(realm),1866 ThumbprintList=[1867 thumbprint,1868 ],1869 )1870 #iam:ResourceTag refers to the tag attached to role, hence the role is allowed to be assumed only when it has a tag matching the policy.1871 policy_document = "{\"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Principal\":{\"Federated\":[\""+oidc_response["OpenIDConnectProviderArn"]+"\"]},\"Action\":[\"sts:AssumeRoleWithWebIdentity\",\"sts:TagSession\"],\"Condition\":{\"StringEquals\":{\"iam:ResourceTag/Department\":\"Engineering\"}}}]}"1872 tags_list = [1873 {'Key':'Department','Value':'Engineering'},1874 {'Key':'Department','Value':'Marketing'}1875 ]1876 (role_error,role_response,general_role_name)=create_role(iam_client,'/',None,policy_document,None,None,None,tags_list)1877 eq(role_response['Role']['Arn'],'arn:aws:iam:::role/'+general_role_name+'')1878 role_policy = "{\"Version\":\"2012-10-17\",\"Statement\":{\"Effect\":\"Allow\",\"Action\":\"s3:*\",\"Resource\":\"arn:aws:s3:::*\",\"Condition\":{\"StringEquals\":{\"s3:ResourceTag/Department\":[\"Engineering\"]}}}}"1879 (role_err,response)=put_role_policy(iam_client,general_role_name,None,role_policy)1880 eq(response['ResponseMetadata']['HTTPStatusCode'],200)1881 resp=sts_client.assume_role_with_web_identity(RoleArn=role_response['Role']['Arn'],RoleSessionName=role_session_name,WebIdentityToken=user_token)1882 eq(resp['ResponseMetadata']['HTTPStatusCode'],200)1883 s3_client = boto3.client('s3',1884 aws_access_key_id = resp['Credentials']['AccessKeyId'],1885 aws_secret_access_key = resp['Credentials']['SecretAccessKey'],1886 aws_session_token = resp['Credentials']['SessionToken'],1887 endpoint_url=default_endpoint,1888 region_name='',1889 )1890 bucket_body = 'this is a test file'1891 s3_put_obj = s3_client.put_object(Body=bucket_body, Bucket=bucket_name, Key="test-1.txt")1892 eq(s3_put_obj['ResponseMetadata']['HTTPStatusCode'],200)1893 oidc_remove=iam_client.delete_open_id_connect_provider(1894 OpenIDConnectProviderArn=oidc_response["OpenIDConnectProviderArn"]1895 )

Full Screen

Full Screen

test_provider_aws.py

Source:test_provider_aws.py Github

copy

Full Screen

1import os2import sys3import unittest4from unittest import mock5from unittest.mock import MagicMock, patch6sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))7from identityexchange import provider8class TestMethods(unittest.TestCase):9 @patch("identityexchange.provider.open")10 def test_login_aws_good(self, mock_open):11 mock_open.close.return_value = None12 expected_access_key = "12345"13 expected_secret_key = "54321"14 expected_session_token = "asdfzxcv"15 expected_username = "nobody"16 expected_token = "123token456"17 expected_duration = 360018 mock_client = MagicMock()19 mock_client.assume_role_with_web_identity.return_value = {20 "Credentials": {21 "AccessKeyId": expected_access_key,22 "SecretAccessKey": expected_secret_key,23 "SessionToken": expected_session_token24 }25 }26 aws = provider.AmazonWebServices(27 config={28 "aws": {29 "duration": expected_duration,30 "profiles": [31 {32 "name": "test-profile",33 "role": "arn:aws:iam::123456789101:role/test-role"34 }35 ]36 }37 },38 client=mock_client,39 credentials={40 "username": expected_username,41 "token": expected_token42 }43 )44 with mock.patch.object(aws, "_AmazonWebServices__write_aws_credentials", return_value=None) as mock_write:45 aws.login_aws()46 mock_write.assert_called_once_with(47 profile="test-profile",48 credentials={49 "access_key": expected_access_key,50 "secret_key": expected_secret_key,51 "session_token": expected_session_token52 }53 )54 mock_client.assume_role_with_web_identity.assert_called_once_with(55 RoleArn="arn:aws:iam::123456789101:role/test-role",56 RoleSessionName=expected_username,57 WebIdentityToken=expected_token,58 DurationSeconds=expected_duration...

Full Screen

Full Screen

minio_s3.py

Source:minio_s3.py Github

copy

Full Screen

...7 aws_access_key_id=config['aws_access_key_id'],8 aws_secret_access_key=config['aws_secret_access_key'], 9 endpoint_url=config['endpoint_url']10 )11 response = client.assume_role_with_web_identity(12 RoleArn='00000-00000-00000-00000', # Not applicable for Minio13 RoleSessionName='0000', # Not applicable for Minio14 WebIdentityToken=jwt_token15 )...

Full Screen

Full Screen

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run localstack automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

... View sample repo... View DocumentationSign up Free
›
_

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

...
Sign up with GoogleSign up with Email
Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful