mirror of
https://github.com/GNS3/gns3-server
synced 2024-12-01 04:38:12 +00:00
261 lines
11 KiB
Python
261 lines
11 KiB
Python
#!/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()
|