gns3-server/gns3server/controller/appliance_manager.py

#!/usr/bin/env python
#
# Copyright (C) 2019 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import os
import json
import uuid
import asyncio
import aiohttp
import shutil
import platformdirs


try:
    import importlib_resources
except ImportError:
    from importlib import resources as importlib_resources


from .appliance import Appliance
from ..config import Config
from ..utils.asyncio import locking

import logging
log = logging.getLogger(__name__)


class ApplianceManager:
    """
    Manages appliances
    """

    def __init__(self):

        self._appliances = {}
        self._appliances_etag = None

    @property
    def appliances_etag(self):
        """
        :returns: ETag for downloaded appliances
        """

        return self._appliances_etag

    @appliances_etag.setter
    def appliances_etag(self, etag):
        """
        :param etag: ETag for downloaded appliances
        """

        self._appliances_etag = etag

    @property
    def appliances(self):
        """
        :returns: The dictionary of appliances managed by GNS3
        """

        return self._appliances

    def _custom_appliances_path(self):
        """
        Get the custom appliance storage directory
        """

        server_config = Config.instance().get_section_config("Server")
        appliances_path = os.path.expanduser(server_config.get("appliances_path", "~/GNS3/appliances"))
        os.makedirs(appliances_path, exist_ok=True)
        return appliances_path

    def builtin_appliances_path(self, delete_first=False):
        """
        Get the built-in appliance storage directory
        """

        appname = vendor = "GNS3"
        appliances_dir = os.path.join(platformdirs.user_data_dir(appname, vendor, roaming=True), "appliances")
        if delete_first:
            shutil.rmtree(appliances_dir, ignore_errors=True)
        os.makedirs(appliances_dir, exist_ok=True)
        return appliances_dir

    def install_builtin_appliances(self):
        """
        At startup we copy the built-in appliances files.
        """

        dst_path = self.builtin_appliances_path(delete_first=True)
        log.info(f"Installing built-in appliances in '{dst_path}'")
        from . import Controller
        try:
            Controller.instance().install_resource_files(dst_path, "appliances")
        except OSError as e:
            log.error(f"Could not install built-in appliance files to {dst_path}: {e}")

    def load_appliances(self, symbol_theme="Classic"):
        """
        Loads appliance files from disk.
        """

        self._appliances = {}
        for directory, builtin in ((self.builtin_appliances_path(), True,), (self._custom_appliances_path(), False,)):
            if directory and os.path.isdir(directory):
                for file in os.listdir(directory):
                    if not file.endswith('.gns3a') and not file.endswith('.gns3appliance'):
                        continue
                    path = os.path.join(directory, file)
                    appliance_id = uuid.uuid3(uuid.NAMESPACE_URL, path)  # Generate UUID from path to avoid change between reboots
                    try:
                        with open(path, 'r', encoding='utf-8') as f:
                            appliance = Appliance(appliance_id, json.load(f), builtin=builtin)
                            json_data = appliance.__json__()  # Check if loaded without error
                            if appliance.status != 'broken':
                                self._appliances[appliance.id] = appliance
                            if not appliance.symbol or appliance.symbol.startswith(":/symbols/"):
                                # apply a default symbol if the appliance has none or a default symbol
                                default_symbol = self._get_default_symbol(json_data, symbol_theme)
                                if default_symbol:
                                    appliance.symbol = default_symbol
                    except (ValueError, OSError, KeyError) as e:
                        log.warning("Cannot load appliance file '%s': %s", path, str(e))
                        continue

    def _get_default_symbol(self, appliance, symbol_theme):
        """
        Returns the default symbol for a given appliance.
        """

        from . import Controller
        controller = Controller.instance()
        category = appliance["category"]
        if category == "guest":
            if "docker" in appliance:
                return controller.symbols.get_default_symbol("docker_guest", symbol_theme)
            elif "qemu" in appliance:
                return controller.symbols.get_default_symbol("qemu_guest", symbol_theme)
        return controller.symbols.get_default_symbol(category, symbol_theme)

    async def download_custom_symbols(self):
        """
        Download custom appliance symbols from our GitHub registry repository.
        """

        from . import Controller
        symbol_dir = Controller.instance().symbols.symbols_path()
        self.load_appliances()
        for appliance in self._appliances.values():
            symbol = appliance.symbol
            if symbol and not symbol.startswith(":/symbols/"):
                destination_path = os.path.join(symbol_dir, symbol)
                if not os.path.exists(destination_path):
                    await self._download_symbol(symbol, destination_path)

        # refresh the symbol cache
        Controller.instance().symbols.list()

    async def _download_symbol(self, symbol, destination_path):
        """
        Download a custom appliance symbol from our GitHub registry repository.
        """

        symbol_url = "https://raw.githubusercontent.com/GNS3/gns3-registry/master/symbols/{}".format(symbol)
        async with aiohttp.ClientSession() as session:
            async with session.get(symbol_url) as response:
                if response.status != 200:
                    log.warning("Could not retrieve appliance symbol {} from GitHub due to HTTP error code {}".format(symbol, response.status))
                else:
                    try:
                        symbol_data = await response.read()
                        log.info("Saving {} symbol to {}".format(symbol, destination_path))
                        with open(destination_path, 'wb') as f:
                            f.write(symbol_data)
                    except asyncio.TimeoutError:
                        log.warning("Timeout while downloading '{}'".format(symbol_url))
                    except OSError as e:
                        log.warning("Could not write appliance symbol '{}': {}".format(destination_path, e))

    @locking
    async def download_appliances(self):
        """
        Downloads appliance files from GitHub registry repository.
        """

        try:
            headers = {}
            if self._appliances_etag:
                log.info("Checking if appliances are up-to-date (ETag {})".format(self._appliances_etag))
                headers["If-None-Match"] = self._appliances_etag
            async with aiohttp.ClientSession() as session:
                async with session.get(
                        'https://api.github.com/repos/GNS3/gns3-registry/contents/appliances',
                        headers=headers
                ) as response:
                    if response.status == 304:
                        log.info("Appliances are already up-to-date (ETag {})".format(self._appliances_etag))
                        return
                    elif response.status != 200:
                        raise aiohttp.web.HTTPConflict(text="Could not retrieve appliances from GitHub due to HTTP error code {}".format(response.status))
                    etag = response.headers.get("ETag")
                    if etag:
                        self._appliances_etag = etag
                        from . import Controller
                        Controller.instance().save()
                    json_data = await response.json()
                appliances_dir = self.builtin_appliances_path()
                downloaded_appliance_files = []
                for appliance in json_data:
                    if appliance["type"] == "file":
                        appliance_name = appliance["name"]
                        log.info("Download appliance file from '{}'".format(appliance["download_url"]))
                        async with session.get(appliance["download_url"]) as response:
                            if response.status != 200:
                                log.warning("Could not download '{}' due to HTTP error code {}".format(appliance["download_url"], response.status))
                                continue
                            try:
                                appliance_data = await response.read()
                            except asyncio.TimeoutError:
                                log.warning("Timeout while downloading '{}'".format(appliance["download_url"]))
                                continue
                            path = os.path.join(appliances_dir, appliance_name)
                            try:
                                log.info("Saving {} file to {}".format(appliance_name, path))
                                with open(path, 'wb') as f:
                                    f.write(appliance_data)
                            except OSError as e:
                                raise aiohttp.web.HTTPConflict(text="Could not write appliance file '{}': {}".format(path, e))
                        downloaded_appliance_files.append(appliance_name)

                # delete old appliance files
                for filename in os.listdir(appliances_dir):
                    file_path = os.path.join(appliances_dir, filename)
                    if filename in downloaded_appliance_files:
                        continue
                    try:
                        if os.path.isfile(file_path) or os.path.islink(file_path):
                            log.info("Deleting old appliance file {}".format(file_path))
                            os.unlink(file_path)
                    except OSError as e:
                        log.warning("Could not delete old appliance file '{}': {}".format(file_path, e))
                        continue

        except ValueError as e:
            raise aiohttp.web.HTTPConflict(text="Could not read appliances information from GitHub: {}".format(e))

        # download the custom symbols
        await self.download_custom_symbols()