Merge pull request #2041 from mikegrima/roletags

IAM Role Tagging support
This commit is contained in:
Mike Grima 2019-02-06 11:40:59 -08:00 committed by GitHub
commit f41c4e756f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 496 additions and 47 deletions

View File

@ -2208,7 +2208,7 @@
- [ ] describe_event_types
- [ ] describe_events
## iam - 48% implemented
## iam - 62% implemented
- [ ] add_client_id_to_open_id_connect_provider
- [X] add_role_to_instance_profile
- [X] add_user_to_group
@ -2247,7 +2247,7 @@
- [X] delete_server_certificate
- [ ] delete_service_linked_role
- [ ] delete_service_specific_credential
- [ ] delete_signing_certificate
- [X] delete_signing_certificate
- [ ] delete_ssh_public_key
- [X] delete_user
- [X] delete_user_policy
@ -2279,7 +2279,7 @@
- [ ] get_ssh_public_key
- [X] get_user
- [X] get_user_policy
- [ ] list_access_keys
- [X] list_access_keys
- [X] list_account_aliases
- [X] list_attached_group_policies
- [X] list_attached_role_policies
@ -2287,19 +2287,21 @@
- [ ] list_entities_for_policy
- [X] list_group_policies
- [X] list_groups
- [ ] list_groups_for_user
- [ ] list_instance_profiles
- [ ] list_instance_profiles_for_role
- [X] list_groups_for_user
- [X] list_instance_profiles
- [X] list_instance_profiles_for_role
- [X] list_mfa_devices
- [ ] list_open_id_connect_providers
- [X] list_policies
- [X] list_policy_versions
- [X] list_role_policies
- [ ] list_roles
- [X] list_roles
- [X] list_role_tags
- [ ] list_user_tags
- [X] list_saml_providers
- [ ] list_server_certificates
- [X] list_server_certificates
- [ ] list_service_specific_credentials
- [ ] list_signing_certificates
- [X] list_signing_certificates
- [ ] list_ssh_public_keys
- [X] list_user_policies
- [X] list_users
@ -2315,6 +2317,10 @@
- [ ] set_default_policy_version
- [ ] simulate_custom_policy
- [ ] simulate_principal_policy
- [X] tag_role
- [ ] tag_user
- [X] untag_role
- [ ] untag_user
- [X] update_access_key
- [ ] update_account_password_policy
- [ ] update_assume_role_policy
@ -2326,11 +2332,11 @@
- [X] update_saml_provider
- [ ] update_server_certificate
- [ ] update_service_specific_credential
- [ ] update_signing_certificate
- [X] update_signing_certificate
- [ ] update_ssh_public_key
- [ ] update_user
- [ ] upload_server_certificate
- [ ] upload_signing_certificate
- [X] upload_server_certificate
- [X] upload_signing_certificate
- [ ] upload_ssh_public_key
## importexport - 0% implemented

View File

@ -32,3 +32,48 @@ class MalformedCertificate(RESTError):
def __init__(self, cert):
super(MalformedCertificate, self).__init__(
'MalformedCertificate', 'Certificate {cert} is malformed'.format(cert=cert))
class DuplicateTags(RESTError):
code = 400
def __init__(self):
super(DuplicateTags, self).__init__(
'InvalidInput', 'Duplicate tag keys found. Please note that Tag keys are case insensitive.')
class TagKeyTooBig(RESTError):
code = 400
def __init__(self, tag, param='tags.X.member.key'):
super(TagKeyTooBig, self).__init__(
'ValidationError', "1 validation error detected: Value '{}' at '{}' failed to satisfy "
"constraint: Member must have length less than or equal to 128.".format(tag, param))
class TagValueTooBig(RESTError):
code = 400
def __init__(self, tag):
super(TagValueTooBig, self).__init__(
'ValidationError', "1 validation error detected: Value '{}' at 'tags.X.member.value' failed to satisfy "
"constraint: Member must have length less than or equal to 256.".format(tag))
class InvalidTagCharacters(RESTError):
code = 400
def __init__(self, tag, param='tags.X.member.key'):
message = "1 validation error detected: Value '{}' at '{}' failed to satisfy ".format(tag, param)
message += "constraint: Member must satisfy regular expression pattern: [\\p{L}\\p{Z}\\p{N}_.:/=+\\-@]+"
super(InvalidTagCharacters, self).__init__('ValidationError', message)
class TooManyTags(RESTError):
code = 400
def __init__(self, tags, param='tags'):
super(TooManyTags, self).__init__(
'ValidationError', "1 validation error detected: Value '{}' at '{}' failed to satisfy "
"constraint: Member must have length less than or equal to 50.".format(tags, param))

View File

@ -3,6 +3,7 @@ import base64
import sys
from datetime import datetime
import json
import re
from cryptography import x509
from cryptography.hazmat.backends import default_backend
@ -12,7 +13,8 @@ from moto.core import BaseBackend, BaseModel
from moto.core.utils import iso_8601_datetime_without_milliseconds
from .aws_managed_policies import aws_managed_policies_data
from .exceptions import IAMNotFoundException, IAMConflictException, IAMReportNotPresentException, MalformedCertificate
from .exceptions import IAMNotFoundException, IAMConflictException, IAMReportNotPresentException, MalformedCertificate, \
DuplicateTags, TagKeyTooBig, InvalidTagCharacters, TooManyTags, TagValueTooBig
from .utils import random_access_key, random_alphanumeric, random_resource_id, random_policy_id
ACCOUNT_ID = 123456789012
@ -32,7 +34,6 @@ class MFADevice(object):
class Policy(BaseModel):
is_attachable = False
def __init__(self,
@ -132,6 +133,7 @@ class Role(BaseModel):
self.policies = {}
self.managed_policies = {}
self.create_date = datetime.now(pytz.utc)
self.tags = {}
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name):
@ -175,6 +177,9 @@ class Role(BaseModel):
raise NotImplementedError('"Fn::GetAtt" : [ "{0}" , "Arn" ]"')
raise UnformattedGetAttTemplateException()
def get_tags(self):
return [self.tags[tag] for tag in self.tags]
class InstanceProfile(BaseModel):
@ -614,6 +619,86 @@ class IAMBackend(BaseBackend):
role = self.get_role(role_name)
return role.policies.keys()
def _validate_tag_key(self, tag_key, exception_param='tags.X.member.key'):
"""Validates the tag key.
:param all_tags: Dict to check if there is a duplicate tag.
:param tag_key: The tag key to check against.
:param exception_param: The exception parameter to send over to help format the message. This is to reflect
the difference between the tag and untag APIs.
:return:
"""
# Validate that the key length is correct:
if len(tag_key) > 128:
raise TagKeyTooBig(tag_key, param=exception_param)
# Validate that the tag key fits the proper Regex:
# [\w\s_.:/=+\-@]+ SHOULD be the same as the Java regex on the AWS documentation: [\p{L}\p{Z}\p{N}_.:/=+\-@]+
match = re.findall(r'[\w\s_.:/=+\-@]+', tag_key)
# Kudos if you can come up with a better way of doing a global search :)
if not len(match) or len(match[0]) < len(tag_key):
raise InvalidTagCharacters(tag_key, param=exception_param)
def _check_tag_duplicate(self, all_tags, tag_key):
"""Validates that a tag key is not a duplicate
:param all_tags: Dict to check if there is a duplicate tag.
:param tag_key: The tag key to check against.
:return:
"""
if tag_key in all_tags:
raise DuplicateTags()
def list_role_tags(self, role_name, marker, max_items=100):
role = self.get_role(role_name)
max_items = int(max_items)
tag_index = sorted(role.tags)
start_idx = int(marker) if marker else 0
tag_index = tag_index[start_idx:start_idx + max_items]
if len(role.tags) <= (start_idx + max_items):
marker = None
else:
marker = str(start_idx + max_items)
# Make the tag list of dict's:
tags = [role.tags[tag] for tag in tag_index]
return tags, marker
def tag_role(self, role_name, tags):
if len(tags) > 50:
raise TooManyTags(tags)
role = self.get_role(role_name)
tag_keys = {}
for tag in tags:
# Need to index by the lowercase tag key since the keys are case insensitive, but their case is retained.
ref_key = tag['Key'].lower()
self._check_tag_duplicate(tag_keys, ref_key)
self._validate_tag_key(tag['Key'])
if len(tag['Value']) > 256:
raise TagValueTooBig(tag['Value'])
tag_keys[ref_key] = tag
role.tags.update(tag_keys)
def untag_role(self, role_name, tag_keys):
if len(tag_keys) > 50:
raise TooManyTags(tag_keys, param='tagKeys')
role = self.get_role(role_name)
for key in tag_keys:
ref_key = key.lower()
self._validate_tag_key(key, exception_param='tagKeys')
role.tags.pop(ref_key, None)
def create_policy_version(self, policy_arn, policy_document, set_as_default):
policy = self.get_policy(policy_arn)
if not policy:

View File

@ -554,7 +554,8 @@ class IamResponse(BaseResponse):
policies=account_details['managed_policies'],
users=account_details['users'],
groups=account_details['groups'],
roles=account_details['roles']
roles=account_details['roles'],
get_groups_for_user=iam_backend.get_groups_for_user
)
def create_saml_provider(self):
@ -625,6 +626,34 @@ class IamResponse(BaseResponse):
template = self.response_template(LIST_SIGNING_CERTIFICATES_TEMPLATE)
return template.render(user_name=user_name, certificates=certs)
def list_role_tags(self):
role_name = self._get_param('RoleName')
marker = self._get_param('Marker')
max_items = self._get_param('MaxItems', 100)
tags, marker = iam_backend.list_role_tags(role_name, marker, max_items)
template = self.response_template(LIST_ROLE_TAG_TEMPLATE)
return template.render(tags=tags, marker=marker)
def tag_role(self):
role_name = self._get_param('RoleName')
tags = self._get_multi_param('Tags.member')
iam_backend.tag_role(role_name, tags)
template = self.response_template(TAG_ROLE_TEMPLATE)
return template.render()
def untag_role(self):
role_name = self._get_param('RoleName')
tag_keys = self._get_multi_param('TagKeys.member')
iam_backend.untag_role(role_name, tag_keys)
template = self.response_template(UNTAG_ROLE_TEMPLATE)
return template.render()
ATTACH_ROLE_POLICY_TEMPLATE = """<AttachRolePolicyResponse>
<ResponseMetadata>
@ -878,6 +907,16 @@ GET_ROLE_TEMPLATE = """<GetRoleResponse xmlns="https://iam.amazonaws.com/doc/201
<AssumeRolePolicyDocument>{{ role.assume_role_policy_document }}</AssumeRolePolicyDocument>
<CreateDate>{{ role.create_date }}</CreateDate>
<RoleId>{{ role.id }}</RoleId>
{% if role.tags %}
<Tags>
{% for tag in role.get_tags() %}
<member>
<Key>{{ tag['Key'] }}</Key>
<Value>{{ tag['Value'] }}</Value>
</member>
{% endfor %}
</Tags>
{% endif %}
</Role>
</GetRoleResult>
<ResponseMetadata>
@ -1461,8 +1500,19 @@ GET_ACCOUNT_AUTHORIZATION_DETAILS_TEMPLATE = """<GetAccountAuthorizationDetailsR
<UserDetailList>
{% for user in users %}
<member>
<GroupList />
<AttachedManagedPolicies/>
<GroupList>
{% for group in get_groups_for_user(user.name) %}
<member>{{ group.name }}</member>
{% endfor %}
</GroupList>
<AttachedManagedPolicies>
{% for policy in user.managed_policies %}
<member>
<PolicyName>{{ user.managed_policies[policy].name }}</PolicyName>
<PolicyArn>{{ policy }}</PolicyArn>
</member>
{% endfor %}
</AttachedManagedPolicies>
<UserId>{{ user.id }}</UserId>
<Path>{{ user.path }}</Path>
<UserName>{{ user.name }}</UserName>
@ -1476,25 +1526,39 @@ GET_ACCOUNT_AUTHORIZATION_DETAILS_TEMPLATE = """<GetAccountAuthorizationDetailsR
<member>
<GroupId>{{ group.id }}</GroupId>
<AttachedManagedPolicies>
{% for policy in group.managed_policies %}
<member>
<PolicyName>{{ policy.name }}</PolicyName>
<PolicyArn>{{ policy.arn }}</PolicyArn>
</member>
{% for policy_arn in group.managed_policies %}
<member>
<PolicyName>{{ group.managed_policies[policy_arn].name }}</PolicyName>
<PolicyArn>{{ policy_arn }}</PolicyArn>
</member>
{% endfor %}
</AttachedManagedPolicies>
<GroupName>{{ group.name }}</GroupName>
<Path>{{ group.path }}</Path>
<Arn>{{ group.arn }}</Arn>
<CreateDate>{{ group.create_date }}</CreateDate>
<GroupPolicyList/>
<GroupPolicyList>
{% for policy in group.policies %}
<member>
<PolicyName>{{ policy }}</PolicyName>
<PolicyDocument>{{ group.get_policy(policy) }}</PolicyDocument>
</member>
{% endfor %}
</GroupPolicyList>
</member>
{% endfor %}
</GroupDetailList>
<RoleDetailList>
{% for role in roles %}
<member>
<RolePolicyList/>
<RolePolicyList>
{% for inline_policy in role.policies %}
<member>
<PolicyName>{{ inline_policy }}</PolicyName>
<PolicyDocument>{{ role.policies[inline_policy] }}</PolicyDocument>
</member>
{% endfor %}
</RolePolicyList>
<AttachedManagedPolicies>
{% for policy in role.managed_policies %}
<member>
@ -1503,6 +1567,14 @@ GET_ACCOUNT_AUTHORIZATION_DETAILS_TEMPLATE = """<GetAccountAuthorizationDetailsR
</member>
{% endfor %}
</AttachedManagedPolicies>
<Tags>
{% for tag in role.get_tags() %}
<member>
<Key>{{ tag['Key'] }}</Key>
<Value>{{ tag['Value'] }}</Value>
</member>
{% endfor %}
</Tags>
<InstanceProfileList>
{% for profile in instance_profiles %}
<member>
@ -1543,19 +1615,14 @@ GET_ACCOUNT_AUTHORIZATION_DETAILS_TEMPLATE = """<GetAccountAuthorizationDetailsR
<PolicyId>{{ policy.id }}</PolicyId>
<Path>{{ policy.path }}</Path>
<PolicyVersionList>
{% for policy_version in policy.versions %}
<member>
<Document>
{"Version":"2012-10-17","Statement":{"Effect":"Allow",
"Action":["iam:CreatePolicy","iam:CreatePolicyVersion",
"iam:DeletePolicy","iam:DeletePolicyVersion","iam:GetPolicy",
"iam:GetPolicyVersion","iam:ListPolicies",
"iam:ListPolicyVersions","iam:SetDefaultPolicyVersion"],
"Resource":"*"}}
</Document>
<IsDefaultVersion>true</IsDefaultVersion>
<VersionId>v1</VersionId>
<CreateDate>2012-05-09T16:27:11Z</CreateDate>
<Document>{{ policy_version.document }}</Document>
<IsDefaultVersion>{{ policy_version.is_default }}</IsDefaultVersion>
<VersionId>{{ policy_version.version_id }}</VersionId>
<CreateDate>{{ policy_version.create_datetime }}</CreateDate>
</member>
{% endfor %}
</PolicyVersionList>
<Arn>{{ policy.arn }}</Arn>
<AttachmentCount>1</AttachmentCount>
@ -1671,3 +1738,38 @@ LIST_SIGNING_CERTIFICATES_TEMPLATE = """<ListSigningCertificatesResponse>
<RequestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</RequestId>
</ResponseMetadata>
</ListSigningCertificatesResponse>"""
TAG_ROLE_TEMPLATE = """<TagRoleResponse xmlns="https://iam.amazonaws.com/doc/2010-05-08/">
<ResponseMetadata>
<RequestId>EXAMPLE8-90ab-cdef-fedc-ba987EXAMPLE</RequestId>
</ResponseMetadata>
</TagRoleResponse>"""
LIST_ROLE_TAG_TEMPLATE = """<ListRoleTagsResponse xmlns="https://iam.amazonaws.com/doc/2010-05-08/">
<ListRoleTagsResult>
<IsTruncated>{{ 'true' if marker else 'false' }}</IsTruncated>
{% if marker %}
<Marker>{{ marker }}</Marker>
{% endif %}
<Tags>
{% for tag in tags %}
<member>
<Key>{{ tag['Key'] }}</Key>
<Value>{{ tag['Value'] }}</Value>
</member>
{% endfor %}
</Tags>
</ListRoleTagsResult>
<ResponseMetadata>
<RequestId>EXAMPLE8-90ab-cdef-fedc-ba987EXAMPLE</RequestId>
</ResponseMetadata>
</ListRoleTagsResponse>"""
UNTAG_ROLE_TEMPLATE = """<UntagRoleResponse xmlns="https://iam.amazonaws.com/doc/2010-05-08/">
<ResponseMetadata>
<RequestId>EXAMPLE8-90ab-cdef-fedc-ba987EXAMPLE</RequestId>
</ResponseMetadata>
</UntagRoleResponse>"""

View File

@ -21,8 +21,8 @@ def read(*parts):
install_requires = [
"Jinja2>=2.7.3",
"boto>=2.36.0",
"boto3>=1.6.16",
"botocore>=1.12.13",
"boto3>=1.9.86",
"botocore>=1.12.86",
"cryptography>=2.3.0",
"requests>=2.5",
"xmltodict",

View File

@ -306,6 +306,7 @@ def test_create_policy_versions():
PolicyDocument='{"some":"policy"}')
version.get('PolicyVersion').get('Document').should.equal({'some': 'policy'})
@mock_iam
def test_get_policy():
conn = boto3.client('iam', region_name='us-east-1')
@ -579,6 +580,7 @@ def test_get_credential_report():
'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')
@ -757,6 +759,17 @@ def test_get_access_key_last_used():
@mock_iam
def test_get_account_authorization_details():
import json
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')
@ -764,21 +777,33 @@ def test_get_account_authorization_details():
conn.create_policy(
PolicyName='testPolicy',
Path='/',
PolicyDocument=json.dumps({
"Version": "2012-10-17",
"Statement": [
{
"Action": "s3:ListBucket",
"Resource": "*",
"Effect": "Allow",
}
]
}),
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)
result = conn.get_account_authorization_details(Filter=['Role'])
assert len(result['RoleDetailList']) == 1
@ -786,10 +811,14 @@ def test_get_account_authorization_details():
assert len(result['GroupDetailList']) == 0
assert len(result['Policies']) == 0
assert len(result['RoleDetailList'][0]['InstanceProfileList']) == 1
assert len(result['RoleDetailList'][0]['Tags']) == 2
assert len(result['RoleDetailList'][0]['RolePolicyList']) == 1
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
@ -797,6 +826,8 @@ def test_get_account_authorization_details():
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
result = conn.get_account_authorization_details(Filter=['LocalManagedPolicy'])
@ -804,6 +835,7 @@ def test_get_account_authorization_details():
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
@ -872,6 +904,7 @@ def test_signing_certs():
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')
@ -881,6 +914,7 @@ def test_create_saml_provider():
)
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')
@ -893,6 +927,7 @@ def test_get_saml_provider():
)
response['SAMLMetadataDocument'].should.equal('a' * 1024)
@mock_iam()
def test_list_saml_providers():
conn = boto3.client('iam', region_name='us-east-1')
@ -903,6 +938,7 @@ def test_list_saml_providers():
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')
@ -929,3 +965,178 @@ def test_delete_saml_provider():
# Verify that it's not in the list:
resp = conn.list_signing_certificates(UserName='testing')
assert not resp['Certificates']
@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'])