glances.plugins.ports.ThreadScanner._web_scan() - Code Metrics - Inspection of "Glances version 4.0.0" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( ee826a...d9056e )

by Nicolas

created 2024-05-12 08:45 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))

# Fields description
# description: human readable description
# short_name: shortname to use un UI
# unit: unit type
# rate: is it a rate ? If yes, // by time_since_update when displayed,
# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
fields_description = {
    'host': {
        'description': 'Measurement is be done on this host (or IP address)',
    },
    'port': {
        'description': 'Measurement is be done on this port (0 for ICMP)',
    },
    'description': {
        'description': 'Human readable description for the host/port',
    },
    'refresh': {
        'description': 'Refresh time (in seconds) for this host/port',
    },
    'timeout': {
        'description': 'Timeout (in seconds) for the measurement',
    },
    'status': {
        'description': 'Measurement result (in seconds)',
        'unit': 'second',
    },
    'rtt_warning': {
        'description': 'Warning threshold (in seconds) for the measurement',
        'unit': 'second',
    },
    'indice': {
        'description': 'Unique indice for the host/port',
    },
}


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=[], fields_description=fields_description
        )
        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()

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

    @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_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
        if max_width:
            name_max_width = max_width - 7
        else:
            # No max_width defined, return an emptu curse message
            logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
            return ret

        # 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 get_key(self):
        """Return the key of the list."""
        return 'indice'

    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)
            # Add the key for every element
            p['key'] = self.get_key()

    @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		# Fields description
35		# description: human readable description
36		# short_name: shortname to use un UI
37		# unit: unit type
38		# rate: is it a rate ? If yes, // by time_since_update when displayed,
39		# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
40		fields_description = {
41		'host': {
42		'description': 'Measurement is be done on this host (or IP address)',
43		},
44		'port': {
45		'description': 'Measurement is be done on this port (0 for ICMP)',
46		},
47		'description': {
48		'description': 'Human readable description for the host/port',
49		},
50		'refresh': {
51		'description': 'Refresh time (in seconds) for this host/port',
52		},
53		'timeout': {
54		'description': 'Timeout (in seconds) for the measurement',
55		},
56		'status': {
57		'description': 'Measurement result (in seconds)',
58		'unit': 'second',
59		},
60		'rtt_warning': {
61		'description': 'Warning threshold (in seconds) for the measurement',
62		'unit': 'second',
63		},
64		'indice': {
65		'description': 'Unique indice for the host/port',
66		},
67		}
68
69
70		class PluginModel(GlancesPluginModel):
71		"""Glances ports scanner plugin."""
72
73		def __init__(self, args=None, config=None):
74		"""Init the plugin."""
75		super(PluginModel, self).__init__(
76		args=args, config=config, stats_init_value=[], fields_description=fields_description
77		)
78		self.args = args
79		self.config = config
80
81		# We want to display the stat in the curse interface
82		self.display_curse = True
83
84		# Init stats
85		self.stats = (
86		GlancesPortsList(config=config, args=args).get_ports_list()
87		+ GlancesWebList(config=config, args=args).get_web_list()
88		)
89
90		# Global Thread running all the scans
91		self._thread = None
92
93		def exit(self):
94		"""Overwrite the exit method to close threads."""
95		if self._thread is not None:
96		self._thread.stop()
97		# Call the father class
98		super(PluginModel, self).exit()
99
100		def get_key(self):
101		"""Return the key of the list."""
102		return 'indice'
103
104		@GlancesPluginModel._check_decorator
105		@GlancesPluginModel._log_result_decorator
106		def update(self):
107		"""Update the ports list."""
108		if self.input_method == 'local':
109		# Only refresh:
110		# * if there is not other scanning thread
111		# * every refresh seconds (define in the configuration file)
112		if self._thread is None:
113		thread_is_running = False
114		else:
115		thread_is_running = self._thread.is_alive()
116		if not thread_is_running:
117		# Run ports scanner
118		self._thread = ThreadScanner(self.stats)
119		self._thread.start()
120		else:
121		# Not available in SNMP mode
122		pass
123
124		return self.stats
125
126	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...
127		"""Return the alert status relative to the port scan return value."""
128		ret = 'OK'
129		if port['status'] is None:
130		ret = 'CAREFUL'
131		elif port['status'] == 0:
132		ret = 'CRITICAL'
133		elif (
134		isinstance(port['status'], (float, int))
135		and port['rtt_warning'] is not None
136		and port['status'] > port['rtt_warning']
137		):
138		ret = 'WARNING'
139
140		# Get stat name
141		stat_name = self.get_stat_name(header=header)
142
143		# Manage threshold
144		self.manage_threshold(stat_name, ret)
145
146		# Manage action
147		self.manage_action(stat_name, ret.lower(), header, port[self.get_key()])
148
149		return ret
150
151	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...
152		"""Return the alert status relative to the web/url scan return value."""
153		ret = 'OK'
154		if web['status'] is None:
155		ret = 'CAREFUL'
156		elif web['status'] not in [200, 301, 302]:
157		ret = 'CRITICAL'
158		elif web['rtt_warning'] is not None and web['elapsed'] > web['rtt_warning']:
159		ret = 'WARNING'
160
161		# Get stat name
162		stat_name = self.get_stat_name(header=header)
163
164		# Manage threshold
165		self.manage_threshold(stat_name, ret)
166
167		# Manage action
168		self.manage_action(stat_name, ret.lower(), header, web[self.get_key()])
169
170		return ret
171
172		def msg_curse(self, args=None, max_width=None):
173		"""Return the dict to display in the curse interface."""
174		# Init the return message
175		# Only process if stats exist and display plugin enable...
176		ret = []
177
178		if not self.stats or args.disable_ports:
179		return ret
180
181		# Max size for the interface name
182		if max_width:
183		name_max_width = max_width - 7
184		else:
185		# No max_width defined, return an emptu curse message
186		logger.debug("No max_width defined for the {} plugin, it will not be displayed.".format(self.plugin_name))
187		return ret
188
189		# Build the string message
190		for p in self.stats:
191		if 'host' in p:
192		if p['host'] is None:
193		status = 'None'
194		elif p['status'] is None:
195		status = 'Scanning'
196		elif isinstance(p['status'], bool_type) and p['status'] is True:
197		status = 'Open'
198		elif p['status'] == 0:
199		status = 'Timeout'
200		else:
201		# Convert second to ms
202		status = '{0:.0f}ms'.format(p['status'] * 1000.0)
203
204		msg = '{:{width}}'.format(p['description'][0:name_max_width], width=name_max_width)
205		ret.append(self.curse_add_line(msg))
206		msg = '{:>9}'.format(status)
207		ret.append(self.curse_add_line(msg, self.get_ports_alert(p, header=p['indice'] + '_rtt')))
208		ret.append(self.curse_new_line())
209		elif 'url' in p:
210		msg = '{:{width}}'.format(p['description'][0:name_max_width], width=name_max_width)
211		ret.append(self.curse_add_line(msg))
212		if isinstance(p['status'], numbers.Number):
213		status = 'Code {}'.format(p['status'])
214		elif p['status'] is None:
215		status = 'Scanning'
216		else:
217		status = p['status']
218		msg = '{:>9}'.format(status)
219		ret.append(self.curse_add_line(msg, self.get_web_alert(p, header=p['indice'] + '_rtt')))
220		ret.append(self.curse_new_line())
221
222		# Delete the last empty line
223		try:
224		ret.pop()
225		except IndexError:
226		pass
227
228		return ret
229
230
231		class ThreadScanner(threading.Thread):
232		"""
233		Specific thread for the port/web scanner.
234
235		stats is a list of dict
236		"""
237
238		def __init__(self, stats):
239		"""Init the class."""
240		logger.debug("ports plugin - Create thread for scan list {}".format(stats))
241		super(ThreadScanner, self).__init__()
242		# Event needed to stop properly the thread
243		self._stopper = threading.Event()
244		# The class return the stats as a list of dict
245		self._stats = stats
246		# Is part of Ports plugin
247		self.plugin_name = "ports"
248
249		def get_key(self):
250		"""Return the key of the list."""
251		return 'indice'
252
253		def run(self):
254		"""Grab the stats.
255
256		Infinite loop, should be stopped by calling the stop() method.
257		"""
258		for p in self._stats:
259		# End of the thread has been asked
260		if self.stopped():
261		break
262		# Scan a port (ICMP or TCP)
263		if 'port' in p:
264		self._port_scan(p)
265		# Had to wait between two scans
266		# If not, result are not ok
267		time.sleep(1)
268		# Scan an URL
269		elif 'url' in p and requests_tag:
270		self._web_scan(p)
271		# Add the key for every element
272		p['key'] = self.get_key()
273
274		@property
275		def stats(self):
276		"""Stats getter."""
277		return self._stats
278
279		@stats.setter
280		def stats(self, value):
281		"""Stats setter."""
282		self._stats = value
283
284		def stop(self, timeout=None):
285		"""Stop the thread."""
286		logger.debug("ports plugin - Close thread for scan list {}".format(self._stats))
287		self._stopper.set()
288
289		def stopped(self):
290		"""Return True is the thread is stopped."""
291		return self._stopper.is_set()
292
293		def _web_scan(self, web):
294		"""Scan the Web/URL (dict) and update the status key."""
295		try:
296		req = requests.head(
297		web['url'],
298		allow_redirects=True,
299		verify=web['ssl_verify'],
300		proxies=web['proxies'],
301		timeout=web['timeout'],
302		)
303		except Exception as e:
304		logger.debug(e)
305		web['status'] = 'Error'
306		web['elapsed'] = 0
307		else:
308		web['status'] = req.status_code
309		web['elapsed'] = req.elapsed.total_seconds()
310		return web
311
312		def _port_scan(self, port):
313		"""Scan the port structure (dict) and update the status key."""
314		if int(port['port']) == 0:
315		return self._port_scan_icmp(port)
316		else:
317		return self._port_scan_tcp(port)
318
319		def _resolv_name(self, hostname):
320		"""Convert hostname to IP address."""
321		ip = hostname
322		try:
323		ip = socket.gethostbyname(hostname)
324		except Exception as e:
325		logger.debug("{}: Cannot convert {} to IP address ({})".format(self.plugin_name, hostname, e))
326		return ip
327
328		def _port_scan_icmp(self, port):
329		"""Scan the (ICMP) port structure (dict) and update the status key."""
330		ret = None
331
332		# Create the ping command
333		# Use the system ping command because it already have the sticky bit set
334		# Python can not create ICMP packet with non root right
335		if WINDOWS:
336		timeout_opt = '-w'
337		count_opt = '-n'
338		elif MACOS or BSD:
339		timeout_opt = '-t'
340		count_opt = '-c'
341		else:
342		# Linux and co...
343		timeout_opt = '-W'
344		count_opt = '-c'
345		# Build the command line
346		# Note: Only string are allowed
347		cmd = [
348		'ping',
349		count_opt,
350		'1',
351		timeout_opt,
352		str(self._resolv_name(port['timeout'])),
353		self._resolv_name(port['host']),
354		]
355		fnull = open(os.devnull, 'w')
356
357		try:
358		counter = Counter()
359		ret = subprocess.check_call(cmd, stdout=fnull, stderr=fnull, close_fds=True)
360		if ret == 0:
361		port['status'] = counter.get()
362		else:
363		port['status'] = False
364		except subprocess.CalledProcessError:
365		# Correct issue #1084: No Offline status for timed-out ports
366		port['status'] = False
367		except Exception as e:
368		logger.debug("{}: Error while pinging host {} ({})".format(self.plugin_name, port['host'], e))
369
370		fnull.close()
371
372		return ret
373
374		def _port_scan_tcp(self, port):
375		"""Scan the (TCP) port structure (dict) and update the status key."""
376		ret = None
377
378		# Create and configure the scanning socket
379		try:
380		socket.setdefaulttimeout(port['timeout'])
381		_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
382		except Exception as e:
383		logger.debug("{}: Error while creating scanning socket ({})".format(self.plugin_name, e))
384
385		# Scan port
386		ip = self._resolv_name(port['host'])
387		counter = Counter()
388		try:
389		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...
390		except Exception as e:
391		logger.debug("{}: Error while scanning port {} ({})".format(self.plugin_name, port, e))
392		else:
393		if ret == 0:
394		port['status'] = counter.get()
395		else:
396		port['status'] = False
397		finally:
398		_socket.close()
399
400		return ret
401

nicolargo / glances

Push — master ( ee826a...d9056e )

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like