# # Copyright (C) 2021 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 . import os import uuid import pydantic from uuid import UUID from fastapi.encoders import jsonable_encoder from typing import List from gns3server import schemas from gns3server.config import Config import gns3server.db.models as models from gns3server.db.repositories.templates import TemplatesRepository from gns3server.controller.controller_error import ( ControllerError, ControllerBadRequestError, ControllerNotFoundError, ControllerForbiddenError, ) TEMPLATE_TYPE_TO_SCHEMA = { "cloud": schemas.CloudTemplate, "ethernet_hub": schemas.EthernetHubTemplate, "ethernet_switch": schemas.EthernetSwitchTemplate, "docker": schemas.DockerTemplate, "dynamips": schemas.DynamipsTemplate, "vpcs": schemas.VPCSTemplate, "virtualbox": schemas.VirtualBoxTemplate, "vmware": schemas.VMwareTemplate, "iou": schemas.IOUTemplate, "qemu": schemas.QemuTemplate, } TEMPLATE_TYPE_TO_UPDATE_SCHEMA = { "cloud": schemas.CloudTemplateUpdate, "ethernet_hub": schemas.EthernetHubTemplateUpdate, "ethernet_switch": schemas.EthernetSwitchTemplateUpdate, "docker": schemas.DockerTemplateUpdate, "vpcs": schemas.VPCSTemplateUpdate, "virtualbox": schemas.VirtualBoxTemplateUpdate, "vmware": schemas.VMwareTemplateUpdate, "iou": schemas.IOUTemplateUpdate, "qemu": schemas.QemuTemplateUpdate, } DYNAMIPS_PLATFORM_TO_SCHEMA = { "c7200": schemas.C7200DynamipsTemplate, "c3745": schemas.C3745DynamipsTemplate, "c3725": schemas.C3725DynamipsTemplate, "c3600": schemas.C3600DynamipsTemplate, "c2691": schemas.C2691DynamipsTemplate, "c2600": schemas.C2600DynamipsTemplate, "c1700": schemas.C1700DynamipsTemplate, } DYNAMIPS_PLATFORM_TO_UPDATE_SCHEMA = { "c7200": schemas.C7200DynamipsTemplateUpdate, "c3745": schemas.C3745DynamipsTemplateUpdate, "c3725": schemas.C3725DynamipsTemplateUpdate, "c3600": schemas.C3600DynamipsTemplateUpdate, "c2691": schemas.C2691DynamipsTemplateUpdate, "c2600": schemas.C2600DynamipsTemplateUpdate, "c1700": schemas.C1700DynamipsTemplateUpdate, } # built-in templates have their compute_id set to None to tell clients to select a compute BUILTIN_TEMPLATES = [ { "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "cloud"), "template_type": "cloud", "name": "Cloud", "default_name_format": "Cloud{0}", "category": "guest", "symbol": "cloud", "compute_id": None, "builtin": True, }, { "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "nat"), "template_type": "nat", "name": "NAT", "default_name_format": "NAT{0}", "category": "guest", "symbol": "nat", "compute_id": None, "builtin": True, }, { "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "vpcs"), "template_type": "vpcs", "name": "VPCS", "default_name_format": "PC{0}", "category": "guest", "symbol": "vpcs_guest", "base_script_file": "vpcs_base_config.txt", "compute_id": None, "builtin": True, }, { "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "ethernet_switch"), "template_type": "ethernet_switch", "name": "Ethernet switch", "console_type": "none", "default_name_format": "Switch{0}", "category": "switch", "symbol": "ethernet_switch", "compute_id": None, "builtin": True, }, { "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "ethernet_hub"), "template_type": "ethernet_hub", "name": "Ethernet hub", "default_name_format": "Hub{0}", "category": "switch", "symbol": "hub", "compute_id": None, "builtin": True, }, { "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "frame_relay_switch"), "template_type": "frame_relay_switch", "name": "Frame Relay switch", "default_name_format": "FRSW{0}", "category": "switch", "symbol": "frame_relay_switch", "compute_id": None, "builtin": True, }, { "template_id": uuid.uuid5(uuid.NAMESPACE_X500, "atm_switch"), "template_type": "atm_switch", "name": "ATM switch", "default_name_format": "ATMSW{0}", "category": "switch", "symbol": "atm_switch", "compute_id": None, "builtin": True, }, ] class TemplatesService: def __init__(self, templates_repo: TemplatesRepository): self._templates_repo = templates_repo from gns3server.controller import Controller self._controller = Controller.instance() # resolve built-in template symbols for builtin_template in BUILTIN_TEMPLATES: builtin_template["symbol"] = self._controller.symbols.resolve_symbol(builtin_template["symbol"]) def get_builtin_template(self, template_id: UUID) -> dict: for builtin_template in BUILTIN_TEMPLATES: if builtin_template["template_id"] == template_id: return jsonable_encoder(builtin_template) async def get_templates(self) -> List[dict]: templates = [] db_templates = await self._templates_repo.get_templates() for db_template in db_templates: templates.append(db_template.asjson()) if Config.instance().settings.Server.enable_builtin_templates: for builtin_template in BUILTIN_TEMPLATES: templates.append(jsonable_encoder(builtin_template)) return templates async def _find_image(self, image_path: str): image = await self._templates_repo.get_image(image_path) if not image: raise ControllerNotFoundError(f"Image '{image_path}' could not be found in the controller database") if not os.path.exists(image.path): raise ControllerNotFoundError(f"Image '{image.path}' could not be found on disk") return image async def _find_images(self, template_type: str, settings: dict) -> List[models.Image]: images_to_add_to_template = [] if template_type == "dynamips": if settings["image"]: image = await self._find_image(settings["image"]) if image.image_type != "ios": raise ControllerBadRequestError( f"Image '{image.filename}' type is not 'ios' but '{image.image_type}'" ) images_to_add_to_template.append(image) elif template_type == "iou": if settings["path"]: image = await self._find_image(settings["path"]) if image.image_type != "iou": raise ControllerBadRequestError( f"Image '{image.filename}' type is not 'iou' but '{image.image_type}'" ) images_to_add_to_template.append(image) elif template_type == "qemu": for key, value in settings.items(): if key.endswith("_image") and value: image = await self._find_image(value) if image.image_type != "qemu": raise ControllerBadRequestError( f"Image '{image.filename}' type is not 'qemu' but '{image.image_type}'" ) if image not in images_to_add_to_template: images_to_add_to_template.append(image) return images_to_add_to_template async def create_template(self, template_create: schemas.TemplateCreate) -> dict: if await self._templates_repo.get_template_by_name_and_version(template_create.name, template_create.version): if template_create.version: raise ControllerError(f"A template with name '{template_create.name}' and " f"version {template_create.version} already exists") else: raise ControllerError(f"A template with name '{template_create.name}' already exists") try: # get the default template settings create_settings = jsonable_encoder(template_create, exclude_unset=True) template_schema = TEMPLATE_TYPE_TO_SCHEMA[template_create.template_type] template_settings = template_schema.model_validate(create_settings).model_dump() if template_create.template_type == "dynamips": # special case for Dynamips to cover all platform types that contain specific settings dynamips_template_schema = DYNAMIPS_PLATFORM_TO_SCHEMA[template_settings["platform"]] template_settings = dynamips_template_schema.model_validate(create_settings).model_dump() except pydantic.ValidationError as e: raise ControllerBadRequestError(f"JSON schema error received while creating new template: {e}") # resolve the template symbol template_settings["symbol"] = self._controller.symbols.resolve_symbol(template_settings["symbol"]) images_to_add_to_template = await self._find_images(template_create.template_type, template_settings) db_template = await self._templates_repo.create_template(template_create.template_type, template_settings) for image in images_to_add_to_template: await self._templates_repo.add_image_to_template(db_template.template_id, image) template = db_template.asjson() self._controller.notification.controller_emit("template.created", template) return template async def get_template(self, template_id: UUID) -> dict: db_template = await self._templates_repo.get_template(template_id) if db_template: template = db_template.asjson() else: template = self.get_builtin_template(template_id) if not template: raise ControllerNotFoundError(f"Template '{template_id}' not found") return template async def _remove_image(self, template_id: UUID, image_path:str) -> None: image = await self._templates_repo.get_image(image_path) await self._templates_repo.remove_image_from_template(template_id, image) async def update_template(self, template_id: UUID, template_update: schemas.TemplateUpdate) -> dict: if self.get_builtin_template(template_id): raise ControllerForbiddenError(f"Template '{template_id}' cannot be updated because it is built-in") db_template = await self._templates_repo.get_template(template_id) if not db_template: raise ControllerNotFoundError(f"Template '{template_id}' not found") try: # validate the update settings update_settings = jsonable_encoder(template_update, exclude_unset=True) if db_template.template_type == "dynamips": template_schema = DYNAMIPS_PLATFORM_TO_UPDATE_SCHEMA[db_template.platform] else: template_schema = TEMPLATE_TYPE_TO_UPDATE_SCHEMA[db_template.template_type] template_settings = template_schema.model_validate(update_settings).model_dump(exclude_unset=True) except pydantic.ValidationError as e: raise ControllerBadRequestError(f"JSON schema error received while updating template: {e}") images_to_add_to_template = await self._find_images(db_template.template_type, template_settings) if db_template.template_type == "dynamips" and "image" in template_settings: await self._remove_image(db_template.template_id, db_template.image) elif db_template.template_type == "iou" and "path" in template_settings: await self._remove_image(db_template.template_id, db_template.path) elif db_template.template_type == "qemu": for key in template_update.model_dump().keys(): if key.endswith("_image") and key in template_settings: await self._remove_image(db_template.template_id, db_template.__dict__[key]) db_template = await self._templates_repo.update_template(db_template, template_settings) for image in images_to_add_to_template: await self._templates_repo.add_image_to_template(db_template.template_id, image) template = db_template.asjson() self._controller.notification.controller_emit("template.updated", template) return template async def duplicate_template(self, template_id: UUID) -> dict: if self.get_builtin_template(template_id): raise ControllerForbiddenError(f"Template '{template_id}' cannot be duplicated because it is built-in") db_template = await self._templates_repo.duplicate_template(template_id) if not db_template: raise ControllerNotFoundError(f"Template '{template_id}' not found") template = db_template.asjson() self._controller.notification.controller_emit("template.created", template) return template async def delete_template(self, template_id: UUID) -> None: if self.get_builtin_template(template_id): raise ControllerForbiddenError(f"Template '{template_id}' cannot be deleted because it is built-in") if await self._templates_repo.delete_template(template_id): self._controller.notification.controller_emit("template.deleted", {"template_id": str(template_id)}) else: raise ControllerNotFoundError(f"Template '{template_id}' not found")