mirror of
https://github.com/jwinterm/modoboa.git
synced 2026-11-13 09:04:04 +01:00
More improvements
This commit is contained in:
@@ -18,7 +18,7 @@ from django.utils.html import conditional_escape
|
||||
from django.utils.translation import gettext as _
|
||||
|
||||
from modoboa.lib import u2u_decode
|
||||
from modoboa.lib.email_utils import Email, EmailAddress
|
||||
from modoboa.lib.email_utils import Email
|
||||
|
||||
from . import imapheader
|
||||
from .attachments import get_storage_path
|
||||
@@ -39,7 +39,7 @@ class ImapEmail(Email):
|
||||
("Subject", True),
|
||||
]
|
||||
|
||||
def __init__(self, request, *args, **kwargs):
|
||||
def __init__(self, request, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
self.request = request
|
||||
self.imapc = get_imapconnector(request)
|
||||
@@ -160,7 +160,9 @@ class ImapEmail(Email):
|
||||
@property
|
||||
def source(self):
|
||||
"""Retrieve email source."""
|
||||
return self.imapc.fetchmail(self.mbox, self.mailid, what="source")["BODY[]"]
|
||||
return self.imapc.fetchmail(
|
||||
self.mbox, self.mailid, readonly=True, what="source"
|
||||
)["BODY[]"]
|
||||
|
||||
def _find_content_charset(self, part):
|
||||
for pos, elem in enumerate(part["params"]):
|
||||
@@ -226,11 +228,11 @@ class ImapEmail(Email):
|
||||
class Modifier(ImapEmail):
|
||||
"""Message modifier."""
|
||||
|
||||
def __init__(self, form, request, *args, **kwargs):
|
||||
def __init__(self, request, *args, **kwargs):
|
||||
kwargs["dformat"] = request.user.parameters.get_value("editor")
|
||||
super().__init__(request, *args, **kwargs)
|
||||
self.form = form
|
||||
self.fetch_headers(raw_addresses=True)
|
||||
self._inject_textheader()
|
||||
getattr(self, f"_modify_{self.dformat}")()
|
||||
|
||||
def _modify_plain(self):
|
||||
@@ -255,32 +257,36 @@ class ReplyModifier(Modifier):
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
self.textheader = f"{self.Form} {_('wrote:')}"
|
||||
if self.dformat == "html":
|
||||
self.textheader = f"<p>{self.textheader}</p>"
|
||||
if hasattr(self, "Message_ID"):
|
||||
self.form.fields["origmsgid"].initial = self.Message_ID
|
||||
if not hasattr(self, "Reply_To"):
|
||||
self.form.fields["to"].initial = self.original_From
|
||||
else:
|
||||
self.form.fields["to"].initial = self.original_ReplyTo
|
||||
if self.request.GET.get("all", "0") == "1": # reply-all
|
||||
self.form.fields["cc"].initial = ""
|
||||
toparse = self.To.split(",")
|
||||
if hasattr(self, "Cc"):
|
||||
toparse += self.Cc.split(",")
|
||||
for addr in toparse:
|
||||
tmp = EmailAddress(addr)
|
||||
if tmp.address and tmp.address == self.request.user.username:
|
||||
continue
|
||||
if self.form.fields["cc"].initial != "":
|
||||
self.form.fields["cc"].initial += ", "
|
||||
self.form.fields["cc"].initial += tmp.fulladdress
|
||||
# if hasattr(self, "Message_ID"):
|
||||
# self.form.fields["origmsgid"].initial = self.Message_ID
|
||||
# if not hasattr(self, "Reply_To"):
|
||||
# self.form.fields["to"].initial = self.original_From
|
||||
# else:
|
||||
# self.form.fields["to"].initial = self.original_ReplyTo
|
||||
# if self.request.GET.get("all", "0") == "1": # reply-all
|
||||
# self.form.fields["cc"].initial = ""
|
||||
# toparse = self.To.split(",")
|
||||
# if hasattr(self, "Cc"):
|
||||
# toparse += self.Cc.split(",")
|
||||
# for addr in toparse:
|
||||
# tmp = EmailAddress(addr)
|
||||
# if tmp.address and tmp.address == self.request.user.username:
|
||||
# continue
|
||||
# if self.form.fields["cc"].initial != "":
|
||||
# self.form.fields["cc"].initial += ", "
|
||||
# self.form.fields["cc"].initial += tmp.fulladdress
|
||||
m = re.match(r"re\s*:\s*.+", self.subject.lower())
|
||||
if m:
|
||||
self.form.fields["subject"].initial = self.subject
|
||||
if not m:
|
||||
self.subject = f"Re: {self.subject}"
|
||||
|
||||
def _inject_textheader(self):
|
||||
sender = self.From.get("name", self.From["address"])
|
||||
textheader = f"{sender} {_('wrote:')}"
|
||||
if self.dformat == "html":
|
||||
textheader = f"<p>{textheader}</p>"
|
||||
else:
|
||||
self.form.fields["subject"].initial = f"Re: {self.subject}"
|
||||
textheader = f"{textheader}\n"
|
||||
self.body = textheader + self.body
|
||||
|
||||
def _modify_plain(self):
|
||||
super()._modify_plain()
|
||||
|
||||
@@ -376,6 +376,7 @@ class IMAPconnector:
|
||||
name = self._encode_mbox_name(name)
|
||||
if readonly:
|
||||
self._cmd("EXAMINE", name)
|
||||
self.m.is_readonly = True
|
||||
else:
|
||||
self._cmd("SELECT", name)
|
||||
self.m.state = "SELECTED"
|
||||
|
||||
+145
-28
@@ -1,15 +1,139 @@
|
||||
from email.header import Header
|
||||
from email.mime.image import MIMEImage
|
||||
import os
|
||||
import pkg_resources
|
||||
import smtplib
|
||||
from typing import Optional, Tuple
|
||||
from urllib.parse import unquote, urlparse
|
||||
|
||||
import lxml
|
||||
|
||||
from django.conf import settings
|
||||
from django.core import mail
|
||||
from django.template.loader import render_to_string
|
||||
from django.core.mail import EmailMessage, EmailMultiAlternatives
|
||||
|
||||
from modoboa.lib.cryptutils import get_password
|
||||
from modoboa.core import models as core_models
|
||||
from modoboa.parameters import tools as param_tools
|
||||
|
||||
from . import get_imapconnector, clean_attachments
|
||||
from . import get_imapconnector
|
||||
|
||||
|
||||
def send_mail(request, form, posturl=None):
|
||||
def html2plaintext(content: str) -> str:
|
||||
"""HTML to plain text translation.
|
||||
|
||||
:param content: some HTML content
|
||||
"""
|
||||
if not content:
|
||||
return ""
|
||||
html = lxml.html.fromstring(content)
|
||||
plaintext = ""
|
||||
for ch in html.iter():
|
||||
p = None
|
||||
if ch.text is not None:
|
||||
p = ch.text.strip("\r\t\n")
|
||||
if ch.tag == "img":
|
||||
p = ch.get("alt")
|
||||
if p is None:
|
||||
continue
|
||||
plaintext += p + "\n"
|
||||
return plaintext
|
||||
|
||||
|
||||
def make_body_images_inline(body: str) -> Tuple[str, list]:
|
||||
"""Look for images inside the body and make them inline.
|
||||
|
||||
Before sending a message in HTML format, it is necessary to find
|
||||
all img tags contained in the body in order to rewrite them. For
|
||||
example, icons provided by CKeditor are stored on the server
|
||||
filesystem and not accessible from the outside. We must embark
|
||||
them as parts of the MIME message if we want recipients to
|
||||
display them correctly.
|
||||
|
||||
:param body: the HTML body to parse
|
||||
"""
|
||||
html = lxml.html.fromstring(body)
|
||||
parts = []
|
||||
for tag in html.iter("img"):
|
||||
src = tag.get("src")
|
||||
if src is None:
|
||||
continue
|
||||
o = urlparse(src)
|
||||
path = unquote(os.path.join(settings.BASE_DIR, o.path[1:]))
|
||||
if not os.path.exists(path):
|
||||
continue
|
||||
fname = os.path.basename(path)
|
||||
cid = f"{os.path.splitext(fname)[0]}@modoboa"
|
||||
tag.set("src", f"cid:{cid}")
|
||||
with open(path, "rb") as fp:
|
||||
part = MIMEImage(fp.read())
|
||||
part["Content-ID"] = f"<{cid}>"
|
||||
part.replace_header("Content-Type", f'{part["Content-Type"]}; name="{fname}"')
|
||||
part["Content-Disposition"] = "inline"
|
||||
parts.append(part)
|
||||
return lxml.html.tostring(html, encoding="unicode"), parts
|
||||
|
||||
|
||||
def html_msg(attributes: dict) -> EmailMultiAlternatives:
|
||||
"""Create a multipart message.
|
||||
|
||||
We attach two alternatives:
|
||||
* text/html
|
||||
* text/plain
|
||||
"""
|
||||
body = attributes.get("body")
|
||||
if body:
|
||||
tbody = html2plaintext(body)
|
||||
body, images = make_body_images_inline(body)
|
||||
else:
|
||||
tbody = ""
|
||||
images = []
|
||||
msg = EmailMultiAlternatives()
|
||||
msg.body = tbody
|
||||
msg.attach_alternative(body, "text/html")
|
||||
for img in images:
|
||||
msg.attach(img)
|
||||
return msg
|
||||
|
||||
|
||||
def plain_msg(attributes: dict) -> EmailMessage:
|
||||
"""Create a simple text message."""
|
||||
msg = EmailMessage()
|
||||
msg.body = attributes.get("body", "")
|
||||
return msg
|
||||
|
||||
|
||||
def format_sender_address(user: core_models.User, address: str) -> str:
|
||||
"""Format address before message is sent."""
|
||||
if user.first_name != "" or user.last_name != "":
|
||||
return f'"{Header(user.fullname, "utf8")}" <{address}>'
|
||||
return address
|
||||
|
||||
|
||||
def create_message(user: core_models.User, attributes: dict):
|
||||
headers = {
|
||||
"User-Agent": "Modoboa {}".format(
|
||||
pkg_resources.get_distribution("modoboa").version
|
||||
)
|
||||
}
|
||||
origmsgid = attributes.get("origmsgid")
|
||||
if origmsgid:
|
||||
headers.update({"References": origmsgid, "In-Reply-To": origmsgid})
|
||||
mode = user.parameters.get_value("editor")
|
||||
sender = format_sender_address(user, attributes["sender"])
|
||||
if mode == "html":
|
||||
msg = html_msg(attributes)
|
||||
else:
|
||||
msg = plain_msg(attributes)
|
||||
msg.from_email = sender
|
||||
msg.to = attributes["to"]
|
||||
msg.headers = headers
|
||||
for hdr in ["subject", "cc", "bcc"]:
|
||||
if hdr in attributes:
|
||||
setattr(msg, hdr, attributes[hdr])
|
||||
return msg
|
||||
|
||||
|
||||
def send_mail(request, attributes: dict) -> Tuple[bool, Optional[str]]:
|
||||
"""Email verification and sending.
|
||||
|
||||
If the form does not present any error, a new MIME message is
|
||||
@@ -20,47 +144,40 @@ def send_mail(request, form, posturl=None):
|
||||
:param posturl: the url to post the message form to
|
||||
:return: a 2-uple (True|False, HttpResponse)
|
||||
"""
|
||||
if not form.is_valid():
|
||||
editormode = request.user.parameters.get_value("editor")
|
||||
listing = render_to_string(
|
||||
"modoboa_webmail/compose.html",
|
||||
{
|
||||
"form": form,
|
||||
"noerrors": True,
|
||||
"body": form.cleaned_data.get("body", "").strip(),
|
||||
"posturl": posturl,
|
||||
},
|
||||
request,
|
||||
)
|
||||
return False, {"status": "ko", "listing": listing, "editor": editormode}
|
||||
|
||||
msg = form.to_msg(request)
|
||||
conf = dict(param_tools.get_global_parameters("modoboa_webmail"))
|
||||
msg = create_message(request.user, attributes)
|
||||
conf = dict(param_tools.get_global_parameters("webmail"))
|
||||
options = {"host": conf["smtp_server"], "port": conf["smtp_port"]}
|
||||
if conf["smtp_secured_mode"] == "ssl":
|
||||
options.update({"use_ssl": True})
|
||||
elif conf["smtp_secured_mode"] == "starttls":
|
||||
options.update({"use_tls": True})
|
||||
if conf["smtp_authentication"]:
|
||||
options.update(
|
||||
{"username": request.user.username, "password": get_password(request)}
|
||||
)
|
||||
if not settings.WEBMAIL_DEV_MODE:
|
||||
# TODO: oauth2 auth
|
||||
pass
|
||||
else:
|
||||
options.update(
|
||||
{
|
||||
"username": settings.WEBMAIL_DEV_USERNAME,
|
||||
"password": settings.WEBMAIL_DEV_PASSWORD,
|
||||
}
|
||||
)
|
||||
try:
|
||||
with mail.get_connection(**options) as connection:
|
||||
msg.connection = connection
|
||||
msg.send()
|
||||
except smtplib.SMTPResponseException as inst:
|
||||
return False, {"status": "ko", "error": inst.smtp_error}
|
||||
return False, str(inst.smtp_error)
|
||||
except smtplib.SMTPRecipientsRefused as inst:
|
||||
error = ", ".join(
|
||||
[f"{rcpt}: {error}" for rcpt, error in inst.recipients.items()]
|
||||
)
|
||||
return False, {"status": "ko", "error": error}
|
||||
return False, error
|
||||
|
||||
# Copy message to sent folder
|
||||
sentfolder = request.user.parameters.get_value("sent_folder")
|
||||
get_imapconnector(request).push_mail(sentfolder, msg.message())
|
||||
clean_attachments(request.session["compose_mail"]["attachments"])
|
||||
del request.session["compose_mail"]
|
||||
# clean_attachments(request.session["compose_mail"]["attachments"])
|
||||
# del request.session["compose_mail"]
|
||||
|
||||
return True, {}
|
||||
return True, None
|
||||
|
||||
Reference in New Issue
Block a user