mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 08:05:31 +02:00
Remove optional dependencies (#4088)
* Stop using already_listening in standalone * remove already_listening * remove psutil entirely * fix #595 * Add basic perform test * make pep8 happy * Add test_perform_eacces * add _setup_perform_error * Add test_perform_unexpected_socket_error * add test_perform_eaddrinuse_no_retry * add test_perform_eaddrinuse_retry * cleanup tests * stop using dnspython * don't install dns extras in tox * remove dns extras from setup.py * Add simple_verify back to DNS response * remove dnspython from oldest tests
This commit is contained in:
Brad Warren
@@ -18,7 +18,6 @@ from certbot import errors
|
||||
from certbot import interfaces
|
||||
|
||||
from certbot.plugins import common
|
||||
from certbot.plugins import util
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
@@ -208,74 +207,38 @@ class Authenticator(common.Plugin):
|
||||
# pylint: disable=unused-argument,missing-docstring
|
||||
return self.supported_challenges
|
||||
|
||||
def _verify_ports_are_available(self, achalls):
|
||||
"""Confirm the ports are available to solve all achalls.
|
||||
|
||||
:param list achalls: list of
|
||||
:class:`~certbot.achallenges.AnnotatedChallenge`
|
||||
|
||||
:raises .errors.MisconfigurationError: if required port is
|
||||
unavailable
|
||||
|
||||
"""
|
||||
ports = []
|
||||
if any(isinstance(ac.chall, challenges.HTTP01) for ac in achalls):
|
||||
ports.append(self.config.http01_port)
|
||||
if any(isinstance(ac.chall, challenges.TLSSNI01) for ac in achalls):
|
||||
ports.append(self.config.tls_sni_01_port)
|
||||
|
||||
renewer = (self.config.verb == "renew")
|
||||
|
||||
if any(util.already_listening(port, renewer) for port in ports):
|
||||
raise errors.MisconfigurationError(
|
||||
"At least one of the required ports is already taken.")
|
||||
|
||||
def perform(self, achalls): # pylint: disable=missing-docstring
|
||||
self._verify_ports_are_available(achalls)
|
||||
return [self._try_perform_single(achall) for achall in achalls]
|
||||
|
||||
try:
|
||||
return self.perform2(achalls)
|
||||
except errors.StandaloneBindError as error:
|
||||
display = zope.component.getUtility(interfaces.IDisplay)
|
||||
def _try_perform_single(self, achall):
|
||||
while True:
|
||||
try:
|
||||
return self._perform_single(achall)
|
||||
except errors.StandaloneBindError as error:
|
||||
_handle_perform_error(error)
|
||||
|
||||
if error.socket_error.errno == socket.errno.EACCES:
|
||||
display.notification(
|
||||
"Could not bind TCP port {0} because you don't have "
|
||||
"the appropriate permissions (for example, you "
|
||||
"aren't running this program as "
|
||||
"root).".format(error.port), force_interactive=True)
|
||||
elif error.socket_error.errno == socket.errno.EADDRINUSE:
|
||||
display.notification(
|
||||
"Could not bind TCP port {0} because it is already in "
|
||||
"use by another process on this system (such as a web "
|
||||
"server). Please stop the program in question and then "
|
||||
"try again.".format(error.port), force_interactive=True)
|
||||
else:
|
||||
raise # XXX: How to handle unknown errors in binding?
|
||||
def _perform_single(self, achall):
|
||||
if isinstance(achall.chall, challenges.HTTP01):
|
||||
server, response = self._perform_http_01(achall)
|
||||
else: # tls-sni-01
|
||||
server, response = self._perform_tls_sni_01(achall)
|
||||
self.served[server].add(achall)
|
||||
return response
|
||||
|
||||
def perform2(self, achalls):
|
||||
"""Perform achallenges without IDisplay interaction."""
|
||||
responses = []
|
||||
def _perform_http_01(self, achall):
|
||||
server = self.servers.run(self.config.http01_port, challenges.HTTP01)
|
||||
response, validation = achall.response_and_validation()
|
||||
resource = acme_standalone.HTTP01RequestHandler.HTTP01Resource(
|
||||
chall=achall.chall, response=response, validation=validation)
|
||||
self.http_01_resources.add(resource)
|
||||
return server, response
|
||||
|
||||
for achall in achalls:
|
||||
if isinstance(achall.chall, challenges.HTTP01):
|
||||
server = self.servers.run(
|
||||
self.config.http01_port, challenges.HTTP01)
|
||||
response, validation = achall.response_and_validation()
|
||||
self.http_01_resources.add(
|
||||
acme_standalone.HTTP01RequestHandler.HTTP01Resource(
|
||||
chall=achall.chall, response=response,
|
||||
validation=validation))
|
||||
else: # tls-sni-01
|
||||
server = self.servers.run(
|
||||
self.config.tls_sni_01_port, challenges.TLSSNI01)
|
||||
response, (cert, _) = achall.response_and_validation(
|
||||
cert_key=self.key)
|
||||
self.certs[response.z_domain] = (self.key, cert)
|
||||
self.served[server].add(achall)
|
||||
responses.append(response)
|
||||
|
||||
return responses
|
||||
def _perform_tls_sni_01(self, achall):
|
||||
port = self.config.tls_sni_01_port
|
||||
server = self.servers.run(port, challenges.TLSSNI01)
|
||||
response, (cert, _) = achall.response_and_validation(cert_key=self.key)
|
||||
self.certs[response.z_domain] = (self.key, cert)
|
||||
return server, response
|
||||
|
||||
def cleanup(self, achalls): # pylint: disable=missing-docstring
|
||||
# reduce self.served and close servers if none challenges are served
|
||||
@@ -286,3 +249,25 @@ class Authenticator(common.Plugin):
|
||||
for port, server in six.iteritems(self.servers.running()):
|
||||
if not self.served[server]:
|
||||
self.servers.stop(port)
|
||||
|
||||
|
||||
def _handle_perform_error(error):
|
||||
if error.socket_error.errno == socket.errno.EACCES:
|
||||
raise errors.PluginError(
|
||||
"Could not bind TCP port {0} because you don't have "
|
||||
"the appropriate permissions (for example, you "
|
||||
"aren't running this program as "
|
||||
"root).".format(error.port))
|
||||
elif error.socket_error.errno == socket.errno.EADDRINUSE:
|
||||
display = zope.component.getUtility(interfaces.IDisplay)
|
||||
msg = (
|
||||
"Could not bind TCP port {0} because it is already in "
|
||||
"use by another process on this system (such as a web "
|
||||
"server). Please stop the program in question and "
|
||||
"then try again.".format(error.port))
|
||||
should_retry = display.yesno(msg, "Retry",
|
||||
"Cancel", default=False)
|
||||
if not should_retry:
|
||||
raise errors.PluginError(msg)
|
||||
else:
|
||||
raise
|
||||
|
||||
@@ -8,11 +8,9 @@ import six
|
||||
|
||||
from acme import challenges
|
||||
from acme import jose
|
||||
from acme import standalone as acme_standalone
|
||||
|
||||
from certbot import achallenges
|
||||
from certbot import errors
|
||||
from certbot import interfaces
|
||||
|
||||
from certbot.tests import acme_util
|
||||
from certbot.tests import util as test_util
|
||||
@@ -114,6 +112,7 @@ def get_open_port():
|
||||
open_socket.close()
|
||||
return port
|
||||
|
||||
|
||||
class AuthenticatorTest(unittest.TestCase):
|
||||
"""Tests for certbot.plugins.standalone.Authenticator."""
|
||||
|
||||
@@ -124,6 +123,7 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
tls_sni_01_port=get_open_port(), http01_port=get_open_port(),
|
||||
standalone_supported_challenges="tls-sni-01,http-01")
|
||||
self.auth = Authenticator(self.config, name="standalone")
|
||||
self.auth.servers = mock.MagicMock()
|
||||
|
||||
def test_supported_challenges(self):
|
||||
self.assertEqual(self.auth.supported_challenges,
|
||||
@@ -146,6 +146,52 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
self.assertEqual(self.auth.get_chall_pref(domain=None),
|
||||
[challenges.TLSSNI01])
|
||||
|
||||
def test_perform(self):
|
||||
achalls = self._get_achalls()
|
||||
response = self.auth.perform(achalls)
|
||||
|
||||
expected = [achall.response(achall.account_key) for achall in achalls]
|
||||
self.assertEqual(response, expected)
|
||||
|
||||
@test_util.patch_get_utility()
|
||||
def test_perform_eaddrinuse_retry(self, mock_get_utility):
|
||||
errno = socket.errno.EADDRINUSE
|
||||
error = errors.StandaloneBindError(mock.MagicMock(errno=errno), -1)
|
||||
self.auth.servers.run.side_effect = [error] + 2 * [mock.MagicMock()]
|
||||
mock_yesno = mock_get_utility.return_value.yesno
|
||||
mock_yesno.return_value = True
|
||||
|
||||
self.test_perform()
|
||||
self._assert_correct_yesno_call(mock_yesno)
|
||||
|
||||
@test_util.patch_get_utility()
|
||||
def test_perform_eaddrinuse_no_retry(self, mock_get_utility):
|
||||
mock_yesno = mock_get_utility.return_value.yesno
|
||||
mock_yesno.return_value = False
|
||||
|
||||
errno = socket.errno.EADDRINUSE
|
||||
self.assertRaises(errors.PluginError, self._fail_perform, errno)
|
||||
self._assert_correct_yesno_call(mock_yesno)
|
||||
|
||||
def _assert_correct_yesno_call(self, mock_yesno):
|
||||
yesno_args, yesno_kwargs = mock_yesno.call_args
|
||||
self.assertTrue("in use" in yesno_args[0])
|
||||
self.assertFalse(yesno_kwargs.get("default", True))
|
||||
|
||||
def test_perform_eacces(self):
|
||||
errno = socket.errno.EACCES
|
||||
self.assertRaises(errors.PluginError, self._fail_perform, errno)
|
||||
|
||||
def test_perform_unexpected_socket_error(self):
|
||||
errno = socket.errno.ENOTCONN
|
||||
self.assertRaises(
|
||||
errors.StandaloneBindError, self._fail_perform, errno)
|
||||
|
||||
def _fail_perform(self, errno):
|
||||
error = errors.StandaloneBindError(mock.MagicMock(errno=errno), -1)
|
||||
self.auth.servers.run.side_effect = error
|
||||
self.auth.perform(self._get_achalls())
|
||||
|
||||
@classmethod
|
||||
def _get_achalls(cls):
|
||||
domain = b'localhost'
|
||||
@@ -157,84 +203,7 @@ class AuthenticatorTest(unittest.TestCase):
|
||||
|
||||
return [http_01, tls_sni_01]
|
||||
|
||||
@mock.patch("certbot.plugins.standalone.util")
|
||||
def test_perform_already_listening(self, mock_util):
|
||||
http_01, tls_sni_01 = self._get_achalls()
|
||||
|
||||
for achall, port in ((http_01, self.config.http01_port,),
|
||||
(tls_sni_01, self.config.tls_sni_01_port)):
|
||||
mock_util.already_listening.return_value = True
|
||||
self.assertRaises(
|
||||
errors.MisconfigurationError, self.auth.perform, [achall])
|
||||
mock_util.already_listening.assert_called_once_with(port, False)
|
||||
mock_util.already_listening.reset_mock()
|
||||
|
||||
@test_util.patch_get_utility()
|
||||
def test_perform(self, unused_mock_get_utility):
|
||||
achalls = self._get_achalls()
|
||||
|
||||
self.auth.perform2 = mock.Mock(return_value=mock.sentinel.responses)
|
||||
self.assertEqual(mock.sentinel.responses, self.auth.perform(achalls))
|
||||
self.auth.perform2.assert_called_once_with(achalls)
|
||||
|
||||
@test_util.patch_get_utility()
|
||||
def _test_perform_bind_errors(self, errno, achalls, mock_get_utility):
|
||||
port = get_open_port()
|
||||
def _perform2(unused_achalls):
|
||||
raise errors.StandaloneBindError(mock.Mock(errno=errno), port)
|
||||
|
||||
self.auth.perform2 = mock.MagicMock(side_effect=_perform2)
|
||||
self.auth.perform(achalls)
|
||||
mock_get_utility.assert_called_once_with(interfaces.IDisplay)
|
||||
notification = mock_get_utility.return_value.notification
|
||||
self.assertEqual(1, notification.call_count)
|
||||
self.assertTrue(str(port) in notification.call_args[0][0])
|
||||
|
||||
def test_perform_eacces(self):
|
||||
# pylint: disable=no-value-for-parameter
|
||||
self._test_perform_bind_errors(socket.errno.EACCES, [])
|
||||
|
||||
def test_perform_eaddrinuse(self):
|
||||
# pylint: disable=no-value-for-parameter
|
||||
self._test_perform_bind_errors(socket.errno.EADDRINUSE, [])
|
||||
|
||||
def test_perfom_unknown_bind_error(self):
|
||||
self.assertRaises(
|
||||
errors.StandaloneBindError, self._test_perform_bind_errors,
|
||||
socket.errno.ENOTCONN, [])
|
||||
|
||||
def test_perform2(self):
|
||||
http_01, tls_sni_01 = self._get_achalls()
|
||||
|
||||
self.auth.servers = mock.MagicMock()
|
||||
|
||||
def _run(port, tls): # pylint: disable=unused-argument
|
||||
return "server{0}".format(port)
|
||||
|
||||
self.auth.servers.run.side_effect = _run
|
||||
responses = self.auth.perform2([http_01, tls_sni_01])
|
||||
|
||||
self.assertTrue(isinstance(responses, list))
|
||||
self.assertEqual(2, len(responses))
|
||||
self.assertTrue(isinstance(responses[0], challenges.HTTP01Response))
|
||||
self.assertTrue(isinstance(responses[1], challenges.TLSSNI01Response))
|
||||
|
||||
self.assertEqual(self.auth.servers.run.mock_calls, [
|
||||
mock.call(self.config.http01_port, challenges.HTTP01),
|
||||
mock.call(self.config.tls_sni_01_port, challenges.TLSSNI01),
|
||||
])
|
||||
self.assertEqual(self.auth.served, {
|
||||
"server" + str(self.config.tls_sni_01_port): set([tls_sni_01]),
|
||||
"server" + str(self.config.http01_port): set([http_01]),
|
||||
})
|
||||
self.assertEqual(1, len(self.auth.http_01_resources))
|
||||
self.assertEqual(1, len(self.auth.certs))
|
||||
self.assertEqual(list(self.auth.http_01_resources), [
|
||||
acme_standalone.HTTP01RequestHandler.HTTP01Resource(
|
||||
acme_util.HTTP01, responses[0], mock.ANY)])
|
||||
|
||||
def test_cleanup(self):
|
||||
self.auth.servers = mock.Mock()
|
||||
self.auth.servers.running.return_value = {
|
||||
1: "server1",
|
||||
2: "server2",
|
||||
|
||||
@@ -1,34 +1,11 @@
|
||||
"""Plugin utilities."""
|
||||
import logging
|
||||
import os
|
||||
import socket
|
||||
|
||||
import zope.component
|
||||
|
||||
from acme import errors as acme_errors
|
||||
from acme import util as acme_util
|
||||
|
||||
from certbot import interfaces
|
||||
from certbot import util
|
||||
|
||||
PSUTIL_REQUIREMENT = "psutil>=2.2.1"
|
||||
|
||||
try:
|
||||
acme_util.activate(PSUTIL_REQUIREMENT)
|
||||
import psutil # pragma: no cover
|
||||
USE_PSUTIL = True
|
||||
except acme_errors.DependencyError: # pragma: no cover
|
||||
USE_PSUTIL = False
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
RENEWER_EXTRA_MSG = (
|
||||
" For automated renewal, you may want to use a script that stops"
|
||||
" and starts your webserver. You can find an example at"
|
||||
" https://certbot.eff.org/docs/using.html#renewal ."
|
||||
" Alternatively you can use the webroot plugin to renew without"
|
||||
" needing to stop and start your webserver.")
|
||||
|
||||
|
||||
def path_surgery(cmd):
|
||||
"""Attempt to perform PATH surgery to find cmd
|
||||
@@ -59,105 +36,3 @@ def path_surgery(cmd):
|
||||
logger.warning("Failed to find %s in%s PATH: %s", cmd,
|
||||
expanded, path)
|
||||
return False
|
||||
|
||||
|
||||
def already_listening(port, renewer=False):
|
||||
"""Check if a process is already listening on the port.
|
||||
|
||||
If so, also tell the user via a display notification.
|
||||
|
||||
.. warning::
|
||||
On some operating systems, this function can only usefully be
|
||||
run as root.
|
||||
|
||||
:param int port: The TCP port in question.
|
||||
:returns: True or False.
|
||||
|
||||
"""
|
||||
|
||||
if USE_PSUTIL:
|
||||
return already_listening_psutil(port, renewer=renewer)
|
||||
else:
|
||||
logger.debug("Psutil not found, using simple socket check.")
|
||||
return already_listening_socket(port, renewer=renewer)
|
||||
|
||||
|
||||
def already_listening_socket(port, renewer=False):
|
||||
"""Simple socket based check to find out if port is already in use
|
||||
|
||||
:param int port: The TCP port in question.
|
||||
:returns: True or False
|
||||
"""
|
||||
|
||||
try:
|
||||
testsocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
|
||||
testsocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
try:
|
||||
testsocket.bind(("", port))
|
||||
except socket.error:
|
||||
display = zope.component.getUtility(interfaces.IDisplay)
|
||||
extra = ""
|
||||
if renewer:
|
||||
extra = RENEWER_EXTRA_MSG
|
||||
display.notification(
|
||||
"Port {0} is already in use by another process. This will "
|
||||
"prevent us from binding to that port. Please stop the "
|
||||
"process that is populating the port in question and try "
|
||||
"again. {1}".format(port, extra), force_interactive=True)
|
||||
return True
|
||||
finally:
|
||||
testsocket.close()
|
||||
except socket.error:
|
||||
pass
|
||||
return False
|
||||
|
||||
|
||||
def already_listening_psutil(port, renewer=False):
|
||||
"""Psutil variant of the open port check
|
||||
|
||||
:param int port: The TCP port in question.
|
||||
:returns: True or False.
|
||||
|
||||
"""
|
||||
try:
|
||||
net_connections = psutil.net_connections()
|
||||
except psutil.AccessDenied as error:
|
||||
logger.info("Access denied when trying to list network "
|
||||
"connections: %s. Are you root?", error)
|
||||
# this function is just a pre-check that often causes false
|
||||
# positives and problems in testing (c.f. #680 on Mac, #255
|
||||
# generally); we will fail later in bind() anyway
|
||||
return False
|
||||
|
||||
listeners = [conn.pid for conn in net_connections
|
||||
if conn.status == 'LISTEN' and
|
||||
conn.type == socket.SOCK_STREAM and
|
||||
conn.laddr[1] == port]
|
||||
try:
|
||||
if listeners and listeners[0] is not None:
|
||||
# conn.pid may be None if the current process doesn't have
|
||||
# permission to identify the listening process! Additionally,
|
||||
# listeners may have more than one element if separate
|
||||
# sockets have bound the same port on separate interfaces.
|
||||
# We currently only have UI to notify the user about one
|
||||
# of them at a time.
|
||||
pid = listeners[0]
|
||||
name = psutil.Process(pid).name()
|
||||
display = zope.component.getUtility(interfaces.IDisplay)
|
||||
extra = ""
|
||||
if renewer:
|
||||
extra = RENEWER_EXTRA_MSG
|
||||
display.notification(
|
||||
"The program {0} (process ID {1}) is already listening "
|
||||
"on TCP port {2}. This will prevent us from binding to "
|
||||
"that port. Please stop the {0} program temporarily "
|
||||
"and then try again.{3}".format(name, pid, port, extra),
|
||||
force_interactive=True)
|
||||
return True
|
||||
except (psutil.NoSuchProcess, psutil.AccessDenied):
|
||||
# Perhaps the result of a race where the process could have
|
||||
# exited or relinquished the port (NoSuchProcess), or the result
|
||||
# of an OS policy where we're not allowed to look up the process
|
||||
# name (AccessDenied).
|
||||
pass
|
||||
return False
|
||||
|
||||
@@ -1,13 +1,9 @@
|
||||
"""Tests for certbot.plugins.util."""
|
||||
import os
|
||||
import socket
|
||||
import unittest
|
||||
|
||||
import mock
|
||||
|
||||
from certbot.plugins.util import PSUTIL_REQUIREMENT
|
||||
from certbot.tests import util as test_util
|
||||
|
||||
|
||||
class PathSurgeryTest(unittest.TestCase):
|
||||
"""Tests for certbot.plugins.path_surgery."""
|
||||
@@ -34,140 +30,5 @@ class PathSurgeryTest(unittest.TestCase):
|
||||
self.assertTrue("/tmp" in os.environ["PATH"])
|
||||
|
||||
|
||||
class AlreadyListeningTest(unittest.TestCase):
|
||||
"""Tests for certbot.plugins.already_listening."""
|
||||
@classmethod
|
||||
def _call(cls, *args, **kwargs):
|
||||
from certbot.plugins.util import already_listening
|
||||
return already_listening(*args, **kwargs)
|
||||
|
||||
|
||||
class AlreadyListeningTestNoPsutil(AlreadyListeningTest):
|
||||
"""Tests for certbot.plugins.already_listening when
|
||||
psutil is not available"""
|
||||
@classmethod
|
||||
def _call(cls, *args, **kwargs):
|
||||
with mock.patch("certbot.plugins.util.USE_PSUTIL", False):
|
||||
return super(
|
||||
AlreadyListeningTestNoPsutil, cls)._call(*args, **kwargs)
|
||||
|
||||
@test_util.patch_get_utility()
|
||||
def test_ports_available(self, mock_getutil):
|
||||
# Ensure we don't get error
|
||||
with mock.patch("socket.socket.bind"):
|
||||
self.assertFalse(self._call(80))
|
||||
self.assertFalse(self._call(80, True))
|
||||
self.assertEqual(mock_getutil.call_count, 0)
|
||||
|
||||
@test_util.patch_get_utility()
|
||||
def test_ports_blocked(self, mock_getutil):
|
||||
with mock.patch("certbot.plugins.util.socket.socket.bind") as mock_bind:
|
||||
mock_bind.side_effect = socket.error
|
||||
self.assertTrue(self._call(80))
|
||||
self.assertTrue(self._call(80, True))
|
||||
with mock.patch("certbot.plugins.util.socket.socket") as mock_socket:
|
||||
mock_socket.side_effect = socket.error
|
||||
self.assertFalse(self._call(80))
|
||||
self.assertEqual(mock_getutil.call_count, 2)
|
||||
|
||||
|
||||
@test_util.skip_unless(test_util.requirement_available(PSUTIL_REQUIREMENT),
|
||||
"optional dependency psutil is not available")
|
||||
class AlreadyListeningTestPsutil(AlreadyListeningTest):
|
||||
"""Tests for certbot.plugins.already_listening."""
|
||||
@mock.patch("certbot.plugins.util.psutil.net_connections")
|
||||
@mock.patch("certbot.plugins.util.psutil.Process")
|
||||
@test_util.patch_get_utility()
|
||||
def test_race_condition(self, mock_get_utility, mock_process, mock_net):
|
||||
# This tests a race condition, or permission problem, or OS
|
||||
# incompatibility in which, for some reason, no process name can be
|
||||
# found to match the identified listening PID.
|
||||
import psutil
|
||||
from psutil._common import sconn
|
||||
conns = [
|
||||
sconn(fd=-1, family=2, type=1, laddr=("0.0.0.0", 30),
|
||||
raddr=(), status="LISTEN", pid=None),
|
||||
sconn(fd=3, family=2, type=1, laddr=("192.168.5.10", 32783),
|
||||
raddr=("20.40.60.80", 22), status="ESTABLISHED", pid=1234),
|
||||
sconn(fd=-1, family=10, type=1, laddr=("::1", 54321),
|
||||
raddr=("::1", 111), status="CLOSE_WAIT", pid=None),
|
||||
sconn(fd=3, family=2, type=1, laddr=("0.0.0.0", 17),
|
||||
raddr=(), status="LISTEN", pid=4416)]
|
||||
mock_net.return_value = conns
|
||||
mock_process.side_effect = psutil.NoSuchProcess("No such PID")
|
||||
# We simulate being unable to find the process name of PID 4416,
|
||||
# which results in returning False.
|
||||
self.assertFalse(self._call(17))
|
||||
self.assertEqual(mock_get_utility.generic_notification.call_count, 0)
|
||||
mock_process.assert_called_once_with(4416)
|
||||
|
||||
@mock.patch("certbot.plugins.util.psutil.net_connections")
|
||||
@mock.patch("certbot.plugins.util.psutil.Process")
|
||||
@test_util.patch_get_utility()
|
||||
def test_not_listening(self, mock_get_utility, mock_process, mock_net):
|
||||
from psutil._common import sconn
|
||||
conns = [
|
||||
sconn(fd=-1, family=2, type=1, laddr=("0.0.0.0", 30),
|
||||
raddr=(), status="LISTEN", pid=None),
|
||||
sconn(fd=3, family=2, type=1, laddr=("192.168.5.10", 32783),
|
||||
raddr=("20.40.60.80", 22), status="ESTABLISHED", pid=1234),
|
||||
sconn(fd=-1, family=10, type=1, laddr=("::1", 54321),
|
||||
raddr=("::1", 111), status="CLOSE_WAIT", pid=None)]
|
||||
mock_net.return_value = conns
|
||||
mock_process.name.return_value = "inetd"
|
||||
self.assertFalse(self._call(17))
|
||||
self.assertEqual(mock_get_utility.generic_notification.call_count, 0)
|
||||
self.assertEqual(mock_process.call_count, 0)
|
||||
|
||||
@mock.patch("certbot.plugins.util.psutil.net_connections")
|
||||
@mock.patch("certbot.plugins.util.psutil.Process")
|
||||
@test_util.patch_get_utility()
|
||||
def test_listening_ipv4(self, mock_get_utility, mock_process, mock_net):
|
||||
from psutil._common import sconn
|
||||
conns = [
|
||||
sconn(fd=-1, family=2, type=1, laddr=("0.0.0.0", 30),
|
||||
raddr=(), status="LISTEN", pid=None),
|
||||
sconn(fd=3, family=2, type=1, laddr=("192.168.5.10", 32783),
|
||||
raddr=("20.40.60.80", 22), status="ESTABLISHED", pid=1234),
|
||||
sconn(fd=-1, family=10, type=1, laddr=("::1", 54321),
|
||||
raddr=("::1", 111), status="CLOSE_WAIT", pid=None),
|
||||
sconn(fd=3, family=2, type=1, laddr=("0.0.0.0", 17),
|
||||
raddr=(), status="LISTEN", pid=4416)]
|
||||
mock_net.return_value = conns
|
||||
mock_process.name.return_value = "inetd"
|
||||
result = self._call(17, True)
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(mock_get_utility.call_count, 1)
|
||||
mock_process.assert_called_once_with(4416)
|
||||
|
||||
@mock.patch("certbot.plugins.util.psutil.net_connections")
|
||||
@mock.patch("certbot.plugins.util.psutil.Process")
|
||||
@test_util.patch_get_utility()
|
||||
def test_listening_ipv6(self, mock_get_utility, mock_process, mock_net):
|
||||
from psutil._common import sconn
|
||||
conns = [
|
||||
sconn(fd=-1, family=2, type=1, laddr=("0.0.0.0", 30),
|
||||
raddr=(), status="LISTEN", pid=None),
|
||||
sconn(fd=3, family=2, type=1, laddr=("192.168.5.10", 32783),
|
||||
raddr=("20.40.60.80", 22), status="ESTABLISHED", pid=1234),
|
||||
sconn(fd=-1, family=10, type=1, laddr=("::1", 54321),
|
||||
raddr=("::1", 111), status="CLOSE_WAIT", pid=None),
|
||||
sconn(fd=3, family=10, type=1, laddr=("::", 12345), raddr=(),
|
||||
status="LISTEN", pid=4420),
|
||||
sconn(fd=3, family=2, type=1, laddr=("0.0.0.0", 17),
|
||||
raddr=(), status="LISTEN", pid=4416)]
|
||||
mock_net.return_value = conns
|
||||
mock_process.name.return_value = "inetd"
|
||||
result = self._call(12345)
|
||||
self.assertTrue(result)
|
||||
self.assertEqual(mock_get_utility.call_count, 1)
|
||||
mock_process.assert_called_once_with(4420)
|
||||
|
||||
@mock.patch("certbot.plugins.util.psutil.net_connections")
|
||||
def test_access_denied_exception(self, mock_net):
|
||||
import psutil
|
||||
mock_net.side_effect = psutil.AccessDenied("")
|
||||
self.assertFalse(self._call(12345))
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main() # pragma: no cover
|
||||
|
||||
Reference in New Issue
Block a user