8fe25d4b-9da7-428d-ad38-de60a26a02cc

Committed 04 Dec 2024 04:09PM UTC coverage: 85.053% (-0.008%) from 85.061%

Build # 8fe25d4b-9da7-428d-ad38-de60a26a02cc

Build Type

Pull #5235

circleci

Committed by

jwhitlock

Commit Message

Remove pem

Pull Request Pull Request #5235: MPP-3852: Use `cryptography` for SNS signature validation, remove `pyopenssl` and `pem`

Run Details

2432 of 3561 branches covered (68.3%)

Branch coverage included in aggregate %.

31 of 33 new or added lines in 4 files covered. (93.94%)

1 existing line in 1 file now uncovered.

16984 of 19267 relevant lines covered (88.15%)

9.93 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

94.2

/emails/sns.py

# Inspired by django-bouncy utils:
# https://github.com/organizerconnect/django-bouncy/blob/master/django_bouncy/utils.py

import base64
import logging
from urllib.request import urlopen

from django.conf import settings
from django.core.cache import caches
from django.core.exceptions import SuspiciousOperation

from cryptography import x509
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.asymmetric import padding, rsa

logger = logging.getLogger("events")

NOTIFICATION_HASH_FORMAT = """\
Message
{Message}
MessageId
{MessageId}
Subject
{Subject}
Timestamp
{Timestamp}
TopicArn
{TopicArn}
Type
{Type}
"""

NOTIFICATION_WITHOUT_SUBJECT_HASH_FORMAT = """\
Message
{Message}
MessageId
{MessageId}
Timestamp
{Timestamp}
TopicArn
{TopicArn}
Type
{Type}
"""

SUBSCRIPTION_HASH_FORMAT = """\
Message
{Message}
MessageId
{MessageId}
SubscribeURL
{SubscribeURL}
Timestamp
{Timestamp}
Token
{Token}
TopicArn
{TopicArn}
Type
{Type}
"""

SUPPORTED_SNS_TYPES = [
    "SubscriptionConfirmation",
    "Notification",
]


class VerificationFailed(ValueError):
    pass


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.
    """
    signing_cert_url = json_body["SigningCertURL"]
    pemfile = _grab_keyfile(signing_cert_url)
    cert = x509.load_pem_x509_certificate(pemfile)
    signature = base64.decodebytes(json_body["Signature"].encode())

    hash_format = _get_hash_format(json_body)
    cert_pubkey = cert.public_key()
    if not isinstance(cert_pubkey, rsa.RSAPublicKey):
        raise VerificationFailed(f"SigningCertURL {signing_cert_url} is not an RSA key")

    try:
        cert_pubkey.verify(
            signature,
            hash_format.format(**json_body).encode(),
            padding.PKCS1v15(),
            hashes.SHA1(),  # noqa: S303  # Use of insecure hash SHA1
        )
    except InvalidSignature as e:
        raise VerificationFailed(
            f"Invalid signature with SigningCertURL {signing_cert_url}"
        ) from e

    return json_body


def _get_hash_format(json_body):
    message_type = json_body["Type"]
    if message_type == "Notification":
        if "Subject" in json_body.keys():
            return NOTIFICATION_HASH_FORMAT
        return NOTIFICATION_WITHOUT_SUBJECT_HASH_FORMAT

    return SUBSCRIPTION_HASH_FORMAT


def _grab_keyfile(cert_url):
    cert_url_origin = f"https://sns.{settings.AWS_REGION}.amazonaws.com/"
    if not (cert_url.startswith(cert_url_origin)):
        raise SuspiciousOperation(
            f'SNS SigningCertURL "{cert_url}" did not start with "{cert_url_origin}"'
        )

    key_cache = caches[getattr(settings, "AWS_SNS_KEY_CACHE", "default")]

    pemfile = key_cache.get(cert_url)
    if not pemfile:
        response = urlopen(cert_url)  # noqa: S310 (check for custom scheme)
        pemfile = response.read()

        # Extract the first certificate in the file and confirm it's a valid
        # PEM certificate
        certs = x509.load_pem_x509_certificates(pemfile)
        # A proper certificate file will contain 1 certificate
        if len(certs) != 1:
            raise VerificationFailed(
                f"SigningCertURL {cert_url} has {len(certs)} certificates."
            )
        cert_pubkey = certs[0].public_key()
        if not isinstance(cert_pubkey, rsa.RSAPublicKey):
            raise VerificationFailed(f"SigningCertURL {cert_url} is not an RSA key")

        key_cache.set(cert_url, pemfile)
    return pemfile

1	# Inspired by django-bouncy utils:
2	# https://github.com/organizerconnect/django-bouncy/blob/master/django_bouncy/utils.py
3
4	import base64	1✔
5	import logging	1✔
6	from urllib.request import urlopen	1✔
7
8	from django.conf import settings	1✔
9	from django.core.cache import caches	1✔
10	from django.core.exceptions import SuspiciousOperation	1✔
11
12	from cryptography import x509	1✔
13	from cryptography.exceptions import InvalidSignature	1✔
14	from cryptography.hazmat.primitives import hashes	1✔
15	from cryptography.hazmat.primitives.asymmetric import padding, rsa	1✔
16
17	logger = logging.getLogger("events")	1✔
18
19	NOTIFICATION_HASH_FORMAT = """\	1✔
20	Message
21	{Message}
22	MessageId
23	{MessageId}
24	Subject
25	{Subject}
26	Timestamp
27	{Timestamp}
28	TopicArn
29	{TopicArn}
30	Type
31	{Type}
32	"""
33
34	NOTIFICATION_WITHOUT_SUBJECT_HASH_FORMAT = """\	1✔
35	Message
36	{Message}
37	MessageId
38	{MessageId}
39	Timestamp
40	{Timestamp}
41	TopicArn
42	{TopicArn}
43	Type
44	{Type}
45	"""
46
47	SUBSCRIPTION_HASH_FORMAT = """\	1✔
48	Message
49	{Message}
50	MessageId
51	{MessageId}
52	SubscribeURL
53	{SubscribeURL}
54	Timestamp
55	{Timestamp}
56	Token
57	{Token}
58	TopicArn
59	{TopicArn}
60	Type
61	{Type}
62	"""
63
64	SUPPORTED_SNS_TYPES = [	1✔
65	"SubscriptionConfirmation",
66	"Notification",
67	]
68
69
70	class VerificationFailed(ValueError):	1✔
71	pass	1✔
72
73
74	def verify_from_sns(json_body):	1✔
75	"""
76	Check that the SNS message was signed by the cetificate.
77
78	https://docs.aws.amazon.com/sns/latest/dg/sns-verify-signature-of-message.html
79
80	Only supports SignatureVersion 1. SignatureVersion 2 (SHA256) was added in
81	September 2022, and requires opt-in.
82	"""
83	signing_cert_url = json_body["SigningCertURL"]	1✔
84	pemfile = _grab_keyfile(signing_cert_url)	1✔
85	cert = x509.load_pem_x509_certificate(pemfile)	1✔
86	signature = base64.decodebytes(json_body["Signature"].encode())	1✔
87
88	hash_format = _get_hash_format(json_body)	1✔
89	cert_pubkey = cert.public_key()	1✔
90	if not isinstance(cert_pubkey, rsa.RSAPublicKey):	1!
NEW 91	raise VerificationFailed(f"SigningCertURL {signing_cert_url} is not an RSA key")	×
92
93	try:	1✔
94	cert_pubkey.verify(	1✔
95	signature,
96	hash_format.format(**json_body).encode(),
97	padding.PKCS1v15(),
98	hashes.SHA1(), # noqa: S303 # Use of insecure hash SHA1
99	)
100	except InvalidSignature as e:	1✔
101	raise VerificationFailed(	1✔
102	f"Invalid signature with SigningCertURL {signing_cert_url}"
103	) from e
104
105	return json_body	1✔
106
107
108	def _get_hash_format(json_body):	1✔
109	message_type = json_body["Type"]	1✔
110	if message_type == "Notification":	1✔
111	if "Subject" in json_body.keys():	1✔
112	return NOTIFICATION_HASH_FORMAT	1✔
113	return NOTIFICATION_WITHOUT_SUBJECT_HASH_FORMAT	1✔
114
115	return SUBSCRIPTION_HASH_FORMAT	1✔
116
117
118	def _grab_keyfile(cert_url):	1✔
119	cert_url_origin = f"https://sns.{settings.AWS_REGION}.amazonaws.com/"	1✔
120	if not (cert_url.startswith(cert_url_origin)):	1✔
121	raise SuspiciousOperation(	1✔
122	f'SNS SigningCertURL "{cert_url}" did not start with "{cert_url_origin}"'
123	)
124
125	key_cache = caches[getattr(settings, "AWS_SNS_KEY_CACHE", "default")]	1✔
126
127	pemfile = key_cache.get(cert_url)	1✔
128	if not pemfile:	1✔
129	response = urlopen(cert_url) # noqa: S310 (check for custom scheme)	1✔
130	pemfile = response.read()	1✔
131
132	# Extract the first certificate in the file and confirm it's a valid
133	# PEM certificate
134	certs = x509.load_pem_x509_certificates(pemfile)	1✔
135	# A proper certificate file will contain 1 certificate
136	if len(certs) != 1:	1✔
137	raise VerificationFailed(	1✔
138	f"SigningCertURL {cert_url} has {len(certs)} certificates."
139	)
140	cert_pubkey = certs[0].public_key()	1✔
141	if not isinstance(cert_pubkey, rsa.RSAPublicKey):	1!
NEW UNCOV 142	raise VerificationFailed(f"SigningCertURL {cert_url} is not an RSA key")	×
143
144	key_cache.set(cert_url, pemfile)	1✔
145	return pemfile	1✔

mozilla / fx-private-relay / 8fe25d4b-9da7-428d-ad38-de60a26a02cc

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous