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 = content.copy()
 312             content.pop("_admin", None)
 313             if isinstance(not_send_msg, list):
 314                 not_send_msg.append((self.topic_msg, action, content))
 315             else:
 316                 self.msg.write(self.topic_msg, action, content)
 317
 318     def check_conflict_on_del(self, session, _id, db_content):
 319         """
 320         Check if deletion can be done because of dependencies if it is not force. To override
 321         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 322         :param _id: internal _id
 323         :param db_content: The database content of this item _id
 324         :return: None if ok or raises EngineException with the conflict
 325         """
 326         pass
 327
 328     @staticmethod
 329     def _update_input_with_kwargs(desc, kwargs, yaml_format=False):
 330         """
 331         Update descriptor with the kwargs. It contains dot separated keys
 332         :param desc: dictionary to be updated
 333         :param kwargs: plain dictionary to be used for updating.
 334         :param yaml_format: get kwargs values as yaml format.
 335         :return: None, 'desc' is modified. It raises EngineException.
 336         """
 337         if not kwargs:
 338             return
 339         try:
 340             for k, v in kwargs.items():
 341                 update_content = desc
 342                 kitem_old = None
 343                 klist = k.split(".")
 344                 for kitem in klist:
 345                     if kitem_old is not None:
 346                         update_content = update_content[kitem_old]
 347                     if isinstance(update_content, dict):
 348                         kitem_old = kitem
 349                         if not isinstance(update_content.get(kitem_old), (dict, list)):
 350                             update_content[kitem_old] = {}
 351                     elif isinstance(update_content, list):
 352                         # key must be an index of the list, must be integer
 353                         kitem_old = int(kitem)
 354                         # if index greater than list, extend the list
 355                         if kitem_old >= len(update_content):
 356                             update_content += [None] * (kitem_old - len(update_content) + 1)
 357                         if not isinstance(update_content[kitem_old], (dict, list)):
 358                             update_content[kitem_old] = {}
 359                     else:
 360                         raise EngineException(
 361                             "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem))
 362                 if v is None:
 363                     del update_content[kitem_old]
 364                 else:
 365                     update_content[kitem_old] = v if not yaml_format else safe_load(v)
 366         except KeyError:
 367             raise EngineException(
 368                 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old))
 369         except ValueError:
 370             raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
 371                 k, kitem))
 372         except IndexError:
 373             raise EngineException(
 374                 "Invalid query string '{}'. Index '{}' out of  range".format(k, kitem_old))
 375         except YAMLError:
 376             raise EngineException("Invalid query string '{}' yaml format".format(k))
 377
 378     def sol005_projection(self, data):
 379         # Projection was moved to child classes
 380         return data
 381
 382     def show(self, session, _id, api_req=False):
 383         """
 384         Get complete information on an topic
 385         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 386         :param _id: server internal id
 387         :param api_req: True if this call is serving an external API request. False if serving internal request.
 388         :return: dictionary, raise exception if not found.
 389         """
 390         if not self.multiproject:
 391             filter_db = {}
 392         else:
 393             filter_db = self._get_project_filter(session)
 394         # To allow project&user addressing by name AS WELL AS _id
 395         filter_db[BaseTopic.id_field(self.topic, _id)] = _id
 396         data = self.db.get_one(self.topic, filter_db)
 397
 398         # Only perform SOL005 projection if we are serving an external request
 399         if api_req:
 400             self.sol005_projection(data)
 401
 402         return data
 403
 404         # TODO transform data for SOL005 URL requests
 405         # TODO remove _admin if not admin
 406
 407     def get_file(self, session, _id, path=None, accept_header=None):
 408         """
 409         Only implemented for descriptor topics. Return the file content of a descriptor
 410         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 411         :param _id: Identity of the item to get content
 412         :param path: artifact path or "$DESCRIPTOR" or None
 413         :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
 414         :return: opened file or raises an exception
 415         """
 416         raise EngineException("Method get_file not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR)
 417
 418     def list(self, session, filter_q=None, api_req=False):
 419         """
 420         Get a list of the topic that matches a filter
 421         :param session: contains the used login username and working project
 422         :param filter_q: filter of data to be applied
 423         :param api_req: True if this call is serving an external API request. False if serving internal request.
 424         :return: The list, it can be empty if no one match the filter.
 425         """
 426         if not filter_q:
 427             filter_q = {}
 428         if self.multiproject:
 429             filter_q.update(self._get_project_filter(session))
 430
 431         # TODO transform data for SOL005 URL requests. Transform filtering
 432         # TODO implement "field-type" query string SOL005
 433         data = self.db.get_list(self.topic, filter_q)
 434
 435         # Only perform SOL005 projection if we are serving an external request
 436         if api_req:
 437             data = [self.sol005_projection(inst) for inst in data]
 438
 439         return data
 440
 441     def new(self, rollback, session, indata=None, kwargs=None, headers=None):
 442         """
 443         Creates a new entry into database.
 444         :param rollback: list to append created items at database in case a rollback may to be done
 445         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 446         :param indata: data to be inserted
 447         :param kwargs: used to override the indata descriptor
 448         :param headers: http request headers
 449         :return: _id, op_id:
 450             _id: identity of the inserted data.
 451              op_id: operation id if this is asynchronous, None otherwise
 452         """
 453         try:
 454             if self.multiproject:
 455                 self.check_quota(session)
 456
 457             content = self._remove_envelop(indata)
 458
 459             # Override descriptor with query string kwargs
 460             self._update_input_with_kwargs(content, kwargs)
 461             content = self._validate_input_new(content, force=session["force"])
 462             self.check_conflict_on_new(session, content)
 463             op_id = self.format_on_new(content, project_id=session["project_id"], make_public=session["public"])
 464             _id = self.db.create(self.topic, content)
 465             rollback.append({"topic": self.topic, "_id": _id})
 466             if op_id:
 467                 content["op_id"] = op_id
 468             self._send_msg("created", content)
 469             return _id, op_id
 470         except ValidationError as e:
 471             raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
 472
 473     def upload_content(self, session, _id, indata, kwargs, headers):
 474         """
 475         Only implemented for descriptor topics.  Used for receiving content by chunks (with a transaction_id header
 476         and/or gzip file. It will store and extract)
 477         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 478         :param _id : the database id of entry to be updated
 479         :param indata: http body request
 480         :param kwargs: user query string to override parameters. NOT USED
 481         :param headers:  http request headers
 482         :return: True package has is completely uploaded or False if partial content has been uplodaed.
 483             Raise exception on error
 484         """
 485         raise EngineException("Method upload_content not valid for this topic", HTTPStatus.INTERNAL_SERVER_ERROR)
 486
 487     def delete_list(self, session, filter_q=None):
 488         """
 489         Delete a several entries of a topic. This is for internal usage and test only, not exposed to NBI API
 490         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 491         :param filter_q: filter of data to be applied
 492         :return: The deleted list, it can be empty if no one match the filter.
 493         """
 494         # TODO add admin to filter, validate rights
 495         if not filter_q:
 496             filter_q = {}
 497         if self.multiproject:
 498             filter_q.update(self._get_project_filter(session))
 499         return self.db.del_list(self.topic, filter_q)
 500
 501     def delete_extra(self, session, _id, db_content, not_send_msg=None):
 502         """
 503         Delete other things apart from database entry of a item _id.
 504         e.g.: other associated elements at database and other file system storage
 505         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 506         :param _id: server internal id
 507         :param db_content: The database content of the _id. It is already deleted when reached this method, but the
 508             content is needed in same cases
 509         :param not_send_msg: To not send message (False) or store content (list) instead
 510         :return: None if ok or raises EngineException with the problem
 511         """
 512         pass
 513
 514     def delete(self, session, _id, dry_run=False, not_send_msg=None):
 515         """
 516         Delete item by its internal _id
 517         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 518         :param _id: server internal id
 519         :param dry_run: make checking but do not delete
 520         :param not_send_msg: To not send message (False) or store content (list) instead
 521         :return: operation id (None if there is not operation), raise exception if error or not found, conflict, ...
 522         """
 523
 524         # To allow addressing projects and users by name AS WELL AS by _id
 525         if not self.multiproject:
 526             filter_q = {}
 527         else:
 528             filter_q = self._get_project_filter(session)
 529         filter_q[self.id_field(self.topic, _id)] = _id
 530         item_content = self.db.get_one(self.topic, filter_q)
 531
 532         self.check_conflict_on_del(session, _id, item_content)
 533         if dry_run:
 534             return None
 535
 536         if self.multiproject and session["project_id"]:
 537             # remove reference from project_read if there are more projects referencing it. If it last one,
 538             # do not remove reference, but delete
 539             other_projects_referencing = next((p for p in item_content["_admin"]["projects_read"]
 540                                                if p not in session["project_id"] and p != "ANY"), None)
 541
 542             # check if there are projects referencing it (apart from ANY, that means, public)....
 543             if other_projects_referencing:
 544                 # remove references but not delete
 545                 update_dict_pull = {"_admin.projects_read": session["project_id"],
 546                                     "_admin.projects_write": session["project_id"]}
 547                 self.db.set_one(self.topic, filter_q, update_dict=None, pull_list=update_dict_pull)
 548                 return None
 549             else:
 550                 can_write = next((p for p in item_content["_admin"]["projects_write"] if p == "ANY" or
 551                                   p in session["project_id"]), None)
 552                 if not can_write:
 553                     raise EngineException("You have not write permission to delete it",
 554                                           http_code=HTTPStatus.UNAUTHORIZED)
 555
 556         # delete
 557         self.db.del_one(self.topic, filter_q)
 558         self.delete_extra(session, _id, item_content, not_send_msg=not_send_msg)
 559         self._send_msg("deleted", {"_id": _id}, not_send_msg=not_send_msg)
 560         return None
 561
 562     def edit(self, session, _id, indata=None, kwargs=None, content=None):
 563         """
 564         Change the content of an item
 565         :param session: contains "username", "admin", "force", "public", "project_id", "set_project"
 566         :param _id: server internal id
 567         :param indata: contains the changes to apply
 568         :param kwargs: modifies indata
 569         :param content: original content of the item
 570         :return: op_id: operation id if this is processed asynchronously, None otherwise
 571         """
 572         indata = self._remove_envelop(indata)
 573
 574         # Override descriptor with query string kwargs
 575         if kwargs:
 576             self._update_input_with_kwargs(indata, kwargs)
 577         try:
 578             if indata and session.get("set_project"):
 579                 raise EngineException("Cannot edit content and set to project (query string SET_PROJECT) at same time",
 580                                       HTTPStatus.UNPROCESSABLE_ENTITY)
 581
 582             # TODO self._check_edition(session, indata, _id, force)
 583             if not content:
 584                 content = self.show(session, _id)
 585
 586             indata = self._validate_input_edit(indata, content, force=session["force"])
 587
 588             deep_update_rfc7396(content, indata)
 589
 590             # To allow project addressing by name AS WELL AS _id. Get the _id, just in case the provided one is a name
 591             _id = content.get("_id") or _id
 592
 593             self.check_conflict_on_edit(session, content, indata, _id=_id)
 594             op_id = self.format_on_edit(content, indata)
 595
 596             self.db.replace(self.topic, _id, content)
 597
 598             indata.pop("_admin", None)
 599             if op_id:
 600                 indata["op_id"] = op_id
 601             indata["_id"] = _id
 602             self._send_msg("edited", indata)
 603             return op_id
 604         except ValidationError as e:
 605             raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)