glances.plugins.ports.ThreadScanner._web_scan() - Code Metrics - Inspection of "Refactor plugin folder in the same way than export..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — develop ( 30cc9f...48251c )

by Nicolas

created 2023-10-15 16:05 UTC

glances.plugins.ports.ThreadScanner._web_scan() A

↳ Parent: glances.plugins.ports

Complexity

Conditions

Size

Total Lines	18
Code Lines	15

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	15
nop	2
dl	0
loc	18
rs	9.65
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2022 Nicolas Hennion <[email protected]>
#
# SPDX-License-Identifier: LGPL-3.0-only
#

"""Ports scanner plugin."""

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

from glances.globals import WINDOWS, MACOS, BSD, bool_type
from glances.ports_list import GlancesPortsList
from glances.web_list import GlancesWebList
from glances.timer import Counter
from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel

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 PluginModel(GlancesPluginModel):
    """Glances ports scanner plugin."""

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super(PluginModel, 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()
        )

        # 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(PluginModel, self).exit()

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._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.is_alive()
            if not thread_is_running:
                # Run ports scanner
                self._thread = ThreadScanner(self.stats)
                self._thread.start()
        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.is_set()

    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 sticky 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:
            # Correct issue #1084: No Offline status for timed-out ports
            port['status'] = False
        except Exception as e:
            logger.debug("{}: Error while pinging host {} ({})".format(self.plugin_name, port['host'], e))

        fnull.close()

        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, e))

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

nicolargo / glances

Push — develop ( 30cc9f...48251c )

glances.plugins.ports.ThreadScanner._web_scan() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like