Use the object_list.total_size to get the total number of Members.
[mailman-postorious.git] / src / postorius / views / list.py
blobd1394d41e0d684ab145cd943632f1d7d9581b68d
1 # -*- coding: utf-8 -*-
2 # Copyright (C) 1998-2021 by the Free Software Foundation, Inc.
4 # This file is part of Postorius.
6 # Postorius is free software: you can redistribute it and/or modify it under
7 # the terms of the GNU General Public License as published by the Free
8 # Software Foundation, either version 3 of the License, or (at your option)
9 # any later version.
11 # Postorius is distributed in the hope that it will be useful, but WITHOUT
12 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 # FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
14 # more details.
16 # You should have received a copy of the GNU General Public License along with
17 # Postorius. If not, see <http://www.gnu.org/licenses/>.
20 import csv
21 import email.utils
22 import logging
23 import sys
24 from urllib.error import HTTPError
26 from django.conf import settings
27 from django.contrib import messages
28 from django.contrib.auth.decorators import login_required
29 from django.core.exceptions import ValidationError
30 from django.core.validators import validate_email
31 from django.forms import formset_factory
32 from django.http import Http404, HttpResponse
33 from django.shortcuts import redirect, render
34 from django.urls import reverse
35 from django.utils.decorators import method_decorator
36 from django.utils.translation import gettext as _
37 from django.views.decorators.http import require_http_methods
39 from allauth.account.models import EmailAddress
40 from django_mailman3.lib.mailman import get_mailman_client, get_mailman_user
41 from django_mailman3.lib.paginator import MailmanPaginator, paginate
42 from django_mailman3.models import MailDomain
43 from django_mailman3.signals import (
44 mailinglist_created, mailinglist_deleted, mailinglist_modified)
46 from postorius.auth.decorators import (
47 list_moderator_required, list_owner_required, superuser_required)
48 from postorius.auth.mixins import ListOwnerMixin
49 from postorius.forms import (
50 AlterMessagesForm, ArchiveSettingsForm, BounceProcessingForm,
51 DigestSettingsForm, DMARCMitigationsForm, ListAnonymousSubscribe,
52 ListAutomaticResponsesForm, ListHeaderMatchForm, ListHeaderMatchFormset,
53 ListIdentityForm, ListMassRemoval, ListMassSubscription, ListNew,
54 ListSubscribe, MemberForm, MemberModeration, MemberPolicyForm,
55 MessageAcceptanceForm, MultipleChoiceForm, UserPreferences)
56 from postorius.forms.list_forms import ACTION_CHOICES, TokenConfirmForm
57 from postorius.models import (
58 Domain, List, Mailman404Error, Style, SubscriptionMode)
59 from postorius.utils import get_member_or_nonmember, set_preferred
60 from postorius.views.generic import MailingListView, bans_view
63 logger = logging.getLogger(__name__)
66 #: DeliveryStatus field values that an admin cannot set.
67 DISABLED_DELIVERY_STATUS_CHOICES_ADMIN = ['by_bounces']
70 class TokenOwner:
71 """Who 'owns' the token returned from the registrar?"""
72 subscriber = 'subscriber'
73 moderator = 'moderator'
76 class ListMembersViews(ListOwnerMixin, MailingListView):
78 # List of allowed roles for the memberships. The string value matches the
79 # exact value Core's REST API expects.
80 allowed_roles = ['owner', 'moderator', 'member', 'nonmember']
82 def _prepare_query(self, request):
83 """Prepare regex based query to search partial email addresses.
85 Core's `members/find` API allows searching for memberships based on
86 regex. This methods prepares a valid regex to pass on the the REST API.
87 """
88 if request.GET.get('q'):
89 query = request.GET['q']
90 if "*" not in query:
91 query = '*{}*'.format(query)
92 else:
93 query = ''
95 return query
97 def get(self, request, list_id, role):
98 """Handle GET for Member view.
100 This includes all the membership roles (self.allowed_roles).
102 member_form = MemberForm()
103 # If the role is misspelled, redirect to the default subscribers.
104 if role not in self.allowed_roles:
105 return redirect('list_members', list_id, 'member')
106 context = dict()
107 context['list'] = self.mailing_list
108 context['role'] = role
109 context['member_form'] = member_form
110 context['page_title'] = _('List {}s'.format(role.capitalize()))
111 context['query'] = self._prepare_query(request)
113 def find_method(count, page):
114 return self.mailing_list.find_members(
115 context['query'], role=role, count=count, page=page)
117 context['members'] = paginate(
118 find_method,
119 request.GET.get('page', 1),
120 request.GET.get('count', 25),
121 paginator_class=MailmanPaginator)
122 context['page_subtitle'] = '({})'.format(
123 context['members'].object_list.total_size)
124 context['form_action'] = _('Add {}'.format(role))
125 if context['query']:
126 context['empty_error'] = _(
127 'No {}s were found matching the search.'.format(role))
128 else:
129 context['empty_error'] = _('List has no {}s'.format(role))
131 return render(request, 'postorius/lists/members.html', context)
133 def _member_post(self, request, role):
134 """Handle POST for members. Unsubscribe all the members selected."""
135 form = MultipleChoiceForm(request.POST)
136 if form.is_valid():
137 members = form.cleaned_data['choices']
138 for member in members:
139 self.mailing_list.unsubscribe(member)
140 messages.success(
141 request,
142 _('The selected members have been unsubscribed'))
143 return redirect('list_members', self.mailing_list.list_id, role)
145 def _non_member_post(self, request, role):
146 """Handle POST for membership roles owner, moderator and non-member.
148 Add memberships if the form is valid otherwise redirect to list_members
149 page with an error message.
151 member_form = MemberForm(request.POST)
152 if member_form.is_valid():
153 try:
154 self.mailing_list.add_role(
155 role=role,
156 address=member_form.cleaned_data['email'],
157 display_name=member_form.cleaned_data['display_name'])
158 messages.success(
159 request,
160 _('{email} has been added with the role {role}'.format(
161 email=member_form.cleaned_data['email'], role=role)))
162 except HTTPError as e:
163 messages.error(request, e.msg)
164 else:
165 messages.error(request, member_form.errors)
166 return redirect('list_members', self.mailing_list.list_id, role)
168 def post(self, request, list_id, role=None):
169 """Handle POST for list members page.
171 List members page have more than one forms, depending on the membership
172 type.
174 - Regular subscribers have a MultipleChoiceForm, which returns a list
175 of emails that needs to be unsubscribed from the MailingList. This is
176 handled by :func:`self._member_post` method.
178 - Owners, moderators and non-members have a MemberForm which allows
179 adding new members with the given roles. This is handled by
180 :func:`self._non_member_post` method.
183 if role not in self.allowed_roles:
184 return redirect('list_members', list_id, 'member')
186 if role in ('member', ):
187 return self._member_post(request, role)
188 else:
189 return self._non_member_post(request, role)
192 @login_required
193 @list_owner_required
194 def list_member_options(request, list_id, email):
195 template_name = 'postorius/lists/memberoptions.html'
196 mm_list = List.objects.get_or_404(fqdn_listname=list_id)
197 # If the role is specified explicitly, use that, otherwise the value is
198 # None and is equivalent to the value not being specified and the lookup
199 # happens using only email. This can cause issues when the email is
200 # subscribed as both a Non-Member and Owner/Moderator returning the wrong
201 # Member object.
202 role = request.GET.get('role')
203 try:
204 mm_member = mm_list.find_members(address=email, role=role)[0]
205 member_prefs = mm_member.preferences
206 except (ValueError, IndexError):
207 raise Http404(_('Member does not exist'))
208 except Mailman404Error:
209 return render(request, template_name, {'nolists': 'true'})
211 initial_moderation = dict([
212 (key, getattr(mm_member, key)) for key in MemberModeration.base_fields
215 if request.method == 'POST':
216 if request.POST.get("formname") == 'preferences':
217 preferences_form = UserPreferences(
218 request.POST, preferences=member_prefs,
219 disabled_delivery_choices=DISABLED_DELIVERY_STATUS_CHOICES_ADMIN) # noqa:E501
220 if preferences_form.is_valid():
221 try:
222 preferences_form.save()
223 except HTTPError as e:
224 messages.error(request, e.msg)
225 else:
226 messages.success(request, _("The member's preferences"
227 " have been updated."))
228 return redirect('list_member_options', list_id, email)
229 elif request.POST.get("formname") == 'moderation':
230 moderation_form = MemberModeration(
231 request.POST, initial=initial_moderation)
232 if moderation_form.is_valid():
233 if not moderation_form.has_changed():
234 messages.info(
235 request, _("No change to the member's moderation."))
236 return redirect('list_member_options', list_id, email)
237 for key in list(moderation_form.fields.keys()):
238 # In general, it would be a very bad idea to loop over the
239 # fields and try to set them one by one, However,
240 # moderation form has only one field.
241 setattr(mm_member, key, moderation_form.cleaned_data[key])
242 try:
243 mm_member.save()
244 except HTTPError as e:
245 messages.error(request, e.msg)
246 else:
247 messages.success(request, _("The member's moderation "
248 "settings have been updated."))
249 return redirect('list_member_options', list_id, email)
250 else:
251 preferences_form = UserPreferences(
252 preferences=member_prefs,
253 disabled_delivery_choices=DISABLED_DELIVERY_STATUS_CHOICES_ADMIN)
254 moderation_form = MemberModeration(initial=initial_moderation)
255 return render(request, template_name, {
256 'mm_member': mm_member,
257 'list': mm_list,
258 'preferences_form': preferences_form,
259 'moderation_form': moderation_form,
263 class ListSummaryView(MailingListView):
264 """Shows common list metrics.
267 def get(self, request, list_id):
268 data = {'list': self.mailing_list,
269 'user_subscribed': False,
270 'subscribed_address': None,
271 'subscribed_preferred': False,
272 'public_archive': False,
273 'hyperkitty_enabled': False}
274 if self.mailing_list.settings['archive_policy'] == 'public':
275 data['public_archive'] = True
276 if (getattr(settings, 'TESTING', False) and # noqa: W504
277 'hyperkitty' not in settings.INSTALLED_APPS):
278 # avoid systematic test failure when HyperKitty is installed
279 # (missing VCR request, see below).
280 list(self.mailing_list.archivers)
281 if ('hyperkitty' in settings.INSTALLED_APPS and # noqa: W504
282 'hyperkitty' in self.mailing_list.archivers and # noqa: W504
283 self.mailing_list.archivers['hyperkitty']):
284 data['hyperkitty_enabled'] = True
285 if request.user.is_authenticated:
286 user_emails = EmailAddress.objects.filter(
287 user=request.user, verified=True).order_by(
288 "email").values_list("email", flat=True)
289 pending_requests = [r['email'] for r in self.mailing_list.requests]
290 for address in user_emails:
291 if address in pending_requests:
292 data['user_request_pending'] = True
293 break
294 try:
295 member = self.mailing_list.get_member(address)
296 except ValueError:
297 pass
298 else:
299 data['user_subscribed'] = True
300 data['subscribed_address'] = address
301 data['subscriber'] = member.member_id
302 data['subscribed_preferred'] = bool(
303 member.subscription_mode ==
304 SubscriptionMode.as_user.name)
305 break # no need to test more addresses
306 mm_user = get_mailman_user(request.user)
307 primary_email = None
308 if mm_user.preferred_address is None:
309 primary_email = set_preferred(request.user, mm_user)
310 else:
311 primary_email = mm_user.preferred_address.email
312 data['subscribe_form'] = ListSubscribe(
313 user_emails, user_id=mm_user.user_id,
314 primary_email=primary_email)
315 else:
316 user_emails = None
317 data['anonymous_subscription_form'] = ListAnonymousSubscribe()
318 return render(request, 'postorius/lists/summary.html', data)
321 class ChangeSubscriptionView(MailingListView):
322 """Change mailing list subscription
325 def _is_subscribed(self, member, subscriber):
326 """Check if the member is same as the subscriber."""
327 return (
328 (member.subscription_mode == SubscriptionMode.as_address.name and
329 member.email == subscriber) or
330 (member.subscription_mode == SubscriptionMode.as_user.name and
331 member.user.user_id == subscriber))
333 @staticmethod
334 def _is_email(subscriber):
335 """Is the subscriber an email address or uuid."""
336 return '@' in subscriber
338 def _change_subscription(self, member, subscriber):
339 """Switch subscriptions to a new email/user.
341 :param member: The currently subscriber Member.
342 :param subscriber: New email or user_id to switch subscription to.
344 # If the Membership is via preferred address and we are switching to an
345 # email or vice versa, we have to do the subscribe-unsubscribe dance.
346 # If the subscription is via an address, then we can simply PATCH the
347 # address in the Member resource.
348 if (member.subscription_mode == SubscriptionMode.as_address.name and
349 self._is_email(subscriber)):
350 member.address = subscriber
351 member.save()
352 return
353 # Keep a copy of the preferences so we can restore them after the
354 # Member switches addresses.
355 if member.preferences:
356 prefs = member.preferences
357 member.unsubscribe()
358 # Unless the subscriber is an an email and it is not-verified, we
359 # should get a Member resource here as response. We should never get to
360 # here with subscriber being an un-verified email.
361 new_member = self.mailing_list.subscribe(
362 subscriber,
363 # Since the email is already verified in Postorius.
364 pre_confirmed=True,
365 # Since this user was already a Member, simply switching Email
366 # addresses shouldn't require another approval.
367 pre_approved=True)
368 self._copy_preferences(new_member.preferences, prefs)
370 def _copy_preferences(self, member_pref, old_member_pref):
371 """Copy the preferences of the old member to new one.
373 :param member_pref: The new member's preference to copy preference to.
374 :param old_member_pref: The old_member's preferences to copy
375 preferences from.
377 # We can't simply switch the Preferences object, so we copy values from
378 # previous one to the new one.
379 for prop in old_member_pref._properties:
380 val = old_member_pref.get(prop)
381 if val:
382 member_pref[prop] = val
383 member_pref.save()
385 @method_decorator(login_required)
386 def post(self, request, list_id):
387 try:
388 user_emails = EmailAddress.objects.filter(
389 user=request.user, verified=True).order_by(
390 "email").values_list("email", flat=True)
391 mm_user = get_mailman_user(request.user)
392 primary_email = None
393 if mm_user.preferred_address is not None:
394 primary_email = mm_user.preferred_address.email
395 form = ListSubscribe(
396 user_emails, mm_user.user_id, primary_email, request.POST)
397 # Find the currently subscribed email
398 old_email = None
399 for address in user_emails:
400 try:
401 member = self.mailing_list.get_member(address)
402 except ValueError:
403 pass
404 else:
405 old_email = address
406 break # no need to test more addresses
407 assert old_email is not None
408 if form.is_valid():
409 subscriber = form.cleaned_data['subscriber']
410 if self._is_subscribed(member, subscriber):
411 messages.error(request, _('You are already subscribed'))
412 else:
413 self._change_subscription(member, subscriber)
414 # If the subscriber is user_id (i.e. not an email
415 # address), Show 'Primary Address ()' in the success
416 # message instead of weird user id.
417 if not self._is_email(subscriber):
418 subscriber = _(
419 'Primary Address ({})').format(primary_email)
420 messages.success(
421 request,
422 _('Subscription changed to %s').format(subscriber))
423 else:
424 messages.error(request,
425 _('Something went wrong. Please try again.'))
426 except HTTPError as e:
427 messages.error(request, e.msg)
428 return redirect('list_summary', self.mailing_list.list_id)
431 class ListSubscribeView(MailingListView):
433 view name: `list_subscribe`
436 @method_decorator(login_required)
437 def post(self, request, list_id):
439 Subscribes an email address to a mailing list via POST and
440 redirects to the `list_summary` view.
442 try:
443 user_emails = EmailAddress.objects.filter(
444 user=request.user, verified=True).order_by(
445 "email").values_list("email", flat=True)
446 mm_user = get_mailman_user(request.user)
447 primary_email = None
448 if mm_user.preferred_address is not None:
449 primary_email = mm_user.preferred_address.email
450 form = ListSubscribe(
451 user_emails, mm_user.user_id, primary_email, request.POST)
452 if form.is_valid():
453 subscriber = request.POST.get('subscriber')
454 display_name = request.POST.get('display_name')
455 response = self.mailing_list.subscribe(
456 subscriber, display_name,
457 pre_verified=True, pre_confirmed=True)
458 if (type(response) == dict and # noqa: W504
459 response.get('token_owner') == TokenOwner.moderator):
460 messages.success(
461 request, _('Your subscription request has been'
462 ' submitted and is waiting for moderator'
463 ' approval.'))
464 else:
465 messages.success(request,
466 _('You are subscribed to %s.') %
467 self.mailing_list.fqdn_listname)
468 else:
469 messages.error(request,
470 _('Something went wrong. Please try again.'))
471 except HTTPError as e:
472 messages.error(request, e.msg)
473 return redirect('list_summary', self.mailing_list.list_id)
476 class ListAnonymousSubscribeView(MailingListView):
478 view name: `list_anonymous_subscribe`
481 def post(self, request, list_id):
483 Subscribes an email address to a mailing list via POST and
484 redirects to the `list_summary` view.
485 This view is used for unauthenticated users and asks Mailman core to
486 verify the supplied email address.
488 try:
489 form = ListAnonymousSubscribe(request.POST)
490 if form.is_valid():
491 email = form.cleaned_data.get('email')
492 display_name = form.cleaned_data.get('display_name')
493 self.mailing_list.subscribe(
494 email, display_name,
495 pre_verified=False, pre_confirmed=False)
496 messages.success(request, _('Please check your inbox for '
497 'further instructions'))
498 else:
499 messages.error(request,
500 _('Something went wrong. Please try again.'))
501 except HTTPError as e:
502 messages.error(request, e.msg)
503 return redirect('list_summary', self.mailing_list.list_id)
506 class ListUnsubscribeView(MailingListView):
508 """Unsubscribe from a mailing list."""
510 @method_decorator(login_required)
511 def post(self, request, *args, **kwargs):
512 email = request.POST['email']
513 if self._has_pending_unsub_req(email):
514 messages.error(
515 request,
516 _('You have a pending unsubscription request waiting for'
517 ' moderator approval.'))
518 return redirect('list_summary', self.mailing_list.list_id)
519 try:
520 response = self.mailing_list.unsubscribe(email)
521 if response is not None and response.get('token') is not None:
522 messages.success(
523 request, _('Your unsubscription request has been'
524 ' submitted and is waiting for moderator'
525 ' approval.'))
526 else:
527 messages.success(request, _('%s has been unsubscribed'
528 ' from this list.') % email)
529 except ValueError as e:
530 messages.error(request, e)
531 return redirect('list_summary', self.mailing_list.list_id)
534 @login_required
535 @list_owner_required
536 def list_mass_subscribe(request, list_id):
537 mailing_list = List.objects.get_or_404(fqdn_listname=list_id)
538 if request.method == 'POST':
539 form = ListMassSubscription(request.POST)
540 if form.is_valid():
541 for data in form.cleaned_data['emails']:
542 try:
543 # Parse the data to get the address and the display name
544 display_name, address = email.utils.parseaddr(data)
545 validate_email(address)
546 p_v = form.cleaned_data['pre_verified']
547 p_c = form.cleaned_data['pre_confirmed']
548 p_a = form.cleaned_data['pre_approved']
549 inv = form.cleaned_data['invitation']
550 mailing_list.subscribe(
551 address=address,
552 display_name=display_name,
553 pre_verified=p_v,
554 pre_confirmed=p_c,
555 pre_approved=p_a,
556 invitation=inv,
557 send_welcome_message=form.cleaned_data[
558 'send_welcome_message'])
559 if inv:
560 message = _('The address %(address)s has been'
561 ' invited to %(list)s.')
562 else:
563 message = _('The address %(address)s has been'
564 ' subscribed to %(list)s.')
565 if not (p_v and p_c and p_a):
566 message += _(' Subscription may be pending address'
567 ' verification, confirmation or'
568 ' approval as appropriate.')
569 messages.success(
570 request, message %
571 {'address': address,
572 'list': mailing_list.fqdn_listname})
573 except HTTPError as e:
574 messages.error(request, e)
575 except ValidationError:
576 messages.error(request, _('The email address %s'
577 ' is not valid.') % address)
578 else:
579 form = ListMassSubscription()
580 return render(request, 'postorius/lists/mass_subscribe.html',
581 {'form': form, 'list': mailing_list})
584 class ListMassRemovalView(MailingListView):
585 """Class For Mass Removal"""
587 @method_decorator(login_required)
588 @method_decorator(list_owner_required)
589 def get(self, request, *args, **kwargs):
590 form = ListMassRemoval()
591 return render(request, 'postorius/lists/mass_removal.html',
592 {'form': form, 'list': self.mailing_list})
594 @method_decorator(list_owner_required)
595 def post(self, request, *args, **kwargs):
596 form = ListMassRemoval(request.POST)
597 if not form.is_valid():
598 messages.error(request, _('Please fill out the form correctly.'))
599 else:
600 valid_emails = []
601 invalid_emails = []
602 for data in form.cleaned_data['emails']:
603 try:
604 # Parse the data to get the address.
605 address = email.utils.parseaddr(data)[1]
606 validate_email(address)
607 valid_emails.append(address)
608 except ValidationError:
609 invalid_emails.append(data)
611 try:
612 self.mailing_list.mass_unsubscribe(valid_emails)
613 messages.success(
614 request, _('These addresses {address} have been'
615 ' unsubscribed from {list}.'.format(
616 address=', '.join(valid_emails),
617 list=self.mailing_list.fqdn_listname)))
618 except (HTTPError, ValueError) as e:
619 messages.error(request, e)
621 if len(invalid_emails) > 0:
622 messages.error(
623 request,
624 _('The email address %s is not valid.') % invalid_emails)
625 return redirect('mass_removal', self.mailing_list.list_id)
628 def _perform_action(message_ids, action):
629 for message_id in message_ids:
630 action(message_id)
633 @login_required
634 @list_moderator_required
635 def list_moderation(request, list_id, held_id=-1):
636 mailing_list = List.objects.get_or_404(fqdn_listname=list_id)
637 if request.method == 'POST':
638 form = MultipleChoiceForm(request.POST)
639 if form.is_valid():
640 message_ids = form.cleaned_data['choices']
641 try:
642 if 'accept' in request.POST:
643 _perform_action(message_ids, mailing_list.accept_message)
644 messages.success(request,
645 _('The selected messages were accepted'))
646 elif 'reject' in request.POST:
647 _perform_action(message_ids, mailing_list.reject_message)
648 messages.success(request,
649 _('The selected messages were rejected'))
650 elif 'discard' in request.POST:
651 _perform_action(message_ids, mailing_list.discard_message)
652 messages.success(request,
653 _('The selected messages were discarded'))
654 except HTTPError:
655 messages.error(request, _('Message could not be found'))
656 else:
657 form = MultipleChoiceForm()
658 held_messages = paginate(
659 mailing_list.get_held_page,
660 request.GET.get('page'), request.GET.get('count'),
661 paginator_class=MailmanPaginator)
662 context = {
663 'list': mailing_list,
664 'held_messages': held_messages,
665 'form': form,
666 'ACTION_CHOICES': ACTION_CHOICES,
668 return render(request, 'postorius/lists/held_messages.html', context)
671 @require_http_methods(['POST'])
672 @login_required
673 @list_moderator_required
674 def moderate_held_message(request, list_id):
675 """Moderate one held message"""
676 mailing_list = List.objects.get_or_404(fqdn_listname=list_id)
677 msg = mailing_list.get_held_message(request.POST['msgid'])
678 moderation_choice = request.POST.get('moderation_choice')
679 reason = request.POST.get('reason')
681 try:
682 if 'accept' in request.POST:
683 mailing_list.accept_message(msg.request_id)
684 messages.success(request, _('The message was accepted'))
685 elif 'reject' in request.POST:
686 mailing_list.reject_message(msg.request_id, reason=reason)
687 messages.success(request, _('The message was rejected'))
688 elif 'discard' in request.POST:
689 mailing_list.discard_message(msg.request_id)
690 messages.success(request, _('The message was discarded'))
691 except HTTPError as e:
692 if e.code == 404:
693 messages.error(
694 request,
695 _('Held message was not found.'))
696 return redirect('list_held_messages', list_id)
697 else:
698 raise
700 moderation_choices = dict(ACTION_CHOICES)
701 if moderation_choice in moderation_choices:
702 member = get_member_or_nonmember(mailing_list, msg.sender)
703 if member is not None:
704 member.moderation_action = moderation_choice
705 member.save()
706 messages.success(
707 request,
708 _('Moderation action for {} set to {}'.format(
709 member, moderation_choices[moderation_choice])))
710 else:
711 messages.error(
712 request,
713 _('Failed to set moderation action for {}'.format(msg.sender)))
714 return redirect('list_held_messages', list_id)
717 @login_required
718 @list_owner_required
719 def csv_view(request, list_id):
720 """Export all the subscriber in csv
722 mm_lists = List.objects.get_or_404(fqdn_listname=list_id)
724 response = HttpResponse(content_type='text/csv')
725 response['Content-Disposition'] = (
726 'attachment; filename="Subscribers.csv"')
728 writer = csv.writer(response)
729 if mm_lists:
730 for i in mm_lists.members:
731 writer.writerow([i.email])
733 return response
736 def _get_choosable_domains(request):
737 domains = Domain.objects.all()
738 return [(d.mail_host, d.mail_host) for d in domains]
741 def _get_choosable_styles(request):
742 styles = Style.objects.all()
743 options = [(style['name'], style['description'])
744 for style in styles['styles']]
745 return options
748 def _get_default_style():
749 return Style.objects.all()['default']
752 @login_required
753 @superuser_required
754 def list_new(request, template='postorius/lists/new.html'):
756 Add a new mailing list.
757 If the request to the function is a GET request an empty form for
758 creating a new list will be displayed. If the request method is
759 POST the form will be evaluated. If the form is considered
760 correct the list gets created and otherwise the form with the data
761 filled in before the last POST request is returned. The user must
762 be logged in to create a new list.
764 mailing_list = None
765 choosable_domains = [('', _('Choose a Domain'))]
766 choosable_domains += _get_choosable_domains(request)
767 choosable_styles = _get_choosable_styles(request)
768 if request.method == 'POST':
769 form = ListNew(choosable_domains, choosable_styles, request.POST)
770 if form.is_valid():
771 # grab domain
772 domain = Domain.objects.get_or_404(
773 mail_host=form.cleaned_data['mail_host'])
774 # creating the list
775 try:
776 mailing_list = domain.create_list(
777 form.cleaned_data['listname'],
778 style_name=form.cleaned_data['list_style'])
779 mailing_list.add_owner(form.cleaned_data['list_owner'])
780 list_settings = mailing_list.settings
781 if form.cleaned_data['description']:
782 list_settings["description"] = \
783 form.cleaned_data['description']
784 list_settings["advertised"] = form.cleaned_data['advertised']
785 list_settings.save()
786 messages.success(request, _("List created"))
787 mailinglist_created.send(sender=List,
788 list_id=mailing_list.list_id)
789 return redirect("list_summary",
790 list_id=mailing_list.list_id)
791 # TODO catch correct Error class:
792 except HTTPError as e:
793 # Right now, there is no good way to detect that this is a
794 # duplicate mailing list request other than checking the
795 # reason for 400 error.
796 if e.reason == 'Mailing list exists':
797 form.add_error(
798 'listname', _('Mailing List already exists.'))
799 return render(request, template, {'form': form})
800 # Otherwise just render the generic error page.
801 return render(request, 'postorius/errors/generic.html',
802 {'error': e.msg})
803 else:
804 messages.error(request, _('Please check the errors below'))
805 else:
806 form = ListNew(choosable_domains, choosable_styles, initial={
807 'list_owner': request.user.email,
808 'advertised': True,
809 'list_style': _get_default_style(),
811 return render(request, template, {'form': form})
814 def _unique_lists(lists):
815 """Return unique lists from a list of mailing lists."""
816 return {mlist.list_id: mlist for mlist in lists}.values()
819 def _get_mail_host(web_host):
820 """Get the mail_host for a web_host if FILTER_VHOST is true and there's
821 only one mail_host for this web_host.
823 if not getattr(settings, 'FILTER_VHOST', False):
824 return None
825 mail_hosts = []
826 use_web_host = False
827 for domain in Domain.objects.all():
828 try:
829 if (MailDomain.objects.get(
830 mail_domain=domain.mail_host).site.domain == web_host):
831 if domain.mail_host not in mail_hosts:
832 mail_hosts.append(domain.mail_host)
833 except MailDomain.DoesNotExist:
834 use_web_host = True
835 if len(mail_hosts) == 1:
836 return mail_hosts[0]
837 elif len(mail_hosts) == 0 and use_web_host:
838 return web_host
839 else:
840 return None
843 @login_required
844 def list_index_authenticated(request):
845 """Index page for authenticated users.
847 Index page for authenticated users is slightly different than
848 un-authenticated ones. Authenticated users will see all their memberships
849 in the index page.
851 This view is not paginated and will show all the lists.
854 role = request.GET.get('role', None)
855 client = get_mailman_client()
856 choosable_domains = _get_choosable_domains(request)
858 # Get all the verified addresses of the user.
859 user_emails = EmailAddress.objects.filter(
860 user=request.user, verified=True).order_by(
861 "email").values_list("email", flat=True)
863 # Get all the mailing lists for the current user.
864 all_lists = []
865 mail_host = _get_mail_host(request.get_host().split(':')[0])
866 for user_email in user_emails:
867 try:
868 all_lists.extend(
869 client.find_lists(user_email,
870 role=role,
871 mail_host=mail_host,
872 count=sys.maxsize))
873 except HTTPError:
874 # No lists exist with the given role for the given user.
875 pass
876 # If the user has no list that they are subscriber/owner/moderator of, we
877 # just redirect them to the index page with all lists.
878 if len(all_lists) == 0 and role is None:
879 return redirect(reverse('list_index') + '?all-lists')
880 # Render the list index page with `check_advertised = False` since we don't
881 # need to check for advertised list given that all the users are related
882 # and know about the existence of the list anyway.
883 context = {
884 'lists': _unique_lists(all_lists),
885 'domain_count': len(choosable_domains),
886 'role': role,
887 'check_advertised': False,
889 return render(
890 request,
891 'postorius/index.html',
892 context
896 def list_index(request, template='postorius/index.html'):
897 """Show a table of all public mailing lists."""
898 # TODO maxking: Figure out why does this view accept POST request and why
899 # can't it be just a GET with list parameter.
900 if request.method == 'POST':
901 return redirect("list_summary", list_id=request.POST["list"])
902 # If the user is logged-in, show them only related lists in the index,
903 # except role is present in requests.GET.
904 if request.user.is_authenticated and 'all-lists' not in request.GET:
905 return list_index_authenticated(request)
907 def _get_list_page(count, page):
908 client = get_mailman_client()
909 advertised = not request.user.is_superuser
910 mail_host = _get_mail_host(request.get_host().split(":")[0])
911 return client.get_list_page(
912 advertised=advertised, mail_host=mail_host, count=count, page=page)
914 lists = paginate(
915 _get_list_page, request.GET.get('page'), request.GET.get('count'),
916 paginator_class=MailmanPaginator)
918 choosable_domains = _get_choosable_domains(request)
920 return render(request, template,
921 {'lists': lists,
922 'check_advertised': True,
923 'all_lists': True,
924 'domain_count': len(choosable_domains)})
927 @login_required
928 @list_owner_required
929 def list_delete(request, list_id):
930 """Deletes a list but asks for confirmation first.
932 the_list = List.objects.get_or_404(fqdn_listname=list_id)
933 if request.method == 'POST':
934 the_list.delete()
935 mailinglist_deleted.send(sender=List, list_id=list_id)
936 return redirect("list_index")
937 else:
938 submit_url = reverse('list_delete',
939 kwargs={'list_id': list_id})
940 cancel_url = reverse('list_index',)
941 return render(request, 'postorius/lists/confirm_delete.html',
942 {'submit_url': submit_url, 'cancel_url': cancel_url,
943 'list': the_list})
946 @login_required
947 @list_moderator_required
948 def list_pending_confirmations(request, list_id):
949 """Shows a list of subscription requests.
951 return _list_subscriptions(
952 request=request,
953 list_id=list_id,
954 token_owner=TokenOwner.subscriber,
955 template='postorius/lists/pending_confirmations.html',
956 page_title=_('Subscriptions pending user confirmation'),
960 @login_required
961 @list_moderator_required
962 def list_subscription_requests(request, list_id):
963 """Shows a list of subscription requests."""
964 return _list_subscriptions(
965 request=request,
966 list_id=list_id,
967 token_owner=TokenOwner.moderator,
968 template='postorius/lists/subscription_requests.html',
969 page_title=_('Subscriptions pending approval'),
973 @login_required
974 @list_moderator_required
975 def list_unsubscription_requests(request, list_id):
976 """Shows a list of pending unsubscription requests."""
977 return _list_subscriptions(
978 request=request,
979 list_id=list_id,
980 token_owner=TokenOwner.moderator,
981 template='postorius/lists/subscription_requests.html',
982 page_title=_('Un-subscriptions pending approval'),
983 request_type='unsubscription'
987 def _list_subscriptions(
988 request, list_id, token_owner, template, page_title,
989 request_type='subscription'):
990 m_list = List.objects.get_or_404(fqdn_listname=list_id)
991 requests = [req
992 for req in m_list.get_requests(
993 token_owner=token_owner, request_type=request_type)]
994 paginated_requests = paginate(
995 requests,
996 request.GET.get('page'),
997 request.GET.get('count', 25))
998 page_subtitle = '(%d)' % len(requests)
999 return render(request, template,
1000 {'list': m_list,
1001 'paginated_requests': paginated_requests,
1002 'page_title': page_title,
1003 'page_subtitle': page_subtitle})
1006 @login_required
1007 @list_moderator_required
1008 @require_http_methods(['POST'])
1009 def handle_subscription_request(request, list_id, request_id):
1011 Handle a subscription request. Possible actions:
1012 - accept
1013 - defer
1014 - reject
1015 - discard
1017 confirmation_messages = {
1018 'accept': _('The request has been accepted.'),
1019 'reject': _('The request has been rejected.'),
1020 'discard': _('The request has been discarded.'),
1021 'defer': _('The request has been defered.'),
1023 # Get the action by the presence of one of the values in POST data.
1024 for each in confirmation_messages:
1025 if each in request.POST:
1026 action = each
1027 # If None of the actions were specified, just raise an error and return.
1028 m_list = List.objects.get_or_404(fqdn_listname=list_id)
1029 if action is None:
1030 messages.error(f'Invalid or missing action {action!r}')
1031 return redirect('list_subscription_requests', m_list.list_id)
1032 reason = request.POST.get('reason')
1033 try:
1034 # Moderate request and add feedback message to session.
1035 m_list.moderate_request(request_id, action, reason)
1036 messages.success(request, confirmation_messages[action])
1037 except HTTPError as e:
1038 if e.code == 409:
1039 messages.success(request,
1040 _('The request was already moderated: %s')
1041 % e.reason)
1042 else:
1043 messages.error(request, _('The request could not be moderated: %s')
1044 % e.reason)
1045 return redirect('list_subscription_requests', m_list.list_id)
1048 SETTINGS_SECTION_NAMES = (
1049 ('list_identity', _('List Identity')),
1050 ('automatic_responses', _('Automatic Responses')),
1051 ('alter_messages', _('Alter Messages')),
1052 ('dmarc_mitigations', _('DMARC Mitigations')),
1053 ('digest', _('Digest')),
1054 ('message_acceptance', _('Message Acceptance')),
1055 ('archiving', _('Archiving')),
1056 ('subscription_policy', _('Member Policy')),
1057 ('bounce_processing', _('Bounce Processing')),
1060 SETTINGS_FORMS = {
1061 'list_identity': ListIdentityForm,
1062 'automatic_responses': ListAutomaticResponsesForm,
1063 'alter_messages': AlterMessagesForm,
1064 'dmarc_mitigations': DMARCMitigationsForm,
1065 'digest': DigestSettingsForm,
1066 'message_acceptance': MessageAcceptanceForm,
1067 'archiving': ArchiveSettingsForm,
1068 'subscription_policy': MemberPolicyForm,
1069 'bounce_processing': BounceProcessingForm,
1073 @login_required
1074 @list_owner_required
1075 def list_settings(request, list_id=None, visible_section=None,
1076 template='postorius/lists/settings.html'):
1078 View and edit the settings of a list.
1079 The function requires the user to be logged in and have the
1080 permissions necessary to perform the action.
1082 Use /<NAMEOFTHESECTION>/<NAMEOFTHEOPTION>
1083 to show only parts of the settings
1084 <param> is optional / is used to differ in between section and option might
1085 result in using //option
1087 if visible_section is None:
1088 visible_section = 'list_identity'
1089 try:
1090 form_class = SETTINGS_FORMS[visible_section]
1091 except KeyError:
1092 raise Http404('No such settings section')
1093 m_list = List.objects.get_or_404(fqdn_listname=list_id)
1094 list_settings = m_list.settings
1095 initial_data = dict((key, value) for key, value in list_settings.items())
1096 # List settings are grouped an processed in different forms.
1097 if request.method == 'POST':
1098 form = form_class(request.POST, mlist=m_list, initial=initial_data)
1099 if form.is_valid():
1100 try:
1101 for key in form.changed_data:
1102 if key in form_class.mlist_properties:
1103 setattr(m_list, key, form.cleaned_data[key])
1104 else:
1105 val = form.cleaned_data[key]
1106 # Empty list isn't really a valid value and Core
1107 # interprets empty string as empty value for
1108 # ListOfStringsField. We are doing it here instead of
1109 # ListOfStringsField.to_python() because output from
1110 # list of strings is expected to be a list and the NULL
1111 # value is hence an empty list. The serialization of
1112 # the empty list is for us empty string, hence we are
1113 # doing this here. Technically, it can be done outside
1114 # of this view, but there aren't any other use cases
1115 # where we'd want an empty list of strings.
1116 if val == []:
1117 val = ''
1118 list_settings[key] = val
1119 list_settings.save()
1120 messages.success(request, _('The settings have been updated.'))
1121 mailinglist_modified.send(sender=List, list_id=m_list.list_id)
1122 except HTTPError as e:
1123 messages.error(
1124 request,
1125 _('An error occurred: ') + e.reason)
1126 return redirect('list_settings', m_list.list_id, visible_section)
1127 else:
1128 form = form_class(initial=initial_data, mlist=m_list)
1130 return render(request, template, {
1131 'form': form,
1132 'section_names': SETTINGS_SECTION_NAMES,
1133 'list': m_list,
1134 'visible_section': visible_section,
1138 @login_required
1139 @list_owner_required
1140 def remove_role(request, list_id=None, role=None, address=None,
1141 template='postorius/lists/confirm_remove_role.html'):
1142 """Removes a list moderator or owner."""
1143 the_list = List.objects.get_or_404(fqdn_listname=list_id)
1144 redirect_on_success = redirect('list_members', the_list.list_id, role)
1145 roster = getattr(the_list, '{}s'.format(role))
1146 all_emails = [each.email for each in roster]
1147 if address not in all_emails:
1148 messages.error(request,
1149 _('The user %(email)s is not in the %(role)s group')
1150 % {'email': address, 'role': role})
1151 return redirect('list_members', the_list.list_id, role)
1153 if role == 'owner':
1154 if len(roster) == 1:
1155 messages.error(request, _('Removing the last owner is impossible'))
1156 return redirect('list_members', the_list.list_id, role)
1157 user_emails = EmailAddress.objects.filter(
1158 user=request.user, verified=True).order_by(
1159 "email").values_list("email", flat=True)
1160 if address in user_emails:
1161 # The user is removing themselves, redirect to the list info page
1162 # because they won't have access to the members page anyway.
1163 redirect_on_success = redirect('list_summary', the_list.list_id)
1165 if request.method == 'POST':
1166 try:
1167 the_list.remove_role(role, address)
1168 except HTTPError as e:
1169 messages.error(request, _('The user could not be removed: %(msg)s')
1170 % {'msg': e.msg})
1171 return redirect('list_members', the_list.list_id, role)
1172 messages.success(request, _('The user %(address)s has been removed'
1173 ' from the %(role)s group.')
1174 % {'address': address, 'role': role})
1175 return redirect_on_success
1176 return render(request, template,
1177 {'role': role, 'address': address,
1178 'list_id': the_list.list_id})
1181 @login_required
1182 @list_owner_required
1183 def remove_all_subscribers(request, list_id):
1184 """Empty the list by unsubscribing all members."""
1186 mlist = List.objects.get_or_404(fqdn_listname=list_id)
1187 if len(mlist.members) == 0:
1188 messages.error(request,
1189 _('No member is subscribed to the list currently.'))
1190 return redirect('mass_removal', mlist.list_id)
1191 if request.method == 'POST':
1192 try:
1193 # TODO maxking: This doesn't scale. Either the Core should provide
1194 # an endpoint to remove all subscribers or there should be some
1195 # better way to do this. Maybe, Core can take a list of email
1196 # addresses in batches of 50 and unsubscribe all of them.
1197 for names in mlist.members:
1198 mlist.unsubscribe(names.email)
1199 messages.success(request, _('All members have been'
1200 ' unsubscribed from the list.'))
1201 return redirect('list_members', mlist.list_id, 'subscriber')
1202 except Exception as e:
1203 messages.error(request, e)
1204 return render(request,
1205 'postorius/lists/confirm_removeall_subscribers.html',
1206 {'list': mlist})
1209 @login_required
1210 @list_owner_required
1211 def list_bans(request, list_id):
1212 return bans_view(
1213 request, list_id=list_id, template='postorius/lists/bans.html')
1216 @login_required
1217 @list_owner_required
1218 def list_header_matches(request, list_id):
1220 View and edit the list's header matches.
1222 m_list = List.objects.get_or_404(fqdn_listname=list_id)
1223 header_matches = m_list.header_matches
1224 HeaderMatchFormset = formset_factory(
1225 ListHeaderMatchForm, extra=1, can_delete=True, can_order=True,
1226 formset=ListHeaderMatchFormset)
1227 initial_data = [
1228 dict([
1229 (key, getattr(hm, key)) for key in ListHeaderMatchForm.base_fields
1230 ]) for hm in header_matches]
1232 # Process form submission.
1233 if request.method == 'POST':
1234 formset = HeaderMatchFormset(request.POST, initial=initial_data)
1235 if formset.is_valid():
1236 if not formset.has_changed():
1237 return redirect('list_header_matches', list_id)
1238 # Purge the existing header_matches
1239 header_matches.clear()
1240 # Add the ones in the form
1242 def form_order(f):
1243 # If ORDER is None (new header match), add it last.
1244 return f.cleaned_data.get('ORDER') or len(formset.forms)
1245 errors = []
1246 for form in sorted(formset, key=form_order):
1247 if 'header' not in form.cleaned_data:
1248 # The new header match form was not filled
1249 continue
1250 if form.cleaned_data.get('DELETE'):
1251 continue
1252 try:
1253 header_matches.add(
1254 header=form.cleaned_data['header'],
1255 pattern=form.cleaned_data['pattern'],
1256 action=form.cleaned_data['action'],
1258 except HTTPError as e:
1259 errors.append(e)
1260 for e in errors:
1261 messages.error(
1262 request, _('An error occurred: %s') % e.reason)
1263 if not errors:
1264 messages.success(request, _('The header matches were'
1265 ' successfully modified.'))
1266 return redirect('list_header_matches', list_id)
1267 else:
1268 formset = HeaderMatchFormset(initial=initial_data)
1269 # Adapt the last form to create new matches
1270 form_new = formset.forms[-1]
1271 form_new.fields['header'].widget.attrs['placeholder'] = _('New header')
1272 form_new.fields['pattern'].widget.attrs['placeholder'] = _('New pattern')
1273 del form_new.fields['ORDER']
1274 del form_new.fields['DELETE']
1276 return render(request, 'postorius/lists/header_matches.html', {
1277 'list': m_list,
1278 'formset': formset,
1282 def confirm_token(request, list_id):
1283 """Confirm token sent via Mailman Core.
1285 This view allows confirming tokens sent via Mailman Core for confirming
1286 subscriptions or verifying email addresses. This is an un-authenticated
1287 view and the authentication is done by assuming the secrecy of the token
1288 sent.
1290 :param request: The Django request object.
1291 :param list_id: The current mailinglist id.
1294 m_list = List.objects.get_or_404(fqdn_listname=list_id)
1295 if request.method == 'POST':
1296 form = TokenConfirmForm(request.POST)
1297 # If the token is None or something, just raise 400 error.
1298 if not form.is_valid():
1299 raise HTTPError(request.path, 400, _('Invalid confirmation token'),
1300 None, None)
1301 token = form.cleaned_data.get('token')
1302 try:
1303 pending_req = m_list.get_request(token)
1304 except HTTPError as e:
1305 if e.code == 404:
1306 raise HTTPError(request.path, 404,
1307 _('Token expired or invalid.'), None, None)
1308 raise
1309 # Since we only accept the token there isn't any need for Form data. We
1310 # just need a POST request at this URL to accept the token.
1311 m_list.moderate_request(token, action='accept')
1312 return redirect('list_summary', m_list.list_id)
1313 # Get the token from url parameter.
1314 token = request.GET.get('token')
1315 try:
1316 pending_req = m_list.get_request(token)
1317 except HTTPError as e:
1318 if e.code == 404:
1319 raise HTTPError(request.path, 404, _('Token expired or invalid.'),
1320 None, None)
1321 raise
1323 # If this is a token pending moderator approval, they need to login and
1324 # approve it from the pending requests page.
1325 if pending_req.get('token_owner') == 'moderator':
1326 return redirect('list_subscription_requests', m_list.list_id)
1328 token_type = pending_req.get('type')
1329 form = TokenConfirmForm(initial=dict(token=token))
1330 # Show the display_name if it is not None or "".
1331 if pending_req.get('display_name'):
1332 addr = email.utils.formataddr((pending_req.get('display_name'),
1333 pending_req.get('email')))
1334 else:
1335 addr = pending_req.get('email')
1337 return render(request, 'postorius/lists/confirm_token.html', {
1338 'mlist': m_list,
1339 'addr': addr,
1340 'token': token,
1341 'type': token_type,
1342 'form': form,