osm_nbi/descriptor_topics.py

   1 # -*- coding: utf-8 -*-
   2
   3 import tarfile
   4 import yaml
   5 import json
   6 # import logging
   7 from hashlib import md5
   8 from osm_common.dbbase import DbException, deep_update_rfc7396
   9 from http import HTTPStatus
  10 from validation import ValidationError, pdu_new_schema, pdu_edit_schema
  11 from base_topic import BaseTopic, EngineException
  12
  13 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  14
  15
  16 class DescriptorTopic(BaseTopic):
  17
  18     def __init__(self, db, fs, msg):
  19         BaseTopic.__init__(self, db, fs, msg)
  20
  21     def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False):
  22         # check that this id is not present
  23         _filter = {"id": final_content["id"]}
  24         if _id:
  25             _filter["_id.neq"] = _id
  26
  27         _filter.update(self._get_project_filter(session, write=False, show_all=False))
  28         if self.db.get_one(self.topic, _filter, fail_on_empty=False):
  29             raise EngineException("{} with id '{}' already exists for this project".format(self.topic[:-1],
  30                                                                                            final_content["id"]),
  31                                   HTTPStatus.CONFLICT)
  32         # TODO validate with pyangbind. Load and dumps to convert data types
  33
  34     @staticmethod
  35     def format_on_new(content, project_id=None, make_public=False):
  36         BaseTopic.format_on_new(content, project_id=project_id, make_public=make_public)
  37         content["_admin"]["onboardingState"] = "CREATED"
  38         content["_admin"]["operationalState"] = "DISABLED"
  39         content["_admin"]["usageSate"] = "NOT_IN_USE"
  40
  41     def delete(self, session, _id, force=False, dry_run=False):
  42         """
  43         Delete item by its internal _id
  44         :param session: contains the used login username, working project, and admin rights
  45         :param _id: server internal id
  46         :param force: indicates if deletion must be forced in case of conflict
  47         :param dry_run: make checking but do not delete
  48         :return: dictionary with deleted item _id. It raises EngineException on error: not found, conflict, ...
  49         """
  50         # TODO add admin to filter, validate rights
  51         v = BaseTopic.delete(self, session, _id, force, dry_run=True)
  52         if dry_run:
  53             return
  54         v = self.db.del_one(self.topic, {"_id": _id})
  55         self.fs.file_delete(_id, ignore_non_exist=True)
  56         self._send_msg("delete", {"_id": _id})
  57         return v
  58
  59     @staticmethod
  60     def get_one_by_id(db, session, topic, id):
  61         # find owned by this project
  62         _filter = BaseTopic._get_project_filter(session, write=False, show_all=False)
  63         _filter["id"] = id
  64         desc_list = db.get_list(topic, _filter)
  65         if len(desc_list) == 1:
  66             return desc_list[0]
  67         elif len(desc_list) > 1:
  68             raise DbException("Found more than one {} with id='{}' belonging to this project".format(topic[:-1], id),
  69                               HTTPStatus.CONFLICT)
  70
  71         # not found any: try to find public
  72         _filter = BaseTopic._get_project_filter(session, write=False, show_all=True)
  73         _filter["id"] = id
  74         desc_list = db.get_list(topic, _filter)
  75         if not desc_list:
  76             raise DbException("Not found any {} with id='{}'".format(topic[:-1], id), HTTPStatus.NOT_FOUND)
  77         elif len(desc_list) == 1:
  78             return desc_list[0]
  79         else:
  80             raise DbException("Found more than one public {} with id='{}'; and no one belonging to this project".format(
  81                 topic[:-1], id), HTTPStatus.CONFLICT)
  82
  83     def new(self, rollback, session, indata=None, kwargs=None, headers=None, force=False, make_public=False):
  84         """
  85         Creates a new almost empty DISABLED  entry into database. Due to SOL005, it does not follow normal procedure.
  86         Creating a VNFD or NSD is done in two steps: 1. Creates an empty descriptor (this step) and 2) upload content
  87         (self.upload_content)
  88         :param rollback: list to append created items at database in case a rollback may to be done
  89         :param session: contains the used login username and working project
  90         :param indata: data to be inserted
  91         :param kwargs: used to override the indata descriptor
  92         :param headers: http request headers
  93         :param force: If True avoid some dependence checks
  94         :param make_public: Make the created descriptor public to all projects
  95         :return: _id: identity of the inserted data.
  96         """
  97
  98         try:
  99             # _remove_envelop
 100             if indata:
 101                 if "userDefinedData" in indata:
 102                     indata = indata['userDefinedData']
 103
 104             # Override descriptor with query string kwargs
 105             self._update_input_with_kwargs(indata, kwargs)
 106             # uncomment when this method is implemented.
 107             # Avoid override in this case as the target is userDefinedData, but not vnfd,nsd descriptors
 108             # indata = DescriptorTopic._validate_input_new(self, indata, force=force)
 109
 110             content = {"_admin": {"userDefinedData": indata}}
 111             self.format_on_new(content, session["project_id"], make_public=make_public)
 112             _id = self.db.create(self.topic, content)
 113             rollback.append({"topic": self.topic, "_id": _id})
 114             return _id
 115         except ValidationError as e:
 116             raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
 117
 118     def upload_content(self, session, _id, indata, kwargs, headers, force=False):
 119         """
 120         Used for receiving content by chunks (with a transaction_id header and/or gzip file. It will store and extract)
 121         :param session: session
 122         :param _id : the nsd,vnfd is already created, this is the id
 123         :param indata: http body request
 124         :param kwargs: user query string to override parameters. NOT USED
 125         :param headers:  http request headers
 126         :param force: to be more tolerant with validation
 127         :return: True package has is completely uploaded or False if partial content has been uplodaed.
 128             Raise exception on error
 129         """
 130         # Check that _id exists and it is valid
 131         current_desc = self.show(session, _id)
 132
 133         content_range_text = headers.get("Content-Range")
 134         expected_md5 = headers.get("Content-File-MD5")
 135         compressed = None
 136         content_type = headers.get("Content-Type")
 137         if content_type and "application/gzip" in content_type or "application/x-gzip" in content_type or \
 138                 "application/zip" in content_type:
 139             compressed = "gzip"
 140         filename = headers.get("Content-Filename")
 141         if not filename:
 142             filename = "package.tar.gz" if compressed else "package"
 143         # TODO change to Content-Disposition filename https://tools.ietf.org/html/rfc6266
 144         file_pkg = None
 145         error_text = ""
 146         try:
 147             if content_range_text:
 148                 content_range = content_range_text.replace("-", " ").replace("/", " ").split()
 149                 if content_range[0] != "bytes":  # TODO check x<y not negative < total....
 150                     raise IndexError()
 151                 start = int(content_range[1])
 152                 end = int(content_range[2]) + 1
 153                 total = int(content_range[3])
 154             else:
 155                 start = 0
 156
 157             if start:
 158                 if not self.fs.file_exists(_id, 'dir'):
 159                     raise EngineException("invalid Transaction-Id header", HTTPStatus.NOT_FOUND)
 160             else:
 161                 self.fs.file_delete(_id, ignore_non_exist=True)
 162                 self.fs.mkdir(_id)
 163
 164             storage = self.fs.get_params()
 165             storage["folder"] = _id
 166
 167             file_path = (_id, filename)
 168             if self.fs.file_exists(file_path, 'file'):
 169                 file_size = self.fs.file_size(file_path)
 170             else:
 171                 file_size = 0
 172             if file_size != start:
 173                 raise EngineException("invalid Content-Range start sequence, expected '{}' but received '{}'".format(
 174                     file_size, start), HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
 175             file_pkg = self.fs.file_open(file_path, 'a+b')
 176             if isinstance(indata, dict):
 177                 indata_text = yaml.safe_dump(indata, indent=4, default_flow_style=False)
 178                 file_pkg.write(indata_text.encode(encoding="utf-8"))
 179             else:
 180                 indata_len = 0
 181                 while True:
 182                     indata_text = indata.read(4096)
 183                     indata_len += len(indata_text)
 184                     if not indata_text:
 185                         break
 186                     file_pkg.write(indata_text)
 187             if content_range_text:
 188                 if indata_len != end-start:
 189                     raise EngineException("Mismatch between Content-Range header {}-{} and body length of {}".format(
 190                         start, end-1, indata_len), HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
 191                 if end != total:
 192                     # TODO update to UPLOADING
 193                     return False
 194
 195             # PACKAGE UPLOADED
 196             if expected_md5:
 197                 file_pkg.seek(0, 0)
 198                 file_md5 = md5()
 199                 chunk_data = file_pkg.read(1024)
 200                 while chunk_data:
 201                     file_md5.update(chunk_data)
 202                     chunk_data = file_pkg.read(1024)
 203                 if expected_md5 != file_md5.hexdigest():
 204                     raise EngineException("Error, MD5 mismatch", HTTPStatus.CONFLICT)
 205             file_pkg.seek(0, 0)
 206             if compressed == "gzip":
 207                 tar = tarfile.open(mode='r', fileobj=file_pkg)
 208                 descriptor_file_name = None
 209                 for tarinfo in tar:
 210                     tarname = tarinfo.name
 211                     tarname_path = tarname.split("/")
 212                     if not tarname_path[0] or ".." in tarname_path:  # if start with "/" means absolute path
 213                         raise EngineException("Absolute path or '..' are not allowed for package descriptor tar.gz")
 214                     if len(tarname_path) == 1 and not tarinfo.isdir():
 215                         raise EngineException("All files must be inside a dir for package descriptor tar.gz")
 216                     if tarname.endswith(".yaml") or tarname.endswith(".json") or tarname.endswith(".yml"):
 217                         storage["pkg-dir"] = tarname_path[0]
 218                         if len(tarname_path) == 2:
 219                             if descriptor_file_name:
 220                                 raise EngineException(
 221                                     "Found more than one descriptor file at package descriptor tar.gz")
 222                             descriptor_file_name = tarname
 223                 if not descriptor_file_name:
 224                     raise EngineException("Not found any descriptor file at package descriptor tar.gz")
 225                 storage["descriptor"] = descriptor_file_name
 226                 storage["zipfile"] = filename
 227                 self.fs.file_extract(tar, _id)
 228                 with self.fs.file_open((_id, descriptor_file_name), "r") as descriptor_file:
 229                     content = descriptor_file.read()
 230             else:
 231                 content = file_pkg.read()
 232                 storage["descriptor"] = descriptor_file_name = filename
 233
 234             if descriptor_file_name.endswith(".json"):
 235                 error_text = "Invalid json format "
 236                 indata = json.load(content)
 237             else:
 238                 error_text = "Invalid yaml format "
 239                 indata = yaml.load(content)
 240
 241             current_desc["_admin"]["storage"] = storage
 242             current_desc["_admin"]["onboardingState"] = "ONBOARDED"
 243             current_desc["_admin"]["operationalState"] = "ENABLED"
 244
 245             indata = self._remove_envelop(indata)
 246
 247             # Override descriptor with query string kwargs
 248             if kwargs:
 249                 self._update_input_with_kwargs(indata, kwargs)
 250             # it will call overrides method at VnfdTopic or NsdTopic
 251             indata = self._validate_input_new(indata, force=force)
 252
 253             deep_update_rfc7396(current_desc, indata)
 254             self.check_conflict_on_edit(session, current_desc, indata, _id=_id, force=force)
 255             self.db.replace(self.topic, _id, current_desc)
 256
 257             indata["_id"] = _id
 258             self._send_msg("created", indata)
 259
 260             # TODO if descriptor has changed because kwargs update content and remove cached zip
 261             # TODO if zip is not present creates one
 262             return True
 263
 264         except EngineException:
 265             raise
 266         except IndexError:
 267             raise EngineException("invalid Content-Range header format. Expected 'bytes start-end/total'",
 268                                   HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
 269         except IOError as e:
 270             raise EngineException("invalid upload transaction sequence: '{}'".format(e), HTTPStatus.BAD_REQUEST)
 271         except tarfile.ReadError as e:
 272             raise EngineException("invalid file content {}".format(e), HTTPStatus.BAD_REQUEST)
 273         except (ValueError, yaml.YAMLError) as e:
 274             raise EngineException(error_text + str(e))
 275         except ValidationError as e:
 276             raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
 277         finally:
 278             if file_pkg:
 279                 file_pkg.close()
 280
 281     def get_file(self, session, _id, path=None, accept_header=None):
 282         """
 283         Return the file content of a vnfd or nsd
 284         :param session: contains the used login username and working project
 285         :param _id: Identity of the vnfd, ndsd
 286         :param path: artifact path or "$DESCRIPTOR" or None
 287         :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
 288         :return: opened file or raises an exception
 289         """
 290         accept_text = accept_zip = False
 291         if accept_header:
 292             if 'text/plain' in accept_header or '*/*' in accept_header:
 293                 accept_text = True
 294             if 'application/zip' in accept_header or '*/*' in accept_header:
 295                 accept_zip = True
 296         if not accept_text and not accept_zip:
 297             raise EngineException("provide request header 'Accept' with 'application/zip' or 'text/plain'",
 298                                   http_code=HTTPStatus.NOT_ACCEPTABLE)
 299
 300         content = self.show(session, _id)
 301         if content["_admin"]["onboardingState"] != "ONBOARDED":
 302             raise EngineException("Cannot get content because this resource is not at 'ONBOARDED' state. "
 303                                   "onboardingState is {}".format(content["_admin"]["onboardingState"]),
 304                                   http_code=HTTPStatus.CONFLICT)
 305         storage = content["_admin"]["storage"]
 306         if path is not None and path != "$DESCRIPTOR":   # artifacts
 307             if not storage.get('pkg-dir'):
 308                 raise EngineException("Packages does not contains artifacts", http_code=HTTPStatus.BAD_REQUEST)
 309             if self.fs.file_exists((storage['folder'], storage['pkg-dir'], *path), 'dir'):
 310                 folder_content = self.fs.dir_ls((storage['folder'], storage['pkg-dir'], *path))
 311                 return folder_content, "text/plain"
 312                 # TODO manage folders in http
 313             else:
 314                 return self.fs.file_open((storage['folder'], storage['pkg-dir'], *path), "rb"),\
 315                     "application/octet-stream"
 316
 317         # pkgtype   accept  ZIP  TEXT    -> result
 318         # manyfiles         yes  X       -> zip
 319         #                   no   yes     -> error
 320         # onefile           yes  no      -> zip
 321         #                   X    yes     -> text
 322
 323         if accept_text and (not storage.get('pkg-dir') or path == "$DESCRIPTOR"):
 324             return self.fs.file_open((storage['folder'], storage['descriptor']), "r"), "text/plain"
 325         elif storage.get('pkg-dir') and not accept_zip:
 326             raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'"
 327                                   "Accept header", http_code=HTTPStatus.NOT_ACCEPTABLE)
 328         else:
 329             if not storage.get('zipfile'):
 330                 # TODO generate zipfile if not present
 331                 raise EngineException("Only allowed 'text/plain' Accept header for this descriptor. To be solved in "
 332                                       "future versions", http_code=HTTPStatus.NOT_ACCEPTABLE)
 333             return self.fs.file_open((storage['folder'], storage['zipfile']), "rb"), "application/zip"
 334
 335
 336 class VnfdTopic(DescriptorTopic):
 337     topic = "vnfds"
 338     topic_msg = "vnfd"
 339
 340     def __init__(self, db, fs, msg):
 341         DescriptorTopic.__init__(self, db, fs, msg)
 342
 343     @staticmethod
 344     def _remove_envelop(indata=None):
 345         if not indata:
 346             return {}
 347         clean_indata = indata
 348         if clean_indata.get('vnfd:vnfd-catalog'):
 349             clean_indata = clean_indata['vnfd:vnfd-catalog']
 350         elif clean_indata.get('vnfd-catalog'):
 351             clean_indata = clean_indata['vnfd-catalog']
 352         if clean_indata.get('vnfd'):
 353             if not isinstance(clean_indata['vnfd'], list) or len(clean_indata['vnfd']) != 1:
 354                 raise EngineException("'vnfd' must be a list only one element")
 355             clean_indata = clean_indata['vnfd'][0]
 356         return clean_indata
 357
 358     def check_conflict_on_del(self, session, _id, force=False):
 359         """
 360         Check that there is not any NSD that uses this VNFD. Only NSDs belonging to this project are considered. Note
 361         that VNFD can be public and be used by NSD of other projects. Also check there are not deployments, or vnfr
 362         that uses this vnfd
 363         :param session:
 364         :param _id: vnfd inernal id
 365         :param force: Avoid this checking
 366         :return: None or raises EngineException with the conflict
 367         """
 368         if force:
 369             return
 370         descriptor = self.db.get_one("vnfds", {"_id": _id})
 371         descriptor_id = descriptor.get("id")
 372         if not descriptor_id:  # empty vnfd not uploaded
 373             return
 374
 375         _filter = self._get_project_filter(session, write=False, show_all=False)
 376         # check vnfrs using this vnfd
 377         _filter["vnfd-id"] = _id
 378         if self.db.get_list("vnfrs", _filter):
 379             raise EngineException("There is some VNFR that depends on this VNFD", http_code=HTTPStatus.CONFLICT)
 380         del _filter["vnfd-id"]
 381         # check NSD using this VNFD
 382         _filter["constituent-vnfd.ANYINDEX.vnfd-id-ref"] = descriptor_id
 383         if self.db.get_list("nsds", _filter):
 384             raise EngineException("There is soame NSD that depends on this VNFD", http_code=HTTPStatus.CONFLICT)
 385
 386     def _validate_input_new(self, indata, force=False):
 387         # TODO validate with pyangbind, serialize
 388         return indata
 389
 390     def _validate_input_edit(self, indata, force=False):
 391         # TODO validate with pyangbind, serialize
 392         return indata
 393
 394
 395 class NsdTopic(DescriptorTopic):
 396     topic = "nsds"
 397     topic_msg = "nsd"
 398
 399     def __init__(self, db, fs, msg):
 400         DescriptorTopic.__init__(self, db, fs, msg)
 401
 402     @staticmethod
 403     def _remove_envelop(indata=None):
 404         if not indata:
 405             return {}
 406         clean_indata = indata
 407
 408         if clean_indata.get('nsd:nsd-catalog'):
 409             clean_indata = clean_indata['nsd:nsd-catalog']
 410         elif clean_indata.get('nsd-catalog'):
 411             clean_indata = clean_indata['nsd-catalog']
 412         if clean_indata.get('nsd'):
 413             if not isinstance(clean_indata['nsd'], list) or len(clean_indata['nsd']) != 1:
 414                 raise EngineException("'nsd' must be a list only one element")
 415             clean_indata = clean_indata['nsd'][0]
 416         return clean_indata
 417
 418     def _validate_input_new(self, indata, force=False):
 419         # transform constituent-vnfd:member-vnf-index to string
 420         if indata.get("constituent-vnfd"):
 421             for constituent_vnfd in indata["constituent-vnfd"]:
 422                 if "member-vnf-index" in constituent_vnfd:
 423                     constituent_vnfd["member-vnf-index"] = str(constituent_vnfd["member-vnf-index"])
 424
 425         # TODO validate with pyangbind, serialize
 426         return indata
 427
 428     def _validate_input_edit(self, indata, force=False):
 429         # TODO validate with pyangbind, serialize
 430         return indata
 431
 432     def _check_descriptor_dependencies(self, session, descriptor):
 433         """
 434         Check that the dependent descriptors exist on a new descriptor or edition
 435         :param session: client session information
 436         :param descriptor: descriptor to be inserted or edit
 437         :return: None or raises exception
 438         """
 439         if not descriptor.get("constituent-vnfd"):
 440             return
 441         for vnf in descriptor["constituent-vnfd"]:
 442             vnfd_id = vnf["vnfd-id-ref"]
 443             filter_q = self._get_project_filter(session, write=False, show_all=True)
 444             filter_q["id"] = vnfd_id
 445             if not self.db.get_list("vnfds", filter_q):
 446                 raise EngineException("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non "
 447                                       "existing vnfd".format(vnfd_id), http_code=HTTPStatus.CONFLICT)
 448
 449     def check_conflict_on_edit(self, session, final_content, edit_content, _id, force=False):
 450         super().check_conflict_on_edit(session, final_content, edit_content, _id, force=force)
 451
 452         self._check_descriptor_dependencies(session, final_content)
 453
 454     def check_conflict_on_del(self, session, _id, force=False):
 455         """
 456         Check that there is not any NSR that uses this NSD. Only NSRs belonging to this project are considered. Note
 457         that NSD can be public and be used by other projects.
 458         :param session:
 459         :param _id: vnfd inernal id
 460         :param force: Avoid this checking
 461         :return: None or raises EngineException with the conflict
 462         """
 463         if force:
 464             return
 465         _filter = self._get_project_filter(session, write=False, show_all=False)
 466         _filter["nsdId"] = _id
 467         if self.db.get_list("nsrs", _filter):
 468             raise EngineException("There is some NSR that depends on this NSD", http_code=HTTPStatus.CONFLICT)
 469
 470
 471 class PduTopic(BaseTopic):
 472     topic = "pdus"
 473     topic_msg = "pdu"
 474     schema_new = pdu_new_schema
 475     schema_edit = pdu_edit_schema
 476
 477     def __init__(self, db, fs, msg):
 478         BaseTopic.__init__(self, db, fs, msg)
 479
 480     @staticmethod
 481     def format_on_new(content, project_id=None, make_public=False):
 482         BaseTopic.format_on_new(content, project_id=None, make_public=make_public)
 483         content["_admin"]["onboardingState"] = "CREATED"
 484         content["_admin"]["operationalState"] = "DISABLED"
 485         content["_admin"]["usageSate"] = "NOT_IN_USE"
 486
 487     def check_conflict_on_del(self, session, _id, force=False):
 488         if force:
 489             return
 490         # TODO Is it needed to check descriptors _admin.project_read/project_write??
 491         _filter = {"vdur.pdu-id": _id}
 492         if self.db.get_list("vnfrs", _filter):
 493             raise EngineException("There is some NSR that uses this PDU", http_code=HTTPStatus.CONFLICT)