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
|
# Copyright (C) 2015-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/>.
"""Common support between subscription and unsubscription."""
import uuid
from datetime import timedelta
from enum import Enum
from mailman.interfaces.address import IAddress
from mailman.interfaces.pending import IPendings
from mailman.interfaces.subscriptions import TokenOwner
from mailman.interfaces.user import IUser
from mailman.interfaces.usermanager import IUserManager
from mailman.utilities.datetime import now
from mailman.workflows.base import Workflow
from zope.component import getUtility
class WhichSubscriber(Enum):
address = 1
user = 2
class SubscriptionWorkflowCommon(Workflow):
"""Common support between subscription and unsubscription."""
PENDABLE_CLASS = None
def __init__(self, mlist, subscriber):
super().__init__()
self.mlist = mlist
self.address = None
self.user = None
self.which = None
self.member = None
self._set_token(TokenOwner.no_one)
# The subscriber must be either an IUser or IAddress.
if IAddress.providedBy(subscriber):
self.address = subscriber
self.user = self.address.user
self.which = WhichSubscriber.address
elif IUser.providedBy(subscriber):
self.address = subscriber.preferred_address
self.user = subscriber
self.which = WhichSubscriber.user
self.subscriber = subscriber
@property
def user_key(self):
# For save.
return self.user.user_id.hex
@user_key.setter
def user_key(self, hex_key):
# For restore.
uid = uuid.UUID(hex_key)
self.user = getUtility(IUserManager).get_user_by_id(uid)
if self.user is None:
self.user = self.address.user
@property
def address_key(self):
# For save.
return self.address.email
@address_key.setter
def address_key(self, email):
# For restore.
self.address = getUtility(IUserManager).get_address(email)
assert self.address is not None
@property
def subscriber_key(self):
return self.which.value
@subscriber_key.setter
def subscriber_key(self, key):
self.which = WhichSubscriber(key)
@property
def token_owner_key(self):
return self.token_owner.value
@token_owner_key.setter
def token_owner_key(self, value):
self.token_owner = TokenOwner(value)
def _set_token(self, token_owner):
assert isinstance(token_owner, TokenOwner)
pendings = getUtility(IPendings)
# Clear out the previous pending token if there is one.
if self.token is not None:
pendings.confirm(self.token)
# Create a new token to prevent replay attacks. It seems like this
# would produce the same token, but it won't because the pending adds a
# bit of randomization.
self.token_owner = token_owner
if token_owner is TokenOwner.no_one:
self.token = None
return
pendable = self.PENDABLE_CLASS(
list_id=self.mlist.list_id,
email=self.address.email,
display_name=self.address.display_name,
when=now().replace(microsecond=0).isoformat(),
token_owner=token_owner.name,
)
self.token = pendings.add(pendable, timedelta(days=3650))
|
