# Copyright (c) 2008, Alexander Dutton # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # * Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # * Redistributions in binary form must reproduce the above copyright # notice, this list of conditions and the following disclaimer in the # documentation and/or other materials provided with the distribution. # * The name of its contributor may be used to endorse or promote products # derived from this software without specific prior written permission. # # THIS SOFTWARE IS PROVIDED BY ALEXANDER DUTTON ''AS IS'' AND ANY # EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE # DISCLAIMED. IN NO EVENT SHALL BE LIABLE FOR ANY # DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES # (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; # LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT # (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS # SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. """ Provides a simple interface to the Oxford Ezmlm web-based mailing list manager, allowing one to update the subcribers, moderators, allow and deny lists for any list you are a manager of. This module handles Webauth authentication, so you needn't worry. However, please note that if ever the Webauth or Maillist pages change, there could be severe breakage. Hopefully, such breakage will be non-damaging, but please don't get too upset if this module eats your cat (and your maillist). Sample usage: # First, we log in >>> manager = OxfordEzmlmManager() >>> manager.connect('username', 'secret') # We can now see what lists we are allowed to administer >>> manager.keys() ['mylist', 'spamlovers', 'egghaters'] # This is how we get to a set-like object representing the subscribers to mylist >>> mylist = manager['mylist'] >>> mylist_subscribers = mylist['subscribers'] >>> mylist_subscribers set(['alice@example.com', 'bob@example.com']) # The changes_made attribute tells us whether an update will do anything >>> mylist_subscribers.changes_made False # We can now add Eve and remove Alice >>> mylist_subscribers.add('eve@example.com') >>> mylist_subscribers.remove('alice@example.com') # Finally, we see that changes may have been made, and perform an update >>> mylist_subscribers.changes_made True >>> mylist_subscribers.save() Note that we can use other set operations (e.g. union, difference) too. """ import urllib, urllib2, ElementSoup, cookielib, re, sys __all__ = ['OxfordEzmlmManager'] class OxfordEzmlmManager(object): SERVER_URL = 'https://maillist.ox.ac.uk' LISTS_URL = SERVER_URL + '/maillist/-/lists/' LIST_CONFIG_RE = re.compile(r'/maillist/(?P.{8})/config/view') class LoginFailedException(Exception): pass def __init__(self): self.__connected = False def __get_parse_tree(self, url, data = None, timeout = None): if sys.version[0:2] >= (2, 6): data = self.__opener.open(url, data, timeout) else: data = self.__opener.open(url, data) return data, ElementSoup.parse(data) def connect(self, username, password): """ Initiates a Webauth session with the maillist server and retrieves the list of maillists that the user is allowed to administer. """ if self.__connected: raise AssertionError # We want to keep track of cookies, so we'll use a non-default Opener self.__opener = urllib2.build_opener(urllib2.HTTPCookieProcessor(cookielib.CookieJar())) # We grab the login page login_page, login_page_tree = self.__get_parse_tree(OxfordEzmlmManager.LISTS_URL) # Here we grab the field data from the Webauth login form and fill in # our username and password. post_data = {} for node in login_page_tree.findall('.//input'): post_data[node.get('name')] = node.get('value') post_data['username'] = username post_data['password'] = password post_data_encoded = urllib.urlencode(post_data) del post_data # We can now resubmit to the same page with our login details intermediate_page, intermediate_page_tree = self.__get_parse_tree(login_page.url, post_data_encoded) # If we can find a go_button, our login succeeded. Else, throw an # exception so people know what happened. go_button = intermediate_page_tree.find('.//td/p/span/a') if go_button is None: raise OxfordEzmlmManager.LoginFailedException('Incorrect username and/or password') go_href = go_button.get('href') lists_page, lists_page_tree = self.__get_parse_tree(go_href) self.__connected, self.__username = True, username # Construct our list of maillists. Ezmlm assigns each an id of the form # list0000, so we need to keep track of those to perform any operations # on them. We grab the listid from the URL of the first link on a given # row in the table of maillists. list_trs = lists_page_tree.findall('.//table/tr/td/table/tr') self.__lists = {} for list_tr in list_trs: tds = list_tr.getchildren() name = tds[0].text listid = OxfordEzmlmManager.LIST_CONFIG_RE.match(tds[1].getchildren()[0].get('href')).group('listid') self.__lists[name] = OxfordEzmlmList(OxfordEzmlmManager.SERVER_URL, name, listid, self.__opener) def require_connection(f): def g(self, *args, **kwargs): if not self.__connected: raise AssertionError('Connection required') return f(self, *args, **kwargs) return g @require_connection def get_username(self): return self.__username username = property(get_username) @require_connection def __getitem__(self, key): return self.__lists[key] @require_connection def __len__(self): return len(self.__lists) @require_connection def __iter__(self): return iter(self.__lists) @require_connection def __contains__(self, key): return key in self.__lists @require_connection def keys(self): return self.__lists.keys() @require_connection def values(self): return self.__lists.values() @require_connection def items(self): return self.__lists.items() class OxfordEzmlmList(object): GROUPS = ['subscribers', 'moderators', 'allow', 'deny'] def __init__(self, server_url, name, listid, opener): self.__server_url, self.__name, self.__listid, self.__opener = server_url, name, listid, opener self.__groups = dict([(group, OxfordEzmlmList.Group(self.__server_url, self.__listid, group, self.__opener)) for group in OxfordEzmlmList.GROUPS]) def __iter__(self): return iter(OxfordEzmlmList.GROUPS) def __getitem__(self, key): group = self.__groups[key] if not group.members_fetched: group.fetch_members() return group class Group(set): def __init__(self, server_url, listid, name, opener): self.__server_url, self.__listid, self.__name, self.__opener = server_url, listid, name, opener self.__members_fetched = False def fetch_members(self): """ Updates the member list for this group, wiping out any previous changes. """ members_url = '%s/maillist/%s/%s/dnload' % (self.__server_url, self.__listid, self.__name) data = self.__opener.open(members_url) # __members is the working copy of the member list for this group self.clear() self |= set([email.strip('\n') for email in data.readlines()]) # __old_members is what we believe to be the current version according to ezmlm self.__old_members, self.__members_fetched = set(self), True def get_changes_made(self): return self != self.__old_members changes_made = property(get_changes_made) def get_members_fetched(self): return self.__members_fetched members_fetched = property(get_members_fetched) def save(self): """ Updates the member list for this group. """ add_url = '%s/maillist/%s/%s/add' % (self.__server_url, self.__listid, self.__name) remove_url = '%s/maillist/%s/%s/remove' % (self.__server_url, self.__listid, self.__name) # Yay for set operations! to_add = set(self) - self.__old_members to_remove = self.__old_members - set(self) # If we need to add anyone, submit them to the relevant URL as a # newline-separated list. if to_add: post_data = urllib.urlencode({'sub':"\n".join(to_add)}) self.__opener.open(add_url, post_data) # Likewise for removing people, only this time we emulate the # checkboxes on the web-interface. if to_remove: post_data = urllib.urlencode(dict([('unsub_member%d' % i, email) for i, email in enumerate(to_remove)])) self.__opener.open(remove_url, post_data) # Now we've updated things we maintain our invariants. self.__old_members = set(self)