glances.plugins.ip.PluginModel.get_private_ipv4() - Code Metrics - Inspection of "Refactored `glances.plugins.ip.PluginModel.update`" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Pull Request — develop (#2998)

unknown

created 2024-11-01 15:46 UTC

glances.plugins.ip.PluginModel.get_private_ipv4() A

↳ Parent: glances.plugins.ip

Complexity

Conditions

Size

Total Lines	12
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	9
nop	2
dl	0
loc	12
rs	9.95
c	0
b	0
f	0

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

"""IP plugin."""

import threading

from glances.globals import json_loads, queue, urlopen_auth
from glances.logger import logger
from glances.plugins.plugin.model import GlancesPluginModel
from glances.timer import Timer, getTimeSinceLastUpdate

# Import plugin specific dependency
try:
    import netifaces
except ImportError as e:
    import_error_tag = True
    logger.warning(f"Missing Python Lib ({e}), IP plugin is disabled")
else:
    import_error_tag = False

# 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 = {
    'address': {
        'description': 'Private IP address',
    },
    'mask': {
        'description': 'Private IP mask',
    },
    'mask_cidr': {
        'description': 'Private IP mask in CIDR format',
        'unit': 'number',
    },
    'gateway': {
        'description': 'Private IP gateway',
    },
    'public_address': {
        'description': 'Public IP address',
    },
    'public_info_human': {
        'description': 'Public IP information',
    },
}


class PluginModel(GlancesPluginModel):
    """Glances IP Plugin.

    stats is a dict
    """

    _default_public_refresh_interval = 300

    def __init__(self, args=None, config=None):
        """Init the plugin."""
        super().__init__(args=args, config=config, fields_description=fields_description)

        # We want to display the stat in the curse interface
        self.display_curse = True

        # Public information (see issue #2732)
        self.public_address = ""
        self.public_info = ""
        self.public_api = self.get_conf_value("public_api", default=[None])[0]
        self.public_username = self.get_conf_value("public_username", default=[None])[0]
        self.public_password = self.get_conf_value("public_password", default=[None])[0]
        self.public_field = self.get_conf_value("public_field", default=[None])
        self.public_template = self.get_conf_value("public_template", default=[None])[0]
        self.public_disabled = (
            self.get_conf_value('public_disabled', default='False')[0].lower() != 'false'
            or self.public_api is None
            or self.public_field is None
        )
        self.public_address_refresh_interval = self.get_conf_value(
            "public_refresh_interval", default=self._default_public_refresh_interval
        )

    def get_private_ipv4(self, init):
        stop, stats = init
        # Get the default gateway thanks to the netifaces lib
        try:
            default_gw = netifaces.gateways()['default'][netifaces.AF_INET]
        except (KeyError, AttributeError) as e:
            logger.debug(f"Cannot grab default gateway IP address ({e})")
            stop = True
        else:
            stats['gateway'] = default_gw[0]

        return (stop, stats)

    def get_first_ipv4(self, init):
        stop, stats = init
        try:
            default_gw = netifaces.gateways()['default'][netifaces.AF_INET]
            address = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['addr']
            mask = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['netmask']
        except (KeyError, AttributeError) as e:
            logger.debug(f"Cannot grab private IP address ({e})")
            stop = True
        else:
            stats['address'] = address
            stats['mask'] = mask
            stats['mask_cidr'] = self.ip_to_cidr(stats['mask'])

        return (stop, stats)

    def get_public_ipv4(self, init):
        stop, stats = init
        time_since_update = getTimeSinceLastUpdate('public-ip')
        try:
            if not self.public_disabled and (
                self.public_address == "" or time_since_update > self.public_address_refresh_interval
            ):
                self.public_info = PublicIpInfo(self.public_api, self.public_username, self.public_password).get()
                self.public_address = self.public_info['ip']
        except (KeyError, AttributeError, TypeError) as e:
            logger.debug(f"Cannot grab public IP information ({e})")
        else:
            stats['public_address'] = (
                self.public_address if not self.args.hide_public_info else self.__hide_ip(self.public_address)
            )
            stats['public_info_human'] = self.public_info_for_human(self.public_info)

        return (True, stats)

    @GlancesPluginModel._check_decorator
    @GlancesPluginModel._log_result_decorator
    def update(self):
        """Update IP stats using the input method.

        :return: the stats dict
        """
        # Init new stats
        stats = self.get_init_value()

        # Revert

        if self.input_method == 'local' and not import_error_tag:
            init = False, stats
            # Private IP address
            # # Get the default gateway thanks to the netifaces lib
            # try:
            #     default_gw = netifaces.gateways()['default'][netifaces.AF_INET]
            # except (KeyError, AttributeError) as e:
            #     logger.debug(f"Cannot grab default gateway IP address ({e})")
            #     return self.get_init_value()
            # else:
            #     stats['gateway'] = default_gw[0]
            # If multiple IP addresses are available, only the one with the default gateway is returned
            init = False, stats
            stop, stats = self.get_private_ipv4(init)

            if not stop:
                # try:
                #     address = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['addr']
                #     mask = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['netmask']
                # except (KeyError, AttributeError) as e:
                #     logger.debug(f"Cannot grab private IP address ({e})")
                #     return self.get_init_value()
                # else:
                #     stats['address'] = address
                #     stats['mask'] = mask
                #     stats['mask_cidr'] = self.ip_to_cidr(stats['mask'])
                stop, stats = self.get_first_ipv4(init)

            # Public IP address
            # time_since_update = getTimeSinceLastUpdate('public-ip')
            # try:
            #     if not self.public_disabled and (
            #         self.public_address == "" or time_since_update > self.public_address_refresh_interval
            #     ):
            #         self.public_info = PublicIpInfo(self.public_api, self.public_username, self.public_password).get()
            #         self.public_address = self.public_info['ip']
            # except (KeyError, AttributeError, TypeError) as e:
            #     logger.debug(f"Cannot grab public IP information ({e})")
            # else:
            #     stats['public_address'] = (
            #         self.public_address if not self.args.hide_public_info else self.__hide_ip(self.public_address)
            #     )
            #     stats['public_info_human'] = self.public_info_for_human(self.public_info)
            if not stop:
                stop, stats = self.get_public_ipv4(init)

        elif self.input_method == 'snmp':
            # Not implemented yet
            pass

        # Update the stats
        self.stats = stats

        return self.stats

    def __hide_ip(self, ip):
        """Hide last to digit of the given IP address"""
        return '.'.join(ip.split('.')[0:2]) + '.*.*'

    def msg_curse(self, args=None, max_width=None):
        """Return the dict to display in the curse interface."""
        # Init the return message
        ret = []

        # Only process if stats exist and display plugin enable...
        if not self.stats or self.is_disabled() or import_error_tag:
            return ret

        # Build the string message
        msg = ' - '
        ret.append(self.curse_add_line(msg, optional=True))

        # Start with the private IP information
        msg = 'IP '
        ret.append(self.curse_add_line(msg, 'TITLE', optional=True))
        if 'address' in self.stats:
            msg = '{}'.format(self.stats['address'])
            ret.append(self.curse_add_line(msg, optional=True))
        if 'mask_cidr' in self.stats:
            # VPN with no internet access (issue #842)
            msg = '/{}'.format(self.stats['mask_cidr'])
            ret.append(self.curse_add_line(msg, optional=True))

        # Then with the public IP information
        try:
            msg_pub = '{}'.format(self.stats['public_address'])
        except (UnicodeEncodeError, KeyError):
            # Add KeyError exception (see https://github.com/nicolargo/glances/issues/1469)
            pass
        else:
            if self.stats['public_address']:
                msg = ' Pub '
                ret.append(self.curse_add_line(msg, 'TITLE', optional=True))
                ret.append(self.curse_add_line(msg_pub, optional=True))

            if self.stats['public_info_human']:
                ret.append(self.curse_add_line(' {}'.format(self.stats['public_info_human']), optional=True))

        return ret

    def public_info_for_human(self, public_info):
        """Return the data to pack to the client."""
        if not public_info:
            return ''

        return self.public_template.format(**public_info)

    @staticmethod
    def ip_to_cidr(ip):
        """Convert IP address to CIDR.

        Example: '255.255.255.0' will return 24
        """
        # Thanks to @Atticfire
        # See https://github.com/nicolargo/glances/issues/1417#issuecomment-469894399
        if ip is None:
            # Correct issue #1528
            return 0
        return sum(bin(int(x)).count('1') for x in ip.split('.'))


class PublicIpInfo:
    """Get public IP information from online service."""

    def __init__(self, url, username, password, timeout=2):
        """Init the class."""
        self.url = url
        self.username = username
        self.password = password
        self.timeout = timeout

    def get(self):
        """Return the public IP information returned by one of the online service."""
        q = queue.Queue()

        t = threading.Thread(target=self._get_ip_public_info, args=(q, self.url, self.username, self.password))
        t.daemon = True
        t.start()

        timer = Timer(self.timeout)
        info = None
        while not timer.finished() and info is None:
            if q.qsize() > 0:
                info = q.get()

        return info

    def _get_ip_public_info(self, queue_target, url, username, password):
        """Request the url service and put the result in the queue_target."""
        try:
            response = urlopen_auth(url, username, password).read()
        except Exception as e:
            logger.debug(f"IP plugin - Cannot get public IP information from {url} ({e})")
            queue_target.put(None)
        else:
            try:
                queue_target.put(json_loads(response))
            except (ValueError, KeyError) as e:
                logger.debug(f"IP plugin - Cannot load public IP information from {url} ({e})")
                queue_target.put(None)


1			#
2			# This file is part of Glances.
3			#
4			# SPDX-FileCopyrightText: 2024 Nicolas Hennion <[email protected]>
5			#
6			# SPDX-License-Identifier: LGPL-3.0-only
7			#
8
9			"""IP plugin."""
10
11			import threading
12
13			from glances.globals import json_loads, queue, urlopen_auth
14			from glances.logger import logger
15			from glances.plugins.plugin.model import GlancesPluginModel
16			from glances.timer import Timer, getTimeSinceLastUpdate
17
18			# Import plugin specific dependency
19			try:
20			import netifaces
21			except ImportError as e:
22			import_error_tag = True
23			logger.warning(f"Missing Python Lib ({e}), IP plugin is disabled")
24			else:
25			import_error_tag = False
26
27			# Fields description
28			# description: human readable description
29			# short_name: shortname to use un UI
30			# unit: unit type
31			# rate: is it a rate ? If yes, // by time_since_update when displayed,
32			# min_symbol: Auto unit should be used if value > than 1 'X' (K, M, G)...
33			fields_description = {
34			'address': {
35			'description': 'Private IP address',
36			},
37			'mask': {
38			'description': 'Private IP mask',
39			},
40			'mask_cidr': {
41			'description': 'Private IP mask in CIDR format',
42			'unit': 'number',
43			},
44			'gateway': {
45			'description': 'Private IP gateway',
46			},
47			'public_address': {
48			'description': 'Public IP address',
49			},
50			'public_info_human': {
51			'description': 'Public IP information',
52			},
53			}
54
55
56			class PluginModel(GlancesPluginModel):
57			"""Glances IP Plugin.
58
59			stats is a dict
60			"""
61
62			_default_public_refresh_interval = 300
63
64			def __init__(self, args=None, config=None):
65			"""Init the plugin."""
66			super().__init__(args=args, config=config, fields_description=fields_description)
67
68			# We want to display the stat in the curse interface
69			self.display_curse = True
70
71			# Public information (see issue #2732)
72			self.public_address = ""
73			self.public_info = ""
74			self.public_api = self.get_conf_value("public_api", default=[None])[0]
75			self.public_username = self.get_conf_value("public_username", default=[None])[0]
76			self.public_password = self.get_conf_value("public_password", default=[None])[0]
77			self.public_field = self.get_conf_value("public_field", default=[None])
78			self.public_template = self.get_conf_value("public_template", default=[None])[0]
79			self.public_disabled = (
80			self.get_conf_value('public_disabled', default='False')[0].lower() != 'false'
81			or self.public_api is None
82			or self.public_field is None
83			)
84			self.public_address_refresh_interval = self.get_conf_value(
85			"public_refresh_interval", default=self._default_public_refresh_interval
86			)
87
88			def get_private_ipv4(self, init):
89			stop, stats = init
90			# Get the default gateway thanks to the netifaces lib
91			try:
92			default_gw = netifaces.gateways()['default'][netifaces.AF_INET]
93			except (KeyError, AttributeError) as e:
94			logger.debug(f"Cannot grab default gateway IP address ({e})")
95			stop = True
96			else:
97			stats['gateway'] = default_gw[0]
98
99			return (stop, stats)
100
101			def get_first_ipv4(self, init):
102			stop, stats = init
103			try:
104			default_gw = netifaces.gateways()['default'][netifaces.AF_INET]
105			address = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['addr']
106			mask = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['netmask']
107			except (KeyError, AttributeError) as e:
108			logger.debug(f"Cannot grab private IP address ({e})")
109			stop = True
110			else:
111			stats['address'] = address
112			stats['mask'] = mask
113			stats['mask_cidr'] = self.ip_to_cidr(stats['mask'])
114
115			return (stop, stats)
116
117			def get_public_ipv4(self, init):
118			stop, stats = init
119			time_since_update = getTimeSinceLastUpdate('public-ip')
120			try:
121			if not self.public_disabled and (
122			self.public_address == "" or time_since_update > self.public_address_refresh_interval
123			):
124			self.public_info = PublicIpInfo(self.public_api, self.public_username, self.public_password).get()
125			self.public_address = self.public_info['ip']
126			except (KeyError, AttributeError, TypeError) as e:
127			logger.debug(f"Cannot grab public IP information ({e})")
128			else:
129			stats['public_address'] = (
130			self.public_address if not self.args.hide_public_info else self.__hide_ip(self.public_address)
131			)
132			stats['public_info_human'] = self.public_info_for_human(self.public_info)
133
134			return (True, stats)
135
136			@GlancesPluginModel._check_decorator
137			@GlancesPluginModel._log_result_decorator
138			def update(self):
139			"""Update IP stats using the input method.
140
141			:return: the stats dict
142			"""
143			# Init new stats
144			stats = self.get_init_value()
145
146			# Revert
147
148			if self.input_method == 'local' and not import_error_tag:
149			init = False, stats
150			# Private IP address
151			# # Get the default gateway thanks to the netifaces lib
152			# try:
153			# default_gw = netifaces.gateways()['default'][netifaces.AF_INET]
154			# except (KeyError, AttributeError) as e:
155			# logger.debug(f"Cannot grab default gateway IP address ({e})")
156			# return self.get_init_value()
157			# else:
158			# stats['gateway'] = default_gw[0]
159			# If multiple IP addresses are available, only the one with the default gateway is returned
160			init = False, stats
161			stop, stats = self.get_private_ipv4(init)
162
163			if not stop:
164			# try:
165			# address = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['addr']
166			# mask = netifaces.ifaddresses(default_gw[1])[netifaces.AF_INET][0]['netmask']
167			# except (KeyError, AttributeError) as e:
168			# logger.debug(f"Cannot grab private IP address ({e})")
169			# return self.get_init_value()
170			# else:
171			# stats['address'] = address
172			# stats['mask'] = mask
173			# stats['mask_cidr'] = self.ip_to_cidr(stats['mask'])
174			stop, stats = self.get_first_ipv4(init)
175
176			# Public IP address
177			# time_since_update = getTimeSinceLastUpdate('public-ip')
178			# try:
179			# if not self.public_disabled and (
180			# self.public_address == "" or time_since_update > self.public_address_refresh_interval
181			# ):
182			# self.public_info = PublicIpInfo(self.public_api, self.public_username, self.public_password).get()
183			# self.public_address = self.public_info['ip']
184			# except (KeyError, AttributeError, TypeError) as e:
185			# logger.debug(f"Cannot grab public IP information ({e})")
186			# else:
187			# stats['public_address'] = (
188			# self.public_address if not self.args.hide_public_info else self.__hide_ip(self.public_address)
189			# )
190			# stats['public_info_human'] = self.public_info_for_human(self.public_info)
191			if not stop:
192			stop, stats = self.get_public_ipv4(init)
193
194			elif self.input_method == 'snmp':
195			# Not implemented yet
196			pass
197
198			# Update the stats
199			self.stats = stats
200
201			return self.stats
202
203			def __hide_ip(self, ip):
204			"""Hide last to digit of the given IP address"""
205			return '.'.join(ip.split('.')[0:2]) + '..'
206
207			def msg_curse(self, args=None, max_width=None):
208			"""Return the dict to display in the curse interface."""
209			# Init the return message
210			ret = []
211
212			# Only process if stats exist and display plugin enable...
213			if not self.stats or self.is_disabled() or import_error_tag:
214			return ret
215
216			# Build the string message
217			msg = ' - '
218			ret.append(self.curse_add_line(msg, optional=True))
219
220			# Start with the private IP information
221			msg = 'IP '
222			ret.append(self.curse_add_line(msg, 'TITLE', optional=True))
223			if 'address' in self.stats:
224			msg = '{}'.format(self.stats['address'])
225			ret.append(self.curse_add_line(msg, optional=True))
226			if 'mask_cidr' in self.stats:
227			# VPN with no internet access (issue #842)
228			msg = '/{}'.format(self.stats['mask_cidr'])
229			ret.append(self.curse_add_line(msg, optional=True))
230
231			# Then with the public IP information
232			try:
233			msg_pub = '{}'.format(self.stats['public_address'])
234			except (UnicodeEncodeError, KeyError):
235			# Add KeyError exception (see https://github.com/nicolargo/glances/issues/1469)
236			pass
237			else:
238			if self.stats['public_address']:
239			msg = ' Pub '
240			ret.append(self.curse_add_line(msg, 'TITLE', optional=True))
241			ret.append(self.curse_add_line(msg_pub, optional=True))
242
243			if self.stats['public_info_human']:
244			ret.append(self.curse_add_line(' {}'.format(self.stats['public_info_human']), optional=True))
245
246			return ret
247
248			def public_info_for_human(self, public_info):
249			"""Return the data to pack to the client."""
250			if not public_info:
251			return ''
252
253			return self.public_template.format(**public_info)
254
255			@staticmethod
256			def ip_to_cidr(ip):
257			"""Convert IP address to CIDR.
258
259			Example: '255.255.255.0' will return 24
260			"""
261			# Thanks to @Atticfire
262			# See https://github.com/nicolargo/glances/issues/1417#issuecomment-469894399
263			if ip is None:
264			# Correct issue #1528
265			return 0
266			return sum(bin(int(x)).count('1') for x in ip.split('.'))
267
268
269			class PublicIpInfo:
270			"""Get public IP information from online service."""
271
272			def __init__(self, url, username, password, timeout=2):
273			"""Init the class."""
274			self.url = url
275			self.username = username
276			self.password = password
277			self.timeout = timeout
278
279			def get(self):
280			"""Return the public IP information returned by one of the online service."""
281			q = queue.Queue()
282
283			t = threading.Thread(target=self._get_ip_public_info, args=(q, self.url, self.username, self.password))
284			t.daemon = True
285			t.start()
286
287			timer = Timer(self.timeout)
288			info = None
289			while not timer.finished() and info is None:
290			if q.qsize() > 0:
291			info = q.get()
292
293			return info
294
295			def _get_ip_public_info(self, queue_target, url, username, password):
296			"""Request the url service and put the result in the queue_target."""
297			try:
298			response = urlopen_auth(url, username, password).read()
299			except Exception as e:
300			logger.debug(f"IP plugin - Cannot get public IP information from {url} ({e})")
301			queue_target.put(None)
302			else:
303			try:
304			queue_target.put(json_loads(response))
305			except (ValueError, KeyError) as e:
306			logger.debug(f"IP plugin - Cannot load public IP information from {url} ({e})")
307			queue_target.put(None)
308

nicolargo / glances

Pull Request — develop (#2998)

glances.plugins.ip.PluginModel.get_private_ipv4() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like