osm_nbi/base_topic.py

   1 # -*- coding: utf-8 -*-
   2
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #    http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # Unless required by applicable law or agreed to in writing, software
  10 # distributed under the License is distributed on an "AS IS" BASIS,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 import logging
  17 from uuid import uuid4
  18 from http import HTTPStatus
  19 from time import time
  20 from osm_common.dbbase import deep_update_rfc7396
  21 from osm_nbi.validation import validate_input, ValidationError, is_valid_uuid
  22 from yaml import safe_load, YAMLError
  23
  24 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
  25
  26
  27 class EngineException(Exception):
  28
  29     def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
  30         self.http_code = http_code
  31         super(Exception, self).__init__(message)
  32
  33
  34 def deep_get(target_dict, key_list):
  35     """
  36     Get a value from target_dict entering in the nested keys. If keys does not exist, it returns None
  37     Example target_dict={a: {b: 5}}; key_list=[a,b] returns 5; both key_list=[a,b,c] and key_list=[f,h] return None
  38     :param target_dict: dictionary to be read
  39     :param key_list: list of keys to read from  target_dict
  40     :return: The wanted value if exist, None otherwise
  41     """
  42     for key in key_list:
  43         if not isinstance(target_dict, dict) or key not in target_dict:
  44             return None
  45         target_dict = target_dict[key]
  46     return target_dict
  47
  48
  49 def get_iterable(input_var):
  50     """
  51     Returns an iterable, in case input_var is None it just returns an empty tuple
  52     :param input_var: can be a list, tuple or None
  53     :return: input_var or () if it is None
  54     """
  55     if input_var is None:
  56         return ()
  57     return input_var
  58
  59
  60 def versiontuple(v):
  61     """utility for compare dot separate versions. Fills with zeros to proper number comparison"""
  62     filled = []
  63     for point in v.split("."):
  64         filled.append(point.zfill(8))
  65     return tuple(filled)
  66
  67
  68 class BaseTopic:
  69     # static variables for all instance classes
  70     topic = None        # to_override
  71     topic_msg = None    # to_override
  72     schema_new = None   # to_override
  73     schema_edit = None  # to_override
  74     multiproject = True  # True if this Topic can be shared by several projects. Then it contains _admin.projects_read
  75
  76     default_quota = 500
  77
  78     # Alternative ID Fields for some Topics
  79     alt_id_field = {
  80         "projects": "name",
  81         "users": "username",
  82         "roles": "name"
  83     }
  84
  85     def __init__(self, db, fs, msg, auth):
  86         self.db = db
  87         self.fs = fs
  88         self.msg = msg
  89         self.logger = logging.getLogger("nbi.engine")
  90         self.auth = auth
  91
  92     @staticmethod
  93     def id_field(topic, value):
  94         """Returns ID Field for given topic and field value"""
  95         if topic in BaseTopic.alt_id_field.keys() and not is_valid_uuid(value):
  96             return BaseTopic.alt_id_field[topic]
  97         else:
  98             return "_id"
  99
 100     @staticmethod
 101     def _remove_envelop(indata=None):
 102         if not indata:
 103             return {}
 104         return indata
 105
 106     def check_quota(self, session):
 107         """
 108         Check whether topic quota is exceeded by the given project
 109         Used by relevant topics' 'new' function to decide whether or not creation of the new item should be allowed
 110         :param projects: projects (tuple) for which quota should be checked
 111         :param override: boolean. If true, don't raise ValidationError even though quota be exceeded
 112         :return: None
 113         :raise:
 114             DbException if project not found
 115             ValidationError if quota exceeded and not overridden
 116         """
 117         if session["force"]:
 118             return
 119         projects = session["project_id"]
 120         for project in projects:
 121             proj = self.auth.get_project(project)
 122             pid = proj["_id"]
 123             quota = proj.get("quotas", {}).get(self.topic, self.default_quota)
 124             count = self.db.count(self.topic, {"_admin.projects_read": pid})
 125             if count >= quota:
 126                 name = proj["name"]
 127                 raise ValidationError("quota ({}={}) exceeded for project {} ({})".format(self.topic, quota, name, pid),
 128                                       http_code=HTTPStatus.UNAUTHORIZED)
 129
 130     def _validate_input_new(self, input, force=False):
 131         """
 132         Validates input user content for a new entry. It uses jsonschema. Some overrides will use pyangbind
 133         :param input: user input content for the new topic
 134         :param force: may be used for being more tolerant
 135         :return: The same input content, or a changed version of it.
 136         """
 137         if self.schema_new:
 138             validate_input(input, self.schema_new)
 139         return input
 140
 141     def _validate_input_edit(self, input, force=False):
 142         """
 143         Validates input user content for an edition. It uses jsonschema. Some overrides will use pyangbind
 144         :param input: user input content for the new topic
 145         :param force: may be used for being more tolerant
 146         :return: The same input content, or a changed version of it.
 147         """
 148         if self.schema_edit:
 149             validate_input(input, self.schema_edit)
 150         return input
 151
 152     @staticmethod
 153     def _get_project_filter(session):
 154         """
 155         Generates a filter dictionary for querying database, so that only allowed items for this project can be
 156         addressed. Only proprietary or public can be used. Allowed projects are at _admin.project_read/write. If it is
 157         not present or contains ANY mean public.
 158         :param session: contains:
 159             project_id: project list this session has rights to access. Can be empty, one or several
 160             set_project: items created will contain this project list
 161             force: True or False
 162             public: True, False or None
 163             method: "list", "show", "write", "delete"
 164             admin: True or False
 165         :return: dictionary with project filter
 166         """
 167         p_filter = {}
 168         project_filter_n = []
 169         project_filter = list(session["project_id"])
 170
 171         if session["method"] not in ("list", "delete"):
 172             if project_filter:
 173                 project_filter.append("ANY")
 174         elif session["public"] is not None:
 175             if session["public"]:
 176                 project_filter.append("ANY")
 177             else:
 178                 project_filter_n.append("ANY")
 179
 180         if session.get("PROJECT.ne"):
 181             project_filter_n.append(session["PROJECT.ne"])
 182
 183         if project_filter:
 184             if session["method"] in ("list", "show", "delete") or session.get("set_project"):
 185                 p_filter["_admin.projects_read.cont"] = project_filter
 186             else:
 187                 p_filter["_admin.projects_write.cont"] = project_filter
 188         if project_filter_n:
 189             if session["method"] in ("list", "show", "delete") or session.get("set_project"):
 190                 p_filter["_admin.projects_read.ncont"] = project_filter_n
 191             else:
 192                 p_filter["_admin.projects_write.ncont"] = project_filter_n
 193
 194         return p_filter
 195
 196     def check_conflict_on_new(self, session, indata):
 197         """
 198         Check that the data to be inserted is valid
 199         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 200         :param indata: data to be inserted
 201         :return: None or raises EngineException
 202         """
 203         pass
 204
 205     def check_conflict_on_edit(self, session, final_content, edit_content, _id):
 206         """
 207         Check that the data to be edited/uploaded is valid
 208         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 209         :param final_content: data once modified. This method may change it.
 210         :param edit_content: incremental data that contains the modifications to apply
 211         :param _id: internal _id
 212         :return: None or raises EngineException
 213         """
 214         if not self.multiproject:
 215             return
 216         # Change public status
 217         if session["public"] is not None:
 218             if session["public"] and "ANY" not in final_content["_admin"]["projects_read"]:
 219                 final_content["_admin"]["projects_read"].append("ANY")
 220                 final_content["_admin"]["projects_write"].clear()
 221             if not session["public"] and "ANY" in final_content["_admin"]["projects_read"]:
 222                 final_content["_admin"]["projects_read"].remove("ANY")
 223
 224         # Change project status
 225         if session.get("set_project"):
 226             for p in session["set_project"]:
 227                 if p not in final_content["_admin"]["projects_read"]:
 228                     final_content["_admin"]["projects_read"].append(p)
 229
 230     def check_unique_name(self, session, name, _id=None):
 231         """
 232         Check that the name is unique for this project
 233         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 234         :param name: name to be checked
 235         :param _id: If not None, ignore this entry that are going to change
 236         :return: None or raises EngineException
 237         """
 238         if not self.multiproject:
 239             _filter = {}
 240         else:
 241             _filter = self._get_project_filter(session)
 242         _filter["name"] = name
 243         if _id:
 244             _filter["_id.neq"] = _id
 245         if self.db.get_one(self.topic, _filter, fail_on_empty=False, fail_on_more=False):
 246             raise EngineException("name '{}' already exists for {}".format(name, self.topic), HTTPStatus.CONFLICT)
 247
 248     @staticmethod
 249     def format_on_new(content, project_id=None, make_public=False):
 250         """
 251         Modifies content descriptor to include _admin
 252         :param content: descriptor to be modified
 253         :param project_id: if included, it add project read/write permissions. Can be None or a list
 254         :param make_public: if included it is generated as public for reading.
 255         :return: op_id: operation id on asynchronous operation, None otherwise. In addition content is modified
 256         """
 257         now = time()
 258         if "_admin" not in content:
 259             content["_admin"] = {}
 260         if not content["_admin"].get("created"):
 261             content["_admin"]["created"] = now
 262         content["_admin"]["modified"] = now
 263         if not content.get("_id"):
 264             content["_id"] = str(uuid4())
 265         if project_id is not None:
 266             if not content["_admin"].get("projects_read"):
 267                 content["_admin"]["projects_read"] = list(project_id)
 268                 if make_public:
 269                     content["_admin"]["projects_read"].append("ANY")
 270             if not content["_admin"].get("projects_write"):
 271                 content["_admin"]["projects_write"] = list(project_id)
 272         return None
 273
 274     @staticmethod
 275     def format_on_edit(final_content, edit_content):
 276         """
 277         Modifies final_content to admin information upon edition
 278         :param final_content: final content to be stored at database
 279         :param edit_content: user requested update content
 280         :return: operation id, if this edit implies an asynchronous operation; None otherwise
 281         """
 282         if final_content.get("_admin"):
 283             now = time()
 284             final_content["_admin"]["modified"] = now
 285         return None
 286
 287     def _send_msg(self, action, content, not_send_msg=None):
 288         if self.topic_msg and not_send_msg is not False:
 289             content.pop("_admin", None)
 290             if isinstance(not_send_msg, list):
 291                 not_send_msg.append((self.topic_msg, action, content))
 292             else:
 293                 self.msg.write(self.topic_msg, action, content)
 294
 295     def check_conflict_on_del(self, session, _id, db_content):
 296         """
 297         Check if deletion can be done because of dependencies if it is not force. To override
 298         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 299         :param _id: internal _id
 300         :param db_content: The database content of this item _id
 301         :return: None if ok or raises EngineException with the conflict
 302         """
 303         pass
 304
 305     @staticmethod
 306     def _update_input_with_kwargs(desc, kwargs, yaml_format=False):
 307         """
 308         Update descriptor with the kwargs. It contains dot separated keys
 309         :param desc: dictionary to be updated
 310         :param kwargs: plain dictionary to be used for updating.
 311         :param yaml_format: get kwargs values as yaml format.
 312         :return: None, 'desc' is modified. It raises EngineException.
 313         """
 314         if not kwargs:
 315             return
 316         try:
 317             for k, v in kwargs.items():
 318                 update_content = desc
 319                 kitem_old = None
 320                 klist = k.split(".")
 321                 for kitem in klist:
 322                     if kitem_old is not None:
 323                         update_content = update_content[kitem_old]
 324                     if isinstance(update_content, dict):
 325                         kitem_old = kitem
 326                     elif isinstance(update_content, list):
 327                         kitem_old = int(kitem)
 328                     else:
 329                         raise EngineException(
 330                             "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem))
 331                 update_content[kitem_old] = v if not yaml_format else safe_load(v)
 332         except KeyError:
 333             raise EngineException(
 334                 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old))
 335         except ValueError:
 336             raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
 337                 k, kitem))
 338         except IndexError:
 339             raise EngineException(
 340                 "Invalid query string '{}'. Index '{}' out of  range".format(k, kitem_old))
 341         except YAMLError:
 342             raise EngineException("Invalid query string '{}' yaml format".format(k))
 343
 344     def show(self, session, _id):
 345         """
 346         Get complete information on an topic
 347         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 348         :param _id: server internal id
 349         :return: dictionary, raise exception if not found.
 350         """
 351         if not self.multiproject:
 352             filter_db = {}
 353         else:
 354             filter_db = self._get_project_filter(session)
 355         # To allow project&user addressing by name AS WELL AS _id
 356         filter_db[BaseTopic.id_field(self.topic, _id)] = _id
 357         return self.db.get_one(self.topic, filter_db)
 358         # TODO transform data for SOL005 URL requests
 359         # TODO remove _admin if not admin
 360
 361     def get_file(self, session, _id, path=None, accept_header=None):
 362         """
 363         Only implemented for descriptor topics. Return the file content of a descriptor
 364         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 365         :param _id: Identity of the item to get content
 366         :param path: artifact path or "$DESCRIPTOR" or None
 367         :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
 368         :return: opened file or raises an exception
 369         """
 370         raise EngineException("Method get_file not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR)
 371
 372     def list(self, session, filter_q=None):
 373         """
 374         Get a list of the topic that matches a filter
 375         :param session: contains the used login username and working project
 376         :param filter_q: filter of data to be applied
 377         :return: The list, it can be empty if no one match the filter.
 378         """
 379         if not filter_q:
 380             filter_q = {}
 381         if self.multiproject:
 382             filter_q.update(self._get_project_filter(session))
 383
 384         # TODO transform data for SOL005 URL requests. Transform filtering
 385         # TODO implement "field-type" query string SOL005
 386         return self.db.get_list(self.topic, filter_q)
 387
 388     def new(self, rollback, session, indata=None, kwargs=None, headers=None):
 389         """
 390         Creates a new entry into database.
 391         :param rollback: list to append created items at database in case a rollback may to be done
 392         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 393         :param indata: data to be inserted
 394         :param kwargs: used to override the indata descriptor
 395         :param headers: http request headers
 396         :return: _id, op_id:
 397             _id: identity of the inserted data.
 398              op_id: operation id if this is asynchronous, None otherwise
 399         """
 400         try:
 401             if self.multiproject:
 402                 self.check_quota(session)
 403
 404             content = self._remove_envelop(indata)
 405
 406             # Override descriptor with query string kwargs
 407             self._update_input_with_kwargs(content, kwargs)
 408             content = self._validate_input_new(content, force=session["force"])
 409             self.check_conflict_on_new(session, content)
 410             op_id = self.format_on_new(content, project_id=session["project_id"], make_public=session["public"])
 411             _id = self.db.create(self.topic, content)
 412             rollback.append({"topic": self.topic, "_id": _id})
 413             if op_id:
 414                 content["op_id"] = op_id
 415             self._send_msg("created", content)
 416             return _id, op_id
 417         except ValidationError as e:
 418             raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
 419
 420     def upload_content(self, session, _id, indata, kwargs, headers):
 421         """
 422         Only implemented for descriptor topics.  Used for receiving content by chunks (with a transaction_id header
 423         and/or gzip file. It will store and extract)
 424         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 425         :param _id : the database id of entry to be updated
 426         :param indata: http body request
 427         :param kwargs: user query string to override parameters. NOT USED
 428         :param headers:  http request headers
 429         :return: True package has is completely uploaded or False if partial content has been uplodaed.
 430             Raise exception on error
 431         """
 432         raise EngineException("Method upload_content not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR)
 433
 434     def delete_list(self, session, filter_q=None):
 435         """
 436         Delete a several entries of a topic. This is for internal usage and test only, not exposed to NBI API
 437         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 438         :param filter_q: filter of data to be applied
 439         :return: The deleted list, it can be empty if no one match the filter.
 440         """
 441         # TODO add admin to filter, validate rights
 442         if not filter_q:
 443             filter_q = {}
 444         if self.multiproject:
 445             filter_q.update(self._get_project_filter(session))
 446         return self.db.del_list(self.topic, filter_q)
 447
 448     def delete_extra(self, session, _id, db_content, not_send_msg=None):
 449         """
 450         Delete other things apart from database entry of a item _id.
 451         e.g.: other associated elements at database and other file system storage
 452         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 453         :param _id: server internal id
 454         :param db_content: The database content of the _id. It is already deleted when reached this method, but the
 455             content is needed in same cases
 456         :param not_send_msg: To not send message (False) or store content (list) instead
 457         :return: None if ok or raises EngineException with the problem
 458         """
 459         pass
 460
 461     def delete(self, session, _id, dry_run=False, not_send_msg=None):
 462         """
 463         Delete item by its internal _id
 464         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 465         :param _id: server internal id
 466         :param dry_run: make checking but do not delete
 467         :param not_send_msg: To not send message (False) or store content (list) instead
 468         :return: operation id (None if there is not operation), raise exception if error or not found, conflict, ...
 469         """
 470
 471         # To allow addressing projects and users by name AS WELL AS by _id
 472         if not self.multiproject:
 473             filter_q = {}
 474         else:
 475             filter_q = self._get_project_filter(session)
 476         filter_q[self.id_field(self.topic, _id)] = _id
 477         item_content = self.db.get_one(self.topic, filter_q)
 478
 479         self.check_conflict_on_del(session, _id, item_content)
 480         if dry_run:
 481             return None
 482
 483         if self.multiproject and session["project_id"]:
 484             # remove reference from project_read if there are more projects referencing it. If it last one,
 485             # do not remove reference, but delete
 486             other_projects_referencing = next((p for p in item_content["_admin"]["projects_read"]
 487                                                if p not in session["project_id"]), None)
 488
 489             # check if there are projects referencing it (apart from ANY, that means, public)....
 490             if other_projects_referencing:
 491                 # remove references but not delete
 492                 update_dict_pull = {"_admin.projects_read.{}".format(p): None for p in session["project_id"]}
 493                 update_dict_pull.update({"_admin.projects_write.{}".format(p): None for p in session["project_id"]})
 494                 self.db.set_one(self.topic, filter_q, update_dict=None, pull=update_dict_pull)
 495                 return None
 496             else:
 497                 can_write = next((p for p in item_content["_admin"]["projects_write"] if p == "ANY" or
 498                                   p in session["project_id"]), None)
 499                 if not can_write:
 500                     raise EngineException("You have not write permission to delete it",
 501                                           http_code=HTTPStatus.UNAUTHORIZED)
 502
 503         # delete
 504         self.db.del_one(self.topic, filter_q)
 505         self.delete_extra(session, _id, item_content, not_send_msg=not_send_msg)
 506         self._send_msg("deleted", {"_id": _id}, not_send_msg=not_send_msg)
 507         return None
 508
 509     def edit(self, session, _id, indata=None, kwargs=None, content=None):
 510         """
 511         Change the content of an item
 512         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 513         :param _id: server internal id
 514         :param indata: contains the changes to apply
 515         :param kwargs: modifies indata
 516         :param content: original content of the item
 517         :return: op_id: operation id if this is processed asynchronously, None otherwise
 518         """
 519         indata = self._remove_envelop(indata)
 520
 521         # Override descriptor with query string kwargs
 522         if kwargs:
 523             self._update_input_with_kwargs(indata, kwargs)
 524         try:
 525             if indata and session.get("set_project"):
 526                 raise EngineException("Cannot edit content and set to project (query string SET_PROJECT) at same time",
 527                                       HTTPStatus.UNPROCESSABLE_ENTITY)
 528             indata = self._validate_input_edit(indata, force=session["force"])
 529
 530             # TODO self._check_edition(session, indata, _id, force)
 531             if not content:
 532                 content = self.show(session, _id)
 533             deep_update_rfc7396(content, indata)
 534
 535             # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
 536             _id = content.get("_id") or _id
 537
 538             self.check_conflict_on_edit(session, content, indata, _id=_id)
 539             op_id = self.format_on_edit(content, indata)
 540
 541             self.db.replace(self.topic, _id, content)
 542
 543             indata.pop("_admin", None)
 544             if op_id:
 545                 indata["op_id"] = op_id
 546             indata["_id"] = _id
 547             self._send_msg("edited", indata)
 548             return op_id
 549         except ValidationError as e:
 550             raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)