kytos.core.api_server.APIServer.update_web_ui() - Code Metrics - Inspection of "Handle URLError when it's not possible to fetch th..." - kytos/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( b4d101...11876d )

by Humberto

created 2020-10-01 19:33 UTC

kytos.core.api_server.APIServer.update_web_ui() C

↳ Parent: kytos.core.api_server

Complexity

Conditions

Size

Total Lines	46
Code Lines	29

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	26
CRAP Score	9.0894

Importance

Changes

Metric	Value
cc	9
eloc	29
nop	3
dl	0
loc	46
ccs	26
cts	29
cp	0.8966
crap	9.0894
rs	6.6666
c	0
b	0
f	0

"""Module used to handle a API Server."""
import json
import logging
import os
import shutil
import sys
import warnings
import zipfile
from datetime import datetime
from glob import glob
from http import HTTPStatus
from urllib.error import HTTPError, URLError
from urllib.request import urlopen, urlretrieve

from flask import Blueprint, Flask, jsonify, request, send_file
from flask_cors import CORS
from flask_socketio import SocketIO, join_room, leave_room
from werkzeug.exceptions import HTTPException


class APIServer:
    """Api server used to provide Kytos Controller routes."""

    #: tuple: Default Flask HTTP methods.
    DEFAULT_METHODS = ('GET',)
    _NAPP_PREFIX = "/api/{napp.username}/{napp.name}/"
    _CORE_PREFIX = "/api/kytos/core/"

    # pylint: disable=too-many-arguments
    def __init__(self, app_name, listen='0.0.0.0', port=8181,
                 napps_manager=None, napps_dir=None):
        """Start a Flask+SocketIO server.

        Require controller to get NApps dir and NAppsManager

        Args:
            app_name(string): String representing a App Name
            listen (string): host name used by api server instance
            port (int): Port number used by api server instance
            controller(kytos.core.controller): A controller instance.
        """
        dirname = os.path.dirname(os.path.abspath(__file__))
        self.napps_manager = napps_manager
        self.napps_dir = napps_dir

        self.flask_dir = os.path.join(dirname, '../web-ui')
        self.log = logging.getLogger(__name__)

        self.listen = listen
        self.port = port

        self.app = Flask(app_name, root_path=self.flask_dir,
                         static_folder="dist", static_url_path="/dist")
        self.server = SocketIO(self.app, async_mode='threading')
        self._enable_websocket_rooms()
        # ENABLE CROSS ORIGIN RESOURCE SHARING
        CORS(self.app)

        # Disable trailing slash
        self.app.url_map.strict_slashes = False

        # Update web-ui if necessary
        self.update_web_ui(force=False)

        @self.app.errorhandler(HTTPException)
        def handle_exception(exception):
            # pylint: disable=unused-variable, invalid-name
            """Return a json for HTTP errors

            This handler allows NApps to return HTTP error messages
            by raising a HTTPException. When raising the Exception,
            create it with a description.
            """
            response = jsonify({
                'code': exception.code,
                'name': exception.name,
                'description': exception.description
            })
            return response, exception.code

    def _enable_websocket_rooms(self):
        socket = self.server
        socket.on_event('join', join_room)
        socket.on_event('leave', leave_room)

    def run(self):
        """Run the Flask API Server."""
        try:
            self.server.run(self.app, self.listen, self.port)
        except OSError as exception:
            msg = "Couldn't start API Server: {}".format(exception)
            self.log.critical(msg)
            sys.exit(msg)

    def register_rest_endpoint(self, url, function, methods):
        """Deprecate in favor of @rest decorator."""
        warnings.warn("From now on, use @rest decorator.", DeprecationWarning,
                      stacklevel=2)
        if url.startswith('/'):
            url = url[1:]
        self._start_endpoint(self.app, f'/kytos/{url}', function,
                             methods=methods)

    def start_api(self):
        """Start this APIServer instance API.

        Start /api/kytos/core/shutdown/ and status/ endpoints, web UI.
        """
        self.register_core_endpoint('shutdown/', self.shutdown_api)
        self.register_core_endpoint('status/', self.status_api)
        self.register_core_endpoint('web/update/<version>/',
                                    self.update_web_ui,
                                    methods=['POST'])
        self.register_core_endpoint('web/update/',
                                    self.update_web_ui,
                                    methods=['POST'])

        self.register_core_napp_services()

        self._register_web_ui()

    def register_core_endpoint(self, rule, function, **options):
        """Register an endpoint with the URL /api/kytos/core/<rule>.

        Not used by NApps, but controller.
        """
        self._start_endpoint(self.app, self._CORE_PREFIX + rule, function,
                             **options)

    def _register_web_ui(self):
        """Register routes to the admin-ui homepage."""
        self.app.add_url_rule('/', self.web_ui.__name__, self.web_ui)
        self.app.add_url_rule('/index.html', self.web_ui.__name__, self.web_ui)
        self.app.add_url_rule('/ui/<username>/<napp_name>/<path:filename>',
                              self.static_web_ui.__name__, self.static_web_ui)
        self.app.add_url_rule('/ui/<path:section_name>',
                              self.get_ui_components.__name__,
                              self.get_ui_components)

    @staticmethod
    def status_api():
        """Display kytos status using the route ``/kytos/status/``."""
        return '{"response": "running"}', HTTPStatus.OK.value

    def stop_api_server(self):
        """Send a shutdown request to stop Api Server."""
        try:
            url = f'http://127.0.0.1:{self.port}/api/kytos/core/shutdown'
            urlopen(url)
        except URLError:
            pass

    def shutdown_api(self):
        """Handle shutdown requests received by Api Server.

        This method must be called by kytos using the method
        stop_api_server, otherwise this request will be ignored.
        """
        allowed_host = ['127.0.0.1:'+str(self.port),
                        'localhost:'+str(self.port)]
        if request.host not in allowed_host:
            return "", HTTPStatus.FORBIDDEN.value

        self.server.stop()

        return 'Server shutting down...', HTTPStatus.OK.value

    def static_web_ui(self, username, napp_name, filename):
        """Serve static files from installed napps."""
        path = f"{self.napps_dir}/{username}/{napp_name}/ui/{filename}"
        if os.path.exists(path):
            return send_file(path)
        return "", HTTPStatus.NOT_FOUND.value

    def get_ui_components(self, section_name):
        """Return all napps ui components from an specific section.

        The component name generated will have the following structure:
        {username}-{nappname}-{component-section}-{filename}`

        Args:
            section_name (str): Specific section name

        Returns:
            str: Json with a list of all components found.

        """
        section_name = '*' if section_name == "all" else section_name
        path = f"{self.napps_dir}/*/*/ui/{section_name}/*.kytos"
        components = []
        for name in glob(path):
            dirs_name = name.split('/')
            dirs_name.remove('ui')

            component_name = '-'.join(dirs_name[-4:]).replace('.kytos', '')
            url = f'ui/{"/".join(dirs_name[-4:])}'
            component = {'name': component_name, 'url': url}
            components.append(component)
        return jsonify(components)

    def web_ui(self):
        """Serve the index.html page for the admin-ui."""
        index_path = f"{self.flask_dir}/index.html"
        if os.path.exists(index_path):
            return send_file(index_path)
        return f"File '{index_path}' not found.", HTTPStatus.NOT_FOUND.value

    def update_web_ui(self, version='latest', force=True):
        """Update the static files for the Web UI.

        Download the latest files from the UI github repository and update them
        in the ui folder.
        The repository link is currently hardcoded here.
        """
        if version == 'latest':
            try:
                url = 'https://api.github.com/repos/kytos/ui/releases/latest'
                response = urlopen(url)
                data = response.readlines()[0]
                version = json.loads(data)['tag_name']
            except URLError:
                version = '1.1.1'

        repository = "https://github.com/kytos/ui"
        uri = repository + f"/releases/download/{version}/latest.zip"

        if not os.path.exists(self.flask_dir) or force:
            # download from github
            try:
                package = urlretrieve(uri)[0]
            except HTTPError:
                return f"Uri not found {uri}."
            except URLError:
                self.log.warning("Error accessing URL %s.", uri)
                return f"Error accessing URL {uri}."

            # test downloaded zip file
            zip_ref = zipfile.ZipFile(package, 'r')

            if zip_ref.testzip() is not None:
                return f'Zip file from {uri} is corrupted.'

            # backup the old web-ui files and create a new web-ui folder
            if os.path.exists(self.flask_dir):
                date = datetime.now().strftime("%Y%m%d%H%M%S")
                shutil.move(self.flask_dir, f"{self.flask_dir}-{date}")
                os.mkdir(self.flask_dir)

            # unzip and extract files to web-ui/*
            zip_ref.extractall(self.flask_dir)
            zip_ref.close()

        return "updated the web ui"

    # BEGIN decorator methods

    @staticmethod
    def decorate_as_endpoint(rule, **options):
        """Decorate methods as REST endpoints.

        Example for URL ``/api/myusername/mynapp/sayhello/World``:

        .. code-block:: python3

           from flask.json import jsonify
           from kytos.core.napps import rest

           @rest('sayhello/<string:name>')
           def say_hello(name):
               return jsonify({"data": f"Hello, {name}!"})

        ``@rest`` parameters are the same as Flask's ``@app.route``. You can
        also add ``methods=['POST']``, for example.

        As we don't have the NApp instance now, we store the parameters in a
        method attribute in order to add the route later, after we have both
        APIServer and NApp instances.
        """
        def store_route_params(function):
            """Store ``Flask`` ``@route`` parameters in a method attribute.

            There can be many @route decorators in a single function.
            """
            # To support any order: @classmethod, @rest or @rest, @classmethod
            # class and static decorators return a descriptor with the function
            # in __func__.
            if isinstance(function, (classmethod, staticmethod)):
                inner = function.__func__
            else:
                inner = function
            # Add route parameters
            if not hasattr(inner, 'route_params'):
                inner.route_params = []
            inner.route_params.append((rule, options))
            # Return the same function, now with "route_params" attribute
            return function
        return store_route_params

    def register_napp_endpoints(self, napp):
        """Add all NApp REST endpoints with @rest decorator.

        We are using Flask Blueprints to register these endpoints. Blueprints
        are essentially the Flask equivalent of Python modules and are used to
        keep related logic and assets grouped and separated from one another.

        URLs will be prefixed with ``/api/{username}/{napp_name}/``.

        Args:
            napp (Napp): Napp instance to register new endpoints.
        """
        # Create a Flask Blueprint for a specific NApp
        napp_blueprint = Blueprint(napp.napp_id, __name__)

        # Start all endpoints for this NApp
        for function in self._get_decorated_functions(napp):
            for rule, options in function.route_params:
                absolute_rule = self.get_absolute_rule(rule, napp)
                self._start_endpoint(napp_blueprint, absolute_rule, function,
                                     **options)

        # Register this Flask Blueprint in the Flask App
        self.app.register_blueprint(napp_blueprint)

    @staticmethod
    def _get_decorated_functions(napp):
        """Return ``napp``'s methods having the @rest decorator."""
        for name in dir(napp):
            if not name.startswith('_'):  # discarding private names
                pub_attr = getattr(napp, name)
                if callable(pub_attr) and hasattr(pub_attr, 'route_params'):
                    yield pub_attr

    @classmethod
    def get_absolute_rule(cls, rule, napp):
        """Prefix the rule, e.g. "flow" to "/api/user/napp/flow".

        This code is used by kytos-utils when generating an OpenAPI skel.
        """
        # Flask does require 2 slashes if specified, so we remove a starting
        # slash if applicable.
        relative_rule = rule[1:] if rule.startswith('/') else rule
        return cls._NAPP_PREFIX.format(napp=napp) + relative_rule

    # END decorator methods

    def _start_endpoint(self, app, rule, function, **options):
        """Start ``function``'s endpoint.

        Forward parameters to ``Flask.add_url_rule`` mimicking Flask
        ``@route`` decorator.
        """
        endpoint = options.pop('endpoint', None)
        app.add_url_rule(rule, endpoint, function, **options)
        self.log.info('Started %s - %s', rule,
                      ', '.join(options.get('methods', self.DEFAULT_METHODS)))

    def remove_napp_endpoints(self, napp):
        """Remove all decorated endpoints.

        Args:
            napp (Napp): Napp instance to look for rest-decorated methods.
        """
        prefix = self._NAPP_PREFIX.format(napp=napp)
        indexes = []
        endpoints = set()
        for index, rule in enumerate(self.app.url_map.iter_rules()):
            if rule.rule.startswith(prefix):
                endpoints.add(rule.endpoint)
                indexes.append(index)
                self.log.info('Stopped %s - %s', rule, ','.join(rule.methods))

        for endpoint in endpoints:
            self.app.view_functions.pop(endpoint)

        for index in reversed(indexes):
            # pylint: disable=protected-access
            self.app.url_map._rules.pop(index)
            # pylint: enable=protected-access

        # Remove the Flask Blueprint of this NApp from the Flask App
        self.app.blueprints.pop(napp.napp_id)

        self.log.info('The Rest endpoints from %s were disabled.', prefix)

    def register_core_napp_services(self):
        """
        Register /kytos/core/ services over NApps.

        It registers enable, disable, install, uninstall NApps that will
        be used by kytos-utils.
        """
        self.register_core_endpoint("napps/<username>/<napp_name>/enable",
                                    self._enable_napp)
        self.register_core_endpoint("napps/<username>/<napp_name>/disable",
                                    self._disable_napp)
        self.register_core_endpoint("napps/<username>/<napp_name>/install",
                                    self._install_napp)
        self.register_core_endpoint("napps/<username>/<napp_name>/uninstall",
                                    self._uninstall_napp)
        self.register_core_endpoint("napps_enabled",
                                    self._list_enabled_napps)
        self.register_core_endpoint("napps_installed",
                                    self._list_installed_napps)
        self.register_core_endpoint(
            "napps/<username>/<napp_name>/metadata/<key>",
            self._get_napp_metadata)

    def _enable_napp(self, username, napp_name):
        """
        Enable an installed NApp.

        :param username: NApps user name
        :param napp_name: NApp name
        :return: JSON content and return code
        """
        # Check if the NApp is installed
        if not self.napps_manager.is_installed(username, napp_name):
            return '{"response": "not installed"}', \
                   HTTPStatus.BAD_REQUEST.value

        # Check if the NApp is already been enabled
        if not self.napps_manager.is_enabled(username, napp_name):
            self.napps_manager.enable(username, napp_name)

        # Check if NApp is enabled
        if not self.napps_manager.is_enabled(username, napp_name):
            # If it is not enabled an admin user must check the log file
            return '{"response": "error"}', \
                   HTTPStatus.INTERNAL_SERVER_ERROR.value

        return '{"response": "enabled"}', HTTPStatus.OK.value

    def _disable_napp(self, username, napp_name):
        """
        Disable an installed NApp.

        :param username: NApps user name
        :param napp_name: NApp name
        :return: JSON content and return code
        """
        # Check if the NApp is installed
        if not self.napps_manager.is_installed(username, napp_name):
            return '{"response": "not installed"}', \
                   HTTPStatus.BAD_REQUEST.value

        # Check if the NApp is enabled
        if self.napps_manager.is_enabled(username, napp_name):
            self.napps_manager.disable(username, napp_name)

        # Check if NApp is still enabled
        if self.napps_manager.is_enabled(username, napp_name):
            # If it is still enabled an admin user must check the log file
            return '{"response": "error"}', \
                   HTTPStatus.INTERNAL_SERVER_ERROR.value

        return '{"response": "disabled"}', \
               HTTPStatus.OK.value

    def _install_napp(self, username, napp_name):
        # Check if the NApp is installed
        if self.napps_manager.is_installed(username, napp_name):
            return '{"response": "installed"}', HTTPStatus.OK.value

        napp = "{}/{}".format(username, napp_name)

        # Try to install and enable the napp
        try:
            if not self.napps_manager.install(napp, enable=True):
                # If it is not installed an admin user must check the log file
                return '{"response": "error"}', \
                       HTTPStatus.INTERNAL_SERVER_ERROR.value
        except HTTPError as exception:
            return '{"response": "error"}', exception.code

        return '{"response": "installed"}', HTTPStatus.OK.value

    def _uninstall_napp(self, username, napp_name):
        # Check if the NApp is installed
        if self.napps_manager.is_installed(username, napp_name):
            # Try to unload/uninstall the napp
            if not self.napps_manager.uninstall(username, napp_name):
                # If it is not uninstalled admin user must check the log file
                return '{"response": "error"}', \
                       HTTPStatus.INTERNAL_SERVER_ERROR.value

        return '{"response": "uninstalled"}', HTTPStatus.OK.value

    def _list_enabled_napps(self):
        """Sorted list of (username, napp_name) of enabled napps."""
        serialized_dict = json.dumps(
                            self.napps_manager.get_enabled_napps(),
                            default=lambda a: [a.username, a.name])

        return '{"napps": %s}' % serialized_dict, HTTPStatus.OK.value

    def _list_installed_napps(self):
        """Sorted list of (username, napp_name) of installed napps."""
        serialized_dict = json.dumps(
                            self.napps_manager.get_installed_napps(),
                            default=lambda a: [a.username, a.name])

        return '{"napps": %s}' % serialized_dict, HTTPStatus.OK.value

    def _get_napp_metadata(self, username, napp_name, key):
        """Get NApp metadata value.

        For safety reasons, only some keys can be retrieved:
        napp_dependencies, description, version.

        """
        valid_keys = ['napp_dependencies', 'description', 'version']

        if not self.napps_manager.is_installed(username, napp_name):
            return "NApp is not installed.", HTTPStatus.BAD_REQUEST.value

        if key not in valid_keys:
            return "Invalid key.", HTTPStatus.BAD_REQUEST.value

        data = self.napps_manager.get_napp_metadata(username, napp_name, key)
        serialized_dict = json.dumps({key: data})

        return '%s' % serialized_dict, HTTPStatus.OK.value


1		"""Module used to handle a API Server."""
2	1	import json
3	1	import logging
4	1	import os
5	1	import shutil
6	1	import sys
7	1	import warnings
8	1	import zipfile
9	1	from datetime import datetime
10	1	from glob import glob
11	1	from http import HTTPStatus
12	1	from urllib.error import HTTPError, URLError
13	1	from urllib.request import urlopen, urlretrieve
14
15	1	from flask import Blueprint, Flask, jsonify, request, send_file
16	1	from flask_cors import CORS
17	1	from flask_socketio import SocketIO, join_room, leave_room
18	1	from werkzeug.exceptions import HTTPException
19
20
21	1	class APIServer:
22		"""Api server used to provide Kytos Controller routes."""
23
24		#: tuple: Default Flask HTTP methods.
25	1	DEFAULT_METHODS = ('GET',)
26	1	_NAPP_PREFIX = "/api/{napp.username}/{napp.name}/"
27	1	_CORE_PREFIX = "/api/kytos/core/"
28
29		# pylint: disable=too-many-arguments
30	1	def __init__(self, app_name, listen='0.0.0.0', port=8181,
31		napps_manager=None, napps_dir=None):
32		"""Start a Flask+SocketIO server.
33
34		Require controller to get NApps dir and NAppsManager
35
36		Args:
37		app_name(string): String representing a App Name
38		listen (string): host name used by api server instance
39		port (int): Port number used by api server instance
40		controller(kytos.core.controller): A controller instance.
41		"""
42	1	dirname = os.path.dirname(os.path.abspath(__file__))
43	1	self.napps_manager = napps_manager
44	1	self.napps_dir = napps_dir
45
46	1	self.flask_dir = os.path.join(dirname, '../web-ui')
47	1	self.log = logging.getLogger(__name__)
48
49	1	self.listen = listen
50	1	self.port = port
51
52	1	self.app = Flask(app_name, root_path=self.flask_dir,
53		static_folder="dist", static_url_path="/dist")
54	1	self.server = SocketIO(self.app, async_mode='threading')
55	1	self._enable_websocket_rooms()
56		# ENABLE CROSS ORIGIN RESOURCE SHARING
57	1	CORS(self.app)
58
59		# Disable trailing slash
60	1	self.app.url_map.strict_slashes = False
61
62		# Update web-ui if necessary
63	1	self.update_web_ui(force=False)
64
65	1	@self.app.errorhandler(HTTPException)
66		def handle_exception(exception):
67		# pylint: disable=unused-variable, invalid-name
68		"""Return a json for HTTP errors
69
70		This handler allows NApps to return HTTP error messages
71		by raising a HTTPException. When raising the Exception,
72		create it with a description.
73		"""
74		response = jsonify({
75		'code': exception.code,
76		'name': exception.name,
77		'description': exception.description
78		})
79		return response, exception.code
80
81	1	def _enable_websocket_rooms(self):
82	1	socket = self.server
83	1	socket.on_event('join', join_room)
84	1	socket.on_event('leave', leave_room)
85
86	1	def run(self):
87		"""Run the Flask API Server."""
88	1	try:
89	1	self.server.run(self.app, self.listen, self.port)
90	1	except OSError as exception:
91	1	msg = "Couldn't start API Server: {}".format(exception)
92	1	self.log.critical(msg)
93	1	sys.exit(msg)
94
95	1	def register_rest_endpoint(self, url, function, methods):
96		"""Deprecate in favor of @rest decorator."""
97	1	warnings.warn("From now on, use @rest decorator.", DeprecationWarning,
98		stacklevel=2)
99	1	if url.startswith('/'):
100		url = url[1:]
101	1	self._start_endpoint(self.app, f'/kytos/{url}', function,
102		methods=methods)
103
104	1	def start_api(self):
105		"""Start this APIServer instance API.
106
107		Start /api/kytos/core/shutdown/ and status/ endpoints, web UI.
108		"""
109	1	self.register_core_endpoint('shutdown/', self.shutdown_api)
110	1	self.register_core_endpoint('status/', self.status_api)
111	1	self.register_core_endpoint('web/update/<version>/',
112		self.update_web_ui,
113		methods=['POST'])
114	1	self.register_core_endpoint('web/update/',
115		self.update_web_ui,
116		methods=['POST'])
117
118	1	self.register_core_napp_services()
119
120	1	self._register_web_ui()
121
122	1	def register_core_endpoint(self, rule, function, **options):
123		"""Register an endpoint with the URL /api/kytos/core/<rule>.
124
125		Not used by NApps, but controller.
126		"""
127	1	self._start_endpoint(self.app, self._CORE_PREFIX + rule, function,
128		**options)
129
130	1	def _register_web_ui(self):
131		"""Register routes to the admin-ui homepage."""
132	1	self.app.add_url_rule('/', self.web_ui.__name__, self.web_ui)
133	1	self.app.add_url_rule('/index.html', self.web_ui.__name__, self.web_ui)
134	1	self.app.add_url_rule('/ui/<username>/<napp_name>/<path:filename>',
135		self.static_web_ui.__name__, self.static_web_ui)
136	1	self.app.add_url_rule('/ui/<path:section_name>',
137		self.get_ui_components.__name__,
138		self.get_ui_components)
139
140	1	@staticmethod
141		def status_api():
142		"""Display kytos status using the route ``/kytos/status/``."""
143	1	return '{"response": "running"}', HTTPStatus.OK.value
144
145	1	def stop_api_server(self):
146		"""Send a shutdown request to stop Api Server."""
147	1	try:
148	1	url = f'http://127.0.0.1:{self.port}/api/kytos/core/shutdown'
149	1	urlopen(url)
150		except URLError:
151		pass
152
153	1	def shutdown_api(self):
154		"""Handle shutdown requests received by Api Server.
155
156		This method must be called by kytos using the method
157		stop_api_server, otherwise this request will be ignored.
158		"""
159	1	allowed_host = ['127.0.0.1:'+str(self.port),
160		'localhost:'+str(self.port)]
161	1	if request.host not in allowed_host:
162	1	return "", HTTPStatus.FORBIDDEN.value
163
164	1	self.server.stop()
165
166	1	return 'Server shutting down...', HTTPStatus.OK.value
167
168	1	def static_web_ui(self, username, napp_name, filename):
169		"""Serve static files from installed napps."""
170	1	path = f"{self.napps_dir}/{username}/{napp_name}/ui/{filename}"
171	1	if os.path.exists(path):
172	1	return send_file(path)
173	1	return "", HTTPStatus.NOT_FOUND.value
174
175	1	def get_ui_components(self, section_name):
176		"""Return all napps ui components from an specific section.
177
178		The component name generated will have the following structure:
179		{username}-{nappname}-{component-section}-{filename}`
180
181		Args:
182		section_name (str): Specific section name
183
184		Returns:
185		str: Json with a list of all components found.
186
187		"""
188	1	section_name = '*' if section_name == "all" else section_name
189	1	path = f"{self.napps_dir}///ui/{section_name}/*.kytos"
190	1	components = []
191	1	for name in glob(path):
192	1	dirs_name = name.split('/')
193	1	dirs_name.remove('ui')
194
195	1	component_name = '-'.join(dirs_name[-4:]).replace('.kytos', '')
196	1	url = f'ui/{"/".join(dirs_name[-4:])}'
197	1	component = {'name': component_name, 'url': url}
198	1	components.append(component)
199	1	return jsonify(components)
200
201	1	def web_ui(self):
202		"""Serve the index.html page for the admin-ui."""
203	1	index_path = f"{self.flask_dir}/index.html"
204	1	if os.path.exists(index_path):
205	1	return send_file(index_path)
206	1	return f"File '{index_path}' not found.", HTTPStatus.NOT_FOUND.value
207
208	1	def update_web_ui(self, version='latest', force=True):
209		"""Update the static files for the Web UI.
210
211		Download the latest files from the UI github repository and update them
212		in the ui folder.
213		The repository link is currently hardcoded here.
214		"""
215	1	if version == 'latest':
216	1	try:
217	1	url = 'https://api.github.com/repos/kytos/ui/releases/latest'
218	1	response = urlopen(url)
219	1	data = response.readlines()[0]
220	1	version = json.loads(data)['tag_name']
221	1	except URLError:
222	1	version = '1.1.1'
223
224	1	repository = "https://github.com/kytos/ui"
225	1	uri = repository + f"/releases/download/{version}/latest.zip"
226
227	1	if not os.path.exists(self.flask_dir) or force:
228		# download from github
229	1	try:
230	1	package = urlretrieve(uri)[0]
231	1	except HTTPError:
232	1	return f"Uri not found {uri}."
233		except URLError:
234		self.log.warning("Error accessing URL %s.", uri)
235		return f"Error accessing URL {uri}."
236
237		# test downloaded zip file
238	1	zip_ref = zipfile.ZipFile(package, 'r')
239
240	1	if zip_ref.testzip() is not None:
241	1	return f'Zip file from {uri} is corrupted.'
242
243		# backup the old web-ui files and create a new web-ui folder
244	1	if os.path.exists(self.flask_dir):
245	1	date = datetime.now().strftime("%Y%m%d%H%M%S")
246	1	shutil.move(self.flask_dir, f"{self.flask_dir}-{date}")
247	1	os.mkdir(self.flask_dir)
248
249		# unzip and extract files to web-ui/*
250	1	zip_ref.extractall(self.flask_dir)
251	1	zip_ref.close()
252
253	1	return "updated the web ui"
254
255		# BEGIN decorator methods
256
257	1	@staticmethod
258		def decorate_as_endpoint(rule, **options):
259		"""Decorate methods as REST endpoints.
260
261		Example for URL ``/api/myusername/mynapp/sayhello/World``:
262
263		.. code-block:: python3
264
265		from flask.json import jsonify
266		from kytos.core.napps import rest
267
268		@rest('sayhello/<string:name>')
269		def say_hello(name):
270		return jsonify({"data": f"Hello, {name}!"})
271
272		``@rest`` parameters are the same as Flask's ``@app.route``. You can
273		also add ``methods=['POST']``, for example.
274
275		As we don't have the NApp instance now, we store the parameters in a
276		method attribute in order to add the route later, after we have both
277		APIServer and NApp instances.
278		"""
279	1	def store_route_params(function):
280		"""Store ``Flask`` ``@route`` parameters in a method attribute.
281
282		There can be many @route decorators in a single function.
283		"""
284		# To support any order: @classmethod, @rest or @rest, @classmethod
285		# class and static decorators return a descriptor with the function
286		# in __func__.
287	1	if isinstance(function, (classmethod, staticmethod)):
288	1	inner = function.__func__
289		else:
290	1	inner = function
291		# Add route parameters
292	1	if not hasattr(inner, 'route_params'):
293	1	inner.route_params = []
294	1	inner.route_params.append((rule, options))
295		# Return the same function, now with "route_params" attribute
296	1	return function
297	1	return store_route_params
298
299	1	def register_napp_endpoints(self, napp):
300		"""Add all NApp REST endpoints with @rest decorator.
301
302		We are using Flask Blueprints to register these endpoints. Blueprints
303		are essentially the Flask equivalent of Python modules and are used to
304		keep related logic and assets grouped and separated from one another.
305
306		URLs will be prefixed with ``/api/{username}/{napp_name}/``.
307
308		Args:
309		napp (Napp): Napp instance to register new endpoints.
310		"""
311		# Create a Flask Blueprint for a specific NApp
312	1	napp_blueprint = Blueprint(napp.napp_id, __name__)
313
314		# Start all endpoints for this NApp
315	1	for function in self._get_decorated_functions(napp):
316	1	for rule, options in function.route_params:
317	1	absolute_rule = self.get_absolute_rule(rule, napp)
318	1	self._start_endpoint(napp_blueprint, absolute_rule, function,
319		**options)
320
321		# Register this Flask Blueprint in the Flask App
322	1	self.app.register_blueprint(napp_blueprint)
323
324	1	@staticmethod
325		def _get_decorated_functions(napp):
326		"""Return ``napp``'s methods having the @rest decorator."""
327	1	for name in dir(napp):
328	1	if not name.startswith('_'): # discarding private names
329	1	pub_attr = getattr(napp, name)
330	1	if callable(pub_attr) and hasattr(pub_attr, 'route_params'):
331	1	yield pub_attr
332
333	1	@classmethod
334		def get_absolute_rule(cls, rule, napp):
335		"""Prefix the rule, e.g. "flow" to "/api/user/napp/flow".
336
337		This code is used by kytos-utils when generating an OpenAPI skel.
338		"""
339		# Flask does require 2 slashes if specified, so we remove a starting
340		# slash if applicable.
341	1	relative_rule = rule[1:] if rule.startswith('/') else rule
342	1	return cls._NAPP_PREFIX.format(napp=napp) + relative_rule
343
344		# END decorator methods
345
346	1	def _start_endpoint(self, app, rule, function, **options):
347		"""Start ``function``'s endpoint.
348
349		Forward parameters to ``Flask.add_url_rule`` mimicking Flask
350		``@route`` decorator.
351		"""
352	1	endpoint = options.pop('endpoint', None)
353	1	app.add_url_rule(rule, endpoint, function, **options)
354	1	self.log.info('Started %s - %s', rule,
355		', '.join(options.get('methods', self.DEFAULT_METHODS)))
356
357	1	def remove_napp_endpoints(self, napp):
358		"""Remove all decorated endpoints.
359
360		Args:
361		napp (Napp): Napp instance to look for rest-decorated methods.
362		"""
363	1	prefix = self._NAPP_PREFIX.format(napp=napp)
364	1	indexes = []
365	1	endpoints = set()
366	1	for index, rule in enumerate(self.app.url_map.iter_rules()):
367	1	if rule.rule.startswith(prefix):
368	1	endpoints.add(rule.endpoint)
369	1	indexes.append(index)
370	1	self.log.info('Stopped %s - %s', rule, ','.join(rule.methods))
371
372	1	for endpoint in endpoints:
373	1	self.app.view_functions.pop(endpoint)
374
375	1	for index in reversed(indexes):
376		# pylint: disable=protected-access
377	1	self.app.url_map._rules.pop(index)
378		# pylint: enable=protected-access
379
380		# Remove the Flask Blueprint of this NApp from the Flask App
381	1	self.app.blueprints.pop(napp.napp_id)
382
383	1	self.log.info('The Rest endpoints from %s were disabled.', prefix)
384
385	1	def register_core_napp_services(self):
386		"""
387		Register /kytos/core/ services over NApps.
388
389		It registers enable, disable, install, uninstall NApps that will
390		be used by kytos-utils.
391		"""
392	1	self.register_core_endpoint("napps/<username>/<napp_name>/enable",
393		self._enable_napp)
394	1	self.register_core_endpoint("napps/<username>/<napp_name>/disable",
395		self._disable_napp)
396	1	self.register_core_endpoint("napps/<username>/<napp_name>/install",
397		self._install_napp)
398	1	self.register_core_endpoint("napps/<username>/<napp_name>/uninstall",
399		self._uninstall_napp)
400	1	self.register_core_endpoint("napps_enabled",
401		self._list_enabled_napps)
402	1	self.register_core_endpoint("napps_installed",
403		self._list_installed_napps)
404	1	self.register_core_endpoint(
405		"napps/<username>/<napp_name>/metadata/<key>",
406		self._get_napp_metadata)
407
408	1	def _enable_napp(self, username, napp_name):
409		"""
410		Enable an installed NApp.
411
412		:param username: NApps user name
413		:param napp_name: NApp name
414		:return: JSON content and return code
415		"""
416		# Check if the NApp is installed
417	1	if not self.napps_manager.is_installed(username, napp_name):
418	1	return '{"response": "not installed"}', \
419		HTTPStatus.BAD_REQUEST.value
420
421		# Check if the NApp is already been enabled
422	1	if not self.napps_manager.is_enabled(username, napp_name):
423	1	self.napps_manager.enable(username, napp_name)
424
425		# Check if NApp is enabled
426	1	if not self.napps_manager.is_enabled(username, napp_name):
427		# If it is not enabled an admin user must check the log file
428	1	return '{"response": "error"}', \
429		HTTPStatus.INTERNAL_SERVER_ERROR.value
430
431	1	return '{"response": "enabled"}', HTTPStatus.OK.value
432
433	1	def _disable_napp(self, username, napp_name):
434		"""
435		Disable an installed NApp.
436
437		:param username: NApps user name
438		:param napp_name: NApp name
439		:return: JSON content and return code
440		"""
441		# Check if the NApp is installed
442	1	if not self.napps_manager.is_installed(username, napp_name):
443	1	return '{"response": "not installed"}', \
444		HTTPStatus.BAD_REQUEST.value
445
446		# Check if the NApp is enabled
447	1	if self.napps_manager.is_enabled(username, napp_name):
448	1	self.napps_manager.disable(username, napp_name)
449
450		# Check if NApp is still enabled
451	1	if self.napps_manager.is_enabled(username, napp_name):
452		# If it is still enabled an admin user must check the log file
453	1	return '{"response": "error"}', \
454		HTTPStatus.INTERNAL_SERVER_ERROR.value
455
456	1	return '{"response": "disabled"}', \
457		HTTPStatus.OK.value
458
459	1	def _install_napp(self, username, napp_name):
460		# Check if the NApp is installed
461	1	if self.napps_manager.is_installed(username, napp_name):
462	1	return '{"response": "installed"}', HTTPStatus.OK.value
463
464	1	napp = "{}/{}".format(username, napp_name)
465
466		# Try to install and enable the napp
467	1	try:
468	1	if not self.napps_manager.install(napp, enable=True):
469		# If it is not installed an admin user must check the log file
470	1	return '{"response": "error"}', \
471		HTTPStatus.INTERNAL_SERVER_ERROR.value
472	1	except HTTPError as exception:
473	1	return '{"response": "error"}', exception.code
474
475	1	return '{"response": "installed"}', HTTPStatus.OK.value
476
477	1	def _uninstall_napp(self, username, napp_name):
478		# Check if the NApp is installed
479	1	if self.napps_manager.is_installed(username, napp_name):
480		# Try to unload/uninstall the napp
481	1	if not self.napps_manager.uninstall(username, napp_name):
482		# If it is not uninstalled admin user must check the log file
483	1	return '{"response": "error"}', \
484		HTTPStatus.INTERNAL_SERVER_ERROR.value
485
486	1	return '{"response": "uninstalled"}', HTTPStatus.OK.value
487
488	1	def _list_enabled_napps(self):
489		"""Sorted list of (username, napp_name) of enabled napps."""
490	1	serialized_dict = json.dumps(
491		self.napps_manager.get_enabled_napps(),
492		default=lambda a: [a.username, a.name])
493
494	1	return '{"napps": %s}' % serialized_dict, HTTPStatus.OK.value
495
496	1	def _list_installed_napps(self):
497		"""Sorted list of (username, napp_name) of installed napps."""
498	1	serialized_dict = json.dumps(
499		self.napps_manager.get_installed_napps(),
500		default=lambda a: [a.username, a.name])
501
502	1	return '{"napps": %s}' % serialized_dict, HTTPStatus.OK.value
503
504	1	def _get_napp_metadata(self, username, napp_name, key):
505		"""Get NApp metadata value.
506
507		For safety reasons, only some keys can be retrieved:
508		napp_dependencies, description, version.
509
510		"""
511	1	valid_keys = ['napp_dependencies', 'description', 'version']
512
513	1	if not self.napps_manager.is_installed(username, napp_name):
514	1	return "NApp is not installed.", HTTPStatus.BAD_REQUEST.value
515
516	1	if key not in valid_keys:
517	1	return "Invalid key.", HTTPStatus.BAD_REQUEST.value
518
519	1	data = self.napps_manager.get_napp_metadata(username, napp_name, key)
520	1	serialized_dict = json.dumps({key: data})
521
522		return '%s' % serialized_dict, HTTPStatus.OK.value
523

kytos / kytos

Push — master ( b4d101...11876d )

kytos.core.api_server.APIServer.update_web_ui() C

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like