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

address PR comments

Browse Source
This commit is contained in:
Nick Stocchero 2020-08-02 21:16:44 -06:00
parent 7bc5b5c08f
commit ff84b63484
4 changed files with 231 additions and 89 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
CONFIG_README.md
View File

@ -53,15 +53,14 @@ An example of the above is implemented for S3. You can see that by looking at:
1. `moto/s3/config.py`
1. `moto/config/models.py`
As well as the corresponding unit tests in:
### Testing
For each resource type, you will need to test write tests for a few separate areas:
1. `tests/s3/test_s3.py`
1. `tests/config/test_config.py`
- Test the backend queries to ensure discovered resources come back (ie for `IAM::Policy`, write `tests.tests_iam.test_policy_list_config_discovered_resources`). For writing these tests, you must not make use of `boto` to create resources. You will need to use the backend model methods to provision the resources. This is to make tests compatible with the moto server. You must make tests for the resource type to test listing and object fetching.
Note for unit testing, you will want to add a test to ensure that you can query all the resources effectively. For testing this feature,
the unit tests for the `ConfigQueryModel` will not make use of `boto` to create resources, such as S3 buckets. You will need to use the
backend model methods to provision the resources. This is to make tests compatible with the moto server. You should absolutely make tests
in the resource type to test listing and object fetching.
- Test the config dict for all scenarios (ie for `IAM::Policy`, write `tests.tests_iam.test_policy_config_dict`). For writing this test, you'll need to create resources in the same way as the first test (without using `boto`), in every meaningful configuration that would produce a different config dict. Then, query the backend and ensure each of the dicts are as you expect.
- Test that everything works end to end with the `boto` clients. (ie for `IAM::Policy`, write `tests.tests_iam.test_policy_config_client`). The main two items to test will be the `boto.client('config').list_discovered_resources()`, `boto.client('config').list_aggregate_discovered_resources()`, `moto.client('config').batch_get_resource_config()`, and `moto.client('config').batch_aggregate_get_resource_config()`. This test doesn't have to be super thorough, but it basically tests that the front end and backend logic all works together and returns correct resources. Beware the aggregate methods all have capital first letters (ie `Limit`), while non-aggregate methods have lowercase first letters (ie `limit`)
### Listing
S3 is currently the model implementation, but it also odd in that S3 is a global resource type with regional resource residency.

24
moto/core/models.py
View File

@ -27,8 +27,8 @@ from .utils import (
convert_flask_to_responses_response,
)
ACCOUNT_ID = os.environ.get("MOTO_ACCOUNT_ID", "123456789012")
CONFIG_BACKEND_DELIM = "\x1e" # Record Seperator "RS" ASCII Character
class BaseMockAWS(object):
@ -768,23 +768,37 @@ class ConfigQueryModel(object):
def aggregate_regions(self, path, backend_region, resource_region):
"""
Returns a list of "region\1eresourcename" strings
This method will is called for both aggregated and non-aggregated calls for config resources.
It will figure out how to return the full list of resources for a given regional backend and append them to a final list.
It produces a list of both the region and the resource name with a delimiter character (CONFIG_BACKEND_DELIM, ASCII Record separator, \x1e).
IE: "us-east-1\x1ei-1234567800"
Each config-enabled resource has a method named `list_config_service_resources` which has to parse the delimiter
...
:param path: - A dict accessor string applied to the backend that locates the resource.
:param backend_region:
:param resource_region:
:return: - Returns a list of "region\x1eresourcename" strings
"""
filter_region = backend_region or resource_region
if filter_region:
filter_resources = list(self.backends[filter_region].__dict__[path].keys())
return map(
lambda resource: "{}\1e{}".format(filter_region, resource),
return list(
map(
lambda resource: "{}{}{}".format(
filter_region, CONFIG_BACKEND_DELIM, resource
),
filter_resources,
)
)
# If we don't have a filter region
ret = []
for region in self.backends:
this_region_resources = list(self.backends[region].__dict__[path].keys())
for resource in this_region_resources:
ret.append("{}\1e{}".format(region, resource))
ret.append("{}{}{}".format(region, CONFIG_BACKEND_DELIM, resource))
return ret

61
moto/iam/config.py
View File

@ -4,6 +4,8 @@ from moto.core.exceptions import InvalidNextTokenException
from moto.core.models import ConfigQueryModel
from moto.iam import iam_backends
CONFIG_BACKEND_DELIM = "\x1e" # Record Seperator "RS" ASCII Character
class RoleConfigQuery(ConfigQueryModel):
def list_config_service_resources(
@ -15,18 +17,17 @@ class RoleConfigQuery(ConfigQueryModel):
backend_region=None,
resource_region=None,
):
# For aggregation -- did we get both a resource ID and a resource name?
if resource_ids and resource_name:
# If the values are different, then return an empty list:
if resource_name not in resource_ids:
return [], None
# IAM roles are "global" and aren't assigned into any availability zone
# The resource ID is a AWS-assigned random string like "AROA0BSVNSZKXVHS00SBJ"
# The resource name is a user-assigned string like "MyDevelopmentAdminRole"
role_list = self.aggregate_regions("roles", backend_region, resource_region)
# Grab roles from backend
role_list = self.aggregate_regions("roles", "global", None)
if not role_list:
return [], None
# Pagination logic:
# Pagination logic
sorted_roles = sorted(role_list)
new_token = None
@ -34,7 +35,7 @@ class RoleConfigQuery(ConfigQueryModel):
if not next_token:
start = 0
else:
# "Tokens" are region + \00 + resource ID.
# "Tokens" are region + \x1e + resource ID.
if next_token not in sorted_roles:
raise InvalidNextTokenException()
@ -46,13 +47,16 @@ class RoleConfigQuery(ConfigQueryModel):
if len(sorted_roles) > (start + limit):
new_token = sorted_roles[start + limit]
# Each element is a string of "region\x1eresource_id"
return (
[
{
"type": "AWS::IAM::Role",
"id": role.split("\1e")[1],
"name": role.split("\1e")[1],
"region": role.split("\1e")[0],
"id": role.split(CONFIG_BACKEND_DELIM)[1],
"name": self.backends["global"]
.roles[role.split(CONFIG_BACKEND_DELIM)[1]]
.name,
"region": role.split(CONFIG_BACKEND_DELIM)[0],
}
for role in role_list
],
@ -71,7 +75,7 @@ class RoleConfigQuery(ConfigQueryModel):
if resource_name and role.name != resource_name:
return
# Format the bucket to the AWS Config format:
# Format the role to the AWS Config format:
config_data = role.to_config_dict()
# The 'configuration' field is also a JSON string:
@ -95,16 +99,19 @@ class PolicyConfigQuery(ConfigQueryModel):
backend_region=None,
resource_region=None,
):
# For aggregation -- did we get both a resource ID and a resource name?
if resource_ids and resource_name:
# If the values are different, then return an empty list:
if resource_name not in resource_ids:
return [], None
# IAM policies are "global" and aren't assigned into any availability zone
# The resource ID is a AWS-assigned random string like "ANPA0BSVNSZK00SJSPVUJ"
# The resource name is a user-assigned string like "my-development-policy"
# We don't want to include AWS Managed Policies
# We don't want to include AWS Managed Policies. This technically needs to
# respect the configuration recorder's 'includeGlobalResourceTypes' setting,
# but it's default set be default, and moto's config doesn't yet support
# custom configuration recorders, we'll just behave as default.
policy_list = filter(
lambda policy: not policy.split("\1e")[1].startswith("arn:aws:iam::aws"),
self.aggregate_regions("managed_policies", backend_region, resource_region),
lambda policy: not policy.split(CONFIG_BACKEND_DELIM)[1].startswith(
"arn:aws:iam::aws"
),
self.aggregate_regions("managed_policies", "global", None),
)
if not policy_list:
@ -118,7 +125,7 @@ class PolicyConfigQuery(ConfigQueryModel):
if not next_token:
start = 0
else:
# "Tokens" are region + \00 + resource ID.
# "Tokens" are region + \x1e + resource ID.
if next_token not in sorted_policies:
raise InvalidNextTokenException()
@ -134,9 +141,13 @@ class PolicyConfigQuery(ConfigQueryModel):
[
{
"type": "AWS::IAM::Policy",
"id": policy.split("\1e")[1],
"name": policy.split("\1e")[1],
"region": policy.split("\1e")[0],
"id": self.backends["global"]
.managed_policies[policy.split(CONFIG_BACKEND_DELIM)[1]]
.id,
"name": self.backends["global"]
.managed_policies[policy.split(CONFIG_BACKEND_DELIM)[1]]
.name,
"region": policy.split(CONFIG_BACKEND_DELIM)[0],
}
for policy in policy_list
],
@ -155,7 +166,7 @@ class PolicyConfigQuery(ConfigQueryModel):
if resource_name and policy.name != resource_name:
return
# Format the bucket to the AWS Config format:
# Format the policy to the AWS Config format:
config_data = policy.to_config_dict()
# The 'configuration' field is also a JSON string:

202
tests/test_iam/test_iam.py
View File

@ -9,7 +9,7 @@ import sure # noqa
from boto.exception import BotoServerError
from botocore.exceptions import ClientError
from moto import mock_iam, mock_iam_deprecated, settings
from moto import mock_config, mock_iam, mock_iam_deprecated, settings
from moto.core import ACCOUNT_ID
from moto.iam.models import aws_managed_policies
from moto.backends import get_backend
@ -2923,48 +2923,14 @@ def test_role_list_config_discovered_resources():
role = result[0]
assert role["type"] == "AWS::IAM::Role"
assert len(role["id"]) == len(random_resource_id())
assert role["id"] == role["name"]
assert role["name"] == "something"
assert role["region"] == "global"
@mock_iam
def test_policy_list_config_discovered_resources():
from moto.iam.config import policy_config_query
# Without any policies
assert policy_config_query.list_config_service_resources(None, None, 100, None) == (
[],
None,
)
basic_policy = {
"Version": "2012-10-17",
"Statement": [
{"Action": ["ec2:DeleteKeyPair"], "Effect": "Deny", "Resource": "*"}
],
}
# Create a role
policy_config_query.backends["global"].create_policy(
description="mypolicy",
path="",
policy_document=json.dumps(basic_policy),
policy_name="mypolicy",
)
result = policy_config_query.list_config_service_resources(None, None, 100, None)[0]
assert len(result) == 1
policy = result[0]
assert policy["type"] == "AWS::IAM::Policy"
assert policy["id"] == policy["name"] == "arn:aws:iam::123456789012:policy/mypolicy"
assert policy["region"] == "global"
@mock_iam
def test_role_config_dict():
from moto.iam.config import role_config_query, policy_config_query
from moto.iam.utils import random_resource_id
from moto.iam.utils import random_resource_id, random_policy_id
# Without any roles
assert not role_config_query.get_config_resource("something")
@ -2986,17 +2952,21 @@ def test_role_config_dict():
}
# Create a policy for use in role permissions boundary
policy_config_query.backends["global"].create_policy(
policy_arn = (
policy_config_query.backends["global"]
.create_policy(
description="basic_policy",
path="/",
policy_document=json.dumps(basic_policy),
policy_name="basic_policy",
)
.arn
)
policy_arn = policy_config_query.list_config_service_resources(
policy_id = policy_config_query.list_config_service_resources(
None, None, 100, None
)[0][0]["id"]
assert policy_arn is not None
assert len(policy_id) == len(random_policy_id())
# Create some roles (and grab them repeatedly since they create with random names)
role_config_query.backends["global"].create_role(
@ -3225,6 +3195,141 @@ def test_role_config_dict():
]
@mock_iam
@mock_config
def test_role_config_client():
from moto.iam.models import ACCOUNT_ID
from moto.iam.utils import random_resource_id
iam_client = boto3.client("iam", region_name="us-west-2")
config_client = boto3.client("config", region_name="us-west-2")
account_aggregation_source = {
"AccountIds": [ACCOUNT_ID],
"AllAwsRegions": True,
}
config_client.put_configuration_aggregator(
ConfigurationAggregatorName="test_aggregator",
AccountAggregationSources=[account_aggregation_source],
)
result = config_client.list_discovered_resources(resourceType="AWS::IAM::Role")
assert not result["resourceIdentifiers"]
role_id = iam_client.create_role(
Path="/",
RoleName="mytestrole",
Description="mytestrole",
AssumeRolePolicyDocument=json.dumps("{ }"),
)["Role"]["RoleId"]
iam_client.create_role(
Path="/",
RoleName="mytestrole2",
Description="zmytestrole",
AssumeRolePolicyDocument=json.dumps("{ }"),
)
# Test non-aggregated query: (everything is getting a random id, so we can't test names by ordering)
result = config_client.list_discovered_resources(
resourceType="AWS::IAM::Role", limit=1
)
first_result = result["resourceIdentifiers"][0]["resourceId"]
assert result["resourceIdentifiers"][0]["resourceType"] == "AWS::IAM::Role"
assert len(first_result) == len(random_resource_id())
# Test non-aggregated pagination
assert (
config_client.list_discovered_resources(
resourceType="AWS::IAM::Role", limit=1, nextToken=result["nextToken"]
)["resourceIdentifiers"][0]["resourceId"]
) != first_result
# Test aggregated query: (everything is getting a random id, so we can't test names by ordering)
agg_result = config_client.list_aggregate_discovered_resources(
ResourceType="AWS::IAM::Role",
ConfigurationAggregatorName="test_aggregator",
Limit=1,
)
first_agg_result = agg_result["ResourceIdentifiers"][0]["ResourceId"]
assert agg_result["ResourceIdentifiers"][0]["ResourceType"] == "AWS::IAM::Role"
assert len(first_agg_result) == len(random_resource_id())
assert agg_result["ResourceIdentifiers"][0]["SourceAccountId"] == ACCOUNT_ID
assert agg_result["ResourceIdentifiers"][0]["SourceRegion"] == "global"
# Test aggregated pagination
assert (
config_client.list_aggregate_discovered_resources(
ConfigurationAggregatorName="test_aggregator",
ResourceType="AWS::IAM::Role",
Limit=1,
NextToken=agg_result["NextToken"],
)["ResourceIdentifiers"][0]["ResourceId"]
!= first_agg_result
)
# Test non-aggregated batch get
assert (
config_client.batch_get_resource_config(
resourceKeys=[{"resourceType": "AWS::IAM::Role", "resourceId": role_id}]
)["baseConfigurationItems"][0]["resourceName"]
== "mytestrole"
)
# Test aggregated batch get
assert (
config_client.batch_get_aggregate_resource_config(
ConfigurationAggregatorName="test_aggregator",
ResourceIdentifiers=[
{
"SourceAccountId": ACCOUNT_ID,
"SourceRegion": "global",
"ResourceId": role_id,
"ResourceType": "AWS::IAM::Role",
}
],
)["BaseConfigurationItems"][0]["resourceName"]
== "mytestrole"
)
@mock_iam
def test_policy_list_config_discovered_resources():
from moto.iam.config import policy_config_query
from moto.iam.utils import random_policy_id
# Without any policies
assert policy_config_query.list_config_service_resources(None, None, 100, None) == (
[],
None,
)
basic_policy = {
"Version": "2012-10-17",
"Statement": [
{"Action": ["ec2:DeleteKeyPair"], "Effect": "Deny", "Resource": "*"}
],
}
# Create a role
policy_config_query.backends["global"].create_policy(
description="mypolicy",
path="",
policy_document=json.dumps(basic_policy),
policy_name="mypolicy",
)
result = policy_config_query.list_config_service_resources(None, None, 100, None)[0]
assert len(result) == 1
policy = result[0]
assert policy["type"] == "AWS::IAM::Policy"
assert len(policy["id"]) == len(random_policy_id())
assert policy["name"] == "mypolicy"
assert policy["region"] == "global"
@mock_iam
def test_policy_config_dict():
from moto.iam.config import role_config_query, policy_config_query
@ -3251,17 +3356,24 @@ def test_policy_config_dict():
],
}
policy_config_query.backends["global"].create_policy(
policy_arn = (
policy_config_query.backends["global"]
.create_policy(
description="basic_policy",
path="/",
policy_document=json.dumps(basic_policy),
policy_name="basic_policy",
)
.arn
)
policy_arn = policy_config_query.list_config_service_resources(
policy_id = policy_config_query.list_config_service_resources(
None, None, 100, None
)[0][0]["id"]
assert len(policy_id) == len(random_policy_id())
assert policy_arn == "arn:aws:iam::123456789012:policy/basic_policy"
assert (
policy_config_query.get_config_resource(
"arn:aws:iam::123456789012:policy/basic_policy"
@ -3330,3 +3442,9 @@ def test_policy_config_dict():
},
]
assert policy["supplementaryConfiguration"] == {}
@mock_iam
@mock_config
def test_policy_config_client():
assert 1 == 1