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
|
# Copyright (C) 2007-2012 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/>.
"""Application level list creation."""
from __future__ import absolute_import, unicode_literals
__metaclass__ = type
__all__ = [
'create_list',
'remove_list',
]
import os
import shutil
import logging
from zope.component import getUtility
from mailman.config import config
from mailman.interfaces.address import IEmailValidator
from mailman.interfaces.domain import (
BadDomainSpecificationError, IDomainManager)
from mailman.interfaces.listmanager import IListManager
from mailman.interfaces.member import MemberRole
from mailman.interfaces.styles import IStyleManager
from mailman.interfaces.usermanager import IUserManager
from mailman.utilities.modules import call_name
log = logging.getLogger('mailman.error')
def create_list(fqdn_listname, owners=None):
"""Create the named list and apply styles.
The mailing may not exist yet, but the domain specified in `fqdn_listname`
must exist.
:param fqdn_listname: The fully qualified name for the new mailing list.
:type fqdn_listname: string
:param owners: The mailing list owners.
:type owners: list of string email addresses
:return: The new mailing list.
:rtype: `IMailingList`
:raises BadDomainSpecificationError: when the hostname part of
`fqdn_listname` does not exist.
:raises ListAlreadyExistsError: when the mailing list already exists.
:raises InvalidEmailAddressError: when the fqdn email address is invalid.
"""
if owners is None:
owners = []
# This raises I
getUtility(IEmailValidator).validate(fqdn_listname)
listname, domain = fqdn_listname.split('@', 1)
if domain not in getUtility(IDomainManager):
raise BadDomainSpecificationError(domain)
mlist = getUtility(IListManager).create(fqdn_listname)
for style in getUtility(IStyleManager).lookup(mlist):
style.apply(mlist)
# Coordinate with the MTA, as defined in the configuration file.
call_name(config.mta.incoming).create(mlist)
# Create any owners that don't yet exist, and subscribe all addresses as
# owners of the mailing list.
user_manager = getUtility(IUserManager)
for owner_address in owners:
address = user_manager.get_address(owner_address)
if address is None:
user = user_manager.create_user(owner_address)
address = list(user.addresses)[0]
mlist.subscribe(address, MemberRole.owner)
return mlist
def remove_list(fqdn_listname, mailing_list=None, archives=True):
"""Remove the list and all associated artifacts and subscriptions."""
removeables = []
# mailing_list will be None when only residual archives are being removed.
if mailing_list is not None:
# Remove all subscriptions, regardless of role.
for member in mailing_list.subscribers.members:
member.unsubscribe()
# Delete the mailing list from the database.
getUtility(IListManager).delete(mailing_list)
# Do the MTA-specific list deletion tasks
call_name(config.mta.incoming).create(mailing_list)
# Remove the list directory.
removeables.append(os.path.join(config.LIST_DATA_DIR, fqdn_listname))
# Remove any stale locks associated with the list.
for filename in os.listdir(config.LOCK_DIR):
fn_listname = filename.split('.')[0]
if fn_listname == fqdn_listname:
removeables.append(os.path.join(config.LOCK_DIR, filename))
if archives:
private_dir = config.PRIVATE_ARCHIVE_FILE_DIR
public_dir = config.PUBLIC_ARCHIVE_FILE_DIR
removeables.extend([
os.path.join(private_dir, fqdn_listname),
os.path.join(private_dir, fqdn_listname + '.mbox'),
os.path.join(public_dir, fqdn_listname),
os.path.join(public_dir, fqdn_listname + '.mbox'),
])
# Now that we know what files and directories to delete, delete them.
for target in removeables:
if not os.path.exists(target):
pass
elif os.path.islink(target):
os.unlink(target)
elif os.path.isdir(target):
shutil.rmtree(target)
elif os.path.isfile(target):
os.unlink(target)
else:
log.error('Could not delete list artifact: %s', target)
|
