From: tierno Date: Thu, 4 Oct 2018 16:39:49 +0000 (+0200) Subject: feature 5956. Split engine in several files X-Git-Tag: v5.0.0~35 X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FNBI.git;a=commitdiff_plain;h=b24258aa9716c1e375fde230a817f7c9faaf6c2a feature 5956. Split engine in several files Change-Id: Ic3d34ae632addd4563b94baad505b87b3cab9ec4 Signed-off-by: tierno --- diff --git a/osm_nbi/admin_topics.py b/osm_nbi/admin_topics.py new file mode 100644 index 0000000..3b5da53 --- /dev/null +++ b/osm_nbi/admin_topics.py @@ -0,0 +1,202 @@ +# -*- coding: utf-8 -*- + +# import logging +from uuid import uuid4 +from hashlib import sha256 +from http import HTTPStatus +from validation import user_new_schema, user_edit_schema, project_new_schema, project_edit_schema +from validation import vim_account_new_schema, vim_account_edit_schema, sdn_new_schema, sdn_edit_schema +from base_topic import BaseTopic, EngineException + +__author__ = "Alfonso Tierno " + + +class UserTopic(BaseTopic): + topic = "users" + topic_msg = "users" + schema_new = user_new_schema + schema_edit = user_edit_schema + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + @staticmethod + def _get_project_filter(session, write=False, show_all=True): + """ + Generates a filter dictionary for querying database users. + Current policy is admin can show all, non admin, only its own user. + :param session: contains "username", if user is "admin" and the working "project_id" + :param write: if operation is for reading (False) or writing (True) + :param show_all: if True it will show public or + :return: + """ + if session["admin"]: # allows all + return {} + else: + return {"username": session["username"]} + + def check_conflict_on_new(self, session, indata, force=False): + # check username not exists + if self.db.get_one(self.topic, {"username": indata.get("username")}, fail_on_empty=False, fail_on_more=False): + raise EngineException("username '{}' exists".format(indata["username"]), HTTPStatus.CONFLICT) + # check projects + if not force: + for p in indata["projects"]: + if p == "admin": + continue + if not self.db.get_one("projects", {"_id": p}, fail_on_empty=False, fail_on_more=False): + raise EngineException("project '{}' does not exists".format(p), HTTPStatus.CONFLICT) + + def check_conflict_on_del(self, session, _id, force=False): + if _id == session["username"]: + raise EngineException("You cannot delete your own user", http_code=HTTPStatus.CONFLICT) + + @staticmethod + def format_on_new(content, project_id=None, make_public=False): + BaseTopic.format_on_new(content, make_public=False) + content["_id"] = content["username"] + salt = uuid4().hex + content["_admin"]["salt"] = salt + if content.get("password"): + content["password"] = sha256(content["password"].encode('utf-8') + salt.encode('utf-8')).hexdigest() + + @staticmethod + def format_on_edit(final_content, edit_content): + BaseTopic.format_on_edit(final_content, edit_content) + if edit_content.get("password"): + salt = uuid4().hex + final_content["_admin"]["salt"] = salt + final_content["password"] = sha256(edit_content["password"].encode('utf-8') + + salt.encode('utf-8')).hexdigest() + + def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None): + if not session["admin"]: + raise EngineException("needed admin privileges", http_code=HTTPStatus.UNAUTHORIZED) + return BaseTopic.edit(self, session, _id, indata=indata, kwargs=kwargs, force=force, content=content) + + def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False): + if not session["admin"]: + raise EngineException("needed admin privileges", http_code=HTTPStatus.UNAUTHORIZED) + return BaseTopic.new(self, rollback, session, indata=indata, kwargs=kwargs, headers=headers, force=force, + make_public=make_public) + + +class ProjectTopic(BaseTopic): + topic = "projects" + topic_msg = "projects" + schema_new = project_new_schema + schema_edit = project_edit_schema + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + def check_conflict_on_new(self, session, indata, force=False): + if not indata.get("name"): + raise EngineException("missing 'name'") + # check name not exists + if self.db.get_one(self.topic, {"name": indata.get("name")}, fail_on_empty=False, fail_on_more=False): + raise EngineException("name '{}' exists".format(indata["name"]), HTTPStatus.CONFLICT) + + @staticmethod + def format_on_new(content, project_id=None, make_public=False): + BaseTopic.format_on_new(content, None) + content["_id"] = content["name"] + + def check_conflict_on_del(self, session, _id, force=False): + if _id == session["project_id"]: + raise EngineException("You cannot delete your own project", http_code=HTTPStatus.CONFLICT) + if force: + return + _filter = {"projects": _id} + if self.db.get_list("users", _filter): + raise EngineException("There is some USER that contains this project", http_code=HTTPStatus.CONFLICT) + + def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None): + if not session["admin"]: + raise EngineException("needed admin privileges", http_code=HTTPStatus.UNAUTHORIZED) + return BaseTopic.edit(self, session, _id, indata=indata, kwargs=kwargs, force=force, content=content) + + def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False): + if not session["admin"]: + raise EngineException("needed admin privileges", http_code=HTTPStatus.UNAUTHORIZED) + return BaseTopic.new(self, rollback, session, indata=indata, kwargs=kwargs, headers=headers, force=force, + make_public=make_public) + + +class VimAccountTopic(BaseTopic): + topic = "vim_accounts" + topic_msg = "vim_account" + schema_new = vim_account_new_schema + schema_edit = vim_account_edit_schema + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + def check_conflict_on_new(self, session, indata, force=False): + self.check_unique_name(session, indata["name"], _id=None) + + def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False): + if edit_content.get("name"): + self.check_unique_name(session, edit_content["name"], _id=_id) + + @staticmethod + def format_on_new(content, project_id=None, make_public=False): + BaseTopic.format_on_new(content, project_id=project_id, make_public=False) + content["_admin"]["operationalState"] = "PROCESSING" + + def delete(self, session, _id, force=False, dry_run=False): + """ + Delete item by its internal _id + :param session: contains the used login username, working project, and admin rights + :param _id: server internal id + :param force: indicates if deletion must be forced in case of conflict + :param dry_run: make checking but do not delete + :return: dictionary with deleted item _id. It raises EngineException on error: not found, conflict, ... + """ + # TODO add admin to filter, validate rights + if dry_run or force: # delete completely + return BaseTopic.delete(self, session, _id, force, dry_run) + else: # if not, sent to kafka + v = BaseTopic.delete(self, session, _id, force, dry_run=True) + self.db.set_one("vim_accounts", {"_id": _id}, {"_admin.to_delete": True}) # TODO change status + self._send_msg("delete", {"_id": _id}) + return v # TODO indicate an offline operation to return 202 ACCEPTED + + +class SdnTopic(BaseTopic): + topic = "sdns" + topic_msg = "sdn" + schema_new = sdn_new_schema + schema_edit = sdn_edit_schema + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + def check_conflict_on_new(self, session, indata, force=False): + self.check_unique_name(session, indata["name"], _id=None) + + def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False): + if edit_content.get("name"): + self.check_unique_name(session, edit_content["name"], _id=_id) + + @staticmethod + def format_on_new(content, project_id=None, make_public=False): + BaseTopic.format_on_new(content, project_id=project_id, make_public=False) + content["_admin"]["operationalState"] = "PROCESSING" + + def delete(self, session, _id, force=False, dry_run=False): + """ + Delete item by its internal _id + :param session: contains the used login username, working project, and admin rights + :param _id: server internal id + :param force: indicates if deletion must be forced in case of conflict + :param dry_run: make checking but do not delete + :return: dictionary with deleted item _id. It raises EngineException on error: not found, conflict, ... + """ + if dry_run or force: # delete completely + return BaseTopic.delete(self, session, _id, force, dry_run) + else: # if not sent to kafka + v = BaseTopic.delete(self, session, _id, force, dry_run=True) + self.db.set_one("sdns", {"_id": _id}, {"_admin.to_delete": True}) # TODO change status + self._send_msg("delete", {"_id": _id}) + return v # TODO indicate an offline operation to return 202 ACCEPTED diff --git a/osm_nbi/base_topic.py b/osm_nbi/base_topic.py new file mode 100644 index 0000000..2060c4c --- /dev/null +++ b/osm_nbi/base_topic.py @@ -0,0 +1,358 @@ +# -*- coding: utf-8 -*- + +import logging +from uuid import uuid4 +from http import HTTPStatus +from time import time +from osm_common.dbbase import deep_update_rfc7396 +from validation import validate_input, ValidationError + +__author__ = "Alfonso Tierno " + + +class EngineException(Exception): + + def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST): + self.http_code = http_code + Exception.__init__(self, message) + + +def get_iterable(input_var): + """ + Returns an iterable, in case input_var is None it just returns an empty tuple + :param input_var: can be a list, tuple or None + :return: input_var or () if it is None + """ + if input_var is None: + return () + return input_var + + +def versiontuple(v): + """utility for compare dot separate versions. Fills with zeros to proper number comparison""" + filled = [] + for point in v.split("."): + filled.append(point.zfill(8)) + return tuple(filled) + + +class BaseTopic: + # static variables for all instance classes + topic = None # to_override + topic_msg = None # to_override + schema_new = None # to_override + schema_edit = None # to_override + + def __init__(self, db, fs, msg): + self.db = db + self.fs = fs + self.msg = msg + self.logger = logging.getLogger("nbi.engine") + + @staticmethod + def _remove_envelop(indata=None): + if not indata: + return {} + return indata + + def _validate_input_new(self, input, force=False): + """ + Validates input user content for a new entry. It uses jsonschema. Some overrides will use pyangbind + :param input: user input content for the new topic + :param force: may be used for being more tolerant + :return: The same input content, or a changed version of it. + """ + if self.schema_new: + validate_input(input, self.schema_new) + return input + + def _validate_input_edit(self, input, force=False): + """ + Validates input user content for an edition. It uses jsonschema. Some overrides will use pyangbind + :param input: user input content for the new topic + :param force: may be used for being more tolerant + :return: The same input content, or a changed version of it. + """ + if self.schema_edit: + validate_input(input, self.schema_edit) + return input + + @staticmethod + def _get_project_filter(session, write=False, show_all=True): + """ + Generates a filter dictionary for querying database, so that only allowed items for this project can be + addressed. Only propietary or public can be used. Allowed projects are at _admin.project_read/write. If it is + not present or contains ANY mean public. + :param session: contains "username", if user is "admin" and the working "project_id" + :param write: if operation is for reading (False) or writing (True) + :param show_all: if True it will show public or + :return: + """ + if write: + k = "_admin.projects_write.cont" + else: + k = "_admin.projects_read.cont" + if not show_all: + return {k: session["project_id"]} + elif session["admin"]: # and show_all: # allows all + return {} + else: + return {k: ["ANY", session["project_id"], None]} + + def check_conflict_on_new(self, session, indata, force=False): + """ + Check that the data to be inserted is valid + :param session: contains "username", if user is "admin" and the working "project_id" + :param indata: data to be inserted + :param force: boolean. With force it is more tolerant + :return: None or raises EngineException + """ + pass + + def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False): + """ + Check that the data to be edited/uploaded is valid + :param session: contains "username", if user is "admin" and the working "project_id" + :param final_content: data once modified + :param edit_content: incremental data that contains the modifications to apply + :param _id: internal _id + :param force: boolean. With force it is more tolerant + :return: None or raises EngineException + """ + pass + + def check_unique_name(self, session, name, _id=None): + """ + Check that the name is unique for this project + :param session: contains "username", if user is "admin" and the working "project_id" + :param name: name to be checked + :param _id: If not None, ignore this entry that are going to change + :return: None or raises EngineException + """ + _filter = self._get_project_filter(session, write=False, show_all=False) + _filter["name"] = name + if _id: + _filter["_id.neq"] = _id + if self.db.get_one(self.topic, _filter, fail_on_empty=False, fail_on_more=False): + raise EngineException("name '{}' already exists for {}".format(name, self.topic), HTTPStatus.CONFLICT) + + @staticmethod + def format_on_new(content, project_id=None, make_public=False): + """ + Modifies content descriptor to include _admin + :param content: descriptor to be modified + :param project_id: if included, it add project read/write permissions + :param make_public: if included it is generated as public for reading. + :return: None, but content is modified + """ + now = time() + if "_admin" not in content: + content["_admin"] = {} + if not content["_admin"].get("created"): + content["_admin"]["created"] = now + content["_admin"]["modified"] = now + if not content.get("_id"): + content["_id"] = str(uuid4()) + if project_id: + if not content["_admin"].get("projects_read"): + content["_admin"]["projects_read"] = [project_id] + if make_public: + content["_admin"]["projects_read"].append("ANY") + if not content["_admin"].get("projects_write"): + content["_admin"]["projects_write"] = [project_id] + + @staticmethod + def format_on_edit(final_content, edit_content): + if final_content.get("_admin"): + now = time() + final_content["_admin"]["modified"] = now + + def _send_msg(self, action, content): + if self.topic_msg: + content.pop("_admin", None) + self.msg.write(self.topic_msg, action, content) + + def check_conflict_on_del(self, session, _id, force=False): + """ + Check if deletion can be done because of dependencies if it is not force. To override + :param session: contains "username", if user is "admin" and the working "project_id" + :param _id: itnernal _id + :param force: Avoid this checking + :return: None if ok or raises EngineException with the conflict + """ + pass + + @staticmethod + def _update_input_with_kwargs(desc, kwargs): + """ + Update descriptor with the kwargs. It contains dot separated keys + :param desc: dictionary to be updated + :param kwargs: plain dictionary to be used for updating. + :return: None, 'desc' is modified. It raises EngineException. + """ + if not kwargs: + return + try: + for k, v in kwargs.items(): + update_content = desc + kitem_old = None + klist = k.split(".") + for kitem in klist: + if kitem_old is not None: + update_content = update_content[kitem_old] + if isinstance(update_content, dict): + kitem_old = kitem + elif isinstance(update_content, list): + kitem_old = int(kitem) + else: + raise EngineException( + "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem)) + update_content[kitem_old] = v + except KeyError: + raise EngineException( + "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old)) + except ValueError: + raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format( + k, kitem)) + except IndexError: + raise EngineException( + "Invalid query string '{}'. Index '{}' out of range".format(k, kitem_old)) + + def show(self, session, _id): + """ + Get complete information on an topic + :param session: contains the used login username and working project + :param _id: server internal id + :return: dictionary, raise exception if not found. + """ + filter_db = self._get_project_filter(session, write=False, show_all=True) + filter_db["_id"] = _id + return self.db.get_one(self.topic, filter_db) + # TODO transform data for SOL005 URL requests + # TODO remove _admin if not admin + + def get_file(self, session, _id, path=None, accept_header=None): + """ + Only implemented for descriptor topics. Return the file content of a descriptor + :param session: contains the used login username and working project + :param _id: Identity of the item to get content + :param path: artifact path or "$DESCRIPTOR" or None + :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain + :return: opened file or raises an exception + """ + raise EngineException("Method get_file not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR) + + def list(self, session, filter_q=None): + """ + Get a list of the topic that matches a filter + :param session: contains the used login username and working project + :param filter_q: filter of data to be applied + :return: The list, it can be empty if no one match the filter. + """ + if not filter_q: + filter_q = {} + + filter_q.update(self._get_project_filter(session, write=False, show_all=True)) + + # TODO transform data for SOL005 URL requests. Transform filtering + # TODO implement "field-type" query string SOL005 + return self.db.get_list(self.topic, filter_q) + + def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False): + """ + Creates a new entry into database. + :param rollback: list to append created items at database in case a rollback may to be done + :param session: contains the used login username and working project + :param indata: data to be inserted + :param kwargs: used to override the indata descriptor + :param headers: http request headers + :param force: If True avoid some dependence checks + :param make_public: Make the created item public to all projects + :return: _id: identity of the inserted data. + """ + try: + content = self._remove_envelop(indata) + + # Override descriptor with query string kwargs + self._update_input_with_kwargs(content, kwargs) + content = self._validate_input_new(content, force=force) + self.check_conflict_on_new(session, content, force=force) + self.format_on_new(content, project_id=session["project_id"], make_public=make_public) + _id = self.db.create(self.topic, content) + rollback.append({"topic": self.topic, "_id": _id}) + self._send_msg("create", content) + return _id + except ValidationError as e: + raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY) + + def upload_content(self, session, _id, indata, kwargs, headers, force=False): + """ + Only implemented for descriptor topics. Used for receiving content by chunks (with a transaction_id header + and/or gzip file. It will store and extract) + :param session: session + :param _id : the database id of entry to be updated + :param indata: http body request + :param kwargs: user query string to override parameters. NOT USED + :param headers: http request headers + :param force: to be more tolerant with validation + :return: True package has is completely uploaded or False if partial content has been uplodaed. + Raise exception on error + """ + raise EngineException("Method upload_content not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR) + + def delete_list(self, session, filter_q=None): + """ + Delete a several entries of a topic. This is for internal usage and test only, not exposed to NBI API + :param session: contains the used login username and working project + :param filter_q: filter of data to be applied + :return: The deleted list, it can be empty if no one match the filter. + """ + # TODO add admin to filter, validate rights + if not filter_q: + filter_q = {} + filter_q.update(self._get_project_filter(session, write=True, show_all=True)) + return self.db.del_list(self.topic, filter_q) + + def delete(self, session, _id, force=False, dry_run=False): + """ + Delete item by its internal _id + :param session: contains the used login username, working project, and admin rights + :param _id: server internal id + :param force: indicates if deletion must be forced in case of conflict + :param dry_run: make checking but do not delete + :return: dictionary with deleted item _id. It raises EngineException on error: not found, conflict, ... + """ + # TODO add admin to filter, validate rights + # data = self.get_item(topic, _id) + self.check_conflict_on_del(session, _id, force) + filter_q = self._get_project_filter(session, write=True, show_all=True) + filter_q["_id"] = _id + if not dry_run: + v = self.db.del_one(self.topic, filter_q) + self._send_msg("deleted", {"_id": _id}) + return v + return None + + def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None): + indata = self._remove_envelop(indata) + + # Override descriptor with query string kwargs + if kwargs: + self._update_input_with_kwargs(indata, kwargs) + try: + indata = self._validate_input_edit(indata, force=force) + + # TODO self._check_edition(session, indata, _id, force) + if not content: + content = self.show(session, _id) + deep_update_rfc7396(content, indata) + self.check_conflict_on_edit(session, content, indata, _id=_id, force=force) + self.format_on_edit(content, indata) + self.db.replace(self.topic, _id, content) + + indata.pop("_admin", None) + indata["_id"] = _id + self._send_msg("edit", indata) + return id + except ValidationError as e: + raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY) diff --git a/osm_nbi/descriptor_topics.py b/osm_nbi/descriptor_topics.py new file mode 100644 index 0000000..b59b4f8 --- /dev/null +++ b/osm_nbi/descriptor_topics.py @@ -0,0 +1,493 @@ +# -*- coding: utf-8 -*- + +import tarfile +import yaml +import json +# import logging +from hashlib import md5 +from osm_common.dbbase import DbException, deep_update_rfc7396 +from http import HTTPStatus +from validation import ValidationError, pdu_new_schema, pdu_edit_schema +from base_topic import BaseTopic, EngineException + +__author__ = "Alfonso Tierno " + + +class DescriptorTopic(BaseTopic): + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False): + # check that this id is not present + _filter = {"id": final_content["id"]} + if _id: + _filter["_id.neq"] = _id + + _filter.update(self._get_project_filter(session, write=False, show_all=False)) + if self.db.get_one(self.topic, _filter, fail_on_empty=False): + raise EngineException("{} with id '{}' already exists for this project".format(self.topic[:-1], + final_content["id"]), + HTTPStatus.CONFLICT) + # TODO validate with pyangbind. Load and dumps to convert data types + + @staticmethod + def format_on_new(content, project_id=None, make_public=False): + BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public) + content["_admin"]["onboardingState"] = "CREATED" + content["_admin"]["operationalState"] = "DISABLED" + content["_admin"]["usageSate"] = "NOT_IN_USE" + + def delete(self, session, _id, force=False, dry_run=False): + """ + Delete item by its internal _id + :param session: contains the used login username, working project, and admin rights + :param _id: server internal id + :param force: indicates if deletion must be forced in case of conflict + :param dry_run: make checking but do not delete + :return: dictionary with deleted item _id. It raises EngineException on error: not found, conflict, ... + """ + # TODO add admin to filter, validate rights + v = BaseTopic.delete(self, session, _id, force, dry_run=True) + if dry_run: + return + v = self.db.del_one(self.topic, {"_id": _id}) + self.fs.file_delete(_id, ignore_non_exist=True) + self._send_msg("delete", {"_id": _id}) + return v + + @staticmethod + def get_one_by_id(db, session, topic, id): + # find owned by this project + _filter = BaseTopic._get_project_filter(session, write=False, show_all=False) + _filter["id"] = id + desc_list = db.get_list(topic, _filter) + if len(desc_list) == 1: + return desc_list[0] + elif len(desc_list) > 1: + raise DbException("Found more than one {} with id='{}' belonging to this project".format(topic[:-1], id), + HTTPStatus.CONFLICT) + + # not found any: try to find public + _filter = BaseTopic._get_project_filter(session, write=False, show_all=True) + _filter["id"] = id + desc_list = db.get_list(topic, _filter) + if not desc_list: + raise DbException("Not found any {} with id='{}'".format(topic[:-1], id), HTTPStatus.NOT_FOUND) + elif len(desc_list) == 1: + return desc_list[0] + else: + raise DbException("Found more than one public {} with id='{}'; and no one belonging to this project".format( + topic[:-1], id), HTTPStatus.CONFLICT) + + def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False): + """ + Creates a new almost empty DISABLED entry into database. Due to SOL005, it does not follow normal procedure. + Creating a VNFD or NSD is done in two steps: 1. Creates an empty descriptor (this step) and 2) upload content + (self.upload_content) + :param rollback: list to append created items at database in case a rollback may to be done + :param session: contains the used login username and working project + :param indata: data to be inserted + :param kwargs: used to override the indata descriptor + :param headers: http request headers + :param force: If True avoid some dependence checks + :param make_public: Make the created descriptor public to all projects + :return: _id: identity of the inserted data. + """ + + try: + # _remove_envelop + if indata: + if "userDefinedData" in indata: + indata = indata['userDefinedData'] + + # Override descriptor with query string kwargs + self._update_input_with_kwargs(indata, kwargs) + # uncomment when this method is implemented. + # Avoid override in this case as the target is userDefinedData, but not vnfd,nsd descriptors + # indata = DescriptorTopic._validate_input_new(self, indata, force=force) + + content = {"_admin": {"userDefinedData": indata}} + self.format_on_new(content, session["project_id"], make_public=make_public) + _id = self.db.create(self.topic, content) + rollback.append({"topic": self.topic, "_id": _id}) + return _id + except ValidationError as e: + raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY) + + def upload_content(self, session, _id, indata, kwargs, headers, force=False): + """ + Used for receiving content by chunks (with a transaction_id header and/or gzip file. It will store and extract) + :param session: session + :param _id : the nsd,vnfd is already created, this is the id + :param indata: http body request + :param kwargs: user query string to override parameters. NOT USED + :param headers: http request headers + :param force: to be more tolerant with validation + :return: True package has is completely uploaded or False if partial content has been uplodaed. + Raise exception on error + """ + # Check that _id exists and it is valid + current_desc = self.show(session, _id) + + content_range_text = headers.get("Content-Range") + expected_md5 = headers.get("Content-File-MD5") + compressed = None + content_type = headers.get("Content-Type") + if content_type and "application/gzip" in content_type or "application/x-gzip" in content_type or \ + "application/zip" in content_type: + compressed = "gzip" + filename = headers.get("Content-Filename") + if not filename: + filename = "package.tar.gz" if compressed else "package" + # TODO change to Content-Disposition filename https://tools.ietf.org/html/rfc6266 + file_pkg = None + error_text = "" + try: + if content_range_text: + content_range = content_range_text.replace("-", " ").replace("/", " ").split() + if content_range[0] != "bytes": # TODO check x result + # manyfiles yes X -> zip + # no yes -> error + # onefile yes no -> zip + # X yes -> text + + if accept_text and (not storage.get('pkg-dir') or path == "$DESCRIPTOR"): + return self.fs.file_open((storage['folder'], storage['descriptor']), "r"), "text/plain" + elif storage.get('pkg-dir') and not accept_zip: + raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'" + "Accept header", http_code=HTTPStatus.NOT_ACCEPTABLE) + else: + if not storage.get('zipfile'): + # TODO generate zipfile if not present + raise EngineException("Only allowed 'text/plain' Accept header for this descriptor. To be solved in " + "future versions", http_code=HTTPStatus.NOT_ACCEPTABLE) + return self.fs.file_open((storage['folder'], storage['zipfile']), "rb"), "application/zip" + + +class VnfdTopic(DescriptorTopic): + topic = "vnfds" + topic_msg = "vnfd" + + def __init__(self, db, fs, msg): + DescriptorTopic.__init__(self, db, fs, msg) + + @staticmethod + def _remove_envelop(indata=None): + if not indata: + return {} + clean_indata = indata + if clean_indata.get('vnfd:vnfd-catalog'): + clean_indata = clean_indata['vnfd:vnfd-catalog'] + elif clean_indata.get('vnfd-catalog'): + clean_indata = clean_indata['vnfd-catalog'] + if clean_indata.get('vnfd'): + if not isinstance(clean_indata['vnfd'], list) or len(clean_indata['vnfd']) != 1: + raise EngineException("'vnfd' must be a list only one element") + clean_indata = clean_indata['vnfd'][0] + return clean_indata + + def check_conflict_on_del(self, session, _id, force=False): + """ + Check that there is not any NSD that uses this VNFD. Only NSDs belonging to this project are considered. Note + that VNFD can be public and be used by NSD of other projects. Also check there are not deployments, or vnfr + that uses this vnfd + :param session: + :param _id: vnfd inernal id + :param force: Avoid this checking + :return: None or raises EngineException with the conflict + """ + if force: + return + descriptor = self.db.get_one("vnfds", {"_id": _id}) + descriptor_id = descriptor.get("id") + if not descriptor_id: # empty vnfd not uploaded + return + + _filter = self._get_project_filter(session, write=False, show_all=False) + # check vnfrs using this vnfd + _filter["vnfd-id"] = _id + if self.db.get_list("vnfrs", _filter): + raise EngineException("There is some VNFR that depends on this VNFD", http_code=HTTPStatus.CONFLICT) + del _filter["vnfd-id"] + # check NSD using this VNFD + _filter["constituent-vnfd.ANYINDEX.vnfd-id-ref"] = descriptor_id + if self.db.get_list("nsds", _filter): + raise EngineException("There is soame NSD that depends on this VNFD", http_code=HTTPStatus.CONFLICT) + + def _validate_input_new(self, indata, force=False): + # TODO validate with pyangbind, serialize + return indata + + def _validate_input_edit(self, indata, force=False): + # TODO validate with pyangbind, serialize + return indata + + +class NsdTopic(DescriptorTopic): + topic = "nsds" + topic_msg = "nsd" + + def __init__(self, db, fs, msg): + DescriptorTopic.__init__(self, db, fs, msg) + + @staticmethod + def _remove_envelop(indata=None): + if not indata: + return {} + clean_indata = indata + + if clean_indata.get('nsd:nsd-catalog'): + clean_indata = clean_indata['nsd:nsd-catalog'] + elif clean_indata.get('nsd-catalog'): + clean_indata = clean_indata['nsd-catalog'] + if clean_indata.get('nsd'): + if not isinstance(clean_indata['nsd'], list) or len(clean_indata['nsd']) != 1: + raise EngineException("'nsd' must be a list only one element") + clean_indata = clean_indata['nsd'][0] + return clean_indata + + def _validate_input_new(self, indata, force=False): + # transform constituent-vnfd:member-vnf-index to string + if indata.get("constituent-vnfd"): + for constituent_vnfd in indata["constituent-vnfd"]: + if "member-vnf-index" in constituent_vnfd: + constituent_vnfd["member-vnf-index"] = str(constituent_vnfd["member-vnf-index"]) + + # TODO validate with pyangbind, serialize + return indata + + def _validate_input_edit(self, indata, force=False): + # TODO validate with pyangbind, serialize + return indata + + def _check_descriptor_dependencies(self, session, descriptor): + """ + Check that the dependent descriptors exist on a new descriptor or edition + :param session: client session information + :param descriptor: descriptor to be inserted or edit + :return: None or raises exception + """ + if not descriptor.get("constituent-vnfd"): + return + for vnf in descriptor["constituent-vnfd"]: + vnfd_id = vnf["vnfd-id-ref"] + filter_q = self._get_project_filter(session, write=False, show_all=True) + filter_q["id"] = vnfd_id + if not self.db.get_list("vnfds", filter_q): + raise EngineException("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non " + "existing vnfd".format(vnfd_id), http_code=HTTPStatus.CONFLICT) + + def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False): + super().check_conflict_on_edit(session, final_content, edit_content, _id, force=force) + + self._check_descriptor_dependencies(session, final_content) + + def check_conflict_on_del(self, session, _id, force=False): + """ + Check that there is not any NSR that uses this NSD. Only NSRs belonging to this project are considered. Note + that NSD can be public and be used by other projects. + :param session: + :param _id: vnfd inernal id + :param force: Avoid this checking + :return: None or raises EngineException with the conflict + """ + if force: + return + _filter = self._get_project_filter(session, write=False, show_all=False) + _filter["nsdId"] = _id + if self.db.get_list("nsrs", _filter): + raise EngineException("There is some NSR that depends on this NSD", http_code=HTTPStatus.CONFLICT) + + +class PduTopic(BaseTopic): + topic = "pdus" + topic_msg = "pdu" + schema_new = pdu_new_schema + schema_edit = pdu_edit_schema + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + @staticmethod + def format_on_new(content, project_id=None, make_public=False): + BaseTopic.format_on_new(content, project_id=None, make_public=make_public) + content["_admin"]["onboardingState"] = "CREATED" + content["_admin"]["operationalState"] = "DISABLED" + content["_admin"]["usageSate"] = "NOT_IN_USE" + + def check_conflict_on_del(self, session, _id, force=False): + if force: + return + # TODO Is it needed to check descriptors _admin.project_read/project_write?? + _filter = {"vdur.pdu-id": _id} + if self.db.get_list("vnfrs", _filter): + raise EngineException("There is some NSR that uses this PDU", http_code=HTTPStatus.CONFLICT) diff --git a/osm_nbi/engine.py b/osm_nbi/engine.py index 6df84c8..ab7eec0 100644 --- a/osm_nbi/engine.py +++ b/osm_nbi/engine.py @@ -1,54 +1,42 @@ # -*- coding: utf-8 -*- -from osm_common import dbmongo -from osm_common import dbmemory -from osm_common import fslocal -from osm_common import msglocal -from osm_common import msgkafka -import tarfile -import yaml -import json import logging -from uuid import uuid4 -from hashlib import sha256, md5 -from osm_common.dbbase import DbException, deep_update +from osm_common import dbmongo, dbmemory, fslocal, msglocal, msgkafka, version as common_version +from osm_common.dbbase import DbException from osm_common.fsbase import FsException from osm_common.msgbase import MsgException from http import HTTPStatus -from time import time -from copy import copy -from validation import validate_input, ValidationError +from base_topic import EngineException, versiontuple +from admin_topics import UserTopic, ProjectTopic, VimAccountTopic, SdnTopic +from descriptor_topics import VnfdTopic, NsdTopic, PduTopic +from instance_topics import NsrTopic, VnfrTopic, NsLcmOpTopic __author__ = "Alfonso Tierno " - - -class EngineException(Exception): - - def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST): - self.http_code = http_code - Exception.__init__(self, message) - - -def get_iterable(input_var): - """ - Returns an iterable, in case input_var is None it just returns an empty tuple - :param input_var: can be a list, tuple or None - :return: input_var or () if it is None - """ - if input_var is None: - return () - return input_var +min_common_version = "0.1.8" class Engine(object): + map_from_topic_to_class = { + "vnfds": VnfdTopic, + "nsds": NsdTopic, + "pdus": PduTopic, + "nsrs": NsrTopic, + "vnfrs": VnfrTopic, + "nslcmops": NsLcmOpTopic, + "vim_accounts": VimAccountTopic, + "sdns": SdnTopic, + "users": UserTopic, + "projects": ProjectTopic, + # [NEW_TOPIC]: add an entry here + } def __init__(self): - self.tokens = {} self.db = None self.fs = None self.msg = None self.config = None self.logger = logging.getLogger("nbi.engine") + self.map_topic = {} def start(self, config): """ @@ -57,6 +45,11 @@ class Engine(object): :return: None """ self.config = config + # check right version of common + if versiontuple(common_version) < versiontuple(min_common_version): + raise EngineException("Not compatible osm/common version '{}'. Needed '{}' or higher".format( + common_version, min_common_version)) + try: if not self.db: if config["database"]["driver"] == "mongo": @@ -85,6 +78,10 @@ class Engine(object): else: raise EngineException("Invalid configuration param '{}' at '[message]':'driver'".format( config["storage"]["driver"])) + + # create one class per topic + for topic, topic_class in self.map_from_topic_to_class.items(): + self.map_topic[topic] = topic_class(self.db, self.fs, self.msg) except (DbException, FsException, MsgException) as e: raise EngineException(str(e), http_code=e.http_code) @@ -99,916 +96,102 @@ class Engine(object): except (DbException, FsException, MsgException) as e: raise EngineException(str(e), http_code=e.http_code) - @staticmethod - def _remove_envelop(item, indata=None): - """ - Obtain the useful data removing the envelop. It goes throw the vnfd or nsd catalog and returns the - vnfd or nsd content - :param item: can be vnfds, nsds, users, projects, userDefinedData (initial content of a vnfds, nsds - :param indata: Content to be inspected - :return: the useful part of indata - """ - clean_indata = indata - if not indata: - return {} - if item == "vnfds": - if clean_indata.get('vnfd:vnfd-catalog'): - clean_indata = clean_indata['vnfd:vnfd-catalog'] - elif clean_indata.get('vnfd-catalog'): - clean_indata = clean_indata['vnfd-catalog'] - if clean_indata.get('vnfd'): - if not isinstance(clean_indata['vnfd'], list) or len(clean_indata['vnfd']) != 1: - raise EngineException("'vnfd' must be a list only one element") - clean_indata = clean_indata['vnfd'][0] - elif item == "nsds": - if clean_indata.get('nsd:nsd-catalog'): - clean_indata = clean_indata['nsd:nsd-catalog'] - elif clean_indata.get('nsd-catalog'): - clean_indata = clean_indata['nsd-catalog'] - if clean_indata.get('nsd'): - if not isinstance(clean_indata['nsd'], list) or len(clean_indata['nsd']) != 1: - raise EngineException("'nsd' must be a list only one element") - clean_indata = clean_indata['nsd'][0] - elif item == "userDefinedData": - if "userDefinedData" in indata: - clean_indata = clean_indata['userDefinedData'] - return clean_indata - - def _check_project_dependencies(self, project_id): - """ - Check if a project can be deleted - :param project_id: - :return: - """ - # TODO Is it needed to check descriptors _admin.project_read/project_write?? - _filter = {"projects": project_id} - if self.db.get_list("users", _filter): - raise EngineException("There are users that uses this project", http_code=HTTPStatus.CONFLICT) - - def _check_pdus_usage(self, pdu_id): - """ - Check if a pdu can be deleted - :param pdu_id: - :return: - """ - # TODO Is it needed to check descriptors _admin.project_read/project_write?? - _filter = {"vdur.pdu-id": pdu_id} - if self.db.get_list("vnfrs", _filter): - raise EngineException("There are ns that uses this pdu", http_code=HTTPStatus.CONFLICT) - - def _check_dependencies_on_descriptor(self, session, item, descriptor_id, _id): - """ - Check that the descriptor to be deleded is not a dependency of others - :param session: client session information - :param item: can be vnfds, nsds - :param descriptor_id: id (provided by client) of descriptor to be deleted - :param _id: internal id of descriptor to be deleted - :return: None or raises exception - """ - if item == "vnfds": - _filter = {"constituent-vnfd.ANYINDEX.vnfd-id-ref": descriptor_id} - if self.get_item_list(session, "nsds", _filter): - raise EngineException("There are nsd that depends on this VNFD", http_code=HTTPStatus.CONFLICT) - if self.get_item_list(session, "vnfrs", {"vnfd-id": _id}): - raise EngineException("There are vnfr that depends on this VNFD", http_code=HTTPStatus.CONFLICT) - elif item == "nsds": - _filter = {"nsdId": _id} - if self.get_item_list(session, "nsrs", _filter): - raise EngineException("There are nsr that depends on this NSD", http_code=HTTPStatus.CONFLICT) - - def _check_descriptor_dependencies(self, session, item, descriptor): - """ - Check that the dependent descriptors exist on a new descriptor or edition - :param session: client session information - :param item: can be nsds, nsrs - :param descriptor: descriptor to be inserted or edit - :return: None or raises exception - """ - if item == "nsds": - if not descriptor.get("constituent-vnfd"): - return - for vnf in descriptor["constituent-vnfd"]: - vnfd_id = vnf["vnfd-id-ref"] - if not self.get_item_list(session, "vnfds", {"id": vnfd_id}): - raise EngineException("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non " - "existing vnfd".format(vnfd_id), http_code=HTTPStatus.CONFLICT) - elif item == "nsrs": - if not descriptor.get("nsdId"): - return - nsd_id = descriptor["nsdId"] - if not self.get_item_list(session, "nsds", {"id": nsd_id}): - raise EngineException("Descriptor error at nsdId='{}' references a non exist nsd".format(nsd_id), - http_code=HTTPStatus.CONFLICT) - - def _check_edition(self, session, item, indata, id, force=False): - if item == "users": - if indata.get("projects"): - if not session["admin"]: - raise EngineException("Needed admin privileges to edit user projects", HTTPStatus.UNAUTHORIZED) - if indata.get("password"): - # regenerate salt and encrypt password - salt = uuid4().hex - indata["_admin"] = {"salt": salt} - indata["password"] = sha256(indata["password"].encode('utf-8') + salt.encode('utf-8')).hexdigest() - - def _validate_new_data(self, session, item, indata, id=None, force=False): - if item == "users": - # check username not exists - if not id and self.db.get_one(item, {"username": indata.get("username")}, fail_on_empty=False, - fail_on_more=False): - raise EngineException("username '{}' exists".format(indata["username"]), HTTPStatus.CONFLICT) - # check projects - if not force: - for p in indata["projects"]: - if p == "admin": - continue - if not self.db.get_one("projects", {"_id": p}, fail_on_empty=False, fail_on_more=False): - raise EngineException("project '{}' does not exists".format(p), HTTPStatus.CONFLICT) - elif item == "projects": - if not indata.get("name"): - raise EngineException("missing 'name'") - # check name not exists - if not id and self.db.get_one(item, {"name": indata.get("name")}, fail_on_empty=False, fail_on_more=False): - raise EngineException("name '{}' exists".format(indata["name"]), HTTPStatus.CONFLICT) - elif item in ("vnfds", "nsds"): - filter = {"id": indata["id"]} - if id: - filter["_id.neq"] = id - # TODO add admin to filter, validate rights - self._add_read_filter(session, item, filter) - if self.db.get_one(item, filter, fail_on_empty=False): - raise EngineException("{} with id '{}' already exists for this tenant".format(item[:-1], indata["id"]), - HTTPStatus.CONFLICT) - # TODO validate with pyangbind. Load and dumps to convert data types - if item == "nsds": - # transform constituent-vnfd:member-vnf-index to string - if indata.get("constituent-vnfd"): - for constituent_vnfd in indata["constituent-vnfd"]: - if "member-vnf-index" in constituent_vnfd: - constituent_vnfd["member-vnf-index"] = str(constituent_vnfd["member-vnf-index"]) - - if item == "nsds" and not force: - self._check_descriptor_dependencies(session, "nsds", indata) - elif item == "userDefinedData": - # TODO validate userDefinedData is a keypair values - pass - - elif item == "nsrs": - pass - elif item == "vim_accounts" or item == "sdns": - filter = {"name": indata.get("name")} - if id: - filter["_id.neq"] = id - if self.db.get_one(item, filter, fail_on_empty=False, fail_on_more=False): - raise EngineException("name '{}' already exists for {}".format(indata["name"], item), - HTTPStatus.CONFLICT) - - def _check_ns_operation(self, session, nsr, operation, indata): - """ - Check that user has enter right parameters for the operation - :param session: - :param operation: it can be: instantiate, terminate, action, TODO: update, heal - :param indata: descriptor with the parameters of the operation - :return: None - """ - vnfds = {} - vim_accounts = [] - nsd = nsr["nsd"] - - def check_valid_vnf_member_index(member_vnf_index): - for vnf in nsd["constituent-vnfd"]: - if member_vnf_index == vnf["member-vnf-index"]: - vnfd_id = vnf["vnfd-id-ref"] - if vnfd_id not in vnfds: - vnfds[vnfd_id] = self.db.get_one("vnfds", {"id": vnfd_id}) - return vnfds[vnfd_id] - else: - raise EngineException("Invalid parameter member_vnf_index='{}' is not one of the " - "nsd:constituent-vnfd".format(member_vnf_index)) - - def check_valid_vim_account(vim_account): - if vim_account in vim_accounts: - return - try: - self.db.get_one("vim_accounts", {"_id": vim_account}) - except Exception: - raise EngineException("Invalid vimAccountId='{}' not present".format(vim_account)) - vim_accounts.append(vim_account) - - if operation == "action": - # check vnf_member_index - if indata.get("vnf_member_index"): - indata["member_vnf_index"] = indata.pop("vnf_member_index") # for backward compatibility - if not indata.get("member_vnf_index"): - raise EngineException("Missing 'member_vnf_index' parameter") - vnfd = check_valid_vnf_member_index(indata["member_vnf_index"]) - # check primitive - for config_primitive in get_iterable(vnfd.get("vnf-configuration", {}).get("config-primitive")): - if indata["primitive"] == config_primitive["name"]: - # check needed primitive_params are provided - if indata.get("primitive_params"): - in_primitive_params_copy = copy(indata["primitive_params"]) - else: - in_primitive_params_copy = {} - for paramd in get_iterable(config_primitive.get("parameter")): - if paramd["name"] in in_primitive_params_copy: - del in_primitive_params_copy[paramd["name"]] - elif not paramd.get("default-value"): - raise EngineException("Needed parameter {} not provided for primitive '{}'".format( - paramd["name"], indata["primitive"])) - # check no extra primitive params are provided - if in_primitive_params_copy: - raise EngineException("parameter/s '{}' not present at vnfd for primitive '{}'".format( - list(in_primitive_params_copy.keys()), indata["primitive"])) - break - else: - raise EngineException("Invalid primitive '{}' is not present at vnfd".format(indata["primitive"])) - if operation == "scale": - vnfd = check_valid_vnf_member_index(indata["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]) - for scaling_group in get_iterable(vnfd.get("scaling-group-descriptor")): - if indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"] == scaling_group["name"]: - break - else: - raise EngineException("Invalid scaleVnfData:scaleByStepData:scaling-group-descriptor '{}' is not " - "present at vnfd:scaling-group-descriptor".format( - indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"])) - if operation == "instantiate": - # check vim_account - check_valid_vim_account(indata["vimAccountId"]) - for in_vnf in get_iterable(indata.get("vnf")): - vnfd = check_valid_vnf_member_index(in_vnf["member-vnf-index"]) - if in_vnf.get("vimAccountId"): - check_valid_vim_account(in_vnf["vimAccountId"]) - for in_vdu in get_iterable(in_vnf.get("vdu")): - for vdud in get_iterable(vnfd.get("vdu")): - if vdud["id"] == in_vdu["id"]: - for volume in get_iterable(in_vdu.get("volume")): - for volumed in get_iterable(vdud.get("volumes")): - if volumed["name"] == volume["name"]: - break - else: - raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu[id='{}']:" - "volume:name='{}' is not present at vnfd:vdu:volumes list". - format(in_vnf["member-vnf-index"], in_vdu["id"], - volume["name"])) - break - else: - raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu:id='{}' is not " - "present at vnfd".format(in_vnf["member-vnf-index"], in_vdu["id"])) - - for in_internal_vld in get_iterable(in_vnf.get("internal-vld")): - for internal_vldd in get_iterable(vnfd.get("internal-vld")): - if in_internal_vld["name"] == internal_vldd["name"] or \ - in_internal_vld["name"] == internal_vldd["id"]: - break - else: - raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:internal-vld:name='{}'" - " is not present at vnfd '{}'".format(in_vnf["member-vnf-index"], - in_internal_vld["name"], - vnfd["id"])) - for in_vld in get_iterable(indata.get("vld")): - for vldd in get_iterable(nsd.get("vld")): - if in_vld["name"] == vldd["name"] or in_vld["name"] == vldd["id"]: - break - else: - raise EngineException("Invalid parameter vld:name='{}' is not present at nsd:vld".format( - in_vld["name"])) - - def _format_new_data(self, session, item, indata): - now = time() - if "_admin" not in indata: - indata["_admin"] = {} - indata["_admin"]["created"] = now - indata["_admin"]["modified"] = now - if item == "users": - indata["_id"] = indata["username"] - salt = uuid4().hex - indata["_admin"]["salt"] = salt - indata["password"] = sha256(indata["password"].encode('utf-8') + salt.encode('utf-8')).hexdigest() - elif item == "projects": - indata["_id"] = indata["name"] - else: - if not indata.get("_id"): - indata["_id"] = str(uuid4()) - if item in ("vnfds", "nsds", "nsrs", "vnfrs", "pdus"): - if not indata["_admin"].get("projects_read"): - indata["_admin"]["projects_read"] = [session["project_id"]] - if not indata["_admin"].get("projects_write"): - indata["_admin"]["projects_write"] = [session["project_id"]] - if item in ("vnfds", "nsds", "pdus"): - indata["_admin"]["onboardingState"] = "CREATED" - indata["_admin"]["operationalState"] = "DISABLED" - indata["_admin"]["usageSate"] = "NOT_IN_USE" - if item == "nsrs": - indata["_admin"]["nsState"] = "NOT_INSTANTIATED" - if item in ("vim_accounts", "sdns"): - indata["_admin"]["operationalState"] = "PROCESSING" - - def upload_content(self, session, item, _id, indata, kwargs, headers): - """ - Used for receiving content by chunks (with a transaction_id header and/or gzip file. It will store and extract) - :param session: session - :param item: can be nsds or vnfds - :param _id : the nsd,vnfd is already created, this is the id - :param indata: http body request - :param kwargs: user query string to override parameters. NOT USED - :param headers: http request headers - :return: True package has is completely uploaded or False if partial content has been uplodaed. - Raise exception on error - """ - # Check that _id exists and it is valid - current_desc = self.get_item(session, item, _id) - - content_range_text = headers.get("Content-Range") - expected_md5 = headers.get("Content-File-MD5") - compressed = None - content_type = headers.get("Content-Type") - if content_type and "application/gzip" in content_type or "application/x-gzip" in content_type or \ - "application/zip" in content_type: - compressed = "gzip" - filename = headers.get("Content-Filename") - if not filename: - filename = "package.tar.gz" if compressed else "package" - # TODO change to Content-Disposition filename https://tools.ietf.org/html/rfc6266 - file_pkg = None - error_text = "" - try: - if content_range_text: - content_range = content_range_text.replace("-", " ").replace("/", " ").split() - if content_range[0] != "bytes": # TODO check x result - # manyfiles yes X -> zip - # no yes -> error - # onefile yes no -> zip - # X yes -> text - - if accept_text and (not storage.get('pkg-dir') or path == "$DESCRIPTOR"): - return self.fs.file_open((storage['folder'], storage['descriptor']), "r"), "text/plain" - elif storage.get('pkg-dir') and not accept_zip: - raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'" - "Accept header", http_code=HTTPStatus.NOT_ACCEPTABLE) - else: - if not storage.get('zipfile'): - # TODO generate zipfile if not present - raise EngineException("Only allowed 'text/plain' Accept header for this descriptor. To be solved in " - "future versions", http_code=HTTPStatus.NOT_ACCEPTABLE) - return self.fs.file_open((storage['folder'], storage['zipfile']), "rb"), "application/zip" - - def get_item_list(self, session, item, filter={}): + def get_item_list(self, session, topic, filter_q=None): """ Get a list of items :param session: contains the used login username and working project - :param item: it can be: users, projects, vnfds, nsds, ... - :param filter: filter of data to be applied - :return: The list, it can be empty if no one match the filter. + :param topic: it can be: users, projects, vnfds, nsds, ... + :param filter_q: filter of data to be applied + :return: The list, it can be empty if no one match the filter_q. """ - # TODO add admin to filter, validate rights - # TODO transform data for SOL005 URL requests. Transform filtering - # TODO implement "field-type" query string SOL005 - - self._add_read_filter(session, item, filter) - return self.db.get_list(item, filter) + if topic not in self.map_topic: + raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) + return self.map_topic[topic].list(session, filter_q) - def get_item(self, session, item, _id): + def get_item(self, session, topic, _id): """ - Get complete information on an items + Get complete information on an item :param session: contains the used login username and working project - :param item: it can be: users, projects, vnfds, nsds, + :param topic: it can be: users, projects, vnfds, nsds, :param _id: server id of the item :return: dictionary, raise exception if not found. """ - filter = {"_id": _id} - # TODO add admin to filter, validate rights - # TODO transform data for SOL005 URL requests - self._add_read_filter(session, item, filter) - return self.db.get_one(item, filter) + if topic not in self.map_topic: + raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) + return self.map_topic[topic].show(session, _id) - def del_item_list(self, session, item, filter={}): + def del_item_list(self, session, topic, _filter=None): """ Delete a list of items :param session: contains the used login username and working project - :param item: it can be: users, projects, vnfds, nsds, ... - :param filter: filter of data to be applied - :return: The deleted list, it can be empty if no one match the filter. + :param topic: it can be: users, projects, vnfds, nsds, ... + :param _filter: filter of data to be applied + :return: The deleted list, it can be empty if no one match the _filter. """ - # TODO add admin to filter, validate rights - self._add_read_filter(session, item, filter) - return self.db.del_list(item, filter) + if topic not in self.map_topic: + raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) + return self.map_topic[topic].delete_list(session, _filter) - def del_item(self, session, item, _id, force=False): + def del_item(self, session, topic, _id, force=False): """ Delete item by its internal id :param session: contains the used login username and working project - :param item: it can be: users, projects, vnfds, nsds, ... + :param topic: it can be: users, projects, vnfds, nsds, ... :param _id: server id of the item :param force: indicates if deletion must be forced in case of conflict :return: dictionary with deleted item _id. It raises exception if not found. """ - # TODO add admin to filter, validate rights - # data = self.get_item(item, _id) - filter = {"_id": _id} - self._add_delete_filter(session, item, filter) - if item in ("vnfds", "nsds") and not force: - descriptor = self.get_item(session, item, _id) - descriptor_id = descriptor.get("id") - if descriptor_id: - self._check_dependencies_on_descriptor(session, item, descriptor_id, _id) - elif item == "projects": - if not force: - self._check_project_dependencies(_id) - elif item == "pdus": - if not force: - self._check_pdus_usage(_id) + if topic not in self.map_topic: + raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) + return self.map_topic[topic].delete(session, _id, force) - if item == "nsrs": - nsr = self.db.get_one(item, filter) - if nsr["_admin"].get("nsState") == "INSTANTIATED" and not force: - raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. " - "Launch 'terminate' operation first; or force deletion".format(_id), - http_code=HTTPStatus.CONFLICT) - v = self.db.del_one(item, {"_id": _id}) - self.db.del_list("nslcmops", {"nsInstanceId": _id}) - self.db.del_list("vnfrs", {"nsr-id-ref": _id}) - # set all used pdus as free - self.db.set_list("pdus", {"_admin.usage.nsr_id": _id}, - {"_admin.usageSate": "NOT_IN_USE", "_admin.usage": None}) - self.msg.write("ns", "deleted", {"_id": _id}) - return v - if item in ("vim_accounts", "sdns") and not force: - self.db.set_one(item, {"_id": _id}, {"_admin.to_delete": True}) # TODO change status - if item == "vim_accounts": - self.msg.write("vim_account", "delete", {"_id": _id}) - elif item == "sdns": - self.msg.write("sdn", "delete", {"_id": _id}) - return {"deleted": 1} # TODO indicate an offline operation to return 202 ACCEPTED - - v = self.db.del_one(item, filter) - if item in ("vnfds", "nsds"): - self.fs.file_delete(_id, ignore_non_exist=True) - if item in ("vim_accounts", "sdns", "vnfds", "nsds"): - self.msg.write(item[:-1], "deleted", {"_id": _id}) - return v + def edit_item(self, session, topic, _id, indata=None, kwargs=None, force=False): + """ + Update an existing entry at database + :param session: contains the used login username and working project + :param topic: it can be: users, projects, vnfds, nsds, ... + :param _id: identifier to be updated + :param indata: data to be inserted + :param kwargs: used to override the indata descriptor + :param force: If True avoid some dependence checks + :return: dictionary, raise exception if not found. + """ + if topic not in self.map_topic: + raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) + return self.map_topic[topic].edit(session, _id, indata, kwargs, force) def prune(self): """ @@ -1026,10 +209,10 @@ class Engine(object): if users: return None # raise EngineException("Unauthorized. Database users is not empty", HTTPStatus.UNAUTHORIZED) - indata = {"username": "admin", "password": "admin", "projects": ["admin"]} - fake_session = {"project_id": "admin", "username": "admin"} - self._format_new_data(fake_session, "users", indata) - _id = self.db.create("users", indata) + user_desc = {"username": "admin", "password": "admin", "projects": ["admin"]} + fake_session = {"project_id": "admin", "username": "admin", "admin": True} + roolback_list = [] + _id = self.map_topic["users"].new(roolback_list, fake_session, user_desc, force=True) return _id def init_db(self, target_version='1.0'): @@ -1059,70 +242,3 @@ class Engine(object): raise EngineException("Wrong database status '{}'".format( version["status"]), HTTPStatus.INTERNAL_SERVER_ERROR) return - - def _edit_item(self, session, item, id, content, indata={}, kwargs=None, force=False): - if indata: - indata = self._remove_envelop(item, indata) - - # Override descriptor with query string kwargs - if kwargs: - try: - for k, v in kwargs.items(): - update_content = indata - kitem_old = None - klist = k.split(".") - for kitem in klist: - if kitem_old is not None: - update_content = update_content[kitem_old] - if isinstance(update_content, dict): - kitem_old = kitem - elif isinstance(update_content, list): - kitem_old = int(kitem) - else: - raise EngineException( - "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem)) - update_content[kitem_old] = v - except KeyError: - raise EngineException( - "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old)) - except ValueError: - raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format( - k, kitem)) - except IndexError: - raise EngineException( - "Invalid query string '{}'. Index '{}' out of range".format(k, kitem_old)) - try: - validate_input(indata, item, new=False) - except ValidationError as e: - raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY) - - self._check_edition(session, item, indata, id, force) - deep_update(content, indata) - self._validate_new_data(session, item, content, id=id, force=force) - # self._format_new_data(session, item, content) - self.db.replace(item, id, content) - if item in ("vim_accounts", "sdns"): - indata.pop("_admin", None) - indata["_id"] = id - if item == "vim_accounts": - self.msg.write("vim_account", "edit", indata) - elif item == "sdns": - self.msg.write("sdn", "edit", indata) - return id - - def edit_item(self, session, item, _id, indata={}, kwargs=None, force=False): - """ - Update an existing entry at database - :param session: contains the used login username and working project - :param item: it can be: users, projects, vnfds, nsds, ... - :param _id: identifier to be updated - :param indata: data to be inserted - :param kwargs: used to override the indata descriptor - :param force: If True avoid some dependence checks - :return: dictionary, raise exception if not found. - """ - if not session["admin"] and item == "projects": - raise EngineException("Needed admin privileges to perform this operation", HTTPStatus.UNAUTHORIZED) - - content = self.get_item(session, item, _id) - return self._edit_item(session, item, _id, content, indata, kwargs, force) diff --git a/osm_nbi/html_out.py b/osm_nbi/html_out.py index 857d954..bfa13d4 100644 --- a/osm_nbi/html_out.py +++ b/osm_nbi/html_out.py @@ -20,9 +20,9 @@ html_start = """ ( {} ) - VNFDs - NSDs - NSs + VNFDs + NSDs + NSs USERs PROJECTs TOKENs @@ -120,10 +120,10 @@ def format(data, request, response, session): if response.status and response.status > 202: body += html_body_error.format(yaml.safe_dump(data, explicit_start=True, indent=4, default_flow_style=False)) elif isinstance(data, (list, tuple)): - if request.path_info == "/vnfpkgm/v1/vnf_packages_content": + if request.path_info == "/vnfpkgm/v1/vnf_packages": body += html_upload_body.format(request.path_info, "VNFD") - elif request.path_info == "/nsd/v1/ns_descriptors_content": - body += html_upload_body.format(request.path_info, "NSD") + elif request.path_info == "/nsd/v1/ns_descriptors": + body += html_upload_body.format(request.path_info + "_content", "NSD") for k in data: if isinstance(k, dict): data_id = k.pop("_id", None) diff --git a/osm_nbi/html_public/version b/osm_nbi/html_public/version index c9d0d9b..332aeb5 100644 --- a/osm_nbi/html_public/version +++ b/osm_nbi/html_public/version @@ -1,2 +1,2 @@ -0.1.20 -2018-10-08 +0.1.21 +2018-10-09 diff --git a/osm_nbi/instance_topics.py b/osm_nbi/instance_topics.py new file mode 100644 index 0000000..345e6a3 --- /dev/null +++ b/osm_nbi/instance_topics.py @@ -0,0 +1,456 @@ +# -*- coding: utf-8 -*- + +# import logging +from uuid import uuid4 +from http import HTTPStatus +from time import time +from copy import copy +from validation import validate_input, ValidationError, ns_instantiate, ns_action, ns_scale +from base_topic import BaseTopic, EngineException, get_iterable +from descriptor_topics import DescriptorTopic + +__author__ = "Alfonso Tierno " + + +class NsrTopic(BaseTopic): + topic = "nsrs" + topic_msg = "ns" + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + def _check_descriptor_dependencies(self, session, descriptor): + """ + Check that the dependent descriptors exist on a new descriptor or edition + :param session: client session information + :param descriptor: descriptor to be inserted or edit + :return: None or raises exception + """ + if not descriptor.get("nsdId"): + return + nsd_id = descriptor["nsdId"] + if not self.get_item_list(session, "nsds", {"id": nsd_id}): + raise EngineException("Descriptor error at nsdId='{}' references a non exist nsd".format(nsd_id), + http_code=HTTPStatus.CONFLICT) + + @staticmethod + def format_on_new(content, project_id=None, make_public=False): + BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public) + content["_admin"]["nsState"] = "NOT_INSTANTIATED" + + def check_conflict_on_del(self, session, _id, force=False): + if force: + return + nsr = self.db.get_one("nsrs", {"_id": _id}) + if nsr["_admin"].get("nsState") == "INSTANTIATED": + raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. " + "Launch 'terminate' operation first; or force deletion".format(_id), + http_code=HTTPStatus.CONFLICT) + + def delete(self, session, _id, force=False, dry_run=False): + """ + Delete item by its internal _id + :param session: contains the used login username, working project, and admin rights + :param _id: server internal id + :param force: indicates if deletion must be forced in case of conflict + :param dry_run: make checking but do not delete + :return: dictionary with deleted item _id. It raises EngineException on error: not found, conflict, ... + """ + # TODO add admin to filter, validate rights + BaseTopic.delete(self, session, _id, force, dry_run=True) + if dry_run: + return + + v = self.db.del_one("nsrs", {"_id": _id}) + self.db.del_list("nslcmops", {"nsInstanceId": _id}) + self.db.del_list("vnfrs", {"nsr-id-ref": _id}) + # set all used pdus as free + self.db.set_list("pdus", {"_admin.usage.nsr_id": _id}, + {"_admin.usageSate": "NOT_IN_USE", "_admin.usage": None}) + self._send_msg("deleted", {"_id": _id}) + return v + + def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False): + """ + Creates a new nsr into database. It also creates needed vnfrs + :param rollback: list to append the created items at database in case a rollback must be done + :param session: contains the used login username and working project + :param indata: params to be used for the nsr + :param kwargs: used to override the indata descriptor + :param headers: http request headers + :param force: If True avoid some dependence checks + :param make_public: Make the created item public to all projects + :return: the _id of nsr descriptor created at database + """ + + try: + ns_request = self._remove_envelop(indata) + # Override descriptor with query string kwargs + self._update_input_with_kwargs(ns_request, kwargs) + self._validate_input_new(ns_request, force) + + step = "" + # look for nsr + step = "getting nsd id='{}' from database".format(ns_request.get("nsdId")) + _filter = {"_id": ns_request["nsdId"]} + _filter.update(BaseTopic._get_project_filter(session, write=False, show_all=True)) + nsd = self.db.get_one("nsds", _filter) + + nsr_id = str(uuid4()) + now = time() + step = "filling nsr from input data" + nsr_descriptor = { + "name": ns_request["nsName"], + "name-ref": ns_request["nsName"], + "short-name": ns_request["nsName"], + "admin-status": "ENABLED", + "nsd": nsd, + "datacenter": ns_request["vimAccountId"], + "resource-orchestrator": "osmopenmano", + "description": ns_request.get("nsDescription", ""), + "constituent-vnfr-ref": [], + + "operational-status": "init", # typedef ns-operational- + "config-status": "init", # typedef config-states + "detailed-status": "scheduled", + + "orchestration-progress": {}, + # {"networks": {"active": 0, "total": 0}, "vms": {"active": 0, "total": 0}}, + + "crete-time": now, + "nsd-name-ref": nsd["name"], + "operational-events": [], # "id", "timestamp", "description", "event", + "nsd-ref": nsd["id"], + "instantiate_params": ns_request, + "ns-instance-config-ref": nsr_id, + "id": nsr_id, + "_id": nsr_id, + # "input-parameter": xpath, value, + "ssh-authorized-key": ns_request.get("key-pair-ref"), + } + ns_request["nsr_id"] = nsr_id + + # Create VNFR + needed_vnfds = {} + for member_vnf in nsd["constituent-vnfd"]: + vnfd_id = member_vnf["vnfd-id-ref"] + step = "getting vnfd id='{}' constituent-vnfd='{}' from database".format( + member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"]) + if vnfd_id not in needed_vnfds: + # Obtain vnfd + vnfd = DescriptorTopic.get_one_by_id(self.db, session, "vnfds", vnfd_id) + vnfd.pop("_admin") + needed_vnfds[vnfd_id] = vnfd + else: + vnfd = needed_vnfds[vnfd_id] + step = "filling vnfr vnfd-id='{}' constituent-vnfd='{}'".format( + member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"]) + vnfr_id = str(uuid4()) + vnfr_descriptor = { + "id": vnfr_id, + "_id": vnfr_id, + "nsr-id-ref": nsr_id, + "member-vnf-index-ref": member_vnf["member-vnf-index"], + "created-time": now, + # "vnfd": vnfd, # at OSM model.but removed to avoid data duplication TODO: revise + "vnfd-ref": vnfd_id, + "vnfd-id": vnfd["_id"], # not at OSM model, but useful + "vim-account-id": None, + "vdur": [], + "connection-point": [], + "ip-address": None, # mgmt-interface filled by LCM + } + for cp in vnfd.get("connection-point", ()): + vnf_cp = { + "name": cp["name"], + "connection-point-id": cp.get("id"), + "id": cp.get("id"), + # "ip-address", "mac-address" # filled by LCM + # vim-id # TODO it would be nice having a vim port id + } + vnfr_descriptor["connection-point"].append(vnf_cp) + for vdu in vnfd["vdu"]: + vdur_id = str(uuid4()) + vdur = { + "id": vdur_id, + "vdu-id-ref": vdu["id"], + # TODO "name": "" Name of the VDU in the VIM + "ip-address": None, # mgmt-interface filled by LCM + # "vim-id", "flavor-id", "image-id", "management-ip" # filled by LCM + "internal-connection-point": [], + "interfaces": [], + } + # TODO volumes: name, volume-id + for icp in vdu.get("internal-connection-point", ()): + vdu_icp = { + "id": icp["id"], + "connection-point-id": icp["id"], + "name": icp.get("name"), + # "ip-address", "mac-address" # filled by LCM + # vim-id # TODO it would be nice having a vim port id + } + vdur["internal-connection-point"].append(vdu_icp) + for iface in vdu.get("interface", ()): + vdu_iface = { + "name": iface.get("name"), + # "ip-address", "mac-address" # filled by LCM + # vim-id # TODO it would be nice having a vim port id + } + vdur["interfaces"].append(vdu_iface) + vnfr_descriptor["vdur"].append(vdur) + + step = "creating vnfr vnfd-id='{}' constituent-vnfd='{}' at database".format( + member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"]) + + # add at database + BaseTopic.format_on_new(vnfr_descriptor, session["project_id"], make_public=make_public) + self.db.create("vnfrs", vnfr_descriptor) + rollback.append({"topic": "vnfrs", "_id": vnfr_id}) + nsr_descriptor["constituent-vnfr-ref"].append(vnfr_id) + + step = "creating nsr at database" + self.format_on_new(nsr_descriptor, session["project_id"], make_public=make_public) + self.db.create("nsrs", nsr_descriptor) + rollback.append({"topic": "nsrs", "_id": nsr_id}) + return nsr_id + except Exception as e: + self.logger.exception("Exception {} at NsrTopic.new()".format(e), exc_info=True) + raise EngineException("Error {}: {}".format(step, e)) + except ValidationError as e: + raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY) + + def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None): + raise EngineException("Method edit called directly", HTTPStatus.INTERNAL_SERVER_ERROR) + + +class VnfrTopic(BaseTopic): + topic = "vnfrs" + topic_msg = None + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + def delete(self, session, _id, force=False, dry_run=False): + raise EngineException("Method delete called directly", HTTPStatus.INTERNAL_SERVER_ERROR) + + def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None): + raise EngineException("Method edit called directly", HTTPStatus.INTERNAL_SERVER_ERROR) + + def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False): + # Not used because vnfrs are created and deleted by NsrTopic class directly + raise EngineException("Method new called directly", HTTPStatus.INTERNAL_SERVER_ERROR) + + +class NsLcmOpTopic(BaseTopic): + topic = "nslcmops" + topic_msg = "ns" + operation_schema = { # mapping between operation and jsonschema to validate + "instantiate": ns_instantiate, + "action": ns_action, + "scale": ns_scale, + "terminate": None, + } + + def __init__(self, db, fs, msg): + BaseTopic.__init__(self, db, fs, msg) + + def _validate_input_new(self, input, force=False): + """ + Validates input user content for a new entry. It uses jsonschema for each type or operation. + :param input: user input content for the new topic + :param force: may be used for being more tolerant + :return: The same input content, or a changed version of it. + """ + if self.schema_new: + validate_input(input, self.schema_new) + return input + + def _check_ns_operation(self, session, nsr, operation, indata): + """ + Check that user has enter right parameters for the operation + :param session: + :param operation: it can be: instantiate, terminate, action, TODO: update, heal + :param indata: descriptor with the parameters of the operation + :return: None + """ + vnfds = {} + vim_accounts = [] + nsd = nsr["nsd"] + + def check_valid_vnf_member_index(member_vnf_index): + # TODO change to vnfR + for vnf in nsd["constituent-vnfd"]: + if member_vnf_index == vnf["member-vnf-index"]: + vnfd_id = vnf["vnfd-id-ref"] + if vnfd_id not in vnfds: + vnfds[vnfd_id] = self.db.get_one("vnfds", {"id": vnfd_id}) + return vnfds[vnfd_id] + else: + raise EngineException("Invalid parameter member_vnf_index='{}' is not one of the " + "nsd:constituent-vnfd".format(member_vnf_index)) + + def check_valid_vim_account(vim_account): + if vim_account in vim_accounts: + return + try: + # TODO add _get_project_filter + self.db.get_one("vim_accounts", {"_id": vim_account}) + except Exception: + raise EngineException("Invalid vimAccountId='{}' not present".format(vim_account)) + vim_accounts.append(vim_account) + + if operation == "action": + # check vnf_member_index + if indata.get("vnf_member_index"): + indata["member_vnf_index"] = indata.pop("vnf_member_index") # for backward compatibility + if not indata.get("member_vnf_index"): + raise EngineException("Missing 'member_vnf_index' parameter") + vnfd = check_valid_vnf_member_index(indata["member_vnf_index"]) + # check primitive + for config_primitive in get_iterable(vnfd.get("vnf-configuration", {}).get("config-primitive")): + if indata["primitive"] == config_primitive["name"]: + # check needed primitive_params are provided + if indata.get("primitive_params"): + in_primitive_params_copy = copy(indata["primitive_params"]) + else: + in_primitive_params_copy = {} + for paramd in get_iterable(config_primitive.get("parameter")): + if paramd["name"] in in_primitive_params_copy: + del in_primitive_params_copy[paramd["name"]] + elif not paramd.get("default-value"): + raise EngineException("Needed parameter {} not provided for primitive '{}'".format( + paramd["name"], indata["primitive"])) + # check no extra primitive params are provided + if in_primitive_params_copy: + raise EngineException("parameter/s '{}' not present at vnfd for primitive '{}'".format( + list(in_primitive_params_copy.keys()), indata["primitive"])) + break + else: + raise EngineException("Invalid primitive '{}' is not present at vnfd".format(indata["primitive"])) + if operation == "scale": + vnfd = check_valid_vnf_member_index(indata["scaleVnfData"]["scaleByStepData"]["member-vnf-index"]) + for scaling_group in get_iterable(vnfd.get("scaling-group-descriptor")): + if indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"] == scaling_group["name"]: + break + else: + raise EngineException("Invalid scaleVnfData:scaleByStepData:scaling-group-descriptor '{}' is not " + "present at vnfd:scaling-group-descriptor".format( + indata["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"])) + if operation == "instantiate": + # check vim_account + check_valid_vim_account(indata["vimAccountId"]) + for in_vnf in get_iterable(indata.get("vnf")): + vnfd = check_valid_vnf_member_index(in_vnf["member-vnf-index"]) + if in_vnf.get("vimAccountId"): + check_valid_vim_account(in_vnf["vimAccountId"]) + for in_vdu in get_iterable(in_vnf.get("vdu")): + for vdud in get_iterable(vnfd.get("vdu")): + if vdud["id"] == in_vdu["id"]: + for volume in get_iterable(in_vdu.get("volume")): + for volumed in get_iterable(vdud.get("volumes")): + if volumed["name"] == volume["name"]: + break + else: + raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu[id='{}']:" + "volume:name='{}' is not present at vnfd:vdu:volumes list". + format(in_vnf["member-vnf-index"], in_vdu["id"], + volume["name"])) + break + else: + raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu:id='{}' is not " + "present at vnfd".format(in_vnf["member-vnf-index"], in_vdu["id"])) + + for in_internal_vld in get_iterable(in_vnf.get("internal-vld")): + for internal_vldd in get_iterable(vnfd.get("internal-vld")): + if in_internal_vld["name"] == internal_vldd["name"] or \ + in_internal_vld["name"] == internal_vldd["id"]: + break + else: + raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:internal-vld:name='{}'" + " is not present at vnfd '{}'".format(in_vnf["member-vnf-index"], + in_internal_vld["name"], + vnfd["id"])) + for in_vld in get_iterable(indata.get("vld")): + for vldd in get_iterable(nsd.get("vld")): + if in_vld["name"] == vldd["name"] or in_vld["name"] == vldd["id"]: + break + else: + raise EngineException("Invalid parameter vld:name='{}' is not present at nsd:vld".format( + in_vld["name"])) + + def _create_nslcmop(self, session, nsInstanceId, operation, params): + now = time() + _id = str(uuid4()) + nslcmop = { + "id": _id, + "_id": _id, + "operationState": "PROCESSING", # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK + "statusEnteredTime": now, + "nsInstanceId": nsInstanceId, + "lcmOperationType": operation, + "startTime": now, + "isAutomaticInvocation": False, + "operationParams": params, + "isCancelPending": False, + "links": { + "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id, + "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsInstanceId, + } + } + return nslcmop + + def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False): + """ + Performs a new operation over a ns + :param rollback: list to append created items at database in case a rollback must to be done + :param session: contains the used login username and working project + :param indata: descriptor with the parameters of the operation. It must contains among others + nsInstanceId: _id of the nsr to perform the operation + operation: it can be: instantiate, terminate, action, TODO: update, heal + :param kwargs: used to override the indata descriptor + :param headers: http request headers + :param force: If True avoid some dependence checks + :param make_public: Make the created item public to all projects + :return: id of the nslcmops + """ + try: + # Override descriptor with query string kwargs + self._update_input_with_kwargs(indata, kwargs) + operation = indata["lcmOperationType"] + nsInstanceId = indata["nsInstanceId"] + + validate_input(indata, self.operation_schema[operation]) + # get ns from nsr_id + _filter = BaseTopic._get_project_filter(session, write=True, show_all=False) + _filter["_id"] = nsInstanceId + nsr = self.db.get_one("nsrs", _filter) + + # initial checking + if not nsr["_admin"].get("nsState") or nsr["_admin"]["nsState"] == "NOT_INSTANTIATED": + if operation == "terminate" and indata.get("autoremove"): + # NSR must be deleted + return self.delete(session, nsInstanceId) + if operation != "instantiate": + raise EngineException("ns_instance '{}' cannot be '{}' because it is not instantiated".format( + nsInstanceId, operation), HTTPStatus.CONFLICT) + else: + if operation == "instantiate" and not indata.get("force"): + raise EngineException("ns_instance '{}' cannot be '{}' because it is already instantiated".format( + nsInstanceId, operation), HTTPStatus.CONFLICT) + self._check_ns_operation(session, nsr, operation, indata) + nslcmop_desc = self._create_nslcmop(session, nsInstanceId, operation, indata) + self.format_on_new(nslcmop_desc, session["project_id"], make_public=make_public) + _id = self.db.create("nslcmops", nslcmop_desc) + rollback.append({"topic": "nslcmops", "_id": _id}) + self.msg.write("ns", operation, nslcmop_desc) + return _id + except ValidationError as e: + raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY) + # except DbException as e: + # raise EngineException("Cannot get ns_instance '{}': {}".format(e), HTTPStatus.NOT_FOUND) + + def delete(self, session, _id, force=False, dry_run=False): + raise EngineException("Method delete called directly", HTTPStatus.INTERNAL_SERVER_ERROR) + + def edit(self, session, _id, indata=None, kwargs=None, force=False, content=None): + raise EngineException("Method edit called directly", HTTPStatus.INTERNAL_SERVER_ERROR) diff --git a/osm_nbi/nbi.py b/osm_nbi/nbi.py index 0a4c402..60a327e 100644 --- a/osm_nbi/nbi.py +++ b/osm_nbi/nbi.py @@ -483,7 +483,7 @@ class Server(object): return f elif len(args) == 2 and args[0] == "db-clear": - return self.engine.del_item_list({"project_id": "admin", "admin": True}, args[1], kwargs) + return self.engine.db.del_list(args[1], kwargs) elif args and args[0] == "prune": return self.engine.prune() elif args and args[0] == "login": @@ -506,17 +506,17 @@ class Server(object): time.sleep(sleep_time) # thread_info elif len(args) >= 2 and args[0] == "message": - topic = args[1] - return_text = "
{} ->\n".format(topic)
+            main_topic = args[1]
+            return_text = "
{} ->\n".format(main_topic)
             try:
                 if cherrypy.request.method == 'POST':
                     to_send = yaml.load(cherrypy.request.body)
                     for k, v in to_send.items():
-                        self.engine.msg.write(topic, k, v)
+                        self.engine.msg.write(main_topic, k, v)
                         return_text += "  {}: {}\n".format(k, v)
                 elif cherrypy.request.method == 'GET':
                     for k, v in kwargs.items():
-                        self.engine.msg.write(topic, k, yaml.load(v))
+                        self.engine.msg.write(main_topic, k, yaml.load(v))
                         return_text += "  {}: {}\n".format(k, yaml.load(v))
             except Exception as e:
                 return_text += "Error: " + str(e)
@@ -545,7 +545,7 @@ class Server(object):
 
     def _check_valid_url_method(self, method, *args):
         if len(args) < 3:
-            raise NbiException("URL must contain at least 'topic/version/item'", HTTPStatus.METHOD_NOT_ALLOWED)
+            raise NbiException("URL must contain at least 'main_topic/version/topic'", HTTPStatus.METHOD_NOT_ALLOWED)
 
         reference = self.valid_methods
         for arg in args:
@@ -571,33 +571,35 @@ class Server(object):
         return
 
     @staticmethod
-    def _set_location_header(topic, version, item, id):
+    def _set_location_header(main_topic, version, topic, id):
         """
         Insert response header Location with the URL of created item base on URL params
-        :param topic:
+        :param main_topic:
         :param version:
-        :param item:
+        :param topic:
         :param id:
         :return: None
         """
         # Use cherrypy.request.base for absoluted path and make use of request.header HOST just in case behind aNAT
-        cherrypy.response.headers["Location"] = "/osm/{}/{}/{}/{}".format(topic, version, item, id)
+        cherrypy.response.headers["Location"] = "/osm/{}/{}/{}/{}".format(main_topic, version, topic, id)
         return
 
     @cherrypy.expose
-    def default(self, topic=None, version=None, item=None, _id=None, item2=None, *args, **kwargs):
+    def default(self, main_topic=None, version=None, topic=None, _id=None, item=None, *args, **kwargs):
         session = None
         outdata = None
         _format = None
         method = "DONE"
-        engine_item = None
+        engine_topic = None
         rollback = []
         session = None
         try:
-            if not topic or not version or not item:
-                raise NbiException("URL must contain at least 'topic/version/item'", HTTPStatus.METHOD_NOT_ALLOWED)
-            if topic not in ("admin", "vnfpkgm", "nsd", "nslcm"):
-                raise NbiException("URL topic '{}' not supported".format(topic), HTTPStatus.METHOD_NOT_ALLOWED)
+            if not main_topic or not version or not topic:
+                raise NbiException("URL must contain at least 'main_topic/version/topic'",
+                                   HTTPStatus.METHOD_NOT_ALLOWED)
+            if main_topic not in ("admin", "vnfpkgm", "nsd", "nslcm"):
+                raise NbiException("URL main_topic '{}' not supported".format(main_topic),
+                                   HTTPStatus.METHOD_NOT_ALLOWED)
             if version != 'v1':
                 raise NbiException("URL version '{}' not supported".format(version), HTTPStatus.METHOD_NOT_ALLOWED)
 
@@ -610,97 +612,107 @@ class Server(object):
             else:
                 force = False
 
-            self._check_valid_url_method(method, topic, version, item, _id, item2, *args)
+            self._check_valid_url_method(method, main_topic, version, topic, _id, item, *args)
 
-            if topic == "admin" and item == "tokens":
+            if main_topic == "admin" and topic == "tokens":
                 return self.token(method, _id, kwargs)
 
             # self.engine.load_dbase(cherrypy.request.app.config)
             session = self.authenticator.authorize()
             indata = self._format_in(kwargs)
-            engine_item = item
-            if item == "subscriptions":
-                engine_item = topic + "_" + item
-            if item2:
-                engine_item = item2
-
-            if topic == "nsd":
-                engine_item = "nsds"
-            elif topic == "vnfpkgm":
-                engine_item = "vnfds"
-            elif topic == "nslcm":
-                engine_item = "nsrs"
-                if item == "ns_lcm_op_occs":
-                    engine_item = "nslcmops"
-                if item == "vnfrs" or item == "vnf_instances":
-                    engine_item = "vnfrs"
-            elif topic == "pdu":
-                engine_item = "pdus"
-            if engine_item == "vims":   # TODO this is for backward compatibility, it will remove in the future
-                engine_item = "vim_accounts"
+            engine_topic = topic
+            if topic == "subscriptions":
+                engine_topic = main_topic + "_" + topic
+            if item:
+                engine_topic = item
+
+            if main_topic == "nsd":
+                engine_topic = "nsds"
+            elif main_topic == "vnfpkgm":
+                engine_topic = "vnfds"
+            elif main_topic == "nslcm":
+                engine_topic = "nsrs"
+                if topic == "ns_lcm_op_occs":
+                    engine_topic = "nslcmops"
+                if topic == "vnfrs" or topic == "vnf_instances":
+                    engine_topic = "vnfrs"
+            elif main_topic == "pdu":
+                engine_topic = "pdus"
+            if engine_topic == "vims":   # TODO this is for backward compatibility, it will remove in the future
+                engine_topic = "vim_accounts"
 
             if method == "GET":
-                if item2 in ("nsd_content", "package_content", "artifacts", "vnfd", "nsd"):
-                    if item2 in ("vnfd", "nsd"):
+                if item in ("nsd_content", "package_content", "artifacts", "vnfd", "nsd"):
+                    if item in ("vnfd", "nsd"):
                         path = "$DESCRIPTOR"
                     elif args:
                         path = args
-                    elif item2 == "artifacts":
+                    elif item == "artifacts":
                         path = ()
                     else:
                         path = None
-                    file, _format = self.engine.get_file(session, engine_item, _id, path,
+                    file, _format = self.engine.get_file(session, engine_topic, _id, path,
                                                          cherrypy.request.headers.get("Accept"))
                     outdata = file
                 elif not _id:
-                    outdata = self.engine.get_item_list(session, engine_item, kwargs)
+                    outdata = self.engine.get_item_list(session, engine_topic, kwargs)
                 else:
-                    outdata = self.engine.get_item(session, engine_item, _id)
+                    outdata = self.engine.get_item(session, engine_topic, _id)
             elif method == "POST":
-                if item in ("ns_descriptors_content", "vnf_packages_content"):
+                if topic in ("ns_descriptors_content", "vnf_packages_content"):
                     _id = cherrypy.request.headers.get("Transaction-Id")
                     if not _id:
-                        _id = self.engine.new_item(rollback, session, engine_item, {}, None, cherrypy.request.headers,
+                        _id = self.engine.new_item(rollback, session, engine_topic, {}, None, cherrypy.request.headers,
                                                    force=force)
-                    completed = self.engine.upload_content(session, engine_item, _id, indata, kwargs,
-                                                           cherrypy.request.headers)
+                    completed = self.engine.upload_content(session, engine_topic, _id, indata, kwargs,
+                                                           cherrypy.request.headers, force=force)
                     if completed:
-                        self._set_location_header(topic, version, item, _id)
+                        self._set_location_header(main_topic, version, topic, _id)
                     else:
                         cherrypy.response.headers["Transaction-Id"] = _id
                     outdata = {"id": _id}
-                elif item == "ns_instances_content":
-                    _id = self.engine.new_item(rollback, session, engine_item, indata, kwargs, force=force)
-                    self.engine.ns_operation(rollback, session, _id, "instantiate", indata, None)
-                    self._set_location_header(topic, version, item, _id)
+                elif topic == "ns_instances_content":
+                    # creates NSR
+                    _id = self.engine.new_item(rollback, session, engine_topic, indata, kwargs, force=force)
+                    # creates nslcmop
+                    indata["lcmOperationType"] = "instantiate"
+                    indata["nsInstanceId"] = _id
+                    self.engine.new_item(rollback, session, "nslcmops", indata, None)
+                    self._set_location_header(main_topic, version, topic, _id)
                     outdata = {"id": _id}
-                elif item == "ns_instances" and item2:
-                    _id = self.engine.ns_operation(rollback, session, _id, item2, indata, kwargs)
-                    self._set_location_header(topic, version, "ns_lcm_op_occs", _id)
+                elif topic == "ns_instances" and item:
+                    indata["lcmOperationType"] = item
+                    indata["nsInstanceId"] = _id
+                    _id = self.engine.new_item(rollback, session, "nslcmops", indata, kwargs)
+                    self._set_location_header(main_topic, version, "ns_lcm_op_occs", _id)
                     outdata = {"id": _id}
                     cherrypy.response.status = HTTPStatus.ACCEPTED.value
                 else:
-                    _id = self.engine.new_item(rollback, session, engine_item, indata, kwargs, cherrypy.request.headers,
-                                               force=force)
-                    self._set_location_header(topic, version, item, _id)
+                    _id = self.engine.new_item(rollback, session, engine_topic, indata, kwargs,
+                                               cherrypy.request.headers, force=force)
+                    self._set_location_header(main_topic, version, topic, _id)
                     outdata = {"id": _id}
-                    # TODO form NsdInfo when item in ("ns_descriptors", "vnf_packages")
+                    # TODO form NsdInfo when topic in ("ns_descriptors", "vnf_packages")
                 cherrypy.response.status = HTTPStatus.CREATED.value
 
             elif method == "DELETE":
                 if not _id:
-                    outdata = self.engine.del_item_list(session, engine_item, kwargs)
+                    outdata = self.engine.del_item_list(session, engine_topic, kwargs)
                     cherrypy.response.status = HTTPStatus.OK.value
                 else:  # len(args) > 1
-                    if item == "ns_instances_content" and not force:
-                        opp_id = self.engine.ns_operation(rollback, session, _id, "terminate", {"autoremove": True},
-                                                          None)
+                    if topic == "ns_instances_content" and not force:
+                        nslcmop_desc = {
+                            "lcmOperationType": "terminate",
+                            "nsInstanceId": _id,
+                            "autoremove": True
+                        }
+                        opp_id = self.engine.new_item(rollback, session, "nslcmops", nslcmop_desc, None)
                         outdata = {"_id": opp_id}
                         cherrypy.response.status = HTTPStatus.ACCEPTED.value
                     else:
-                        self.engine.del_item(session, engine_item, _id, force)
+                        self.engine.del_item(session, engine_topic, _id, force)
                         cherrypy.response.status = HTTPStatus.NO_CONTENT.value
-                if engine_item in ("vim_accounts", "sdns"):
+                if engine_topic in ("vim_accounts", "sdns"):
                     cherrypy.response.status = HTTPStatus.ACCEPTED.value
 
             elif method in ("PUT", "PATCH"):
@@ -708,35 +720,44 @@ class Server(object):
                 if not indata and not kwargs:
                     raise NbiException("Nothing to update. Provide payload and/or query string",
                                        HTTPStatus.BAD_REQUEST)
-                if item2 in ("nsd_content", "package_content") and method == "PUT":
-                    completed = self.engine.upload_content(session, engine_item, _id, indata, kwargs,
-                                                           cherrypy.request.headers)
+                if item in ("nsd_content", "package_content") and method == "PUT":
+                    completed = self.engine.upload_content(session, engine_topic, _id, indata, kwargs,
+                                                           cherrypy.request.headers, force=force)
                     if not completed:
                         cherrypy.response.headers["Transaction-Id"] = id
                 else:
-                    self.engine.edit_item(session, engine_item, _id, indata, kwargs, force=force)
+                    self.engine.edit_item(session, engine_topic, _id, indata, kwargs, force=force)
                 cherrypy.response.status = HTTPStatus.NO_CONTENT.value
             else:
                 raise NbiException("Method {} not allowed".format(method), HTTPStatus.METHOD_NOT_ALLOWED)
             return self._format_out(outdata, session, _format)
-        except (NbiException, EngineException, DbException, FsException, MsgException, AuthException) as e:
-            cherrypy.log("Exception {}".format(e))
-            cherrypy.response.status = e.http_code.value
+        except Exception as e:
+            if isinstance(e, (NbiException, EngineException, DbException, FsException, MsgException, AuthException)):
+                http_code_value = cherrypy.response.status = e.http_code.value
+                http_code_name = e.http_code.name
+                cherrypy.log("Exception {}".format(e))
+            else:
+                http_code_value = cherrypy.response.status = HTTPStatus.BAD_REQUEST.value  # INTERNAL_SERVER_ERROR
+                cherrypy.log("CRITICAL: Exception {}".format(e))
+                http_code_name = HTTPStatus.BAD_REQUEST.name
             if hasattr(outdata, "close"):  # is an open file
                 outdata.close()
+            error_text = str(e)
+            rollback.reverse()
             for rollback_item in rollback:
                 try:
                     self.engine.del_item(**rollback_item, session=session, force=True)
                 except Exception as e2:
-                    cherrypy.log("Rollback Exception {}: {}".format(rollback_item, e2))
-            error_text = str(e)
-            if isinstance(e, MsgException):
-                error_text = "{} has been '{}' but other modules cannot be informed because an error on bus".format(
-                    engine_item[:-1], method, error_text)
+                    rollback_error_text = "Rollback Exception {}: {}".format(rollback_item, e2)
+                    cherrypy.log(rollback_error_text)
+                    error_text += ". " + rollback_error_text
+            # if isinstance(e, MsgException):
+            #     error_text = "{} has been '{}' but other modules cannot be informed because an error on bus".format(
+            #         engine_topic[:-1], method, error_text)
             problem_details = {
-                "code": e.http_code.name,
-                "status": e.http_code.value,
-                "detail": str(e),
+                "code": http_code_name,
+                "status": http_code_value,
+                "detail": error_text,
             }
             return self._format_out(problem_details, session)
             # raise cherrypy.HTTPError(e.http_code.value, str(e))
@@ -808,16 +829,16 @@ def _start_service():
         file_handler.setFormatter(log_formatter_simple)
         logger_cherry.addHandler(file_handler)
         logger_nbi.addHandler(file_handler)
-    else:
-        for format_, logger in {"nbi.server": logger_server,
-                                "nbi.access": logger_access,
-                                "%(name)s %(filename)s:%(lineno)s": logger_nbi
-                                }.items():
-            log_format_cherry = "%(asctime)s %(levelname)s {} %(message)s".format(format_)
-            log_formatter_cherry = logging.Formatter(log_format_cherry, datefmt='%Y-%m-%dT%H:%M:%S')
-            str_handler = logging.StreamHandler()
-            str_handler.setFormatter(log_formatter_cherry)
-            logger.addHandler(str_handler)
+    # log always to standard output
+    for format_, logger in {"nbi.server %(filename)s:%(lineno)s": logger_server,
+                            "nbi.access %(filename)s:%(lineno)s": logger_access,
+                            "%(name)s %(filename)s:%(lineno)s": logger_nbi
+                            }.items():
+        log_format_cherry = "%(asctime)s %(levelname)s {} %(message)s".format(format_)
+        log_formatter_cherry = logging.Formatter(log_format_cherry, datefmt='%Y-%m-%dT%H:%M:%S')
+        str_handler = logging.StreamHandler()
+        str_handler.setFormatter(log_formatter_cherry)
+        logger.addHandler(str_handler)
 
     if engine_config["global"].get("log.level"):
         logger_cherry.setLevel(engine_config["global"]["log.level"])
diff --git a/osm_nbi/tests/clear-all.sh b/osm_nbi/tests/clear-all.sh
index 5a9a9c6..e53df12 100755
--- a/osm_nbi/tests/clear-all.sh
+++ b/osm_nbi/tests/clear-all.sh
@@ -10,6 +10,7 @@ function usage(){
     echo -e "  OPTIONS"
     echo -e "     -h --help:   show this help"
     echo -e "     -f --force:  Do not ask for confirmation"
+    echo -e "     --completely:  It cleans also user admin. NBI will need to be restarted to init database"
     echo -e "     --clean-RO:  clean RO content. RO client (openmano) must be installed and configured"
     echo -e "     --clean-VCA: clean VCA content. juju  must be installed and configured"
     echo -e "  ENV variable 'OSMNBI_URL' is used for the URL of the NBI server. If missing, it uses" \
@@ -38,6 +39,7 @@ do
     shift
     ( [ "$option" == -h ] || [ "$option" == --help ] ) && usage && exit
     ( [ "$option" == -f ] || [ "$option" == --force ] ) && OSMNBI_CLEAN_FORCE=yes && continue
+    [ "$option" == --completely ] && OSMNBI_COMPLETELY=yes && continue
     [ "$option" == --clean-RO ] && OSMNBI_CLEAN_RO=yes && continue
     [ "$option" == --clean-VCA ] && OSMNBI_CLEAN_VCA=yes && continue
     echo "Unknown option '$option'. Type $0 --help" 2>&1 && exit 1
@@ -64,13 +66,18 @@ then
     done
 fi
 
-for item in vim_accounts vims sdns nsrs vnfrs nslcmops nsds vnfds projects
+for item in vim_accounts sdns nsrs vnfrs nslcmops nsds vnfds projects # vims
 do
     curl --insecure ${OSMNBI_URL}/test/db-clear/${item}
     echo " ${item}"
 done
-# delete all users except admin
-curl --insecure ${OSMNBI_URL}/test/db-clear/users?username.ne=admin
+if [ -n "$OSMNBI_COMPLETELY" ] ; then
+    curl --insecure ${OSMNBI_URL}/test/db-clear/users && echo " ${item}"
+    curl --insecure ${OSMNBI_URL}/test/db-clear/version && echo " ${item}"
+else
+    # delete all users except admin
+    curl --insecure ${OSMNBI_URL}/test/db-clear/users?username.ne=admin
+fi
 
 if [ -n "$OSMNBI_CLEAN_RO" ]
 then
diff --git a/osm_nbi/validation.py b/osm_nbi/validation.py
index 2eb7034..fa812bf 100644
--- a/osm_nbi/validation.py
+++ b/osm_nbi/validation.py
@@ -188,6 +188,8 @@ ns_instantiate = {
     "$schema": "http://json-schema.org/draft-04/schema#",
     "type": "object",
     "properties": {
+        "lcmOperationType": string_schema,
+        "nsInstanceId": id_schema,
         "nsName": name_schema,
         "nsDescription": {"oneOf": [description_schema, {"type": "null"}]},
         "nsdId": id_schema,
@@ -258,6 +260,8 @@ ns_action = {   # TODO for the moment it is only contemplated the vnfd primitive
     "$schema": "http://json-schema.org/draft-04/schema#",
     "type": "object",
     "properties": {
+        "lcmOperationType": string_schema,
+        "nsInstanceId": id_schema,
         "member_vnf_index": name_schema,
         "vnf_member_index": name_schema,  # TODO for backward compatibility. To remove in future
         "vdu_id": name_schema,
@@ -272,6 +276,8 @@ ns_scale = {   # TODO for the moment it is only VDU-scaling
     "$schema": "http://json-schema.org/draft-04/schema#",
     "type": "object",
     "properties": {
+        "lcmOperationType": string_schema,
+        "nsInstanceId": id_schema,
         "scaleType": {"enum": ["SCALE_VNF"]},
         "scaleVnfData": {
             "type": "object",
@@ -555,19 +561,14 @@ class ValidationError(Exception):
     pass
 
 
-def validate_input(indata, item, new=True):
+def validate_input(indata, schema_to_use):
     """
     Validates input data against json schema
     :param indata: user input data. Should be a dictionary
-    :param item: can be users, projects, vims, sdns, ns_xxxxx
-    :param new: True if the validation is for creating or False if it is for editing
-    :return: None if ok, raises ValidationError exception otherwise
+    :param schema_to_use: jsonschema to test
+    :return: None if ok, raises ValidationError exception on error
     """
     try:
-        if new:
-            schema_to_use = nbi_new_input_schemas.get(item)
-        else:
-            schema_to_use = nbi_edit_input_schemas.get(item)
         if schema_to_use:
             js_v(indata, schema_to_use)
         return None
diff --git a/tox.ini b/tox.ini
index 0207a67..b864700 100644
--- a/tox.ini
+++ b/tox.ini
@@ -10,8 +10,8 @@ commands=nosetests
 [testenv:flake8]
 basepython = python3
 deps = flake8
-commands =
-    flake8 osm_nbi/ setup.py --max-line-length 120 --exclude .svn,CVS,.gz,.git,__pycache__,.tox,local,temp,vnfd_catalog.py,nsd_catalog.py --ignore W291,W293,E226
+commands = flake8 osm_nbi/ setup.py --max-line-length 120 \
+    --exclude .svn,CVS,.gz,.git,__pycache__,.tox,local,temp,vnfd_catalog.py,nsd_catalog.py --ignore W291,W293,E226,E402
 
 [testenv:build]
 basepython = python3