Merge pull request #5234 from mozilla/test-verify-from-sns-mpp-3852

MPP-3852: Test `verify_from_sns`

Author: groovecoder

emails/sns.py

@@ -64,6 +64,16 @@
 
 
 def verify_from_sns(json_body):
+    """
+    Check that the SNS message was signed by the cetificate.
+
+    https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message.html
+
+    Only supports SignatureVersion 1. SignatureVersion 2 (SHA256) was added in
+    September 2022, and requires opt-in.
+
+    TODO MPP-3852: Stop using OpenSSL.crypto
+    """
     pemfile = _grab_keyfile(json_body["SigningCertURL"])
     cert = crypto.load_certificate(crypto.FILETYPE_PEM, pemfile)
     signature = base64.decodebytes(json_body["Signature"].encode("utf-8"))

emails/tests/sns_tests.py

@@ -1,15 +1,220 @@
-from unittest.mock import patch
+# ruff: noqa: S303  # Use of insecure SHA1 hash function
 
+from base64 import b64encode
+from collections.abc import Iterator
+from datetime import UTC, datetime, timedelta
+from io import BytesIO
+from unittest.mock import Mock, patch
+
+from django.core.cache import BaseCache, caches
 from django.core.exceptions import SuspiciousOperation
-from django.test import TestCase
 
-from ..sns import _grab_keyfile
+import pytest
+from cryptography import x509
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import padding, rsa
+from cryptography.x509.oid import NameOID
+from OpenSSL.crypto import Error
+from pytest_django.fixtures import SettingsWrapper
+
+from ..sns import (
+    NOTIFICATION_HASH_FORMAT,
+    NOTIFICATION_WITHOUT_SUBJECT_HASH_FORMAT,
+    SUBSCRIPTION_HASH_FORMAT,
+    _grab_keyfile,
+    verify_from_sns,
+)
+
+
+@pytest.fixture(autouse=True)
+def key_cache(settings: SettingsWrapper) -> Iterator[BaseCache]:
+    """
+    Return the cache used for signing certificates.
+
+    Clear the cache before and after tests.
+    """
+    key_cache = caches[getattr(settings, "AWS_SNS_KEY_CACHE", "default")]
+    key_cache.clear()
+    yield key_cache
+    key_cache.clear()
+
+
+@pytest.fixture
+def key_and_cert() -> tuple[rsa.RSAPrivateKey, x509.Certificate]:
+    """Generate an RSA key and a signing certificate"""
+    key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
+    name_attributes = [
+        x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
+        x509.NameAttribute(NameOID.STATE_OR_PROVINCE_NAME, "Oklahoma"),
+        x509.NameAttribute(NameOID.LOCALITY_NAME, "Tulsa"),
+        x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Firefox Private Relay Test"),
+        x509.NameAttribute(NameOID.COMMON_NAME, "github.com/mozilla/fx-private-relay/"),
+    ]
+    subject = issuer = x509.Name(name_attributes)
+    cert = (
+        x509.CertificateBuilder()
+        .subject_name(subject)
+        .issuer_name(issuer)
+        .public_key(key.public_key())
+        .serial_number(x509.random_serial_number())
+        .not_valid_before(datetime.now(UTC) - timedelta(seconds=1))
+        .not_valid_after(datetime.now(UTC) + timedelta(seconds=60))
+        .sign(key, hashes.SHA256())
+    )
+    return key, cert
+
+
+@pytest.fixture
+def signing_cert_url_and_private_key(
+    key_and_cert: tuple[rsa.RSAPrivateKey, x509.Certificate],
+    key_cache: BaseCache,
+    settings: SettingsWrapper,
+) -> tuple[str, rsa.RSAPrivateKey]:
+    """Return the URL and private key for a cached signing certificate."""
+    cert_url = f"https://sns.{settings.AWS_REGION}.amazonaws.com/cert.pem"
+    key, cert = key_and_cert
+    cert_pem = cert.public_bytes(serialization.Encoding.PEM)
+    key_cache.set(cert_url, cert_pem)
+    return cert_url, key
+
+
+@pytest.fixture
+def mock_urlopen() -> Iterator[Mock]:
+    with patch("emails.sns.urlopen") as mock_urlopen:
+        yield mock_urlopen
+
+
+def test_grab_keyfile_checks_cert_url_origin(mock_urlopen: Mock) -> None:
+    cert_url = "https://attacker.com/cert.pem"
+    with pytest.raises(SuspiciousOperation):
+        _grab_keyfile(cert_url)
+    mock_urlopen.assert_not_called()
+
+
+def test_grab_keyfile_downloads_valid_certificate(
+    mock_urlopen: Mock,
+    key_and_cert: tuple[rsa.RSAPrivateKey, x509.Certificate],
+    key_cache: BaseCache,
+    settings: SettingsWrapper,
+) -> None:
+    cert_url = f"https://sns.{settings.AWS_REGION}.amazonaws.com/cert.pem"
+    key, cert = key_and_cert
+    cert_pem = cert.public_bytes(serialization.Encoding.PEM)
+    mock_urlopen.return_value = BytesIO(cert_pem)
+    ret_value = _grab_keyfile(cert_url)
+    mock_urlopen.assert_called_once_with(cert_url)
+    assert ret_value == cert_pem
+    assert key_cache.get(cert_url) == cert_pem
+
+
+def test_grab_keyfile_reads_from_cache(
+    mock_urlopen: Mock,
+    key_cache: BaseCache,
+    settings: SettingsWrapper,
+) -> None:
+    cert_url = f"https://sns.{settings.AWS_REGION}.amazonaws.com/cert.pem"
+    fake_pem = b"I am fake"
+    key_cache.set(cert_url, fake_pem)
+    ret_value = _grab_keyfile(cert_url)
+    assert ret_value == fake_pem
+    mock_urlopen.assert_not_called()
+
+
+def test_grab_keyfile_cert_chain_fails(
+    mock_urlopen: Mock,
+    key_and_cert: tuple[rsa.RSAPrivateKey, x509.Certificate],
+    key_cache: BaseCache,
+    settings: SettingsWrapper,
+) -> None:
+    cert_url = f"https://sns.{settings.AWS_REGION}.amazonaws.com/cert.pem"
+    key, cert = key_and_cert
+    cert_pem = cert.public_bytes(serialization.Encoding.PEM)
+    two_cert_pem = b"\n".join((cert_pem, cert_pem))
+    mock_urlopen.return_value = BytesIO(two_cert_pem)
+    with pytest.raises(ValueError, match="Invalid Certificate File"):
+        _grab_keyfile(cert_url)
+
+
+def test_verify_from_sns_notification_with_subject_ver1(
+    signing_cert_url_and_private_key: tuple[str, rsa.RSAPrivateKey],
+) -> None:
+    cert_url, key = signing_cert_url_and_private_key
+    json_body = {
+        "Type": "Notification",
+        "Message": "message",
+        "MessageId": "message_id",
+        "Subject": "subject",
+        "Timestamp": "timestamp",
+        "TopicArn": "topic_arn",
+        "SigningCertURL": cert_url,
+        "SignatureVersion": 1,
+    }
+    text_to_sign = NOTIFICATION_HASH_FORMAT.format(**json_body)
+    signature = key.sign(text_to_sign.encode(), padding.PKCS1v15(), hashes.SHA1())
+    json_body["Signature"] = b64encode(signature).decode()
+    ret = verify_from_sns(json_body)
+    assert ret == json_body
+
+
+def test_verify_from_sns_notification_with_subject_ver1_fails(
+    signing_cert_url_and_private_key: tuple[str, rsa.RSAPrivateKey],
+) -> None:
+    cert_url, key = signing_cert_url_and_private_key
+    json_body = {
+        "Type": "Notification",
+        "Message": "message",
+        "MessageId": "message_id",
+        "Subject": "subject",
+        "Timestamp": "timestamp",
+        "TopicArn": "topic_arn",
+        "SigningCertURL": cert_url,
+        "SignatureVersion": 1,
+    }
+    text_to_sign = NOTIFICATION_HASH_FORMAT.format(**json_body)
+    signature = key.sign(text_to_sign.encode(), padding.PKCS1v15(), hashes.SHA1())
+    json_body["Signature"] = b64encode(signature).decode()
+    json_body["Message"] = "different message"
+    with pytest.raises(Error):
+        verify_from_sns(json_body)
+
+
+def test_verify_from_sns_notification_without_subject_ver1(
+    signing_cert_url_and_private_key: tuple[str, rsa.RSAPrivateKey],
+) -> None:
+    cert_url, key = signing_cert_url_and_private_key
+    json_body = {
+        "Type": "Notification",
+        "Message": "message",
+        "MessageId": "message_id",
+        "Timestamp": "timestamp",
+        "TopicArn": "topic_arn",
+        "SigningCertURL": cert_url,
+        "SignatureVersion": 1,
+    }
+    text_to_sign = NOTIFICATION_WITHOUT_SUBJECT_HASH_FORMAT.format(**json_body)
+    signature = key.sign(text_to_sign.encode(), padding.PKCS1v15(), hashes.SHA1())
+    json_body["Signature"] = b64encode(signature).decode()
+    ret = verify_from_sns(json_body)
+    assert ret == json_body
 
 
-class GrabKeyfileTest(TestCase):
-    @patch("emails.sns.urlopen")
-    def test_grab_keyfile_checks_cert_url_origin(self, mock_urlopen):
-        cert_url = "https://attacker.com/cert.pem"
-        with self.assertRaises(SuspiciousOperation):
-            _grab_keyfile(cert_url)
-        mock_urlopen.assert_not_called()
+def test_verify_from_sns_subscription_ver1(
+    signing_cert_url_and_private_key: tuple[str, rsa.RSAPrivateKey],
+) -> None:
+    cert_url, key = signing_cert_url_and_private_key
+    json_body = {
+        "Type": "Subscription",
+        "Message": "message",
+        "MessageId": "message_id",
+        "SubscribeURL": "subscribe_url",
+        "Timestamp": "timestamp",
+        "Token": "token",
+        "TopicArn": "topic_arn",
+        "SigningCertURL": cert_url,
+        "SignatureVersion": 1,
+    }
+    text_to_sign = SUBSCRIPTION_HASH_FORMAT.format(**json_body)
+    signature = key.sign(text_to_sign.encode(), padding.PKCS1v15(), hashes.SHA1())
+    json_body["Signature"] = b64encode(signature).decode()
+    ret = verify_from_sns(json_body)
+    assert ret == json_body