kytos.core.api_server.APIServer.get_absolute_rule() - 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-05-01 14:28 UTC

APIServer.get_absolute_rule() A

↳ Parent: kytos.core.api_server

Complexity

Conditions

Size

Total Lines	10
Code Lines	4

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	4
nop	3
dl	0
loc	10
rs	10
c	0
b	0
f	0

"""Module used to handle a API Server."""
import logging
import os
import shutil
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/"
    _route_index_count = 0

    # 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)

    @classmethod
    def _get_next_route_index(cls) -> int:
        """Get next route index.

        This classmethod is meant to ensure route ordering when allocating
        an index for each route, the @rest decorator will use it. Decorated
        routes are imported sequentially so this won't need a threading Lock.
        """
        index = cls._route_index_count
        cls._route_index_count += 1
        return index

    async def serve(self):
        """Start serving."""
        await self.server.serve()

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

kytos-ng / kytos

Pull Request — master (#375)

APIServer.get_absolute_rule() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like