glances.plugins.ports

Complexity

Total Complexity

63

Size/Duplication

Total Lines	353
Duplicated Lines	12.46 %

Importance

Changes

0

17 Methods

Rating	Name	Duplication	Size	Complexity
A	PluginModel.get_web_alert()	20	20	5
A	ThreadScanner.stats()	0	4	1
A	PluginModel.update()	0	21	4
B	ThreadScanner.run()	0	18	6
A	PluginModel.exit()	0	6	2
A	ThreadScanner._port_scan()	0	6	2
A	ThreadScanner.stop()	0	4	1
F	PluginModel.msg_curse()	0	52	14
A	PluginModel.get_key()	0	3	1
A	ThreadScanner._resolv_name()	0	8	2
A	ThreadScanner.__init__()	0	10	1
A	ThreadScanner.stopped()	0	3	1
A	ThreadScanner._web_scan()	0	18	3
B	ThreadScanner._port_scan_icmp()	0	45	7
A	ThreadScanner._port_scan_tcp()	0	27	5
B	PluginModel.get_ports_alert()	24	24	6
A	PluginModel.__init__()	0	17	1

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# SPDX-FileCopyrightText: 2022 Nicolas Hennion <nicolas@nicolargo.com>
#
# 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):

This code seems to be duplicated in your project.

        """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):

This code seems to be duplicated in your project.

        """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'])))

The variable _socket does not seem to be defined for all execution paths.

        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