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

Organizations - implement AWS Service Access functionality (#3122)

Browse Source 
* Add organizations.enable_aws_service_access

* Add organizations.list_aws_service_access_for_organization

* Add organizations.disable_aws_service_access
This commit is contained in:
Anton Grübel 2020-07-14 11:27:39 +02:00 committed by GitHub
parent 55bb4eb08d
commit f31f8e08c1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 215 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
moto/organizations/exceptions.py
View File

@ -5,11 +5,8 @@ from moto.core.exceptions import JsonRESTError
class InvalidInputException(JsonRESTError):
code = 400
def __init__(self):
super(InvalidInputException, self).__init__(
"InvalidInputException",
"You provided a value that does not match the required pattern.",
)
def __init__(self, message):
super(InvalidInputException, self).__init__("InvalidInputException", message)
class DuplicateOrganizationalUnitException(JsonRESTError):

93
moto/organizations/models.py
View File

@ -173,12 +173,60 @@ class FakeServiceControlPolicy(BaseModel):
}
class FakeServiceAccess(BaseModel):
# List of trusted services, which support trusted access with Organizations
# https://docs.aws.amazon.com/organizations/latest/userguide/orgs_integrated-services-list.html
TRUSTED_SERVICES = [
"aws-artifact-account-sync.amazonaws.com",
"backup.amazonaws.com",
"member.org.stacksets.cloudformation.amazonaws.com",
"cloudtrail.amazonaws.com",
"compute-optimizer.amazonaws.com",
"config.amazonaws.com",
"config-multiaccountsetup.amazonaws.com",
"controltower.amazonaws.com",
"ds.amazonaws.com",
"fms.amazonaws.com",
"guardduty.amazonaws.com",
"access-analyzer.amazonaws.com",
"license-manager.amazonaws.com",
"license-manager.member-account.amazonaws.com.",
"macie.amazonaws.com",
"ram.amazonaws.com",
"servicecatalog.amazonaws.com",
"servicequotas.amazonaws.com",
"sso.amazonaws.com",
"ssm.amazonaws.com",
"tagpolicies.tag.amazonaws.com",
]
def __init__(self, **kwargs):
if not self.trusted_service(kwargs["ServicePrincipal"]):
raise InvalidInputException(
"You specified an unrecognized service principal."
)
self.service_principal = kwargs["ServicePrincipal"]
self.date_enabled = datetime.datetime.utcnow()
def describe(self):
return {
"ServicePrincipal": self.service_principal,
"DateEnabled": unix_time(self.date_enabled),
}
@staticmethod
def trusted_service(service_principal):
return service_principal in FakeServiceAccess.TRUSTED_SERVICES
class OrganizationsBackend(BaseBackend):
def __init__(self):
self.org = None
self.accounts = []
self.ou = []
self.policies = []
self.services = []
def create_organization(self, **kwargs):
self.org = FakeOrganization(kwargs["FeatureSet"])
@ -459,7 +507,9 @@ class OrganizationsBackend(BaseBackend):
account = next((a for a in self.accounts if a.id == kwargs["ResourceId"]), None)
if account is None:
raise InvalidInputException
raise InvalidInputException(
"You provided a value that does not match the required pattern."
)
new_tags = {tag["Key"]: tag["Value"] for tag in kwargs["Tags"]}
account.tags.update(new_tags)
@ -468,7 +518,9 @@ class OrganizationsBackend(BaseBackend):
account = next((a for a in self.accounts if a.id == kwargs["ResourceId"]), None)
if account is None:
raise InvalidInputException
raise InvalidInputException(
"You provided a value that does not match the required pattern."
)
tags = [{"Key": key, "Value": value} for key, value in account.tags.items()]
return dict(Tags=tags)
@ -477,10 +529,45 @@ class OrganizationsBackend(BaseBackend):
account = next((a for a in self.accounts if a.id == kwargs["ResourceId"]), None)
if account is None:
raise InvalidInputException
raise InvalidInputException(
"You provided a value that does not match the required pattern."
)
for key in kwargs["TagKeys"]:
account.tags.pop(key, None)
def enable_aws_service_access(self, **kwargs):
service = FakeServiceAccess(**kwargs)
# enabling an existing service results in no changes
if any(
service["ServicePrincipal"] == kwargs["ServicePrincipal"]
for service in self.services
):
return
self.services.append(service.describe())
def list_aws_service_access_for_organization(self):
return dict(EnabledServicePrincipals=self.services)
def disable_aws_service_access(self, **kwargs):
if not FakeServiceAccess.trusted_service(kwargs["ServicePrincipal"]):
raise InvalidInputException(
"You specified an unrecognized service principal."
)
service_principal = next(
(
service
for service in self.services
if service["ServicePrincipal"] == kwargs["ServicePrincipal"]
),
None,
)
if service_principal:
self.services.remove(service_principal)
organizations_backend = OrganizationsBackend()

15
moto/organizations/responses.py
View File

@ -139,3 +139,18 @@ class OrganizationsResponse(BaseResponse):
return json.dumps(
self.organizations_backend.untag_resource(**self.request_params)
)
def enable_aws_service_access(self):
return json.dumps(
self.organizations_backend.enable_aws_service_access(**self.request_params)
)
def list_aws_service_access_for_organization(self):
return json.dumps(
self.organizations_backend.list_aws_service_access_for_organization()
)
def disable_aws_service_access(self):
return json.dumps(
self.organizations_backend.disable_aws_service_access(**self.request_params)
)

108
tests/test_organizations/test_organizations_boto3.py
View File

@ -1,5 +1,7 @@
from __future__ import unicode_literals
from datetime import datetime
import boto3
import json
import six
@ -751,3 +753,109 @@ def test_update_organizational_unit_duplicate_error():
exc.response["Error"]["Message"].should.equal(
"An OU with the same name already exists."
)
@mock_organizations
def test_enable_aws_service_access():
# given
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
# when
client.enable_aws_service_access(ServicePrincipal="config.amazonaws.com")
# then
response = client.list_aws_service_access_for_organization()
response["EnabledServicePrincipals"].should.have.length_of(1)
service = response["EnabledServicePrincipals"][0]
service["ServicePrincipal"].should.equal("config.amazonaws.com")
date_enabled = service["DateEnabled"]
date_enabled["DateEnabled"].should.be.a(datetime)
# enabling the same service again should not result in any error or change
# when
client.enable_aws_service_access(ServicePrincipal="config.amazonaws.com")
# then
response = client.list_aws_service_access_for_organization()
response["EnabledServicePrincipals"].should.have.length_of(1)
service = response["EnabledServicePrincipals"][0]
service["ServicePrincipal"].should.equal("config.amazonaws.com")
service["DateEnabled"].should.equal(date_enabled)
@mock_organizations
def test_enable_aws_service_access():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
with assert_raises(ClientError) as e:
client.enable_aws_service_access(ServicePrincipal="moto.amazonaws.com")
ex = e.exception
ex.operation_name.should.equal("EnableAWSServiceAccess")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("InvalidInputException")
ex.response["Error"]["Message"].should.equal(
"You specified an unrecognized service principal."
)
@mock_organizations
def test_enable_aws_service_access():
# given
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
client.enable_aws_service_access(ServicePrincipal="config.amazonaws.com")
client.enable_aws_service_access(ServicePrincipal="ram.amazonaws.com")
# when
response = client.list_aws_service_access_for_organization()
# then
response["EnabledServicePrincipals"].should.have.length_of(2)
services = sorted(
response["EnabledServicePrincipals"], key=lambda i: i["ServicePrincipal"]
)
services[0]["ServicePrincipal"].should.equal("config.amazonaws.com")
services[0]["DateEnabled"].should.be.a(datetime)
services[1]["ServicePrincipal"].should.equal("ram.amazonaws.com")
services[1]["DateEnabled"].should.be.a(datetime)
@mock_organizations
def test_disable_aws_service_access():
# given
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
client.enable_aws_service_access(ServicePrincipal="config.amazonaws.com")
# when
client.disable_aws_service_access(ServicePrincipal="config.amazonaws.com")
# then
response = client.list_aws_service_access_for_organization()
response["EnabledServicePrincipals"].should.have.length_of(0)
# disabling the same service again should not result in any error
# when
client.disable_aws_service_access(ServicePrincipal="config.amazonaws.com")
# then
response = client.list_aws_service_access_for_organization()
response["EnabledServicePrincipals"].should.have.length_of(0)
@mock_organizations
def test_disable_aws_service_access_errors():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")
with assert_raises(ClientError) as e:
client.disable_aws_service_access(ServicePrincipal="moto.amazonaws.com")
ex = e.exception
ex.operation_name.should.equal("DisableAWSServiceAccess")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("InvalidInputException")
ex.response["Error"]["Message"].should.equal(
"You specified an unrecognized service principal."
)