Bug: CloudWatchLogsEnvelope raises ValidationError on non-JSON logEvents[*].message (e.g. Lambda START/END/REPORT) #8695

Expected Behaviour

CloudWatchLogsEnvelope should decode awslogs.data, extract each logEvents[*].message, and validate it against the provided model. When model=str, every message string should be returned as-is, including plain-text Lambda runtime messages (START, END, REPORT).

Current Behaviour

Real CloudWatch Logs subscriptions forward all log events from the log group, including Lambda runtime messages (START, END, REPORT) which are plain text, not JSON. When any logEvents[*].message is not valid JSON, the envelope raises a ValidationError instead of returning the string.

Location (aws-powertools/powertools-lambda-python @ default branch 5785377):

• functions.py:71-74 — unconditionally calls adapter.validate_json(data) for string input.
• Reached via base.py:44 (_parse) from cloudwatch.py:44, which passes record.message.

Current code:

if isinstance(data, str):
    ...
    return adapter.validate_json(data)

Problem: The envelope's internal _parse path assumes the message is always a JSON string and calls adapter.validate_json(data) on it. For any plain-text message (such as a Lambda START/END/REPORT line), validate_json fails because the input is not JSON.

Code snippet

import base64, json, zlib
from aws_lambda_powertools.utilities.parser import envelopes, parse

inner = {
    "messageType": "DATA_MESSAGE",
    "owner": "123456789012",
    "logGroup": "/aws/lambda/my-function",
    "logStream": "2024/10/21/[$LATEST]abcdef1234567890",
    "subscriptionFilters": ["my-filter"],
    "logEvents": [
        {"id": "evt-0", "timestamp": 1729520895158, "message": "START RequestId: eadc91c8-8273-4a2d-b170-b0a7883b619e Version: $LATEST\n"},
        {"id": "evt-1", "timestamp": 1729520895159, "message": "END RequestId: eadc91c8-8273-4a2d-b170-b0a7883b619e\n"},
    ],
}
compressed = zlib.compress(json.dumps(inner).encode("utf-8"), -1)
raw_event = {"awslogs": {"data": base64.b64encode(compressed)}}

# This raises ValidationError: Invalid JSON
parsed = parse(event=raw_event, model=str, envelope=envelopes.CloudWatchLogsEnvelope)

Steps to Reproduce

Add the following test on latest HEAD and run it (the two tests below fail; the same suite passes without this patch, exit code 0):

def test_cloudwatch_envelope_handles_non_json_messages():
    plain_text_messages = [
        "START RequestId: eadc91c8-8273-4a2d-b170-b0a7883b619e Version: $LATEST\n",
        "2024-10-21T14:28:15.158Z\teadc91c8-8273-4a2d-b170-b0a7883b619e\tINFO\tHello from handler\n",
        "END RequestId: eadc91c8-8273-4a2d-b170-b0a7883b619e\n",
        "REPORT RequestId: eadc91c8-8273-4a2d-b170-b0a7883b619e\tDuration: 2.07 ms\tBilled Duration: 3 ms\tMemory Size: 128 MB\tMax Memory Used: 67 MB\t\n",
    ]

    log_events = [
        {"id": f"event-{i}", "timestamp": 1729520895158 + i, "message": msg}
        for i, msg in enumerate(plain_text_messages)
    ]

    raw_event = _make_cloudwatch_event(log_events)

    parsed = parse(
        event=raw_event,
        model=str,
        envelope=envelopes.CloudWatchLogsEnvelope,
    )

    assert len(parsed) == len(plain_text_messages)
    for i, msg in enumerate(plain_text_messages):
        assert parsed[i] == msg

(A second test, test_cloudwatch_envelope_handles_mixed_json_and_plain_text, mixing JSON and plain-text messages, fails identically.)

Failing output:

tests/unit/parser/_pydantic/test_cloudwatch_non_json_messages.py::test_cloudwatch_envelope_handles_non_json_messages FAILED
tests/unit/parser/_pydantic/test_cloudwatch_non_json_messages.py::test_cloudwatch_envelope_handles_mixed_json_and_plain_text FAILED

E   pydantic_core._pydantic_core.ValidationError: 1 validation error for str
E     Invalid JSON: expected value at line 1 column 1 [type=json_invalid, input_value='START RequestId: eadc91c...619e Version: $LATEST\n', input_type=str]

=================== 2 failed, 625 passed, 1 skipped in 2.64s ===================

Environment:

• Commit: 5785377 (default branch, 2026-10-25)
• Toolchain: Python 3.14.10, dependencies pinned via uv.lock

Possible Solution

Already fixed in aws-powertools/powertools-lambda-typescript

PR: aws-powertools/powertools-lambda-typescript#3505 — "fix(parser): CloudWatch Log Envelope handles non-JSON" (merged 2025-01-22T15:19:24Z, merge commit 781a14e11b5).

The fix removes the forced JSON parsing of each logEvents[*].message and instead validates the raw message directly against the user-supplied schema, so plain-text messages are no longer rejected. The relevant part for this behaviour is the change in cloudwatch.ts from Envelope.parse(record.message, schema) (which forced JSON parsing) to schema.parse(record.message).

How powertools-lambda-typescript fixed it:

   parse<T extends ZodSchema>(data: unknown, schema: T): z.infer<T>[] {
     const parsedEnvelope = CloudWatchLogsSchema.parse(data);
 
-    return parsedEnvelope.awslogs.data.logEvents.map((record) => {
-      return Envelope.parse(record.message, schema);
+    return parsedEnvelope.awslogs.data.logEvents.map((record, index) => {
+      try {
+        return schema.parse(record.message);
+      } catch (error) {
+        throw new ParseError(
+          `Failed to parse CloudWatch log event at index ${index}`,
...
+      }
     });
   },

Suggested fix

When validate_json rejects a payload that is not JSON, fall back to validate_python on the raw value, so string-compatible models (e.g. model=str) receive plain-text log lines; if that also fails, the original error is raised:

diff --git a/aws_lambda_powertools/utilities/parser/functions.py b/aws_lambda_powertools/utilities/parser/functions.py
--- a/aws_lambda_powertools/utilities/parser/functions.py
+++ b/aws_lambda_powertools/utilities/parser/functions.py
@@ -4,7 +4,7 @@ import json
 import logging
 from typing import TYPE_CHECKING, Any
 
-from pydantic import IPvAnyNetwork, TypeAdapter
+from pydantic import IPvAnyNetwork, TypeAdapter, ValidationError
 
 from aws_lambda_powertools.shared.cache_dict import LRUDict
 
@@ -80,6 +80,15 @@ def _parse_and_validate_event(data: dict[str, Any] | Any, adapter: TypeAdapter[T
                 "See issue: https://github.com/aws-powertools/powertools-lambda-python/issues/5303",
             )
             data = json.loads(data)
+        except ValidationError as exc:
+            # validate_json fails for plain-text payloads (e.g., CloudWatch Logs
+            # subscriptions forward Lambda runtime START/END/REPORT lines). Fall
+            # back to validate_python so string-compatible models still receive
+            # the raw payload; otherwise surface the original error.
+            try:
+                return adapter.validate_python(data)
+            except ValidationError:
+                raise exc from None
 
     return adapter.validate_python(data)

Verified on 5785377 (Python 3.14.10, dependencies from uv.lock): the test above fails without this patch and passes with it; no other test results change.

Powertools for AWS Lambda (Python) version

latest (5785377)

AWS Lambda function runtime

3.14

Packaging format used

PyPi