kytos.core.api_server.APIServer.update_web_ui() - Code Metrics - Inspection of "feat: replaced `werkzeug/flask` with `uvicorn/star..." - kytos-ng/kytos - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — master (#375)

by Vinicius

created 2023-04-25 19:30 UTC

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

↳ Parent: kytos.core.api_server

Complexity

Conditions

Size

Total Lines	44
Code Lines	33

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	8
eloc	33
nop	4
dl	0
loc	44
rs	7.2213
c	0
b	0
f	0

"""Module used to handle a API Server."""
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 typing import Optional, Union
from urllib.error import HTTPError, URLError
from urllib.request import urlretrieve

import httpx
from starlette.applications import Starlette
from starlette.exceptions import HTTPException
from starlette.middleware import Middleware
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import FileResponse
from starlette.staticfiles import StaticFiles
from uvicorn import Config as UvicornConfig
from uvicorn import Server

from kytos.core.auth import authenticated
from kytos.core.config import KytosConfig
from kytos.core.rest_api import JSONResponse, Request

LOG = logging.getLogger(__name__)


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, 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:
            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.listen = listen
        self.port = port
        self.web_ui_dir = os.path.join(dirname, '../web-ui')
        self.app = Starlette(
            exception_handlers={HTTPException: self._http_exc_handler},
            middleware=[
                Middleware(CORSMiddleware, allow_origins=["*"]),
            ],
        )
        self.server = Server(
            UvicornConfig(
                self.app,
                host=self.listen,
                port=self.port,
            )
        )

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

    async def _http_exc_handler(self, _request: Request, exc: HTTPException):
        """HTTPException handler.

        The response format is still partly compatible with how werkzeug used
        to format http exceptions.
        """
        return JSONResponse({"description": exc.detail,
                             "code": exc.status_code},
                            status_code=exc.status_code)

    def run(self):
        """Run API Server."""
        try:
            self._run_uvicorn()
        except (RuntimeError, ValueError, OSError) as exception:
            msg = f"Couldn't start API Server: {exception}"
            LOG.critical(msg)
            sys.exit(msg)

    def _run_uvicorn(self) -> None:
        self.server.run()

    def _stop_uvicorn(self) -> None:
        self.server.should_exit = True

    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."""
        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()

    def start_web_ui(self) -> None:
        """Start Web UI endpoints."""
        self._start_endpoint(self.app,
                             "/ui/{username}/{napp_name}/{filename:path}",
                             self.static_web_ui)
        self._start_endpoint(self.app,
                             "/ui/{section_name}/",
                             self.get_ui_components)
        self._start_endpoint(self.app, '/', self.web_ui)
        self._start_endpoint(self.app, '/index.html', self.web_ui)
        self.start_web_ui_static_files()

    def start_web_ui_static_files(self) -> None:
        """Start Web UI static files."""
        self.app.router.mount("/", app=StaticFiles(directory=self.web_ui_dir),
                              name="dist")

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

        Not used by NApps, but controller.
        """
        self._start_endpoint(self.app, f"{self._CORE_PREFIX}{route}",
                             function, **options)

    def status_api(self, _request: Request):
        """Display kytos status using the route ``/kytos/status/``."""
        return JSONResponse({"response": "running"})

    def stop_api_server(self):
        """Stop API server."""
        self._stop_uvicorn()

    def static_web_ui(self,
                      request: Request) -> Union[FileResponse, JSONResponse]:
        """Serve static files from installed napps."""
        username = request.path_params["username"]
        napp_name = request.path_params["napp_name"]
        filename = request.path_params["filename"]
        path = f"{self.napps_dir}/{username}/{napp_name}/ui/{filename}"
        if os.path.exists(path):
            return FileResponse(path)
        return JSONResponse("", status_code=HTTPStatus.NOT_FOUND.value)

    def get_ui_components(self, request: Request) -> JSONResponse:
        """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 = request.path_params["section_name"]
        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 JSONResponse(components)

    def web_ui(self, _request: Request) -> Union[JSONResponse, FileResponse]:
        """Serve the index.html page for the admin-ui."""
        index_path = f"{self.web_ui_dir}/index.html"
        if os.path.exists(index_path):
            return FileResponse(index_path)
        return JSONResponse(f"File '{index_path}' not found.",
                            status_code=HTTPStatus.NOT_FOUND.value)

    def _unzip_backup_web_ui(self, package: str, uri: str) -> None:
        """Unzip and backup web ui files.

        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.
        """
        with zipfile.ZipFile(package, 'r') as zip_ref:
            if zip_ref.testzip() is not None:
                LOG.error("Web update - Zip file from %s "
                          "is corrupted.", uri)
                raise ValueError(f'Zip file from {uri} is corrupted.')
            if os.path.exists(self.web_ui_dir):
                LOG.info("Web update - Performing UI backup.")
                date = datetime.now().strftime("%Y%m%d%H%M%S")
                shutil.move(self.web_ui_dir, f"{self.web_ui_dir}-{date}")
                os.mkdir(self.web_ui_dir)
            # unzip and extract files to web-ui/*
            zip_ref.extractall(self.web_ui_dir)
            zip_ref.close()

    def _fetch_latest_ui_tag(self) -> str:
        """Fetch latest ui tag version from GitHub."""
        version = '2022.3.0'
        try:
            url = ('https://api.github.com/repos/kytos-ng/'
                   'ui/releases/latest')
            response = httpx.get(url, timeout=10)
            version = response.json()['tag_name']
        except (httpx.RequestError, KeyError):
            msg = "Failed to fetch latest tag from GitHub, " \
                  f"falling back to {version}"
            LOG.warning(f"Web update - {msg}")
        return version

    def update_web_ui(self, request: Optional[Request],
                      version='latest',
                      force=True) -> JSONResponse:
        """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 not os.path.exists(self.web_ui_dir) or force:
            if request:
                version = request.path_params.get("version", "latest")
            if version == 'latest':
                version = self._fetch_latest_ui_tag()

            repository = "https://github.com/kytos-ng/ui"
            uri = repository + f"/releases/download/{version}/latest.zip"
            try:
                LOG.info("Web update - Downloading UI from %s.", uri)
                package = urlretrieve(uri)[0]
                self._unzip_backup_web_ui(package, uri)
            except HTTPError:
                LOG.error("Web update - Uri not found %s.", uri)
                return JSONResponse(
                        f"Uri not found {uri}.",
                        status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value
                        )
            except URLError:
                LOG.error("Web update - Error accessing URL %s.", uri)
                return JSONResponse(
                        f"Error accessing URL {uri}.",
                        status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value
                       )
            except ValueError as exc:
                return JSONResponse(
                        str(exc),
                        status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value
                       )
            LOG.info("Web update - Updated")
            return JSONResponse("Web UI was updated")

        LOG.error("Web update - Web UI was not updated")
        return JSONResponse("Web ui was not updated",
                            status_code=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 kytos.core.napps import rest
           from kytos.core.rest_api import JSONResponse, Request

           @rest("sayhello/{name}")
           def say_hello(request: Request) -> JSONResponse:
               name = request.path_params["name"]
               return JSONResponse({"data": f"Hello, {name}!"})

        ``@rest`` parameters are the same as Starlette's ``add_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))
            inner.created_at = datetime.utcnow()
            # 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.

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

        Args:
            napp (Napp): Napp instance to register new endpoints.
        """
        # 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(self.app, absolute_rule,
                                     function, **options)

    @staticmethod
    def _get_decorated_functions(napp):
        """Return ``napp``'s methods having the @rest decorator.

        The callables are yielded based on their decorated order (created_at),
        this ensures deterministic routing matching order.
        """
        callables = []
        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'):
                    callables.append(pub_attr)
        try:
            callables = sorted(callables, key=lambda f: f.created_at)
        except TypeError:
            pass

        for pub_attr in callables:
            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 _add_non_strict_slash_route(
        self,
        app: Starlette,
        route: str,
        function,
        **options
    ) -> None:
        """Try to add a non strict slash route."""
        if route == "/":
            return
        non_strict = route[:-1] if route.endswith("/") else f"{route}/"
        app.router.add_route(non_strict, function, **options,
                             include_in_schema=False)

    def _start_endpoint(
        self,
        app: Starlette,
        route: str,
        function,
        **options
    ):
        """Start ``function``'s endpoint."""
        app.router.add_route(route, function, **options)
        self._add_non_strict_slash_route(app, route, function, **options)
        LOG.info('Started %s - %s', route,
                 ', '.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 = []
        for index, route in enumerate(self.app.routes):
            if route.path.startswith(prefix):
                indexes.append(index)
        for index in reversed(indexes):
            self.app.routes.pop(index)
        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, request: Request) -> JSONResponse:

        """Enable an installed NApp."""

        username = request.path_params["username"]
        napp_name = request.path_params["napp_name"]
        # Check if the NApp is installed
        if not self.napps_manager.is_installed(username, napp_name):
            return JSONResponse({"response": "not installed"},
                                status_code=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
            status_code = HTTPStatus.INTERNAL_SERVER_ERROR.value
            return JSONResponse({"response": "error"},
                                status_code=status_code)

        return JSONResponse({"response": "enabled"})

    def _disable_napp(self, request: Request) -> JSONResponse:

        """Disable an installed NApp."""

        username = request.path_params["username"]
        napp_name = request.path_params["napp_name"]
        # Check if the NApp is installed
        if not self.napps_manager.is_installed(username, napp_name):
            return JSONResponse({"response": "not installed"},
                                status_code=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
            status_code = HTTPStatus.INTERNAL_SERVER_ERROR.value
            return JSONResponse({"response": "error"},
                                status_code=status_code)

        return JSONResponse({"response": "disabled"})

    def _install_napp(self, request: Request) -> JSONResponse:
        username = request.path_params["username"]
        napp_name = request.path_params["napp_name"]
        # Check if the NApp is installed
        if self.napps_manager.is_installed(username, napp_name):
            return JSONResponse({"response": "installed"})

        napp = f"{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
                status_code = HTTPStatus.INTERNAL_SERVER_ERROR.value
                return JSONResponse({"response": "error"},
                                    status_code=status_code)
        except HTTPError as exception:
            return JSONResponse({"response": "error"},
                                status_code=exception.code)

        return JSONResponse({"response": "installed"})

    def _uninstall_napp(self, request: Request) -> JSONResponse:
        username = request.path_params["username"]
        napp_name = request.path_params["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
                status_code = HTTPStatus.INTERNAL_SERVER_ERROR.value
                return JSONResponse({"response": "error"},
                                    status_code=status_code)

        return JSONResponse({"response": "uninstalled"})

    def _list_enabled_napps(self, _request: Request) -> JSONResponse:
        """Sorted list of (username, napp_name) of enabled napps."""
        napps = self.napps_manager.get_enabled_napps()
        napps = [[n.username, n.name] for n in napps]
        return JSONResponse({"napps": napps})

    def _list_installed_napps(self, _request: Request) -> JSONResponse:
        """Sorted list of (username, napp_name) of installed napps."""
        napps = self.napps_manager.get_installed_napps()
        napps = [[n.username, n.name] for n in napps]
        return JSONResponse({"napps": napps})

    def _get_napp_metadata(self, request: Request) -> JSONResponse:
        """Get NApp metadata value.

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

        """
        username = request.path_params["username"]
        napp_name = request.path_params["napp_name"]
        key = request.path_params["key"]
        valid_keys = ['napp_dependencies', 'description', 'version']

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

        if key not in valid_keys:
            return JSONResponse("Invalid key.",
                                status_code=HTTPStatus.BAD_REQUEST.value)

        data = self.napps_manager.get_napp_metadata(username, napp_name, key)
        return JSONResponse({key: data})


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

kytos-ng / kytos

Pull Request — master (#375)

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

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like