[IMP] base: test against a real SMTP server
A previous commit broke the smtp authentication using a TLS certificate and we only figured it out after that a client created a support ticket several weeks later. It turns out that there are no tests that validate the various ways outgoing mail servers can be configured. In this work, we add a test suite where a local smtp server is started and controlled during the test execution. This makes it possible to test all the possible outgoing mail server configurations, including TLS. This work revealed several problems that have been sorted in other PRs, a problem that is left to solve is to verify those certificates as shown by the `test_man_in_the_middle` test. This will be sorted in a future work. We chose [aiosmtpd] which is a pure-python lightweight SMTP server that aims at providing a programming API that is well-suited to be used inside unittests. task-3703209 opw-3640374 [aiosmtpd]: https://aiosmtpd.readthedocs.io Part-of: odoo/odoo#151483
This commit is contained in:
Julien Castiaux
@@ -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:
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
*.pem
|
||||
index*
|
||||
serial*
|
||||
Executable
+84
@@ -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'
|
||||
@@ -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 <https://en.wikipedia.org/wiki/Certificate_signing_request>.
|
||||
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
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user