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

Organizations - implement Policy Type functionality (#3207)

Browse source 
* Add organizations.enable_policy_type

* Add organizations.disable_policy_type

* Add support for AISERVICES_OPT_OUT_POLICY
This commit is contained in:
Anton Grübel 2020-08-02 11:56:19 +02:00 • committed by GitHub
commit 252d679b27
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 419 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

106
moto/organizations/models.py
View file

@ -17,6 +17,10 @@ from moto.organizations.exceptions import (
AccountAlreadyRegisteredException,
AWSOrganizationsNotInUseException,
AccountNotRegisteredException,
RootNotFoundException,
PolicyTypeAlreadyEnabledException,
PolicyTypeNotEnabledException,
TargetNotFoundException,
)
@ -124,6 +128,13 @@ class FakeOrganizationalUnit(BaseModel):
class FakeRoot(FakeOrganizationalUnit):
SUPPORTED_POLICY_TYPES = [
"AISERVICES_OPT_OUT_POLICY",
"BACKUP_POLICY",
"SERVICE_CONTROL_POLICY",
"TAG_POLICY",
]
def __init__(self, organization, **kwargs):
super(FakeRoot, self).__init__(organization, **kwargs)
self.type = "ROOT"
@ -141,20 +152,55 @@ class FakeRoot(FakeOrganizationalUnit):
"PolicyTypes": self.policy_types,
}
def add_policy_type(self, policy_type):
if policy_type not in self.SUPPORTED_POLICY_TYPES:
raise InvalidInputException("You specified an invalid value.")
if any(type["Type"] == policy_type for type in self.policy_types):
raise PolicyTypeAlreadyEnabledException
self.policy_types.append({"Type": policy_type, "Status": "ENABLED"})
def remove_policy_type(self, policy_type):
if not FakePolicy.supported_policy_type(policy_type):
raise InvalidInputException("You specified an invalid value.")
if all(type["Type"] != policy_type for type in self.policy_types):
raise PolicyTypeNotEnabledException
self.policy_types.remove({"Type": policy_type, "Status": "ENABLED"})
class FakePolicy(BaseModel):
SUPPORTED_POLICY_TYPES = [
"AISERVICES_OPT_OUT_POLICY",
"BACKUP_POLICY",
"SERVICE_CONTROL_POLICY",
"TAG_POLICY",
]
class FakeServiceControlPolicy(BaseModel):
def __init__(self, organization, **kwargs):
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.id = utils.make_random_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 = []
if not FakePolicy.supported_policy_type(self.type):
raise InvalidInputException("You specified an invalid value.")
elif self.type == "AISERVICES_OPT_OUT_POLICY":
self._arn_format = utils.AI_POLICY_ARN_FORMAT
elif self.type == "SERVICE_CONTROL_POLICY":
self._arn_format = utils.SCP_ARN_FORMAT
else:
raise NotImplementedError(
"The {0} policy type has not been implemented".format(self.type)
)
@property
def arn(self):
return self._arn_format.format(
@ -176,6 +222,10 @@ class FakeServiceControlPolicy(BaseModel):
}
}
@staticmethod
def supported_policy_type(policy_type):
return policy_type in FakePolicy.SUPPORTED_POLICY_TYPES
class FakeServiceAccess(BaseModel):
# List of trusted services, which support trusted access with Organizations
@ -283,6 +333,13 @@ class OrganizationsBackend(BaseBackend):
self.services = []
self.admins = []
def _get_root_by_id(self, root_id):
root = next((ou for ou in self.ou if ou.id == root_id), None)
if not root:
raise RootNotFoundException
return root
def create_organization(self, **kwargs):
self.org = FakeOrganization(kwargs["FeatureSet"])
root_ou = FakeRoot(self.org)
@ -292,7 +349,7 @@ class OrganizationsBackend(BaseBackend):
)
master_account.id = self.org.master_account_id
self.accounts.append(master_account)
default_policy = FakeServiceControlPolicy(
default_policy = FakePolicy(
self.org,
Name="FullAWSAccess",
Description="Allows access to every operation",
@ -452,7 +509,7 @@ class OrganizationsBackend(BaseBackend):
)
def create_policy(self, **kwargs):
new_policy = FakeServiceControlPolicy(self.org, **kwargs)
new_policy = FakePolicy(self.org, **kwargs)
for policy in self.policies:
if kwargs["Name"] == policy.name:
raise DuplicatePolicyException
@ -460,7 +517,7 @@ class OrganizationsBackend(BaseBackend):
return new_policy.describe()
def describe_policy(self, **kwargs):
if re.compile(utils.SCP_ID_REGEX).match(kwargs["PolicyId"]):
if re.compile(utils.POLICY_ID_REGEX).match(kwargs["PolicyId"]):
policy = next(
(p for p in self.policies if p.id == kwargs["PolicyId"]), None
)
@ -540,7 +597,13 @@ class OrganizationsBackend(BaseBackend):
)
def list_policies_for_target(self, **kwargs):
if re.compile(utils.OU_ID_REGEX).match(kwargs["TargetId"]):
filter = kwargs["Filter"]
if re.match(utils.ROOT_ID_REGEX, kwargs["TargetId"]):
obj = next((ou for ou in self.ou if ou.id == kwargs["TargetId"]), None)
if obj is None:
raise TargetNotFoundException
elif 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(
@ -553,14 +616,25 @@ class OrganizationsBackend(BaseBackend):
raise AccountNotFoundException
else:
raise InvalidInputException("You specified an invalid value.")
if not FakePolicy.supported_policy_type(filter):
raise InvalidInputException("You specified an invalid value.")
if filter not in ["AISERVICES_OPT_OUT_POLICY", "SERVICE_CONTROL_POLICY"]:
raise NotImplementedError(
"The {0} policy type has not been implemented".format(filter)
)
return dict(
Policies=[
p.describe()["Policy"]["PolicySummary"] for p in obj.attached_policies
p.describe()["Policy"]["PolicySummary"]
for p in obj.attached_policies
if p.type == filter
]
)
def list_targets_for_policy(self, **kwargs):
if re.compile(utils.SCP_ID_REGEX).match(kwargs["PolicyId"]):
if re.compile(utils.POLICY_ID_REGEX).match(kwargs["PolicyId"]):
policy = next(
(p for p in self.policies if p.id == kwargs["PolicyId"]), None
)
@ -733,5 +807,19 @@ class OrganizationsBackend(BaseBackend):
if not admin.services:
self.admins.remove(admin)
def enable_policy_type(self, **kwargs):
root = self._get_root_by_id(kwargs["RootId"])
root.add_policy_type(kwargs["PolicyType"])
return dict(Root=root.describe())
def disable_policy_type(self, **kwargs):
root = self._get_root_by_id(kwargs["RootId"])
root.remove_policy_type(kwargs["PolicyType"])
return dict(Root=root.describe())
organizations_backend = OrganizationsBackend()