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)
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
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/>.
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']
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.
88 if request
.GET
.get('q'):
89 query
= request
.GET
['q']
91 query
= '*{}*'.format(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')
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(
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
))
126 context
['empty_error'] = _(
127 'No {}s were found matching the search.'.format(role
))
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
)
137 members
= form
.cleaned_data
['choices']
138 for member
in members
:
139 self
.mailing_list
.unsubscribe(member
)
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():
154 self
.mailing_list
.add_role(
156 address
=member_form
.cleaned_data
['email'],
157 display_name
=member_form
.cleaned_data
['display_name'])
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
)
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
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
)
189 return self
._non
_member
_post
(request
, role
)
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
202 role
= request
.GET
.get('role')
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():
222 preferences_form
.save()
223 except HTTPError
as e
:
224 messages
.error(request
, e
.msg
)
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():
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
])
244 except HTTPError
as e
:
245 messages
.error(request
, e
.msg
)
247 messages
.success(request
, _("The member's moderation "
248 "settings have been updated."))
249 return redirect('list_member_options', list_id
, email
)
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
,
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
295 member
= self
.mailing_list
.get_member(address
)
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
)
308 if mm_user
.preferred_address
is None:
309 primary_email
= set_preferred(request
.user
, mm_user
)
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
)
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."""
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
))
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
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
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(
363 # Since the email is already verified in Postorius.
365 # Since this user was already a Member, simply switching Email
366 # addresses shouldn't require another approval.
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
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
)
382 member_pref
[prop
] = val
385 @method_decorator(login_required
)
386 def post(self
, request
, list_id
):
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
)
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
399 for address
in user_emails
:
401 member
= self
.mailing_list
.get_member(address
)
406 break # no need to test more addresses
407 assert old_email
is not None
409 subscriber
= form
.cleaned_data
['subscriber']
410 if self
._is
_subscribed
(member
, subscriber
):
411 messages
.error(request
, _('You are already subscribed'))
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
):
419 'Primary Address ({})').format(primary_email
)
422 _('Subscription changed to %s').format(subscriber
))
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.
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
)
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
)
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
):
461 request
, _('Your subscription request has been'
462 ' submitted and is waiting for moderator'
465 messages
.success(request
,
466 _('You are subscribed to %s.') %
467 self
.mailing_list
.fqdn_listname
)
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.
489 form
= ListAnonymousSubscribe(request
.POST
)
491 email
= form
.cleaned_data
.get('email')
492 display_name
= form
.cleaned_data
.get('display_name')
493 self
.mailing_list
.subscribe(
495 pre_verified
=False, pre_confirmed
=False)
496 messages
.success(request
, _('Please check your inbox for '
497 'further instructions'))
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
):
516 _('You have a pending unsubscription request waiting for'
517 ' moderator approval.'))
518 return redirect('list_summary', self
.mailing_list
.list_id
)
520 response
= self
.mailing_list
.unsubscribe(email
)
521 if response
is not None and response
.get('token') is not None:
523 request
, _('Your unsubscription request has been'
524 ' submitted and is waiting for moderator'
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
)
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
)
541 for data
in form
.cleaned_data
['emails']:
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(
552 display_name
=display_name
,
557 send_welcome_message
=form
.cleaned_data
[
558 'send_welcome_message'])
560 message
= _('The address %(address)s has been'
561 ' invited to %(list)s.')
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.')
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
)
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.'))
602 for data
in form
.cleaned_data
['emails']:
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
)
612 self
.mailing_list
.mass_unsubscribe(valid_emails
)
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:
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
:
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
)
640 message_ids
= form
.cleaned_data
['choices']
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'))
655 messages
.error(request
, _('Message could not be found'))
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
)
663 'list': mailing_list
,
664 'held_messages': held_messages
,
666 'ACTION_CHOICES': ACTION_CHOICES
,
668 return render(request
, 'postorius/lists/held_messages.html', context
)
671 @require_http_methods(['POST'])
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')
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
:
695 _('Held message was not found.'))
696 return redirect('list_held_messages', list_id
)
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
708 _('Moderation action for {} set to {}'.format(
709 member
, moderation_choices
[moderation_choice
])))
713 _('Failed to set moderation action for {}'.format(msg
.sender
)))
714 return redirect('list_held_messages', list_id
)
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
)
730 for i
in mm_lists
.members
:
731 writer
.writerow([i
.email
])
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']]
748 def _get_default_style():
749 return Style
.objects
.all()['default']
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.
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
)
772 domain
= Domain
.objects
.get_or_404(
773 mail_host
=form
.cleaned_data
['mail_host'])
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']
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':
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',
804 messages
.error(request
, _('Please check the errors below'))
806 form
= ListNew(choosable_domains
, choosable_styles
, initial
={
807 'list_owner': request
.user
.email
,
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):
827 for domain
in Domain
.objects
.all():
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
:
835 if len(mail_hosts
) == 1:
837 elif len(mail_hosts
) == 0 and use_web_host
:
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
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.
865 mail_host
= _get_mail_host(request
.get_host().split(':')[0])
866 for user_email
in user_emails
:
869 client
.find_lists(user_email
,
874 # No lists exist with the given role for the given user.
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.
884 'lists': _unique_lists(all_lists
),
885 'domain_count': len(choosable_domains
),
887 'check_advertised': False,
891 'postorius/index.html',
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
)
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
,
922 'check_advertised': True,
924 'domain_count': len(choosable_domains
)})
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':
935 mailinglist_deleted
.send(sender
=List
, list_id
=list_id
)
936 return redirect("list_index")
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
,
947 @list_moderator_required
948 def list_pending_confirmations(request
, list_id
):
949 """Shows a list of subscription requests.
951 return _list_subscriptions(
954 token_owner
=TokenOwner
.subscriber
,
955 template
='postorius/lists/pending_confirmations.html',
956 page_title
=_('Subscriptions pending user confirmation'),
961 @list_moderator_required
962 def list_subscription_requests(request
, list_id
):
963 """Shows a list of subscription requests."""
964 return _list_subscriptions(
967 token_owner
=TokenOwner
.moderator
,
968 template
='postorius/lists/subscription_requests.html',
969 page_title
=_('Subscriptions pending approval'),
974 @list_moderator_required
975 def list_unsubscription_requests(request
, list_id
):
976 """Shows a list of pending unsubscription requests."""
977 return _list_subscriptions(
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
)
992 for req
in m_list
.get_requests(
993 token_owner
=token_owner
, request_type
=request_type
)]
994 paginated_requests
= paginate(
996 request
.GET
.get('page'),
997 request
.GET
.get('count', 25))
998 page_subtitle
= '(%d)' % len(requests
)
999 return render(request
, template
,
1001 'paginated_requests': paginated_requests
,
1002 'page_title': page_title
,
1003 'page_subtitle': page_subtitle
})
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:
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
:
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
)
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')
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
:
1039 messages
.success(request
,
1040 _('The request was already moderated: %s')
1043 messages
.error(request
, _('The request could not be moderated: %s')
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')),
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
,
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'
1090 form_class
= SETTINGS_FORMS
[visible_section
]
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
)
1101 for key
in form
.changed_data
:
1102 if key
in form_class
.mlist_properties
:
1103 setattr(m_list
, key
, form
.cleaned_data
[key
])
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.
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
:
1125 _('An error occurred: ') + e
.reason
)
1126 return redirect('list_settings', m_list
.list_id
, visible_section
)
1128 form
= form_class(initial
=initial_data
, mlist
=m_list
)
1130 return render(request
, template
, {
1132 'section_names': SETTINGS_SECTION_NAMES
,
1134 'visible_section': visible_section
,
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
)
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':
1167 the_list
.remove_role(role
, address
)
1168 except HTTPError
as e
:
1169 messages
.error(request
, _('The user could not be removed: %(msg)s')
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
})
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':
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',
1210 @list_owner_required
1211 def list_bans(request
, list_id
):
1213 request
, list_id
=list_id
, template
='postorius/lists/bans.html')
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
)
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
1243 # If ORDER is None (new header match), add it last.
1244 return f
.cleaned_data
.get('ORDER') or len(formset
.forms
)
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
1250 if form
.cleaned_data
.get('DELETE'):
1254 header
=form
.cleaned_data
['header'],
1255 pattern
=form
.cleaned_data
['pattern'],
1256 action
=form
.cleaned_data
['action'],
1258 except HTTPError
as e
:
1262 request
, _('An error occurred: %s') % e
.reason
)
1264 messages
.success(request
, _('The header matches were'
1265 ' successfully modified.'))
1266 return redirect('list_header_matches', list_id
)
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', {
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
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'),
1301 token
= form
.cleaned_data
.get('token')
1303 pending_req
= m_list
.get_request(token
)
1304 except HTTPError
as e
:
1306 raise HTTPError(request
.path
, 404,
1307 _('Token expired or invalid.'), None, None)
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')
1316 pending_req
= m_list
.get_request(token
)
1317 except HTTPError
as e
:
1319 raise HTTPError(request
.path
, 404, _('Token expired or invalid.'),
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')))
1335 addr
= pending_req
.get('email')
1337 return render(request
, 'postorius/lists/confirm_token.html', {