Added NS parameter validation
[osm/NBI.git] / osm_nbi / engine.py
1 # -*- coding: utf-8 -*-
2
3 from osm_common import dbmongo
4 from osm_common import dbmemory
5 from osm_common import fslocal
6 from osm_common import msglocal
7 from osm_common import msgkafka
8 import tarfile
9 import yaml
10 import json
11 import logging
12 from random import choice as random_choice
13 from uuid import uuid4
14 from hashlib import sha256, md5
15 from osm_common.dbbase import DbException
16 from osm_common.fsbase import FsException
17 from osm_common.msgbase import MsgException
18 from http import HTTPStatus
19 from time import time
20 from copy import deepcopy
21 from validation import validate_input, ValidationError
22
23 __author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
24
25
26 class EngineException(Exception):
27
28 def __init__(self, message, http_code=HTTPStatus.BAD_REQUEST):
29 self.http_code = http_code
30 Exception.__init__(self, message)
31
32
33 def _deep_update(dict_to_change, dict_reference):
34 """
35 Modifies one dictionary with the information of the other following https://tools.ietf.org/html/rfc7396
36 :param dict_to_change: Ends modified
37 :param dict_reference: reference
38 :return: none
39 """
40 for k in dict_reference:
41 if dict_reference[k] is None: # None->Anything
42 if k in dict_to_change:
43 del dict_to_change[k]
44 elif not isinstance(dict_reference[k], dict): # NotDict->Anything
45 dict_to_change[k] = dict_reference[k]
46 elif k not in dict_to_change: # Dict->Empty
47 dict_to_change[k] = deepcopy(dict_reference[k])
48 _deep_update(dict_to_change[k], dict_reference[k])
49 elif isinstance(dict_to_change[k], dict): # Dict->Dict
50 _deep_update(dict_to_change[k], dict_reference[k])
51 else: # Dict->NotDict
52 dict_to_change[k] = deepcopy(dict_reference[k])
53 _deep_update(dict_to_change[k], dict_reference[k])
54
55
56 class Engine(object):
57
58 def __init__(self):
59 self.tokens = {}
60 self.db = None
61 self.fs = None
62 self.msg = None
63 self.config = None
64 self.logger = logging.getLogger("nbi.engine")
65
66 def start(self, config):
67 """
68 Connect to database, filesystem storage, and messaging
69 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
70 :return: None
71 """
72 self.config = config
73 try:
74 if not self.db:
75 if config["database"]["driver"] == "mongo":
76 self.db = dbmongo.DbMongo()
77 self.db.db_connect(config["database"])
78 elif config["database"]["driver"] == "memory":
79 self.db = dbmemory.DbMemory()
80 self.db.db_connect(config["database"])
81 else:
82 raise EngineException("Invalid configuration param '{}' at '[database]':'driver'".format(
83 config["database"]["driver"]))
84 if not self.fs:
85 if config["storage"]["driver"] == "local":
86 self.fs = fslocal.FsLocal()
87 self.fs.fs_connect(config["storage"])
88 else:
89 raise EngineException("Invalid configuration param '{}' at '[storage]':'driver'".format(
90 config["storage"]["driver"]))
91 if not self.msg:
92 if config["message"]["driver"] == "local":
93 self.msg = msglocal.MsgLocal()
94 self.msg.connect(config["message"])
95 elif config["message"]["driver"] == "kafka":
96 self.msg = msgkafka.MsgKafka()
97 self.msg.connect(config["message"])
98 else:
99 raise EngineException("Invalid configuration param '{}' at '[message]':'driver'".format(
100 config["storage"]["driver"]))
101 except (DbException, FsException, MsgException) as e:
102 raise EngineException(str(e), http_code=e.http_code)
103
104 def stop(self):
105 try:
106 if self.db:
107 self.db.db_disconnect()
108 if self.fs:
109 self.fs.fs_disconnect()
110 if self.fs:
111 self.fs.fs_disconnect()
112 except (DbException, FsException, MsgException) as e:
113 raise EngineException(str(e), http_code=e.http_code)
114
115 def authorize(self, token):
116 try:
117 if not token:
118 raise EngineException("Needed a token or Authorization http header",
119 http_code=HTTPStatus.UNAUTHORIZED)
120 if token not in self.tokens:
121 raise EngineException("Invalid token or Authorization http header",
122 http_code=HTTPStatus.UNAUTHORIZED)
123 session = self.tokens[token]
124 now = time()
125 if session["expires"] < now:
126 del self.tokens[token]
127 raise EngineException("Expired Token or Authorization http header",
128 http_code=HTTPStatus.UNAUTHORIZED)
129 return session
130 except EngineException:
131 if self.config["global"].get("test.user_not_authorized"):
132 return {"id": "fake-token-id-for-test",
133 "project_id": self.config["global"].get("test.project_not_authorized", "admin"),
134 "username": self.config["global"]["test.user_not_authorized"]}
135 else:
136 raise
137
138 def new_token(self, session, indata, remote):
139 now = time()
140 user_content = None
141
142 # Try using username/password
143 if indata.get("username"):
144 user_rows = self.db.get_list("users", {"username": indata.get("username")})
145 user_content = None
146 if user_rows:
147 user_content = user_rows[0]
148 salt = user_content["_admin"]["salt"]
149 shadow_password = sha256(indata.get("password", "").encode('utf-8') + salt.encode('utf-8')).hexdigest()
150 if shadow_password != user_content["password"]:
151 user_content = None
152 if not user_content:
153 raise EngineException("Invalid username/password", http_code=HTTPStatus.UNAUTHORIZED)
154 elif session:
155 user_rows = self.db.get_list("users", {"username": session["username"]})
156 if user_rows:
157 user_content = user_rows[0]
158 else:
159 raise EngineException("Invalid token", http_code=HTTPStatus.UNAUTHORIZED)
160 else:
161 raise EngineException("Provide credentials: username/password or Authorization Bearer token",
162 http_code=HTTPStatus.UNAUTHORIZED)
163
164 token_id = ''.join(random_choice('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789')
165 for _ in range(0, 32))
166 if indata.get("project_id"):
167 project_id = indata.get("project_id")
168 if project_id not in user_content["projects"]:
169 raise EngineException("project {} not allowed for this user".format(project_id),
170 http_code=HTTPStatus.UNAUTHORIZED)
171 else:
172 project_id = user_content["projects"][0]
173 if project_id == "admin":
174 session_admin = True
175 else:
176 project = self.db.get_one("projects", {"_id": project_id})
177 session_admin = project.get("admin", False)
178 new_session = {"issued_at": now, "expires": now+3600,
179 "_id": token_id, "id": token_id, "project_id": project_id, "username": user_content["username"],
180 "remote_port": remote.port, "admin": session_admin}
181 if remote.name:
182 new_session["remote_host"] = remote.name
183 elif remote.ip:
184 new_session["remote_host"] = remote.ip
185
186 self.tokens[token_id] = new_session
187 return deepcopy(new_session)
188
189 def get_token_list(self, session):
190 token_list = []
191 for token_id, token_value in self.tokens.items():
192 if token_value["username"] == session["username"]:
193 token_list.append(deepcopy(token_value))
194 return token_list
195
196 def get_token(self, session, token_id):
197 token_value = self.tokens.get(token_id)
198 if not token_value:
199 raise EngineException("token not found", http_code=HTTPStatus.NOT_FOUND)
200 if token_value["username"] != session["username"] and not session["admin"]:
201 raise EngineException("needed admin privileges", http_code=HTTPStatus.UNAUTHORIZED)
202 return token_value
203
204 def del_token(self, token_id):
205 try:
206 del self.tokens[token_id]
207 return "token '{}' deleted".format(token_id)
208 except KeyError:
209 raise EngineException("Token '{}' not found".format(token_id), http_code=HTTPStatus.NOT_FOUND)
210
211 @staticmethod
212 def _remove_envelop(item, indata=None):
213 """
214 Obtain the useful data removing the envelop. It goes throw the vnfd or nsd catalog and returns the
215 vnfd or nsd content
216 :param item: can be vnfds, nsds, users, projects, userDefinedData (initial content of a vnfds, nsds
217 :param indata: Content to be inspected
218 :return: the useful part of indata
219 """
220 clean_indata = indata
221 if not indata:
222 return {}
223 if item == "vnfds":
224 if clean_indata.get('vnfd:vnfd-catalog'):
225 clean_indata = clean_indata['vnfd:vnfd-catalog']
226 elif clean_indata.get('vnfd-catalog'):
227 clean_indata = clean_indata['vnfd-catalog']
228 if clean_indata.get('vnfd'):
229 if not isinstance(clean_indata['vnfd'], list) or len(clean_indata['vnfd']) != 1:
230 raise EngineException("'vnfd' must be a list only one element")
231 clean_indata = clean_indata['vnfd'][0]
232 elif item == "nsds":
233 if clean_indata.get('nsd:nsd-catalog'):
234 clean_indata = clean_indata['nsd:nsd-catalog']
235 elif clean_indata.get('nsd-catalog'):
236 clean_indata = clean_indata['nsd-catalog']
237 if clean_indata.get('nsd'):
238 if not isinstance(clean_indata['nsd'], list) or len(clean_indata['nsd']) != 1:
239 raise EngineException("'nsd' must be a list only one element")
240 clean_indata = clean_indata['nsd'][0]
241 elif item == "userDefinedData":
242 if "userDefinedData" in indata:
243 clean_indata = clean_indata['userDefinedData']
244 return clean_indata
245
246 def _check_dependencies_on_descriptor(self, session, item, descriptor_id):
247 """
248 Check that the descriptor to be deleded is not a dependency of others
249 :param session: client session information
250 :param item: can be vnfds, nsds
251 :param descriptor_id: id of descriptor to be deleted
252 :return: None or raises exception
253 """
254 if item == "vnfds":
255 _filter = {"constituent-vnfd.ANYINDEX.vnfd-id-ref": descriptor_id}
256 if self.get_item_list(session, "nsds", _filter):
257 raise EngineException("There are nsd that depends on this VNFD", http_code=HTTPStatus.CONFLICT)
258 elif item == "nsds":
259 _filter = {"nsdId": descriptor_id}
260 if self.get_item_list(session, "nsrs", _filter):
261 raise EngineException("There are nsr that depends on this NSD", http_code=HTTPStatus.CONFLICT)
262
263 def _check_descriptor_dependencies(self, session, item, descriptor):
264 """
265 Check that the dependent descriptors exist on a new descriptor or edition
266 :param session: client session information
267 :param item: can be nsds, nsrs
268 :param descriptor: descriptor to be inserted or edit
269 :return: None or raises exception
270 """
271 if item == "nsds":
272 if not descriptor.get("constituent-vnfd"):
273 return
274 for vnf in descriptor["constituent-vnfd"]:
275 vnfd_id = vnf["vnfd-id-ref"]
276 if not self.get_item_list(session, "vnfds", {"id": vnfd_id}):
277 raise EngineException("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non "
278 "existing vnfd".format(vnfd_id), http_code=HTTPStatus.CONFLICT)
279 elif item == "nsrs":
280 if not descriptor.get("nsdId"):
281 return
282 nsd_id = descriptor["nsdId"]
283 if not self.get_item_list(session, "nsds", {"id": nsd_id}):
284 raise EngineException("Descriptor error at nsdId='{}' references a non exist nsd".format(nsd_id),
285 http_code=HTTPStatus.CONFLICT)
286
287 def _validate_new_data(self, session, item, indata, id=None, force=False):
288 if item == "users":
289 if not indata.get("username"):
290 raise EngineException("missing 'username'", HTTPStatus.UNPROCESSABLE_ENTITY)
291 if not indata.get("password"):
292 raise EngineException("missing 'password'", HTTPStatus.UNPROCESSABLE_ENTITY)
293 if not indata.get("projects"):
294 raise EngineException("missing 'projects'", HTTPStatus.UNPROCESSABLE_ENTITY)
295 # check username not exists
296 if self.db.get_one(item, {"username": indata.get("username")}, fail_on_empty=False, fail_on_more=False):
297 raise EngineException("username '{}' exists".format(indata["username"]), HTTPStatus.CONFLICT)
298 elif item == "projects":
299 if not indata.get("name"):
300 raise EngineException("missing 'name'")
301 # check name not exists
302 if self.db.get_one(item, {"name": indata.get("name")}, fail_on_empty=False, fail_on_more=False):
303 raise EngineException("name '{}' exists".format(indata["name"]), HTTPStatus.CONFLICT)
304 elif item in ("vnfds", "nsds"):
305 filter = {"id": indata["id"]}
306 if id:
307 filter["_id.neq"] = id
308 # TODO add admin to filter, validate rights
309 self._add_read_filter(session, item, filter)
310 if self.db.get_one(item, filter, fail_on_empty=False):
311 raise EngineException("{} with id '{}' already exists for this tenant".format(item[:-1], indata["id"]),
312 HTTPStatus.CONFLICT)
313 # TODO validate with pyangbind. Load and dumps to convert data types
314 if item == "nsds":
315 # transform constituent-vnfd:member-vnf-index to string
316 if indata.get("constituent-vnfd"):
317 for constituent_vnfd in indata["constituent-vnfd"]:
318 if "member-vnf-index" in constituent_vnfd:
319 constituent_vnfd["member-vnf-index"] = str(constituent_vnfd["member-vnf-index"])
320
321 if item == "nsds" and not force:
322 self._check_descriptor_dependencies(session, "nsds", indata)
323 elif item == "userDefinedData":
324 # TODO validate userDefinedData is a keypair values
325 pass
326
327 elif item == "nsrs":
328 pass
329 elif item == "vim_accounts" or item == "sdns":
330 filter = {"name": indata.get("name")}
331 if id:
332 filter["_id.neq"] = id
333 if self.db.get_one(item, filter, fail_on_empty=False, fail_on_more=False):
334 raise EngineException("name '{}' already exists for {}".format(indata["name"], item),
335 HTTPStatus.CONFLICT)
336
337 def _check_ns_operation(self, session, nsr, operation, indata):
338 """
339 Check that user has enter right parameters for the operation
340 :param session:
341 :param operation: it can be: instantiate, terminate, action, TODO: update, heal
342 :param indata: descriptor with the parameters of the operation
343 :return: None
344 """
345 if operation == "action":
346 if indata.get("vnf_member_index"):
347 indata["member_vnf_index"] = indata.pop("vnf_member_index") # for backward compatibility
348 for vnf in nsr["nsd"]["constituent-vnfd"]:
349 if indata["member_vnf_index"] == vnf["member-vnf-index"]:
350 # TODO get vnfd, check primitives
351 break
352 else:
353 raise EngineException("Invalid parameter member_vnf_index='{}' is not one of the nsd "
354 "constituent-vnfd".format(indata["member_vnf_index"]))
355
356 def _format_new_data(self, session, item, indata):
357 now = time()
358 if "_admin" not in indata:
359 indata["_admin"] = {}
360 indata["_admin"]["created"] = now
361 indata["_admin"]["modified"] = now
362 if item == "users":
363 indata["_id"] = indata["username"]
364 salt = uuid4().hex
365 indata["_admin"]["salt"] = salt
366 indata["password"] = sha256(indata["password"].encode('utf-8') + salt.encode('utf-8')).hexdigest()
367 elif item == "projects":
368 indata["_id"] = indata["name"]
369 else:
370 if not indata.get("_id"):
371 indata["_id"] = str(uuid4())
372 if item in ("vnfds", "nsds", "nsrs", "vnfrs"):
373 if not indata["_admin"].get("projects_read"):
374 indata["_admin"]["projects_read"] = [session["project_id"]]
375 if not indata["_admin"].get("projects_write"):
376 indata["_admin"]["projects_write"] = [session["project_id"]]
377 if item in ("vnfds", "nsds"):
378 indata["_admin"]["onboardingState"] = "CREATED"
379 indata["_admin"]["operationalState"] = "DISABLED"
380 indata["_admin"]["usageSate"] = "NOT_IN_USE"
381 if item == "nsrs":
382 indata["_admin"]["nsState"] = "NOT_INSTANTIATED"
383 if item in ("vim_accounts", "sdns"):
384 indata["_admin"]["operationalState"] = "PROCESSING"
385
386 def upload_content(self, session, item, _id, indata, kwargs, headers):
387 """
388 Used for receiving content by chunks (with a transaction_id header and/or gzip file. It will store and extract)
389 :param session: session
390 :param item: can be nsds or vnfds
391 :param _id : the nsd,vnfd is already created, this is the id
392 :param indata: http body request
393 :param kwargs: user query string to override parameters. NOT USED
394 :param headers: http request headers
395 :return: True package has is completely uploaded or False if partial content has been uplodaed.
396 Raise exception on error
397 """
398 # Check that _id exists and it is valid
399 current_desc = self.get_item(session, item, _id)
400
401 content_range_text = headers.get("Content-Range")
402 expected_md5 = headers.get("Content-File-MD5")
403 compressed = None
404 content_type = headers.get("Content-Type")
405 if content_type and "application/gzip" in content_type or "application/x-gzip" in content_type or \
406 "application/zip" in content_type:
407 compressed = "gzip"
408 filename = headers.get("Content-Filename")
409 if not filename:
410 filename = "package.tar.gz" if compressed else "package"
411 # TODO change to Content-Disposition filename https://tools.ietf.org/html/rfc6266
412 file_pkg = None
413 error_text = ""
414 try:
415 if content_range_text:
416 content_range = content_range_text.replace("-", " ").replace("/", " ").split()
417 if content_range[0] != "bytes": # TODO check x<y not negative < total....
418 raise IndexError()
419 start = int(content_range[1])
420 end = int(content_range[2]) + 1
421 total = int(content_range[3])
422 else:
423 start = 0
424
425 if start:
426 if not self.fs.file_exists(_id, 'dir'):
427 raise EngineException("invalid Transaction-Id header", HTTPStatus.NOT_FOUND)
428 else:
429 self.fs.file_delete(_id, ignore_non_exist=True)
430 self.fs.mkdir(_id)
431
432 storage = self.fs.get_params()
433 storage["folder"] = _id
434
435 file_path = (_id, filename)
436 if self.fs.file_exists(file_path, 'file'):
437 file_size = self.fs.file_size(file_path)
438 else:
439 file_size = 0
440 if file_size != start:
441 raise EngineException("invalid Content-Range start sequence, expected '{}' but received '{}'".format(
442 file_size, start), HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
443 file_pkg = self.fs.file_open(file_path, 'a+b')
444 if isinstance(indata, dict):
445 indata_text = yaml.safe_dump(indata, indent=4, default_flow_style=False)
446 file_pkg.write(indata_text.encode(encoding="utf-8"))
447 else:
448 indata_len = 0
449 while True:
450 indata_text = indata.read(4096)
451 indata_len += len(indata_text)
452 if not indata_text:
453 break
454 file_pkg.write(indata_text)
455 if content_range_text:
456 if indata_len != end-start:
457 raise EngineException("Mismatch between Content-Range header {}-{} and body length of {}".format(
458 start, end-1, indata_len), HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
459 if end != total:
460 # TODO update to UPLOADING
461 return False
462
463 # PACKAGE UPLOADED
464 if expected_md5:
465 file_pkg.seek(0, 0)
466 file_md5 = md5()
467 chunk_data = file_pkg.read(1024)
468 while chunk_data:
469 file_md5.update(chunk_data)
470 chunk_data = file_pkg.read(1024)
471 if expected_md5 != file_md5.hexdigest():
472 raise EngineException("Error, MD5 mismatch", HTTPStatus.CONFLICT)
473 file_pkg.seek(0, 0)
474 if compressed == "gzip":
475 tar = tarfile.open(mode='r', fileobj=file_pkg)
476 descriptor_file_name = None
477 for tarinfo in tar:
478 tarname = tarinfo.name
479 tarname_path = tarname.split("/")
480 if not tarname_path[0] or ".." in tarname_path: # if start with "/" means absolute path
481 raise EngineException("Absolute path or '..' are not allowed for package descriptor tar.gz")
482 if len(tarname_path) == 1 and not tarinfo.isdir():
483 raise EngineException("All files must be inside a dir for package descriptor tar.gz")
484 if tarname.endswith(".yaml") or tarname.endswith(".json") or tarname.endswith(".yml"):
485 storage["pkg-dir"] = tarname_path[0]
486 if len(tarname_path) == 2:
487 if descriptor_file_name:
488 raise EngineException(
489 "Found more than one descriptor file at package descriptor tar.gz")
490 descriptor_file_name = tarname
491 if not descriptor_file_name:
492 raise EngineException("Not found any descriptor file at package descriptor tar.gz")
493 storage["descriptor"] = descriptor_file_name
494 storage["zipfile"] = filename
495 self.fs.file_extract(tar, _id)
496 with self.fs.file_open((_id, descriptor_file_name), "r") as descriptor_file:
497 content = descriptor_file.read()
498 else:
499 content = file_pkg.read()
500 storage["descriptor"] = descriptor_file_name = filename
501
502 if descriptor_file_name.endswith(".json"):
503 error_text = "Invalid json format "
504 indata = json.load(content)
505 else:
506 error_text = "Invalid yaml format "
507 indata = yaml.load(content)
508
509 current_desc["_admin"]["storage"] = storage
510 current_desc["_admin"]["onboardingState"] = "ONBOARDED"
511 current_desc["_admin"]["operationalState"] = "ENABLED"
512
513 self._edit_item(session, item, _id, current_desc, indata, kwargs)
514 # TODO if descriptor has changed because kwargs update content and remove cached zip
515 # TODO if zip is not present creates one
516 return True
517
518 except EngineException:
519 raise
520 except IndexError:
521 raise EngineException("invalid Content-Range header format. Expected 'bytes start-end/total'",
522 HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
523 except IOError as e:
524 raise EngineException("invalid upload transaction sequence: '{}'".format(e), HTTPStatus.BAD_REQUEST)
525 except tarfile.ReadError as e:
526 raise EngineException("invalid file content {}".format(e), HTTPStatus.BAD_REQUEST)
527 except (ValueError, yaml.YAMLError) as e:
528 raise EngineException(error_text + str(e))
529 finally:
530 if file_pkg:
531 file_pkg.close()
532
533 def new_nsr(self, rollback, session, ns_request):
534 """
535 Creates a new nsr into database. It also creates needed vnfrs
536 :param rollback: list where this method appends created items at database in case a rollback may to be done
537 :param session: contains the used login username and working project
538 :param ns_request: params to be used for the nsr
539 :return: the _id of nsr descriptor stored at database
540 """
541 rollback_index = len(rollback)
542 step = ""
543 try:
544 # look for nsr
545 step = "getting nsd id='{}' from database".format(ns_request.get("nsdId"))
546 nsd = self.get_item(session, "nsds", ns_request["nsdId"])
547 nsr_id = str(uuid4())
548 now = time()
549 step = "filling nsr from input data"
550 nsr_descriptor = {
551 "name": ns_request["nsName"],
552 "name-ref": ns_request["nsName"],
553 "short-name": ns_request["nsName"],
554 "admin-status": "ENABLED",
555 "nsd": nsd,
556 "datacenter": ns_request["vimAccountId"],
557 "resource-orchestrator": "osmopenmano",
558 "description": ns_request.get("nsDescription", ""),
559 "constituent-vnfr-ref": [],
560
561 "operational-status": "init", # typedef ns-operational-
562 "config-status": "init", # typedef config-states
563 "detailed-status": "scheduled",
564
565 "orchestration-progress": {},
566 # {"networks": {"active": 0, "total": 0}, "vms": {"active": 0, "total": 0}},
567
568 "crete-time": now,
569 "nsd-name-ref": nsd["name"],
570 "operational-events": [], # "id", "timestamp", "description", "event",
571 "nsd-ref": nsd["id"],
572 "instantiate_params": ns_request,
573 "ns-instance-config-ref": nsr_id,
574 "id": nsr_id,
575 "_id": nsr_id,
576 # "input-parameter": xpath, value,
577 "ssh-authorized-key": ns_request.get("key-pair-ref"),
578 }
579 ns_request["nsr_id"] = nsr_id
580
581 # Create VNFR
582 needed_vnfds = {}
583 for member_vnf in nsd["constituent-vnfd"]:
584 vnfd_id = member_vnf["vnfd-id-ref"]
585 step = "getting vnfd id='{}' constituent-vnfd='{}' from database".format(
586 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
587 if vnfd_id not in needed_vnfds:
588 # Obtain vnfd
589 vnf_filter = {"id": vnfd_id}
590 self._add_read_filter(session, "vnfds", vnf_filter)
591 vnfd = self.db.get_one("vnfds", vnf_filter)
592 vnfd.pop("_admin")
593 needed_vnfds[vnfd_id] = vnfd
594 else:
595 vnfd = needed_vnfds[vnfd_id]
596 step = "filling vnfr vnfd-id='{}' constituent-vnfd='{}'".format(
597 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
598 vnfr_id = str(uuid4())
599 vnfr_descriptor = {
600 "id": vnfr_id,
601 "_id": vnfr_id,
602 "nsr-id-ref": nsr_id,
603 "member-vnf-index-ref": member_vnf["member-vnf-index"],
604 "created-time": now,
605 # "vnfd": vnfd, # at OSM model.but removed to avoid data duplication TODO: revise
606 "vnfd-ref": vnfd_id,
607 "vnfd-id": vnfd["_id"], # not at OSM model, but useful
608 "vim-account-id": None,
609 "vdur": [],
610 "connection-point": [],
611 "ip-address": None, # mgmt-interface filled by LCM
612 }
613 for cp in vnfd.get("connection-point", ()):
614 vnf_cp = {
615 "name": cp["name"],
616 "connection-point-id": cp.get("id"),
617 "id": cp.get("id"),
618 # "ip-address", "mac-address" # filled by LCM
619 # vim-id # TODO it would be nice having a vim port id
620 }
621 vnfr_descriptor["connection-point"].append(vnf_cp)
622 for vdu in vnfd["vdu"]:
623 vdur_id = str(uuid4())
624 vdur = {
625 "id": vdur_id,
626 "vdu-id-ref": vdu["id"],
627 "ip-address": None, # mgmt-interface filled by LCM
628 # "vim-id", "flavor-id", "image-id", "management-ip" # filled by LCM
629 "internal-connection-point": [],
630 }
631 # TODO volumes: name, volume-id
632 for icp in vdu.get("internal-connection-point", ()):
633 vdu_icp = {
634 "id": icp["id"],
635 "connection-point-id": icp["id"],
636 "name": icp.get("name"),
637 # "ip-address", "mac-address" # filled by LCM
638 # vim-id # TODO it would be nice having a vim port id
639 }
640 vdur["internal-connection-point"].append(vdu_icp)
641 vnfr_descriptor["vdur"].append(vdur)
642
643 step = "creating vnfr vnfd-id='{}' constituent-vnfd='{}' at database".format(
644 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
645 self._format_new_data(session, "vnfrs", vnfr_descriptor)
646 self.db.create("vnfrs", vnfr_descriptor)
647 rollback.insert(0, {"item": "vnfrs", "_id": vnfr_id})
648 nsr_descriptor["constituent-vnfr-ref"].append(vnfr_id)
649
650 step = "creating nsr at database"
651 self._format_new_data(session, "nsrs", nsr_descriptor)
652 self.db.create("nsrs", nsr_descriptor)
653 rollback.insert(rollback_index, {"item": "nsrs", "_id": nsr_id})
654 return nsr_id
655 except Exception as e:
656 raise EngineException("Error {}: {}".format(step, e))
657
658 @staticmethod
659 def _update_descriptor(desc, kwargs):
660 """
661 Update descriptor with the kwargs. It contains dot separated keys
662 :param desc: dictionary to be updated
663 :param kwargs: plain dictionary to be used for updating.
664 :return:
665 """
666 if not kwargs:
667 return
668 try:
669 for k, v in kwargs.items():
670 update_content = desc
671 kitem_old = None
672 klist = k.split(".")
673 for kitem in klist:
674 if kitem_old is not None:
675 update_content = update_content[kitem_old]
676 if isinstance(update_content, dict):
677 kitem_old = kitem
678 elif isinstance(update_content, list):
679 kitem_old = int(kitem)
680 else:
681 raise EngineException(
682 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem))
683 update_content[kitem_old] = v
684 except KeyError:
685 raise EngineException(
686 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old))
687 except ValueError:
688 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
689 k, kitem))
690 except IndexError:
691 raise EngineException(
692 "Invalid query string '{}'. Index '{}' out of range".format(k, kitem_old))
693
694 def new_item(self, rollback, session, item, indata={}, kwargs=None, headers={}, force=False):
695 """
696 Creates a new entry into database. For nsds and vnfds it creates an almost empty DISABLED entry,
697 that must be completed with a call to method upload_content
698 :param rollback: list where this method appends created items at database in case a rollback may to be done
699 :param session: contains the used login username and working project
700 :param item: it can be: users, projects, vim_accounts, sdns, nsrs, nsds, vnfds
701 :param indata: data to be inserted
702 :param kwargs: used to override the indata descriptor
703 :param headers: http request headers
704 :param force: If True avoid some dependence checks
705 :return: _id: identity of the inserted data.
706 """
707
708 try:
709 item_envelop = item
710 if item in ("nsds", "vnfds"):
711 item_envelop = "userDefinedData"
712 content = self._remove_envelop(item_envelop, indata)
713
714 # Override descriptor with query string kwargs
715 self._update_descriptor(content, kwargs)
716 if not indata and item not in ("nsds", "vnfds"):
717 raise EngineException("Empty payload")
718
719 validate_input(content, item, new=True)
720
721 if item == "nsrs":
722 # in this case the input descriptor is not the data to be stored
723 return self.new_nsr(rollback, session, ns_request=content)
724
725 self._validate_new_data(session, item_envelop, content, force)
726 if item in ("nsds", "vnfds"):
727 content = {"_admin": {"userDefinedData": content}}
728 self._format_new_data(session, item, content)
729 _id = self.db.create(item, content)
730 rollback.insert(0, {"item": item, "_id": _id})
731
732 if item == "vim_accounts":
733 msg_data = self.db.get_one(item, {"_id": _id})
734 msg_data.pop("_admin", None)
735 self.msg.write("vim_account", "create", msg_data)
736 elif item == "sdns":
737 msg_data = self.db.get_one(item, {"_id": _id})
738 msg_data.pop("_admin", None)
739 self.msg.write("sdn", "create", msg_data)
740 return _id
741 except ValidationError as e:
742 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
743
744 def new_nslcmop(self, session, nsInstanceId, operation, params):
745 now = time()
746 _id = str(uuid4())
747 nslcmop = {
748 "id": _id,
749 "_id": _id,
750 "operationState": "PROCESSING", # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
751 "statusEnteredTime": now,
752 "nsInstanceId": nsInstanceId,
753 "lcmOperationType": operation,
754 "startTime": now,
755 "isAutomaticInvocation": False,
756 "operationParams": params,
757 "isCancelPending": False,
758 "links": {
759 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id,
760 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsInstanceId,
761 }
762 }
763 return nslcmop
764
765 def ns_operation(self, rollback, session, nsInstanceId, operation, indata, kwargs=None):
766 """
767 Performs a new operation over a ns
768 :param rollback: list where this method appends created items at database in case a rollback may to be done
769 :param session: contains the used login username and working project
770 :param nsInstanceId: _id of the nsr to perform the operation
771 :param operation: it can be: instantiate, terminate, action, TODO: update, heal
772 :param indata: descriptor with the parameters of the operation
773 :param kwargs: used to override the indata descriptor
774 :return: id of the nslcmops
775 """
776 try:
777 # Override descriptor with query string kwargs
778 self._update_descriptor(indata, kwargs)
779 validate_input(indata, "ns_" + operation, new=True)
780 # get ns from nsr_id
781 nsr = self.get_item(session, "nsrs", nsInstanceId)
782 if not nsr["_admin"].get("nsState") or nsr["_admin"]["nsState"] == "NOT_INSTANTIATED":
783 if operation == "terminate" and indata.get("autoremove"):
784 # NSR must be deleted
785 return self.del_item(session, "nsrs", nsInstanceId)
786 if operation != "instantiate":
787 raise EngineException("ns_instance '{}' cannot be '{}' because it is not instantiated".format(
788 nsInstanceId, operation), HTTPStatus.CONFLICT)
789 else:
790 if operation == "instantiate" and not indata.get("force"):
791 raise EngineException("ns_instance '{}' cannot be '{}' because it is already instantiated".format(
792 nsInstanceId, operation), HTTPStatus.CONFLICT)
793 indata["nsInstanceId"] = nsInstanceId
794 self._check_ns_operation(session, nsr, operation, indata)
795 nslcmop = self.new_nslcmop(session, nsInstanceId, operation, indata)
796 self._format_new_data(session, "nslcmops", nslcmop)
797 _id = self.db.create("nslcmops", nslcmop)
798 rollback.insert(0, {"item": "nslcmops", "_id": _id})
799 indata["_id"] = _id
800 self.msg.write("ns", operation, nslcmop)
801 return _id
802 except ValidationError as e:
803 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
804 # except DbException as e:
805 # raise EngineException("Cannot get ns_instance '{}': {}".format(e), HTTPStatus.NOT_FOUND)
806
807 def _add_read_filter(self, session, item, filter):
808 if session["project_id"] == "admin": # allows all
809 return filter
810 if item == "users":
811 filter["username"] = session["username"]
812 elif item in ("vnfds", "nsds", "nsrs"):
813 filter["_admin.projects_read.cont"] = ["ANY", session["project_id"]]
814
815 def _add_delete_filter(self, session, item, filter):
816 if session["project_id"] != "admin" and item in ("users", "projects"):
817 raise EngineException("Only admin users can perform this task", http_code=HTTPStatus.FORBIDDEN)
818 if item == "users":
819 if filter.get("_id") == session["username"] or filter.get("username") == session["username"]:
820 raise EngineException("You cannot delete your own user", http_code=HTTPStatus.CONFLICT)
821 elif item == "project":
822 if filter.get("_id") == session["project_id"]:
823 raise EngineException("You cannot delete your own project", http_code=HTTPStatus.CONFLICT)
824 elif item in ("vnfds", "nsds") and session["project_id"] != "admin":
825 filter["_admin.projects_write.cont"] = ["ANY", session["project_id"]]
826
827 def get_file(self, session, item, _id, path=None, accept_header=None):
828 """
829 Return the file content of a vnfd or nsd
830 :param session: contains the used login username and working project
831 :param item: it can be vnfds or nsds
832 :param _id: Identity of the vnfd, ndsd
833 :param path: artifact path or "$DESCRIPTOR" or None
834 :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
835 :return: opened file or raises an exception
836 """
837 accept_text = accept_zip = False
838 if accept_header:
839 if 'text/plain' in accept_header or '*/*' in accept_header:
840 accept_text = True
841 if 'application/zip' in accept_header or '*/*' in accept_header:
842 accept_zip = True
843 if not accept_text and not accept_zip:
844 raise EngineException("provide request header 'Accept' with 'application/zip' or 'text/plain'",
845 http_code=HTTPStatus.NOT_ACCEPTABLE)
846
847 content = self.get_item(session, item, _id)
848 if content["_admin"]["onboardingState"] != "ONBOARDED":
849 raise EngineException("Cannot get content because this resource is not at 'ONBOARDED' state. "
850 "onboardingState is {}".format(content["_admin"]["onboardingState"]),
851 http_code=HTTPStatus.CONFLICT)
852 storage = content["_admin"]["storage"]
853 if path is not None and path != "$DESCRIPTOR": # artifacts
854 if not storage.get('pkg-dir'):
855 raise EngineException("Packages does not contains artifacts", http_code=HTTPStatus.BAD_REQUEST)
856 if self.fs.file_exists((storage['folder'], storage['pkg-dir'], *path), 'dir'):
857 folder_content = self.fs.dir_ls((storage['folder'], storage['pkg-dir'], *path))
858 return folder_content, "text/plain"
859 # TODO manage folders in http
860 else:
861 return self.fs.file_open((storage['folder'], storage['pkg-dir'], *path), "rb"),\
862 "application/octet-stream"
863
864 # pkgtype accept ZIP TEXT -> result
865 # manyfiles yes X -> zip
866 # no yes -> error
867 # onefile yes no -> zip
868 # X yes -> text
869
870 if accept_text and (not storage.get('pkg-dir') or path == "$DESCRIPTOR"):
871 return self.fs.file_open((storage['folder'], storage['descriptor']), "r"), "text/plain"
872 elif storage.get('pkg-dir') and not accept_zip:
873 raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'"
874 "Accept header", http_code=HTTPStatus.NOT_ACCEPTABLE)
875 else:
876 if not storage.get('zipfile'):
877 # TODO generate zipfile if not present
878 raise EngineException("Only allowed 'text/plain' Accept header for this descriptor. To be solved in "
879 "future versions", http_code=HTTPStatus.NOT_ACCEPTABLE)
880 return self.fs.file_open((storage['folder'], storage['zipfile']), "rb"), "application/zip"
881
882 def get_item_list(self, session, item, filter={}):
883 """
884 Get a list of items
885 :param session: contains the used login username and working project
886 :param item: it can be: users, projects, vnfds, nsds, ...
887 :param filter: filter of data to be applied
888 :return: The list, it can be empty if no one match the filter.
889 """
890 # TODO add admin to filter, validate rights
891 # TODO transform data for SOL005 URL requests. Transform filtering
892 # TODO implement "field-type" query string SOL005
893
894 self._add_read_filter(session, item, filter)
895 return self.db.get_list(item, filter)
896
897 def get_item(self, session, item, _id):
898 """
899 Get complete information on an items
900 :param session: contains the used login username and working project
901 :param item: it can be: users, projects, vnfds, nsds,
902 :param _id: server id of the item
903 :return: dictionary, raise exception if not found.
904 """
905 filter = {"_id": _id}
906 # TODO add admin to filter, validate rights
907 # TODO transform data for SOL005 URL requests
908 self._add_read_filter(session, item, filter)
909 return self.db.get_one(item, filter)
910
911 def del_item_list(self, session, item, filter={}):
912 """
913 Delete a list of items
914 :param session: contains the used login username and working project
915 :param item: it can be: users, projects, vnfds, nsds, ...
916 :param filter: filter of data to be applied
917 :return: The deleted list, it can be empty if no one match the filter.
918 """
919 # TODO add admin to filter, validate rights
920 self._add_read_filter(session, item, filter)
921 return self.db.del_list(item, filter)
922
923 def del_item(self, session, item, _id, force=False):
924 """
925 Delete item by its internal id
926 :param session: contains the used login username and working project
927 :param item: it can be: users, projects, vnfds, nsds, ...
928 :param _id: server id of the item
929 :param force: indicates if deletion must be forced in case of conflict
930 :return: dictionary with deleted item _id. It raises exception if not found.
931 """
932 # TODO add admin to filter, validate rights
933 # data = self.get_item(item, _id)
934 filter = {"_id": _id}
935 self._add_delete_filter(session, item, filter)
936 if item in ("vnfds", "nsds") and not force:
937 descriptor = self.get_item(session, item, _id)
938 descriptor_id = descriptor.get("id")
939 if descriptor_id:
940 self._check_dependencies_on_descriptor(session, item, descriptor_id)
941
942 if item == "nsrs":
943 nsr = self.db.get_one(item, filter)
944 if nsr["_admin"].get("nsState") == "INSTANTIATED" and not force:
945 raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. "
946 "Launch 'terminate' operation first; or force deletion".format(_id),
947 http_code=HTTPStatus.CONFLICT)
948 v = self.db.del_one(item, {"_id": _id})
949 self.db.del_list("nslcmops", {"nsInstanceId": _id})
950 self.db.del_list("vnfrs", {"nsr-id-ref": _id})
951 self.msg.write("ns", "deleted", {"_id": _id})
952 return v
953 if item in ("vim_accounts", "sdns") and not force:
954 self.db.set_one(item, {"_id": _id}, {"_admin.to_delete": True}) # TODO change status
955 if item == "vim_accounts":
956 self.msg.write("vim_account", "delete", {"_id": _id})
957 elif item == "sdns":
958 self.msg.write("sdn", "delete", {"_id": _id})
959 return {"deleted": 1} # TODO indicate an offline operation to return 202 ACCEPTED
960
961 v = self.db.del_one(item, filter)
962 if item in ("vnfds", "nsds"):
963 self.fs.file_delete(_id, ignore_non_exist=True)
964 if item in ("vim_accounts", "sdns", "vnfds", "nsds"):
965 self.msg.write(item[:-1], "deleted", {"_id": _id})
966 return v
967
968 def prune(self):
969 """
970 Prune database not needed content
971 :return: None
972 """
973 return self.db.del_list("nsrs", {"_admin.to_delete": True})
974
975 def create_admin(self):
976 """
977 Creates a new user admin/admin into database if database is empty. Useful for initialization
978 :return: _id identity of the inserted data, or None
979 """
980 users = self.db.get_one("users", fail_on_empty=False, fail_on_more=False)
981 if users:
982 return None
983 # raise EngineException("Unauthorized. Database users is not empty", HTTPStatus.UNAUTHORIZED)
984 indata = {"username": "admin", "password": "admin", "projects": ["admin"]}
985 fake_session = {"project_id": "admin", "username": "admin"}
986 self._format_new_data(fake_session, "users", indata)
987 _id = self.db.create("users", indata)
988 return _id
989
990 def init_db(self, target_version='1.0'):
991 """
992 Init database if empty. If not empty it checks that database version is ok.
993 If empty, it creates a new user admin/admin at 'users' and a new entry at 'version'
994 :return: None if ok, exception if error or if the version is different.
995 """
996 version = self.db.get_one("version", fail_on_empty=False, fail_on_more=False)
997 if not version:
998 # create user admin
999 self.create_admin()
1000 # create database version
1001 version_data = {
1002 "_id": '1.0', # version text
1003 "version": 1000, # version number
1004 "date": "2018-04-12", # version date
1005 "description": "initial design", # changes in this version
1006 'status': 'ENABLED' # ENABLED, DISABLED (migration in process), ERROR,
1007 }
1008 self.db.create("version", version_data)
1009 elif version["_id"] != target_version:
1010 # TODO implement migration process
1011 raise EngineException("Wrong database version '{}'. Expected '{}'".format(
1012 version["_id"], target_version), HTTPStatus.INTERNAL_SERVER_ERROR)
1013 elif version["status"] != 'ENABLED':
1014 raise EngineException("Wrong database status '{}'".format(
1015 version["status"]), HTTPStatus.INTERNAL_SERVER_ERROR)
1016 return
1017
1018 def _edit_item(self, session, item, id, content, indata={}, kwargs=None, force=False):
1019 if indata:
1020 indata = self._remove_envelop(item, indata)
1021
1022 # Override descriptor with query string kwargs
1023 if kwargs:
1024 try:
1025 for k, v in kwargs.items():
1026 update_content = indata
1027 kitem_old = None
1028 klist = k.split(".")
1029 for kitem in klist:
1030 if kitem_old is not None:
1031 update_content = update_content[kitem_old]
1032 if isinstance(update_content, dict):
1033 kitem_old = kitem
1034 elif isinstance(update_content, list):
1035 kitem_old = int(kitem)
1036 else:
1037 raise EngineException(
1038 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem))
1039 update_content[kitem_old] = v
1040 except KeyError:
1041 raise EngineException(
1042 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old))
1043 except ValueError:
1044 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
1045 k, kitem))
1046 except IndexError:
1047 raise EngineException(
1048 "Invalid query string '{}'. Index '{}' out of range".format(k, kitem_old))
1049 try:
1050 validate_input(indata, item, new=False)
1051 except ValidationError as e:
1052 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
1053
1054 _deep_update(content, indata)
1055 self._validate_new_data(session, item, content, id, force)
1056 # self._format_new_data(session, item, content)
1057 self.db.replace(item, id, content)
1058 if item in ("vim_accounts", "sdns"):
1059 indata.pop("_admin", None)
1060 indata["_id"] = id
1061 if item == "vim_accounts":
1062 self.msg.write("vim_account", "edit", indata)
1063 elif item == "sdns":
1064 self.msg.write("sdn", "edit", indata)
1065 return id
1066
1067 def edit_item(self, session, item, _id, indata={}, kwargs=None, force=False):
1068 """
1069 Update an existing entry at database
1070 :param session: contains the used login username and working project
1071 :param item: it can be: users, projects, vnfds, nsds, ...
1072 :param _id: identifier to be updated
1073 :param indata: data to be inserted
1074 :param kwargs: used to override the indata descriptor
1075 :param force: If True avoid some dependence checks
1076 :return: dictionary, raise exception if not found.
1077 """
1078
1079 content = self.get_item(session, item, _id)
1080 return self._edit_item(session, item, _id, content, indata, kwargs, force)