Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1224 from JackDanger/jack/implement-attached-group-policies

Browse Source 
Implement IAM attached group policies
This commit is contained in:
Jack Danger 2017-10-01 17:00:27 -07:00 committed by GitHub
commit 0c07c4467a
3 changed files with 124 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

88
moto/iam/models.py
View File

@ -74,21 +74,13 @@ class ManagedPolicy(Policy):
is_attachable = True
def attach_to_role(self, role):
def attach_to(self, obj):
self.attachment_count += 1
role.managed_policies[self.name] = self
obj.managed_policies[self.name] = self
def detach_from_role(self, role):
def detach_from(self, obj):
self.attachment_count -= 1
del role.managed_policies[self.name]
def attach_to_user(self, user):
self.attachment_count += 1
user.managed_policies[self.name] = self
def detach_from_user(self, user):
self.attachment_count -= 1
del user.managed_policies[self.name]
del obj.managed_policies[self.name]
class AWSManagedPolicy(ManagedPolicy):
@ -249,6 +241,7 @@ class Group(BaseModel):
)
self.users = []
self.managed_policies = {}
self.policies = {}
def get_cfn_attribute(self, attribute_name):
@ -423,25 +416,47 @@ class IAMBackend(BaseBackend):
def attach_role_policy(self, policy_arn, role_name):
arns = dict((p.arn, p) for p in self.managed_policies.values())
policy = arns[policy_arn]
policy.attach_to_role(self.get_role(role_name))
policy.attach_to(self.get_role(role_name))
def detach_role_policy(self, policy_arn, role_name):
arns = dict((p.arn, p) for p in self.managed_policies.values())
try:
policy = arns[policy_arn]
policy.detach_from_role(self.get_role(role_name))
policy.detach_from(self.get_role(role_name))
except KeyError:
raise IAMNotFoundException("Policy {0} was not found.".format(policy_arn))
def attach_group_policy(self, policy_arn, group_name):
arns = dict((p.arn, p) for p in self.managed_policies.values())
try:
policy = arns[policy_arn]
except KeyError:
raise IAMNotFoundException("Policy {0} was not found.".format(policy_arn))
policy.attach_to(self.get_group(group_name))
def detach_group_policy(self, policy_arn, group_name):
arns = dict((p.arn, p) for p in self.managed_policies.values())
try:
policy = arns[policy_arn]
except KeyError:
raise IAMNotFoundException("Policy {0} was not found.".format(policy_arn))
policy.detach_from(self.get_group(group_name))
def attach_user_policy(self, policy_arn, user_name):
arns = dict((p.arn, p) for p in self.managed_policies.values())
policy = arns[policy_arn]
policy.attach_to_user(self.get_user(user_name))
try:
policy = arns[policy_arn]
except KeyError:
raise IAMNotFoundException("Policy {0} was not found.".format(policy_arn))
policy.attach_to(self.get_user(user_name))
def detach_user_policy(self, policy_arn, user_name):
arns = dict((p.arn, p) for p in self.managed_policies.values())
policy = arns[policy_arn]
policy.detach_from_user(self.get_user(user_name))
try:
policy = arns[policy_arn]
except KeyError:
raise IAMNotFoundException("Policy {0} was not found.".format(policy_arn))
policy.detach_from(self.get_user(user_name))
def create_policy(self, description, path, policy_document, policy_name):
policy = ManagedPolicy(
@ -458,39 +473,15 @@ class IAMBackend(BaseBackend):
def list_attached_role_policies(self, role_name, marker=None, max_items=100, path_prefix='/'):
policies = self.get_role(role_name).managed_policies.values()
return self._filter_attached_policies(policies, marker, max_items, path_prefix)
if path_prefix:
policies = [p for p in policies if p.path.startswith(path_prefix)]
policies = sorted(policies, key=lambda policy: policy.name)
start_idx = int(marker) if marker else 0
policies = policies[start_idx:start_idx + max_items]
if len(policies) < max_items:
marker = None
else:
marker = str(start_idx + max_items)
return policies, marker
def list_attached_group_policies(self, group_name, marker=None, max_items=100, path_prefix='/'):
policies = self.get_group(group_name).managed_policies.values()
return self._filter_attached_policies(policies, marker, max_items, path_prefix)
def list_attached_user_policies(self, user_name, marker=None, max_items=100, path_prefix='/'):
policies = self.get_user(user_name).managed_policies.values()
if path_prefix:
policies = [p for p in policies if p.path.startswith(path_prefix)]
policies = sorted(policies, key=lambda policy: policy.name)
start_idx = int(marker) if marker else 0
policies = policies[start_idx:start_idx + max_items]
if len(policies) < max_items:
marker = None
else:
marker = str(start_idx + max_items)
return policies, marker
return self._filter_attached_policies(policies, marker, max_items, path_prefix)
def list_policies(self, marker, max_items, only_attached, path_prefix, scope):
policies = self.managed_policies.values()
@ -504,6 +495,9 @@ class IAMBackend(BaseBackend):
policies = [p for p in policies if not isinstance(
p, AWSManagedPolicy)]
return self._filter_attached_policies(policies, marker, max_items, path_prefix)
def _filter_attached_policies(self, policies, marker, max_items, path_prefix):
if path_prefix:
policies = [p for p in policies if p.path.startswith(path_prefix)]

59
moto/iam/responses.py
View File

@ -20,6 +20,20 @@ class IamResponse(BaseResponse):
template = self.response_template(GENERIC_EMPTY_TEMPLATE)
return template.render(name="DetachRolePolicyResponse")
def attach_group_policy(self):
policy_arn = self._get_param('PolicyArn')
group_name = self._get_param('GroupName')
iam_backend.attach_group_policy(policy_arn, group_name)
template = self.response_template(ATTACH_GROUP_POLICY_TEMPLATE)
return template.render()
def detach_group_policy(self):
policy_arn = self._get_param('PolicyArn')
group_name = self._get_param('GroupName')
iam_backend.detach_group_policy(policy_arn, group_name)
template = self.response_template(DETACH_GROUP_POLICY_TEMPLATE)
return template.render()
def attach_user_policy(self):
policy_arn = self._get_param('PolicyArn')
user_name = self._get_param('UserName')
@ -54,6 +68,17 @@ class IamResponse(BaseResponse):
template = self.response_template(LIST_ATTACHED_ROLE_POLICIES_TEMPLATE)
return template.render(policies=policies, marker=marker)
def list_attached_group_policies(self):
marker = self._get_param('Marker')
max_items = self._get_int_param('MaxItems', 100)
path_prefix = self._get_param('PathPrefix', '/')
group_name = self._get_param('GroupName')
policies, marker = iam_backend.list_attached_group_policies(
group_name, marker=marker, max_items=max_items,
path_prefix=path_prefix)
template = self.response_template(LIST_ATTACHED_GROUP_POLICIES_TEMPLATE)
return template.render(policies=policies, marker=marker)
def list_attached_user_policies(self):
marker = self._get_param('Marker')
max_items = self._get_int_param('MaxItems', 100)
@ -520,6 +545,18 @@ DETACH_USER_POLICY_TEMPLATE = """<DetachUserPolicyResponse>
</ResponseMetadata>
</DetachUserPolicyResponse>"""
ATTACH_GROUP_POLICY_TEMPLATE = """<AttachGroupPolicyResponse>
<ResponseMetadata>
<RequestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</RequestId>
</ResponseMetadata>
</AttachGroupPolicyResponse>"""
DETACH_GROUP_POLICY_TEMPLATE = """<DetachGroupPolicyResponse>
<ResponseMetadata>
<RequestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</RequestId>
</ResponseMetadata>
</DetachGroupPolicyResponse>"""
CREATE_POLICY_TEMPLATE = """<CreatePolicyResponse>
<CreatePolicyResult>
<Policy>
@ -560,6 +597,28 @@ LIST_ATTACHED_ROLE_POLICIES_TEMPLATE = """<ListAttachedRolePoliciesResponse>
</ResponseMetadata>
</ListAttachedRolePoliciesResponse>"""
LIST_ATTACHED_GROUP_POLICIES_TEMPLATE = """<ListAttachedGroupPoliciesResponse>
<ListAttachedGroupPoliciesResult>
{% if marker is none %}
<IsTruncated>false</IsTruncated>
{% else %}
<IsTruncated>true</IsTruncated>
<Marker>{{ marker }}</Marker>
{% endif %}
<AttachedPolicies>
{% for policy in policies %}
<member>
<PolicyName>{{ policy.name }}</PolicyName>
<PolicyArn>{{ policy.arn }}</PolicyArn>
</member>
{% endfor %}
</AttachedPolicies>
</ListAttachedGroupPoliciesResult>
<ResponseMetadata>
<RequestId>7a62c49f-347e-4fc4-9331-6e8eEXAMPLE</RequestId>
</ResponseMetadata>
</ListAttachedGroupPoliciesResponse>"""
LIST_ATTACHED_USER_POLICIES_TEMPLATE = """<ListAttachedUserPoliciesResponse>
<ListAttachedUserPoliciesResult>
{% if marker is none %}

27
tests/test_iam/test_iam_groups.py
View File

@ -82,6 +82,26 @@ def test_put_group_policy():
conn.put_group_policy('my-group', 'my-policy', '{"some": "json"}')
@mock_iam
def test_attach_group_policies():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_group(GroupName='my-group')
conn.list_attached_group_policies(GroupName='my-group')['AttachedPolicies'].should.be.empty
policy_arn = 'arn:aws:iam::aws:policy/service-role/AmazonElasticMapReduceforEC2Role'
conn.list_attached_group_policies(GroupName='my-group')['AttachedPolicies'].should.be.empty
conn.attach_group_policy(GroupName='my-group', PolicyArn=policy_arn)
conn.list_attached_group_policies(GroupName='my-group')['AttachedPolicies'].should.equal(
[
{
'PolicyName': 'AmazonElasticMapReduceforEC2Role',
'PolicyArn': policy_arn,
}
])
conn.detach_group_policy(GroupName='my-group', PolicyArn=policy_arn)
conn.list_attached_group_policies(GroupName='my-group')['AttachedPolicies'].should.be.empty
@mock_iam_deprecated()
def test_get_group_policy():
conn = boto.connect_iam()
@ -90,7 +110,8 @@ def test_get_group_policy():
conn.get_group_policy('my-group', 'my-policy')
conn.put_group_policy('my-group', 'my-policy', '{"some": "json"}')
policy = conn.get_group_policy('my-group', 'my-policy')
conn.get_group_policy('my-group', 'my-policy')
@mock_iam_deprecated()
def test_get_all_group_policies():
@ -107,6 +128,6 @@ def test_get_all_group_policies():
def test_list_group_policies():
conn = boto3.client('iam', region_name='us-east-1')
conn.create_group(GroupName='my-group')
policies = conn.list_group_policies(GroupName='my-group')['PolicyNames'].should.be.empty
conn.list_group_policies(GroupName='my-group')['PolicyNames'].should.be.empty
conn.put_group_policy(GroupName='my-group', PolicyName='my-policy', PolicyDocument='{"some": "json"}')
policies = conn.list_group_policies(GroupName='my-group')['PolicyNames'].should.equal(['my-policy'])
conn.list_group_policies(GroupName='my-group')['PolicyNames'].should.equal(['my-policy'])