moto/tests/test_config/test_config_rules.py

"""Unit tests specific to the ConfigService ConfigRule APIs.

 These APIs include:
   put_config_rule
   describe_config_rule
   delete_config_rule
"""
import json
from string import ascii_lowercase

import boto3
from botocore.exceptions import ClientError
import pytest

from moto.config import mock_config
from moto.config.models import random_string
from moto.config.models import ConfigRule, CONFIG_RULE_PAGE_SIZE
from moto import settings

TEST_REGION = "us-east-1" if settings.TEST_SERVER_MODE else "us-west-2"


def managed_config_rule():
    """Return a valid managed AWS Config Rule."""
    return {
        "ConfigRuleName": f"managed_rule_{random_string()}",
        "Description": "Managed S3 Public Read Prohibited Bucket Rule",
        "Scope": {"ComplianceResourceTypes": ["AWS::S3::Bucket", "AWS::IAM::Group"]},
        "Source": {
            "Owner": "AWS",
            "SourceIdentifier": "S3_BUCKET_PUBLIC_READ_PROHIBITED",
        },
        "MaximumExecutionFrequency": "One_Hour",
    }


@mock_config
def test_put_config_rule_errors():
    """Test various error conditions in put_config_rule API call."""
    client = boto3.client("config", region_name=TEST_REGION)

    rule_name_base = "cf_limit_test"
    for idx in range(ConfigRule.MAX_RULES):
        managed_rule = managed_config_rule()
        managed_rule["ConfigRuleName"] = f"{rule_name_base}_{idx}"
        client.put_config_rule(ConfigRule=managed_rule)

    with pytest.raises(ClientError) as exc:
        managed_rule = managed_config_rule()
        managed_rule["ConfigRuleName"] = f"{rule_name_base}_{ConfigRule.MAX_RULES}"
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "MaxNumberOfConfigRulesExceededException"
    assert "maximum number of config rules" in err["Message"]

    # Free up the memory from the limits test.
    for idx in range(ConfigRule.MAX_RULES):
        client.delete_config_rule(ConfigRuleName=f"{rule_name_base}_{idx}")

    # Rule name that exceeds 128 chars in length.
    rule_name = ascii_lowercase * 5
    managed_rule = managed_config_rule()
    managed_rule["ConfigRuleName"] = rule_name
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "ValidationException"
    assert "Member must have length less than or equal to 128" in err["Message"]


@mock_config
def test_put_config_rule_update_errors():
    """Test various error conditions when updating ConfigRule."""
    client = boto3.client("config", region_name=TEST_REGION)

    # No name, arn or id.
    managed_rule = {
        "Description": "Managed S3 Public Read Prohibited Bucket Rule",
        "Scope": {"ComplianceResourceTypes": ["AWS::S3::Bucket"]},
        "Source": {
            "Owner": "AWS",
            "SourceIdentifier": "S3_BUCKET_PUBLIC_READ_PROHIBITED",
        },
    }
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        "One or more identifiers needs to be provided. Provide Name or Id or Arn"
        in err["Message"]
    )

    # Provide an id for a rule that does not exist.
    managed_rule = {
        "ConfigRuleId": "foo",
        "Description": "Managed S3 Public Read Prohibited Bucket Rule",
        "Scope": {"ComplianceResourceTypes": ["AWS::S3::Bucket"]},
        "Source": {
            "Owner": "AWS",
            "SourceIdentifier": "S3_BUCKET_PUBLIC_READ_PROHIBITED",
        },
    }
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        "One or more identifiers needs to be provided. Provide Name or Id or Arn"
        in err["Message"]
    )


@mock_config
def test_config_rule_errors():  # pylint: disable=too-many-statements
    """Test various error conditions in ConfigRule instantiation."""
    client = boto3.client("config", region_name=TEST_REGION)

    # Missing fields (ParamValidationError) caught by botocore and not
    # tested here:  ConfigRule.Source, ConfigRule.ConfigRuleName

    managed_rule = managed_config_rule()
    managed_rule["ConfigRuleArn"] = "arn"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        "ConfigRule Arn and Id can not be specified when creating a new "
        "ConfigRule." in err["Message"]
    )

    managed_rule = managed_config_rule()
    bad_json_string = "{'name': 'test', 'type': null, }"
    managed_rule["InputParameters"] = bad_json_string
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        f"Invalid json {bad_json_string} passed in the InputParameters field"
        in err["Message"]
    )

    managed_rule = managed_config_rule()
    managed_rule["MaximumExecutionFrequency"] = "HOUR"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "ValidationException"
    assert (
        "Member must satisfy enum value set: {One_Hour, Six_Hours, "
        "Three_Hours, Twelve_Hours, TwentyFour_Hours}" in err["Message"]
    )

    managed_rule = managed_config_rule()
    managed_rule["ConfigRuleState"] = "BOGUS"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "ValidationException"
    assert (
        "Value 'BOGUS' at 'configRule.configRuleState' failed to satisfy "
        "constraint: Member must satisfy enum value set: {ACTIVE, "
        "DELETING, DELETING_RESULTS, EVALUATING}" in err["Message"]
    )

    managed_rule = managed_config_rule()
    managed_rule["ConfigRuleState"] = "DELETING"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        "The ConfigRuleState DELETING is invalid.  Only the following values "
        "are permitted: ACTIVE" in err["Message"]
    )

    managed_rule = managed_config_rule()
    managed_rule["CreatedBy"] = "tester"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        "AWS Config populates the CreatedBy field for ServiceLinkedConfigRule. "
        "Try again without populating the CreatedBy field" in err["Message"]
    )


@mock_config
def test_aws_managed_rule_errors():
    """Test various error conditions in ConfigRule instantiation."""
    client = boto3.client("config", region_name=TEST_REGION)

    # Extra, unknown input parameter should raise an error.
    managed_rule = managed_config_rule()
    managed_rule["Source"]["SourceIdentifier"] = "IAM_PASSWORD_POLICY"
    managed_rule["InputParameters"] = '{"RequireNumbers":"true","Extra":"10"}'
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        f"Unknown parameters provided in the inputParameters: "
        f"{managed_rule['InputParameters']}" in err["Message"]
    )

    # Missing required parameters should raise an error.
    managed_rule = managed_config_rule()
    managed_rule["Source"]["SourceIdentifier"] = "CLOUDWATCH_ALARM_ACTION_CHECK"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        "The required parameter [alarmActionRequired, "
        "insufficientDataActionRequired, okActionRequired] is not present "
        "in the inputParameters" in err["Message"]
    )

    # If no MaxExecutionFrequency specified, set it to the default.
    # rule_name = f"managed_rule_{random_string()}"
    # managed_rule = {
    #     "ConfigRuleName": rule_name,
    #     "Description": "Managed S3 Public Read Prohibited Bucket Rule",
    #     "Scope": {"ComplianceResourceTypes": ["AWS::IAM::Group"]},
    #     "Source": {
    #         "Owner": "AWS",
    #         "SourceIdentifier": "IAM_PASSWORD_POLICY",
    #     },
    # }
    # client.put_config_rule(ConfigRule=managed_rule)
    # rsp = client.describe_config_rules(ConfigRuleNames=[rule_name])
    # new_config_rule = rsp["ConfigRules"][0]
    # assert new_config_rule["MaximumExecutionFrequency"] == "TwentyFour_Hours"


@mock_config
def test_config_rules_scope_errors():  # pylint: disable=too-many-statements
    """Test various error conditions in ConfigRule.Scope instantiation."""
    client = boto3.client("config", region_name=TEST_REGION)

    managed_rule = managed_config_rule()
    managed_rule["Scope"]["TagValue"] = "tester"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        "Tag key should not be empty when tag value is provided in scope"
        in err["Message"]
    )

    managed_rule = managed_config_rule()
    managed_rule["Scope"]["ComplianceResourceId"] = "12345"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert (
        "A single resourceType should be provided when resourceId is provided "
        "in scope" in err["Message"]
    )

    managed_rule = managed_config_rule()
    tag_key = "hellobye" * 16 + "x"
    managed_rule["Scope"]["TagKey"] = tag_key
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "ValidationException"
    assert (
        f"Value '{tag_key}' at 'ConfigRule.Scope.TagKey' failed to satisfy "
        f"constraint: Member must have length less than or equal to 128"
        in err["Message"]
    )

    managed_rule = managed_config_rule()
    managed_rule["Scope"]["TagKey"] = "test"
    tag_value = "01234567890123456" * 16 + "x"
    managed_rule["Scope"]["TagValue"] = tag_value
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "ValidationException"
    assert (
        f"Value '{tag_value}' at 'ConfigRule.Scope.TagValue' failed to "
        f"satisfy constraint: Member must have length less than or equal to "
        f"256" in err["Message"]
    )

    managed_rule = managed_config_rule()
    managed_rule["Scope"]["TagKey"] = "test"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert "Scope cannot be applied to both resource and tag" in err["Message"]

    managed_rule = managed_config_rule()
    managed_rule["Scope"]["TagKey"] = "test_key"
    managed_rule["Scope"]["ComplianceResourceTypes"] = []
    managed_rule["Scope"]["ComplianceResourceId"] = "12345"
    with pytest.raises(ClientError) as exc:
        client.put_config_rule(ConfigRule=managed_rule)
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidParameterValueException"
    assert "Scope cannot be applied to both resource and tag" in err["Message"]


@mock_config
def test_valid_put_config_managed_rule():
    """Test valid put_config_rule API calls for managed rules."""
    client = boto3.client("config", region_name=TEST_REGION)

    # Create managed rule and compare input against describe_config_rule()
    # output.
    managed_rule = managed_config_rule()
    managed_rule["Source"]["SourceIdentifier"] = "IAM_PASSWORD_POLICY"
    managed_rule["Scope"]["ComplianceResourceTypes"] = ["AWS::IAM::Group"]
    managed_rule["Scope"]["ComplianceResourceId"] = "basic_test"
    managed_rule["InputParameters"] = '{"RequireUppercaseCharacters":"true"}'
    managed_rule["ConfigRuleState"] = "ACTIVE"
    client.put_config_rule(ConfigRule=managed_rule)

    rsp = client.describe_config_rules(ConfigRuleNames=[managed_rule["ConfigRuleName"]])
    managed_rule_json = json.dumps(managed_rule, sort_keys=True)
    new_config_rule = rsp["ConfigRules"][0]
    rule_arn = new_config_rule.pop("ConfigRuleArn")
    rule_id = new_config_rule.pop("ConfigRuleId")
    rsp_json = json.dumps(new_config_rule, sort_keys=True)
    assert managed_rule_json == rsp_json

    # Update managed rule and compare again.
    managed_rule["ConfigRuleArn"] = rule_arn
    managed_rule["ConfigRuleId"] = rule_id
    managed_rule["Description"] = "Updated Managed S3 Public Read Rule"
    managed_rule["Scope"]["ComplianceResourceTypes"] = ["AWS::S3::Bucket"]
    managed_rule["MaximumExecutionFrequency"] = "Six_Hours"
    managed_rule["InputParameters"] = "{}"
    client.put_config_rule(ConfigRule=managed_rule)

    rsp = client.describe_config_rules(ConfigRuleNames=[managed_rule["ConfigRuleName"]])
    managed_rule_json = json.dumps(managed_rule, sort_keys=True)
    rsp_json = json.dumps(rsp["ConfigRules"][0], sort_keys=True)
    assert managed_rule_json == rsp_json

    # Valid InputParameters.
    managed_rule = {
        "ConfigRuleName": f"input_param_test_{random_string()}",
        "Description": "Provide subset of allowed input parameters",
        "InputParameters": '{"blockedPort1":"22","blockedPort2":"3389"}',
        "Scope": {"ComplianceResourceTypes": ["AWS::IAM::SecurityGroup"]},
        "Source": {"Owner": "AWS", "SourceIdentifier": "RESTRICTED_INCOMING_TRAFFIC"},
    }
    client.put_config_rule(ConfigRule=managed_rule)

    rsp = client.describe_config_rules(ConfigRuleNames=[managed_rule["ConfigRuleName"]])
    managed_rule_json = json.dumps(managed_rule, sort_keys=True)
    new_config_rule = rsp["ConfigRules"][0]
    del new_config_rule["ConfigRuleArn"]
    del new_config_rule["ConfigRuleId"]
    del new_config_rule["ConfigRuleState"]
    rsp_json = json.dumps(new_config_rule, sort_keys=True)
    assert managed_rule_json == rsp_json


@mock_config
def test_describe_config_rules():
    """Test the describe_config_rules API."""
    client = boto3.client("config", region_name=TEST_REGION)

    response = client.describe_config_rules()
    assert len(response["ConfigRules"]) == 0

    rule_name_base = "describe_test"
    for idx in range(ConfigRule.MAX_RULES):
        managed_rule = managed_config_rule()
        managed_rule["ConfigRuleName"] = f"{rule_name_base}_{idx}"
        client.put_config_rule(ConfigRule=managed_rule)

    with pytest.raises(ClientError) as exc:
        client.describe_config_rules(
            ConfigRuleNames=[
                f"{rule_name_base}_1",
                f"{rule_name_base}_10",
                "fooey",
                f"{rule_name_base}_20",
            ]
        )
    err = exc.value.response["Error"]
    assert err["Code"] == "NoSuchConfigRuleException"
    assert "The ConfigRule 'fooey' provided in the request is invalid" in err["Message"]

    # Request three specific ConfigRules.
    response = client.describe_config_rules(
        ConfigRuleNames=[
            f"{rule_name_base}_1",
            f"{rule_name_base}_10",
            f"{rule_name_base}_20",
        ]
    )
    assert len(response["ConfigRules"]) == 3

    # By default, if no ConfigRules are specified, all that can be fit on a
    # "page" will be returned.
    response = client.describe_config_rules()
    assert len(response["ConfigRules"]) == CONFIG_RULE_PAGE_SIZE

    # Test a bad token.
    with pytest.raises(ClientError) as exc:
        client.describe_config_rules(NextToken="foo")
    err = exc.value.response["Error"]
    assert err["Code"] == "InvalidNextTokenException"
    assert "The nextToken provided is invalid" in err["Message"]

    # Loop using tokens, verifying the tokens are as expected.
    expected_tokens = [  # Non-alphanumeric sorted token numbers
        f"{rule_name_base}_120",
        f"{rule_name_base}_143",
        f"{rule_name_base}_31",
        f"{rule_name_base}_54",
        f"{rule_name_base}_77",
        None,
    ]
    idx = 0
    token = f"{rule_name_base}_0"
    while token:
        rsp = client.describe_config_rules(NextToken=token)
        token = rsp.get("NextToken")
        assert token == expected_tokens[idx]
        idx += 1


@mock_config
def test_delete_config_rules():
    """Test the delete_config_rule API."""
    client = boto3.client("config", region_name=TEST_REGION)

    # Create a ConfigRule:
    rule_name = "test_delete_config_rule"
    managed_rule = managed_config_rule()
    managed_rule["ConfigRuleName"] = rule_name
    client.put_config_rule(ConfigRule=managed_rule)

    # Delete it:
    client.delete_config_rule(ConfigRuleName=rule_name)

    # Verify that none are there:
    assert not client.describe_config_rules()["ConfigRules"]

    # Try it again -- it should error indicating the rule could not be found.
    with pytest.raises(ClientError) as exc:
        client.delete_config_rule(ConfigRuleName=rule_name)
    err = exc.value.response["Error"]
    assert err["Code"] == "NoSuchConfigRuleException"
    assert (
        f"The ConfigRule '{rule_name}' provided in the request is invalid"
        in err["Message"]
    )