+ @staticmethod
+ def validate_helm_chart(indata):
+ kdus = indata.get("kdu", [])
+ for kdu in kdus:
+ helm_chart_value = kdu.get("helm-chart")
+ if not helm_chart_value:
+ continue
+ if not valid_helm_chart_re.match(helm_chart_value):
+ raise EngineException(
+ "helm-chart '{}' is not valid".format(helm_chart_value),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def validate_mgmt_interface_connection_point(indata):
+ if not indata.get("vdu"):
+ return
+ if not indata.get("mgmt-cp"):
+ raise EngineException(
+ "'mgmt-cp' is a mandatory field and it is not defined",
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ for cp in get_iterable(indata.get("ext-cpd")):
+ if cp["id"] == indata["mgmt-cp"]:
+ break
+ else:
+ raise EngineException(
+ "mgmt-cp='{}' must match an existing ext-cpd".format(indata["mgmt-cp"]),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def validate_vdu_internal_connection_points(vdu):
+ int_cpds = set()
+ for cpd in get_iterable(vdu.get("int-cpd")):
+ cpd_id = cpd.get("id")
+ if cpd_id and cpd_id in int_cpds:
+ raise EngineException(
+ "vdu[id='{}']:int-cpd[id='{}'] is already used by other int-cpd".format(
+ vdu["id"], cpd_id
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ int_cpds.add(cpd_id)
+
+ @staticmethod
+ def validate_external_connection_points(indata):
+ all_vdus_int_cpds = set()
+ for vdu in get_iterable(indata.get("vdu")):
+ for int_cpd in get_iterable(vdu.get("int-cpd")):
+ all_vdus_int_cpds.add((vdu.get("id"), int_cpd.get("id")))
+
+ ext_cpds = set()
+ for cpd in get_iterable(indata.get("ext-cpd")):
+ cpd_id = cpd.get("id")
+ if cpd_id and cpd_id in ext_cpds:
+ raise EngineException(
+ "ext-cpd[id='{}'] is already used by other ext-cpd".format(cpd_id),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ ext_cpds.add(cpd_id)
+
+ int_cpd = cpd.get("int-cpd")
+ if int_cpd:
+ if (int_cpd.get("vdu-id"), int_cpd.get("cpd")) not in all_vdus_int_cpds:
+ raise EngineException(
+ "ext-cpd[id='{}']:int-cpd must match an existing vdu int-cpd".format(
+ cpd_id
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ # TODO: Validate k8s-cluster-net points to a valid k8s-cluster:nets ?
+
+ def _validate_vdu_charms_in_package(self, storage_params, indata):
+ for df in indata["df"]:
+ if (
+ "lcm-operations-configuration" in df
+ and "operate-vnf-op-config" in df["lcm-operations-configuration"]
+ ):
+ configs = df["lcm-operations-configuration"][
+ "operate-vnf-op-config"
+ ].get("day1-2", [])
+ vdus = df.get("vdu-profile", [])
+ for vdu in vdus:
+ for config in configs:
+ if config["id"] == vdu["id"] and utils.find_in_list(
+ config.get("execution-environment-list", []),
+ lambda ee: "juju" in ee,
+ ):
+ if not self._validate_package_folders(
+ storage_params, "charms"
+ ) and not self._validate_package_folders(
+ storage_params, "Scripts/charms"
+ ):
+ raise EngineException(
+ "Charm defined in vnf[id={}] but not present in "
+ "package".format(indata["id"])
+ )
+
+ def _validate_vdu_cloud_init_in_package(self, storage_params, vdu, indata):
+ if not vdu.get("cloud-init-file"):
+ return
+ if not self._validate_package_folders(
+ storage_params, "cloud_init", vdu["cloud-init-file"]
+ ) and not self._validate_package_folders(
+ storage_params, "Scripts/cloud_init", vdu["cloud-init-file"]
+ ):
+ raise EngineException(
+ "Cloud-init defined in vnf[id={}]:vdu[id={}] but not present in "
+ "package".format(indata["id"], vdu["id"])
+ )
+
+ def _validate_vnf_charms_in_package(self, storage_params, indata):
+ # Get VNF configuration through new container
+ for deployment_flavor in indata.get("df", []):
+ if "lcm-operations-configuration" not in deployment_flavor:
+ return
+ if (
+ "operate-vnf-op-config"
+ not in deployment_flavor["lcm-operations-configuration"]
+ ):
+ return
+ for day_1_2_config in deployment_flavor["lcm-operations-configuration"][
+ "operate-vnf-op-config"
+ ]["day1-2"]:
+ if day_1_2_config["id"] == indata["id"]:
+ if utils.find_in_list(
+ day_1_2_config.get("execution-environment-list", []),
+ lambda ee: "juju" in ee,
+ ):
+ if not self._validate_package_folders(
+ storage_params, "charms"
+ ) and not self._validate_package_folders(
+ storage_params, "Scripts/charms"
+ ):
+ raise EngineException(
+ "Charm defined in vnf[id={}] but not present in "
+ "package".format(indata["id"])
+ )
+
+ def _validate_package_folders(self, storage_params, folder, file=None):
+ if not storage_params:
+ return False
+ elif not storage_params.get("pkg-dir"):
+ if self.fs.file_exists("{}_".format(storage_params["folder"]), "dir"):
+ f = "{}_/{}".format(storage_params["folder"], folder)
+ else:
+ f = "{}/{}".format(storage_params["folder"], folder)
+ if file:
+ return self.fs.file_exists("{}/{}".format(f, file), "file")
+ else:
+ if self.fs.file_exists(f, "dir"):
+ if self.fs.dir_ls(f):
+ return True
+ return False
+ else:
+ if self.fs.file_exists("{}_".format(storage_params["folder"]), "dir"):
+ f = "{}_/{}/{}".format(
+ storage_params["folder"], storage_params["pkg-dir"], folder
+ )
+ else:
+ f = "{}/{}/{}".format(
+ storage_params["folder"], storage_params["pkg-dir"], folder
+ )
+ if file:
+ return self.fs.file_exists("{}/{}".format(f, file), "file")
+ else:
+ if self.fs.file_exists(f, "dir"):
+ if self.fs.dir_ls(f):
+ return True
+ return False
+
+ @staticmethod
+ def validate_internal_virtual_links(indata):
+ all_ivld_ids = set()
+ for ivld in get_iterable(indata.get("int-virtual-link-desc")):
+ ivld_id = ivld.get("id")
+ if ivld_id and ivld_id in all_ivld_ids:
+ raise EngineException(
+ "Duplicated VLD id in int-virtual-link-desc[id={}]".format(ivld_id),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ else:
+ all_ivld_ids.add(ivld_id)
+
+ for vdu in get_iterable(indata.get("vdu")):
+ for int_cpd in get_iterable(vdu.get("int-cpd")):
+ int_cpd_ivld_id = int_cpd.get("int-virtual-link-desc")
+ if int_cpd_ivld_id and int_cpd_ivld_id not in all_ivld_ids:
+ raise EngineException(
+ "vdu[id='{}']:int-cpd[id='{}']:int-virtual-link-desc='{}' must match an existing "
+ "int-virtual-link-desc".format(
+ vdu["id"], int_cpd["id"], int_cpd_ivld_id
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ for df in get_iterable(indata.get("df")):
+ for vlp in get_iterable(df.get("virtual-link-profile")):
+ vlp_ivld_id = vlp.get("id")
+ if vlp_ivld_id and vlp_ivld_id not in all_ivld_ids:
+ raise EngineException(
+ "df[id='{}']:virtual-link-profile='{}' must match an existing "
+ "int-virtual-link-desc".format(df["id"], vlp_ivld_id),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def validate_monitoring_params(indata):
+ all_monitoring_params = set()
+ for ivld in get_iterable(indata.get("int-virtual-link-desc")):
+ for mp in get_iterable(ivld.get("monitoring-parameters")):
+ mp_id = mp.get("id")
+ if mp_id and mp_id in all_monitoring_params:
+ raise EngineException(
+ "Duplicated monitoring-parameter id in "
+ "int-virtual-link-desc[id='{}']:monitoring-parameters[id='{}']".format(
+ ivld["id"], mp_id
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ else:
+ all_monitoring_params.add(mp_id)
+
+ for vdu in get_iterable(indata.get("vdu")):
+ for mp in get_iterable(vdu.get("monitoring-parameter")):
+ mp_id = mp.get("id")
+ if mp_id and mp_id in all_monitoring_params:
+ raise EngineException(
+ "Duplicated monitoring-parameter id in "
+ "vdu[id='{}']:monitoring-parameter[id='{}']".format(
+ vdu["id"], mp_id
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ else:
+ all_monitoring_params.add(mp_id)
+
+ for df in get_iterable(indata.get("df")):
+ for mp in get_iterable(df.get("monitoring-parameter")):
+ mp_id = mp.get("id")
+ if mp_id and mp_id in all_monitoring_params:
+ raise EngineException(
+ "Duplicated monitoring-parameter id in "
+ "df[id='{}']:monitoring-parameter[id='{}']".format(
+ df["id"], mp_id
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ else:
+ all_monitoring_params.add(mp_id)
+
+ @staticmethod
+ def validate_scaling_group_descriptor(indata):
+ all_monitoring_params = set()
+ all_vdu_ids = set()
+ for df in get_iterable(indata.get("df")):
+ for il in get_iterable(df.get("instantiation-level")):
+ for vl in get_iterable(il.get("vdu-level")):
+ all_vdu_ids.add(vl.get("vdu-id"))
+
+ for ivld in get_iterable(indata.get("int-virtual-link-desc")):
+ for mp in get_iterable(ivld.get("monitoring-parameters")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for vdu in get_iterable(indata.get("vdu")):
+ for mp in get_iterable(vdu.get("monitoring-parameter")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for df in get_iterable(indata.get("df")):
+ for mp in get_iterable(df.get("monitoring-parameter")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for df in get_iterable(indata.get("df")):
+ for sa in get_iterable(df.get("scaling-aspect")):
+ for deltas in get_iterable(
+ sa.get("aspect-delta-details").get("deltas")
+ ):
+ for vds in get_iterable(deltas.get("vdu-delta")):
+ sa_vdu_id = vds.get("id")
+ if sa_vdu_id and sa_vdu_id not in all_vdu_ids:
+ raise EngineException(
+ "df[id='{}']:scaling-aspect[id='{}']:aspect-delta-details"
+ "[delta='{}']: "
+ "vdu-id='{}' not defined in vdu".format(
+ df["id"],
+ sa["id"],
+ deltas["id"],
+ sa_vdu_id,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ for df in get_iterable(indata.get("df")):
+ for sa in get_iterable(df.get("scaling-aspect")):
+ for sp in get_iterable(sa.get("scaling-policy")):
+ for sc in get_iterable(sp.get("scaling-criteria")):
+ sc_monitoring_param = sc.get("vnf-monitoring-param-ref")
+ if (
+ sc_monitoring_param
+ and sc_monitoring_param not in all_monitoring_params
+ ):
+ raise EngineException(
+ "df[id='{}']:scaling-aspect[id='{}']:scaling-policy"
+ "[name='{}']:scaling-criteria[name='{}']: "
+ "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
+ df["id"],
+ sa["id"],
+ sp["name"],
+ sc["name"],
+ sc_monitoring_param,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ for sca in get_iterable(sa.get("scaling-config-action")):
+ if (
+ "lcm-operations-configuration" not in df
+ or "operate-vnf-op-config"
+ not in df["lcm-operations-configuration"]
+ or not utils.find_in_list(
+ df["lcm-operations-configuration"][
+ "operate-vnf-op-config"
+ ].get("day1-2", []),
+ lambda config: config["id"] == indata["id"],
+ )
+ ):
+ raise EngineException(
+ "'day1-2 configuration' not defined in the descriptor but it is "
+ "referenced by df[id='{}']:scaling-aspect[id='{}']:scaling-config-action".format(
+ df["id"], sa["id"]
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ for configuration in get_iterable(
+ df["lcm-operations-configuration"]["operate-vnf-op-config"].get(
+ "day1-2", []
+ )
+ ):
+ for primitive in get_iterable(
+ configuration.get("config-primitive")
+ ):
+ if (
+ primitive["name"]
+ == sca["vnf-config-primitive-name-ref"]
+ ):
+ break
+ else:
+ raise EngineException(
+ "df[id='{}']:scaling-aspect[id='{}']:scaling-config-action:vnf-"
+ "config-primitive-name-ref='{}' does not match any "
+ "day1-2 configuration:config-primitive:name".format(
+ df["id"],
+ sa["id"],
+ sca["vnf-config-primitive-name-ref"],
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def validate_healing_group_descriptor(indata):
+ all_vdu_ids = set()
+ for df in get_iterable(indata.get("df")):
+ for il in get_iterable(df.get("instantiation-level")):
+ for vl in get_iterable(il.get("vdu-level")):
+ all_vdu_ids.add(vl.get("vdu-id"))
+
+ for df in get_iterable(indata.get("df")):
+ for ha in get_iterable(df.get("healing-aspect")):
+ for hp in get_iterable(ha.get("healing-policy")):
+ hp_monitoring_param = hp.get("vdu-id")
+ if hp_monitoring_param and hp_monitoring_param not in all_vdu_ids:
+ raise EngineException(
+ "df[id='{}']:healing-aspect[id='{}']:healing-policy"
+ "[name='{}']: "
+ "vdu-id='{}' not defined in vdu".format(
+ df["id"],
+ ha["id"],
+ hp["event-name"],
+ hp_monitoring_param,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def validate_alarm_group_descriptor(indata):
+ all_monitoring_params = set()
+ for ivld in get_iterable(indata.get("int-virtual-link-desc")):
+ for mp in get_iterable(ivld.get("monitoring-parameters")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for vdu in get_iterable(indata.get("vdu")):
+ for mp in get_iterable(vdu.get("monitoring-parameter")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for df in get_iterable(indata.get("df")):
+ for mp in get_iterable(df.get("monitoring-parameter")):
+ all_monitoring_params.add(mp.get("id"))
+
+ for vdus in get_iterable(indata.get("vdu")):
+ for alarms in get_iterable(vdus.get("alarm")):
+ alarm_monitoring_param = alarms.get("vnf-monitoring-param-ref")
+ if (
+ alarm_monitoring_param
+ and alarm_monitoring_param not in all_monitoring_params
+ ):
+ raise EngineException(
+ "vdu[id='{}']:alarm[id='{}']:"
+ "vnf-monitoring-param-ref='{}' not defined in any monitoring-param".format(
+ vdus["id"],
+ alarms["alarm-id"],
+ alarm_monitoring_param,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ @staticmethod
+ def validate_storage_compute_descriptor(indata):
+ all_vsd_ids = set()
+ for vsd in get_iterable(indata.get("virtual-storage-desc")):
+ all_vsd_ids.add(vsd.get("id"))
+
+ all_vcd_ids = set()
+ for vcd in get_iterable(indata.get("virtual-compute-desc")):
+ all_vcd_ids.add(vcd.get("id"))
+
+ for vdus in get_iterable(indata.get("vdu")):
+ for vsd_ref in vdus.get("virtual-storage-desc"):
+ if vsd_ref and vsd_ref not in all_vsd_ids:
+ raise EngineException(
+ "vdu[virtual-storage-desc='{}']"
+ "not defined in vnfd".format(
+ vsd_ref,
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ for vdus in get_iterable(indata.get("vdu")):
+ vcd_ref = vdus.get("virtual-compute-desc")
+ if vcd_ref and vcd_ref not in all_vcd_ids:
+ raise EngineException(
+ "vdu[virtual-compute-desc='{}']"
+ "not defined in vnfd".format(
+ vdus["virtual-compute-desc"],
+ ),
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+
+ def delete_extra(self, session, _id, db_content, not_send_msg=None):
+ """
+ Deletes associate file system storage (via super)
+ Deletes associated vnfpkgops from database.
+ :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
+ :param _id: server internal id
+ :param db_content: The database content of the descriptor
+ :return: None
+ :raises: FsException in case of error while deleting associated storage
+ """
+ super().delete_extra(session, _id, db_content, not_send_msg)
+ self.db.del_list("vnfpkgops", {"vnfPkgId": _id})
+ self.db.del_list(self.topic + "_revisions", {"_id": {"$regex": _id}})
+
+ def sol005_projection(self, data):
+ data["onboardingState"] = data["_admin"]["onboardingState"]
+ data["operationalState"] = data["_admin"]["operationalState"]
+ data["usageState"] = data["_admin"]["usageState"]
+
+ links = {}
+ links["self"] = {"href": "/vnfpkgm/v1/vnf_packages/{}".format(data["_id"])}
+ links["vnfd"] = {"href": "/vnfpkgm/v1/vnf_packages/{}/vnfd".format(data["_id"])}
+ links["packageContent"] = {
+ "href": "/vnfpkgm/v1/vnf_packages/{}/package_content".format(data["_id"])
+ }
+ data["_links"] = links
+
+ return super().sol005_projection(data)
+
+ @staticmethod
+ def find_software_version(vnfd: dict) -> str:
+ """Find the sotware version in the VNFD descriptors
+
+ Args:
+ vnfd (dict): Descriptor as a dictionary
+
+ Returns:
+ software-version (str)
+ """
+ default_sw_version = "1.0"
+ if vnfd.get("vnfd"):
+ vnfd = vnfd["vnfd"]
+ if vnfd.get("software-version"):
+ return vnfd["software-version"]
+ else:
+ return default_sw_version
+
+ @staticmethod
+ def extract_policies(vnfd: dict) -> dict:
+ """Removes the policies from the VNFD descriptors
+
+ Args:
+ vnfd (dict): Descriptor as a dictionary
+
+ Returns:
+ vnfd (dict): VNFD which does not include policies
+ """
+ for df in vnfd.get("df", {}):
+ for policy in ["scaling-aspect", "healing-aspect"]:
+ if df.get(policy, {}):
+ df.pop(policy)
+ for vdu in vnfd.get("vdu", {}):
+ for alarm_policy in ["alarm", "monitoring-parameter"]:
+ if vdu.get(alarm_policy, {}):
+ vdu.pop(alarm_policy)
+ return vnfd
+
+ @staticmethod
+ def extract_day12_primitives(vnfd: dict) -> dict:
+ """Removes the day12 primitives from the VNFD descriptors
+
+ Args:
+ vnfd (dict): Descriptor as a dictionary
+
+ Returns:
+ vnfd (dict)
+ """
+ for df_id, df in enumerate(vnfd.get("df", {})):
+ if (
+ df.get("lcm-operations-configuration", {})
+ .get("operate-vnf-op-config", {})
+ .get("day1-2")
+ ):
+ day12 = df["lcm-operations-configuration"]["operate-vnf-op-config"].get(
+ "day1-2"
+ )
+ for config_id, config in enumerate(day12):
+ for key in [
+ "initial-config-primitive",
+ "config-primitive",
+ "terminate-config-primitive",
+ ]:
+ config.pop(key, None)
+ day12[config_id] = config
+ df["lcm-operations-configuration"]["operate-vnf-op-config"][
+ "day1-2"
+ ] = day12
+ vnfd["df"][df_id] = df
+ return vnfd
+
+ def remove_modifiable_items(self, vnfd: dict) -> dict:
+ """Removes the modifiable parts from the VNFD descriptors
+
+ It calls different extract functions according to different update types
+ to clear all the modifiable items from VNFD
+
+ Args:
+ vnfd (dict): Descriptor as a dictionary
+
+ Returns:
+ vnfd (dict): Descriptor which does not include modifiable contents
+ """
+ if vnfd.get("vnfd"):
+ vnfd = vnfd["vnfd"]
+ vnfd.pop("_admin", None)
+ # If the other extractions need to be done from VNFD,
+ # the new extract methods could be appended to below list.
+ for extract_function in [self.extract_day12_primitives, self.extract_policies]:
+ vnfd_temp = extract_function(vnfd)
+ vnfd = vnfd_temp
+ return vnfd
+
+ def _validate_descriptor_changes(
+ self,
+ descriptor_id: str,
+ descriptor_file_name: str,
+ old_descriptor_directory: str,
+ new_descriptor_directory: str,
+ ):
+ """Compares the old and new VNFD descriptors and validates the new descriptor.
+
+ Args:
+ old_descriptor_directory (str): Directory of descriptor which is in-use
+ new_descriptor_directory (str): Directory of descriptor which is proposed to update (new revision)
+
+ Returns:
+ None
+
+ Raises:
+ EngineException: In case of error when there are unallowed changes
+ """
+ try:
+ # If VNFD does not exist in DB or it is not in use by any NS,
+ # validation is not required.
+ vnfd = self.db.get_one("vnfds", {"_id": descriptor_id})
+ if not vnfd or not detect_descriptor_usage(vnfd, "vnfds", self.db):
+ return
+
+ # Get the old and new descriptor contents in order to compare them.
+ with self.fs.file_open(
+ (old_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
+ ) as old_descriptor_file:
+ with self.fs.file_open(
+ (new_descriptor_directory.rstrip("/"), descriptor_file_name), "r"
+ ) as new_descriptor_file:
+ old_content = yaml.safe_load(old_descriptor_file.read())
+ new_content = yaml.safe_load(new_descriptor_file.read())
+
+ # If software version has changed, we do not need to validate
+ # the differences anymore.
+ if old_content and new_content:
+ if self.find_software_version(
+ old_content
+ ) != self.find_software_version(new_content):
+ return
+
+ disallowed_change = DeepDiff(
+ self.remove_modifiable_items(old_content),
+ self.remove_modifiable_items(new_content),
+ )
+
+ if disallowed_change:
+ changed_nodes = functools.reduce(
+ lambda a, b: a + " , " + b,
+ [
+ node.lstrip("root")
+ for node in disallowed_change.get(
+ "values_changed"
+ ).keys()
+ ],
+ )
+
+ raise EngineException(
+ f"Error in validating new descriptor: {changed_nodes} cannot be modified, "
+ "there are disallowed changes in the vnf descriptor.",
+ http_code=HTTPStatus.UNPROCESSABLE_ENTITY,
+ )
+ except (
+ DbException,
+ AttributeError,
+ IndexError,
+ KeyError,
+ ValueError,
+ ) as e:
+ raise type(e)(
+ "VNF Descriptor could not be processed with error: {}.".format(e)
+ )
+