diff --git a/odoo/addons/base/models/ir_mail_server.py b/odoo/addons/base/models/ir_mail_server.py index 25dafa77ebf..c0ee60c802f 100644 --- a/odoo/addons/base/models/ir_mail_server.py +++ b/odoo/addons/base/models/ir_mail_server.py @@ -238,7 +238,7 @@ class IrMailServer(models.Model): # Testing the MAIL FROM step should detect sender filter problems (code, repl) = smtp.mail(email_from) if code != 250: - raise UserError(_('The server refused the sender address (%(email_from)s) with error %(repl)s', email_from=email_from, repl=repl)) + raise UserError(_('The server refused the sender address (%(email_from)s) with error %(repl)s', email_from=email_from, repl=repl)) # noqa: TRY301 # Testing the RCPT TO step should detect most relaying problems (code, repl) = smtp.rcpt(email_to) if code not in (250, 251): @@ -248,26 +248,26 @@ class IrMailServer(models.Model): smtp.putcmd("data") (code, repl) = smtp.getreply() if code != 354: - raise UserError(_('The server refused the test connection with error %(repl)s', repl=repl)) - except UserError as e: - # let UserErrors (messages) bubble up - raise e + raise UserError(_('The server refused the test connection with error %(repl)s', repl=repl)) # noqa: TRY301 except (UnicodeError, idna.core.InvalidCodepoint) as e: - raise UserError(_("Invalid server name!\n %s", ustr(e))) + raise UserError(_("Invalid server name!\n %s", e)) from e except (gaierror, timeout) as e: - raise UserError(_("No response received. Check server address and port number.\n %s", ustr(e))) + raise UserError(_("No response received. Check server address and port number.\n %s", e)) from e except smtplib.SMTPServerDisconnected as e: - raise UserError(_("The server has closed the connection unexpectedly. Check configuration served on this port number.\n %s", ustr(e.strerror))) + raise UserError(_("The server has closed the connection unexpectedly. Check configuration served on this port number.\n %s", e)) from e except smtplib.SMTPResponseException as e: - raise UserError(_("Server replied with following exception:\n %s", ustr(e.smtp_error))) + raise UserError(_("Server replied with following exception:\n %s", e)) from e except smtplib.SMTPNotSupportedError as e: - raise UserError(_("An option is not supported by the server:\n %s", e.strerror)) + raise UserError(_("An option is not supported by the server:\n %s", e)) from e except smtplib.SMTPException as e: - raise UserError(_("An SMTP exception occurred. Check port number and connection security type.\n %s", ustr(e))) - except SSLError as e: - raise UserError(_("An SSL exception occurred. Check connection security type.\n %s", ustr(e))) + raise UserError(_("An SMTP exception occurred. Check port number and connection security type.\n %s", e)) from e + except (ssl.SSLError, SSLError) as e: + raise UserError(_("An SSL exception occurred. Check connection security type.\n %s", e)) from e + except UserError: + raise except Exception as e: - raise UserError(_("Connection Test Failed! Here is what we got instead:\n %s", ustr(e))) + _logger.warning("Connection test on %s failed with a generic error.", server, exc_info=True) + raise UserError(_("Connection Test Failed! Here is what we got instead:\n %s", e)) from e finally: try: if smtp: diff --git a/odoo/addons/base/tests/__init__.py b/odoo/addons/base/tests/__init__.py index d61483e8439..4f1d9aba68f 100644 --- a/odoo/addons/base/tests/__init__.py +++ b/odoo/addons/base/tests/__init__.py @@ -24,6 +24,7 @@ from . import test_ir_cron from . import test_ir_filters from . import test_ir_http from . import test_ir_mail_server +from . import test_ir_mail_server_smtpd from . import test_ir_model from . import test_ir_module from . import test_ir_sequence diff --git a/odoo/addons/base/tests/ssl/.gitignore b/odoo/addons/base/tests/ssl/.gitignore new file mode 100644 index 00000000000..4476a0fed98 --- /dev/null +++ b/odoo/addons/base/tests/ssl/.gitignore @@ -0,0 +1,3 @@ +*.pem +index* +serial* diff --git a/odoo/addons/base/tests/ssl/gencert.sh b/odoo/addons/base/tests/ssl/gencert.sh new file mode 100755 index 00000000000..d0755f040bd --- /dev/null +++ b/odoo/addons/base/tests/ssl/gencert.sh @@ -0,0 +1,84 @@ +# Use openssl to generate multiple signed RSA key/cert pairs. +# +# usage: cd path/to/dir/of/gencert.sh/; ./gencert.sh +# +# This script runs multiple openssl commands to create a fake certificate +# authority (CA) for the fake city of "Houtsiplou". +# https://fr.wiktionary.org/wiki/Houte-Si-Plou +# +# It then uses that CA to generate and sign 3 other certificates for various +# purposes: a client pair, a server pair, and an "any purpose" pair that is not +# restricted to client or server side. It also generates a self-signed pair to +# use as an "invalid" pair for testing. +# +# In Python you'll want to `load_verify_locations(cafile='path/to/ca.cert.pem')` +# to trust the certification authority and the other pairs that were signed +# with it. Note that it won't trust the autorities embedded in your OS anymore. +# https://docs.python.org/3/library/ssl.html#ssl.SSLContext.load_verify_locations +# +# This script and configuration were largely inspired by +# https://jamielinux.com/docs/openssl-certificate-authority/ + +set -o errexit +set -o nounset + +if test ! -f openssl.conf +then + echo Cannot find openssl.conf in the working directory. + exit 1 +fi + +# reset auto generated files +rm -f *.pem index* serial* +touch index.txt +echo 1000 > serial + +set -o xtrace + +# create root cA +openssl genpkey -config openssl.conf -quiet -algorithm RSA -out ca.key.pem +openssl req -config openssl.conf -x509 -days 1000 \ + -extensions ca_cert -key ca.key.pem -out ca.cert.pem \ + -subj '/C=BE/L=Houtesiplou/CN=Houtesiplou Certification Authority' +openssl x509 -noout -text -in ca.cert.pem + +# create any purpose +#openssl genpkey -config openssl.conf -quiet -algorithm RSA -out any_purpose.key.pem +#openssl req -config openssl.conf -new \ +# -key any_purpose.key.pem -out any_purpose.csr.pem \ +# -subj '/C=BE/L=Houtesiplou/CN=Houtesiplou Any Purpose' +#openssl ca -config openssl.conf -notext -md sha256 -batch -days 1000 -notext \ +# -extensions any_cert -keyfile ca.key.pem -cert ca.cert.pem \ +# -in any_purpose.csr.pem -out any_purpose.cert.pem +#openssl x509 -noout -text -in any_purpose.cert.pem + +# create client +openssl genpkey -config openssl.conf -quiet -algorithm RSA -out client.key.pem +openssl req -config openssl.conf -new \ + -key client.key.pem -out client.csr.pem \ + -subj '/C=BE/L=Houtesiplou/CN=Houtesiplou Client' +openssl ca -config openssl.conf -notext -md sha256 -batch -days 1000 -notext \ + -extensions client_cert -keyfile ca.key.pem -cert ca.cert.pem \ + -in client.csr.pem -out client.cert.pem +openssl x509 -noout -text -in client.cert.pem + +# create server +openssl genpkey -config openssl.conf -quiet -algorithm RSA -out server.key.pem +openssl req -config openssl.conf -new \ + -key server.key.pem -out server.csr.pem \ + -subj '/C=BE/L=Houtesiplou/CN=Houtesiplou Server' +openssl ca -config openssl.conf -notext -md sha256 -batch -days 1000 -notext \ + -extensions server_cert -keyfile ca.key.pem -cert ca.cert.pem \ + -in server.csr.pem -out server.cert.pem +openssl x509 -noout -text -in server.cert.pem + +# create untrusted self-signed pair +openssl req -new -x509 -noenc -days 1000 -subj '/CN=SelfSigned Lmtd' \ + -out self_signed.cert.pem -keyout self_signed.key.pem + +# remove useless files +rm *.csr.pem *ca.key.pem + +set +o xtrace +echo -e "\nDone! Here are your files:" +find $(pwd) -name '*.*.pem' diff --git a/odoo/addons/base/tests/ssl/openssl.conf b/odoo/addons/base/tests/ssl/openssl.conf new file mode 100644 index 00000000000..b0cf28abd48 --- /dev/null +++ b/odoo/addons/base/tests/ssl/openssl.conf @@ -0,0 +1,84 @@ +# OpenSSL configuration file. +# Largely inspired from https://jamielinux.com/docs/openssl-certificate-authority/appendix/root-configuration-file.html + +[ ca ] +# `man ca` +default_ca = CA_default + +[ CA_default ] +# Directory and file locations. +certs = . +new_certs_dir = . +database = ./index.txt +serial = ./serial +RANDFILE = ./.rand + +name_opt = ca_default +cert_opt = ca_default +default_days = 1 +preserve = no +default_md = sha256 +policy = policy0 + +[ policy0 ] +# See the POLICY FORMAT section of `man ca`. +countryName = match +localityName = match +commonName = supplied +emailAddress = optional + +[ req ] +# Options for the `req` tool (`man req`). +default_bits = 2048 +distinguished_name = req_distinguished_name +string_mask = utf8only +default_md = sha256 + +# Extension to add when the -x509 option is used. +x509_extensions = ca_cert + +[ req_distinguished_name ] +# See . +countryName = Country Name (2 letter code) +localityName = Locality Name +commonName = Common Name +emailAddress = Email Address + +[ ca_cert ] +# Extensions for a typical CA (`man x509v3_config`). +subjectKeyIdentifier = hash +authorityKeyIdentifier = keyid:always,issuer +basicConstraints = critical, CA:true +keyUsage = critical, digitalSignature, cRLSign, keyCertSign + +[ any_cert ] +# Extensions for client certificates (`man x509v3_config`). +basicConstraints = CA:FALSE +nsCertType = server, client, email +nsComment = "OpenSSL Generated Any Purpose Certificate" +subjectKeyIdentifier = hash +authorityKeyIdentifier = keyid,issuer +keyUsage = critical, nonRepudiation, digitalSignature, keyEncipherment + + +[ client_cert ] +# Extensions for client certificates (`man x509v3_config`). +basicConstraints = CA:FALSE +nsCertType = client, email +nsComment = "OpenSSL Generated Client Certificate" +subjectKeyIdentifier = hash +authorityKeyIdentifier = keyid,issuer +keyUsage = critical, nonRepudiation, digitalSignature, keyEncipherment +extendedKeyUsage = clientAuth, emailProtection + + +[ server_cert ] +# Extensions for server certificates (`man x509v3_config`). +basicConstraints = CA:FALSE +nsCertType = server +nsComment = "OpenSSL Generated Server Certificate" +subjectKeyIdentifier = hash +authorityKeyIdentifier = keyid,issuer:always +keyUsage = critical, digitalSignature, keyEncipherment +extendedKeyUsage = serverAuth +subjectAltName = DNS:localhost diff --git a/odoo/addons/base/tests/test_ir_mail_server_smtpd.py b/odoo/addons/base/tests/test_ir_mail_server_smtpd.py new file mode 100644 index 00000000000..814046b46b9 --- /dev/null +++ b/odoo/addons/base/tests/test_ir_mail_server_smtpd.py @@ -0,0 +1,462 @@ +import contextlib +import logging +import shutil +import smtplib +import socket +import ssl +import subprocess +import sys +import tempfile +import unittest +import warnings +from base64 import b64encode +from pathlib import Path +from unittest.mock import patch +from socket import getaddrinfo # keep a reference on the non-patched function + +from odoo.exceptions import UserError +from odoo.tools import file_path, mute_logger +from .common import TransactionCaseWithUserDemo + +try: + import aiosmtpd + import aiosmtpd.controller + import aiosmtpd.smtp + import aiosmtpd.handlers +except ImportError: + aiosmtpd = None + + +PASSWORD = 'secretpassword' +_openssl = shutil.which('openssl') +_logger = logging.getLogger(__name__) + + +def _find_free_local_address(): + """ Get a triple (family, address, port) on which it possible to bind + a local tcp service. """ + addr = aiosmtpd.controller.get_localhost() # it returns 127.0.0.1 or ::1 + family = socket.AF_INET if addr == '127.0.0.1' else socket.AF_INET6 + with socket.socket(family, socket.SOCK_STREAM) as sock: + sock.bind((addr, 0)) + port = sock.getsockname()[1] + return family, addr, port + + +def _smtp_authenticate(server, session, enveloppe, mechanism, data): + """ Callback method used by aiosmtpd to validate a login/password pair. """ + result = aiosmtpd.smtp.AuthResult(success=data.password == PASSWORD.encode()) + _logger.debug("AUTH %s", "successfull" if result.success else "failed") + return result + + +class Certificate: + def __init__(self, key, cert): + self.key = key + self.cert = cert + + def __repr__(self): + return f"Certificate({self.key=}, {self.cert=})" + + +def _generate_certificates(tempdir): + """Generate and sign a few RSA key pairs inside a temporary directory. + + It invokes the openssl command line to create a new Certification + Authority and then use that CA to generate extra keys and + certificates for a user, a server. + + It also creates a self-signed certificate, to test that invalid + certificates are correctly rejected. + + It returns 4 Certificate: CA, client, server, self-signed. Each + Certificate has two attributes: key and cert, both are paths. + """ + certdir = Path(tempdir.name) + + # Create a fake certificate authority and use it to generate a few + # signed certificates. + shutil.copy(file_path('base/tests/ssl/openssl.conf'), certdir) + shutil.copy(file_path('base/tests/ssl/gencert.sh'), certdir) + subprocess.run( + [shutil.which('sh'), certdir / 'gencert.sh'], + check=True, + cwd=certdir, + stdout=sys.stdout if _logger.isEnabledFor(logging.DEBUG) else subprocess.DEVNULL, + stderr=subprocess.STDOUT, + ) + + # Collect files and return them + return ( + Certificate(key=None, cert=certdir / 'ca.cert.pem'), + Certificate(key=certdir / 'client.key.pem', cert=certdir / 'client.cert.pem'), + Certificate(key=certdir / 'server.key.pem', cert=certdir / 'server.cert.pem'), + Certificate(key=certdir / 'self_signed.key.pem', cert=certdir / 'self_signed.cert.pem'), + ) + + +# skip when optional dependencies are not found +@unittest.skipUnless(aiosmtpd, "aiosmtpd couldn't be imported") +@unittest.skipUnless(_openssl, "openssl not found in path") +# fail fast for timeout errors +@patch('odoo.addons.base.models.ir_mail_server.SMTP_TIMEOUT', .1) +# prevent the CLI from interfering with the tests +@patch.dict('odoo.tools.config.options', {'smtp_server': ''}) +class TestIrMailServerSMTPD(TransactionCaseWithUserDemo): + @classmethod + def setUpClass(cls): + super().setUpClass() + + # aiosmtpd emits deprecation warnings because it uses its own + # deprecated features, mute those logs. + # https://github.com/aio-libs/aiosmtpd/issues/347 + class Session(aiosmtpd.smtp.Session): + @property + def login_data(self): + return self._login_data + @login_data.setter + def login_data(self, value): + self._login_data = value + patcher = patch('aiosmtpd.smtp.Session', Session) + patcher.start() + cls.addClassCleanup(patcher.stop) + + # aiosmtpd emits warnings for some unusual configuration, like + # requiring AUTH on a clear-text transport. Mute those logs + # since we also test those unusual configurations. + warnings.filterwarnings( + 'ignore', + "Requiring AUTH while not requiring TLS can lead to security vulnerabilities!", + category=UserWarning + ) + class CustomFilter(logging.Filter): + def filter(self, record): + if record.msg == "auth_required == True but auth_require_tls == False": + return False + if record.msg == "tls_context.verify_mode not in {CERT_NONE, CERT_OPTIONAL}; this might cause client connection problems": + return False + return True + logging.getLogger('mail.log').addFilter(CustomFilter()) + + # decrease aiosmtpd verbosity, odoo INFO = aiosmtpd WARNING + logging.getLogger('mail.log').setLevel(_logger.getEffectiveLevel() + 10) + + # create a few certificates for ssl/tls+starttls and mock the + # various smtplib classes into trusting the created root CA. + _logger.info("Using openssl to generate fake certificates, show " + "subprocess output with: --log-handler %s:DEBUG", __name__) + tempdir = tempfile.TemporaryDirectory(prefix='odoo-test-smtpd-') + cls.addClassCleanup(tempdir.cleanup) + cls.ssl_ca, cls.ssl_client, cls.ssl_server, cls.ssl_self_signed = \ + _generate_certificates(tempdir) + + class TEST_SMTP(smtplib.SMTP): + def starttls(self, *, context): + if context is None: + context = ssl._create_stdlib_context() # what SMTP_SSL does + # context = ssl.create_default_context() # what it should do + context.load_verify_locations(cafile=str(cls.ssl_ca.cert)) + super().starttls(context=context) + patcher = patch('smtplib.SMTP', TEST_SMTP) + patcher.start() + cls.addClassCleanup(patcher.stop) + + class TEST_SMTP_SSL(smtplib.SMTP_SSL): + def _get_socket(self, *args, **kwargs): + # self.context = ssl.create_default_context() # what it should do + self.context.load_verify_locations(cafile=str(cls.ssl_ca.cert)) + return super()._get_socket(*args, **kwargs) + patcher = patch('smtplib.SMTP_SSL', TEST_SMTP_SSL) + patcher.start() + cls.addClassCleanup(patcher.stop) + + # reactivate sending emails during this test suite, make sure + # NOT TO send emails using another ir.mail_server than the one + # created in setUp! + patcher = patch.object(cls.registry['ir.mail_server'], '_is_test_mode') + mock = patcher.start() + mock.return_value = False + cls.addClassCleanup(patcher.stop) + + # fix runbot, docker uses a single ipv4 stack but it gives ::1 + # when resolving "localhost" (so stupid), use the following to + # force aiosmtpd/odoo to bind/connect to a fixed ipv4 OR ipv6 + # address. + family, _, cls.port = _find_free_local_address() + cls.localhost = getaddrinfo('localhost', cls.port, family) + cls.startClassPatcher(patch('socket.getaddrinfo', cls.getaddrinfo)) + + @classmethod + def getaddrinfo(cls, host, port, *args, **kwargs): + """ + Resolve both "localhost" and "notlocalhost" on the ip address + bound by aiosmtpd inside `start_smtpd`. + """ + if host in ('localhost', 'notlocalhost') and port == cls.port: + return cls.localhost + return getaddrinfo(host, port, family=0, type=0, proto=0, flags=0) + + @contextlib.contextmanager + def start_smtpd( + self, encryption, ssl_context=None, auth_required=True, stop_on_cleanup=True + ): + """ + Start a smtp daemon in a background thread, stop it upon exiting + the context manager. + + :param encryption: 'none', 'ssl' or 'starttls', the kind of + server to start. + :param ssl_context: the ``ssl.SSLContext`` object to use with + 'ssl' or 'starttls'. + :param auth_required: whether the server enforces password + authentication or not. + """ + assert encryption in ('none', 'ssl', 'starttls') + assert encryption == 'none' or ssl_context + + kwargs = {} + if encryption == 'starttls': + # for aiosmtpd.smtp.SMTP + kwargs.update({ + 'require_starttls': True, + 'tls_context': ssl_context, + }) + elif encryption == 'ssl': + # for aiosmtpd.controller.InetMixin + kwargs['ssl_context'] = ssl_context + if auth_required: + kwargs['authenticator'] = _smtp_authenticate + + smtpd_thread = aiosmtpd.controller.Controller( + aiosmtpd.handlers.Debugging(), + hostname=aiosmtpd.controller.get_localhost(), + server_hostname='localhost', + port=self.port, + auth_required=auth_required, + auth_require_tls=False, + enable_SMTPUTF8=True, + **kwargs, + ) + try: + smtpd_thread.start() + yield smtpd_thread + finally: + smtpd_thread.stop() + + @mute_logger('mail.log') + def test_authentication_certificate_matrix(self): + """ + Connect to a server that is authenticating users via a TLS + certificate. Test the various possible configurations (missing + cert, invalid cert and valid cert) against both a STARTTLS and + a SSL/TLS SMTP server. + """ + mail_server = self.env['ir.mail_server'].create({ + 'name': 'test smtpd', + 'from_filter': 'localhost', + 'smtp_host': 'localhost', + 'smtp_port': self.port, + 'smtp_authentication': 'login', + 'smtp_user': '', + 'smtp_pass': '', + }) + + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ssl_context.load_cert_chain(self.ssl_server.cert, self.ssl_server.key) + ssl_context.load_verify_locations(cafile=self.ssl_ca.cert) + ssl_context.verify_mode = ssl.CERT_REQUIRED + + self_signed_key = b64encode(self.ssl_self_signed.key.read_bytes()) + self_signed_cert = b64encode(self.ssl_self_signed.cert.read_bytes()) + client_key = b64encode(self.ssl_client.key.read_bytes()) + client_cert = b64encode(self.ssl_client.cert.read_bytes()) + matrix = [ + # authentication, name, certificate, private key, error pattern + ('login', "missing", '', '', + r"The server has closed the connection unexpectedly\. " + r"Check configuration served on this port number\.\n " + r"Connection unexpectedly closed"), + ('certificate', "self signed", self_signed_cert, self_signed_key, + r"The server has closed the connection unexpectedly\. " + r"Check configuration served on this port number\.\n " + r"Connection unexpectedly closed"), + ('certificate', "valid client", client_cert, client_key, None), + ] + + for encryption in ('starttls', 'ssl'): + mail_server.smtp_encryption = encryption + with self.start_smtpd(encryption, ssl_context, auth_required=False): + for authentication, name, certificate, private_key, error_pattern in matrix: + with self.subTest(encryption=encryption, certificate=name): + mail_server.write({ + 'smtp_authentication': authentication, + 'smtp_ssl_certificate': certificate, + 'smtp_ssl_private_key': private_key, + }) + if error_pattern: + with self.assertRaises(UserError) as error_capture: + mail_server.test_smtp_connection() + self.assertRegex(error_capture.exception.args[0], error_pattern) + else: + mail_server.test_smtp_connection() + + + def test_authentication_login_matrix(self): + """ + Connect to a server that is authenticating users via a login/pwd + pair. Test the various possible configurations (missing pair, + invalid pair and valid pair) against both a SMTP server without + encryption, a STARTTLS and a SSL/TLS SMTP server. + """ + mail_server = self.env['ir.mail_server'].create({ + 'name': 'test smtpd', + 'from_filter': 'localhost', + 'smtp_host': 'localhost', + 'smtp_port': self.port, + 'smtp_authentication': 'login', + 'smtp_user': '', + 'smtp_pass': '', + }) + + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ssl_context.load_cert_chain(self.ssl_server.cert, self.ssl_server.key) + + MISSING = '' + INVALID = 'bad password' + matrix = [ + # auth_required, password, error_pattern + (False, MISSING, None), + (True, MISSING, + r"The server refused the sender address \(noreply@localhost\) " + r"with error b'5\.7\.0 Authentication required'"), + (True, INVALID, + r"The server has closed the connection unexpectedly\. " + r"Check configuration served on this port number\.\n " + r"Connection unexpectedly closed:.* timed out"), + (True, PASSWORD, None), + ] + + for encryption in ('none', 'starttls', 'ssl'): + mail_server.smtp_encryption = encryption + for auth_required, password, error_pattern in matrix: + mail_server.smtp_user = password and self.user_demo.email + mail_server.smtp_pass = password + with self.subTest(encryption=encryption, + auth_required=auth_required, + password=password): + with self.start_smtpd(encryption, ssl_context, auth_required): + if error_pattern: + with self.assertRaises(UserError) as capture: + mail_server.test_smtp_connection() + self.assertRegex(capture.exception.args[0], error_pattern) + else: + mail_server.test_smtp_connection() + + @mute_logger('mail.log') + def test_encryption_matrix(self): + """ + Connect to a server on a different encryption configuration than + the server is configured. Verify that it crashes with a good + error message. + """ + mail_server = self.env['ir.mail_server'].create({ + 'name': 'test smtpd', + 'from_filter': 'localhost', + 'smtp_host': 'localhost', + 'smtp_port': self.port, + 'smtp_authentication': 'login', + 'smtp_user': '', + 'smtp_pass': '', + }) + + ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ssl_context.load_cert_chain(self.ssl_server.cert, self.ssl_server.key) + + matrix = [ + # client, server, error_pattern + ('none', 'ssl', + r"The server has closed the connection unexpectedly\. " + r"Check configuration served on this port number\.\n " + r"Connection unexpectedly closed: timed out"), + ('none', 'starttls', + r"The server refused the sender address \(noreply@localhost\) with error " + r"b'Must issue a STARTTLS command first'"), + ('starttls', 'none', + r"An option is not supported by the server:\n " + r"STARTTLS extension not supported by server\."), + ('starttls', 'ssl', + r"The server has closed the connection unexpectedly\. " + r"Check configuration served on this port number\.\n " + r"Connection unexpectedly closed: timed out"), + ('ssl', 'none', + r"An SSL exception occurred\. " + r"Check connection security type\.\n " + r".*?wrong version number"), + ('ssl', 'starttls', + r"An SSL exception occurred\. " + r"Check connection security type\.\n " + r".*?wrong version number"), + ] + + for client_encryption, server_encryption, error_pattern in matrix: + with self.subTest(server_encryption=server_encryption, + client_encryption=client_encryption): + mail_server.smtp_encryption = client_encryption + with self.start_smtpd(server_encryption, ssl_context, auth_required=False): + with self.assertRaises(UserError) as capture: + mail_server.test_smtp_connection() + self.assertRegex(capture.exception.args[0], error_pattern) + + def test_man_in_the_middle_matrix(self): + """ + Simulate that a pirate was successful at intercepting the live + traffic in between the Odoo server and the legitimate SMTP + server. + """ + mail_server = self.env['ir.mail_server'].create({ + 'name': 'test smtpd', + 'from_filter': 'localhost', + 'smtp_host': 'localhost', + 'smtp_port': self.port, + 'smtp_authentication': 'login', + 'smtp_user': self.user_demo.email, + 'smtp_pass': PASSWORD, + 'smtp_ssl_certificate': b64encode(self.ssl_client.cert.read_bytes()), + 'smtp_ssl_private_key': b64encode(self.ssl_client.key.read_bytes()), + }) + + cert_good = self.ssl_server + cert_bad = self.ssl_self_signed + host_good = 'localhost' + host_bad = 'notlocalhost' + + # for now it doesn't raise any error for bad cert/host + matrix = [ + # authentication, certificate, hostname, error_pattern + ('login', cert_bad, host_good, None), + ('login', cert_good, host_bad, None), + ('certificate', cert_bad, host_good, None), + ('certificate', cert_good, host_bad, None), + ] + + for encryption in ('starttls', 'ssl'): + for authentication, certificate, hostname, error_pattern in matrix: + mail_server.smtp_host = hostname + mail_server.smtp_authentication = authentication + mail_server.smtp_encryption = encryption + with self.subTest( + encryption=encryption, + authentication=authentication, + cert_good=certificate == cert_good, + host_good=hostname == host_good, + ): + mitm_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + mitm_context.load_cert_chain(certificate.cert, certificate.key) + auth_required = authentication == 'login' + with self.start_smtpd(encryption, mitm_context, auth_required): + if error_pattern: + with self.assertRaises(UserError) as capture: + mail_server.test_smtp_connection() + self.assertRegex(capture.exception.args[0], error_pattern) + else: + mail_server.test_smtp_connection()