IAM Role Tagging support

This commit is contained in:
Mike Grima 2019-01-29 18:09:31 -08:00
parent 7e211eb6ea
commit 1a36c0c377
6 changed files with 426 additions and 16 deletions

View File

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

View File

@ -32,3 +32,48 @@ class MalformedCertificate(RESTError):
def __init__(self, cert): def __init__(self, cert):
super(MalformedCertificate, self).__init__( super(MalformedCertificate, self).__init__(
'MalformedCertificate', 'Certificate {cert} is malformed'.format(cert=cert)) '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 import sys
from datetime import datetime from datetime import datetime
import json import json
import re
from cryptography import x509 from cryptography import x509
from cryptography.hazmat.backends import default_backend 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 moto.core.utils import iso_8601_datetime_without_milliseconds
from .aws_managed_policies import aws_managed_policies_data 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 from .utils import random_access_key, random_alphanumeric, random_resource_id, random_policy_id
ACCOUNT_ID = 123456789012 ACCOUNT_ID = 123456789012
@ -32,7 +34,6 @@ class MFADevice(object):
class Policy(BaseModel): class Policy(BaseModel):
is_attachable = False is_attachable = False
def __init__(self, def __init__(self,
@ -132,6 +133,7 @@ class Role(BaseModel):
self.policies = {} self.policies = {}
self.managed_policies = {} self.managed_policies = {}
self.create_date = datetime.now(pytz.utc) self.create_date = datetime.now(pytz.utc)
self.tags = {}
@classmethod @classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json, region_name): 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 NotImplementedError('"Fn::GetAtt" : [ "{0}" , "Arn" ]"')
raise UnformattedGetAttTemplateException() raise UnformattedGetAttTemplateException()
def get_tags(self):
return [self.tags[tag] for tag in self.tags]
class InstanceProfile(BaseModel): class InstanceProfile(BaseModel):
@ -614,6 +619,86 @@ class IAMBackend(BaseBackend):
role = self.get_role(role_name) role = self.get_role(role_name)
return role.policies.keys() 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): def create_policy_version(self, policy_arn, policy_document, set_as_default):
policy = self.get_policy(policy_arn) policy = self.get_policy(policy_arn)
if not policy: if not policy:

View File

@ -625,6 +625,34 @@ class IamResponse(BaseResponse):
template = self.response_template(LIST_SIGNING_CERTIFICATES_TEMPLATE) template = self.response_template(LIST_SIGNING_CERTIFICATES_TEMPLATE)
return template.render(user_name=user_name, certificates=certs) 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> ATTACH_ROLE_POLICY_TEMPLATE = """<AttachRolePolicyResponse>
<ResponseMetadata> <ResponseMetadata>
@ -878,6 +906,16 @@ GET_ROLE_TEMPLATE = """<GetRoleResponse xmlns="https://iam.amazonaws.com/doc/201
<AssumeRolePolicyDocument>{{ role.assume_role_policy_document }}</AssumeRolePolicyDocument> <AssumeRolePolicyDocument>{{ role.assume_role_policy_document }}</AssumeRolePolicyDocument>
<CreateDate>{{ role.create_date }}</CreateDate> <CreateDate>{{ role.create_date }}</CreateDate>
<RoleId>{{ role.id }}</RoleId> <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> </Role>
</GetRoleResult> </GetRoleResult>
<ResponseMetadata> <ResponseMetadata>
@ -1503,6 +1541,14 @@ GET_ACCOUNT_AUTHORIZATION_DETAILS_TEMPLATE = """<GetAccountAuthorizationDetailsR
</member> </member>
{% endfor %} {% endfor %}
</AttachedManagedPolicies> </AttachedManagedPolicies>
<Tags>
{% for tag in role.get_tags() %}
<member>
<Key>{{ tag['Key'] }}</Key>
<Value>{{ tag['Value'] }}</Value>
</member>
{% endfor %}
</Tags>
<InstanceProfileList> <InstanceProfileList>
{% for profile in instance_profiles %} {% for profile in instance_profiles %}
<member> <member>
@ -1671,3 +1717,38 @@ LIST_SIGNING_CERTIFICATES_TEMPLATE = """<ListSigningCertificatesResponse>
<RequestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</RequestId> <RequestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</RequestId>
</ResponseMetadata> </ResponseMetadata>
</ListSigningCertificatesResponse>""" </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 = [ install_requires = [
"Jinja2>=2.7.3", "Jinja2>=2.7.3",
"boto>=2.36.0", "boto>=2.36.0",
"boto3>=1.6.16", "boto3>=1.9.86",
"botocore>=1.12.13", "botocore>=1.12.86",
"cryptography>=2.3.0", "cryptography>=2.3.0",
"requests>=2.5", "requests>=2.5",
"xmltodict", "xmltodict",

View File

@ -306,6 +306,7 @@ def test_create_policy_versions():
PolicyDocument='{"some":"policy"}') PolicyDocument='{"some":"policy"}')
version.get('PolicyVersion').get('Document').should.equal({'some': 'policy'}) version.get('PolicyVersion').get('Document').should.equal({'some': 'policy'})
@mock_iam @mock_iam
def test_get_policy(): def test_get_policy():
conn = boto3.client('iam', region_name='us-east-1') 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') 'get_credential_report_result']['content'].encode('ascii')).decode('ascii')
report.should.match(r'.*my-user.*') report.should.match(r'.*my-user.*')
@mock_iam @mock_iam
def test_boto3_get_credential_report(): def test_boto3_get_credential_report():
conn = boto3.client('iam', region_name='us-east-1') conn = boto3.client('iam', region_name='us-east-1')
@ -780,12 +782,24 @@ def test_get_account_authorization_details():
conn.create_instance_profile(InstanceProfileName='ipn') conn.create_instance_profile(InstanceProfileName='ipn')
conn.add_role_to_instance_profile(InstanceProfileName='ipn', RoleName='my-role') 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'
}
])
result = conn.get_account_authorization_details(Filter=['Role']) result = conn.get_account_authorization_details(Filter=['Role'])
assert len(result['RoleDetailList']) == 1 assert len(result['RoleDetailList']) == 1
assert len(result['UserDetailList']) == 0 assert len(result['UserDetailList']) == 0
assert len(result['GroupDetailList']) == 0 assert len(result['GroupDetailList']) == 0
assert len(result['Policies']) == 0 assert len(result['Policies']) == 0
assert len(result['RoleDetailList'][0]['InstanceProfileList']) == 1 assert len(result['RoleDetailList'][0]['InstanceProfileList']) == 1
assert len(result['RoleDetailList'][0]['Tags']) == 2
result = conn.get_account_authorization_details(Filter=['User']) result = conn.get_account_authorization_details(Filter=['User'])
assert len(result['RoleDetailList']) == 0 assert len(result['RoleDetailList']) == 0
@ -872,6 +886,7 @@ def test_signing_certs():
with assert_raises(ClientError): with assert_raises(ClientError):
client.delete_signing_certificate(UserName='notauser', CertificateId=cert_id) client.delete_signing_certificate(UserName='notauser', CertificateId=cert_id)
@mock_iam() @mock_iam()
def test_create_saml_provider(): def test_create_saml_provider():
conn = boto3.client('iam', region_name='us-east-1') conn = boto3.client('iam', region_name='us-east-1')
@ -881,6 +896,7 @@ def test_create_saml_provider():
) )
response['SAMLProviderArn'].should.equal("arn:aws:iam::123456789012:saml-provider/TestSAMLProvider") response['SAMLProviderArn'].should.equal("arn:aws:iam::123456789012:saml-provider/TestSAMLProvider")
@mock_iam() @mock_iam()
def test_get_saml_provider(): def test_get_saml_provider():
conn = boto3.client('iam', region_name='us-east-1') conn = boto3.client('iam', region_name='us-east-1')
@ -893,6 +909,7 @@ def test_get_saml_provider():
) )
response['SAMLMetadataDocument'].should.equal('a' * 1024) response['SAMLMetadataDocument'].should.equal('a' * 1024)
@mock_iam() @mock_iam()
def test_list_saml_providers(): def test_list_saml_providers():
conn = boto3.client('iam', region_name='us-east-1') conn = boto3.client('iam', region_name='us-east-1')
@ -903,6 +920,7 @@ def test_list_saml_providers():
response = conn.list_saml_providers() response = conn.list_saml_providers()
response['SAMLProviderList'][0]['Arn'].should.equal("arn:aws:iam::123456789012:saml-provider/TestSAMLProvider") response['SAMLProviderList'][0]['Arn'].should.equal("arn:aws:iam::123456789012:saml-provider/TestSAMLProvider")
@mock_iam() @mock_iam()
def test_delete_saml_provider(): def test_delete_saml_provider():
conn = boto3.client('iam', region_name='us-east-1') conn = boto3.client('iam', region_name='us-east-1')
@ -929,3 +947,178 @@ def test_delete_saml_provider():
# Verify that it's not in the list: # Verify that it's not in the list:
resp = conn.list_signing_certificates(UserName='testing') resp = conn.list_signing_certificates(UserName='testing')
assert not resp['Certificates'] 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'])