feat(cloudformation): support logs resource policy (#4427)

moto/logs/models.py

@@ -4,6 +4,7 @@ from boto3 import Session
 
 from moto import core as moto_core
 from moto.core import BaseBackend, BaseModel
+from moto.core.models import CloudFormationModel
 from moto.core.utils import unix_time_millis
 from moto.utilities.paginator import paginate
 from moto.logs.metric_filters import MetricFilters
@@ -540,6 +541,72 @@ class LogGroup(BaseModel):
         self.subscription_filters = []
 
 
+class LogResourcePolicy(CloudFormationModel):
+    def __init__(self, policy_name, policy_document):
+        self.policy_name = policy_name
+        self.policy_document = policy_document
+        self.last_updated_time = int(unix_time_millis())
+
+    def update(self, policy_document):
+        self.policy_document = policy_document
+        self.last_updated_time = int(unix_time_millis())
+
+    def describe(self):
+        return {
+            "policyName": self.policy_name,
+            "policyDocument": self.policy_document,
+            "lastUpdatedTime": self.last_updated_time,
+        }
+
+    @property
+    def physical_resource_id(self):
+        return self.policy_name
+
+    @staticmethod
+    def cloudformation_name_type():
+        return "PolicyName"
+
+    @staticmethod
+    def cloudformation_type():
+        # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-logs-resourcepolicy.html
+        return "AWS::Logs::ResourcePolicy"
+
+    @classmethod
+    def create_from_cloudformation_json(
+        cls, resource_name, cloudformation_json, region_name
+    ):
+        properties = cloudformation_json["Properties"]
+        policy_name = properties["PolicyName"]
+        policy_document = properties["PolicyDocument"]
+        return logs_backends[region_name].put_resource_policy(
+            policy_name, policy_document
+        )
+
+    @classmethod
+    def update_from_cloudformation_json(
+        cls, original_resource, new_resource_name, cloudformation_json, region_name
+    ):
+        properties = cloudformation_json["Properties"]
+        policy_name = properties["PolicyName"]
+        policy_document = properties["PolicyDocument"]
+
+        updated = logs_backends[region_name].put_resource_policy(
+            policy_name, policy_document
+        )
+        # TODO: move `update by replacement logic` to cloudformation. this is required for implementing rollbacks
+        if original_resource.policy_name != policy_name:
+            logs_backends[region_name].delete_resource_policy(
+                original_resource.policy_name
+            )
+        return updated
+
+    @classmethod
+    def delete_from_cloudformation_json(
+        cls, resource_name, cloudformation_json, region_name
+    ):
+        return logs_backends[region_name].delete_resource_policy(resource_name)
+
+
 class LogsBackend(BaseBackend):
     def __init__(self, region_name):
         self.region_name = region_name
@@ -741,29 +808,19 @@ class LogsBackend(BaseBackend):
         """
         limit = limit or MAX_RESOURCE_POLICIES_PER_REGION
 
-        policies = []
+        return list(self.resource_policies.values())
-        for policy_name, policy_info in self.resource_policies.items():
-            policies.append(
-                {
-                    "policyName": policy_name,
-                    "policyDocument": policy_info["policyDocument"],
-                    "lastUpdatedTime": policy_info["lastUpdatedTime"],
-                }
-            )
-        return policies
 
     def put_resource_policy(self, policy_name, policy_doc):
-        """Create resource policy and return dict of policy name and doc."""
+        """Creates/updates resource policy and return policy object"""
+        if policy_name in self.resource_policies:
+            policy = self.resource_policies[policy_name]
+            policy.update(policy_doc)
+            return policy
         if len(self.resource_policies) == MAX_RESOURCE_POLICIES_PER_REGION:
             raise LimitExceededException()
-
+        policy = LogResourcePolicy(policy_name, policy_doc)
-        policy = {
-            "policyName": policy_name,
-            "policyDocument": policy_doc,
-            "lastUpdatedTime": int(unix_time_millis()),
-        }
         self.resource_policies[policy_name] = policy
-        return {"resourcePolicy": policy}
+        return policy
 
     def delete_resource_policy(self, policy_name):
         """Remove resource policy with a policy name matching given name."""

moto/logs/responses.py

@@ -293,13 +293,13 @@ class LogsResponse(BaseResponse):
         next_token = self._get_param("nextToken")
         limit = self._get_param("limit")
         policies = self.logs_backend.describe_resource_policies(next_token, limit)
-        return json.dumps({"resourcePolicies": policies})
+        return json.dumps({"resourcePolicies": [p.describe() for p in policies]})
 
     def put_resource_policy(self):
         policy_name = self._get_param("policyName")
         policy_doc = self._get_param("policyDocument")
-        result = self.logs_backend.put_resource_policy(policy_name, policy_doc)
+        policy = self.logs_backend.put_resource_policy(policy_name, policy_doc)
-        return json.dumps(result)
+        return json.dumps({"resourcePolicy": policy.describe()})
 
     def delete_resource_policy(self):
         policy_name = self._get_param("policyName")

tests/__init__.py

@@ -2,6 +2,8 @@ from __future__ import unicode_literals
 
 import logging
 
+from . import helpers  # noqa
+
 # Disable extra logging for tests
 logging.getLogger("boto").setLevel(logging.CRITICAL)
 logging.getLogger("boto3").setLevel(logging.CRITICAL)

tests/helpers.py

@@ -1,6 +1,8 @@
 from __future__ import unicode_literals
 import boto
 from unittest import SkipTest
+from collections.abc import Iterable
+from sure import assertion
 
 
 def version_tuple(v):
@@ -25,3 +27,22 @@ class requires_boto_gte(object):
         if boto_version >= required:
             return test
         return skip_test
+
+
+@assertion
+def containing_item_with_attributes(context, **kwargs):
+    contains = False
+    if kwargs and isinstance(context.obj, Iterable):
+        for item in context.obj:
+            if not isinstance(item, dict):
+                continue
+            for k, v in kwargs.items():
+                if k not in item or item[k] != v:
+                    break
+            else:
+                contains = True
+    if context.negative:
+        assert not contains, f"{context.obj} contains matching item {kwargs}"
+    else:
+        assert contains, f"{context.obj} does not contain matching item {kwargs}"
+    return True

tests/test_cloudformation/test_cloudformation_stack_integration.py

@@ -2849,13 +2849,108 @@ def test_create_log_group_using_fntransform():
         },
     }
 
-    cf_conn = boto3.client("cloudformation", "us-west-2")
-    cf_conn.create_stack(StackName="test_stack", TemplateBody=json.dumps(template))
 
-    logs_conn = boto3.client("logs", region_name="us-west-2")
+@mock_cloudformation
-    log_group = logs_conn.describe_log_groups()["logGroups"][0]
+@mock_logs
-    log_group["logGroupName"].should.equal("some-log-group")
+def test_create_cloudwatch_logs_resource_policy():
-    log_group["retentionInDays"].should.be.equal(90)
+    policy_document = json.dumps(
+        {
+            "Statement": [
+                {
+                    "Action": ["logs:CreateLogStream", "logs:PutLogEvents",],
+                    "Effect": "Allow",
+                    "Principal": {"Service": "es.amazonaws.com"},
+                    "Resource": "*",
+                }
+            ]
+        }
+    )
+    template1 = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Resources": {
+            "LogGroupPolicy1": {
+                "Type": "AWS::Logs::ResourcePolicy",
+                "Properties": {
+                    "PolicyDocument": policy_document,
+                    "PolicyName": "TestPolicyA",
+                },
+            }
+        },
+    }
+    template2 = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Resources": {
+            "LogGroupPolicy1": {
+                "Type": "AWS::Logs::ResourcePolicy",
+                "Properties": {
+                    "PolicyDocument": policy_document,
+                    "PolicyName": "TestPolicyB",
+                },
+            },
+            "LogGroupPolicy2": {
+                "Type": "AWS::Logs::ResourcePolicy",
+                "Properties": {
+                    "PolicyDocument": policy_document,
+                    "PolicyName": "TestPolicyC",
+                },
+            },
+        },
+    }
+
+    cf_conn = boto3.client("cloudformation", "us-east-1")
+    cf_conn.create_stack(StackName="test_stack", TemplateBody=json.dumps(template1))
+
+    logs_conn = boto3.client("logs", region_name="us-east-1")
+    policies = logs_conn.describe_resource_policies()["resourcePolicies"]
+    policies.should.have.length_of(1)
+    policies.should.be.containing_item_with_attributes(
+        policyName="TestPolicyA", policyDocument=policy_document
+    )
+
+    cf_conn.update_stack(StackName="test_stack", TemplateBody=json.dumps(template2))
+    policies = logs_conn.describe_resource_policies()["resourcePolicies"]
+    policies.should.have.length_of(2)
+    policies.should.be.containing_item_with_attributes(
+        policyName="TestPolicyB", policyDocument=policy_document
+    )
+    policies.should.be.containing_item_with_attributes(
+        policyName="TestPolicyC", policyDocument=policy_document
+    )
+
+    cf_conn.update_stack(StackName="test_stack", TemplateBody=json.dumps(template1))
+    policies = logs_conn.describe_resource_policies()["resourcePolicies"]
+    policies.should.have.length_of(1)
+    policies.should.be.containing_item_with_attributes(
+        policyName="TestPolicyA", policyDocument=policy_document
+    )
+
+
+@mock_cloudformation
+@mock_logs
+def test_delete_stack_containing_cloudwatch_logs_resource_policy():
+    template1 = {
+        "AWSTemplateFormatVersion": "2010-09-09",
+        "Resources": {
+            "LogGroupPolicy1": {
+                "Type": "AWS::Logs::ResourcePolicy",
+                "Properties": {
+                    "PolicyDocument": '{"Statement":[{"Action":"logs:*","Effect":"Allow","Principal":"*","Resource":"*"}]}',
+                    "PolicyName": "TestPolicyA",
+                },
+            }
+        },
+    }
+
+    cf_conn = boto3.client("cloudformation", "us-east-1")
+    cf_conn.create_stack(StackName="test_stack", TemplateBody=json.dumps(template1))
+
+    logs_conn = boto3.client("logs", region_name="us-east-1")
+    policies = logs_conn.describe_resource_policies()["resourcePolicies"]
+    policies.should.have.length_of(1)
+
+    cf_conn.delete_stack(StackName="test_stack")
+    policies = logs_conn.describe_resource_policies()["resourcePolicies"]
+    policies.should.have.length_of(0)
 
 
 @mock_cloudformation

tests/test_logs/test_logs.py

@@ -3,10 +3,12 @@ import os
 import time
 import sure  # noqa
 from unittest import SkipTest
+from datetime import timedelta
 
 import boto3
 import pytest
 from botocore.exceptions import ClientError
+from freezegun import freeze_time
 
 from moto import mock_logs, settings
 from moto.core.utils import unix_time_millis
@@ -592,6 +594,17 @@ def test_put_resource_policy():
 
     client.delete_log_group(logGroupName=log_group_name)
 
+    # put_resource_policy with same policy name should update the resouce
+    created_time = response["resourcePolicy"]["lastUpdatedTime"]
+    with freeze_time(timedelta(minutes=1)):
+        new_document = '{"Statement":[{"Action":"logs:*","Effect":"Allow","Principal":"*","Resource":"*"}]}'
+        policy_info = client.put_resource_policy(
+            policyName=policy_name, policyDocument=new_document,
+        )["resourcePolicy"]
+        assert policy_info["policyName"] == policy_name
+        assert policy_info["policyDocument"] == new_document
+        assert created_time < policy_info["lastUpdatedTime"] <= int(unix_time_millis())
+
 
 @mock_logs
 def test_put_resource_policy_too_many(json_policy_doc):
@@ -615,6 +628,11 @@ def test_put_resource_policy_too_many(json_policy_doc):
     exc_value.response["Error"]["Code"].should.equal("LimitExceededException")
     exc_value.response["Error"]["Message"].should.contain("Resource limit exceeded.")
 
+    # put_resource_policy on already created policy, shouldnt throw any error
+    client.put_resource_policy(
+        policyName="test_policy_1", policyDocument=json.dumps(json_policy_doc)
+    )
+
 
 @mock_logs
 def test_delete_resource_policy(json_policy_doc):