Skip to content

Commit

Permalink
fix(parser): raise ValidationError when SNS->SQS keys are intentional…
Browse files Browse the repository at this point in the history
…ly missing (#1299)

Co-authored-by: Heitor Lessa <[email protected]>
  • Loading branch information
rubenfonseca and heitorlessa authored Jul 18, 2022
1 parent c912034 commit a659b21
Show file tree
Hide file tree
Showing 2 changed files with 32 additions and 2 deletions.
7 changes: 5 additions & 2 deletions aws_lambda_powertools/utilities/parser/models/sns.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,8 +31,11 @@ class SnsNotificationModel(BaseModel):
def check_sqs_protocol(cls, values):
sqs_rewritten_keys = ("UnsubscribeURL", "SigningCertURL")
if any(key in sqs_rewritten_keys for key in values):
values["UnsubscribeUrl"] = values.pop("UnsubscribeURL")
values["SigningCertUrl"] = values.pop("SigningCertURL")
# The sentinel value 'None' forces the validator to fail with
# ValidatorError instead of KeyError when the key is missing from
# the SQS payload
values["UnsubscribeUrl"] = values.pop("UnsubscribeURL", None)
values["SigningCertUrl"] = values.pop("SigningCertURL", None)
return values


Expand Down
27 changes: 27 additions & 0 deletions tests/functional/parser/test_sns.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import json
from typing import Any, List

import pytest
Expand Down Expand Up @@ -103,3 +104,29 @@ def handle_sns_sqs_json_body(event: List[MySnsBusiness], _: LambdaContext):
def test_handle_sns_sqs_trigger_event_json_body(): # noqa: F811
event_dict = load_event("snsSqsEvent.json")
handle_sns_sqs_json_body(event_dict, LambdaContext())


def test_handle_sns_sqs_trigger_event_json_body_missing_signing_cert_url():
# GIVEN an event is tampered with a missing SigningCertURL
event_dict = load_event("snsSqsEvent.json")
payload = json.loads(event_dict["Records"][0]["body"])
payload.pop("SigningCertURL")
event_dict["Records"][0]["body"] = json.dumps(payload)

# WHEN parsing the payload
# THEN raise a ValidationError error
with pytest.raises(ValidationError):
handle_sns_sqs_json_body(event_dict, LambdaContext())


def test_handle_sns_sqs_trigger_event_json_body_missing_unsubscribe_url():
# GIVEN an event is tampered with a missing UnsubscribeURL
event_dict = load_event("snsSqsEvent.json")
payload = json.loads(event_dict["Records"][0]["body"])
payload.pop("UnsubscribeURL")
event_dict["Records"][0]["body"] = json.dumps(payload)

# WHEN parsing the payload
# THEN raise a ValidationError error
with pytest.raises(ValidationError):
handle_sns_sqs_json_body(event_dict, LambdaContext())

0 comments on commit a659b21

Please sign in to comment.