beenull/simple-login
1
0
Fork 0
Code Issues Pull requests Packages Projects Releases Wiki Activity

catch all exception in to_bytes

Browse source
This commit is contained in:
Son 2022-04-22 10:20:43 +02:00
parent 0f7ccec51a
commit 68ec159d91
1 changed files with 13 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
app/email_utils.py
View file

@ -8,9 +8,6 @@ import random
import time import time
import uuid import uuid
from copy import deepcopy from copy import deepcopy
from aiosmtpd.smtp import Envelope
from email import policy, message_from_bytes, message_from_string from email import policy, message_from_bytes, message_from_string
from email.header import decode_header, Header from email.header import decode_header, Header
from email.message import Message, EmailMessage from email.message import Message, EmailMessage
@ -25,6 +22,7 @@ import dkim
import newrelic.agent import newrelic.agent
import re2 as re import re2 as re
import spf import spf
from aiosmtpd.smtp import Envelope
from cachetools import cached, TTLCache from cachetools import cached, TTLCache
from email_validator import ( from email_validator import (
validate_email, validate_email,
@ -839,24 +837,19 @@ def copy(msg: Message) -> Message:
def to_bytes(msg: Message): def to_bytes(msg: Message):
"""replace Message.as_bytes() method by trying different policies""" """replace Message.as_bytes() method by trying different policies"""
try: for generator_policy in [None, policy.SMTP, policy.SMTPUTF8]:
return msg.as_bytes()
except (UnicodeEncodeError, AttributeError):
LOG.w("as_bytes fails with default policy, try SMTP policy")
try: try:
return msg.as_bytes(policy=policy.SMTP) return msg.as_bytes(policy=generator_policy)
except (UnicodeEncodeError, AttributeError): except:
LOG.w("as_bytes fails with SMTP policy, try SMTPUTF8 policy") LOG.w("as_bytes() fails with %s policy", policy, exc_info=True)
try:
return msg.as_bytes(policy=policy.SMTPUTF8) msg_string = msg.as_string()
except (UnicodeEncodeError, AttributeError): try:
LOG.w("as_bytes fails with SMTPUTF8 policy, try converting to string") return msg_string.encode()
msg_string = msg.as_string() except:
try: LOG.w("as_string().encode() fails", exc_info=True)
return msg_string.encode()
except (UnicodeEncodeError, AttributeError) as e: return msg_string.encode(errors="replace")
LOG.w("can't encode msg, err:%s", e)
return msg_string.encode(errors="replace")
def should_add_dkim_signature(domain: str) -> bool: def should_add_dkim_signature(domain: str) -> bool: