combine smtpd and aiosmtpd tests; encapsulate smtp facilities to setUpClass/tearDownClass (behaves like a singleton, doesn't start smtp server per test); don't generate cert every time (too slow by RSA:2048, use short ECC:256 instead);

drastically speedup all smtp-action tests

.github/workflows/main.yml

@@ -68,6 +68,11 @@ jobs:
             #python -m pip install pyasynchat || echo "can't install pyasynchat";
             #python -m pip install pyasyncore || echo "can't install pyasyncore";
           fi
+          # aiosmtpd in test_smtp (for 3.10+, no need to test it everywhere):
+          if dpkg --compare-versions "$F2B_PYV" ge 3.10; then
+            #sudo apt-get -y install python${F2B_PY/2/}-aiosmtpd || echo 'aiosmtpd not available'
+            python -m pip install aiosmtpd || echo 'aiosmtpd not available'
+          fi
 
       - name: Before scripts
         run: |

fail2ban/tests/action_d/test_smtp.py

@@ -30,61 +30,14 @@ else:
 from ..dummyjail import DummyJail
 from ..utils import CONFIG_DIR, asyncserver, Utils, uni_decode
 
-try:
-	import smtpd
 
-	class TestSMTPServer(smtpd.SMTPServer):
+class _SMTPActionTestCase():
 
-		def __init__(self, *args):
+	def _reset_smtpd(self):
-			smtpd.SMTPServer.__init__(self, *args)
+		for a in ('mailfrom', 'org_data', 'data'):
+			if hasattr(self.smtpd, a): delattr(self.smtpd, a)
 		self.ready = False
 
-		def process_message(self, peer, mailfrom, rcpttos, data, **kwargs):
-			self.peer = peer
-			self.mailfrom = mailfrom
-			self.rcpttos = rcpttos
-			self.org_data = data
-			# replace new line (with tab or space) for possible mime translations (word wrap),
-			self.data = re.sub(r"\n[\t ]", " ", uni_decode(data))
-			self.ready = True
-
-
-	class SMTPActionTest(unittest.TestCase):
-
-		def setUp(self):
-			"""Call before every test case."""
-			unittest.F2B.SkipIfCfgMissing(action='smtp.py')
-			super(SMTPActionTest, self).setUp()
-			self.jail = DummyJail()
-			pythonModule = os.path.join(CONFIG_DIR, "action.d", "smtp.py")
-			pythonModuleName = os.path.basename(pythonModule.rstrip(".py"))
-			if sys.version_info >= (3, 3):
-				customActionModule = importlib.machinery.SourceFileLoader(
-					pythonModuleName, pythonModule).load_module()
-			else:
-				customActionModule = imp.load_source(
-					pythonModuleName, pythonModule)
-
-			self.smtpd = TestSMTPServer(("localhost", 0), None)
-			port = self.smtpd.socket.getsockname()[1]
-
-			self.action = customActionModule.Action(
-				self.jail, "test", host="localhost:%i" % port)
-
-			## because of bug in loop (see loop in asyncserver.py) use it's loop instead of asyncore.loop:
-			self._active = True
-			self._loop_thread = threading.Thread(
-				target=asyncserver.loop, kwargs={'active': lambda: self._active})
-			self._loop_thread.daemon = True
-			self._loop_thread.start()
-
-		def tearDown(self):
-			"""Call after every test case."""
-			self.smtpd.close()
-			self._active = False
-			self._loop_thread.join()
-			super(SMTPActionTest, self).tearDown()
-
 	def _exec_and_wait(self, doaction, timeout=3, short=False):
 		if short: timeout /= 25
 		self.smtpd.ready = False
@@ -95,6 +48,7 @@ try:
 		self._exec_and_wait(self.action.start)
 		self.assertEqual(self.smtpd.mailfrom, "fail2ban")
 		self.assertEqual(self.smtpd.rcpttos, ["root"])
+		self.action.ssl = False # ensure it works without TLS as a sanity check
 		self.assertTrue(
 			"Subject: [Fail2Ban] %s: started" % self.jail.name
 			in self.smtpd.data)
@@ -162,6 +116,67 @@ try:
 			(self.action.fromname, self.action.fromaddr) in self.smtpd.data)
 		self.assertEqual(set(self.smtpd.rcpttos), set(["test@example.com", "test2@example.com"]))
 
+try:
+	import smtpd
+
+	class TestSMTPServer(smtpd.SMTPServer):
+
+		def __init__(self, *args):
+			smtpd.SMTPServer.__init__(self, *args)
+			self.ready = False
+
+		def process_message(self, peer, mailfrom, rcpttos, data, **kwargs):
+			self.peer = peer
+			self.mailfrom = mailfrom
+			self.rcpttos = rcpttos
+			self.org_data = data
+			# replace new line (with tab or space) for possible mime translations (word wrap),
+			self.data = re.sub(r"\n[\t ]", " ", uni_decode(data))
+			self.ready = True
+
+
+	class SMTPActionTest(unittest.TestCase, _SMTPActionTestCase):
+
+		def setUpClass():
+			"""Call before tests."""
+			unittest.F2B.SkipIfCfgMissing(action='smtp.py')
+
+			cls = SMTPActionTest
+			cls.smtpd = TestSMTPServer(("localhost", 0), None)
+			cls.port = cls.smtpd.socket.getsockname()[1]
+
+			## because of bug in loop (see loop in asyncserver.py) use it's loop instead of asyncore.loop:
+			cls._active = True
+			cls._loop_thread = threading.Thread(
+				target=asyncserver.loop, kwargs={'active': lambda: cls._active})
+			cls._loop_thread.daemon = True
+			cls._loop_thread.start()
+
+		def tearDownClass():
+			"""Call after tests."""
+			cls = SMTPActionTest
+			cls.smtpd.close()
+			cls._active = False
+			cls._loop_thread.join()
+
+		def setUp(self):
+			"""Call before every test case."""
+			unittest.F2B.SkipIfCfgMissing(action='smtp.py')
+			super(SMTPActionTest, self).setUp()
+			self.jail = DummyJail()
+			pythonModule = os.path.join(CONFIG_DIR, "action.d", "smtp.py")
+			pythonModuleName = os.path.basename(pythonModule.rstrip(".py"))
+			customActionModule = importlib.machinery.SourceFileLoader(
+				pythonModuleName, pythonModule).load_module()
+
+			self.action = customActionModule.Action(
+				self.jail, "test", host="localhost:%i" % self.port)
+
+		def tearDown(self):
+			"""Call after every test case."""
+			self._reset_smtpd()
+			super(SMTPActionTest, self).tearDown()
+
 except ImportError as e:
 	print("I: Skipping smtp tests: %s" % e)
 
@@ -169,12 +184,10 @@ except ImportError as e:
 try:
 	import asyncio
 	from aiosmtpd.controller import Controller
-	import os
+	import socket
 	import ssl
 	import tempfile
 
-	from OpenSSL import crypto
-
 	class TestSMTPHandler:
 		def __init__(self, *args):
 			self.ready = False
@@ -183,7 +196,9 @@ try:
 			self.peer = session.peer
 			self.mailfrom = envelope.mail_from
 			self.rcpttos = envelope.rcpt_tos
-			self.data = envelope.content.decode()
+			self.org_data = envelope.content.decode()
+			# normalize CRLF -> LF:
+			self.data = re.sub(r"\r\n", "\n", uni_decode(self.org_data))
 			self.ready = True
 			return '250 OK'
 		
@@ -191,55 +206,90 @@ try:
 			print(error)
 			return '542 Internal server error'
 
-	class AIOSMTPActionTest(unittest.TestCase):
+
-		def create_temp_self_signed_cert(self):
+	class AIOSMTPActionTest(unittest.TestCase, _SMTPActionTestCase):
+
+		@classmethod
+		def create_temp_self_signed_cert(cls):
 			"""
-			https://aliceh75.github.io/testing-asyncio-with-ssl
 			Create a self signed SSL certificate in temporary files for host
 				'localhost'
 
 			Returns a tuple containing the certificate file name and the key
 			file name.
 
-			It is the caller's responsibility to delete the files after use
+			The cert (ECC:256, 100years) created with:
+			openssl req -x509 -out /tmp/f2b-localhost.crt -keyout /tmp/f2b-localhost.key -days 36500 -newkey ec:<(openssl ecparam -name prime256v1) -nodes -sha256 \
+			  -subj '/CN=localhost' -extensions EXT -config <( \
+			      printf "[dn]\nCN=localhost\n[req]\ndistinguished_name = dn\n[EXT]\nsubjectAltName=DNS:localhost\nkeyUsage=digitalSignature\nextendedKeyUsage=serverAuth" \
+			    )
+			cat /tmp/f2b-localhost.*
+			rm /tmp/f2b-localhost.*
+
 			"""
-			# create a key pair
+			if hasattr(cls, 'crtfiles'): return cls.crtfiles
-			key = crypto.PKey()
+			cls.crtfiles = crtfiles = (tempfile.mktemp(".crt", "f2b_cert_"), tempfile.mktemp(".key", "f2b_cert_"))
-			key.generate_key(crypto.TYPE_RSA, 2048)
+			with open(crtfiles[0], 'w') as f:
-
+				f.write(
-			# create a self-signed cert
+					'-----BEGIN CERTIFICATE-----\n'
-			cert = crypto.X509()
+					'MIIBhDCCASugAwIBAgIUCuW168kD3G7XrpFwGHwE6vGfoJkwCgYIKoZIzj0EAwIw\n'
-			cert.get_subject().C = "UK"
+					'FDESMBAGA1UEAwwJbG9jYWxob3N0MCAXDTIzMTIzMDE3NDUzNFoYDzIxMjMxMjA2\n'
-			cert.get_subject().ST = "London"
+					'MTc0NTM0WjAUMRIwEAYDVQQDDAlsb2NhbGhvc3QwWTATBgcqhkjOPQIBBggqhkjO\n'
-			cert.get_subject().L = "London"
+					'PQMBBwNCAARDa8BO/UE4axzvnOQ/pCc/ZTp351X1TqIfjEFaMoZOItz1/MW3ZCuS\n'
-			cert.get_subject().O = "myapp"
+					'2vuby3rMn0WZ59RWVotBqA6lcMVcgDq3o1kwVzAUBgNVHREEDTALgglsb2NhbGhv\n'
-			cert.get_subject().OU = "myapp"
+					'c3QwCwYDVR0PBAQDAgeAMBMGA1UdJQQMMAoGCCsGAQUFBwMBMB0GA1UdDgQWBBS8\n'
-			cert.get_subject().CN = 'localhost'
+					'kH1Ucuq+wlex5DxxHDe1kKGdcjAKBggqhkjOPQQDAgNHADBEAiBmv05+BvXWMzLg\n'
-			cert.set_serial_number(1000)
+					'TtF4McoQNrU/0TTKhV8o+mgd+47tMAIgaaSNRnfjGIfJMbXg7Bh53qOIu5+lnm1b\n'
-			cert.gmtime_adj_notBefore(0)
+					'ySygMgFmePs=\n'
-			cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60)
+					'-----END CERTIFICATE-----\n'
-			cert.set_issuer(cert.get_subject())
-			cert.set_pubkey(key)
-			cert.sign(key, 'sha1')
-
-			# Save certificate in temporary file
-			(cert_file_fd, cert_file_name) = tempfile.mkstemp(suffix='.crt', prefix='cert')
-			cert_file = os.fdopen(cert_file_fd, 'wb')
-			cert_file.write(
-				crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
 				)
-			cert_file.close()
+			with open(crtfiles[1], 'w') as f:
-
+				f.write(
-			# Save key in temporary file
+					'-----BEGIN PRIVATE KEY-----\n'
-			(key_file_fd, key_file_name) = tempfile.mkstemp(suffix='.key', prefix='cert')
+					'MIGHAgEAMBMGByqGSM49AgEGCCqGSM49AwEHBG0wawIBAQQgoBGcojKPZMYut7aP\n'
-			key_file = os.fdopen(key_file_fd, 'wb')
+					'JGe2GW+2lVV0zJpgCsZ7816a9uqhRANCAARDa8BO/UE4axzvnOQ/pCc/ZTp351X1\n'
-			key_file.write(
+					'TqIfjEFaMoZOItz1/MW3ZCuS2vuby3rMn0WZ59RWVotBqA6lcMVcgDq3\n'
-				crypto.dump_privatekey(crypto.FILETYPE_PEM, key)
+					'-----END PRIVATE KEY-----\n'
 				)
-			key_file.close()
+			# return file names
+			return crtfiles
 
-			# Return file names
+		@classmethod
-			return (cert_file_name, key_file_name)
+		def _del_cert(cls):
+			if hasattr(cls, 'crtfiles') and cls.crtfiles:
+				for f in cls.crtfiles:
+					try:
+						os.unlink(f)
+					except FileNotFoundError: pass
+				cls.crtfiles = None
+
+		@staticmethod
+		def _free_port():
+			with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+				s.bind(('localhost', 0))
+				return s.getsockname()[1]
+
+		def setUpClass():
+			"""Call before tests."""
+			unittest.F2B.SkipIfCfgMissing(action='smtp.py')
+
+			cert_file, cert_key = AIOSMTPActionTest.create_temp_self_signed_cert()
+			ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+			ssl_context.load_cert_chain(cert_file, cert_key)
+
+			cls = AIOSMTPActionTest
+			cls.port = cls._free_port()
+			cls.smtpd = TestSMTPHandler()
+			cls.controller = Controller(cls.smtpd, hostname='localhost', server_hostname='localhost', port=cls.port,
+				server_kwargs={'tls_context': ssl_context, 'require_starttls': False})
+			# Run the event loop in a separate thread.
+			cls.controller.start()
+
+		def tearDownClass():
+			"""Call after tests."""
+			cls = AIOSMTPActionTest
+			cls.controller.stop()
+			cls._del_cert()
 	
 		def setUp(self):
 			"""Call before every test case."""
@@ -248,72 +298,18 @@ try:
 			self.jail = DummyJail()
 			pythonModule = os.path.join(CONFIG_DIR, "action.d", "smtp.py")
 			pythonModuleName = os.path.basename(pythonModule.rstrip(".py"))
-			if sys.version_info >= (3, 3):
 			customActionModule = importlib.machinery.SourceFileLoader(
 				pythonModuleName, pythonModule).load_module()
-			else:
-				customActionModule = imp.load_source(
-					pythonModuleName, pythonModule)
-
-			cert_file, cert_key = self.create_temp_self_signed_cert()
-			ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
-			ssl_context.load_cert_chain(cert_file, cert_key)
-
-			port = 8025
-			self.smtpd = TestSMTPHandler()
-			self.controller = Controller(self.smtpd, hostname='localhost', server_hostname='localhost', port=port, server_kwargs={'tls_context': ssl_context, 'require_starttls': False})
-			# Run the event loop in a separate thread.
-			self.controller.start()
 
 			self.action = customActionModule.Action(
-				self.jail, "test", host="localhost:%i" % port)
+				self.jail, "test", host="localhost:%i" % self.port)
 
-			## because of bug in loop (see loop in asyncserver.py) use it's loop instead of asyncore.loop:
+			self.action.ssl = True
-			self._active = True
-			self._loop_thread = threading.Thread(
-				target=asyncserver.loop, kwargs={'active': lambda: self._active})
-			self._loop_thread.daemon = True
-			self._loop_thread.start()
 
 		def tearDown(self):
 			"""Call after every test case."""
-			self.controller.stop()
+			self._reset_smtpd()
-			self._active = False
-			self._loop_thread.join()
 			super(AIOSMTPActionTest, self).tearDown()
 	
-		def _exec_and_wait(self, doaction, timeout=3, short=False):
-			if short: timeout /= 25
-			self.smtpd.ready = False
-			doaction()
-			Utils.wait_for(lambda: self.smtpd.ready, timeout)
-
-		def testStart(self):
-			"""
-			Make sure aiosmtpd starts without TLS as a sanity check
-			"""
-
-			self._exec_and_wait(self.action.start)
-			self.assertEqual(self.smtpd.mailfrom, "fail2ban")
-			self.assertEqual(self.smtpd.rcpttos, ["root"])
-			self.assertTrue(
-				"Subject: [Fail2Ban] %s: started" % self.jail.name
-				in self.smtpd.data)
-
-		def testOptionsTls(self):
-			self._exec_and_wait(self.action.start)
-			self.assertEqual(self.smtpd.mailfrom, "fail2ban")
-			self.assertEqual(self.smtpd.rcpttos, ["root"])
-
-			self.action.fromname = "Test"
-			self.action.fromaddr = "test@example.com"
-			self.action.toaddr = "test@example.com, test2@example.com"
-			self.action.ssl = True # Important part
-			self._exec_and_wait(self.action.start)
-			self.assertEqual(self.smtpd.mailfrom, "test@example.com")
-			self.assertTrue("From: %s <%s>" %
-				(self.action.fromname, self.action.fromaddr) in self.smtpd.data)
-			self.assertEqual(set(self.smtpd.rcpttos), set(["test@example.com", "test2@example.com"]))
-			
 except ImportError as e:
 	print("I: Skipping SSL smtp tests: %s" % e)