1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
|
# Copyright (C) 2009-2017 by the Free Software Foundation, Inc.
#
# This file is part of GNU Mailman.
#
# GNU Mailman is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option)
# any later version.
#
# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# GNU Mailman. If not, see <http://www.gnu.org/licenses/>.
"""Handle subscriptions."""
from mailman.database.transaction import flush
from mailman.email.message import UserNotification
from mailman.interfaces.listmanager import ListDeletingEvent
from mailman.interfaces.pending import IPendings
from mailman.interfaces.subscriptions import (
ISubscriptionManager, ISubscriptionService,
SubscriptionConfirmationNeededEvent,
UnsubscriptionConfirmationNeededEvent)
from mailman.interfaces.template import ITemplateLoader
from mailman.interfaces.workflows import IWorkflowStateManager
from mailman.utilities.string import expand
from mailman.workflows.builtin import (PendableSubscription,
PendableUnsubscription,
SubscriptionWorkflow,
UnSubscriptionWorkflow)
from public import public
from zope.component import getUtility
from zope.interface import implementer
@public
@implementer(ISubscriptionManager)
class SubscriptionManager:
def __init__(self, mlist):
self._mlist = mlist
def register(self, subscriber=None, *,
pre_verified=False, pre_confirmed=False, pre_approved=False):
"""See `ISubscriptionManager`."""
workflow = SubscriptionWorkflow(
self._mlist, subscriber,
pre_verified=pre_verified,
pre_confirmed=pre_confirmed,
pre_approved=pre_approved)
list(workflow)
return workflow.token, workflow.token_owner, workflow.member
def unregister(self, subscriber=None, *,
pre_confirmed=False, pre_approved=False):
workflow = UnSubscriptionWorkflow(
self._mlist, subscriber,
pre_confirmed=pre_confirmed,
pre_approved=pre_approved)
list(workflow)
return workflow.token, workflow.token_owner, workflow.member
def confirm(self, token):
if token is None:
raise LookupError
pendable = getUtility(IPendings).confirm(token, expunge=False)
if pendable is None:
raise LookupError
workflow_type = pendable.get('type')
assert workflow_type in (PendableSubscription.PEND_TYPE,
PendableUnsubscription.PEND_TYPE)
workflow = (SubscriptionWorkflow
if workflow_type == PendableSubscription.PEND_TYPE
else UnSubscriptionWorkflow)(self._mlist)
workflow.token = token
workflow.restore()
# In order to just run the whole workflow, all we need to do
# is iterate over the workflow object. On calling the __next__
# over the workflow iterator it automatically executes the steps
# that needs to be done.
list(workflow)
return workflow.token, workflow.token_owner, workflow.member
def discard(self, token):
with flush():
getUtility(IPendings).confirm(token)
getUtility(IWorkflowStateManager).discard(token)
def _handle_confirmation_needed_events(event, template_name):
subject = 'confirm {}'.format(event.token)
confirm_address = event.mlist.confirm_address(event.token)
email_address = event.email
# Send a verification email to the address.
template = getUtility(ITemplateLoader).get(template_name, event.mlist)
text = expand(template, event.mlist, dict(
token=event.token,
subject=subject,
confirm_email=confirm_address,
user_email=email_address,
# For backward compatibility.
confirm_address=confirm_address,
email_address=email_address,
domain_name=event.mlist.domain.mail_host,
contact_address=event.mlist.owner_address,
))
msg = UserNotification(email_address, confirm_address, subject, text)
msg.send(event.mlist, add_precedence=False)
@public
def handle_SubscriptionConfirmationNeededEvent(event):
if not isinstance(event, SubscriptionConfirmationNeededEvent):
return
_handle_confirmation_needed_events(event, 'list:user:action:subscribe')
@public
def handle_UnsubscriptionConfirmationNeededEvent(event):
if not isinstance(event, UnsubscriptionConfirmationNeededEvent):
return
_handle_confirmation_needed_events(event, 'list:user:action:unsubscribe')
@public
def handle_ListDeletingEvent(event):
"""Delete a mailing list's members when the list is being deleted."""
if not isinstance(event, ListDeletingEvent):
return
# Find all the members still associated with the mailing list.
members = getUtility(ISubscriptionService).find_members(
list_id=event.mailing_list.list_id)
for member in members:
member.unsubscribe()
|
