starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-28 00:35:50 +02:00
Code Issues Packages Projects Releases Wiki Activity

Use psutil instead of netstat subprocess

Browse Source
This commit is contained in:
Seth Schoen
2015-02-19 17:49:27 -08:00
parent 5ab9b7c331
commit fc1617531e
3 changed files with 103 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
letsencrypt/client/standalone_authenticator.py
+16 -25
View File
@@ -1,8 +1,8 @@
"""Standalone authenticator."""
import os
import psutil
import signal
import socket
import subprocess
import sys
import time
@@ -266,30 +266,23 @@ class StandaloneAuthenticator(object):
If so, also tell the user via a display notification.
.. warning::
The current implementation is Linux-specific. (On other
operating systems, it will simply not detect bound ports.)
This function can only usefully be run as root.
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."""
listeners = [conn.pid for conn in psutil.net_connections() \
if conn.status == 'LISTEN' and \
conn.type == socket.SOCK_STREAM and \
(conn.laddr == ('0.0.0.0', port) or \
conn.laddr == ('::', port))]
try:
proc = subprocess.Popen(
[constants.NETSTAT, "-nta", "--program"],
stdout=subprocess.PIPE, stderr=subprocess.PIPE)
stdout, _ = proc.communicate()
if proc.wait() != 0:
raise OSError("netstat subprocess failed")
lines = [x.split() for x in stdout.split("\n")[2:] if x]
listeners = [L[6] for L in lines if
# IPv4 socket case
(L[0] == 'tcp' and L[5] == 'LISTEN' \
and L[3] == '0.0.0.0:{0}'.format(port)) or \
# IPv6 socket case
(L[0] == 'tcp6' and L[5] == 'LISTEN' \
and L[3] == ':::{0}'.format(port))]
if listeners:
pid, name = listeners[0].split("/")
if listeners and listeners[0]:
# conn.pid may be None if the current process doesn't have
# permission to identify the listening process!
pid = listeners[0]
name = psutil.Process(pid).name()
display = zope.component.getUtility(interfaces.IDisplay)
display.generic_notification(
"The program {0} (process ID {1}) is already listening "
@@ -297,11 +290,9 @@ class StandaloneAuthenticator(object):
"that port. Please stop the {0} program temporarily "
"and then try again.".format(name, pid, port))
return True
except (OSError, ValueError, IndexError):
# A sign that this command isn't available or usable this
# way on this operating system, or there was something
# unexpected about the format of the netstat output; we will
# not be able to recover from this condition.
except psutil.NoSuchProcess:
# Perhaps the result of a race where the process could have
# exited or relinquished the port.
pass
return False
letsencrypt/client/tests/standalone_authenticator_test.py
+86 -55
View File
@@ -187,71 +187,102 @@ class AlreadyListeningTest(unittest.TestCase):
StandaloneAuthenticator
self.authenticator = StandaloneAuthenticator()
@mock.patch("letsencrypt.client.standalone_authenticator.subprocess.Popen")
def test_subprocess_fails(self, mock_popen):
subprocess_object = mock.MagicMock()
subprocess_object.communicate.return_value = ("foo", "bar")
subprocess_object.wait.return_value = 1
mock_popen.return_value = subprocess_object
result = self.authenticator.already_listening(17)
self.assertFalse(result)
subprocess_object.wait.assert_called_once_with()
@mock.patch("letsencrypt.client.standalone_authenticator.subprocess.Popen")
def test_no_relevant_line(self, mock_popen):
# pylint: disable=line-too-long,trailing-whitespace
subprocess_object = mock.MagicMock()
subprocess_object.communicate.return_value = (
"""Active Internet connections (servers and established)
Proto Recv-Q Send-Q Local Address Foreign Address State PID/Program name
tcp 0 0 127.0.1.1:53 0.0.0.0:* LISTEN 1234/foo
tcp 0 0 127.0.0.1:631 0.0.0.0:* LISTEN 2345/bar
tcp 0 0 0.0.0.0:180 0.0.0.0:* LISTEN 11111/hello """,
"I am the standard error")
subprocess_object.wait.return_value = 0
mock_popen.return_value = subprocess_object
result = self.authenticator.already_listening(17)
self.assertFalse(result)
@mock.patch("letsencrypt.client.standalone_authenticator.subprocess.Popen")
@mock.patch("letsencrypt.client.standalone_authenticator.psutil."
"net_connections")
@mock.patch("letsencrypt.client.standalone_authenticator.psutil.Process")
@mock.patch("letsencrypt.client.standalone_authenticator."
"zope.component.getUtility")
def test_has_relevant_line(self, mock_get_utility, mock_popen):
# pylint: disable=line-too-long,trailing-whitespace
subprocess_object = mock.MagicMock()
subprocess_object.communicate.return_value = (
"""Active Internet connections (servers and established)
Proto Recv-Q Send-Q Local Address Foreign Address State PID/Program name
tcp 0 0 127.0.1.1:53 0.0.0.0:* LISTEN 1234/foo
tcp 0 0 127.0.0.1:631 0.0.0.0:* LISTEN 2345/bar
tcp 0 0 0.0.0.0:17 0.0.0.0:* LISTEN 11111/hello
tcp 0 0 0.0.0.0:1728 0.0.0.0:* LISTEN 2345/bar """,
"I am the standard error")
subprocess_object.wait.return_value = 0
mock_popen.return_value = subprocess_object
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.
from psutil._common import sconn
from psutil import NoSuchProcess
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 = NoSuchProcess("No such PID")
# We simulate being unable to find the process name of PID 4416,
# which results in returning False.
self.assertFalse(self.authenticator.already_listening(17))
self.assertEqual(mock_get_utility.generic_notification.call_count, 0)
mock_process.assert_called_once_with(4416)
@mock.patch("letsencrypt.client.standalone_authenticator.psutil."
"net_connections")
@mock.patch("letsencrypt.client.standalone_authenticator.psutil.Process")
@mock.patch("letsencrypt.client.standalone_authenticator."
"zope.component.getUtility")
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.authenticator.already_listening(17))
self.assertEqual(mock_get_utility.generic_notification.call_count, 0)
self.assertEqual(mock_process.call_count, 0)
@mock.patch("letsencrypt.client.standalone_authenticator.psutil."
"net_connections")
@mock.patch("letsencrypt.client.standalone_authenticator.psutil.Process")
@mock.patch("letsencrypt.client.standalone_authenticator."
"zope.component.getUtility")
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.authenticator.already_listening(17)
self.assertTrue(result)
self.assertEqual(mock_get_utility.call_count, 1)
mock_process.assert_called_once_with(4416)
@mock.patch("letsencrypt.client.standalone_authenticator.subprocess.Popen")
@mock.patch("letsencrypt.client.standalone_authenticator.psutil."
"net_connections")
@mock.patch("letsencrypt.client.standalone_authenticator.psutil.Process")
@mock.patch("letsencrypt.client.standalone_authenticator."
"zope.component.getUtility")
def test_has_relevant_ipv6_line(self, mock_get_utility, mock_popen):
# pylint: disable=line-too-long,trailing-whitespace
subprocess_object = mock.MagicMock()
subprocess_object.communicate.return_value = (
"""Active Internet connections (servers and established)
Proto Recv-Q Send-Q Local Address Foreign Address State PID/Program name
tcp 0 0 127.0.1.1:53 0.0.0.0:* LISTEN 1234/foo
tcp 0 0 127.0.0.1:631 0.0.0.0:* LISTEN 2345/bar
tcp6 0 0 :::17 :::* LISTEN 11111/hello
tcp 0 0 0.0.0.0:1728 0.0.0.0:* LISTEN 2345/bar """,
"I am the standard error")
subprocess_object.wait.return_value = 0
mock_popen.return_value = subprocess_object
result = self.authenticator.already_listening(17)
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.authenticator.already_listening(12345)
self.assertTrue(result)
self.assertEqual(mock_get_utility.call_count, 1)
mock_process.assert_called_once_with(4420)
class PerformTest(unittest.TestCase):
"""Tests for perform() method."""
setup.py
+1
View File
@@ -26,6 +26,7 @@ install_requires = [
'ConfArgParse',
'jsonschema',
'mock',
'psutil',
'pycrypto',
'PyOpenSSL',
'python-augeas',