glances.server.GlancesInstance.__getattr__() - Code Metrics - Inspection of "Glances 4.0.8 - Update release note" - nicolargo/glances - Measure and Improve Code Quality continuously with Scrutinizer

Test Failed

Push — master ( b1b8ea...183265 )

by Nicolas

created 2024-06-08 08:28 UTC

glances.server.GlancesInstance.getattr() A

↳ Parent: glances.server

Complexity

Conditions

Size

Total Lines	19
Code Lines	9

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	9
nop	2
dl	0
loc	19
rs	9.95
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
#

"""Manage the Glances server."""

import json
import socket
import sys
from base64 import b64decode

from glances import __version__
from glances.autodiscover import GlancesAutoDiscoverClient
from glances.globals import SimpleXMLRPCRequestHandler, SimpleXMLRPCServer
from glances.logger import logger
from glances.stats_server import GlancesStatsServer
from glances.timer import Timer


class GlancesXMLRPCHandler(SimpleXMLRPCRequestHandler):
    """Main XML-RPC handler."""

    rpc_paths = ('/RPC2',)

    def end_headers(self):
        # Hack to add a specific header
        # Thk to: https://gist.github.com/rca/4063325
        self.send_my_headers()
        super().end_headers()

    def send_my_headers(self):
        # Specific header is here (solved the issue #227)
        self.send_header("Access-Control-Allow-Origin", "*")

    def authenticate(self, headers):
        # auth = headers.get('Authorization')
        try:
            (basic, _, encoded) = headers.get('Authorization').partition(' ')
        except Exception:
            # Client did not ask for authentication
            # If server need it then exit
            return not self.server.isAuth
        else:
            # Client authentication
            (basic, _, encoded) = headers.get('Authorization').partition(' ')
            assert basic == 'Basic', 'Only basic authentication supported'
            # Encoded portion of the header is a string
            # Need to convert to byte-string
            encoded_byte_string = encoded.encode()
            # Decode base64 byte string to a decoded byte string
            decoded_bytes = b64decode(encoded_byte_string)
            # Convert from byte string to a regular string
            decoded_string = decoded_bytes.decode()
            # Get the username and password from the string
            (username, _, password) = decoded_string.partition(':')
            # Check that username and password match internal global dictionary
            return self.check_user(username, password)

    def check_user(self, username, password):
        # Check username and password in the dictionary
        if username in self.server.user_dict:
            from glances.password import GlancesPassword

            # TODO: config is not taken into account: it may be a problem ?
            pwd = GlancesPassword(username=username, config=None)
            return pwd.check_password(self.server.user_dict[username], password)
        return False

    def parse_request(self):
        if SimpleXMLRPCRequestHandler.parse_request(self):
            # Next we authenticate
            if self.authenticate(self.headers):
                return True
            # if authentication fails, tell the client
            self.send_error(401, 'Authentication failed')
        return False

    def log_message(self, log_format, *args):
        # No message displayed on the server side
        pass


class GlancesXMLRPCServer(SimpleXMLRPCServer):
    """Init a SimpleXMLRPCServer instance (IPv6-ready)."""

    finished = False

    def __init__(self, bind_address, bind_port=61209, requestHandler=GlancesXMLRPCHandler, config=None):
        self.bind_address = bind_address
        self.bind_port = bind_port
        self.config = config
        try:
            self.address_family = socket.getaddrinfo(bind_address, bind_port)[0][0]
        except OSError as e:
            logger.error(f"Couldn't open socket: {e}")
            sys.exit(1)

        super().__init__((bind_address, bind_port), requestHandler)

    def end(self):
        """Stop the server"""
        self.server_close()
        self.finished = True

    def serve_forever(self):
        """Main loop"""
        while not self.finished:
            self.handle_request()


class GlancesInstance:
    """All the methods of this class are published as XML-RPC methods."""

    def __init__(self, config=None, args=None):
        # Init stats
        self.stats = GlancesStatsServer(config=config, args=args)

        # Initial update
        self.stats.update()

        # cached_time is the minimum time interval between stats updates
        # i.e. XML/RPC calls will not retrieve updated info until the time
        # since last update is passed (will retrieve old cached info instead)
        self.timer = Timer(0)
        self.cached_time = args.cached_time

    def __update__(self):
        # Never update more than 1 time per cached_time
        if self.timer.finished():
            self.stats.update()
            self.timer = Timer(self.cached_time)

    def init(self):
        # Return the Glances version
        return __version__

    def getAll(self):
        # Update and return all the stats
        self.__update__()
        return json.dumps(self.stats.getAll())

    def getAllPlugins(self):
        # Return the plugins list
        return json.dumps(self.stats.getPluginsList())

    def getAllLimits(self):
        # Return all the plugins limits
        return json.dumps(self.stats.getAllLimitsAsDict())

    def getAllViews(self):
        # Return all the plugins views
        return json.dumps(self.stats.getAllViewsAsDict())

    def getPlugin(self, plugin):
        # Update and return the plugin stat
        self.__update__()
        return json.dumps(self.stats.get_plugin(plugin).get_raw())

    def getPluginView(self, plugin):
        # Update and return the plugin view
        return json.dumps(self.stats.get_plugin(plugin).get_views())


class GlancesServer:
    """This class creates and manages the TCP server."""

    def __init__(self, requestHandler=GlancesXMLRPCHandler, config=None, args=None):
        # Args
        self.args = args

        # Init the XML RPC server
        try:
            self.server = GlancesXMLRPCServer(args.bind_address, args.port, requestHandler, config=config)
        except Exception as e:
            logger.critical(f"Cannot start Glances server: {e}")
            sys.exit(2)
        else:
            print(f'Glances XML-RPC server is running on {args.bind_address}:{args.port}')

        # The users dict
        # username / password couple
        # By default, no auth is needed
        self.server.user_dict = {}
        self.server.isAuth = False

        # Register functions
        self.server.register_introspection_functions()
        self.server.register_instance(GlancesInstance(config, args))

        if not self.args.disable_autodiscover:
            # Note: The Zeroconf service name will be based on the hostname
            # Correct issue: Zeroconf problem with zeroconf service name #889
            logger.info('Autodiscover is enabled with service name {}'.format(socket.gethostname().split('.', 1)[0]))
            self.autodiscover_client = GlancesAutoDiscoverClient(socket.gethostname().split('.', 1)[0], args)
        else:
            logger.info("Glances autodiscover announce is disabled")

    def add_user(self, username, password):
        """Add an user to the dictionary."""
        self.server.user_dict[username] = password
        self.server.isAuth = True

    def serve_forever(self):
        """Call the main loop."""
        # Set the server login/password (if -P/--password tag)
        if self.args.password != "":
            self.add_user(self.args.username, self.args.password)
        # Serve forever
        self.server.serve_forever()

    def end(self):
        """End of the Glances server session."""
        if not self.args.disable_autodiscover:
            self.autodiscover_client.close()
        self.server.end()


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			"""Manage the Glances server."""
10
11			import json
12			import socket
13			import sys
14			from base64 import b64decode
15
16			from glances import __version__
17			from glances.autodiscover import GlancesAutoDiscoverClient
18			from glances.globals import SimpleXMLRPCRequestHandler, SimpleXMLRPCServer
19			from glances.logger import logger
20			from glances.stats_server import GlancesStatsServer
21			from glances.timer import Timer
22
23
24			class GlancesXMLRPCHandler(SimpleXMLRPCRequestHandler):
25			"""Main XML-RPC handler."""
26
27			rpc_paths = ('/RPC2',)
28
29			def end_headers(self):
30			# Hack to add a specific header
31			# Thk to: https://gist.github.com/rca/4063325
32			self.send_my_headers()
33			super().end_headers()
34
35			def send_my_headers(self):
36			# Specific header is here (solved the issue #227)
37			self.send_header("Access-Control-Allow-Origin", "*")
38
39			def authenticate(self, headers):
40			# auth = headers.get('Authorization')
41			try:
42			(basic, _, encoded) = headers.get('Authorization').partition(' ')
43			except Exception:
44			# Client did not ask for authentication
45			# If server need it then exit
46			return not self.server.isAuth
47			else:
48			# Client authentication
49			(basic, _, encoded) = headers.get('Authorization').partition(' ')
50			assert basic == 'Basic', 'Only basic authentication supported'
51			# Encoded portion of the header is a string
52			# Need to convert to byte-string
53			encoded_byte_string = encoded.encode()
54			# Decode base64 byte string to a decoded byte string
55			decoded_bytes = b64decode(encoded_byte_string)
56			# Convert from byte string to a regular string
57			decoded_string = decoded_bytes.decode()
58			# Get the username and password from the string
59			(username, _, password) = decoded_string.partition(':')
60			# Check that username and password match internal global dictionary
61			return self.check_user(username, password)
62
63			def check_user(self, username, password):
64			# Check username and password in the dictionary
65			if username in self.server.user_dict:
66			from glances.password import GlancesPassword
67
68			# TODO: config is not taken into account: it may be a problem ?
69			pwd = GlancesPassword(username=username, config=None)
70			return pwd.check_password(self.server.user_dict[username], password)
71			return False
72
73			def parse_request(self):
74			if SimpleXMLRPCRequestHandler.parse_request(self):
75			# Next we authenticate
76			if self.authenticate(self.headers):
77			return True
78			# if authentication fails, tell the client
79			self.send_error(401, 'Authentication failed')
80			return False
81
82			def log_message(self, log_format, *args):
83			# No message displayed on the server side
84			pass
85
86
87			class GlancesXMLRPCServer(SimpleXMLRPCServer):
88			"""Init a SimpleXMLRPCServer instance (IPv6-ready)."""
89
90			finished = False
91
92			def __init__(self, bind_address, bind_port=61209, requestHandler=GlancesXMLRPCHandler, config=None):
93			self.bind_address = bind_address
94			self.bind_port = bind_port
95			self.config = config
96			try:
97			self.address_family = socket.getaddrinfo(bind_address, bind_port)[0][0]
98			except OSError as e:
99			logger.error(f"Couldn't open socket: {e}")
100			sys.exit(1)
101
102			super().__init__((bind_address, bind_port), requestHandler)
103
104			def end(self):
105			"""Stop the server"""
106			self.server_close()
107			self.finished = True
108
109			def serve_forever(self):
110			"""Main loop"""
111			while not self.finished:
112			self.handle_request()
113
114
115			class GlancesInstance:
116			"""All the methods of this class are published as XML-RPC methods."""
117
118			def __init__(self, config=None, args=None):
119			# Init stats
120			self.stats = GlancesStatsServer(config=config, args=args)
121
122			# Initial update
123			self.stats.update()
124
125			# cached_time is the minimum time interval between stats updates
126			# i.e. XML/RPC calls will not retrieve updated info until the time
127			# since last update is passed (will retrieve old cached info instead)
128			self.timer = Timer(0)
129			self.cached_time = args.cached_time
130
131			def __update__(self):
132			# Never update more than 1 time per cached_time
133			if self.timer.finished():
134			self.stats.update()
135			self.timer = Timer(self.cached_time)
136
137			def init(self):
138			# Return the Glances version
139			return __version__
140
141			def getAll(self):
142			# Update and return all the stats
143			self.__update__()
144			return json.dumps(self.stats.getAll())
145
146			def getAllPlugins(self):
147			# Return the plugins list
148			return json.dumps(self.stats.getPluginsList())
149
150			def getAllLimits(self):
151			# Return all the plugins limits
152			return json.dumps(self.stats.getAllLimitsAsDict())
153
154			def getAllViews(self):
155			# Return all the plugins views
156			return json.dumps(self.stats.getAllViewsAsDict())
157
158			def getPlugin(self, plugin):
159			# Update and return the plugin stat
160			self.__update__()
161			return json.dumps(self.stats.get_plugin(plugin).get_raw())
162
163			def getPluginView(self, plugin):
164			# Update and return the plugin view
165			return json.dumps(self.stats.get_plugin(plugin).get_views())
166
167
168			class GlancesServer:
169			"""This class creates and manages the TCP server."""
170
171			def __init__(self, requestHandler=GlancesXMLRPCHandler, config=None, args=None):
172			# Args
173			self.args = args
174
175			# Init the XML RPC server
176			try:
177			self.server = GlancesXMLRPCServer(args.bind_address, args.port, requestHandler, config=config)
178			except Exception as e:
179			logger.critical(f"Cannot start Glances server: {e}")
180			sys.exit(2)
181			else:
182			print(f'Glances XML-RPC server is running on {args.bind_address}:{args.port}')
183
184			# The users dict
185			# username / password couple
186			# By default, no auth is needed
187			self.server.user_dict = {}
188			self.server.isAuth = False
189
190			# Register functions
191			self.server.register_introspection_functions()
192			self.server.register_instance(GlancesInstance(config, args))
193
194			if not self.args.disable_autodiscover:
195			# Note: The Zeroconf service name will be based on the hostname
196			# Correct issue: Zeroconf problem with zeroconf service name #889
197			logger.info('Autodiscover is enabled with service name {}'.format(socket.gethostname().split('.', 1)[0]))
198			self.autodiscover_client = GlancesAutoDiscoverClient(socket.gethostname().split('.', 1)[0], args)
199			else:
200			logger.info("Glances autodiscover announce is disabled")
201
202			def add_user(self, username, password):
203			"""Add an user to the dictionary."""
204			self.server.user_dict[username] = password
205			self.server.isAuth = True
206
207			def serve_forever(self):
208			"""Call the main loop."""
209			# Set the server login/password (if -P/--password tag)
210			if self.args.password != "":
211			self.add_user(self.args.username, self.args.password)
212			# Serve forever
213			self.server.serve_forever()
214
215			def end(self):
216			"""End of the Glances server session."""
217			if not self.args.disable_autodiscover:
218			self.autodiscover_client.close()
219			self.server.end()
220

nicolargo / glances

Push — master ( b1b8ea...183265 )

glances.server.GlancesInstance.__getattr__() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

glances.server.GlancesInstance.getattr() A