From c7e794caecb8b12d250be92f698fed8fa1f8a101 Mon Sep 17 00:00:00 2001 From: Barry Warsaw Date: Tue, 4 Jan 2011 19:10:13 -0500 Subject: Remove some unused/untested stuff regarding password creation from Utils.py. Also, remove a obsolete test. --- src/mailman/tests/test_security_mgr.py | 241 --------------------------------- 1 file changed, 241 deletions(-) delete mode 100644 src/mailman/tests/test_security_mgr.py (limited to 'src/mailman/tests') diff --git a/src/mailman/tests/test_security_mgr.py b/src/mailman/tests/test_security_mgr.py deleted file mode 100644 index a4f9c1cf4..000000000 --- a/src/mailman/tests/test_security_mgr.py +++ /dev/null @@ -1,241 +0,0 @@ -# Copyright (C) 2001-2011 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 . - -"""Unit tests for the SecurityManager module.""" - -from __future__ import absolute_import, unicode_literals - -__metaclass__ = type -__all__ = [ - 'test_suite', - ] - - -import os -import errno -import unittest - -# Don't use cStringIO because we're going to inherit -from StringIO import StringIO - -from mailman import Utils -from mailman import passwords -from mailman.config import config - - - -def password(cleartext): - return passwords.make_secret(cleartext, passwords.Schemes.ssha) - - - -class TestSecurityManager(unittest.TestCase): - def test_init_vars(self): - eq = self.assertEqual - eq(self._mlist.mod_password, None) - eq(self._mlist.passwords, {}) - - def test_auth_context_info_authuser(self): - mlist = self._mlist - self.assertRaises(TypeError, mlist.AuthContextInfo, config.AuthUser) - # Add a member - mlist.addNewMember('aperson@dom.ain', password='xxXXxx') - self.assertEqual( - mlist.AuthContextInfo(config.AuthUser, 'aperson@dom.ain'), - ('_xtest%40example.com+user+aperson--at--dom.ain', 'xxXXxx')) - - def test_auth_context_moderator(self): - mlist = self._mlist - mlist.mod_password = 'yyYYyy' - self.assertEqual( - mlist.AuthContextInfo(config.AuthListModerator), - ('_xtest%40example.com+moderator', 'yyYYyy')) - - def test_auth_context_admin(self): - mlist = self._mlist - mlist.password = 'zzZZzz' - self.assertEqual( - mlist.AuthContextInfo(config.AuthListAdmin), - ('_xtest%40example.com+admin', 'zzZZzz')) - - def test_auth_context_site(self): - mlist = self._mlist - mlist.password = 'aaAAaa' - self.assertEqual( - mlist.AuthContextInfo(config.AuthSiteAdmin), - ('_xtest%40example.com+admin', 'aaAAaa')) - - def test_auth_context_huh(self): - self.assertEqual( - self._mlist.AuthContextInfo('foo'), - (None, None)) - - - -class TestAuthenticate(unittest.TestCase): - def setUp(self): - Utils.set_global_password('bbBBbb', siteadmin=True) - Utils.set_global_password('ccCCcc', siteadmin=False) - - def tearDown(self): - try: - os.unlink(config.SITE_PW_FILE) - except OSError, e: - if e.errno <> errno.ENOENT: - raise - try: - os.unlink(config.LISTCREATOR_PW_FILE) - except OSError, e: - if e.errno <> errno.ENOENT: - raise - - def test_auth_creator(self): - self.assertEqual(self._mlist.Authenticate( - [config.AuthCreator], 'ccCCcc'), config.AuthCreator) - - def test_auth_creator_unauth(self): - self.assertEqual(self._mlist.Authenticate( - [config.AuthCreator], 'xxxxxx'), config.UnAuthorized) - - def test_auth_site_admin(self): - self.assertEqual(self._mlist.Authenticate( - [config.AuthSiteAdmin], 'bbBBbb'), config.AuthSiteAdmin) - - def test_auth_site_admin_unauth(self): - self.assertEqual(self._mlist.Authenticate( - [config.AuthSiteAdmin], 'xxxxxx'), config.UnAuthorized) - - def test_list_admin(self): - self._mlist.password = password('ttTTtt') - self.assertEqual(self._mlist.Authenticate( - [config.AuthListAdmin], 'ttTTtt'), config.AuthListAdmin) - - def test_list_admin_unauth(self): - self._mlist.password = password('ttTTtt') - self.assertEqual(self._mlist.Authenticate( - [config.AuthListAdmin], 'xxxxxx'), config.UnAuthorized) - - def test_list_moderator(self): - self._mlist.mod_password = password('mmMMmm') - self.assertEqual(self._mlist.Authenticate( - [config.AuthListModerator], 'mmMMmm'), config.AuthListModerator) - - def test_user(self): - mlist = self._mlist - mlist.addNewMember('aperson@dom.ain', password=password('nosrepa')) - self.assertEqual(mlist.Authenticate( - [config.AuthUser], 'nosrepa', 'aperson@dom.ain'), config.AuthUser) - - def test_wrong_user(self): - mlist = self._mlist - mlist.addNewMember('aperson@dom.ain', password='nosrepa') - self.assertEqual( - mlist.Authenticate([config.AuthUser], 'nosrepa', 'bperson@dom.ain'), - config.UnAuthorized) - - def test_no_user(self): - mlist = self._mlist - mlist.addNewMember('aperson@dom.ain', password='nosrepa') - self.assertEqual(mlist.Authenticate([config.AuthUser], 'norespa'), - config.UnAuthorized) - - def test_user_unauth(self): - mlist = self._mlist - mlist.addNewMember('aperson@dom.ain', password='nosrepa') - self.assertEqual(mlist.Authenticate( - [config.AuthUser], 'xxxxxx', 'aperson@dom.ain'), - config.UnAuthorized) - - def test_value_error(self): - self.assertRaises(ValueError, self._mlist.Authenticate, - ['spooge'], 'xxxxxx', 'zperson@dom.ain') - - - -class StripperIO(StringIO): - HEAD = 'Set-Cookie: ' - def write(self, s): - if s.startswith(self.HEAD): - s = s[len(self.HEAD):] - StringIO.write(self, s) - - -class TestWebAuthenticate(unittest.TestCase): - def setUp(self): - Utils.set_global_password('bbBBbb', siteadmin=True) - Utils.set_global_password('ccCCcc', siteadmin=False) - mlist = self._mlist - mlist.mod_password = password('abcdefg') - mlist.addNewMember('aperson@dom.ain', password='qqQQqq') - # Set up the cookie data - sfp = StripperIO() - print >> sfp, mlist.MakeCookie(config.AuthSiteAdmin) - # AuthCreator isn't handled in AuthContextInfo() - print >> sfp, mlist.MakeCookie(config.AuthListAdmin) - print >> sfp, mlist.MakeCookie(config.AuthListModerator) - print >> sfp, mlist.MakeCookie(config.AuthUser, 'aperson@dom.ain') - # Strip off the "Set-Cookie: " prefix - cookie = sfp.getvalue() - os.environ['HTTP_COOKIE'] = cookie - - def tearDown(self): - try: - os.unlink(config.SITE_PW_FILE) - except OSError, e: - if e.errno <> errno.ENOENT: - raise - try: - os.unlink(config.LISTCREATOR_PW_FILE) - except OSError, e: - if e.errno <> errno.ENOENT: - raise - del os.environ['HTTP_COOKIE'] - - def test_auth_site_admin(self): - self.failUnless(self._mlist.WebAuthenticate( - [config.AuthSiteAdmin], 'does not matter')) - - def test_list_admin(self): - self.failUnless(self._mlist.WebAuthenticate( - [config.AuthListAdmin], 'does not matter')) - - def test_list_moderator(self): - self.failUnless(self._mlist.WebAuthenticate( - [config.AuthListModerator], 'does not matter')) - - def test_user(self): - self.failUnless(self._mlist.WebAuthenticate( - [config.AuthUser], 'does not matter')) - - def test_not_a_user(self): - self._mlist.removeMember('aperson@dom.ain') - self.failIf(self._mlist.WebAuthenticate( - [config.AuthUser], 'does not matter', 'aperson@dom.ain')) - - - -# TBD: Tests for MakeCookie(), ZapCookie(), CheckCookie() -- although the -# latter is implicitly tested by testing WebAuthenticate() above. - - - -def test_suite(): - suite = unittest.TestSuite() -## suite.addTest(unittest.makeSuite(TestSecurityManager)) -## suite.addTest(unittest.makeSuite(TestAuthenticate)) -## suite.addTest(unittest.makeSuite(TestWebAuthenticate)) - return suite -- cgit v1.2.3-70-g09d2 From 51140e885c9e1dc074e1fb3f288f50a8e9add884 Mon Sep 17 00:00:00 2001 From: Barry Warsaw Date: Fri, 25 Feb 2011 15:29:19 -0500 Subject: Move passwords module to the utilities subpackage. --- src/mailman/Utils.py | 4 - src/mailman/passwords.py | 317 -------------------------- src/mailman/tests/test_membership.py | 2 +- src/mailman/tests/test_passwords.py | 150 ------------ src/mailman/utilities/passwords.py | 317 ++++++++++++++++++++++++++ src/mailman/utilities/tests/test_passwords.py | 150 ++++++++++++ 6 files changed, 468 insertions(+), 472 deletions(-) delete mode 100644 src/mailman/passwords.py delete mode 100644 src/mailman/tests/test_passwords.py create mode 100644 src/mailman/utilities/passwords.py create mode 100644 src/mailman/utilities/tests/test_passwords.py (limited to 'src/mailman/tests') diff --git a/src/mailman/Utils.py b/src/mailman/Utils.py index 384d25702..49b7d082e 100644 --- a/src/mailman/Utils.py +++ b/src/mailman/Utils.py @@ -33,20 +33,16 @@ import os import re import cgi import errno -import base64 -import random import logging # pylint: disable-msg=E0611,W0403 from email.errors import HeaderParseError from email.header import decode_header, make_header -from lazr.config import as_boolean from string import ascii_letters, digits, whitespace from zope.component import getUtility import mailman.templates -from mailman import passwords from mailman.config import config from mailman.core.i18n import _ from mailman.interfaces.languages import ILanguageManager diff --git a/src/mailman/passwords.py b/src/mailman/passwords.py deleted file mode 100644 index c14584748..000000000 --- a/src/mailman/passwords.py +++ /dev/null @@ -1,317 +0,0 @@ -# Copyright (C) 2007-2011 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 . - -"""Password hashing and verification schemes. - -Represents passwords using RFC 2307 syntax (as best we can tell). -""" - -from __future__ import absolute_import, unicode_literals - -__metaclass__ = type -__all__ = [ - 'Schemes', - 'check_response', - 'make_secret', - ] - - -import os -import re -import hmac -import hashlib - -from array import array -from base64 import urlsafe_b64decode as decode -from base64 import urlsafe_b64encode as encode -from flufl.enum import Enum - -from mailman.core import errors - -SALT_LENGTH = 20 # bytes -ITERATIONS = 2000 - - - -# pylint: disable-msg=W0232 -class PasswordScheme: - """Password scheme base class.""" - TAG = b'' - - @staticmethod - def make_secret(password): - """Return the hashed password. - - :param password: The clear text password. - :type password: string - :return: The encrypted password. - :rtype: string - """ - raise NotImplementedError - - @staticmethod - def check_response(challenge, response): - """Check a response against a challenge. - - It is expected that the scheme specifier prefix is already stripped - from the response string. - - :param challenge: The challenge. - :type challenge: string - :param response: The response. - :type response: string - :return: True if the response matches the challenge. - :rtype: bool - """ - raise NotImplementedError - - - -class NoPasswordScheme(PasswordScheme): - """A password scheme without passwords.""" - - TAG = b'NONE' - - @staticmethod - def make_secret(password): - """See `PasswordScheme`.""" - return b'' - - # pylint: disable-msg=W0613 - @staticmethod - def check_response(challenge, response): - """See `PasswordScheme`.""" - return False - - - -class ClearTextPasswordScheme(PasswordScheme): - """A password scheme that stores clear text passwords.""" - - TAG = b'CLEARTEXT' - - @staticmethod - def make_secret(password): - """See `PasswordScheme`.""" - return password - - @staticmethod - def check_response(challenge, response): - """See `PasswordScheme`.""" - return challenge == response - - - -class SHAPasswordScheme(PasswordScheme): - """A password scheme that encodes the password using SHA1.""" - - TAG = b'SHA' - - @staticmethod - def make_secret(password): - """See `PasswordScheme`.""" - h = hashlib.sha1(password) - return encode(h.digest()) - - @staticmethod - def check_response(challenge, response): - """See `PasswordScheme`.""" - h = hashlib.sha1(response) - return challenge == encode(h.digest()) - - - -class SSHAPasswordScheme(PasswordScheme): - """A password scheme that encodes the password using salted SHA1.""" - - TAG = b'SSHA' - - @staticmethod - def make_secret(password): - """See `PasswordScheme`.""" - salt = os.urandom(SALT_LENGTH) - h = hashlib.sha1(password) - h.update(salt) - return encode(h.digest() + salt) - - @staticmethod - def check_response(challenge, response): - """See `PasswordScheme`.""" - # Get the salt from the challenge - challenge_bytes = decode(challenge) - digest = challenge_bytes[:20] - salt = challenge_bytes[20:] - h = hashlib.sha1(response) - h.update(salt) - return digest == h.digest() - - - -# Basic algorithm given by Bob Fleck -class PBKDF2PasswordScheme(PasswordScheme): - """RFC 2989 password encoding scheme.""" - - # This is a bit nasty if we wanted a different prf or iterations. OTOH, - # we really have no clue what the standard LDAP-ish specification for - # those options is. - TAG = b'PBKDF2 SHA {0}'.format(ITERATIONS) - - @staticmethod - def _pbkdf2(password, salt, iterations): - """From RFC2898 sec. 5.2. Simplified to handle only 20 byte output - case. Output of 20 bytes means always exactly one block to handle, - and a constant block counter appended to the salt in the initial hmac - update. - """ - h = hmac.new(password, None, hashlib.sha1) - prf = h.copy() - prf.update(salt + b'\x00\x00\x00\x01') - T = U = array(b'i', prf.digest()) - while iterations: - prf = h.copy() - prf.update(U.tostring()) - U = array(b'i', prf.digest()) - T = array(b'i', (t ^ u for t, u in zip(T, U))) - iterations -= 1 - return T.tostring() - - @staticmethod - def make_secret(password): - """See `PasswordScheme`. - - From RFC2898 sec. 5.2. Simplified to handle only 20 byte output - case. Output of 20 bytes means always exactly one block to handle, - and a constant block counter appended to the salt in the initial hmac - update. - """ - salt = os.urandom(SALT_LENGTH) - digest = PBKDF2PasswordScheme._pbkdf2(password, salt, ITERATIONS) - derived_key = encode(digest + salt) - return derived_key - - # pylint: disable-msg=W0221 - @staticmethod - def check_response(challenge, response, prf, iterations): - """See `PasswordScheme`.""" - # Decode the challenge to get the number of iterations and salt. We - # don't support anything but SHA PRF. - if prf.lower() != b'sha': - return False - try: - iterations = int(iterations) - except (ValueError, TypeError): - return False - challenge_bytes = decode(challenge) - digest = challenge_bytes[:20] - salt = challenge_bytes[20:] - key = PBKDF2PasswordScheme._pbkdf2(response, salt, iterations) - return digest == key - - - -class Schemes(Enum): - """List of password schemes.""" - # no_scheme is deliberately ugly because no one should be using it. Yes, - # this makes cleartext inconsistent, but that's a common enough - # terminology to justify the missing underscore. - no_scheme = 1 - cleartext = 2 - sha = 3 - ssha = 4 - pbkdf2 = 5 - - -_SCHEMES_BY_ENUM = { - Schemes.no_scheme : NoPasswordScheme, - Schemes.cleartext : ClearTextPasswordScheme, - Schemes.sha : SHAPasswordScheme, - Schemes.ssha : SSHAPasswordScheme, - Schemes.pbkdf2 : PBKDF2PasswordScheme, - } - - -# Some scheme tags have arguments, but the key for this dictionary should just -# be the lowercased scheme name. -_SCHEMES_BY_TAG = dict((_SCHEMES_BY_ENUM[e].TAG.split(' ')[0].lower(), e) - for e in _SCHEMES_BY_ENUM) - -_DEFAULT_SCHEME = NoPasswordScheme - - - -def make_secret(password, scheme=None): - """Encrypt a password. - - :param password: The clear text password. - :type password: string - :param scheme: The password scheme name. - :type scheme: string - :return: The encrypted password. - :rtype: string - """ - # The hash algorithms operate on bytes not strings. The password argument - # as provided here by the client will be a string (in Python 2 either - # unicode or 8-bit, in Python 3 always unicode). We need to encode this - # string into a byte array, and the way to spell that in Python 2 is to - # encode the string to utf-8. The returned secret is a string, so it must - # be a unicode. - if isinstance(password, unicode): - password = password.encode('utf-8') - scheme_class = _SCHEMES_BY_ENUM.get(scheme) - if not scheme_class: - raise errors.BadPasswordSchemeError(scheme) - secret = scheme_class.make_secret(password) - return b'{{{0}}}{1}'.format(scheme_class.TAG, secret) - - -def check_response(challenge, response): - """Check a response against a challenge. - - :param challenge: The challenge. - :type challenge: string - :param response: The response. - :type response: string - :return: True if the response matches the challenge. - :rtype: bool - """ - mo = re.match(r'{(?P[^}]+?)}(?P.*)', - challenge, re.IGNORECASE) - if not mo: - return False - # See above for why we convert here. However because we should have - # generated the challenge, we assume that it is already a byte string. - if isinstance(response, unicode): - response = response.encode('utf-8') - scheme_group, rest_group = mo.group('scheme', 'rest') - scheme_parts = scheme_group.split() - scheme = scheme_parts[0].lower() - scheme_enum = _SCHEMES_BY_TAG.get(scheme, _DEFAULT_SCHEME) - scheme_class = _SCHEMES_BY_ENUM[scheme_enum] - if isinstance(rest_group, unicode): - rest_group = rest_group.encode('utf-8') - return scheme_class.check_response(rest_group, response, *scheme_parts[1:]) - - -def lookup_scheme(scheme_name): - """Look up a password scheme. - - :param scheme_name: The password scheme name. - :type scheme_name: string - :return: Password scheme class. - :rtype: `PasswordScheme` - """ - return _SCHEMES_BY_TAG.get(scheme_name.lower()) diff --git a/src/mailman/tests/test_membership.py b/src/mailman/tests/test_membership.py index c3e40cfcc..8425cf65f 100644 --- a/src/mailman/tests/test_membership.py +++ b/src/mailman/tests/test_membership.py @@ -28,11 +28,11 @@ __all__ = [ import time import unittest -from mailman import passwords from mailman.app.lifecycle import create_list, remove_list from mailman.config import config from mailman.interfaces.member import NotAMemberError from mailman.testing.layers import ConfigLayer +from mailman.utilities import passwords diff --git a/src/mailman/tests/test_passwords.py b/src/mailman/tests/test_passwords.py deleted file mode 100644 index 49f55e82f..000000000 --- a/src/mailman/tests/test_passwords.py +++ /dev/null @@ -1,150 +0,0 @@ -# Copyright (C) 2007-2011 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 . - -"""Unit tests for the passwords module.""" - -from __future__ import absolute_import, unicode_literals - -__metaclass__ = type -__all__ = [ - 'test_suite', - ] - - -import unittest - -from mailman import passwords -from mailman.core import errors - - - -class TestPasswordsBase(unittest.TestCase): - scheme = None - - def setUp(self): - # passwords; 8-bit or unicode strings; ascii or binary - self.pw8a = b'abc' - self.pw8b = b'abc\xc3\xbf' # 'abc\xff' - self.pwub = b'abc\xff' - # bad password; 8-bit or unicode; ascii or binary - self.bad8a = b'xyz' - self.bad8b = b'xyz\xc3\xbf' # 'xyz\xff' - self.badub = b'xyz\xff' - - def test_passwords(self): - unless = self.failUnless - failif = self.failIf - secret = passwords.make_secret(self.pw8a, self.scheme) - unless(passwords.check_response(secret, self.pw8a)) - failif(passwords.check_response(secret, self.bad8a)) - - def test_passwords_with_funky_chars(self): - unless = self.failUnless - failif = self.failIf - secret = passwords.make_secret(self.pw8b, self.scheme) - unless(passwords.check_response(secret, self.pw8b)) - failif(passwords.check_response(secret, self.bad8b)) - - def test_unicode_passwords_with_funky_chars(self): - unless = self.failUnless - failif = self.failIf - secret = passwords.make_secret(self.pwub, self.scheme) - unless(passwords.check_response(secret, self.pwub)) - failif(passwords.check_response(secret, self.badub)) - - - -class TestBogusPasswords(TestPasswordsBase): - scheme = -1 - - def test_passwords(self): - self.assertRaises(errors.BadPasswordSchemeError, - passwords.make_secret, self.pw8a, self.scheme) - - def test_passwords_with_funky_chars(self): - self.assertRaises(errors.BadPasswordSchemeError, - passwords.make_secret, self.pw8b, self.scheme) - - def test_unicode_passwords_with_funky_chars(self): - self.assertRaises(errors.BadPasswordSchemeError, - passwords.make_secret, self.pwub, self.scheme) - - - -class TestNonePasswords(TestPasswordsBase): - scheme = passwords.Schemes.no_scheme - - def test_passwords(self): - failif = self.failIf - secret = passwords.make_secret(self.pw8a, self.scheme) - failif(passwords.check_response(secret, self.pw8a)) - failif(passwords.check_response(secret, self.bad8a)) - - def test_passwords_with_funky_chars(self): - failif = self.failIf - secret = passwords.make_secret(self.pw8b, self.scheme) - failif(passwords.check_response(secret, self.pw8b)) - failif(passwords.check_response(secret, self.bad8b)) - - def test_unicode_passwords_with_funky_chars(self): - failif = self.failIf - secret = passwords.make_secret(self.pwub, self.scheme) - failif(passwords.check_response(secret, self.pwub)) - failif(passwords.check_response(secret, self.badub)) - - - -class TestCleartextPasswords(TestPasswordsBase): - scheme = passwords.Schemes.cleartext - - -class TestSHAPasswords(TestPasswordsBase): - scheme = passwords.Schemes.sha - - -class TestSSHAPasswords(TestPasswordsBase): - scheme = passwords.Schemes.ssha - - -class TestPBKDF2Passwords(TestPasswordsBase): - scheme = passwords.Schemes.pbkdf2 - - - -class TestSchemeLookup(unittest.TestCase): - def test_scheme_name_lookup(self): - unless = self.failUnless - unless(passwords.lookup_scheme('NONE') is passwords.Schemes.no_scheme) - unless(passwords.lookup_scheme('CLEARTEXT') is - passwords.Schemes.cleartext) - unless(passwords.lookup_scheme('SHA') is passwords.Schemes.sha) - unless(passwords.lookup_scheme('SSHA') is passwords.Schemes.ssha) - unless(passwords.lookup_scheme('PBKDF2') is passwords.Schemes.pbkdf2) - unless(passwords.lookup_scheme(' -bogus- ') is None) - - - -def test_suite(): - suite = unittest.TestSuite() - suite.addTest(unittest.makeSuite(TestBogusPasswords)) - suite.addTest(unittest.makeSuite(TestNonePasswords)) - suite.addTest(unittest.makeSuite(TestCleartextPasswords)) - suite.addTest(unittest.makeSuite(TestSHAPasswords)) - suite.addTest(unittest.makeSuite(TestSSHAPasswords)) - suite.addTest(unittest.makeSuite(TestPBKDF2Passwords)) - suite.addTest(unittest.makeSuite(TestSchemeLookup)) - return suite diff --git a/src/mailman/utilities/passwords.py b/src/mailman/utilities/passwords.py new file mode 100644 index 000000000..c14584748 --- /dev/null +++ b/src/mailman/utilities/passwords.py @@ -0,0 +1,317 @@ +# Copyright (C) 2007-2011 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 . + +"""Password hashing and verification schemes. + +Represents passwords using RFC 2307 syntax (as best we can tell). +""" + +from __future__ import absolute_import, unicode_literals + +__metaclass__ = type +__all__ = [ + 'Schemes', + 'check_response', + 'make_secret', + ] + + +import os +import re +import hmac +import hashlib + +from array import array +from base64 import urlsafe_b64decode as decode +from base64 import urlsafe_b64encode as encode +from flufl.enum import Enum + +from mailman.core import errors + +SALT_LENGTH = 20 # bytes +ITERATIONS = 2000 + + + +# pylint: disable-msg=W0232 +class PasswordScheme: + """Password scheme base class.""" + TAG = b'' + + @staticmethod + def make_secret(password): + """Return the hashed password. + + :param password: The clear text password. + :type password: string + :return: The encrypted password. + :rtype: string + """ + raise NotImplementedError + + @staticmethod + def check_response(challenge, response): + """Check a response against a challenge. + + It is expected that the scheme specifier prefix is already stripped + from the response string. + + :param challenge: The challenge. + :type challenge: string + :param response: The response. + :type response: string + :return: True if the response matches the challenge. + :rtype: bool + """ + raise NotImplementedError + + + +class NoPasswordScheme(PasswordScheme): + """A password scheme without passwords.""" + + TAG = b'NONE' + + @staticmethod + def make_secret(password): + """See `PasswordScheme`.""" + return b'' + + # pylint: disable-msg=W0613 + @staticmethod + def check_response(challenge, response): + """See `PasswordScheme`.""" + return False + + + +class ClearTextPasswordScheme(PasswordScheme): + """A password scheme that stores clear text passwords.""" + + TAG = b'CLEARTEXT' + + @staticmethod + def make_secret(password): + """See `PasswordScheme`.""" + return password + + @staticmethod + def check_response(challenge, response): + """See `PasswordScheme`.""" + return challenge == response + + + +class SHAPasswordScheme(PasswordScheme): + """A password scheme that encodes the password using SHA1.""" + + TAG = b'SHA' + + @staticmethod + def make_secret(password): + """See `PasswordScheme`.""" + h = hashlib.sha1(password) + return encode(h.digest()) + + @staticmethod + def check_response(challenge, response): + """See `PasswordScheme`.""" + h = hashlib.sha1(response) + return challenge == encode(h.digest()) + + + +class SSHAPasswordScheme(PasswordScheme): + """A password scheme that encodes the password using salted SHA1.""" + + TAG = b'SSHA' + + @staticmethod + def make_secret(password): + """See `PasswordScheme`.""" + salt = os.urandom(SALT_LENGTH) + h = hashlib.sha1(password) + h.update(salt) + return encode(h.digest() + salt) + + @staticmethod + def check_response(challenge, response): + """See `PasswordScheme`.""" + # Get the salt from the challenge + challenge_bytes = decode(challenge) + digest = challenge_bytes[:20] + salt = challenge_bytes[20:] + h = hashlib.sha1(response) + h.update(salt) + return digest == h.digest() + + + +# Basic algorithm given by Bob Fleck +class PBKDF2PasswordScheme(PasswordScheme): + """RFC 2989 password encoding scheme.""" + + # This is a bit nasty if we wanted a different prf or iterations. OTOH, + # we really have no clue what the standard LDAP-ish specification for + # those options is. + TAG = b'PBKDF2 SHA {0}'.format(ITERATIONS) + + @staticmethod + def _pbkdf2(password, salt, iterations): + """From RFC2898 sec. 5.2. Simplified to handle only 20 byte output + case. Output of 20 bytes means always exactly one block to handle, + and a constant block counter appended to the salt in the initial hmac + update. + """ + h = hmac.new(password, None, hashlib.sha1) + prf = h.copy() + prf.update(salt + b'\x00\x00\x00\x01') + T = U = array(b'i', prf.digest()) + while iterations: + prf = h.copy() + prf.update(U.tostring()) + U = array(b'i', prf.digest()) + T = array(b'i', (t ^ u for t, u in zip(T, U))) + iterations -= 1 + return T.tostring() + + @staticmethod + def make_secret(password): + """See `PasswordScheme`. + + From RFC2898 sec. 5.2. Simplified to handle only 20 byte output + case. Output of 20 bytes means always exactly one block to handle, + and a constant block counter appended to the salt in the initial hmac + update. + """ + salt = os.urandom(SALT_LENGTH) + digest = PBKDF2PasswordScheme._pbkdf2(password, salt, ITERATIONS) + derived_key = encode(digest + salt) + return derived_key + + # pylint: disable-msg=W0221 + @staticmethod + def check_response(challenge, response, prf, iterations): + """See `PasswordScheme`.""" + # Decode the challenge to get the number of iterations and salt. We + # don't support anything but SHA PRF. + if prf.lower() != b'sha': + return False + try: + iterations = int(iterations) + except (ValueError, TypeError): + return False + challenge_bytes = decode(challenge) + digest = challenge_bytes[:20] + salt = challenge_bytes[20:] + key = PBKDF2PasswordScheme._pbkdf2(response, salt, iterations) + return digest == key + + + +class Schemes(Enum): + """List of password schemes.""" + # no_scheme is deliberately ugly because no one should be using it. Yes, + # this makes cleartext inconsistent, but that's a common enough + # terminology to justify the missing underscore. + no_scheme = 1 + cleartext = 2 + sha = 3 + ssha = 4 + pbkdf2 = 5 + + +_SCHEMES_BY_ENUM = { + Schemes.no_scheme : NoPasswordScheme, + Schemes.cleartext : ClearTextPasswordScheme, + Schemes.sha : SHAPasswordScheme, + Schemes.ssha : SSHAPasswordScheme, + Schemes.pbkdf2 : PBKDF2PasswordScheme, + } + + +# Some scheme tags have arguments, but the key for this dictionary should just +# be the lowercased scheme name. +_SCHEMES_BY_TAG = dict((_SCHEMES_BY_ENUM[e].TAG.split(' ')[0].lower(), e) + for e in _SCHEMES_BY_ENUM) + +_DEFAULT_SCHEME = NoPasswordScheme + + + +def make_secret(password, scheme=None): + """Encrypt a password. + + :param password: The clear text password. + :type password: string + :param scheme: The password scheme name. + :type scheme: string + :return: The encrypted password. + :rtype: string + """ + # The hash algorithms operate on bytes not strings. The password argument + # as provided here by the client will be a string (in Python 2 either + # unicode or 8-bit, in Python 3 always unicode). We need to encode this + # string into a byte array, and the way to spell that in Python 2 is to + # encode the string to utf-8. The returned secret is a string, so it must + # be a unicode. + if isinstance(password, unicode): + password = password.encode('utf-8') + scheme_class = _SCHEMES_BY_ENUM.get(scheme) + if not scheme_class: + raise errors.BadPasswordSchemeError(scheme) + secret = scheme_class.make_secret(password) + return b'{{{0}}}{1}'.format(scheme_class.TAG, secret) + + +def check_response(challenge, response): + """Check a response against a challenge. + + :param challenge: The challenge. + :type challenge: string + :param response: The response. + :type response: string + :return: True if the response matches the challenge. + :rtype: bool + """ + mo = re.match(r'{(?P[^}]+?)}(?P.*)', + challenge, re.IGNORECASE) + if not mo: + return False + # See above for why we convert here. However because we should have + # generated the challenge, we assume that it is already a byte string. + if isinstance(response, unicode): + response = response.encode('utf-8') + scheme_group, rest_group = mo.group('scheme', 'rest') + scheme_parts = scheme_group.split() + scheme = scheme_parts[0].lower() + scheme_enum = _SCHEMES_BY_TAG.get(scheme, _DEFAULT_SCHEME) + scheme_class = _SCHEMES_BY_ENUM[scheme_enum] + if isinstance(rest_group, unicode): + rest_group = rest_group.encode('utf-8') + return scheme_class.check_response(rest_group, response, *scheme_parts[1:]) + + +def lookup_scheme(scheme_name): + """Look up a password scheme. + + :param scheme_name: The password scheme name. + :type scheme_name: string + :return: Password scheme class. + :rtype: `PasswordScheme` + """ + return _SCHEMES_BY_TAG.get(scheme_name.lower()) diff --git a/src/mailman/utilities/tests/test_passwords.py b/src/mailman/utilities/tests/test_passwords.py new file mode 100644 index 000000000..60c201a5a --- /dev/null +++ b/src/mailman/utilities/tests/test_passwords.py @@ -0,0 +1,150 @@ +# Copyright (C) 2007-2011 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 . + +"""Unit tests for the passwords module.""" + +from __future__ import absolute_import, unicode_literals + +__metaclass__ = type +__all__ = [ + 'test_suite', + ] + + +import unittest + +from mailman.core import errors +from mailman.utilities import passwords + + + +class TestPasswordsBase(unittest.TestCase): + scheme = None + + def setUp(self): + # passwords; 8-bit or unicode strings; ascii or binary + self.pw8a = b'abc' + self.pw8b = b'abc\xc3\xbf' # 'abc\xff' + self.pwub = b'abc\xff' + # bad password; 8-bit or unicode; ascii or binary + self.bad8a = b'xyz' + self.bad8b = b'xyz\xc3\xbf' # 'xyz\xff' + self.badub = b'xyz\xff' + + def test_passwords(self): + unless = self.failUnless + failif = self.failIf + secret = passwords.make_secret(self.pw8a, self.scheme) + unless(passwords.check_response(secret, self.pw8a)) + failif(passwords.check_response(secret, self.bad8a)) + + def test_passwords_with_funky_chars(self): + unless = self.failUnless + failif = self.failIf + secret = passwords.make_secret(self.pw8b, self.scheme) + unless(passwords.check_response(secret, self.pw8b)) + failif(passwords.check_response(secret, self.bad8b)) + + def test_unicode_passwords_with_funky_chars(self): + unless = self.failUnless + failif = self.failIf + secret = passwords.make_secret(self.pwub, self.scheme) + unless(passwords.check_response(secret, self.pwub)) + failif(passwords.check_response(secret, self.badub)) + + + +class TestBogusPasswords(TestPasswordsBase): + scheme = -1 + + def test_passwords(self): + self.assertRaises(errors.BadPasswordSchemeError, + passwords.make_secret, self.pw8a, self.scheme) + + def test_passwords_with_funky_chars(self): + self.assertRaises(errors.BadPasswordSchemeError, + passwords.make_secret, self.pw8b, self.scheme) + + def test_unicode_passwords_with_funky_chars(self): + self.assertRaises(errors.BadPasswordSchemeError, + passwords.make_secret, self.pwub, self.scheme) + + + +class TestNonePasswords(TestPasswordsBase): + scheme = passwords.Schemes.no_scheme + + def test_passwords(self): + failif = self.failIf + secret = passwords.make_secret(self.pw8a, self.scheme) + failif(passwords.check_response(secret, self.pw8a)) + failif(passwords.check_response(secret, self.bad8a)) + + def test_passwords_with_funky_chars(self): + failif = self.failIf + secret = passwords.make_secret(self.pw8b, self.scheme) + failif(passwords.check_response(secret, self.pw8b)) + failif(passwords.check_response(secret, self.bad8b)) + + def test_unicode_passwords_with_funky_chars(self): + failif = self.failIf + secret = passwords.make_secret(self.pwub, self.scheme) + failif(passwords.check_response(secret, self.pwub)) + failif(passwords.check_response(secret, self.badub)) + + + +class TestCleartextPasswords(TestPasswordsBase): + scheme = passwords.Schemes.cleartext + + +class TestSHAPasswords(TestPasswordsBase): + scheme = passwords.Schemes.sha + + +class TestSSHAPasswords(TestPasswordsBase): + scheme = passwords.Schemes.ssha + + +class TestPBKDF2Passwords(TestPasswordsBase): + scheme = passwords.Schemes.pbkdf2 + + + +class TestSchemeLookup(unittest.TestCase): + def test_scheme_name_lookup(self): + unless = self.failUnless + unless(passwords.lookup_scheme('NONE') is passwords.Schemes.no_scheme) + unless(passwords.lookup_scheme('CLEARTEXT') is + passwords.Schemes.cleartext) + unless(passwords.lookup_scheme('SHA') is passwords.Schemes.sha) + unless(passwords.lookup_scheme('SSHA') is passwords.Schemes.ssha) + unless(passwords.lookup_scheme('PBKDF2') is passwords.Schemes.pbkdf2) + unless(passwords.lookup_scheme(' -bogus- ') is None) + + + +def test_suite(): + suite = unittest.TestSuite() + suite.addTest(unittest.makeSuite(TestBogusPasswords)) + suite.addTest(unittest.makeSuite(TestNonePasswords)) + suite.addTest(unittest.makeSuite(TestCleartextPasswords)) + suite.addTest(unittest.makeSuite(TestSHAPasswords)) + suite.addTest(unittest.makeSuite(TestSSHAPasswords)) + suite.addTest(unittest.makeSuite(TestPBKDF2Passwords)) + suite.addTest(unittest.makeSuite(TestSchemeLookup)) + return suite -- cgit v1.2.3-70-g09d2 From db2777e4aea3b516906a9500a0156d388779292e Mon Sep 17 00:00:00 2001 From: Barry Warsaw Date: Fri, 25 Feb 2011 18:15:58 -0500 Subject: Eliminate Utils.get_pattern() and in the process, completely revamp subscription bans so as not to have to rely on BLOBS or pickles in the database. Also, be sure to include .rst files in both doctests and in the packaged tarballs. With the now awesome reST mode for Emacs, I plan to rename all .txt doctest files to .rst. --- MANIFEST.in | 2 +- src/mailman/Utils.py | 26 ----- src/mailman/app/docs/bans.rst | 167 +++++++++++++++++++++++++++++++ src/mailman/app/membership.py | 7 +- src/mailman/app/tests/test_membership.py | 57 ++++++++++- src/mailman/config/configure.zcml | 5 + src/mailman/database/mailman.sql | 10 +- src/mailman/interfaces/bans.py | 110 ++++++++++++++++++++ src/mailman/model/bans.py | 110 ++++++++++++++++++++ src/mailman/model/mailinglist.py | 3 +- src/mailman/styles/default.py | 2 - src/mailman/tests/test_documentation.py | 2 +- 12 files changed, 462 insertions(+), 39 deletions(-) create mode 100644 src/mailman/app/docs/bans.rst create mode 100644 src/mailman/interfaces/bans.py create mode 100644 src/mailman/model/bans.py (limited to 'src/mailman/tests') diff --git a/MANIFEST.in b/MANIFEST.in index 5d594934b..658a4f700 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -4,7 +4,7 @@ recursive-include .buildout * recursive-include contrib * recursive-include cron * recursive-include data * -global-include *.txt *.po *.mo *.cfg *.sql *.zcml *.html +global-include *.txt *.rst *.po *.mo *.cfg *.sql *.zcml *.html global-exclude *.egg-info exclude MANIFEST.in prune src/attic diff --git a/src/mailman/Utils.py b/src/mailman/Utils.py index 49b7d082e..e10a9fde5 100644 --- a/src/mailman/Utils.py +++ b/src/mailman/Utils.py @@ -407,29 +407,3 @@ def strip_verbose_pattern(pattern): newpattern += c i += 1 return newpattern - - - -def get_pattern(email, pattern_list): - """Returns matched entry in pattern_list if email matches. - Otherwise returns None. - """ - if not pattern_list: - return None - matched = None - for pattern in pattern_list: - if pattern.startswith('^'): - # This is a regular expression match - try: - if re.search(pattern, email, re.IGNORECASE): - matched = pattern - break - except re.error: - # BAW: we should probably remove this pattern - pass - else: - # Do the comparison case insensitively - if pattern.lower() == email.lower(): - matched = pattern - break - return matched diff --git a/src/mailman/app/docs/bans.rst b/src/mailman/app/docs/bans.rst new file mode 100644 index 000000000..bb6d5902e --- /dev/null +++ b/src/mailman/app/docs/bans.rst @@ -0,0 +1,167 @@ +======================= +Banning email addresses +======================= + +Email addresses can be banned from ever subscribing, either to a specific +mailing list or globally within the Mailman system. Both explicit email +addresses and email address patterns can be banned. + +Bans are managed through the `Ban Manager`. + + >>> from zope.component import getUtility + >>> from mailman.interfaces.bans import IBanManager + >>> ban_manager = getUtility(IBanManager) + +At first, no email addresses are banned, either globally... + + >>> ban_manager.is_banned('anne@example.com') + False + +...or for a specific mailing list. + + >>> ban_manager.is_banned('bart@example.com', 'test@example.com') + False + + +Specific bans +============= + +An email address can be banned from a specific mailing list by adding a ban to +the ban manager. + + >>> ban_manager.ban('cris@example.com', 'test@example.com') + >>> ban_manager.is_banned('cris@example.com', 'test@example.com') + True + >>> ban_manager.is_banned('bart@example.com', 'test@example.com') + False + +However, this is not a global ban. + + >>> ban_manager.is_banned('cris@example.com') + False + + +Global bans +=========== + +An email address can be banned globally, so that it cannot be subscribed to +any mailing list. + + >>> ban_manager.ban('dave@example.com') + +Dave is banned from the test mailing list... + + >>> ban_manager.is_banned('dave@example.com', 'test@example.com') + True + +...and the sample mailing list. + + >>> ban_manager.is_banned('dave@example.com', 'sample@example.com') + True + +Dave is also banned globally. + + >>> ban_manager.is_banned('dave@example.com') + True + +Cris however is not banned globally. + + >>> ban_manager.is_banned('cris@example.com') + False + +Even though Cris is not banned globally, we can add a global ban for her. + + >>> ban_manager.ban('cris@example.com') + >>> ban_manager.is_banned('cris@example.com') + True + +Cris is obviously still banned from specific mailing lists. + + >>> ban_manager.is_banned('cris@example.com', 'test@example.com') + True + >>> ban_manager.is_banned('cris@example.com', 'sample@example.com') + True + +We can remove the global ban to once again just ban her address from the test +list. + + >>> ban_manager.unban('cris@example.com') + >>> ban_manager.is_banned('cris@example.com', 'test@example.com') + True + >>> ban_manager.is_banned('cris@example.com', 'sample@example.com') + False + + +Regular expression bans +======================= + +Entire email address patterns can be banned, both for a specific mailing list +and globally, just as specific addresses can be banned. Use this for example, +when an entire domain is a spam faucet. When using a pattern, the email +address must start with a caret (^). + + >>> ban_manager.ban('^.*@example.org', 'test@example.com') + +Now, no one from example.org can subscribe to the test list. + + >>> ban_manager.is_banned('elle@example.org', 'test@example.com') + True + >>> ban_manager.is_banned('eperson@example.org', 'test@example.com') + True + >>> ban_manager.is_banned('elle@example.com', 'test@example.com') + False + +They are not, however banned globally. + + >>> ban_manager.is_banned('elle@example.org', 'sample@example.com') + False + >>> ban_manager.is_banned('elle@example.org') + False + +Of course, we can ban everyone from example.org globally too. + + >>> ban_manager.ban('^.*@example.org') + >>> ban_manager.is_banned('elle@example.org', 'sample@example.com') + True + >>> ban_manager.is_banned('elle@example.org') + True + +We can remove the mailing list ban on the pattern, though the global ban will +still be in place. + + >>> ban_manager.unban('^.*@example.org', 'test@example.com') + >>> ban_manager.is_banned('elle@example.org', 'test@example.com') + True + >>> ban_manager.is_banned('elle@example.org', 'sample@example.com') + True + >>> ban_manager.is_banned('elle@example.org') + True + +But once the global ban is removed, everyone from example.org can subscribe to +the mailing lists. + + >>> ban_manager.unban('^.*@example.org') + >>> ban_manager.is_banned('elle@example.org', 'test@example.com') + False + >>> ban_manager.is_banned('elle@example.org', 'sample@example.com') + False + >>> ban_manager.is_banned('elle@example.org') + False + + +Adding and removing bans +======================== + +It is not an error to add a ban more than once. These are just ignored. + + >>> ban_manager.ban('fred@example.com', 'test@example.com') + >>> ban_manager.ban('fred@example.com', 'test@example.com') + >>> ban_manager.is_banned('fred@example.com', 'test@example.com') + True + +Nor is it an error to remove a ban more than once. + + >>> ban_manager.unban('fred@example.com', 'test@example.com') + >>> ban_manager.unban('fred@example.com', 'test@example.com') + >>> ban_manager.is_banned('fred@example.com', 'test@example.com') + False diff --git a/src/mailman/app/membership.py b/src/mailman/app/membership.py index 32ba1ff42..8723fd781 100644 --- a/src/mailman/app/membership.py +++ b/src/mailman/app/membership.py @@ -34,6 +34,7 @@ from mailman.app.notifications import send_goodbye_message from mailman.core.i18n import _ from mailman.email.message import OwnerNotification from mailman.interfaces.address import IEmailValidator +from mailman.interfaces.bans import IBanManager from mailman.interfaces.member import ( AlreadySubscribedError, MemberRole, MembershipIsBannedError, NotAMemberError) @@ -71,10 +72,8 @@ def add_member(mlist, email, realname, password, delivery_mode, language): if mlist.members.get_member(email) is not None: raise AlreadySubscribedError( mlist.fqdn_listname, email, MemberRole.member) - # Check for banned email addresses here too for administrative mass - # subscribes and confirmations. - pattern = Utils.get_pattern(email, mlist.ban_list) - if pattern: + # Check to see if the email address is banned. + if getUtility(IBanManager).is_banned(email, mlist.fqdn_listname): raise MembershipIsBannedError(mlist, email) # See if there's already a user linked with the given address. user_manager = getUtility(IUserManager) diff --git a/src/mailman/app/tests/test_membership.py b/src/mailman/app/tests/test_membership.py index dc8009f65..b0e1bae5d 100644 --- a/src/mailman/app/tests/test_membership.py +++ b/src/mailman/app/tests/test_membership.py @@ -32,7 +32,8 @@ from zope.component import getUtility from mailman.app.lifecycle import create_list from mailman.app.membership import add_member from mailman.core.constants import system_preferences -from mailman.interfaces.member import DeliveryMode +from mailman.interfaces.bans import IBanManager +from mailman.interfaces.member import DeliveryMode, MembershipIsBannedError from mailman.interfaces.usermanager import IUserManager from mailman.testing.helpers import reset_the_world from mailman.testing.layers import ConfigLayer @@ -68,6 +69,60 @@ class AddMemberTest(unittest.TestCase): self.assertEqual(member.address.email, 'aperson@example.com') self.assertEqual(member.mailing_list, 'test@example.com') + def test_add_member_banned(self): + # Test that members who are banned by specific address cannot + # subscribe to the mailing list. + getUtility(IBanManager).ban('anne@example.com', 'test@example.com') + self.assertRaises( + MembershipIsBannedError, + add_member, self._mlist, 'anne@example.com', 'Anne Person', + '123', DeliveryMode.regular, system_preferences.preferred_language) + + def test_add_member_globally_banned(self): + # Test that members who are banned by specific address cannot + # subscribe to the mailing list. + getUtility(IBanManager).ban('anne@example.com') + self.assertRaises( + MembershipIsBannedError, + add_member, self._mlist, 'anne@example.com', 'Anne Person', + '123', DeliveryMode.regular, system_preferences.preferred_language) + + def test_add_member_banned_from_different_list(self): + # Test that members who are banned by specific address cannot + # subscribe to the mailing list. + getUtility(IBanManager).ban('anne@example.com', 'sample@example.com') + member = add_member(self._mlist, 'anne@example.com', + 'Anne Person', '123', DeliveryMode.regular, + system_preferences.preferred_language) + self.assertEqual(member.address.email, 'anne@example.com') + + def test_add_member_banned_by_pattern(self): + # Test that members who are banned by specific address cannot + # subscribe to the mailing list. + getUtility(IBanManager).ban('^.*@example.com', 'test@example.com') + self.assertRaises( + MembershipIsBannedError, + add_member, self._mlist, 'anne@example.com', 'Anne Person', + '123', DeliveryMode.regular, system_preferences.preferred_language) + + def test_add_member_globally_banned_by_pattern(self): + # Test that members who are banned by specific address cannot + # subscribe to the mailing list. + getUtility(IBanManager).ban('^.*@example.com') + self.assertRaises( + MembershipIsBannedError, + add_member, self._mlist, 'anne@example.com', 'Anne Person', + '123', DeliveryMode.regular, system_preferences.preferred_language) + + def test_add_member_banned_from_different_list_by_pattern(self): + # Test that members who are banned by specific address cannot + # subscribe to the mailing list. + getUtility(IBanManager).ban('^.*@example.com', 'sample@example.com') + member = add_member(self._mlist, 'anne@example.com', + 'Anne Person', '123', DeliveryMode.regular, + system_preferences.preferred_language) + self.assertEqual(member.address.email, 'anne@example.com') + def test_suite(): diff --git a/src/mailman/config/configure.zcml b/src/mailman/config/configure.zcml index bf42d6635..3b4497ab8 100644 --- a/src/mailman/config/configure.zcml +++ b/src/mailman/config/configure.zcml @@ -16,6 +16,11 @@ provides="mailman.interfaces.mailinglist.IAcceptableAliasSet" /> + + . + +"""Manager of email address bans.""" + +from __future__ import absolute_import, unicode_literals + +__metaclass__ = type +__all__ = [ + 'IBan', + 'IBanManager', + ] + + +from zope.interface import Attribute, Interface + + + +class IBan(Interface): + """A specific ban. + + In general, this interface isn't publicly useful. + """ + + email = Attribute('The banned email address, or pattern.') + + mailing_list = Attribute( + """The fqdn name of the mailing list the ban applies to. + + Use None if this is a global ban. + """) + + + +class IBanManager(Interface): + """The global manager of email address bans.""" + + def ban(email, mailing_list=None): + """Ban an email address from subscribing to a mailing list. + + When an email address is banned, it will not be allowed to subscribe + to a the named mailing list. This does not affect any email address + already subscribed to the mailing list. With the default arguments, + an email address can be banned globally from subscribing to any + mailing list on the system. + + It is also possible to add a 'ban pattern' whereby all email addresses + matching a Python regular expression can be banned. This is + accomplished by using a `^` as the first character in `email`. + + When an email address is already banned for the given mailing list (or + globally), then this method does nothing. However, it is possible to + extend a ban for a specific mailing list into a global ban; both bans + would be in place and they can be removed individually. + + :param email: The text email address being banned or, if the string + starts with a caret (^), the email address pattern to ban. + :type email: str + :param mailing_list: The fqdn name of the mailing list to which the + ban applies. If None, then the ban is global. + :type mailing_list: string + """ + + def unban(email, mailing_list=None): + """Remove an email address ban. + + This removes a specific or global email address ban, which would have + been added with the `ban()` method. If a ban is lifted which did not + exist, this method does nothing. + + :param email: The text email address being unbanned or, if the string + starts with a caret (^), the email address pattern to unban. + :type email: str + :param mailing_list: The fqdn name of the mailing list to which the + unban applies. If None, then the unban is global. + :type mailing_list: string + """ + + def is_banned(email, mailing_list=None): + """Check whether a specific email address is banned. + + `email` must be a text email address; it cannot be a pattern. The + given email address is checked against all registered bans, both + specific and regular expression, both for the named mailing list (if + given), and globally. + + :param email: The text email address being checked. + :type email: str + :param mailing_list: The fqdn name of the mailing list being checked. + Note that if not None, both specific and global bans will be + checked. If None, then only global bans will be checked. + :type mailing_list: string + :return: A flag indicating whether the given email address is banned + or not. + :rtype: bool + """ diff --git a/src/mailman/model/bans.py b/src/mailman/model/bans.py new file mode 100644 index 000000000..95711cfa9 --- /dev/null +++ b/src/mailman/model/bans.py @@ -0,0 +1,110 @@ +# Copyright (C) 2011 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 . + +"""Ban manager.""" + +from __future__ import absolute_import, unicode_literals + +__metaclass__ = type +__all__ = [ + 'BanManager', + ] + + +import re + +from storm.locals import Int, Unicode +from zope.interface import implements + +from mailman.config import config +from mailman.database.model import Model +from mailman.interfaces.bans import IBan, IBanManager + + + +class Ban(Model): + implements(IBan) + + id = Int(primary=True) + email = Unicode() + mailing_list = Unicode() + + def __init__(self, email, mailing_list): + super(Ban, self).__init__() + self.email = email + self.mailing_list = mailing_list + + + +class BanManager: + implements(IBanManager) + + def ban(self, email, mailing_list=None): + """See `IBanManager`.""" + bans = config.db.store.find( + Ban, email=email, mailing_list=mailing_list) + if bans.count() == 0: + ban = Ban(email, mailing_list) + config.db.store.add(ban) + + def unban(self, email, mailing_list=None): + """See `IBanManager`.""" + ban = config.db.store.find( + Ban, email=email, mailing_list=mailing_list).one() + if ban is not None: + config.db.store.remove(ban) + + def is_banned(self, email, mailing_list=None): + """See `IBanManager`.""" + # A specific mailing list ban is being checked, however the email + # address could be banned specifically, or globally. + if mailing_list is not None: + # Try specific bans first. + bans = config.db.store.find( + Ban, email=email, mailing_list=mailing_list) + if bans.count() > 0: + return True + # Try global bans next. + bans = config.db.store.find(Ban, email=email, mailing_list=None) + if bans.count() > 0: + return True + # Now try specific mailing list bans, but with a pattern. + bans = config.db.store.find(Ban, mailing_list=mailing_list) + for ban in bans: + if (ban.email.startswith('^') and + re.match(ban.email, email, re.IGNORECASE) is not None): + return True + # And now try global pattern bans. + bans = config.db.store.find(Ban, mailing_list=None) + for ban in bans: + if (ban.email.startswith('^') and + re.match(ban.email, email, re.IGNORECASE) is not None): + return True + else: + # The client is asking for global bans. Look up bans on the + # specific email address first. + bans = config.db.store.find( + Ban, email=email, mailing_list=None) + if bans.count() > 0: + return True + # And now look for global pattern bans. + bans = config.db.store.find(Ban, mailing_list=None) + for ban in bans: + if (ban.email.startswith('^') and + re.match(ban.email, email, re.IGNORECASE) is not None): + return True + return False diff --git a/src/mailman/model/mailinglist.py b/src/mailman/model/mailinglist.py index cdebd5ca6..488b6da3d 100644 --- a/src/mailman/model/mailinglist.py +++ b/src/mailman/model/mailinglist.py @@ -108,8 +108,7 @@ class MailingList(Model): filter_content = Bool() collapse_alternatives = Bool() convert_html_to_plaintext = Bool() - # Bounces and bans. - ban_list = Pickle() # XXX + # Bounces. bounce_info_stale_after = TimeDelta() # XXX bounce_matching_headers = Unicode() # XXX bounce_notify_owner_on_disable = Bool() # XXX diff --git a/src/mailman/styles/default.py b/src/mailman/styles/default.py index f35b2519a..4f2b3d7d5 100644 --- a/src/mailman/styles/default.py +++ b/src/mailman/styles/default.py @@ -131,8 +131,6 @@ ${listinfo_page} mlist.forward_auto_discards = True mlist.generic_nonmember_action = 1 mlist.nonmember_rejection_notice = '' - # Ban lists - mlist.ban_list = [] # Max autoresponses per day. A mapping between addresses and a # 2-tuple of the date of the last autoresponse and the number of # autoresponses sent on that date. diff --git a/src/mailman/tests/test_documentation.py b/src/mailman/tests/test_documentation.py index 30bcdccc3..23cc189d0 100644 --- a/src/mailman/tests/test_documentation.py +++ b/src/mailman/tests/test_documentation.py @@ -253,7 +253,7 @@ def test_suite(): layer = getattr(sys.modules[package_path], 'layer', SMTPLayer) for filename in os.listdir(docsdir): base, extension = os.path.splitext(filename) - if os.path.splitext(filename)[1] == '.txt': + if os.path.splitext(filename)[1] in ('.txt', '.rst'): module_path = package_path + '.' + base doctest_files[module_path] = ( os.path.join(docsdir, filename), layer) -- cgit v1.2.3-70-g09d2