simple-login/app
1
1
Fork 0
mirror of https://github.com/simple-login/app.git synced 2025-10-21 12:47:07 +08:00
Code Issues Projects Releases Packages Wiki Activity

use to_bytes instead of .as_bytes()

Browse source
This commit is contained in:
Son NK 2020-11-09 17:02:10 +01:00
parent 15466903d1
commit 2cc7cb6a37
2 changed files with 16 additions and 18 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
app/email_utils.py
View file

@ -243,7 +243,7 @@ def send_email(
email_domain = SUPPORT_EMAIL[SUPPORT_EMAIL.find("@") + 1 :] email_domain = SUPPORT_EMAIL[SUPPORT_EMAIL.find("@") + 1 :]
add_dkim_signature(msg, email_domain) add_dkim_signature(msg, email_domain)
msg_raw = msg.as_bytes() msg_raw = to_bytes(msg)
if SENDER: if SENDER:
smtp.sendmail(SENDER, to_email, msg_raw) smtp.sendmail(SENDER, to_email, msg_raw)
else: else:
@ -346,7 +346,7 @@ def add_dkim_signature(msg: Message, email_domain: str):
# Specify headers in "byte" form # Specify headers in "byte" form
# Generate message signature # Generate message signature
sig = dkim.sign( sig = dkim.sign(
msg.as_bytes(), to_bytes(msg),
DKIM_SELECTOR, DKIM_SELECTOR,
email_domain.encode(), email_domain.encode(),
DKIM_PRIVATE_KEY.encode(), DKIM_PRIVATE_KEY.encode(),
@ -622,7 +622,7 @@ def parseaddr_unicode(addr) -> (str, str):
def copy(msg: Message) -> Message: def copy(msg: Message) -> Message:
"""return a copy of message""" """return a copy of message"""
return email.message_from_bytes(msg.as_bytes()) return email.message_from_bytes(to_bytes(msg))
def to_bytes(msg: Message): def to_bytes(msg: Message):

28
email_handler.py
View file

@ -442,7 +442,7 @@ def prepare_pgp_message(
# encrypt # encrypt
# use pgpy as fallback # use pgpy as fallback
msg_bytes = clone_msg.as_bytes() msg_bytes = to_bytes(clone_msg)
try: try:
encrypted_data = pgp_utils.encrypt_file(BytesIO(msg_bytes), pgp_fingerprint) encrypted_data = pgp_utils.encrypt_file(BytesIO(msg_bytes), pgp_fingerprint)
second.set_payload(encrypted_data) second.set_payload(encrypted_data)
@ -468,11 +468,11 @@ def sign_msg(msg: Message) -> Message:
signature.add_header("Content-Disposition", 'attachment; filename="signature.asc"') signature.add_header("Content-Disposition", 'attachment; filename="signature.asc"')
try: try:
signature.set_payload(sign_data(msg.as_bytes().replace(b"\n", b"\r\n"))) signature.set_payload(sign_data(to_bytes(msg).replace(b"\n", b"\r\n")))
except Exception: except Exception:
LOG.exception("Cannot sign, try using pgpy") LOG.exception("Cannot sign, try using pgpy")
signature.set_payload( signature.set_payload(
sign_data_with_pgpy(msg.as_bytes().replace(b"\n", b"\r\n")) sign_data_with_pgpy(to_bytes(msg).replace(b"\n", b"\r\n"))
) )
container.attach(signature) container.attach(signature)
@ -484,7 +484,7 @@ def handle_email_sent_to_ourself(alias, mailbox, msg: Message, user):
# store the refused email # store the refused email
random_name = str(uuid.uuid4()) random_name = str(uuid.uuid4())
full_report_path = f"refused-emails/cycle-{random_name}.eml" full_report_path = f"refused-emails/cycle-{random_name}.eml"
s3.upload_email_from_bytesio(full_report_path, BytesIO(msg.as_bytes()), random_name) s3.upload_email_from_bytesio(full_report_path, BytesIO(to_bytes(msg)), random_name)
refused_email = RefusedEmail.create( refused_email = RefusedEmail.create(
path=None, full_report_path=full_report_path, user_id=alias.user_id path=None, full_report_path=full_report_path, user_id=alias.user_id
) )
@ -559,7 +559,7 @@ def handle_forward(envelope, msg: Message, rcpt_to: str) -> List[Tuple[bool, str
# save the data for debugging # save the data for debugging
file_path = f"/tmp/{random_string(10)}.eml" file_path = f"/tmp/{random_string(10)}.eml"
with open(file_path, "wb") as f: with open(file_path, "wb") as f:
f.write(msg.as_bytes()) f.write(to_bytes(msg))
LOG.exception( LOG.exception(
"Cannot create contact for %s %s %s %s", "Cannot create contact for %s %s %s %s",
@ -1183,7 +1183,7 @@ def handle_bounce(contact: Contact, alias: Alias, msg: Message, user: User):
random_name = str(uuid.uuid4()) random_name = str(uuid.uuid4())
full_report_path = f"refused-emails/full-{random_name}.eml" full_report_path = f"refused-emails/full-{random_name}.eml"
s3.upload_email_from_bytesio(full_report_path, BytesIO(msg.as_bytes()), random_name) s3.upload_email_from_bytesio(full_report_path, BytesIO(to_bytes(msg)), random_name)
file_path = None file_path = None
mailbox = None mailbox = None
@ -1201,9 +1201,7 @@ def handle_bounce(contact: Contact, alias: Alias, msg: Message, user: User):
) )
else: else:
file_path = f"refused-emails/{random_name}.eml" file_path = f"refused-emails/{random_name}.eml"
s3.upload_email_from_bytesio( s3.upload_email_from_bytesio(file_path, BytesIO(to_bytes(msg)), random_name)
file_path, BytesIO(orig_msg.as_bytes()), random_name
)
try: try:
mailbox_id = int(orig_msg[_MAILBOX_ID_HEADER]) mailbox_id = int(orig_msg[_MAILBOX_ID_HEADER])
except TypeError: except TypeError:
@ -1372,7 +1370,7 @@ def handle_bounce_reply_phase(alias: Alias, msg: Message, user: User):
# save the data for debugging # save the data for debugging
file_path = f"/tmp/{random_string(10)}.eml" file_path = f"/tmp/{random_string(10)}.eml"
with open(file_path, "wb") as f: with open(file_path, "wb") as f:
f.write(msg.as_bytes()) f.write(to_bytes(msg))
LOG.exception( LOG.exception(
"Cannot get email-log-id from bounced report, %s %s %s", "Cannot get email-log-id from bounced report, %s %s %s",
@ -1391,14 +1389,14 @@ def handle_bounce_reply_phase(alias: Alias, msg: Message, user: User):
random_name = str(uuid.uuid4()) random_name = str(uuid.uuid4())
full_report_path = f"refused-emails/full-{random_name}.eml" full_report_path = f"refused-emails/full-{random_name}.eml"
s3.upload_email_from_bytesio(full_report_path, BytesIO(msg.as_bytes()), random_name) s3.upload_email_from_bytesio(full_report_path, BytesIO(to_bytes(msg)), random_name)
orig_msg = get_orig_message_from_bounce(msg) orig_msg = get_orig_message_from_bounce(msg)
file_path = None file_path = None
if orig_msg: if orig_msg:
file_path = f"refused-emails/{random_name}.eml" file_path = f"refused-emails/{random_name}.eml"
s3.upload_email_from_bytesio( s3.upload_email_from_bytesio(
file_path, BytesIO(orig_msg.as_bytes()), random_name file_path, BytesIO(to_bytes(orig_msg)), random_name
) )
refused_email = RefusedEmail.create( refused_email = RefusedEmail.create(
@ -1468,13 +1466,13 @@ def handle_spam(
random_name = str(uuid.uuid4()) random_name = str(uuid.uuid4())
full_report_path = f"spams/full-{random_name}.eml" full_report_path = f"spams/full-{random_name}.eml"
s3.upload_email_from_bytesio(full_report_path, BytesIO(msg.as_bytes()), random_name) s3.upload_email_from_bytesio(full_report_path, BytesIO(to_bytes(msg)), random_name)
file_path = None file_path = None
if orig_msg: if orig_msg:
file_path = f"spams/{random_name}.eml" file_path = f"spams/{random_name}.eml"
s3.upload_email_from_bytesio( s3.upload_email_from_bytesio(
file_path, BytesIO(orig_msg.as_bytes()), random_name file_path, BytesIO(to_bytes(orig_msg)), random_name
) )
refused_email = RefusedEmail.create( refused_email = RefusedEmail.create(
@ -1782,7 +1780,7 @@ def sl_sendmail(from_addr, to_addr, msg: Message, mail_options, rcpt_options):
smtp.sendmail( smtp.sendmail(
from_addr, from_addr,
to_addr, to_addr,
msg.as_bytes(), to_bytes(msg),
mail_options, mail_options,
rcpt_options, rcpt_options,
) )