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

Test Failed

Pull Request — master (#375)

by Vinicius

created 2023-04-20 19:12 UTC

kytos.core.api_server.APIServer.start_web_ui() A

↳ Parent: kytos.core.api_server

Complexity

Conditions

Size

Total Lines	12
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	1
eloc	11
nop	1
dl	0
loc	12
rs	9.85
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.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))
            # 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."""
        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 _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.app.router.mount("/", app=StaticFiles(directory=self.web_ui_dir),
132		name="dist")
133
134		def register_core_endpoint(self, route, function, **options):
135		"""Register an endpoint with the URL /api/kytos/core/<route>.
136
137		Not used by NApps, but controller.
138		"""
139		self._start_endpoint(self.app, f"{self._CORE_PREFIX}{route}",
140		function, **options)
141
142		def status_api(self, _request: Request):
143		"""Display kytos status using the route ``/kytos/status/``."""
144		return JSONResponse({"response": "running"})
145
146		def stop_api_server(self):
147		"""Stop API server."""
148		self._stop_uvicorn()
149
150		def static_web_ui(self,
151		request: Request) -> Union[FileResponse, JSONResponse]:
152		"""Serve static files from installed napps."""
153		username = request.path_params["username"]
154		napp_name = request.path_params["napp_name"]
155		filename = request.path_params["filename"]
156		path = f"{self.napps_dir}/{username}/{napp_name}/ui/{filename}"
157		if os.path.exists(path):
158		return FileResponse(path)
159		return JSONResponse("", status_code=HTTPStatus.NOT_FOUND.value)
160
161		def get_ui_components(self, request: Request) -> JSONResponse:
162		"""Return all napps ui components from an specific section.
163
164		The component name generated will have the following structure:
165		{username}-{nappname}-{component-section}-{filename}`
166
167		Args:
168		section_name (str): Specific section name
169
170		Returns:
171		str: Json with a list of all components found.
172
173		"""
174		section_name = request.path_params["section_name"]
175		section_name = '*' if section_name == "all" else section_name
176		path = f"{self.napps_dir}///ui/{section_name}/*.kytos"
177		components = []
178		for name in glob(path):
179		dirs_name = name.split('/')
180		dirs_name.remove('ui')
181
182		component_name = '-'.join(dirs_name[-4:]).replace('.kytos', '')
183		url = f'ui/{"/".join(dirs_name[-4:])}'
184		component = {'name': component_name, 'url': url}
185		components.append(component)
186		return JSONResponse(components)
187
188		def web_ui(self, _request: Request) -> Union[JSONResponse, FileResponse]:
189		"""Serve the index.html page for the admin-ui."""
190		index_path = f"{self.web_ui_dir}/index.html"
191		if os.path.exists(index_path):
192		return FileResponse(index_path)
193		return JSONResponse(f"File '{index_path}' not found.",
194		status_code=HTTPStatus.NOT_FOUND.value)
195
196		def _unzip_backup_web_ui(self, package: str, uri: str) -> None:
197		"""Unzip and backup web ui files.
198
199		backup the old web-ui files and create a new web-ui folder
200		if there is no path to backup, zip.extractall will
201		create the path.
202		"""
203		with zipfile.ZipFile(package, 'r') as zip_ref:
204		if zip_ref.testzip() is not None:
205		LOG.error("Web update - Zip file from %s "
206		"is corrupted.", uri)
207		raise ValueError(f'Zip file from {uri} is corrupted.')
208		if os.path.exists(self.web_ui_dir):
209		LOG.info("Web update - Performing UI backup.")
210		date = datetime.now().strftime("%Y%m%d%H%M%S")
211		shutil.move(self.web_ui_dir, f"{self.web_ui_dir}-{date}")
212		os.mkdir(self.web_ui_dir)
213		# unzip and extract files to web-ui/*
214		zip_ref.extractall(self.web_ui_dir)
215		zip_ref.close()
216
217		def _fetch_latest_ui_tag(self) -> str:
218		"""Fetch latest ui tag version from GitHub."""
219		version = '2022.3.0'
220		try:
221		url = ('https://api.github.com/repos/kytos-ng/'
222		'ui/releases/latest')
223		response = httpx.get(url, timeout=10)
224		version = response.json()['tag_name']
225		except (httpx.RequestError, KeyError):
226		msg = "Failed to fetch latest tag from GitHub, " \
227		f"falling back to {version}"
228		LOG.warning(f"Web update - {msg}")
229		return version
230
231		def update_web_ui(self, request: Optional[Request],
232		version='latest',
233		force=True) -> JSONResponse:
234		"""Update the static files for the Web UI.
235
236		Download the latest files from the UI github repository and update them
237		in the ui folder.
238		The repository link is currently hardcoded here.
239		"""
240		if not os.path.exists(self.web_ui_dir) or force:
241		if request:
242		version = request.path_params.get("version", "latest")
243		if version == 'latest':
244		version = self._fetch_latest_ui_tag()
245
246		repository = "https://github.com/kytos-ng/ui"
247		uri = repository + f"/releases/download/{version}/latest.zip"
248		try:
249		LOG.info("Web update - Downloading UI from %s.", uri)
250		package = urlretrieve(uri)[0]
251		self._unzip_backup_web_ui(package, uri)
252		except HTTPError:
253		LOG.error("Web update - Uri not found %s.", uri)
254		return JSONResponse(
255		f"Uri not found {uri}.",
256		status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value
257		)
258		except URLError:
259		LOG.error("Web update - Error accessing URL %s.", uri)
260		return JSONResponse(
261		f"Error accessing URL {uri}.",
262		status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value
263		)
264		except ValueError as exc:
265		return JSONResponse(
266		str(exc),
267		status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value
268		)
269		LOG.info("Web update - Updated")
270		return JSONResponse("Web UI was updated")
271
272		LOG.error("Web update - Web UI was not updated")
273		return JSONResponse("Web ui was not updated",
274		status_code=HTTPStatus.INTERNAL_SERVER_ERROR.value)
275
276		# BEGIN decorator methods
277		@staticmethod
278		def decorate_as_endpoint(rule, **options):
279		"""Decorate methods as REST endpoints.
280
281		Example for URL ``/api/myusername/mynapp/sayhello/World``:
282
283		.. code-block:: python3
284
285
286		from kytos.core.napps import rest
287		from kytos.core.rest_api import JSONResponse, Request
288
289		@rest("sayhello/{name}")
290		def say_hello(request: Request) -> JSONResponse:
291		name = request.path_params["name"]
292		return JSONResponse({"data": f"Hello, {name}!"})
293
294		``@rest`` parameters are the same as Starlette's ``add_route``. You can
295		also add ``methods=['POST']``, for example.
296
297		As we don't have the NApp instance now, we store the parameters in a
298		method attribute in order to add the route later, after we have both
299		APIServer and NApp instances.
300		"""
301		def store_route_params(function):
302		"""Store ``Flask`` ``@route`` parameters in a method attribute.
303
304		There can be many @route decorators in a single function.
305		"""
306		# To support any order: @classmethod, @rest or @rest, @classmethod
307		# class and static decorators return a descriptor with the function
308		# in __func__.
309		if isinstance(function, (classmethod, staticmethod)):
310		inner = function.__func__
311		else:
312		inner = function
313		# Add route parameters
314		if not hasattr(inner, 'route_params'):
315		inner.route_params = []
316		inner.route_params.append((rule, options))
317		# Return the same function, now with "route_params" attribute
318		return function
319		return store_route_params
320
321		@staticmethod
322		def get_authenticate_options():
323		"""Return configuration options related to authentication."""
324		options = KytosConfig().options['daemon']
325		return options.authenticate_urls
326
327		def authenticate_endpoints(self, napp):
328		"""Add authentication to defined REST endpoints.
329
330		If any URL marked for authentication uses a function,
331		that function will require authentication.
332		"""
333		authenticate_urls = self.get_authenticate_options()
334		for function in self._get_decorated_functions(napp):
335		inner = getattr(function, '__func__', function)
336		inner.authenticated = False
337		for rule, _ in function.route_params:
338		if inner.authenticated:
339		break
340		absolute_rule = self.get_absolute_rule(rule, napp)
341		for url in authenticate_urls:
342		if url in absolute_rule:
343		inner.authenticated = True
344		break
345
346		def register_napp_endpoints(self, napp):
347		"""Add all NApp REST endpoints with @rest decorator.
348
349		URLs will be prefixed with ``/api/{username}/{napp_name}/``.
350
351		Args:
352		napp (Napp): Napp instance to register new endpoints.
353		"""
354		# Start all endpoints for this NApp
355		for function in self._get_decorated_functions(napp):
356		for rule, options in function.route_params:
357		absolute_rule = self.get_absolute_rule(rule, napp)
358		if getattr(function, 'authenticated', False):
359		function = authenticated(function)
360		self._start_endpoint(self.app, absolute_rule,
361		function, **options)
362
363		@staticmethod
364		def _get_decorated_functions(napp):
365		"""Return ``napp``'s methods having the @rest decorator."""
366		for name in dir(napp):
367		if not name.startswith('_'): # discarding private names
368		pub_attr = getattr(napp, name)
369		if callable(pub_attr) and hasattr(pub_attr, 'route_params'):
370		yield pub_attr
371
372		@classmethod
373		def get_absolute_rule(cls, rule, napp):
374		"""Prefix the rule, e.g. "flow" to "/api/user/napp/flow".
375
376		This code is used by kytos-utils when generating an OpenAPI skel.
377		"""
378		# Flask does require 2 slashes if specified, so we remove a starting
379		# slash if applicable.
380		relative_rule = rule[1:] if rule.startswith('/') else rule
381		return cls._NAPP_PREFIX.format(napp=napp) + relative_rule
382
383		# END decorator methods
384
385		def _add_non_strict_slash_route(
386		self,
387		app: Starlette,
388		route: str,
389		function,
390		**options
391		) -> None:
392		"""Try to add a non strict slash route."""
393		if route == "/":
394		return
395		non_strict = route[:-1] if route.endswith("/") else f"{route}/"
396		app.router.add_route(non_strict, function, **options,
397		include_in_schema=False)
398
399		def _start_endpoint(
400		self,
401		app: Starlette,
402		route: str,
403		function,
404		**options
405		):
406		"""Start ``function``'s endpoint."""
407		app.router.add_route(route, function, **options)
408		self._add_non_strict_slash_route(app, route, function, **options)
409		LOG.info('Started %s - %s', route,
410		', '.join(options.get('methods', self.DEFAULT_METHODS)))
411
412		def remove_napp_endpoints(self, napp):
413		"""Remove all decorated endpoints.
414
415		Args:
416		napp (Napp): Napp instance to look for rest-decorated methods.
417		"""
418		prefix = self._NAPP_PREFIX.format(napp=napp)
419		indexes = []
420		for index, route in enumerate(self.app.routes):
421		if route.path.startswith(prefix):
422		indexes.append(index)
423		for index in reversed(indexes):
424		self.app.routes.pop(index)
425		LOG.info('The Rest endpoints from %s were disabled.', prefix)
426
427		def register_core_napp_services(self):
428		"""
429		Register /kytos/core/ services over NApps.
430
431		It registers enable, disable, install, uninstall NApps that will
432		be used by kytos-utils.
433		"""
434		self.register_core_endpoint("napps/{username}/{napp_name}/enable",
435		self._enable_napp)
436		self.register_core_endpoint("napps/{username}/{napp_name}/disable",
437		self._disable_napp)
438		self.register_core_endpoint("napps/{username}/{napp_name}/install",
439		self._install_napp)
440		self.register_core_endpoint("napps/{username}/{napp_name}/uninstall",
441		self._uninstall_napp)
442		self.register_core_endpoint("napps_enabled",
443		self._list_enabled_napps)
444		self.register_core_endpoint("napps_installed",
445		self._list_installed_napps)
446		self.register_core_endpoint(
447		"napps/{username}/{napp_name}/metadata/{key}",
448		self._get_napp_metadata)
449
450	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...
451		"""Enable an installed NApp."""
452
453		username = request.path_params["username"]
454		napp_name = request.path_params["napp_name"]
455		# Check if the NApp is installed
456		if not self.napps_manager.is_installed(username, napp_name):
457		return JSONResponse({"response": "not installed"},
458		status_code=HTTPStatus.BAD_REQUEST.value)
459
460		# Check if the NApp is already been enabled
461		if not self.napps_manager.is_enabled(username, napp_name):
462		self.napps_manager.enable(username, napp_name)
463
464		# Check if NApp is enabled
465		if not self.napps_manager.is_enabled(username, napp_name):
466		# If it is not enabled an admin user must check the log file
467		status_code = HTTPStatus.INTERNAL_SERVER_ERROR.value
468		return JSONResponse({"response": "error"},
469		status_code=status_code)
470
471		return JSONResponse({"response": "enabled"})
472
473	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...
474		"""Disable an installed NApp."""
475
476		username = request.path_params["username"]
477		napp_name = request.path_params["napp_name"]
478		# Check if the NApp is installed
479		if not self.napps_manager.is_installed(username, napp_name):
480		return JSONResponse({"response": "not installed"},
481		status_code=HTTPStatus.BAD_REQUEST.value)
482
483		# Check if the NApp is enabled
484		if self.napps_manager.is_enabled(username, napp_name):
485		self.napps_manager.disable(username, napp_name)
486
487		# Check if NApp is still enabled
488		if self.napps_manager.is_enabled(username, napp_name):
489		# If it is still enabled an admin user must check the log file
490		status_code = HTTPStatus.INTERNAL_SERVER_ERROR.value
491		return JSONResponse({"response": "error"},
492		status_code=status_code)
493
494		return JSONResponse({"response": "disabled"})
495
496		def _install_napp(self, request: Request) -> JSONResponse:
497		username = request.path_params["username"]
498		napp_name = request.path_params["napp_name"]
499		# Check if the NApp is installed
500		if self.napps_manager.is_installed(username, napp_name):
501		return JSONResponse({"response": "installed"})
502
503		napp = f"{username}/{napp_name}"
504
505		# Try to install and enable the napp
506		try:
507		if not self.napps_manager.install(napp, enable=True):
508		# If it is not installed an admin user must check the log file
509		status_code = HTTPStatus.INTERNAL_SERVER_ERROR.value
510		return JSONResponse({"response": "error"},
511		status_code=status_code)
512		except HTTPError as exception:
513		return JSONResponse({"response": "error"},
514		status_code=exception.code)
515
516		return JSONResponse({"response": "installed"})
517
518		def _uninstall_napp(self, request: Request) -> JSONResponse:
519		username = request.path_params["username"]
520		napp_name = request.path_params["napp_name"]
521		# Check if the NApp is installed
522		if self.napps_manager.is_installed(username, napp_name):
523		# Try to unload/uninstall the napp
524		if not self.napps_manager.uninstall(username, napp_name):
525		# If it is not uninstalled 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
530		return JSONResponse({"response": "uninstalled"})
531
532		def _list_enabled_napps(self, _request: Request) -> JSONResponse:
533		"""Sorted list of (username, napp_name) of enabled napps."""
534		napps = self.napps_manager.get_enabled_napps()
535		napps = [[n.username, n.name] for n in napps]
536		return JSONResponse({"napps": napps})
537
538		def _list_installed_napps(self, _request: Request) -> JSONResponse:
539		"""Sorted list of (username, napp_name) of installed napps."""
540		napps = self.napps_manager.get_installed_napps()
541		napps = [[n.username, n.name] for n in napps]
542		return JSONResponse({"napps": napps})
543
544		def _get_napp_metadata(self, request: Request) -> JSONResponse:
545		"""Get NApp metadata value.
546
547		For safety reasons, only some keys can be retrieved:
548		napp_dependencies, description, version.
549
550		"""
551		username = request.path_params["username"]
552		napp_name = request.path_params["napp_name"]
553		key = request.path_params["key"]
554		valid_keys = ['napp_dependencies', 'description', 'version']
555
556		if not self.napps_manager.is_installed(username, napp_name):
557		return JSONResponse("NApp is not installed.",
558		status_code=HTTPStatus.BAD_REQUEST.value)
559
560		if key not in valid_keys:
561		return JSONResponse("Invalid key.",
562		status_code=HTTPStatus.BAD_REQUEST.value)
563
564		data = self.napps_manager.get_napp_metadata(username, napp_name, key)
565		return JSONResponse({key: data})
566

kytos-ng / kytos

Pull Request — master (#375)

kytos.core.api_server.APIServer.start_web_ui() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like