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