ThreadScanner._port_scan_tcp() - Code Metrics - Inspection of "version 2.7" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 3dcd25...20576f )

by Nicolas

created 2016-09-10 09:14 UTC

ThreadScanner._port_scan_tcp() B

↳ Parent: ThreadScanner

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	5
dl	0
loc	27
rs	8.0894
c	0
b	0
f	0

# -*- coding: utf-8 -*-
#
# This file is part of Glances.
#
# Copyright (C) 2016 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 types
import time

from glances.globals import WINDOWS
from glances.ports_list import GlancesPortsList
from glances.timer import Timer, Counter
from glances.logger import logger
from glances.plugins.glances_plugin import GlancesPlugin


class Plugin(GlancesPlugin):

    """Glances ports scanner plugin."""

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

        # 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_alert(self, port, header="", log=False):
        """Return the alert status relative to the port scan return value."""

        if port['status'] is None:
            return 'CAREFUL'
        elif port['status'] == 0:
            return 'CRITICAL'
        elif isinstance(port['status'], (float, int)) and \
                port['rtt_warning'] is not None and \
                port['status'] > port['rtt_warning']:
            return 'WARNING'

        return 'OK'

    def msg_curse(self, args=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

        # Build the string message
        for p in self.stats:
            if p['status'] is None:
                status = 'Scanning'
            elif isinstance(p['status'], types.BooleanType) 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 = '{:14.14} '.format(p['description'])
            ret.append(self.curse_add_line(msg))
            msg = '{:>8}'.format(status)
            ret.append(self.curse_add_line(msg, self.get_alert(p)))
            ret.append(self.curse_new_line())

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

        return ret

    def _port_scan_all(self, stats):
        """Scan all host/port of the given stats"""
        for p in stats:
            self._port_scan(p)
            # Had to wait between two scans
            # If not, result are not ok
            time.sleep(1)


class ThreadScanner(threading.Thread):
    """
    Specific thread for the port 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):
        """Function called to grab stats.
        Infinite loop, should be stopped by calling the stop() method"""

        for p in self._stats:
            self._port_scan(p)
            if self.stopped():
                break
            # Had to wait between two scans
            # If not, result are not ok
            time.sleep(1)

    @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 _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("{0}: Can not convert {1} to IP address ({2})".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
        cmd = ['ping', '-n' if WINDOWS else '-c', '1', 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 Exception as e:
            logger.debug("{0}: Error while pinging host ({2})".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("{0}: 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("{0}: Error while scanning port {1} ({2})".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) 2016 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 types
27			import time
28
29			from glances.globals import WINDOWS
30			from glances.ports_list import GlancesPortsList
31			from glances.timer import Timer, Counter
32			from glances.logger import logger
33			from glances.plugins.glances_plugin import GlancesPlugin
34
35
36			class Plugin(GlancesPlugin):
37
38			"""Glances ports scanner plugin."""
39
40			def __init__(self, args=None, config=None):
41			"""Init the plugin."""
42			super(Plugin, self).__init__(args=args)
43			self.args = args
44			self.config = config
45
46			# We want to display the stat in the curse interface
47			self.display_curse = True
48
49			# Init stats
50			self.stats = GlancesPortsList(config=config, args=args).get_ports_list()
51
52			# Init global Timer
53			self.timer_ports = Timer(0)
54
55			# Global Thread running all the scans
56			self._thread = None
57
58			def exit(self):
59			"""Overwrite the exit method to close threads"""
60			if self._thread is not None:
61			self._thread.stop()
62			# Call the father class
63			super(Plugin, self).exit()
64
65			@GlancesPlugin._log_result_decorator
66			def update(self):
67			"""Update the ports list."""
68
69			if self.input_method == 'local':
70			# Only refresh:
71			# * if there is not other scanning thread
72			# * every refresh seconds (define in the configuration file)
73			if self._thread is None:
74			thread_is_running = False
75			else:
76			thread_is_running = self._thread.isAlive()
77			if self.timer_ports.finished() and not thread_is_running:
78			# Run ports scanner
79			self._thread = ThreadScanner(self.stats)
80			self._thread.start()
81			# Restart timer
82			if len(self.stats) > 0:
83			self.timer_ports = Timer(self.stats[0]['refresh'])
84			else:
85			self.timer_ports = Timer(0)
86			else:
87			# Not available in SNMP mode
88			pass
89
90			return self.stats
91
92			def get_alert(self, port, header="", log=False):
93			"""Return the alert status relative to the port scan return value."""
94
95			if port['status'] is None:
96			return 'CAREFUL'
97			elif port['status'] == 0:
98			return 'CRITICAL'
99			elif isinstance(port['status'], (float, int)) and \
100			port['rtt_warning'] is not None and \
101			port['status'] > port['rtt_warning']:
102			return 'WARNING'
103
104			return 'OK'
105
106			def msg_curse(self, args=None):
107			"""Return the dict to display in the curse interface."""
108			# Init the return message
109			# Only process if stats exist and display plugin enable...
110			ret = []
111
112			if not self.stats or args.disable_ports:
113			return ret
114
115			# Build the string message
116			for p in self.stats:
117			if p['status'] is None:
118			status = 'Scanning'
119			elif isinstance(p['status'], types.BooleanType) and p['status'] is True:
120			status = 'Open'
121			elif p['status'] == 0:
122			status = 'Timeout'
123			else:
124			# Convert second to ms
125			status = '{0:.0f}ms'.format(p['status'] * 1000.0)
126
127			msg = '{:14.14} '.format(p['description'])
128			ret.append(self.curse_add_line(msg))
129			msg = '{:>8}'.format(status)
130			ret.append(self.curse_add_line(msg, self.get_alert(p)))
131			ret.append(self.curse_new_line())
132
133			# Delete the last empty line
134			try:
135			ret.pop()
136			except IndexError:
137			pass
138
139			return ret
140
141			def _port_scan_all(self, stats):
142			"""Scan all host/port of the given stats"""
143			for p in stats:
144			self._port_scan(p)
145			# Had to wait between two scans
146			# If not, result are not ok
147			time.sleep(1)
148
149
150			class ThreadScanner(threading.Thread):
151			"""
152			Specific thread for the port scanner.
153
154			stats is a list of dict
155			"""
156
157			def __init__(self, stats):
158			"""Init the class"""
159			logger.debug("ports plugin - Create thread for scan list {}".format(stats))
160			super(ThreadScanner, self).__init__()
161			# Event needed to stop properly the thread
162			self._stopper = threading.Event()
163			# The class return the stats as a list of dict
164			self._stats = stats
165			# Is part of Ports plugin
166			self.plugin_name = "ports"
167
168			def run(self):
169			"""Function called to grab stats.
170			Infinite loop, should be stopped by calling the stop() method"""
171
172			for p in self._stats:
173			self._port_scan(p)
174			if self.stopped():
175			break
176			# Had to wait between two scans
177			# If not, result are not ok
178			time.sleep(1)
179
180			@property
181			def stats(self):
182			"""Stats getter"""
183			return self._stats
184
185			@stats.setter
186			def stats(self, value):
187			"""Stats setter"""
188			self._stats = value
189
190			def stop(self, timeout=None):
191			"""Stop the thread"""
192			logger.debug("ports plugin - Close thread for scan list {}".format(self._stats))
193			self._stopper.set()
194
195			def stopped(self):
196			"""Return True is the thread is stopped"""
197			return self._stopper.isSet()
198
199			def _port_scan(self, port):
200			"""Scan the port structure (dict) and update the status key"""
201			if int(port['port']) == 0:
202			return self._port_scan_icmp(port)
203			else:
204			return self._port_scan_tcp(port)
205
206			def _resolv_name(self, hostname):
207			"""Convert hostname to IP address"""
208			ip = hostname
209			try:
210			ip = socket.gethostbyname(hostname)
211			except Exception as e:
212			logger.debug("{0}: Can not convert {1} to IP address ({2})".format(self.plugin_name, hostname, e))
213			return ip
214
215			def _port_scan_icmp(self, port):
216			"""Scan the (ICMP) port structure (dict) and update the status key"""
217			ret = None
218
219			# Create the ping command
220			# Use the system ping command because it already have the steacky bit set
221			# Python can not create ICMP packet with non root right
222			cmd = ['ping', '-n' if WINDOWS else '-c', '1', self._resolv_name(port['host'])]
223			fnull = open(os.devnull, 'w')
224
225			try:
226			counter = Counter()
227			ret = subprocess.check_call(cmd, stdout=fnull, stderr=fnull, close_fds=True)
228			if ret == 0:
229			port['status'] = counter.get()
230			else:
231			port['status'] = False
232			except Exception as e:
233			logger.debug("{0}: Error while pinging host ({2})".format(self.plugin_name, port['host'], e))
234
235			return ret
236
237			def _port_scan_tcp(self, port):
238			"""Scan the (TCP) port structure (dict) and update the status key"""
239			ret = None
240
241			# Create and configure the scanning socket
242			try:
243			socket.setdefaulttimeout(port['timeout'])
244			_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
245			except Exception as e:
246			logger.debug("{0}: Error while creating scanning socket".format(self.plugin_name))
247
248			# Scan port
249			ip = self._resolv_name(port['host'])
250			counter = Counter()
251			try:
252			ret = _socket.connect_ex((ip, int(port['port'])))
253			except Exception as e:
254			logger.debug("{0}: Error while scanning port {1} ({2})".format(self.plugin_name, port, e))
255			else:
256			if ret == 0:
257			port['status'] = counter.get()
258			else:
259			port['status'] = False
260			finally:
261			_socket.close()
262
263			return ret
264

nicolargo / glances

Push — master ( 3dcd25...20576f )

ThreadScanner._port_scan_tcp() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like