kytos.core.api_server.CustomJSONEncoder.default() - Code Metrics - Inspection of "Merge pull request #253 from kytos-ng/feature/flas..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 4871c2...1a61e6 )

by Vinicius

created 2022-07-25 13:52 UTC

kytos.core.api_server.CustomJSONEncoder.default() A

↳ Parent: kytos.core.api_server

Complexity

Conditions

Size

Total Lines	4
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	3
CRAP Score	2.0625

Importance

Changes

Metric	Value
cc	2
eloc	4
nop	2
dl	0
loc	4
ccs	3
cts	4
cp	0.75
crap	2.0625
rs	10
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.json import JSONEncoder
from flask_cors import CORS
from flask_socketio import SocketIO, join_room, leave_room
from werkzeug.exceptions import HTTPException

from kytos.core.auth import authenticated
from kytos.core.config import KytosConfig


# pylint: disable=method-hidden,arguments-differ
class CustomJSONEncoder(JSONEncoder):
    """CustomJSONEncoder."""

    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.strftime("%Y-%m-%dT%H:%M:%S")
        return JSONEncoder.default(self, obj)


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.app.json_encoder = CustomJSONEncoder
        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.
        The '_shutdown' endpoint should not be public and is intended to
        shutdown the APIServer.
        """
        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 requests received to shutdown the 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 'API 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-ng/'
                       'ui/releases/latest')
                with urlopen(url) as response:
                    data = response.readlines()[0]
                    version = json.loads(data)['tag_name']
            except URLError:
                version = '1.4.1'

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

        if not os.path.exists(self.flask_dir) or force:
            # download from github
            try:
                self.log.info("Web update - Downloading UI from %s.", uri)
                package = urlretrieve(uri)[0]
            except HTTPError:
                self.log.error("Web update - Uri not found %s.", uri)
                return (f"Uri not found {uri}.",
                        HTTPStatus.INTERNAL_SERVER_ERROR.value)
            except URLError:
                self.log.error("Web update - Error accessing URL %s.", uri)
                return (f"Error accessing URL {uri}.",
                        HTTPStatus.INTERNAL_SERVER_ERROR.value)

            # test downloaded zip file
            with zipfile.ZipFile(package, 'r') as zip_ref:
                if zip_ref.testzip() is not None:
                    self.log.error("Web update - Zip file from %s "
                                   "is corrupted.", uri)
                    return (f'Zip file from {uri} is corrupted.',
                            HTTPStatus.INTERNAL_SERVER_ERROR.value)

                # backup the old web-ui files and create a new web-ui folder
                # if there is no path to backup, zip.extractall will
                # create the path.
                if os.path.exists(self.flask_dir):
                    self.log.info("Web update - Performing UI backup.")
                    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()

            self.log.info("Web update - Updated")
            return "updated the web ui", HTTPStatus.OK.value

        self.log.error("Web update - Web UI was not updated")
        return ("Web ui was not updated",
                HTTPStatus.INTERNAL_SERVER_ERROR.value)

    # 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

    @staticmethod
    def get_authenticate_options():
        """Return configuration options related to authentication."""
        options = KytosConfig().options['daemon']
        return options.authenticate_urls

    def authenticate_endpoints(self, napp):
        """Add authentication to defined REST endpoints.

        If any URL marked for authentication uses a function,
        that function will require authentication.
        """
        authenticate_urls = self.get_authenticate_options()
        for function in self._get_decorated_functions(napp):
            inner = getattr(function, '__func__', function)
            inner.authenticated = False
            for rule, _ in function.route_params:
                if inner.authenticated:
                    break
                absolute_rule = self.get_absolute_rule(rule, napp)
                for url in authenticate_urls:
                    if url in absolute_rule:
                        inner.authenticated = True
                        break

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

kytos-ng / kytos

Push — master ( 4871c2...1a61e6 )

kytos.core.api_server.CustomJSONEncoder.default() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like