Issues in glances_ports.py - All Issues - Inspection of "Update doc for new connections plugin" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 66c9ff...e21229 )

by Nicolas

created 2019-10-07 11:02 UTC

glances/plugins/glances_ports.py (1 issue)

Severity

Nicolas Hennion 1

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2019 Nicolargo <[email protected]>
#
# Glances is free software; you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# Glances is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.

"""Ports scanner plugin."""

import os
import subprocess
import threading
import socket
import time
import numbers

from glances.globals import WINDOWS, MACOS, BSD
from glances.ports_list import GlancesPortsList
from glances.web_list import GlancesWebList
from glances.timer import Timer, Counter
from glances.compat import bool_type
from glances.logger import logger
from glances.plugins.glances_plugin import GlancesPlugin

try:
    import requests
    requests_tag = True
except ImportError as e:
    requests_tag = False
    logger.warning("Missing Python Lib ({}), Ports plugin is limited to port scanning".format(e))


class Plugin(GlancesPlugin):
    """Glances ports scanner plugin."""

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super(Plugin, self).__init__(args=args,
                                     config=config,
                                     stats_init_value=[])
        self.args = args
        self.config = config

        # We want to display the stat in the curse interface
        self.display_curse = True

        # Init stats
        self.stats = GlancesPortsList(config=config, args=args).get_ports_list() + GlancesWebList(config=config, args=args).get_web_list()

        # Init global Timer
        self.timer_ports = Timer(0)

        # Global Thread running all the scans
        self._thread = None

    def exit(self):
        """Overwrite the exit method to close threads."""
        if self._thread is not None:
            self._thread.stop()
        # Call the father class
        super(Plugin, self).exit()

    @GlancesPlugin._log_result_decorator
    def update(self):
        """Update the ports list."""
        if self.input_method == 'local':
            # Only refresh:
            # * if there is not other scanning thread
            # * every refresh seconds (define in the configuration file)
            if self._thread is None:
                thread_is_running = False
            else:
                thread_is_running = self._thread.isAlive()
            if self.timer_ports.finished() and not thread_is_running:
                # Run ports scanner
                self._thread = ThreadScanner(self.stats)
                self._thread.start()
                # Restart timer
                if len(self.stats) > 0:
                    self.timer_ports = Timer(self.stats[0]['refresh'])
                else:
                    self.timer_ports = Timer(0)
        else:
            # Not available in SNMP mode
            pass

        return self.stats

    def get_key(self):
        """Return the key of the list."""
        return 'indice'

    def get_ports_alert(self, port, header="", log=False):
        """Return the alert status relative to the port scan return value."""
        ret = 'OK'
        if port['status'] is None:
            ret = 'CAREFUL'
        elif port['status'] == 0:
            ret = 'CRITICAL'
        elif (isinstance(port['status'], (float, int)) and
              port['rtt_warning'] is not None and
              port['status'] > port['rtt_warning']):
            ret = 'WARNING'

        # Get stat name
        stat_name = self.get_stat_name(header=header)

        # Manage threshold
        self.manage_threshold(stat_name, ret)

        # Manage action
        self.manage_action(stat_name,
                           ret.lower(),
                           header,
                           port[self.get_key()])

        return ret

    def get_web_alert(self, web, header="", log=False):
        """Return the alert status relative to the web/url scan return value."""
        ret = 'OK'
        if web['status'] is None:
            ret = 'CAREFUL'
        elif web['status'] not in [200, 301, 302]:
            ret = 'CRITICAL'
        elif web['rtt_warning'] is not None and web['elapsed'] > web['rtt_warning']:
            ret = 'WARNING'

        # Get stat name
        stat_name = self.get_stat_name(header=header)

        # Manage threshold
        self.manage_threshold(stat_name, ret)

        # Manage action
        self.manage_action(stat_name,
                           ret.lower(),
                           header,
                           web[self.get_key()])

        return ret

    def msg_curse(self, args=None, max_width=None):
        """Return the dict to display in the curse interface."""
        # Init the return message
        # Only process if stats exist and display plugin enable...
        ret = []

        if not self.stats or args.disable_ports:
            return ret

        # Max size for the interface name
        name_max_width = max_width - 7

        # Build the string message
        for p in self.stats:
            if 'host' in p:
                if p['host'] is None:
                    status = 'None'
                elif p['status'] is None:
                    status = 'Scanning'
                elif isinstance(p['status'], bool_type) and p['status'] is True:
                    status = 'Open'
                elif p['status'] == 0:
                    status = 'Timeout'
                else:
                    # Convert second to ms
                    status = '{0:.0f}ms'.format(p['status'] * 1000.0)

                msg = '{:{width}}'.format(p['description'][0:name_max_width],
                                          width=name_max_width)
                ret.append(self.curse_add_line(msg))
                msg = '{:>9}'.format(status)
                ret.append(self.curse_add_line(msg,
                                               self.get_ports_alert(p,
                                                                    header=p['indice'] + '_rtt')))
                ret.append(self.curse_new_line())
            elif 'url' in p:
                msg = '{:{width}}'.format(p['description'][0:name_max_width],
                                          width=name_max_width)
                ret.append(self.curse_add_line(msg))
                if isinstance(p['status'], numbers.Number):
                    status = 'Code {}'.format(p['status'])
                elif p['status'] is None:
                    status = 'Scanning'
                else:
                    status = p['status']
                msg = '{:>9}'.format(status)
                ret.append(self.curse_add_line(msg,
                                               self.get_web_alert(p,
                                                                  header=p['indice'] + '_rtt')))
                ret.append(self.curse_new_line())

        # Delete the last empty line
        try:
            ret.pop()
        except IndexError:
            pass

        return ret


class ThreadScanner(threading.Thread):
    """
    Specific thread for the port/web scanner.

    stats is a list of dict
    """

    def __init__(self, stats):
        """Init the class."""
        logger.debug("ports plugin - Create thread for scan list {}".format(stats))
        super(ThreadScanner, self).__init__()
        # Event needed to stop properly the thread
        self._stopper = threading.Event()
        # The class return the stats as a list of dict
        self._stats = stats
        # Is part of Ports plugin
        self.plugin_name = "ports"

    def run(self):
        """Grab the stats.

        Infinite loop, should be stopped by calling the stop() method.
        """
        for p in self._stats:
            # End of the thread has been asked
            if self.stopped():
                break
            # Scan a port (ICMP or TCP)
            if 'port' in p:
                self._port_scan(p)
                # Had to wait between two scans
                # If not, result are not ok
                time.sleep(1)
            # Scan an URL
            elif 'url' in p and requests_tag:
                self._web_scan(p)

    @property
    def stats(self):
        """Stats getter."""
        return self._stats

    @stats.setter
    def stats(self, value):
        """Stats setter."""
        self._stats = value

    def stop(self, timeout=None):
        """Stop the thread."""
        logger.debug("ports plugin - Close thread for scan list {}".format(self._stats))
        self._stopper.set()

    def stopped(self):
        """Return True is the thread is stopped."""
        return self._stopper.isSet()

    def _web_scan(self, web):
        """Scan the  Web/URL (dict) and update the status key."""
        try:
            req = requests.head(web['url'],
                                allow_redirects=True,
                                verify=web['ssl_verify'],
                                proxies=web['proxies'],
                                timeout=web['timeout'])
        except Exception as e:
            logger.debug(e)
            web['status'] = 'Error'
            web['elapsed'] = 0
        else:
            web['status'] = req.status_code
            web['elapsed'] = req.elapsed.total_seconds()
        return web

    def _port_scan(self, port):
        """Scan the port structure (dict) and update the status key."""
        if int(port['port']) == 0:
            return self._port_scan_icmp(port)
        else:
            return self._port_scan_tcp(port)

    def _resolv_name(self, hostname):
        """Convert hostname to IP address."""
        ip = hostname
        try:
            ip = socket.gethostbyname(hostname)
        except Exception as e:
            logger.debug("{}: Cannot convert {} to IP address ({})".format(self.plugin_name, hostname, e))
        return ip

    def _port_scan_icmp(self, port):
        """Scan the (ICMP) port structure (dict) and update the status key."""
        ret = None

        # Create the ping command
        # Use the system ping command because it already have the steacky bit set
        # Python can not create ICMP packet with non root right
        if WINDOWS:
            timeout_opt = '-w'
            count_opt = '-n'
        elif MACOS or BSD:
            timeout_opt = '-t'
            count_opt = '-c'
        else:
            # Linux and co...
            timeout_opt = '-W'
            count_opt = '-c'
        # Build the command line
        # Note: Only string are allowed
        cmd = ['ping',
               count_opt, '1',
               timeout_opt, str(self._resolv_name(port['timeout'])),
               self._resolv_name(port['host'])]
        fnull = open(os.devnull, 'w')

        try:
            counter = Counter()
            ret = subprocess.check_call(cmd, stdout=fnull, stderr=fnull, close_fds=True)
            if ret == 0:
                port['status'] = counter.get()
            else:
                port['status'] = False
        except subprocess.CalledProcessError as e:
            # Correct issue #1084: No Offline status for timeouted ports
            port['status'] = False
        except Exception as e:
            logger.debug("{}: Error while pinging host {} ({})".format(self.plugin_name, port['host'], e))

        return ret

    def _port_scan_tcp(self, port):
        """Scan the (TCP) port structure (dict) and update the status key."""
        ret = None

        # Create and configure the scanning socket
        try:
            socket.setdefaulttimeout(port['timeout'])
            _socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        except Exception as e:
            logger.debug("{}: Error while creating scanning socket".format(self.plugin_name))

        # Scan port
        ip = self._resolv_name(port['host'])
        counter = Counter()
        try:
            ret = _socket.connect_ex((ip, int(port['port'])))

        except Exception as e:
            logger.debug("{}: Error while scanning port {} ({})".format(self.plugin_name, port, e))
        else:
            if ret == 0:
                port['status'] = counter.get()
            else:
                port['status'] = False
        finally:
            _socket.close()

        return ret


1		# -- coding: utf-8 --
2		#
3		# This file is part of Glances.
4		#
5		# Copyright (C) 2019 Nicolargo <[email protected]>
6		#
7		# Glances is free software; you can redistribute it and/or modify
8		# it under the terms of the GNU Lesser General Public License as published by
9		# the Free Software Foundation, either version 3 of the License, or
10		# (at your option) any later version.
11		#
12		# Glances is distributed in the hope that it will be useful,
13		# but WITHOUT ANY WARRANTY; without even the implied warranty of
14		# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15		# GNU Lesser General Public License for more details.
16		#
17		# You should have received a copy of the GNU Lesser General Public License
18		# along with this program. If not, see <http://www.gnu.org/licenses/>.
19
20		"""Ports scanner plugin."""
21
22		import os
23		import subprocess
24		import threading
25		import socket
26		import time
27		import numbers
28
29		from glances.globals import WINDOWS, MACOS, BSD
30		from glances.ports_list import GlancesPortsList
31		from glances.web_list import GlancesWebList
32		from glances.timer import Timer, Counter
33		from glances.compat import bool_type
34		from glances.logger import logger
35		from glances.plugins.glances_plugin import GlancesPlugin
36
37		try:
38		import requests
39		requests_tag = True
40		except ImportError as e:
41		requests_tag = False
42		logger.warning("Missing Python Lib ({}), Ports plugin is limited to port scanning".format(e))
43
44
45		class Plugin(GlancesPlugin):
46		"""Glances ports scanner plugin."""
47
48		def __init__(self, args=None, config=None):
49		"""Init the plugin."""
50		super(Plugin, self).__init__(args=args,
51		config=config,
52		stats_init_value=[])
53		self.args = args
54		self.config = config
55
56		# We want to display the stat in the curse interface
57		self.display_curse = True
58
59		# Init stats
60		self.stats = GlancesPortsList(config=config, args=args).get_ports_list() + GlancesWebList(config=config, args=args).get_web_list()
61
62		# Init global Timer
63		self.timer_ports = Timer(0)
64
65		# Global Thread running all the scans
66		self._thread = None
67
68		def exit(self):
69		"""Overwrite the exit method to close threads."""
70		if self._thread is not None:
71		self._thread.stop()
72		# Call the father class
73		super(Plugin, self).exit()
74
75		@GlancesPlugin._log_result_decorator
76		def update(self):
77		"""Update the ports list."""
78		if self.input_method == 'local':
79		# Only refresh:
80		# * if there is not other scanning thread
81		# * every refresh seconds (define in the configuration file)
82		if self._thread is None:
83		thread_is_running = False
84		else:
85		thread_is_running = self._thread.isAlive()
86		if self.timer_ports.finished() and not thread_is_running:
87		# Run ports scanner
88		self._thread = ThreadScanner(self.stats)
89		self._thread.start()
90		# Restart timer
91		if len(self.stats) > 0:
92		self.timer_ports = Timer(self.stats[0]['refresh'])
93		else:
94		self.timer_ports = Timer(0)
95		else:
96		# Not available in SNMP mode
97		pass
98
99		return self.stats
100
101		def get_key(self):
102		"""Return the key of the list."""
103		return 'indice'
104
105	View Code Duplication	def get_ports_alert(self, port, header="", log=False):
106		"""Return the alert status relative to the port scan return value."""
107		ret = 'OK'
108		if port['status'] is None:
109		ret = 'CAREFUL'
110		elif port['status'] == 0:
111		ret = 'CRITICAL'
112		elif (isinstance(port['status'], (float, int)) and
113		port['rtt_warning'] is not None and
114		port['status'] > port['rtt_warning']):
115		ret = 'WARNING'
116
117		# Get stat name
118		stat_name = self.get_stat_name(header=header)
119
120		# Manage threshold
121		self.manage_threshold(stat_name, ret)
122
123		# Manage action
124		self.manage_action(stat_name,
125		ret.lower(),
126		header,
127		port[self.get_key()])
128
129		return ret
130
131	View Code Duplication	def get_web_alert(self, web, header="", log=False):
132		"""Return the alert status relative to the web/url scan return value."""
133		ret = 'OK'
134		if web['status'] is None:
135		ret = 'CAREFUL'
136		elif web['status'] not in [200, 301, 302]:
137		ret = 'CRITICAL'
138		elif web['rtt_warning'] is not None and web['elapsed'] > web['rtt_warning']:
139		ret = 'WARNING'
140
141		# Get stat name
142		stat_name = self.get_stat_name(header=header)
143
144		# Manage threshold
145		self.manage_threshold(stat_name, ret)
146
147		# Manage action
148		self.manage_action(stat_name,
149		ret.lower(),
150		header,
151		web[self.get_key()])
152
153		return ret
154
155		def msg_curse(self, args=None, max_width=None):
156		"""Return the dict to display in the curse interface."""
157		# Init the return message
158		# Only process if stats exist and display plugin enable...
159		ret = []
160
161		if not self.stats or args.disable_ports:
162		return ret
163
164		# Max size for the interface name
165		name_max_width = max_width - 7
166
167		# Build the string message
168		for p in self.stats:
169		if 'host' in p:
170		if p['host'] is None:
171		status = 'None'
172		elif p['status'] is None:
173		status = 'Scanning'
174		elif isinstance(p['status'], bool_type) and p['status'] is True:
175		status = 'Open'
176		elif p['status'] == 0:
177		status = 'Timeout'
178		else:
179		# Convert second to ms
180		status = '{0:.0f}ms'.format(p['status'] * 1000.0)
181
182		msg = '{:{width}}'.format(p['description'][0:name_max_width],
183		width=name_max_width)
184		ret.append(self.curse_add_line(msg))
185		msg = '{:>9}'.format(status)
186		ret.append(self.curse_add_line(msg,
187		self.get_ports_alert(p,
188		header=p['indice'] + '_rtt')))
189		ret.append(self.curse_new_line())
190		elif 'url' in p:
191		msg = '{:{width}}'.format(p['description'][0:name_max_width],
192		width=name_max_width)
193		ret.append(self.curse_add_line(msg))
194		if isinstance(p['status'], numbers.Number):
195		status = 'Code {}'.format(p['status'])
196		elif p['status'] is None:
197		status = 'Scanning'
198		else:
199		status = p['status']
200		msg = '{:>9}'.format(status)
201		ret.append(self.curse_add_line(msg,
202		self.get_web_alert(p,
203		header=p['indice'] + '_rtt')))
204		ret.append(self.curse_new_line())
205
206		# Delete the last empty line
207		try:
208		ret.pop()
209		except IndexError:
210		pass
211
212		return ret
213
214
215		class ThreadScanner(threading.Thread):
216		"""
217		Specific thread for the port/web scanner.
218
219		stats is a list of dict
220		"""
221
222		def __init__(self, stats):
223		"""Init the class."""
224		logger.debug("ports plugin - Create thread for scan list {}".format(stats))
225		super(ThreadScanner, self).__init__()
226		# Event needed to stop properly the thread
227		self._stopper = threading.Event()
228		# The class return the stats as a list of dict
229		self._stats = stats
230		# Is part of Ports plugin
231		self.plugin_name = "ports"
232
233		def run(self):
234		"""Grab the stats.
235
236		Infinite loop, should be stopped by calling the stop() method.
237		"""
238		for p in self._stats:
239		# End of the thread has been asked
240		if self.stopped():
241		break
242		# Scan a port (ICMP or TCP)
243		if 'port' in p:
244		self._port_scan(p)
245		# Had to wait between two scans
246		# If not, result are not ok
247		time.sleep(1)
248		# Scan an URL
249		elif 'url' in p and requests_tag:
250		self._web_scan(p)
251
252		@property
253		def stats(self):
254		"""Stats getter."""
255		return self._stats
256
257		@stats.setter
258		def stats(self, value):
259		"""Stats setter."""
260		self._stats = value
261
262		def stop(self, timeout=None):
263		"""Stop the thread."""
264		logger.debug("ports plugin - Close thread for scan list {}".format(self._stats))
265		self._stopper.set()
266
267		def stopped(self):
268		"""Return True is the thread is stopped."""
269		return self._stopper.isSet()
270
271		def _web_scan(self, web):
272		"""Scan the Web/URL (dict) and update the status key."""
273		try:
274		req = requests.head(web['url'],
275		allow_redirects=True,
276		verify=web['ssl_verify'],
277		proxies=web['proxies'],
278		timeout=web['timeout'])
279		except Exception as e:
280		logger.debug(e)
281		web['status'] = 'Error'
282		web['elapsed'] = 0
283		else:
284		web['status'] = req.status_code
285		web['elapsed'] = req.elapsed.total_seconds()
286		return web
287
288		def _port_scan(self, port):
289		"""Scan the port structure (dict) and update the status key."""
290		if int(port['port']) == 0:
291		return self._port_scan_icmp(port)
292		else:
293		return self._port_scan_tcp(port)
294
295		def _resolv_name(self, hostname):
296		"""Convert hostname to IP address."""
297		ip = hostname
298		try:
299		ip = socket.gethostbyname(hostname)
300		except Exception as e:
301		logger.debug("{}: Cannot convert {} to IP address ({})".format(self.plugin_name, hostname, e))
302		return ip
303
304		def _port_scan_icmp(self, port):
305		"""Scan the (ICMP) port structure (dict) and update the status key."""
306		ret = None
307
308		# Create the ping command
309		# Use the system ping command because it already have the steacky bit set
310		# Python can not create ICMP packet with non root right
311		if WINDOWS:
312		timeout_opt = '-w'
313		count_opt = '-n'
314		elif MACOS or BSD:
315		timeout_opt = '-t'
316		count_opt = '-c'
317		else:
318		# Linux and co...
319		timeout_opt = '-W'
320		count_opt = '-c'
321		# Build the command line
322		# Note: Only string are allowed
323		cmd = ['ping',
324		count_opt, '1',
325		timeout_opt, str(self._resolv_name(port['timeout'])),
326		self._resolv_name(port['host'])]
327		fnull = open(os.devnull, 'w')
328
329		try:
330		counter = Counter()
331		ret = subprocess.check_call(cmd, stdout=fnull, stderr=fnull, close_fds=True)
332		if ret == 0:
333		port['status'] = counter.get()
334		else:
335		port['status'] = False
336		except subprocess.CalledProcessError as e:
337		# Correct issue #1084: No Offline status for timeouted ports
338		port['status'] = False
339		except Exception as e:
340		logger.debug("{}: Error while pinging host {} ({})".format(self.plugin_name, port['host'], e))
341
342		return ret
343
344		def _port_scan_tcp(self, port):
345		"""Scan the (TCP) port structure (dict) and update the status key."""
346		ret = None
347
348		# Create and configure the scanning socket
349		try:
350		socket.setdefaulttimeout(port['timeout'])
351		_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
352		except Exception as e:
353		logger.debug("{}: Error while creating scanning socket".format(self.plugin_name))
354
355		# Scan port
356		ip = self._resolv_name(port['host'])
357		counter = Counter()
358		try:
359		ret = _socket.connect_ex((ip, int(port['port'])))
		0 ignored issues – show introduced 2018-11-11 15:28 UTC by Report Bug Copy Issue Report Show Similar Issues like this The variable `_socket` does not seem to be defined for all execution paths. Loading history...
360		except Exception as e:
361		logger.debug("{}: Error while scanning port {} ({})".format(self.plugin_name, port, e))
362		else:
363		if ret == 0:
364		port['status'] = counter.get()
365		else:
366		port['status'] = False
367		finally:
368		_socket.close()
369
370		return ret
371

nicolargo / glances

Push — develop ( 66c9ff...e21229 )

glances/plugins/glances_ports.py (1 issue)

Severity

Introduced By

Duplication Side-by-Side

Filter issues like