github-jwinterm/modoboa
1
0
Fork 0
mirror of https://github.com/jwinterm/modoboa.git synced 2026-10-18 05:51:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

Ticket #57 and #59.

Browse Source 
Go back to passwords stored in session. This time, we use pycrypto to encrypt passwords using the AES algorithm.
http://www.dlitz.net/software/pycrypto/

IMAP connections are now refreshed.
This commit is contained in:
tonio
2010-06-10 16:07:00 +02:00
parent 574e012088
commit ace8448efe
7 changed files with 134 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files
extensions/webmail/imap_listing.py → extensions/webmail/lib.py
+49 -9
View File
@@ -122,7 +122,7 @@ class ConnectionsManager(type):
super(ConnectionsManager, cls).__init__(name, bases, dict)
cls.instances = {}
def __call__(cls, *args, **kwargs):
def __call__(cls, **kwargs):
key = None
if kwargs.has_key("user"):
key = kwargs["user"]
@@ -130,10 +130,14 @@ class ConnectionsManager(type):
return None
if not cls.instances.has_key(key):
cls.instances[key] = None
if cls.instances[key] is None or cls.instances[key].invalid:
# I hope python's garbage collector will detect this correclty
if kwargs.has_key("password"):
kwargs["password"] = decrypt(kwargs["password"])
if cls.instances[key] is None:
cls.instances[key] = \
super(ConnectionsManager, cls).__call__(*args, **kwargs)
super(ConnectionsManager, cls).__call__(**kwargs)
else:
cls.instances[key].refresh(key, kwargs["password"])
return cls.instances[key]
class IMAPconnector(object):
@@ -143,11 +147,24 @@ class IMAPconnector(object):
self.criterions = []
self.address = parameters.get_admin("webmail", "IMAP_SERVER")
self.port = int(parameters.get_admin("webmail", "IMAP_PORT"))
self.invalid = False
status, msg = self.login(user, password)
if not status:
raise Exception(msg)
def refresh(self, user, password):
"""Check if current connection needs a refresh
Is it really secure?
"""
if self.m is not None:
try:
self.m.select()
return
except imaplib.IMAP4.error, error:
print error
print self.login(user, password)
def login(self, user, passwd):
import imaplib
try:
@@ -167,11 +184,12 @@ class IMAPconnector(object):
def logout(self):
try:
self.m.close()
self.m.select()
except imaplib.IMAP4.error:
pass
self.m.logout()
self.invalid = True
del self.m
self.m = None
def _encodefolder(self, folder):
if not folder:
@@ -318,9 +336,9 @@ class ImapListing(EmailListing):
defcallback = "wm_updatelisting"
reset_wm_url = False
def __init__(self, user, **kwargs):
def __init__(self, user, password, **kwargs):
self.user = user
self.mbc = IMAPconnector(user=user.username)
self.mbc = IMAPconnector(user=user.username, password=password)
if kwargs.has_key("pattern"):
self.parse_search_parameters(kwargs["criteria"],
kwargs["pattern"])
@@ -414,3 +432,25 @@ class ImapEmail(Email):
"folder" : kwargs["folder"], "mail_id" : kwargs["mail_id"],
"attachments" : attachments != [] and attachments or None
})
def encrypt(clear):
from Crypto.Cipher import AES
import base64
obj = AES.new(parameters.get_admin("webmail", "SECRET_KEY"),
AES.MODE_ECB)
if len(clear) % AES.block_size:
clear += "\0" * (AES.block_size - len(clear) % AES.block_size)
ciph = obj.encrypt(clear)
ciph = base64.b64encode(ciph)
return ciph
def decrypt(ciph):
from Crypto.Cipher import AES
import base64
obj = AES.new(parameters.get_admin("webmail", "SECRET_KEY"),
AES.MODE_ECB)
ciph = base64.b64decode(ciph)
clear = obj.decrypt(ciph)
return clear.rstrip('\0')
extensions/webmail/main.py
+13 -9
View File
@@ -8,6 +8,10 @@ def init():
events.register("UserMenuDisplay", menu)
events.register("UserLogin", userlogin)
events.register("UserLogout", userlogout)
parameters.register_admin("webmail", "SECRET_KEY", type="string",
deflt="abcdefghijklmnop",
help=_("Key used to encrypt/decrypt passwords"))
parameters.register_admin("webmail", "IMAP_SERVER", type="string",
deflt="127.0.0.1",
help=_("Address of your IMAP server"))
@@ -62,17 +66,17 @@ def menu(**kwargs):
]
def userlogin(**kwargs):
from imap_listing import IMAPconnector
from lib import IMAPconnector, encrypt
if kwargs["request"].user.id == 1:
return
kwargs["request"].session["password"] = encrypt(kwargs["password"])
def userlogout(**kwargs):
from lib import IMAPconnector
if kwargs["request"].user.id == 1:
return
m = IMAPconnector(user=kwargs["request"].user.username,
password=kwargs["password"])
def userlogout(**kwargs):
from imap_listing import IMAPconnector
if kwargs["request"].user.id == 1:
return
m = IMAPconnector(user=kwargs["request"].user.username)
password=kwargs["request"].session["password"])
m.logout()
extensions/webmail/templates/compose.html
+1 -2
View File
@@ -43,8 +43,7 @@
{% include "common/genericform.html" %}
</div>
{% if body %}
<textarea id="id_body" name="id_body">
{% if bodyheader %}{{ bodyheader }}{% endif %}
<textarea id="id_body" name="id_body">{% if bodyheader %}{{ bodyheader }}{% endif %}
{{ body }}
</textarea>
{% else %}
extensions/webmail/templates/headers.html
+1 -1
View File
@@ -6,7 +6,7 @@
<tr>
<th>{% trans "Attachments" %}</th>
<td>
{% url extensions.webmail.main.getattachment folder mail_id as test %}
{% url extensions.webmail.views.getattachment folder mail_id as test %}
{% for attachment in attachments %}
<a href="{{ test }}?fname={{ attachment }}">{{ attachment }}</a>
{% endfor %}
extensions/webmail/templates/index.html
+3
View File
@@ -185,6 +185,9 @@ ul.visible {
if (resp.status == "ko") {
wm_updatelisting(resp);
window.fireEvent("resize");
if ($defined(resp.error)) {
infobox.error(resp.error);
}
return;
}
current_anchor.parse_string(resp.url, true).setparams(navparams);
extensions/webmail/templatetags/webextras.py
+4 -2
View File
@@ -3,7 +3,8 @@ from django.template.loader import render_to_string
from django.utils.translation import ugettext as _
from django.core.urlresolvers import reverse
from mailng.extensions import webmail
from mailng.extensions.webmail.imap_listing import IMAPheader
from mailng.extensions.webmail.lib import IMAPheader
from mailng.lib import parameters
register = template.Library()
@@ -85,7 +86,8 @@ def listing_menu(selection, folder, user):
{"name" : "fdaction",
"label" : _("Empty folder"),
"img" : "/static/pics/clear.png",
"url" : folder == "Trash" \
"url" : folder == parameters.get_user(user, "webmail",
"TRASH_FOLDER") \
and reverse(webmail.views.empty, args=[folder]) or ""}
]
}
extensions/webmail/views.py
+63 -41
View File
@@ -10,10 +10,10 @@ from django.core.urlresolvers import reverse
from django.contrib.auth.decorators import login_required
from mailng.admin.models import Mailbox
from mailng.lib import parameters, _render, _render_error, getctx, is_not_localadmin
from imap_listing import *
from mailng.lib.email_listing import parse_search_parameters
from lib import *
from forms import *
from templatetags.webextras import *
from lib.email_listing import parse_search_parameters
def __get_current_url(request):
res = "%s?page=%s" % (request.session["folder"], request.session["page"])
@@ -47,7 +47,7 @@ def folder(request, name, updatenav=True):
optparams["criteria"] = request.session["criteria"]
else:
optparams["reset"] = True
lst = ImapListing(request.user,
lst = ImapListing(request.user, request.session["password"],
baseurl=name, folder=name, order=order, **optparams)
page = lst.paginator.getpage(request.session["page"])
@@ -55,31 +55,33 @@ def folder(request, name, updatenav=True):
content = lst.fetch(request, page.id_start, page.id_stop)
navbar = lst.render_navbar(page)
else:
content = "Empty folder"
content = _("Empty folder")
navbar = ""
folders = render_to_string("webmail/folders.html", {
"folders" : lst.getfolders()
})
ctx = getctx("ok", folders=folders, listing=content, navbar=navbar,
menu=listing_menu("", name, request.user.get_all_permissions()))
menu=listing_menu("", name, request.user))
return HttpResponse(simplejson.dumps(ctx), mimetype="application/json")
@login_required
@is_not_localadmin()
def index(request):
try:
navp = request.session.has_key("navparams") \
and request.session["navparams"] or {}
lst = ImapListing(request.user,
baseurl="INBOX", navparams=navp, folder="INBOX",
empty=True)
except Exception, exp:
return _render_error(request, user_context={"error" : exp})
#try:
navp = request.session.has_key("navparams") \
and request.session["navparams"] or {}
lst = ImapListing(request.user, request.session["password"],
baseurl="INBOX", navparams=navp, folder="INBOX",
empty=True)
#except Exception, exp:
# return _render_error(request, user_context={"error" : exp})
return lst.render(request)
def fetchmail(request, folder, mail_id, all=False):
res = IMAPconnector(user=request.user.username).fetch(start=mail_id,
folder=folder, all=all)
res = IMAPconnector(user=request.user.username,
password=request.session["password"]).fetch(start=mail_id,
folder=folder,
all=all)
if len(res):
return res[0]
return None
@@ -102,7 +104,8 @@ def viewmail(request, folder, mail_id=None):
def getmailcontent(request, folder, mail_id):
msg = fetchmail(request, folder, mail_id, True)
if "class" in msg.keys() and msg["class"] == "unseen":
IMAPconnector(user=request.user.username).msg_read(folder, mail_id)
IMAPconnector(user=request.user.username,
password=request.session["password"]).msg_read(folder, mail_id)
email = ImapEmail(msg, mode="html", links="1")
try:
pageid = request.session["page"]
@@ -141,14 +144,16 @@ def move(request):
for arg in ["msgset", "to"]:
if not request.GET.has_key(arg):
return
mbc = IMAPconnector(user=request.user.username)
mbc = IMAPconnector(user=request.user.username,
password=request.session["password"])
mbc.move(request.GET["msgset"], request.session["folder"], request.GET["to"])
return folder(request, request.session["folder"], False)
@login_required
@is_not_localadmin()
def delete(request, fdname, mail_id):
mbc = IMAPconnector(user=request.user.username)
mbc = IMAPconnector(user=request.user.username,
password=request.session["password"])
mbc.move(mail_id, fdname, parameters.get_user(request.user, "webmail",
"TRASH_FOLDER"))
return folder(request, fdname, False)
@@ -158,7 +163,8 @@ def delete(request, fdname, mail_id):
def mark(request, name):
if not request.GET.has_key("status") or not request.GET.has_key("ids"):
return
mbc = IMAPconnector(user=request.user.username)
mbc = IMAPconnector(user=request.user.username,
password=request.session["password"])
try:
getattr(mbc, "msg_%s" % request.GET["status"])(name, request.GET["ids"])
except AttributeError:
@@ -168,15 +174,17 @@ def mark(request, name):
@login_required
@is_not_localadmin()
def empty(request, name):
if name == "Trash":
mbc = IMAPconnector(user=request.user.username)
if name == parameters.get_user(request.user, "webmail", "TRASH_FOLDER"):
mbc = IMAPconnector(user=request.user.username,
password=request.session["password"])
mbc.empty(name)
return folder(request, name, False)
@login_required
@is_not_localadmin()
def compact(request, name):
mbc = IMAPconnector(user=request.user.username)
mbc = IMAPconnector(user=request.user.username,
password=request.session["password"])
mbc.compact(name)
return folder(request, name, False)
@@ -190,8 +198,10 @@ def render_compose(request, form, posturl, bodyheader=None, body=None):
ctx = getctx("ok", level=2, menu=menu, listing=content)
return HttpResponse(simplejson.dumps(ctx), mimetype="application/json")
def send_mail(request, withctx=False, origmsg=None):
def send_mail(request, withctx=False, origmsg=None, posturl=None):
form = ComposeMailForm(request.POST)
error = None
ctx = None
if form.is_valid():
from email.mime.text import MIMEText
from email.utils import make_msgid, formatdate
@@ -211,24 +221,30 @@ def send_mail(request, withctx=False, origmsg=None):
if "cc" in request.POST.keys():
msg["Cc"] = request.POST["cc"]
rcpts += msg["Cc"].split(",")
error = None
try:
s = smtplib.SMTP(parameters.get_admin("webmail", "SMTP_SERVER"))
if parameters.get_admin("webmail", "SMTP_SECURED") == "yes":
s.starttls()
except (smtplib.SMTPException, ssl.SSLError), error:
print error
# Prévoir la remontée de cette erreur au niveau du client!!
except (smtplib.SMTPException, ssl.SSLError), text:
error = text
if error is None:
if parameters.get_admin("webmail", "SMTP_AUTHENTICATION") == "yes":
s.login(request.user.username, decrypt(request.session["password"]))
s.sendmail(msg['From'], rcpts, msg.as_string())
s.quit()
sentfolder = parameters.get_user(request.user, "webmail", "SENT_FOLDER")
IMAPconnector(user=request.user.username,
password=request.session["password"]).push_mail(sentfolder, msg)
ctx = getctx("ok", url=__get_current_url(request))
if ctx is None:
listing = render_to_string("webmail/compose.html",
{"form" : form,
"body" : request.POST["id_body"].strip(),
"posturl" : posturl})
ctx = getctx("ko", level=2, error=error, listing=listing)
if parameters.get_admin("webmail", "SMTP_AUTHENTICATION") == "yes":
s.login(request.user.username, request.session["password"])
s.sendmail(msg['From'], rcpts, msg.as_string())
s.quit()
sentfolder = parameters.get_user(request.user, "webmail", "SENT_FOLDER")
IMAPconnector(user=request.user.username).push_mail(sentfolder, msg)
ctx = getctx("ok", url=__get_current_url(request))
else:
ctx = getctx("ko", level=2, listing=render_to_string("webmail/compose.html",
{"form" : form}))
if not withctx:
return HttpResponse(simplejson.dumps(ctx), mimetype="application/json")
return ctx, HttpResponse(simplejson.dumps(ctx), mimetype="application/json")
@@ -237,7 +253,7 @@ def send_mail(request, withctx=False, origmsg=None):
@is_not_localadmin()
def compose(request):
if request.method == "POST":
return send_mail(request)
return send_mail(request, posturl=reverse(compose))
form = ComposeMailForm()
form.fields["from_"].initial = request.user.username
@@ -248,9 +264,12 @@ def compose(request):
def reply(request, folder, mail_id):
msg = fetchmail(request, folder, mail_id, True)
if request.method == "POST":
ctx, r = send_mail(request, True, origmsg=msg)
ctx, r = send_mail(request, True, origmsg=msg,
posturl=reverse(reply, args=[folder, mail_id]))
if ctx["status"] == "ok":
IMAPconnector(user=request.user.usernam).msg_answered(folder, mail_id)
IMAPconnector(user=request.user.username,
password=request.session["password"]).msg_answered(folder,
mail_id)
return r
email = ImapEmail(msg, True)
lines = email.body.split('\n')
@@ -290,9 +309,12 @@ def reply(request, folder, mail_id):
@is_not_localadmin()
def forward(request, folder, mail_id):
if request.method == "POST":
ctx, response = send_mail(request, True)
ctx, response = send_mail(request, True,
posturl=reverse(forward, args=[folder, mail_id]))
if ctx["status"] == "ok":
IMAPconnector(user=request.user.username).msgforwarded(folder, mail_id)
IMAPconnector(user=request.user.username,
password=request.session["password"]).msgforwarded(folder,
mail_id)
return response
msg = fetchmail(request, folder, mail_id, True)