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

Better fix. It think it should cover all cases now.

Browse Source 
see #298
This commit is contained in:
Antoine Nguyen
2012-09-16 19:18:45 +02:00
parent 5200740ee2
commit f08e3367ff
5 changed files with 115 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modoboa/extensions/amavis/__init__.py
+2 -2
View File
@@ -152,12 +152,12 @@ $(document).ready(function() {
@events.observe("TopNotifications")
def display_requests(user):
from lib import get_nb_requests
from sql_listing import get_wrapper
if parameters.get_admin("USER_CAN_RELEASE") == "yes" \
or user.group == "SimpleUsers":
return []
nbrequests = get_nb_requests(user)
nbrequests = get_wrapper().get_pending_requests(user)
url = reverse("modoboa.extensions.amavis.views.index")
url += "#listing/?viewrequests=1"
modoboa/extensions/amavis/lib.py
-16
View File
@@ -7,7 +7,6 @@ from functools import wraps
from django.core.urlresolvers import reverse
from django.db.models import Q
from modoboa.lib import parameters
from modoboa.lib.dbutils import db_type
from models import Msgrcpt
def selfservice(ssfunc=None):
@@ -33,21 +32,6 @@ def selfservice(ssfunc=None):
return wrapped_f
return decorator
def get_nb_requests(user):
"""Return the number of current pending requests
:param user: a ``User`` instance
"""
rq = Q(rs='p')
if not user.is_superuser:
doms = user.get_domains()
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
if db_type() == 'postgres':
return len(Msgrcpt.objects.filter(rq).extra(where=["convert_from(maddr.email, 'UTF8') ~ '%s'" % (regexp,)], tables=['maddr']))
doms_q = Q(rid__email__regex=regexp)
rq &= doms_q
return len(Msgrcpt.objects.filter(rq))
class AMrelease(object):
def __init__(self):
mode = parameters.get_admin("AM_PDP_MODE")
modoboa/extensions/amavis/management/commands/amnotify.py
+2 -2
View File
@@ -12,6 +12,7 @@ from modoboa.lib.emailutils import sendmail_simple
from modoboa.admin.models import User, Domain
from modoboa.extensions.amavis import Amavis
from modoboa.extensions.amavis.models import *
from modoboa.extensions.amavis.sql_listing import get_wrapper
class Command(BaseCommand):
help = 'Amavis notification tool'
@@ -55,8 +56,7 @@ class Command(BaseCommand):
if not da.has_mailbox:
continue
rcpt = da.mailbox_set.all()[0].full_address
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, da.get_domains()))
reqs = Msgrcpt.objects.filter(rs='p', rid__email__regex=regexp)
reqs = get_wrapper().get_domains_pending_requests(da.get_domains())
if len(reqs):
self.send_pr_notification(rcpt, reqs)
modoboa/extensions/amavis/sql_listing.py
+100
View File
@@ -7,6 +7,7 @@ from modoboa.lib import tables
from modoboa.lib.webutils import static_url
from modoboa.lib.email_listing import MBconnector, EmailListing
from modoboa.lib.emailutils import *
from modoboa.lib.dbutils import db_type
from models import *
class Qtable(tables.Table):
@@ -80,6 +81,105 @@ class SQLconnector(MBconnector):
emails.append(m)
return emails
class SQLWrapper(object):
def get_mails(self, request, rcptfilter=None):
if request.GET.get("viewrequests", None) == "1":
q = Q(rs='p')
else:
q = ~Q(rs='D')
if request.user.group == 'SimpleUsers':
q &= Q(rid__email=request.user.email)
else:
if not request.user.is_superuser:
doms = request.user.get_domains()
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
doms_q = Q(rid__email__regex=regexp)
q &= doms_q
if rcptfilter is not None:
q &= Q(rid__email__contains=rcptfilter)
return Msgrcpt.objects.filter(q).values("mail_id")
def get_recipient_message(self, address, mailid):
return Msgrcpt.objects.get(mail=mailid, rid__email=address)
def get_recipient_messages(self, address, mailids):
return Msgrcpt.objects.filter(mail__in=mailids, rid__email=address)
def get_domains_pending_requests(self, domains):
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, domains))
return Msgrcpt.objects.filter(rs='p', rid__email__regex=regexp)
def get_pending_requests(self, user):
"""Return the number of current pending requests
:param user: a ``User`` instance
"""
rq = Q(rs='p')
if not user.is_superuser:
doms = user.get_domains()
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
doms_q = Q(rid__email__regex=regexp)
rq &= doms_q
return len(Msgrcpt.objects.filter(rq))
class PgWrapper(SQLWrapper):
def get_mails(self, request, rcptfilter=None):
if request.GET.get("viewrequests", None) == "1":
q = Q(rs='p')
else:
q = ~Q(rs='D')
if request.user.group == 'SimpleUsers':
msgs = Msgrcpt.objects.filter(q).extra(
where=["convert_from(maddr.email, 'UTF8') = '%s'" % request.user.email],
tables=['maddr']
)
else:
where = []
if not request.user.is_superuser:
doms = request.user.get_domains()
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
where.append("convert_from(maddr.email, 'UTF8') ~ '%s'" % regexp)
if rcptfilter is not None:
where.append("convert_from(maddr.email, 'UTF8') LIKE '%%%s%%'" % rcptfilter)
return Msgrcpt.objects.filter(q).extra(where=where, tables=['maddr']).values("mail_id")
def get_recipient_message(self, address, mailid):
return Msgrcpt.objects.filter(mail=mailid).extra(
where=["convert_from(maddr.email, 'UTF8') = '%s'" % address],
tables=['maddr']
)
def get_recipient_messages(self, address, mailids):
return Msgrcpt.objects.filter(mail__in=mailids).extra(
where=["convert_from(maddr.email, 'UTF8') = '%s'" % address],
tables=['maddr']
)
def get_domains_pending_requests(self, domains):
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, domains))
return Msgrcpt.objects.filter(rs='p').extra(
where=["convert_from(maddr.email, 'UTF8') ~ '%s'" % regexp],
tables=['maddr']
)
def get_pending_requests(self, user):
rq = Q(rs='p')
if not user.is_superuser:
doms = user.get_domains()
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
return len(Msgrcpt.objects.filter(rq).extra(where=["convert_from(maddr.email, 'UTF8') ~ '%s'" % (regexp,)], tables=['maddr']))
return len(Msgrcpt.objects.filter(rq))
def get_wrapper():
if db_type() == 'postgres':
return PgWrapper()
return SQLWrapper()
class SQLlisting(EmailListing):
tpl = "amavis/index.html"
tbltype = Qtable
modoboa/extensions/amavis/views.py
+11 -38
View File
@@ -66,36 +66,7 @@ def _listing(request):
raise Exception("unsupported search criteria %s" % c)
flt = nfilter if flt is None else flt | nfilter
if request.GET.get("viewrequests", None) == "1":
q = Q(rs='p')
else:
q = ~Q(rs='D')
if db_type() != 'postgres':
if request.user.group == 'SimpleUsers':
q &= Q(rid__email=request.user.email)
else:
if not request.user.is_superuser:
doms = request.user.get_domains()
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
doms_q = Q(rid__email__regex=regexp)
q &= doms_q
if rcptfilter is not None:
q &= Q(rid__email__contains=rcptfilter)
msgs = Msgrcpt.objects.filter(q).values("mail_id")
else:
if request.user.group == 'SimpleUsers':
msgs = Msgrcpt.objects.filter(q).extra(where=["convert_from(maddr.email, 'UTF8') = '%s'" % request.user.email], tables=['maddr'])
else:
where = []
if not request.user.is_superuser:
doms = request.user.get_domains()
regexp = "(%s)" % '|'.join(map(lambda dom: dom.name, doms))
where.append("convert_from(maddr.email, 'UTF8') ~ '%s'" % regexp)
if rcptfilter is not None:
where.append("convert_from(maddr.email, 'UTF8') LIKE '%%%s%%'" % rcptfilter)
msgs = Msgrcpt.objects.filter(q).extra(where=where, tables=['maddr']).values("mail_id")
msgs = get_wrapper().get_mails(request, rcptfilter)
if request.GET.has_key("page"):
request.session["page"] = request.GET["page"]
@@ -178,7 +149,7 @@ def viewmail(request, mail_id):
if request.user.has_mailbox:
mb = Mailbox.objects.get(user=request.user)
if not rcpt or rcpt == mb.full_address:
msgrcpt = Msgrcpt.objects.get(mail=mail_id, rid__email=mb.full_address)
msgrcpt = get_wrapper().get_recipient_message(mb.full_address, mail_id)
msgrcpt.rs = 'V'
msgrcpt.save()
@@ -212,7 +183,7 @@ def delete_selfservice(request, mail_id):
if rcpt is None:
raise ModoboaException(_("Invalid request"))
try:
msgrcpt = Msgrcpt.objects.get(mail=mail_id, rid__email=rcpt)
msgrcpt = get_wrapper().get_recipient_message(rcpt, mail_id)
msgrcpt.rs = 'D'
msgrcpt.save()
except Msgrcpt.DoesNotExist:
@@ -224,12 +195,13 @@ def delete(request, mail_id):
mail_id = check_mail_id(request, mail_id)
if request.user.group == 'SimpleUsers':
mb = Mailbox.objects.get(user=request.user)
msgrcpts = Msgrcpt.objects.filter(mail__in=mail_id, rid__email=mb.full_address)
msgrcpts = get_wrapper().get_recipient_messages(mb.full_address, mail_id)
msgrcpts.update(rs='D')
else:
wrapper = get_wrapper()
for mid in mail_id:
r, i = mid.split()
msgrcpt = Msgrcpt.objects.get(mail=i, rid__email=r)
msgrcpt = wrapper.get_recipient_message(r, i)
msgrcpt.rs = 'D'
msgrcpt.save()
@@ -245,7 +217,7 @@ def release_selfservice(request, mail_id):
if rcpt is None or secret_id is None:
raise ModoboaException(_("Invalid request"))
try:
msgrcpt = Msgrcpt.objects.get(mail=mail_id, rid__email=rcpt)
msgrcpt = get_wrapper().get_recipient_message(rcpt, mail_id)
except Msgrcpt.DoesNotExist:
raise ModoboaException(_("Invalid request"))
if secret_id != msgrcpt.mail.secret_id:
@@ -269,7 +241,7 @@ def release(request, mail_id):
mail_id = check_mail_id(request, mail_id)
if request.user.group == 'SimpleUsers':
mb = Mailbox.objects.get(user=request.user)
msgrcpts = Msgrcpt.objects.filter(mail__in=mail_id, rid__email=mb.full_address)
msgrcpts = get_wrapper().get_recipient_messages(mb.full_address, mail_id)
if parameters.get_admin("USER_CAN_RELEASE") == "no":
msgrcpts.update(rs='p')
message = ungettext("%(count)d request sent",
@@ -279,9 +251,10 @@ def release(request, mail_id):
url=__get_current_url(request))
else:
msgrcpts = []
wrapper = get_wrapper()
for mid in mail_id:
r, i = mid.split()
msgrcpts += [Msgrcpt.objects.get(mail=i, rid__email=r)]
msgrcpts += [wrapper.get_recipient_message(r, i)]
amr = AMrelease()
error = None
@@ -319,5 +292,5 @@ def process(request):
@login_required
@user_passes_test(lambda u: u.group != 'SimpleUsers')
def nbrequests(request):
nbrequests = get_nb_requests(request.user)
nbrequests = get_wrapper().get_pending_requests(request.user)
return ajax_simple_response(dict(status="ok", requests=nbrequests))