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