feat(event_source): Add support for policyLevel field in CloudWatch Logs event and parser

aws_lambda_powertools/utilities/data_classes/cloud_watch_logs_event.py

@@ -58,6 +58,11 @@ def message_type(self) -> str:
         """
         return self["messageType"]
 
+    @property
+    def policy_level(self) -> Optional[str]:
+        """The level at which the policy was enforced."""
+        return self.get("policyLevel")
+
     @property
     def log_events(self) -> List[CloudWatchLogsLogEvent]:
         """The actual log data, represented as an array of log event records.

aws_lambda_powertools/utilities/parser/models/cloudwatch.py

@@ -3,7 +3,7 @@
 import logging
 import zlib
 from datetime import datetime
-from typing import Type, Union
+from typing import Optional, Type, Union
 
 from pydantic import BaseModel, Field, validator
 
@@ -25,6 +25,7 @@ class CloudWatchLogsDecode(BaseModel):
     logStream: str
     subscriptionFilters: List[str]
     logEvents: List[CloudWatchLogsLogEvent]
+    policyLevel: Optional[str] = None
 
 
 class CloudWatchLogsData(BaseModel):

tests/events/cloudWatchLogEventWithPolicyLevel.json

@@ -0,0 +1,5 @@
+{
+  "awslogs": {
+    "data": "eNqFj0+LwjAQxe9+ipCzh7Yb1z+30o0ixO3SdhdEpNR2kEDblCTuIuJ33yS1i4dFL0Nmfi/vzVxGCOEGlCqOkJ07wAuE38IszDc0TcMVxWMrED8tSIv84IVMXqezuXn0qBbHlRSnzlINSrOhH2iqJRTNHb4NHFengyol7zQX7ZLXGqQyyp2T9j3eD0b0G1rtsBkgdHHVIF5Zc7B0XfnO1801N3fporGr+YR4hATz2dTzvD/F7W77fUeTJE72aMml0sjGo4E69XX8IDR4Huo/Ck2hFG31X6qp/f2dqHl5ZiawxgscRlH8+Z7ljH5Rln/EbB1t8ej6C87if5I="
+  }
+}

tests/unit/data_classes/test_cloud_watch_logs_event.py

@@ -19,6 +19,7 @@ def test_cloud_watch_trigger_event():
     assert json_logs_data.log_stream == "testLogStream"
     assert json_logs_data.subscription_filters == ["testFilter"]
     assert json_logs_data.message_type == "DATA_MESSAGE"
+    assert json_logs_data.policy_level is None
 
     assert log_event.get_id == "eventId1"
     assert log_event.timestamp == 1440442987000
@@ -27,3 +28,31 @@ def test_cloud_watch_trigger_event():
 
     event2 = CloudWatchLogsEvent(load_event("cloudWatchLogEvent.json"))
     assert parsed_event.raw_event == event2.raw_event
+
+
+def test_cloud_watch_trigger_event_with_policy_level():
+    raw_event = load_event("cloudWatchLogEventWithPolicyLevel.json")
+    parsed_event = CloudWatchLogsEvent(raw_event)
+
+    decompressed_logs_data = parsed_event.decompress_logs_data
+    assert parsed_event.decompress_logs_data == decompressed_logs_data
+
+    json_logs_data = parsed_event.parse_logs_data()
+    assert parsed_event.parse_logs_data().raw_event == json_logs_data.raw_event
+    log_events = json_logs_data.log_events
+    log_event = log_events[0]
+
+    assert json_logs_data.owner == "123456789123"
+    assert json_logs_data.log_group == "testLogGroup"
+    assert json_logs_data.log_stream == "testLogStream"
+    assert json_logs_data.subscription_filters == ["testFilter"]
+    assert json_logs_data.message_type == "DATA_MESSAGE"
+    assert json_logs_data.policy_level == "ACCOUNT_LEVEL_POLICY"
+
+    assert log_event.get_id == "eventId1"
+    assert log_event.timestamp == 1440442987000
+    assert log_event.message == "[ERROR] First test message"
+    assert log_event.extracted_fields is None
+
+    event2 = CloudWatchLogsEvent(load_event("cloudWatchLogEventWithPolicyLevel.json"))
+    assert parsed_event.raw_event == event2.raw_event

tests/unit/parser/test_cloudwatch.py

@@ -63,6 +63,37 @@ def test_handle_cloudwatch_trigger_event_no_envelope():
     assert parsed_event.awslogs.decoded_data.logStream == raw_event_decoded["logStream"]
     assert parsed_event.awslogs.decoded_data.subscriptionFilters == raw_event_decoded["subscriptionFilters"]
     assert parsed_event.awslogs.decoded_data.messageType == raw_event_decoded["messageType"]
+    assert parsed_event.awslogs.decoded_data.policyLevel is None
+
+    assert len(parsed_event.awslogs.decoded_data.logEvents) == 2
+
+    log_record: CloudWatchLogsLogEvent = parsed_event.awslogs.decoded_data.logEvents[0]
+    raw_log_record = raw_event_decoded["logEvents"][0]
+    assert log_record.id == raw_log_record["id"]
+    convert_time = int(round(log_record.timestamp.timestamp() * 1000))
+    assert convert_time == raw_log_record["timestamp"]
+    assert log_record.message == raw_log_record["message"]
+
+    log_record: CloudWatchLogsLogEvent = parsed_event.awslogs.decoded_data.logEvents[1]
+    raw_log_record = raw_event_decoded["logEvents"][1]
+    assert log_record.id == raw_log_record["id"]
+    convert_time = int(round(log_record.timestamp.timestamp() * 1000))
+    assert convert_time == raw_log_record["timestamp"]
+    assert log_record.message == raw_log_record["message"]
+
+
+def test_handle_cloudwatch_trigger_event_no_envelope_with_policylevel():
+    raw_event = load_event("cloudWatchLogEventWithPolicyLevel.json")
+    parsed_event: CloudWatchLogsModel = CloudWatchLogsModel(**raw_event)
+
+    raw_event_decoded = decode_cloudwatch_raw_event(raw_event["awslogs"]["data"])
+
+    assert parsed_event.awslogs.decoded_data.owner == raw_event_decoded["owner"]
+    assert parsed_event.awslogs.decoded_data.logGroup == raw_event_decoded["logGroup"]
+    assert parsed_event.awslogs.decoded_data.logStream == raw_event_decoded["logStream"]
+    assert parsed_event.awslogs.decoded_data.subscriptionFilters == raw_event_decoded["subscriptionFilters"]
+    assert parsed_event.awslogs.decoded_data.messageType == raw_event_decoded["messageType"]
+    assert parsed_event.awslogs.decoded_data.policyLevel == "ACCOUNT_LEVEL_POLICY"
 
     assert len(parsed_event.awslogs.decoded_data.logEvents) == 2