catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

[Resolves #2196] - endpoints for querying organizations SC policies (#2197)

Browse source 
adding support for organizations service control policies

* [Resolves #2196] - endpoints for querying organizations SC policies

I have added the following mock endpoints to the Organizations service:

- create_policy
- list_policies
- describe_policy
- attach_policy
- list_policies_for_target
- list_targets_for_policy
This commit is contained in:
Ashley Gould 2019-05-25 02:20:19 -07:00 • committed by Terry Cain
commit a3f6d2c110
6 changed files with 508 additions and 26 deletions
Download patch file Download diff file Expand all files Collapse all files

147
moto/organizations/models.py
View file

@ -47,6 +47,7 @@ class FakeOrganization(BaseModel):
class FakeAccount(BaseModel):
def __init__(self, organization, **kwargs):
self.type = 'ACCOUNT'
self.organization_id = organization.id
self.master_account_id = organization.master_account_id
self.create_account_status_id = utils.make_random_create_account_status_id()
@ -57,6 +58,7 @@ class FakeAccount(BaseModel):
self.status = 'ACTIVE'
self.joined_method = 'CREATED'
self.parent_id = organization.root_id
self.attached_policies = []
@property
def arn(self):
@ -103,6 +105,7 @@ class FakeOrganizationalUnit(BaseModel):
self.name = kwargs.get('Name')
self.parent_id = kwargs.get('ParentId')
self._arn_format = utils.OU_ARN_FORMAT
self.attached_policies = []
@property
def arn(self):
@ -134,6 +137,7 @@ class FakeRoot(FakeOrganizationalUnit):
'Status': 'ENABLED'
}]
self._arn_format = utils.ROOT_ARN_FORMAT
self.attached_policies = []
def describe(self):
return {
@ -144,12 +148,52 @@ class FakeRoot(FakeOrganizationalUnit):
}
class FakeServiceControlPolicy(BaseModel):
def __init__(self, organization, **kwargs):
self.type = 'POLICY'
self.content = kwargs.get('Content')
self.description = kwargs.get('Description')
self.name = kwargs.get('Name')
self.type = kwargs.get('Type')
self.id = utils.make_random_service_control_policy_id()
self.aws_managed = False
self.organization_id = organization.id
self.master_account_id = organization.master_account_id
self._arn_format = utils.SCP_ARN_FORMAT
self.attachments = []
@property
def arn(self):
return self._arn_format.format(
self.master_account_id,
self.organization_id,
self.id
)
def describe(self):
return {
'Policy': {
'PolicySummary': {
'Id': self.id,
'Arn': self.arn,
'Name': self.name,
'Description': self.description,
'Type': self.type,
'AwsManaged': self.aws_managed,
},
'Content': self.content
}
}
class OrganizationsBackend(BaseBackend):
def __init__(self):
self.org = None
self.accounts = []
self.ou = []
self.policies = []
def create_organization(self, **kwargs):
self.org = FakeOrganization(kwargs['FeatureSet'])
@ -292,5 +336,108 @@ class OrganizationsBackend(BaseBackend):
]
)
def create_policy(self, **kwargs):
new_policy = FakeServiceControlPolicy(self.org, **kwargs)
self.policies.append(new_policy)
return new_policy.describe()
def describe_policy(self, **kwargs):
if re.compile(utils.SCP_ID_REGEX).match(kwargs['PolicyId']):
policy = next((p for p in self.policies if p.id == kwargs['PolicyId']), None)
if policy is None:
raise RESTError(
'PolicyNotFoundException',
"You specified a policy that doesn't exist."
)
else:
raise RESTError(
'InvalidInputException',
'You specified an invalid value.'
)
return policy.describe()
def attach_policy(self, **kwargs):
policy = next((p for p in self.policies if p.id == kwargs['PolicyId']), None)
if (re.compile(utils.ROOT_ID_REGEX).match(kwargs['TargetId']) or
re.compile(utils.OU_ID_REGEX).match(kwargs['TargetId'])):
ou = next((ou for ou in self.ou if ou.id == kwargs['TargetId']), None)
if ou is not None:
if ou not in ou.attached_policies:
ou.attached_policies.append(policy)
policy.attachments.append(ou)
else:
raise RESTError(
'OrganizationalUnitNotFoundException',
"You specified an organizational unit that doesn't exist."
)
elif re.compile(utils.ACCOUNT_ID_REGEX).match(kwargs['TargetId']):
account = next((a for a in self.accounts if a.id == kwargs['TargetId']), None)
if account is not None:
if account not in account.attached_policies:
account.attached_policies.append(policy)
policy.attachments.append(account)
else:
raise RESTError(
'AccountNotFoundException',
"You specified an account that doesn't exist."
)
else:
raise RESTError(
'InvalidInputException',
'You specified an invalid value.'
)
def list_policies(self, **kwargs):
return dict(Policies=[
p.describe()['Policy']['PolicySummary'] for p in self.policies
])
def list_policies_for_target(self, **kwargs):
if re.compile(utils.OU_ID_REGEX).match(kwargs['TargetId']):
obj = next((ou for ou in self.ou if ou.id == kwargs['TargetId']), None)
if obj is None:
raise RESTError(
'OrganizationalUnitNotFoundException',
"You specified an organizational unit that doesn't exist."
)
elif re.compile(utils.ACCOUNT_ID_REGEX).match(kwargs['TargetId']):
obj = next((a for a in self.accounts if a.id == kwargs['TargetId']), None)
if obj is None:
raise RESTError(
'AccountNotFoundException',
"You specified an account that doesn't exist."
)
else:
raise RESTError(
'InvalidInputException',
'You specified an invalid value.'
)
return dict(Policies=[
p.describe()['Policy']['PolicySummary'] for p in obj.attached_policies
])
def list_targets_for_policy(self, **kwargs):
if re.compile(utils.SCP_ID_REGEX).match(kwargs['PolicyId']):
policy = next((p for p in self.policies if p.id == kwargs['PolicyId']), None)
if policy is None:
raise RESTError(
'PolicyNotFoundException',
"You specified a policy that doesn't exist."
)
else:
raise RESTError(
'InvalidInputException',
'You specified an invalid value.'
)
objects = [
{
'TargetId': obj.id,
'Arn': obj.arn,
'Name': obj.name,
'Type': obj.type,
} for obj in policy.attachments
]
return dict(Targets=objects)
organizations_backend = OrganizationsBackend()