glances.plugins.ports.ThreadScanner.stop() - Code Metrics - Inspection of "Fixed typos, refactored `glances.plugins.ports.Pl..." - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — develop (#2957)

unknown

created 2024-10-04 19:36 UTC

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

↳ Parent: glances.plugins.ports

Complexity

Conditions

Size

Total Lines	4
Code Lines	3

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	3
nop	2
dl	0
loc	4
rs	10
c	0
b	0
f	0

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

"""Ports scanner plugin."""

import numbers
import os
import socket
import subprocess
import threading
import time
from functools import partial, reduce

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

try:
    import requests

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

# 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().__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().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 set_status_if_host(self, 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 = '{:.0f}ms'.format(p['status'] * 1000.0)

        return status

    def set_status_if_url(self, p):
        if isinstance(p['status'], numbers.Number):
            status = 'Code {}'.format(p['status'])
        elif p['status'] is None:
            status = 'Scanning'
        else:
            status = p['status']

        return status

    def build_str(self, name_max_width, ret, p):
        if 'host' in p:
            status = self.set_status_if_host(p)
        elif 'url' in p:
            status = self.set_status_if_url(p)

        msg = '{:{width}}'.format(p['description'][0:name_max_width], width=name_max_width)
        ret.append(self.curse_add_line(msg))
        msg = f'{status:>9}'
        ret.append(self.curse_add_line(msg, self.get_ports_alert(p, header=p['indice'] + '_rtt')))
        ret.append(self.curse_new_line())

        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...
        init = []

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

        # Max size for the interface name
        if max_width:
            name_max_width = max_width - 7
        else:
            # No max_width defined, return an empty curse message
            logger.debug(f"No max_width defined for the {self.plugin_name} plugin, it will not be displayed.")
            return init

        # Build the string message
        build_str_with_this_max_width = partial(self.build_str, name_max_width)
        ret = reduce(build_str_with_this_max_width, self.stats, init)

        # 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(f"ports plugin - Create thread for scan list {stats}")
        super().__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(f"ports plugin - Close thread for scan list {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)
        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(f"{self.plugin_name}: Cannot convert {hostname} to IP address ({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(f"{self.plugin_name}: Error while creating scanning socket ({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(f"{self.plugin_name}: Error while scanning port {port} ({e})")
        else:
            if ret == 0:
                port['status'] = counter.get()
            else:
                port['status'] = False
        finally:
            _socket.close()

        return ret


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

nicolargo / glances

Pull Request — develop (#2957)

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like