60660c4b-c854-4188-a1f1-714f4fd18596

Committed 13 Nov 2024 08:28PM CUT coverage: 84.855% (+0.003%) from 84.852%

Build # 60660c4b-c854-4188-a1f1-714f4fd18596

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #5180 from mozilla/python-312-mpp-3487

MPP-3487: Update to Python 3.12

Run Details

2419 of 3555 branches covered (68.05%)

Branch coverage included in aggregate %.

16855 of 19159 relevant lines covered (87.97%)

9.98 hits per line

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

83.87

/emails/policy.py

"""
Implement Python's email.policy.Policy for Relay

The library provides email.policy.default, which parses and generates RFC-compliant
email headers. However, Relay needs to be able to handle emails with non-compliant
headers as well.

The main change is to use a custom header_factory, which:

* For each header, adds the UnstructuredHeader variant as .as_unstructured
* Handles non-compliant Message-IDs generated by Microsoft alerts

See:
https://docs.python.org/3/library/email.headerregistry.html
https://docs.python.org/3/library/email.policy.html
https://github.com/python/cpython/blob/main/Lib/email/_policybase.py
https://github.com/python/cpython/blob/main/Lib/email/headerregistry.py
https://github.com/python/cpython/blob/main/Lib/email/policy.py
"""

from email import errors
from email._header_value_parser import InvalidMessageID, get_unstructured
from email.headerregistry import BaseHeader, UnstructuredHeader
from email.headerregistry import HeaderRegistry as PythonHeaderRegistry
from email.headerregistry import MessageIDHeader as PythonMessageIDHeader
from email.policy import EmailPolicy
from typing import TYPE_CHECKING, cast

if TYPE_CHECKING:
    # _HeaderParser is a protocol from mypy's typeshed
    # https://github.com/python/typeshed/blob/main/stdlib/email/headerregistry.pyi
    from email.headerregistry import _HeaderParser


class RelayMessageIDHeader(PythonMessageIDHeader):
    """
    Handle an IndexError raised by parsing an invalid Message-ID header.

    This issue is tracked in
    https://github.com/python/cpython/issues/105802

    A fix is unmerged as of October 2023:
    https://github.com/python/cpython/pull/108133
    """

    @classmethod
    def parse(cls, value, kwds):
        try:
            parse_tree = cls.value_parser(value)
        except IndexError:
            token = get_unstructured(value)
            message_id = InvalidMessageID(token)
            message_id.defects.append(
                errors.InvalidHeaderDefect(
                    f"IndexError for invalid msg-id in '{value}'"
                )
            )
            parse_tree = message_id
        kwds["parse_tree"] = parse_tree
        kwds["decoded"] = str(parse_tree)
        kwds["defects"].extend(parse_tree.all_defects)


class RelayHeaderRegistry(PythonHeaderRegistry):
    """Extend the HeaderRegistry to store the unstructured header."""

    def __call__(self, name: str, value: str) -> BaseHeader:
        """Add the unstructured header as .as_unstructured."""
        header_instance = super().__call__(name, value)
        as_unstructured_cls = type(
            "_UnstructuredHeader", (UnstructuredHeader, self.base_class), {}
        )
        as_unstructured = as_unstructured_cls(name, value)
        # Avoid mypy attr-defined error for setting a dynamic attribute
        setattr(header_instance, "as_unstructured", as_unstructured)
        setattr(header_instance, "as_raw", value)
        return header_instance


relay_header_factory = RelayHeaderRegistry()
relay_header_factory.registry["message-id"] = cast(
    type["_HeaderParser"], RelayMessageIDHeader
)

relay_policy = EmailPolicy(header_factory=relay_header_factory)

1	"""
2	Implement Python's email.policy.Policy for Relay
3
4	The library provides email.policy.default, which parses and generates RFC-compliant
5	email headers. However, Relay needs to be able to handle emails with non-compliant
6	headers as well.
7
8	The main change is to use a custom header_factory, which:
9
10	* For each header, adds the UnstructuredHeader variant as .as_unstructured
11	* Handles non-compliant Message-IDs generated by Microsoft alerts
12
13	See:
14	https://docs.python.org/3/library/email.headerregistry.html
15	https://docs.python.org/3/library/email.policy.html
16	https://github.com/python/cpython/blob/main/Lib/email/_policybase.py
17	https://github.com/python/cpython/blob/main/Lib/email/headerregistry.py
18	https://github.com/python/cpython/blob/main/Lib/email/policy.py
19	"""
20
21	from email import errors	1✔
22	from email._header_value_parser import InvalidMessageID, get_unstructured	1✔
23	from email.headerregistry import BaseHeader, UnstructuredHeader	1✔
24	from email.headerregistry import HeaderRegistry as PythonHeaderRegistry	1✔
25	from email.headerregistry import MessageIDHeader as PythonMessageIDHeader	1✔
26	from email.policy import EmailPolicy	1✔
27	from typing import TYPE_CHECKING, cast	1✔
28
29	if TYPE_CHECKING:
30	# _HeaderParser is a protocol from mypy's typeshed
31	# https://github.com/python/typeshed/blob/main/stdlib/email/headerregistry.pyi
32	from email.headerregistry import _HeaderParser
33
34
35	class RelayMessageIDHeader(PythonMessageIDHeader):	1✔
36	"""
37	Handle an IndexError raised by parsing an invalid Message-ID header.
38
39	This issue is tracked in
40	https://github.com/python/cpython/issues/105802
41
42	A fix is unmerged as of October 2023:
43	https://github.com/python/cpython/pull/108133
44	"""
45
46	@classmethod	1✔
47	def parse(cls, value, kwds):	1✔
48	try:	1✔
49	parse_tree = cls.value_parser(value)	1✔
50	except IndexError:	×
51	token = get_unstructured(value)	×
52	message_id = InvalidMessageID(token)	×
53	message_id.defects.append(	×
54	errors.InvalidHeaderDefect(
55	f"IndexError for invalid msg-id in '{value}'"
56	)
57	)
58	parse_tree = message_id	×
59	kwds["parse_tree"] = parse_tree	1✔
60	kwds["decoded"] = str(parse_tree)	1✔
61	kwds["defects"].extend(parse_tree.all_defects)	1✔
62
63
64	class RelayHeaderRegistry(PythonHeaderRegistry):	1✔
65	"""Extend the HeaderRegistry to store the unstructured header."""
66
67	def __call__(self, name: str, value: str) -> BaseHeader:	1✔
68	"""Add the unstructured header as .as_unstructured."""
69	header_instance = super().__call__(name, value)	1✔
70	as_unstructured_cls = type(	1✔
71	"_UnstructuredHeader", (UnstructuredHeader, self.base_class), {}
72	)
73	as_unstructured = as_unstructured_cls(name, value)	1✔
74	# Avoid mypy attr-defined error for setting a dynamic attribute
75	setattr(header_instance, "as_unstructured", as_unstructured)	1✔
76	setattr(header_instance, "as_raw", value)	1✔
77	return header_instance	1✔
78
79
80	relay_header_factory = RelayHeaderRegistry()	1✔
81	relay_header_factory.registry["message-id"] = cast(	1✔
82	type["_HeaderParser"], RelayMessageIDHeader
83	)
84
85	relay_policy = EmailPolicy(header_factory=relay_header_factory)	1✔

mozilla / fx-private-relay / 60660c4b-c854-4188-a1f1-714f4fd18596

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