from __future__ import unicode_literals import base64 import json import boto import boto3 import os import sure # noqa import sys from boto.exception import BotoServerError from botocore.exceptions import ClientError from dateutil.tz import tzutc from moto import mock_iam, mock_iam_deprecated from moto.iam.models import aws_managed_policies from nose.tools import assert_raises, assert_equals from nose.tools import raises from datetime import datetime from tests.helpers import requires_boto_gte MOCK_CERT = """-----BEGIN CERTIFICATE----- MIIBpzCCARACCQCY5yOdxCTrGjANBgkqhkiG9w0BAQsFADAXMRUwEwYDVQQKDAxt b3RvIHRlc3RpbmcwIBcNMTgxMTA1MTkwNTIwWhgPMjI5MjA4MTkxOTA1MjBaMBcx FTATBgNVBAoMDG1vdG8gdGVzdGluZzCBnzANBgkqhkiG9w0BAQEFAAOBjQAwgYkC gYEA1Jn3g2h7LD3FLqdpcYNbFXCS4V4eDpuTCje9vKFcC3pi/01147X3zdfPy8Mt ZhKxcREOwm4NXykh23P9KW7fBovpNwnbYsbPqj8Hf1ZaClrgku1arTVhEnKjx8zO vaR/bVLCss4uE0E0VM1tJn/QGQsfthFsjuHtwx8uIWz35tUCAwEAATANBgkqhkiG 9w0BAQsFAAOBgQBWdOQ7bDc2nWkUhFjZoNIZrqjyNdjlMUndpwREVD7FQ/DuxJMj FyDHrtlrS80dPUQWNYHw++oACDpWO01LGLPPrGmuO/7cOdojPEd852q5gd+7W9xt 8vUH+pBa6IBLbvBp+szli51V3TLSWcoyy4ceJNQU2vCkTLoFdS0RLd/7tQ== -----END CERTIFICATE-----""" MOCK_POLICY = """ { "Version": "2012-10-17", "Statement": { "Effect": "Allow", "Action": "s3:ListBucket", "Resource": "arn:aws:s3:::example_bucket" } } """ MOCK_POLICY_2 = """ { "Version": "2012-10-17", "Id": "2", "Statement": { "Effect": "Allow", "Action": "s3:ListBucket", "Resource": "arn:aws:s3:::example_bucket" } } """ MOCK_POLICY_3 = """ { "Version": "2012-10-17", "Id": "3", "Statement": { "Effect": "Allow", "Action": "s3:ListBucket", "Resource": "arn:aws:s3:::example_bucket" } } """ @mock_iam_deprecated() def test_get_all_server_certs(): conn = boto.connect_iam() conn.upload_server_cert("certname", "certbody", "privatekey") certs = conn.get_all_server_certs()['list_server_certificates_response'][ 'list_server_certificates_result']['server_certificate_metadata_list'] certs.should.have.length_of(1) cert1 = certs[0] cert1.server_certificate_name.should.equal("certname") cert1.arn.should.equal( "arn:aws:iam::123456789012:server-certificate/certname") @mock_iam_deprecated() def test_get_server_cert_doesnt_exist(): conn = boto.connect_iam() with assert_raises(BotoServerError): conn.get_server_certificate("NonExistant") @mock_iam_deprecated() def test_get_server_cert(): conn = boto.connect_iam() conn.upload_server_cert("certname", "certbody", "privatekey") cert = conn.get_server_certificate("certname") cert.server_certificate_name.should.equal("certname") cert.arn.should.equal( "arn:aws:iam::123456789012:server-certificate/certname") @mock_iam_deprecated() def test_upload_server_cert(): conn = boto.connect_iam() conn.upload_server_cert("certname", "certbody", "privatekey") cert = conn.get_server_certificate("certname") cert.server_certificate_name.should.equal("certname") cert.arn.should.equal( "arn:aws:iam::123456789012:server-certificate/certname") @mock_iam_deprecated() def test_delete_server_cert(): conn = boto.connect_iam() conn.upload_server_cert("certname", "certbody", "privatekey") conn.get_server_certificate("certname") conn.delete_server_cert("certname") with assert_raises(BotoServerError): conn.get_server_certificate("certname") with assert_raises(BotoServerError): conn.delete_server_cert("certname") @mock_iam_deprecated() @raises(BotoServerError) def test_get_role__should_throw__when_role_does_not_exist(): conn = boto.connect_iam() conn.get_role('unexisting_role') @mock_iam_deprecated() @raises(BotoServerError) def test_get_instance_profile__should_throw__when_instance_profile_does_not_exist(): conn = boto.connect_iam() conn.get_instance_profile('unexisting_instance_profile') @mock_iam_deprecated() def test_create_role_and_instance_profile(): conn = boto.connect_iam() conn.create_instance_profile("my-profile", path="my-path") conn.create_role( "my-role", assume_role_policy_document="some policy", path="my-path") conn.add_role_to_instance_profile("my-profile", "my-role") role = conn.get_role("my-role") role.path.should.equal("my-path") role.assume_role_policy_document.should.equal("some policy") profile = conn.get_instance_profile("my-profile") profile.path.should.equal("my-path") role_from_profile = list(profile.roles.values())[0] role_from_profile['role_id'].should.equal(role.role_id) role_from_profile['role_name'].should.equal("my-role") conn.list_roles().roles[0].role_name.should.equal('my-role') # Test with an empty path: profile = conn.create_instance_profile('my-other-profile') profile.path.should.equal('/') @mock_iam_deprecated() def test_remove_role_from_instance_profile(): conn = boto.connect_iam() conn.create_instance_profile("my-profile", path="my-path") conn.create_role( "my-role", assume_role_policy_document="some policy", path="my-path") conn.add_role_to_instance_profile("my-profile", "my-role") profile = conn.get_instance_profile("my-profile") role_from_profile = list(profile.roles.values())[0] role_from_profile['role_name'].should.equal("my-role") conn.remove_role_from_instance_profile("my-profile", "my-role") profile = conn.get_instance_profile("my-profile") dict(profile.roles).should.be.empty @mock_iam() def test_get_login_profile(): conn = boto3.client('iam', region_name='us-east-1') conn.create_user(UserName='my-user') conn.create_login_profile(UserName='my-user', Password='my-pass') response = conn.get_login_profile(UserName='my-user') response['LoginProfile']['UserName'].should.equal('my-user') @mock_iam() def test_update_login_profile(): conn = boto3.client('iam', region_name='us-east-1') conn.create_user(UserName='my-user') conn.create_login_profile(UserName='my-user', Password='my-pass') response = conn.get_login_profile(UserName='my-user') response['LoginProfile'].get('PasswordResetRequired').should.equal(None) conn.update_login_profile(UserName='my-user', Password='new-pass', PasswordResetRequired=True) response = conn.get_login_profile(UserName='my-user') response['LoginProfile'].get('PasswordResetRequired').should.equal(True) @mock_iam() def test_delete_role(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(conn.exceptions.NoSuchEntityException): conn.delete_role(RoleName="my-role") # Test deletion failure with a managed policy conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/") response = conn.create_policy(PolicyName="my-managed-policy", PolicyDocument=MOCK_POLICY) conn.attach_role_policy(PolicyArn=response['Policy']['Arn'], RoleName="my-role") with assert_raises(conn.exceptions.DeleteConflictException): conn.delete_role(RoleName="my-role") conn.detach_role_policy(PolicyArn=response['Policy']['Arn'], RoleName="my-role") conn.delete_policy(PolicyArn=response['Policy']['Arn']) conn.delete_role(RoleName="my-role") with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_role(RoleName="my-role") # Test deletion failure with an inline policy conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/") conn.put_role_policy(RoleName="my-role", PolicyName="my-role-policy", PolicyDocument=MOCK_POLICY) with assert_raises(conn.exceptions.DeleteConflictException): conn.delete_role(RoleName="my-role") conn.delete_role_policy(RoleName="my-role", PolicyName="my-role-policy") conn.delete_role(RoleName="my-role") with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_role(RoleName="my-role") # Test deletion failure with attachment to an instance profile conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/") conn.create_instance_profile(InstanceProfileName="my-profile") conn.add_role_to_instance_profile(InstanceProfileName="my-profile", RoleName="my-role") with assert_raises(conn.exceptions.DeleteConflictException): conn.delete_role(RoleName="my-role") conn.remove_role_from_instance_profile(InstanceProfileName="my-profile", RoleName="my-role") conn.delete_role(RoleName="my-role") with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_role(RoleName="my-role") # Test deletion with no conflicts conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/") conn.delete_role(RoleName="my-role") with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_role(RoleName="my-role") @mock_iam_deprecated() def test_list_instance_profiles(): conn = boto.connect_iam() conn.create_instance_profile("my-profile", path="my-path") conn.create_role("my-role", path="my-path") conn.add_role_to_instance_profile("my-profile", "my-role") profiles = conn.list_instance_profiles().instance_profiles len(profiles).should.equal(1) profiles[0].instance_profile_name.should.equal("my-profile") profiles[0].roles.role_name.should.equal("my-role") @mock_iam_deprecated() def test_list_instance_profiles_for_role(): conn = boto.connect_iam() conn.create_role(role_name="my-role", assume_role_policy_document="some policy", path="my-path") conn.create_role(role_name="my-role2", assume_role_policy_document="some policy2", path="my-path2") profile_name_list = ['my-profile', 'my-profile2'] profile_path_list = ['my-path', 'my-path2'] for profile_count in range(0, 2): conn.create_instance_profile( profile_name_list[profile_count], path=profile_path_list[profile_count]) for profile_count in range(0, 2): conn.add_role_to_instance_profile( profile_name_list[profile_count], "my-role") profile_dump = conn.list_instance_profiles_for_role(role_name="my-role") profile_list = profile_dump['list_instance_profiles_for_role_response'][ 'list_instance_profiles_for_role_result']['instance_profiles'] for profile_count in range(0, len(profile_list)): profile_name_list.remove(profile_list[profile_count][ "instance_profile_name"]) profile_path_list.remove(profile_list[profile_count]["path"]) profile_list[profile_count]["roles"]["member"][ "role_name"].should.equal("my-role") len(profile_name_list).should.equal(0) len(profile_path_list).should.equal(0) profile_dump2 = conn.list_instance_profiles_for_role(role_name="my-role2") profile_list = profile_dump2['list_instance_profiles_for_role_response'][ 'list_instance_profiles_for_role_result']['instance_profiles'] len(profile_list).should.equal(0) @mock_iam_deprecated() def test_list_role_policies(): conn = boto.connect_iam() conn.create_role("my-role") conn.put_role_policy("my-role", "test policy", MOCK_POLICY) role = conn.list_role_policies("my-role") role.policy_names.should.have.length_of(1) role.policy_names[0].should.equal("test policy") conn.put_role_policy("my-role", "test policy 2", MOCK_POLICY) role = conn.list_role_policies("my-role") role.policy_names.should.have.length_of(2) conn.delete_role_policy("my-role", "test policy") role = conn.list_role_policies("my-role") role.policy_names.should.have.length_of(1) role.policy_names[0].should.equal("test policy 2") with assert_raises(BotoServerError): conn.delete_role_policy("my-role", "test policy") @mock_iam_deprecated() def test_put_role_policy(): conn = boto.connect_iam() conn.create_role( "my-role", assume_role_policy_document="some policy", path="my-path") conn.put_role_policy("my-role", "test policy", MOCK_POLICY) policy = conn.get_role_policy( "my-role", "test policy")['get_role_policy_response']['get_role_policy_result']['policy_name'] policy.should.equal("test policy") @mock_iam def test_get_role_policy(): conn = boto3.client('iam', region_name='us-east-1') conn.create_role( RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="my-path") with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_role_policy(RoleName="my-role", PolicyName="does-not-exist") @mock_iam_deprecated() def test_update_assume_role_policy(): conn = boto.connect_iam() role = conn.create_role("my-role") conn.update_assume_role_policy(role.role_name, "my-policy") role = conn.get_role("my-role") role.assume_role_policy_document.should.equal("my-policy") @mock_iam def test_create_policy(): conn = boto3.client('iam', region_name='us-east-1') response = conn.create_policy( PolicyName="TestCreatePolicy", PolicyDocument=MOCK_POLICY) response['Policy']['Arn'].should.equal("arn:aws:iam::123456789012:policy/TestCreatePolicy") @mock_iam def test_delete_policy(): conn = boto3.client('iam', region_name='us-east-1') response = conn.create_policy(PolicyName="TestCreatePolicy", PolicyDocument=MOCK_POLICY) [pol['PolicyName'] for pol in conn.list_policies(Scope='Local')['Policies']].should.equal(['TestCreatePolicy']) conn.delete_policy(PolicyArn=response['Policy']['Arn']) assert conn.list_policies(Scope='Local')['Policies'].should.be.empty @mock_iam def test_create_policy_versions(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(ClientError): conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestCreatePolicyVersion", PolicyDocument='{"some":"policy"}') conn.create_policy( PolicyName="TestCreatePolicyVersion", PolicyDocument=MOCK_POLICY) version = conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestCreatePolicyVersion", PolicyDocument=MOCK_POLICY, SetAsDefault=True) version.get('PolicyVersion').get('Document').should.equal(json.loads(MOCK_POLICY)) version.get('PolicyVersion').get('VersionId').should.equal("v2") version.get('PolicyVersion').get('IsDefaultVersion').should.be.ok conn.delete_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestCreatePolicyVersion", VersionId="v1") version = conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestCreatePolicyVersion", PolicyDocument=MOCK_POLICY) version.get('PolicyVersion').get('VersionId').should.equal("v3") version.get('PolicyVersion').get('IsDefaultVersion').shouldnt.be.ok @mock_iam def test_create_many_policy_versions(): conn = boto3.client('iam', region_name='us-east-1') conn.create_policy( PolicyName="TestCreateManyPolicyVersions", PolicyDocument=MOCK_POLICY) for _ in range(0, 4): conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestCreateManyPolicyVersions", PolicyDocument=MOCK_POLICY) with assert_raises(ClientError): conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestCreateManyPolicyVersions", PolicyDocument=MOCK_POLICY) @mock_iam def test_set_default_policy_version(): conn = boto3.client('iam', region_name='us-east-1') conn.create_policy( PolicyName="TestSetDefaultPolicyVersion", PolicyDocument=MOCK_POLICY) conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestSetDefaultPolicyVersion", PolicyDocument=MOCK_POLICY_2, SetAsDefault=True) conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestSetDefaultPolicyVersion", PolicyDocument=MOCK_POLICY_3, SetAsDefault=True) versions = conn.list_policy_versions( PolicyArn="arn:aws:iam::123456789012:policy/TestSetDefaultPolicyVersion") versions.get('Versions')[0].get('Document').should.equal(json.loads(MOCK_POLICY)) versions.get('Versions')[0].get('IsDefaultVersion').shouldnt.be.ok versions.get('Versions')[1].get('Document').should.equal(json.loads(MOCK_POLICY_2)) versions.get('Versions')[1].get('IsDefaultVersion').shouldnt.be.ok versions.get('Versions')[2].get('Document').should.equal(json.loads(MOCK_POLICY_3)) versions.get('Versions')[2].get('IsDefaultVersion').should.be.ok @mock_iam def test_get_policy(): conn = boto3.client('iam', region_name='us-east-1') response = conn.create_policy( PolicyName="TestGetPolicy", PolicyDocument=MOCK_POLICY) policy = conn.get_policy( PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicy") policy['Policy']['Arn'].should.equal("arn:aws:iam::123456789012:policy/TestGetPolicy") @mock_iam def test_get_aws_managed_policy(): conn = boto3.client('iam', region_name='us-east-1') managed_policy_arn = 'arn:aws:iam::aws:policy/IAMUserChangePassword' managed_policy_create_date = datetime.strptime("2016-11-15T00:25:16+00:00", "%Y-%m-%dT%H:%M:%S+00:00") policy = conn.get_policy( PolicyArn=managed_policy_arn) policy['Policy']['Arn'].should.equal(managed_policy_arn) policy['Policy']['CreateDate'].replace(tzinfo=None).should.equal(managed_policy_create_date) @mock_iam def test_get_policy_version(): conn = boto3.client('iam', region_name='us-east-1') conn.create_policy( PolicyName="TestGetPolicyVersion", PolicyDocument=MOCK_POLICY) version = conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicyVersion", PolicyDocument=MOCK_POLICY) with assert_raises(ClientError): conn.get_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicyVersion", VersionId='v2-does-not-exist') retrieved = conn.get_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestGetPolicyVersion", VersionId=version.get('PolicyVersion').get('VersionId')) retrieved.get('PolicyVersion').get('Document').should.equal(json.loads(MOCK_POLICY)) retrieved.get('PolicyVersion').get('IsDefaultVersion').shouldnt.be.ok @mock_iam def test_get_aws_managed_policy_version(): conn = boto3.client('iam', region_name='us-east-1') managed_policy_arn = 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole' managed_policy_version_create_date = datetime.strptime("2015-04-09T15:03:43+00:00", "%Y-%m-%dT%H:%M:%S+00:00") with assert_raises(ClientError): conn.get_policy_version( PolicyArn=managed_policy_arn, VersionId='v2-does-not-exist') retrieved = conn.get_policy_version( PolicyArn=managed_policy_arn, VersionId="v1") retrieved['PolicyVersion']['CreateDate'].replace(tzinfo=None).should.equal(managed_policy_version_create_date) retrieved['PolicyVersion']['Document'].should.be.an(dict) @mock_iam def test_get_aws_managed_policy_v4_version(): conn = boto3.client('iam', region_name='us-east-1') managed_policy_arn = 'arn:aws:iam::aws:policy/job-function/SystemAdministrator' managed_policy_version_create_date = datetime.strptime("2018-10-08T21:33:45+00:00", "%Y-%m-%dT%H:%M:%S+00:00") with assert_raises(ClientError): conn.get_policy_version( PolicyArn=managed_policy_arn, VersionId='v2-does-not-exist') retrieved = conn.get_policy_version( PolicyArn=managed_policy_arn, VersionId="v4") retrieved['PolicyVersion']['CreateDate'].replace(tzinfo=None).should.equal(managed_policy_version_create_date) retrieved['PolicyVersion']['Document'].should.be.an(dict) @mock_iam def test_list_policy_versions(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(ClientError): versions = conn.list_policy_versions( PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions") conn.create_policy( PolicyName="TestListPolicyVersions", PolicyDocument=MOCK_POLICY) versions = conn.list_policy_versions( PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions") versions.get('Versions')[0].get('VersionId').should.equal('v1') versions.get('Versions')[0].get('IsDefaultVersion').should.be.ok conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions", PolicyDocument=MOCK_POLICY_2) conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions", PolicyDocument=MOCK_POLICY_3) versions = conn.list_policy_versions( PolicyArn="arn:aws:iam::123456789012:policy/TestListPolicyVersions") versions.get('Versions')[1].get('Document').should.equal(json.loads(MOCK_POLICY_2)) versions.get('Versions')[1].get('IsDefaultVersion').shouldnt.be.ok versions.get('Versions')[2].get('Document').should.equal(json.loads(MOCK_POLICY_3)) versions.get('Versions')[2].get('IsDefaultVersion').shouldnt.be.ok @mock_iam def test_delete_policy_version(): conn = boto3.client('iam', region_name='us-east-1') conn.create_policy( PolicyName="TestDeletePolicyVersion", PolicyDocument=MOCK_POLICY) conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion", PolicyDocument=MOCK_POLICY) with assert_raises(ClientError): conn.delete_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion", VersionId='v2-nope-this-does-not-exist') conn.delete_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion", VersionId='v2') versions = conn.list_policy_versions( PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion") len(versions.get('Versions')).should.equal(1) @mock_iam def test_delete_default_policy_version(): conn = boto3.client('iam', region_name='us-east-1') conn.create_policy( PolicyName="TestDeletePolicyVersion", PolicyDocument=MOCK_POLICY) conn.create_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion", PolicyDocument=MOCK_POLICY_2) with assert_raises(ClientError): conn.delete_policy_version( PolicyArn="arn:aws:iam::123456789012:policy/TestDeletePolicyVersion", VersionId='v1') @mock_iam_deprecated() def test_create_user(): conn = boto.connect_iam() conn.create_user('my-user') with assert_raises(BotoServerError): conn.create_user('my-user') @mock_iam_deprecated() def test_get_user(): conn = boto.connect_iam() with assert_raises(BotoServerError): conn.get_user('my-user') conn.create_user('my-user') conn.get_user('my-user') @mock_iam() def test_update_user(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(conn.exceptions.NoSuchEntityException): conn.update_user(UserName='my-user') conn.create_user(UserName='my-user') conn.update_user(UserName='my-user', NewPath='/new-path/', NewUserName='new-user') response = conn.get_user(UserName='new-user') response['User'].get('Path').should.equal('/new-path/') with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_user(UserName='my-user') @mock_iam_deprecated() def test_get_current_user(): """If no user is specific, IAM returns the current user""" conn = boto.connect_iam() user = conn.get_user()['get_user_response']['get_user_result']['user'] user['user_name'].should.equal('default_user') @mock_iam() def test_list_users(): path_prefix = '/' max_items = 10 conn = boto3.client('iam', region_name='us-east-1') conn.create_user(UserName='my-user') response = conn.list_users(PathPrefix=path_prefix, MaxItems=max_items) user = response['Users'][0] user['UserName'].should.equal('my-user') user['Path'].should.equal('/') user['Arn'].should.equal('arn:aws:iam::123456789012:user/my-user') @mock_iam() def test_user_policies(): policy_name = 'UserManagedPolicy' user_name = 'my-user' conn = boto3.client('iam', region_name='us-east-1') conn.create_user(UserName=user_name) conn.put_user_policy( UserName=user_name, PolicyName=policy_name, PolicyDocument=MOCK_POLICY ) policy_doc = conn.get_user_policy( UserName=user_name, PolicyName=policy_name ) policy_doc['PolicyDocument'].should.equal(json.loads(MOCK_POLICY)) policies = conn.list_user_policies(UserName=user_name) len(policies['PolicyNames']).should.equal(1) policies['PolicyNames'][0].should.equal(policy_name) conn.delete_user_policy( UserName=user_name, PolicyName=policy_name ) policies = conn.list_user_policies(UserName=user_name) len(policies['PolicyNames']).should.equal(0) @mock_iam_deprecated() def test_create_login_profile(): conn = boto.connect_iam() with assert_raises(BotoServerError): conn.create_login_profile('my-user', 'my-pass') conn.create_user('my-user') conn.create_login_profile('my-user', 'my-pass') with assert_raises(BotoServerError): conn.create_login_profile('my-user', 'my-pass') @mock_iam_deprecated() def test_delete_login_profile(): conn = boto.connect_iam() conn.create_user('my-user') with assert_raises(BotoServerError): conn.delete_login_profile('my-user') conn.create_login_profile('my-user', 'my-pass') conn.delete_login_profile('my-user') @mock_iam() def test_create_access_key(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(ClientError): conn.create_access_key(UserName='my-user') conn.create_user(UserName='my-user') access_key = conn.create_access_key(UserName='my-user')["AccessKey"] (datetime.utcnow() - access_key["CreateDate"].replace(tzinfo=None)).seconds.should.be.within(0, 10) access_key["AccessKeyId"].should.have.length_of(20) access_key["SecretAccessKey"].should.have.length_of(40) assert access_key["AccessKeyId"].startswith("AKIA") @mock_iam_deprecated() def test_get_all_access_keys(): """If no access keys exist there should be none in the response, if an access key is present it should have the correct fields present""" conn = boto.connect_iam() conn.create_user('my-user') response = conn.get_all_access_keys('my-user') assert_equals( response['list_access_keys_response'][ 'list_access_keys_result']['access_key_metadata'], [] ) conn.create_access_key('my-user') response = conn.get_all_access_keys('my-user') assert_equals( sorted(response['list_access_keys_response'][ 'list_access_keys_result']['access_key_metadata'][0].keys()), sorted(['status', 'create_date', 'user_name', 'access_key_id']) ) @mock_iam_deprecated() def test_delete_access_key(): conn = boto.connect_iam() conn.create_user('my-user') access_key_id = conn.create_access_key('my-user')['create_access_key_response'][ 'create_access_key_result']['access_key']['access_key_id'] conn.delete_access_key(access_key_id, 'my-user') @mock_iam() def test_mfa_devices(): # Test enable device conn = boto3.client('iam', region_name='us-east-1') conn.create_user(UserName='my-user') conn.enable_mfa_device( UserName='my-user', SerialNumber='123456789', AuthenticationCode1='234567', AuthenticationCode2='987654' ) # Test list mfa devices response = conn.list_mfa_devices(UserName='my-user') device = response['MFADevices'][0] device['SerialNumber'].should.equal('123456789') # Test deactivate mfa device conn.deactivate_mfa_device(UserName='my-user', SerialNumber='123456789') response = conn.list_mfa_devices(UserName='my-user') len(response['MFADevices']).should.equal(0) @mock_iam def test_create_virtual_mfa_device(): client = boto3.client('iam', region_name='us-east-1') response = client.create_virtual_mfa_device( VirtualMFADeviceName='test-device' ) device = response['VirtualMFADevice'] device['SerialNumber'].should.equal('arn:aws:iam::123456789012:mfa/test-device') device['Base32StringSeed'].decode('ascii').should.match('[A-Z234567]') device['QRCodePNG'].should_not.be.empty response = client.create_virtual_mfa_device( Path='/', VirtualMFADeviceName='test-device-2' ) device = response['VirtualMFADevice'] device['SerialNumber'].should.equal('arn:aws:iam::123456789012:mfa/test-device-2') device['Base32StringSeed'].decode('ascii').should.match('[A-Z234567]') device['QRCodePNG'].should_not.be.empty response = client.create_virtual_mfa_device( Path='/test/', VirtualMFADeviceName='test-device' ) device = response['VirtualMFADevice'] device['SerialNumber'].should.equal('arn:aws:iam::123456789012:mfa/test/test-device') device['Base32StringSeed'].decode('ascii').should.match('[A-Z234567]') device['QRCodePNG'].should_not.be.empty @mock_iam def test_create_virtual_mfa_device_errors(): client = boto3.client('iam', region_name='us-east-1') client.create_virtual_mfa_device( VirtualMFADeviceName='test-device' ) client.create_virtual_mfa_device.when.called_with( VirtualMFADeviceName='test-device' ).should.throw( ClientError, 'MFADevice entity at the same path and name already exists.' ) client.create_virtual_mfa_device.when.called_with( Path='test', VirtualMFADeviceName='test-device' ).should.throw( ClientError, 'The specified value for path is invalid. ' 'It must begin and end with / and contain only alphanumeric characters and/or / characters.' ) client.create_virtual_mfa_device.when.called_with( Path='/test//test/', VirtualMFADeviceName='test-device' ).should.throw( ClientError, 'The specified value for path is invalid. ' 'It must begin and end with / and contain only alphanumeric characters and/or / characters.' ) too_long_path = '/{}/'.format('b' * 511) client.create_virtual_mfa_device.when.called_with( Path=too_long_path, VirtualMFADeviceName='test-device' ).should.throw( ClientError, '1 validation error detected: ' 'Value "{}" at "path" failed to satisfy constraint: ' 'Member must have length less than or equal to 512' ) @mock_iam def test_delete_virtual_mfa_device(): client = boto3.client('iam', region_name='us-east-1') response = client.create_virtual_mfa_device( VirtualMFADeviceName='test-device' ) serial_number = response['VirtualMFADevice']['SerialNumber'] client.delete_virtual_mfa_device( SerialNumber=serial_number ) response = client.list_virtual_mfa_devices() response['VirtualMFADevices'].should.have.length_of(0) response['IsTruncated'].should_not.be.ok @mock_iam def test_delete_virtual_mfa_device_errors(): client = boto3.client('iam', region_name='us-east-1') serial_number = 'arn:aws:iam::123456789012:mfa/not-existing' client.delete_virtual_mfa_device.when.called_with( SerialNumber=serial_number ).should.throw( ClientError, 'VirtualMFADevice with serial number {0} doesn\'t exist.'.format(serial_number) ) @mock_iam def test_list_virtual_mfa_devices(): client = boto3.client('iam', region_name='us-east-1') response = client.create_virtual_mfa_device( VirtualMFADeviceName='test-device' ) serial_number_1 = response['VirtualMFADevice']['SerialNumber'] response = client.create_virtual_mfa_device( Path='/test/', VirtualMFADeviceName='test-device' ) serial_number_2 = response['VirtualMFADevice']['SerialNumber'] response = client.list_virtual_mfa_devices() response['VirtualMFADevices'].should.equal([ { 'SerialNumber': serial_number_1 }, { 'SerialNumber': serial_number_2 } ]) response['IsTruncated'].should_not.be.ok response = client.list_virtual_mfa_devices( AssignmentStatus='Assigned' ) response['VirtualMFADevices'].should.have.length_of(0) response['IsTruncated'].should_not.be.ok response = client.list_virtual_mfa_devices( AssignmentStatus='Unassigned' ) response['VirtualMFADevices'].should.equal([ { 'SerialNumber': serial_number_1 }, { 'SerialNumber': serial_number_2 } ]) response['IsTruncated'].should_not.be.ok response = client.list_virtual_mfa_devices( AssignmentStatus='Any', MaxItems=1 ) response['VirtualMFADevices'].should.equal([ { 'SerialNumber': serial_number_1 } ]) response['IsTruncated'].should.be.ok response['Marker'].should.equal('1') response = client.list_virtual_mfa_devices( AssignmentStatus='Any', Marker=response['Marker'] ) response['VirtualMFADevices'].should.equal([ { 'SerialNumber': serial_number_2 } ]) response['IsTruncated'].should_not.be.ok @mock_iam def test_list_virtual_mfa_devices_errors(): client = boto3.client('iam', region_name='us-east-1') client.create_virtual_mfa_device( VirtualMFADeviceName='test-device' ) client.list_virtual_mfa_devices.when.called_with( Marker='100' ).should.throw( ClientError, 'Invalid Marker.' ) @mock_iam def test_enable_virtual_mfa_device(): client = boto3.client('iam', region_name='us-east-1') response = client.create_virtual_mfa_device( VirtualMFADeviceName='test-device' ) serial_number = response['VirtualMFADevice']['SerialNumber'] client.create_user(UserName='test-user') client.enable_mfa_device( UserName='test-user', SerialNumber=serial_number, AuthenticationCode1='234567', AuthenticationCode2='987654' ) response = client.list_virtual_mfa_devices( AssignmentStatus='Unassigned' ) response['VirtualMFADevices'].should.have.length_of(0) response['IsTruncated'].should_not.be.ok response = client.list_virtual_mfa_devices( AssignmentStatus='Assigned' ) device = response['VirtualMFADevices'][0] device['SerialNumber'].should.equal(serial_number) device['User']['Path'].should.equal('/') device['User']['UserName'].should.equal('test-user') device['User']['UserId'].should_not.be.empty device['User']['Arn'].should.equal('arn:aws:iam::123456789012:user/test-user') device['User']['CreateDate'].should.be.a(datetime) device['EnableDate'].should.be.a(datetime) response['IsTruncated'].should_not.be.ok client.deactivate_mfa_device( UserName='test-user', SerialNumber=serial_number ) response = client.list_virtual_mfa_devices( AssignmentStatus='Assigned' ) response['VirtualMFADevices'].should.have.length_of(0) response['IsTruncated'].should_not.be.ok response = client.list_virtual_mfa_devices( AssignmentStatus = 'Unassigned' ) response['VirtualMFADevices'].should.equal([ { 'SerialNumber': serial_number } ]) response['IsTruncated'].should_not.be.ok @mock_iam_deprecated() def test_delete_user_deprecated(): conn = boto.connect_iam() with assert_raises(BotoServerError): conn.delete_user('my-user') conn.create_user('my-user') conn.delete_user('my-user') @mock_iam() def test_delete_user(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(conn.exceptions.NoSuchEntityException): conn.delete_user(UserName='my-user') # Test deletion failure with a managed policy conn.create_user(UserName='my-user') response = conn.create_policy(PolicyName="my-managed-policy", PolicyDocument=MOCK_POLICY) conn.attach_user_policy(PolicyArn=response['Policy']['Arn'], UserName="my-user") with assert_raises(conn.exceptions.DeleteConflictException): conn.delete_user(UserName='my-user') conn.detach_user_policy(PolicyArn=response['Policy']['Arn'], UserName="my-user") conn.delete_policy(PolicyArn=response['Policy']['Arn']) conn.delete_user(UserName='my-user') with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_user(UserName='my-user') # Test deletion failure with an inline policy conn.create_user(UserName='my-user') conn.put_user_policy( UserName='my-user', PolicyName='my-user-policy', PolicyDocument=MOCK_POLICY ) with assert_raises(conn.exceptions.DeleteConflictException): conn.delete_user(UserName='my-user') conn.delete_user_policy(UserName='my-user', PolicyName='my-user-policy') conn.delete_user(UserName='my-user') with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_user(UserName='my-user') # Test deletion with no conflicts conn.create_user(UserName='my-user') conn.delete_user(UserName='my-user') with assert_raises(conn.exceptions.NoSuchEntityException): conn.get_user(UserName='my-user') @mock_iam_deprecated() def test_generate_credential_report(): conn = boto.connect_iam() result = conn.generate_credential_report() result['generate_credential_report_response'][ 'generate_credential_report_result']['state'].should.equal('STARTED') result = conn.generate_credential_report() result['generate_credential_report_response'][ 'generate_credential_report_result']['state'].should.equal('COMPLETE') @mock_iam def test_boto3_generate_credential_report(): conn = boto3.client('iam', region_name='us-east-1') result = conn.generate_credential_report() result['State'].should.equal('STARTED') result = conn.generate_credential_report() result['State'].should.equal('COMPLETE') @mock_iam_deprecated() def test_get_credential_report(): conn = boto.connect_iam() conn.create_user('my-user') with assert_raises(BotoServerError): conn.get_credential_report() result = conn.generate_credential_report() while result['generate_credential_report_response']['generate_credential_report_result']['state'] != 'COMPLETE': result = conn.generate_credential_report() result = conn.get_credential_report() report = base64.b64decode(result['get_credential_report_response'][ 'get_credential_report_result']['content'].encode('ascii')).decode('ascii') report.should.match(r'.*my-user.*') @mock_iam def test_boto3_get_credential_report(): conn = boto3.client('iam', region_name='us-east-1') conn.create_user(UserName='my-user') with assert_raises(ClientError): conn.get_credential_report() result = conn.generate_credential_report() while result['State'] != 'COMPLETE': result = conn.generate_credential_report() result = conn.get_credential_report() report = result['Content'].decode('utf-8') report.should.match(r'.*my-user.*') @requires_boto_gte('2.39') @mock_iam_deprecated() def test_managed_policy(): conn = boto.connect_iam() conn.create_policy(policy_name='UserManagedPolicy', policy_document=MOCK_POLICY, path='/mypolicy/', description='my user managed policy') marker = 0 aws_policies = [] while marker is not None: response = conn.list_policies(scope='AWS', marker=marker)[ 'list_policies_response']['list_policies_result'] for policy in response['policies']: aws_policies.append(policy) marker = response.get('marker') set(p.name for p in aws_managed_policies).should.equal( set(p['policy_name'] for p in aws_policies)) user_policies = conn.list_policies(scope='Local')['list_policies_response'][ 'list_policies_result']['policies'] set(['UserManagedPolicy']).should.equal( set(p['policy_name'] for p in user_policies)) marker = 0 all_policies = [] while marker is not None: response = conn.list_policies(marker=marker)[ 'list_policies_response']['list_policies_result'] for policy in response['policies']: all_policies.append(policy) marker = response.get('marker') set(p['policy_name'] for p in aws_policies + user_policies).should.equal(set(p['policy_name'] for p in all_policies)) role_name = 'my-role' conn.create_role(role_name, assume_role_policy_document={ 'policy': 'test'}, path="my-path") for policy_name in ['AmazonElasticMapReduceRole', 'AmazonElasticMapReduceforEC2Role']: policy_arn = 'arn:aws:iam::aws:policy/service-role/' + policy_name conn.attach_role_policy(policy_arn, role_name) rows = conn.list_policies(only_attached=True)['list_policies_response'][ 'list_policies_result']['policies'] rows.should.have.length_of(2) for x in rows: int(x['attachment_count']).should.be.greater_than(0) # boto has not implemented this end point but accessible this way resp = conn.get_response('ListAttachedRolePolicies', {'RoleName': role_name}, list_marker='AttachedPolicies') resp['list_attached_role_policies_response']['list_attached_role_policies_result'][ 'attached_policies'].should.have.length_of(2) conn.detach_role_policy( "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole", role_name) rows = conn.list_policies(only_attached=True)['list_policies_response'][ 'list_policies_result']['policies'] rows.should.have.length_of(1) for x in rows: int(x['attachment_count']).should.be.greater_than(0) # boto has not implemented this end point but accessible this way resp = conn.get_response('ListAttachedRolePolicies', {'RoleName': role_name}, list_marker='AttachedPolicies') resp['list_attached_role_policies_response']['list_attached_role_policies_result'][ 'attached_policies'].should.have.length_of(1) with assert_raises(BotoServerError): conn.detach_role_policy( "arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceRole", role_name) with assert_raises(BotoServerError): conn.detach_role_policy( "arn:aws:iam::aws:policy/Nonexistent", role_name) @mock_iam def test_boto3_create_login_profile(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(ClientError): conn.create_login_profile(UserName='my-user', Password='Password') conn.create_user(UserName='my-user') conn.create_login_profile(UserName='my-user', Password='Password') with assert_raises(ClientError): conn.create_login_profile(UserName='my-user', Password='Password') @mock_iam() def test_attach_detach_user_policy(): iam = boto3.resource('iam', region_name='us-east-1') client = boto3.client('iam', region_name='us-east-1') user = iam.create_user(UserName='test-user') policy_name = 'UserAttachedPolicy' policy = iam.create_policy(PolicyName=policy_name, PolicyDocument=MOCK_POLICY, Path='/mypolicy/', Description='my user attached policy') client.attach_user_policy(UserName=user.name, PolicyArn=policy.arn) resp = client.list_attached_user_policies(UserName=user.name) resp['AttachedPolicies'].should.have.length_of(1) attached_policy = resp['AttachedPolicies'][0] attached_policy['PolicyArn'].should.equal(policy.arn) attached_policy['PolicyName'].should.equal(policy_name) client.detach_user_policy(UserName=user.name, PolicyArn=policy.arn) resp = client.list_attached_user_policies(UserName=user.name) resp['AttachedPolicies'].should.have.length_of(0) @mock_iam def test_update_access_key(): iam = boto3.resource('iam', region_name='us-east-1') client = iam.meta.client username = 'test-user' iam.create_user(UserName=username) with assert_raises(ClientError): client.update_access_key(UserName=username, AccessKeyId='non-existent-key', Status='Inactive') key = client.create_access_key(UserName=username)['AccessKey'] client.update_access_key(UserName=username, AccessKeyId=key['AccessKeyId'], Status='Inactive') resp = client.list_access_keys(UserName=username) resp['AccessKeyMetadata'][0]['Status'].should.equal('Inactive') @mock_iam def test_get_access_key_last_used(): iam = boto3.resource('iam', region_name='us-east-1') client = iam.meta.client username = 'test-user' iam.create_user(UserName=username) with assert_raises(ClientError): client.get_access_key_last_used(AccessKeyId='non-existent-key-id') create_key_response = client.create_access_key(UserName=username)['AccessKey'] resp = client.get_access_key_last_used(AccessKeyId=create_key_response['AccessKeyId']) datetime.strftime(resp["AccessKeyLastUsed"]["LastUsedDate"], "%Y-%m-%d").should.equal(datetime.strftime( datetime.utcnow(), "%Y-%m-%d" )) resp["UserName"].should.equal(create_key_response["UserName"]) @mock_iam def test_get_account_authorization_details(): test_policy = json.dumps({ "Version": "2012-10-17", "Statement": [ { "Action": "s3:ListBucket", "Resource": "*", "Effect": "Allow", } ] }) conn = boto3.client('iam', region_name='us-east-1') boundary = 'arn:aws:iam::123456789012:policy/boundary' conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/", Description='testing', PermissionsBoundary=boundary) conn.create_user(Path='/', UserName='testUser') conn.create_group(Path='/', GroupName='testGroup') conn.create_policy( PolicyName='testPolicy', Path='/', PolicyDocument=test_policy, Description='Test Policy' ) # Attach things to the user and group: conn.put_user_policy(UserName='testUser', PolicyName='testPolicy', PolicyDocument=test_policy) conn.put_group_policy(GroupName='testGroup', PolicyName='testPolicy', PolicyDocument=test_policy) conn.attach_user_policy(UserName='testUser', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy') conn.attach_group_policy(GroupName='testGroup', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy') conn.add_user_to_group(UserName='testUser', GroupName='testGroup') # Add things to the role: conn.create_instance_profile(InstanceProfileName='ipn') conn.add_role_to_instance_profile(InstanceProfileName='ipn', RoleName='my-role') conn.tag_role(RoleName='my-role', Tags=[ { 'Key': 'somekey', 'Value': 'somevalue' }, { 'Key': 'someotherkey', 'Value': 'someothervalue' } ]) conn.put_role_policy(RoleName='my-role', PolicyName='test-policy', PolicyDocument=test_policy) conn.attach_role_policy(RoleName='my-role', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy') result = conn.get_account_authorization_details(Filter=['Role']) assert len(result['RoleDetailList']) == 1 assert len(result['UserDetailList']) == 0 assert len(result['GroupDetailList']) == 0 assert len(result['Policies']) == 0 assert len(result['RoleDetailList'][0]['InstanceProfileList']) == 1 assert result['RoleDetailList'][0]['InstanceProfileList'][0]['Roles'][0]['Description'] == 'testing' assert result['RoleDetailList'][0]['InstanceProfileList'][0]['Roles'][0]['PermissionsBoundary'] == { 'PermissionsBoundaryType': 'PermissionsBoundaryPolicy', 'PermissionsBoundaryArn': 'arn:aws:iam::123456789012:policy/boundary' } assert len(result['RoleDetailList'][0]['Tags']) == 2 assert len(result['RoleDetailList'][0]['RolePolicyList']) == 1 assert len(result['RoleDetailList'][0]['AttachedManagedPolicies']) == 1 assert result['RoleDetailList'][0]['AttachedManagedPolicies'][0]['PolicyName'] == 'testPolicy' assert result['RoleDetailList'][0]['AttachedManagedPolicies'][0]['PolicyArn'] == \ 'arn:aws:iam::123456789012:policy/testPolicy' result = conn.get_account_authorization_details(Filter=['User']) assert len(result['RoleDetailList']) == 0 assert len(result['UserDetailList']) == 1 assert len(result['UserDetailList'][0]['GroupList']) == 1 assert len(result['UserDetailList'][0]['AttachedManagedPolicies']) == 1 assert len(result['GroupDetailList']) == 0 assert len(result['Policies']) == 0 assert result['UserDetailList'][0]['AttachedManagedPolicies'][0]['PolicyName'] == 'testPolicy' assert result['UserDetailList'][0]['AttachedManagedPolicies'][0]['PolicyArn'] == \ 'arn:aws:iam::123456789012:policy/testPolicy' result = conn.get_account_authorization_details(Filter=['Group']) assert len(result['RoleDetailList']) == 0 assert len(result['UserDetailList']) == 0 assert len(result['GroupDetailList']) == 1 assert len(result['GroupDetailList'][0]['GroupPolicyList']) == 1 assert len(result['GroupDetailList'][0]['AttachedManagedPolicies']) == 1 assert len(result['Policies']) == 0 assert result['GroupDetailList'][0]['AttachedManagedPolicies'][0]['PolicyName'] == 'testPolicy' assert result['GroupDetailList'][0]['AttachedManagedPolicies'][0]['PolicyArn'] == \ 'arn:aws:iam::123456789012:policy/testPolicy' result = conn.get_account_authorization_details(Filter=['LocalManagedPolicy']) assert len(result['RoleDetailList']) == 0 assert len(result['UserDetailList']) == 0 assert len(result['GroupDetailList']) == 0 assert len(result['Policies']) == 1 assert len(result['Policies'][0]['PolicyVersionList']) == 1 # Check for greater than 1 since this should always be greater than one but might change. # See iam/aws_managed_policies.py result = conn.get_account_authorization_details(Filter=['AWSManagedPolicy']) assert len(result['RoleDetailList']) == 0 assert len(result['UserDetailList']) == 0 assert len(result['GroupDetailList']) == 0 assert len(result['Policies']) > 1 result = conn.get_account_authorization_details() assert len(result['RoleDetailList']) == 1 assert len(result['UserDetailList']) == 1 assert len(result['GroupDetailList']) == 1 assert len(result['Policies']) > 1 @mock_iam def test_signing_certs(): client = boto3.client('iam', region_name='us-east-1') # Create the IAM user first: client.create_user(UserName='testing') # Upload the cert: resp = client.upload_signing_certificate(UserName='testing', CertificateBody=MOCK_CERT)['Certificate'] cert_id = resp['CertificateId'] assert resp['UserName'] == 'testing' assert resp['Status'] == 'Active' assert resp['CertificateBody'] == MOCK_CERT assert resp['CertificateId'] # Upload a the cert with an invalid body: with assert_raises(ClientError) as ce: client.upload_signing_certificate(UserName='testing', CertificateBody='notacert') assert ce.exception.response['Error']['Code'] == 'MalformedCertificate' # Upload with an invalid user: with assert_raises(ClientError): client.upload_signing_certificate(UserName='notauser', CertificateBody=MOCK_CERT) # Update: client.update_signing_certificate(UserName='testing', CertificateId=cert_id, Status='Inactive') with assert_raises(ClientError): client.update_signing_certificate(UserName='notauser', CertificateId=cert_id, Status='Inactive') with assert_raises(ClientError) as ce: client.update_signing_certificate(UserName='testing', CertificateId='x' * 32, Status='Inactive') assert ce.exception.response['Error']['Message'] == 'The Certificate with id {id} cannot be found.'.format( id='x' * 32) # List the certs: resp = client.list_signing_certificates(UserName='testing')['Certificates'] assert len(resp) == 1 assert resp[0]['CertificateBody'] == MOCK_CERT assert resp[0]['Status'] == 'Inactive' # Changed with the update call above. with assert_raises(ClientError): client.list_signing_certificates(UserName='notauser') # Delete: client.delete_signing_certificate(UserName='testing', CertificateId=cert_id) with assert_raises(ClientError): client.delete_signing_certificate(UserName='notauser', CertificateId=cert_id) @mock_iam() def test_create_saml_provider(): conn = boto3.client('iam', region_name='us-east-1') response = conn.create_saml_provider( Name="TestSAMLProvider", SAMLMetadataDocument='a' * 1024 ) response['SAMLProviderArn'].should.equal("arn:aws:iam::123456789012:saml-provider/TestSAMLProvider") @mock_iam() def test_get_saml_provider(): conn = boto3.client('iam', region_name='us-east-1') saml_provider_create = conn.create_saml_provider( Name="TestSAMLProvider", SAMLMetadataDocument='a' * 1024 ) response = conn.get_saml_provider( SAMLProviderArn=saml_provider_create['SAMLProviderArn'] ) response['SAMLMetadataDocument'].should.equal('a' * 1024) @mock_iam() def test_list_saml_providers(): conn = boto3.client('iam', region_name='us-east-1') conn.create_saml_provider( Name="TestSAMLProvider", SAMLMetadataDocument='a' * 1024 ) response = conn.list_saml_providers() response['SAMLProviderList'][0]['Arn'].should.equal("arn:aws:iam::123456789012:saml-provider/TestSAMLProvider") @mock_iam() def test_delete_saml_provider(): conn = boto3.client('iam', region_name='us-east-1') saml_provider_create = conn.create_saml_provider( Name="TestSAMLProvider", SAMLMetadataDocument='a' * 1024 ) response = conn.list_saml_providers() len(response['SAMLProviderList']).should.equal(1) conn.delete_saml_provider( SAMLProviderArn=saml_provider_create['SAMLProviderArn'] ) response = conn.list_saml_providers() len(response['SAMLProviderList']).should.equal(0) conn.create_user(UserName='testing') cert_id = '123456789012345678901234' with assert_raises(ClientError) as ce: conn.delete_signing_certificate(UserName='testing', CertificateId=cert_id) assert ce.exception.response['Error']['Message'] == 'The Certificate with id {id} cannot be found.'.format( id=cert_id) # Verify that it's not in the list: resp = conn.list_signing_certificates(UserName='testing') assert not resp['Certificates'] @mock_iam() def test_create_role_with_tags(): """Tests both the tag_role and get_role_tags capability""" conn = boto3.client('iam', region_name='us-east-1') conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="{}", Tags=[ { 'Key': 'somekey', 'Value': 'somevalue' }, { 'Key': 'someotherkey', 'Value': 'someothervalue' } ], Description='testing') # Get role: role = conn.get_role(RoleName='my-role')['Role'] assert len(role['Tags']) == 2 assert role['Tags'][0]['Key'] == 'somekey' assert role['Tags'][0]['Value'] == 'somevalue' assert role['Tags'][1]['Key'] == 'someotherkey' assert role['Tags'][1]['Value'] == 'someothervalue' assert role['Description'] == 'testing' # Empty is good: conn.create_role(RoleName="my-role2", AssumeRolePolicyDocument="{}", Tags=[ { 'Key': 'somekey', 'Value': '' } ]) tags = conn.list_role_tags(RoleName='my-role2') assert len(tags['Tags']) == 1 assert tags['Tags'][0]['Key'] == 'somekey' assert tags['Tags'][0]['Value'] == '' # Test creating tags with invalid values: # With more than 50 tags: with assert_raises(ClientError) as ce: too_many_tags = list(map(lambda x: {'Key': str(x), 'Value': str(x)}, range(0, 51))) conn.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", Tags=too_many_tags) assert 'failed to satisfy constraint: Member must have length less than or equal to 50.' \ in ce.exception.response['Error']['Message'] # With a duplicate tag: with assert_raises(ClientError) as ce: conn.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", Tags=[{'Key': '0', 'Value': ''}, {'Key': '0', 'Value': ''}]) assert 'Duplicate tag keys found. Please note that Tag keys are case insensitive.' \ in ce.exception.response['Error']['Message'] # Duplicate tag with different casing: with assert_raises(ClientError) as ce: conn.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", Tags=[{'Key': 'a', 'Value': ''}, {'Key': 'A', 'Value': ''}]) assert 'Duplicate tag keys found. Please note that Tag keys are case insensitive.' \ in ce.exception.response['Error']['Message'] # With a really big key: with assert_raises(ClientError) as ce: conn.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", Tags=[{'Key': '0' * 129, 'Value': ''}]) assert 'Member must have length less than or equal to 128.' in ce.exception.response['Error']['Message'] # With a really big value: with assert_raises(ClientError) as ce: conn.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", Tags=[{'Key': '0', 'Value': '0' * 257}]) assert 'Member must have length less than or equal to 256.' in ce.exception.response['Error']['Message'] # With an invalid character: with assert_raises(ClientError) as ce: conn.create_role(RoleName="my-role3", AssumeRolePolicyDocument="{}", Tags=[{'Key': 'NOWAY!', 'Value': ''}]) assert 'Member must satisfy regular expression pattern: [\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]+' \ in ce.exception.response['Error']['Message'] @mock_iam() def test_tag_role(): """Tests both the tag_role and get_role_tags capability""" conn = boto3.client('iam', region_name='us-east-1') conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="{}") # Get without tags: role = conn.get_role(RoleName='my-role')['Role'] assert not role.get('Tags') # With proper tag values: conn.tag_role(RoleName='my-role', Tags=[ { 'Key': 'somekey', 'Value': 'somevalue' }, { 'Key': 'someotherkey', 'Value': 'someothervalue' } ]) # Get role: role = conn.get_role(RoleName='my-role')['Role'] assert len(role['Tags']) == 2 assert role['Tags'][0]['Key'] == 'somekey' assert role['Tags'][0]['Value'] == 'somevalue' assert role['Tags'][1]['Key'] == 'someotherkey' assert role['Tags'][1]['Value'] == 'someothervalue' # Same -- but for list_role_tags: tags = conn.list_role_tags(RoleName='my-role') assert len(tags['Tags']) == 2 assert role['Tags'][0]['Key'] == 'somekey' assert role['Tags'][0]['Value'] == 'somevalue' assert role['Tags'][1]['Key'] == 'someotherkey' assert role['Tags'][1]['Value'] == 'someothervalue' assert not tags['IsTruncated'] assert not tags.get('Marker') # Test pagination: tags = conn.list_role_tags(RoleName='my-role', MaxItems=1) assert len(tags['Tags']) == 1 assert tags['IsTruncated'] assert tags['Tags'][0]['Key'] == 'somekey' assert tags['Tags'][0]['Value'] == 'somevalue' assert tags['Marker'] == '1' tags = conn.list_role_tags(RoleName='my-role', Marker=tags['Marker']) assert len(tags['Tags']) == 1 assert tags['Tags'][0]['Key'] == 'someotherkey' assert tags['Tags'][0]['Value'] == 'someothervalue' assert not tags['IsTruncated'] assert not tags.get('Marker') # Test updating an existing tag: conn.tag_role(RoleName='my-role', Tags=[ { 'Key': 'somekey', 'Value': 'somenewvalue' } ]) tags = conn.list_role_tags(RoleName='my-role') assert len(tags['Tags']) == 2 assert tags['Tags'][0]['Key'] == 'somekey' assert tags['Tags'][0]['Value'] == 'somenewvalue' # Empty is good: conn.tag_role(RoleName='my-role', Tags=[ { 'Key': 'somekey', 'Value': '' } ]) tags = conn.list_role_tags(RoleName='my-role') assert len(tags['Tags']) == 2 assert tags['Tags'][0]['Key'] == 'somekey' assert tags['Tags'][0]['Value'] == '' # Test creating tags with invalid values: # With more than 50 tags: with assert_raises(ClientError) as ce: too_many_tags = list(map(lambda x: {'Key': str(x), 'Value': str(x)}, range(0, 51))) conn.tag_role(RoleName='my-role', Tags=too_many_tags) assert 'failed to satisfy constraint: Member must have length less than or equal to 50.' \ in ce.exception.response['Error']['Message'] # With a duplicate tag: with assert_raises(ClientError) as ce: conn.tag_role(RoleName='my-role', Tags=[{'Key': '0', 'Value': ''}, {'Key': '0', 'Value': ''}]) assert 'Duplicate tag keys found. Please note that Tag keys are case insensitive.' \ in ce.exception.response['Error']['Message'] # Duplicate tag with different casing: with assert_raises(ClientError) as ce: conn.tag_role(RoleName='my-role', Tags=[{'Key': 'a', 'Value': ''}, {'Key': 'A', 'Value': ''}]) assert 'Duplicate tag keys found. Please note that Tag keys are case insensitive.' \ in ce.exception.response['Error']['Message'] # With a really big key: with assert_raises(ClientError) as ce: conn.tag_role(RoleName='my-role', Tags=[{'Key': '0' * 129, 'Value': ''}]) assert 'Member must have length less than or equal to 128.' in ce.exception.response['Error']['Message'] # With a really big value: with assert_raises(ClientError) as ce: conn.tag_role(RoleName='my-role', Tags=[{'Key': '0', 'Value': '0' * 257}]) assert 'Member must have length less than or equal to 256.' in ce.exception.response['Error']['Message'] # With an invalid character: with assert_raises(ClientError) as ce: conn.tag_role(RoleName='my-role', Tags=[{'Key': 'NOWAY!', 'Value': ''}]) assert 'Member must satisfy regular expression pattern: [\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]+' \ in ce.exception.response['Error']['Message'] # With a role that doesn't exist: with assert_raises(ClientError): conn.tag_role(RoleName='notarole', Tags=[{'Key': 'some', 'Value': 'value'}]) @mock_iam def test_untag_role(): conn = boto3.client('iam', region_name='us-east-1') conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="{}") # With proper tag values: conn.tag_role(RoleName='my-role', Tags=[ { 'Key': 'somekey', 'Value': 'somevalue' }, { 'Key': 'someotherkey', 'Value': 'someothervalue' } ]) # Remove them: conn.untag_role(RoleName='my-role', TagKeys=['somekey']) tags = conn.list_role_tags(RoleName='my-role') assert len(tags['Tags']) == 1 assert tags['Tags'][0]['Key'] == 'someotherkey' assert tags['Tags'][0]['Value'] == 'someothervalue' # And again: conn.untag_role(RoleName='my-role', TagKeys=['someotherkey']) tags = conn.list_role_tags(RoleName='my-role') assert not tags['Tags'] # Test removing tags with invalid values: # With more than 50 tags: with assert_raises(ClientError) as ce: conn.untag_role(RoleName='my-role', TagKeys=[str(x) for x in range(0, 51)]) assert 'failed to satisfy constraint: Member must have length less than or equal to 50.' \ in ce.exception.response['Error']['Message'] assert 'tagKeys' in ce.exception.response['Error']['Message'] # With a really big key: with assert_raises(ClientError) as ce: conn.untag_role(RoleName='my-role', TagKeys=['0' * 129]) assert 'Member must have length less than or equal to 128.' in ce.exception.response['Error']['Message'] assert 'tagKeys' in ce.exception.response['Error']['Message'] # With an invalid character: with assert_raises(ClientError) as ce: conn.untag_role(RoleName='my-role', TagKeys=['NOWAY!']) assert 'Member must satisfy regular expression pattern: [\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]+' \ in ce.exception.response['Error']['Message'] assert 'tagKeys' in ce.exception.response['Error']['Message'] # With a role that doesn't exist: with assert_raises(ClientError): conn.untag_role(RoleName='notarole', TagKeys=['somevalue']) @mock_iam() def test_update_role_description(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(ClientError): conn.delete_role(RoleName="my-role") conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/") response = conn.update_role_description(RoleName="my-role", Description="test") assert response['Role']['RoleName'] == 'my-role' @mock_iam() def test_update_role(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(ClientError): conn.delete_role(RoleName="my-role") conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/") response = conn.update_role_description(RoleName="my-role", Description="test") assert response['Role']['RoleName'] == 'my-role' @mock_iam() def test_update_role(): conn = boto3.client('iam', region_name='us-east-1') with assert_raises(ClientError): conn.delete_role(RoleName="my-role") conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/") response = conn.update_role(RoleName="my-role", Description="test") assert len(response.keys()) == 1 @mock_iam() def test_list_entities_for_policy(): test_policy = json.dumps({ "Version": "2012-10-17", "Statement": [ { "Action": "s3:ListBucket", "Resource": "*", "Effect": "Allow", } ] }) conn = boto3.client('iam', region_name='us-east-1') conn.create_role(RoleName="my-role", AssumeRolePolicyDocument="some policy", Path="/my-path/") conn.create_user(Path='/', UserName='testUser') conn.create_group(Path='/', GroupName='testGroup') conn.create_policy( PolicyName='testPolicy', Path='/', PolicyDocument=test_policy, Description='Test Policy' ) # Attach things to the user and group: conn.put_user_policy(UserName='testUser', PolicyName='testPolicy', PolicyDocument=test_policy) conn.put_group_policy(GroupName='testGroup', PolicyName='testPolicy', PolicyDocument=test_policy) conn.attach_user_policy(UserName='testUser', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy') conn.attach_group_policy(GroupName='testGroup', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy') conn.add_user_to_group(UserName='testUser', GroupName='testGroup') # Add things to the role: conn.create_instance_profile(InstanceProfileName='ipn') conn.add_role_to_instance_profile(InstanceProfileName='ipn', RoleName='my-role') conn.tag_role(RoleName='my-role', Tags=[ { 'Key': 'somekey', 'Value': 'somevalue' }, { 'Key': 'someotherkey', 'Value': 'someothervalue' } ]) conn.put_role_policy(RoleName='my-role', PolicyName='test-policy', PolicyDocument=test_policy) conn.attach_role_policy(RoleName='my-role', PolicyArn='arn:aws:iam::123456789012:policy/testPolicy') response = conn.list_entities_for_policy( PolicyArn='arn:aws:iam::123456789012:policy/testPolicy', EntityFilter='Role' ) assert response['PolicyRoles'] == [{'RoleName': 'my-role'}] response = conn.list_entities_for_policy( PolicyArn='arn:aws:iam::123456789012:policy/testPolicy', EntityFilter='User', ) assert response['PolicyUsers'] == [{'UserName': 'testUser'}] response = conn.list_entities_for_policy( PolicyArn='arn:aws:iam::123456789012:policy/testPolicy', EntityFilter='Group', ) assert response['PolicyGroups'] == [{'GroupName': 'testGroup'}] response = conn.list_entities_for_policy( PolicyArn='arn:aws:iam::123456789012:policy/testPolicy', EntityFilter='LocalManagedPolicy', ) assert response['PolicyGroups'] == [{'GroupName': 'testGroup'}] assert response['PolicyUsers'] == [{'UserName': 'testUser'}] assert response['PolicyRoles'] == [{'RoleName': 'my-role'}] @mock_iam() def test_create_role_no_path(): conn = boto3.client('iam', region_name='us-east-1') resp = conn.create_role(RoleName='my-role', AssumeRolePolicyDocument='some policy', Description='test') resp.get('Role').get('Arn').should.equal('arn:aws:iam::123456789012:role/my-role') resp.get('Role').should_not.have.key('PermissionsBoundary') resp.get('Role').get('Description').should.equal('test') @mock_iam() def test_create_role_with_permissions_boundary(): conn = boto3.client('iam', region_name='us-east-1') boundary = 'arn:aws:iam::123456789012:policy/boundary' resp = conn.create_role(RoleName='my-role', AssumeRolePolicyDocument='some policy', Description='test', PermissionsBoundary=boundary) expected = { 'PermissionsBoundaryType': 'PermissionsBoundaryPolicy', 'PermissionsBoundaryArn': boundary } resp.get('Role').get('PermissionsBoundary').should.equal(expected) resp.get('Role').get('Description').should.equal('test') invalid_boundary_arn = 'arn:aws:iam::123456789:not_a_boundary' with assert_raises(ClientError): conn.create_role(RoleName='bad-boundary', AssumeRolePolicyDocument='some policy', Description='test', PermissionsBoundary=invalid_boundary_arn) # Ensure the PermissionsBoundary is included in role listing as well conn.list_roles().get('Roles')[0].get('PermissionsBoundary').should.equal(expected) @mock_iam def test_create_open_id_connect_provider(): client = boto3.client('iam', region_name='us-east-1') response = client.create_open_id_connect_provider( Url='https://example.com', ThumbprintList=[] # even it is required to provide at least one thumbprint, AWS accepts an empty list ) response['OpenIDConnectProviderArn'].should.equal( 'arn:aws:iam::123456789012:oidc-provider/example.com' ) response = client.create_open_id_connect_provider( Url='http://example.org', ThumbprintList=[ 'b' * 40 ], ClientIDList=[ 'b' ] ) response['OpenIDConnectProviderArn'].should.equal( 'arn:aws:iam::123456789012:oidc-provider/example.org' ) response = client.create_open_id_connect_provider( Url='http://example.org/oidc', ThumbprintList=[] ) response['OpenIDConnectProviderArn'].should.equal( 'arn:aws:iam::123456789012:oidc-provider/example.org/oidc' ) response = client.create_open_id_connect_provider( Url='http://example.org/oidc-query?test=true', ThumbprintList=[] ) response['OpenIDConnectProviderArn'].should.equal( 'arn:aws:iam::123456789012:oidc-provider/example.org/oidc-query' ) @mock_iam def test_create_open_id_connect_provider_errors(): client = boto3.client('iam', region_name='us-east-1') client.create_open_id_connect_provider( Url='https://example.com', ThumbprintList=[] ) client.create_open_id_connect_provider.when.called_with( Url='https://example.com', ThumbprintList=[] ).should.throw( ClientError, 'Unknown' ) client.create_open_id_connect_provider.when.called_with( Url='example.org', ThumbprintList=[] ).should.throw( ClientError, 'Invalid Open ID Connect Provider URL' ) client.create_open_id_connect_provider.when.called_with( Url='example', ThumbprintList=[] ).should.throw( ClientError, 'Invalid Open ID Connect Provider URL' ) client.create_open_id_connect_provider.when.called_with( Url='http://example.org', ThumbprintList=[ 'a' * 40, 'b' * 40, 'c' * 40, 'd' * 40, 'e' * 40, 'f' * 40, ] ).should.throw( ClientError, 'Thumbprint list must contain fewer than 5 entries.' ) too_many_client_ids = ['{}'.format(i) for i in range(101)] client.create_open_id_connect_provider.when.called_with( Url='http://example.org', ThumbprintList=[], ClientIDList=too_many_client_ids ).should.throw( ClientError, 'Cannot exceed quota for ClientIdsPerOpenIdConnectProvider: 100' ) too_long_url = 'b' * 256 too_long_thumbprint = 'b' * 41 too_long_client_id = 'b' * 256 client.create_open_id_connect_provider.when.called_with( Url=too_long_url, ThumbprintList=[ too_long_thumbprint ], ClientIDList=[ too_long_client_id ] ).should.throw( ClientError, '3 validation errors detected: ' 'Value "{0}" at "clientIDList" failed to satisfy constraint: ' 'Member must satisfy constraint: ' '[Member must have length less than or equal to 255, ' 'Member must have length greater than or equal to 1]; ' 'Value "{1}" at "thumbprintList" failed to satisfy constraint: ' 'Member must satisfy constraint: ' '[Member must have length less than or equal to 40, ' 'Member must have length greater than or equal to 40]; ' 'Value "{2}" at "url" failed to satisfy constraint: ' 'Member must have length less than or equal to 255'.format([too_long_client_id], [too_long_thumbprint], too_long_url) ) @mock_iam def test_delete_open_id_connect_provider(): client = boto3.client('iam', region_name='us-east-1') response = client.create_open_id_connect_provider( Url='https://example.com', ThumbprintList=[] ) open_id_arn = response['OpenIDConnectProviderArn'] client.delete_open_id_connect_provider( OpenIDConnectProviderArn=open_id_arn ) client.get_open_id_connect_provider.when.called_with( OpenIDConnectProviderArn=open_id_arn ).should.throw( ClientError, 'OpenIDConnect Provider not found for arn {}'.format(open_id_arn) ) # deleting a non existing provider should be successful client.delete_open_id_connect_provider( OpenIDConnectProviderArn=open_id_arn ) @mock_iam def test_get_open_id_connect_provider(): client = boto3.client('iam', region_name='us-east-1') response = client.create_open_id_connect_provider( Url='https://example.com', ThumbprintList=[ 'b' * 40 ], ClientIDList=[ 'b' ] ) open_id_arn = response['OpenIDConnectProviderArn'] response = client.get_open_id_connect_provider( OpenIDConnectProviderArn=open_id_arn ) response['Url'].should.equal('example.com') response['ThumbprintList'].should.equal([ 'b' * 40 ]) response['ClientIDList'].should.equal([ 'b' ]) response.should.have.key('CreateDate').should.be.a(datetime) @mock_iam def test_get_open_id_connect_provider_errors(): client = boto3.client('iam', region_name='us-east-1') response = client.create_open_id_connect_provider( Url='https://example.com', ThumbprintList=[ 'b' * 40 ], ClientIDList=[ 'b' ] ) open_id_arn = response['OpenIDConnectProviderArn'] client.get_open_id_connect_provider.when.called_with( OpenIDConnectProviderArn=open_id_arn + '-not-existing' ).should.throw( ClientError, 'OpenIDConnect Provider not found for arn {}'.format(open_id_arn + '-not-existing') ) @mock_iam def test_list_open_id_connect_providers(): client = boto3.client('iam', region_name='us-east-1') response = client.create_open_id_connect_provider( Url='https://example.com', ThumbprintList=[] ) open_id_arn_1 = response['OpenIDConnectProviderArn'] response = client.create_open_id_connect_provider( Url='http://example.org', ThumbprintList=[ 'b' * 40 ], ClientIDList=[ 'b' ] ) open_id_arn_2 = response['OpenIDConnectProviderArn'] response = client.create_open_id_connect_provider( Url='http://example.org/oidc', ThumbprintList=[] ) open_id_arn_3 = response['OpenIDConnectProviderArn'] response = client.list_open_id_connect_providers() sorted(response['OpenIDConnectProviderList'], key=lambda i: i['Arn']).should.equal( [ { 'Arn': open_id_arn_1 }, { 'Arn': open_id_arn_2 }, { 'Arn': open_id_arn_3 } ] )