aaa94f0ed76d671693eb9b91d482fe3c9d5ee0f4
[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 def check_valid_vnf_member_index(member_vnf_index):
346 for vnf in nsr["nsd"]["constituent-vnfd"]:
347 if member_vnf_index == vnf["member-vnf-index"]:
348 break
349 else:
350 raise EngineException("Invalid parameter member_vnf_index='{}' is not one of the "
351 "nsd:constituent-vnfd".format(member_vnf_index))
352
353 if operation == "action":
354 if indata.get("vnf_member_index"):
355 indata["member_vnf_index"] = indata.pop("vnf_member_index") # for backward compatibility
356 check_valid_vnf_member_index(indata["member_vnf_index"])
357 # TODO get vnfd, check primitives
358 if operation == "scale":
359 check_valid_vnf_member_index(indata["scaleVnfData"]["scaleByStepData"]["member-vnf-index"])
360 # TODO check vnf scaling primitives
361
362 def _format_new_data(self, session, item, indata):
363 now = time()
364 if "_admin" not in indata:
365 indata["_admin"] = {}
366 indata["_admin"]["created"] = now
367 indata["_admin"]["modified"] = now
368 if item == "users":
369 indata["_id"] = indata["username"]
370 salt = uuid4().hex
371 indata["_admin"]["salt"] = salt
372 indata["password"] = sha256(indata["password"].encode('utf-8') + salt.encode('utf-8')).hexdigest()
373 elif item == "projects":
374 indata["_id"] = indata["name"]
375 else:
376 if not indata.get("_id"):
377 indata["_id"] = str(uuid4())
378 if item in ("vnfds", "nsds", "nsrs", "vnfrs"):
379 if not indata["_admin"].get("projects_read"):
380 indata["_admin"]["projects_read"] = [session["project_id"]]
381 if not indata["_admin"].get("projects_write"):
382 indata["_admin"]["projects_write"] = [session["project_id"]]
383 if item in ("vnfds", "nsds"):
384 indata["_admin"]["onboardingState"] = "CREATED"
385 indata["_admin"]["operationalState"] = "DISABLED"
386 indata["_admin"]["usageSate"] = "NOT_IN_USE"
387 if item == "nsrs":
388 indata["_admin"]["nsState"] = "NOT_INSTANTIATED"
389 if item in ("vim_accounts", "sdns"):
390 indata["_admin"]["operationalState"] = "PROCESSING"
391
392 def upload_content(self, session, item, _id, indata, kwargs, headers):
393 """
394 Used for receiving content by chunks (with a transaction_id header and/or gzip file. It will store and extract)
395 :param session: session
396 :param item: can be nsds or vnfds
397 :param _id : the nsd,vnfd is already created, this is the id
398 :param indata: http body request
399 :param kwargs: user query string to override parameters. NOT USED
400 :param headers: http request headers
401 :return: True package has is completely uploaded or False if partial content has been uplodaed.
402 Raise exception on error
403 """
404 # Check that _id exists and it is valid
405 current_desc = self.get_item(session, item, _id)
406
407 content_range_text = headers.get("Content-Range")
408 expected_md5 = headers.get("Content-File-MD5")
409 compressed = None
410 content_type = headers.get("Content-Type")
411 if content_type and "application/gzip" in content_type or "application/x-gzip" in content_type or \
412 "application/zip" in content_type:
413 compressed = "gzip"
414 filename = headers.get("Content-Filename")
415 if not filename:
416 filename = "package.tar.gz" if compressed else "package"
417 # TODO change to Content-Disposition filename https://tools.ietf.org/html/rfc6266
418 file_pkg = None
419 error_text = ""
420 try:
421 if content_range_text:
422 content_range = content_range_text.replace("-", " ").replace("/", " ").split()
423 if content_range[0] != "bytes": # TODO check x<y not negative < total....
424 raise IndexError()
425 start = int(content_range[1])
426 end = int(content_range[2]) + 1
427 total = int(content_range[3])
428 else:
429 start = 0
430
431 if start:
432 if not self.fs.file_exists(_id, 'dir'):
433 raise EngineException("invalid Transaction-Id header", HTTPStatus.NOT_FOUND)
434 else:
435 self.fs.file_delete(_id, ignore_non_exist=True)
436 self.fs.mkdir(_id)
437
438 storage = self.fs.get_params()
439 storage["folder"] = _id
440
441 file_path = (_id, filename)
442 if self.fs.file_exists(file_path, 'file'):
443 file_size = self.fs.file_size(file_path)
444 else:
445 file_size = 0
446 if file_size != start:
447 raise EngineException("invalid Content-Range start sequence, expected '{}' but received '{}'".format(
448 file_size, start), HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
449 file_pkg = self.fs.file_open(file_path, 'a+b')
450 if isinstance(indata, dict):
451 indata_text = yaml.safe_dump(indata, indent=4, default_flow_style=False)
452 file_pkg.write(indata_text.encode(encoding="utf-8"))
453 else:
454 indata_len = 0
455 while True:
456 indata_text = indata.read(4096)
457 indata_len += len(indata_text)
458 if not indata_text:
459 break
460 file_pkg.write(indata_text)
461 if content_range_text:
462 if indata_len != end-start:
463 raise EngineException("Mismatch between Content-Range header {}-{} and body length of {}".format(
464 start, end-1, indata_len), HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
465 if end != total:
466 # TODO update to UPLOADING
467 return False
468
469 # PACKAGE UPLOADED
470 if expected_md5:
471 file_pkg.seek(0, 0)
472 file_md5 = md5()
473 chunk_data = file_pkg.read(1024)
474 while chunk_data:
475 file_md5.update(chunk_data)
476 chunk_data = file_pkg.read(1024)
477 if expected_md5 != file_md5.hexdigest():
478 raise EngineException("Error, MD5 mismatch", HTTPStatus.CONFLICT)
479 file_pkg.seek(0, 0)
480 if compressed == "gzip":
481 tar = tarfile.open(mode='r', fileobj=file_pkg)
482 descriptor_file_name = None
483 for tarinfo in tar:
484 tarname = tarinfo.name
485 tarname_path = tarname.split("/")
486 if not tarname_path[0] or ".." in tarname_path: # if start with "/" means absolute path
487 raise EngineException("Absolute path or '..' are not allowed for package descriptor tar.gz")
488 if len(tarname_path) == 1 and not tarinfo.isdir():
489 raise EngineException("All files must be inside a dir for package descriptor tar.gz")
490 if tarname.endswith(".yaml") or tarname.endswith(".json") or tarname.endswith(".yml"):
491 storage["pkg-dir"] = tarname_path[0]
492 if len(tarname_path) == 2:
493 if descriptor_file_name:
494 raise EngineException(
495 "Found more than one descriptor file at package descriptor tar.gz")
496 descriptor_file_name = tarname
497 if not descriptor_file_name:
498 raise EngineException("Not found any descriptor file at package descriptor tar.gz")
499 storage["descriptor"] = descriptor_file_name
500 storage["zipfile"] = filename
501 self.fs.file_extract(tar, _id)
502 with self.fs.file_open((_id, descriptor_file_name), "r") as descriptor_file:
503 content = descriptor_file.read()
504 else:
505 content = file_pkg.read()
506 storage["descriptor"] = descriptor_file_name = filename
507
508 if descriptor_file_name.endswith(".json"):
509 error_text = "Invalid json format "
510 indata = json.load(content)
511 else:
512 error_text = "Invalid yaml format "
513 indata = yaml.load(content)
514
515 current_desc["_admin"]["storage"] = storage
516 current_desc["_admin"]["onboardingState"] = "ONBOARDED"
517 current_desc["_admin"]["operationalState"] = "ENABLED"
518
519 self._edit_item(session, item, _id, current_desc, indata, kwargs)
520 # TODO if descriptor has changed because kwargs update content and remove cached zip
521 # TODO if zip is not present creates one
522 return True
523
524 except EngineException:
525 raise
526 except IndexError:
527 raise EngineException("invalid Content-Range header format. Expected 'bytes start-end/total'",
528 HTTPStatus.REQUESTED_RANGE_NOT_SATISFIABLE)
529 except IOError as e:
530 raise EngineException("invalid upload transaction sequence: '{}'".format(e), HTTPStatus.BAD_REQUEST)
531 except tarfile.ReadError as e:
532 raise EngineException("invalid file content {}".format(e), HTTPStatus.BAD_REQUEST)
533 except (ValueError, yaml.YAMLError) as e:
534 raise EngineException(error_text + str(e))
535 finally:
536 if file_pkg:
537 file_pkg.close()
538
539 def new_nsr(self, rollback, session, ns_request):
540 """
541 Creates a new nsr into database. It also creates needed vnfrs
542 :param rollback: list where this method appends created items at database in case a rollback may to be done
543 :param session: contains the used login username and working project
544 :param ns_request: params to be used for the nsr
545 :return: the _id of nsr descriptor stored at database
546 """
547 rollback_index = len(rollback)
548 step = ""
549 try:
550 # look for nsr
551 step = "getting nsd id='{}' from database".format(ns_request.get("nsdId"))
552 nsd = self.get_item(session, "nsds", ns_request["nsdId"])
553 nsr_id = str(uuid4())
554 now = time()
555 step = "filling nsr from input data"
556 nsr_descriptor = {
557 "name": ns_request["nsName"],
558 "name-ref": ns_request["nsName"],
559 "short-name": ns_request["nsName"],
560 "admin-status": "ENABLED",
561 "nsd": nsd,
562 "datacenter": ns_request["vimAccountId"],
563 "resource-orchestrator": "osmopenmano",
564 "description": ns_request.get("nsDescription", ""),
565 "constituent-vnfr-ref": [],
566
567 "operational-status": "init", # typedef ns-operational-
568 "config-status": "init", # typedef config-states
569 "detailed-status": "scheduled",
570
571 "orchestration-progress": {},
572 # {"networks": {"active": 0, "total": 0}, "vms": {"active": 0, "total": 0}},
573
574 "crete-time": now,
575 "nsd-name-ref": nsd["name"],
576 "operational-events": [], # "id", "timestamp", "description", "event",
577 "nsd-ref": nsd["id"],
578 "instantiate_params": ns_request,
579 "ns-instance-config-ref": nsr_id,
580 "id": nsr_id,
581 "_id": nsr_id,
582 # "input-parameter": xpath, value,
583 "ssh-authorized-key": ns_request.get("key-pair-ref"),
584 }
585 ns_request["nsr_id"] = nsr_id
586
587 # Create VNFR
588 needed_vnfds = {}
589 for member_vnf in nsd["constituent-vnfd"]:
590 vnfd_id = member_vnf["vnfd-id-ref"]
591 step = "getting vnfd id='{}' constituent-vnfd='{}' from database".format(
592 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
593 if vnfd_id not in needed_vnfds:
594 # Obtain vnfd
595 vnf_filter = {"id": vnfd_id}
596 self._add_read_filter(session, "vnfds", vnf_filter)
597 vnfd = self.db.get_one("vnfds", vnf_filter)
598 vnfd.pop("_admin")
599 needed_vnfds[vnfd_id] = vnfd
600 else:
601 vnfd = needed_vnfds[vnfd_id]
602 step = "filling vnfr vnfd-id='{}' constituent-vnfd='{}'".format(
603 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
604 vnfr_id = str(uuid4())
605 vnfr_descriptor = {
606 "id": vnfr_id,
607 "_id": vnfr_id,
608 "nsr-id-ref": nsr_id,
609 "member-vnf-index-ref": member_vnf["member-vnf-index"],
610 "created-time": now,
611 # "vnfd": vnfd, # at OSM model.but removed to avoid data duplication TODO: revise
612 "vnfd-ref": vnfd_id,
613 "vnfd-id": vnfd["_id"], # not at OSM model, but useful
614 "vim-account-id": None,
615 "vdur": [],
616 "connection-point": [],
617 "ip-address": None, # mgmt-interface filled by LCM
618 }
619 for cp in vnfd.get("connection-point", ()):
620 vnf_cp = {
621 "name": cp["name"],
622 "connection-point-id": cp.get("id"),
623 "id": cp.get("id"),
624 # "ip-address", "mac-address" # filled by LCM
625 # vim-id # TODO it would be nice having a vim port id
626 }
627 vnfr_descriptor["connection-point"].append(vnf_cp)
628 for vdu in vnfd["vdu"]:
629 vdur_id = str(uuid4())
630 vdur = {
631 "id": vdur_id,
632 "vdu-id-ref": vdu["id"],
633 # TODO "name": "" Name of the VDU in the VIM
634 "ip-address": None, # mgmt-interface filled by LCM
635 # "vim-id", "flavor-id", "image-id", "management-ip" # filled by LCM
636 "internal-connection-point": [],
637 "interfaces": [],
638 }
639 # TODO volumes: name, volume-id
640 for icp in vdu.get("internal-connection-point", ()):
641 vdu_icp = {
642 "id": icp["id"],
643 "connection-point-id": icp["id"],
644 "name": icp.get("name"),
645 # "ip-address", "mac-address" # filled by LCM
646 # vim-id # TODO it would be nice having a vim port id
647 }
648 vdur["internal-connection-point"].append(vdu_icp)
649 for iface in vdu.get("interface", ()):
650 vdu_iface = {
651 "name": iface.get("name"),
652 # "ip-address", "mac-address" # filled by LCM
653 # vim-id # TODO it would be nice having a vim port id
654 }
655 vdur["interfaces"].append(vdu_iface)
656 vnfr_descriptor["vdur"].append(vdur)
657
658 step = "creating vnfr vnfd-id='{}' constituent-vnfd='{}' at database".format(
659 member_vnf["vnfd-id-ref"], member_vnf["member-vnf-index"])
660 self._format_new_data(session, "vnfrs", vnfr_descriptor)
661 self.db.create("vnfrs", vnfr_descriptor)
662 rollback.insert(0, {"item": "vnfrs", "_id": vnfr_id})
663 nsr_descriptor["constituent-vnfr-ref"].append(vnfr_id)
664
665 step = "creating nsr at database"
666 self._format_new_data(session, "nsrs", nsr_descriptor)
667 self.db.create("nsrs", nsr_descriptor)
668 rollback.insert(rollback_index, {"item": "nsrs", "_id": nsr_id})
669 return nsr_id
670 except Exception as e:
671 raise EngineException("Error {}: {}".format(step, e))
672
673 @staticmethod
674 def _update_descriptor(desc, kwargs):
675 """
676 Update descriptor with the kwargs. It contains dot separated keys
677 :param desc: dictionary to be updated
678 :param kwargs: plain dictionary to be used for updating.
679 :return:
680 """
681 if not kwargs:
682 return
683 try:
684 for k, v in kwargs.items():
685 update_content = desc
686 kitem_old = None
687 klist = k.split(".")
688 for kitem in klist:
689 if kitem_old is not None:
690 update_content = update_content[kitem_old]
691 if isinstance(update_content, dict):
692 kitem_old = kitem
693 elif isinstance(update_content, list):
694 kitem_old = int(kitem)
695 else:
696 raise EngineException(
697 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem))
698 update_content[kitem_old] = v
699 except KeyError:
700 raise EngineException(
701 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old))
702 except ValueError:
703 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
704 k, kitem))
705 except IndexError:
706 raise EngineException(
707 "Invalid query string '{}'. Index '{}' out of range".format(k, kitem_old))
708
709 def new_item(self, rollback, session, item, indata={}, kwargs=None, headers={}, force=False):
710 """
711 Creates a new entry into database. For nsds and vnfds it creates an almost empty DISABLED entry,
712 that must be completed with a call to method upload_content
713 :param rollback: list where this method appends created items at database in case a rollback may to be done
714 :param session: contains the used login username and working project
715 :param item: it can be: users, projects, vim_accounts, sdns, nsrs, nsds, vnfds
716 :param indata: data to be inserted
717 :param kwargs: used to override the indata descriptor
718 :param headers: http request headers
719 :param force: If True avoid some dependence checks
720 :return: _id: identity of the inserted data.
721 """
722
723 try:
724 item_envelop = item
725 if item in ("nsds", "vnfds"):
726 item_envelop = "userDefinedData"
727 content = self._remove_envelop(item_envelop, indata)
728
729 # Override descriptor with query string kwargs
730 self._update_descriptor(content, kwargs)
731 if not indata and item not in ("nsds", "vnfds"):
732 raise EngineException("Empty payload")
733
734 validate_input(content, item, new=True)
735
736 if item == "nsrs":
737 # in this case the input descriptor is not the data to be stored
738 return self.new_nsr(rollback, session, ns_request=content)
739
740 self._validate_new_data(session, item_envelop, content, force)
741 if item in ("nsds", "vnfds"):
742 content = {"_admin": {"userDefinedData": content}}
743 self._format_new_data(session, item, content)
744 _id = self.db.create(item, content)
745 rollback.insert(0, {"item": item, "_id": _id})
746
747 if item == "vim_accounts":
748 msg_data = self.db.get_one(item, {"_id": _id})
749 msg_data.pop("_admin", None)
750 self.msg.write("vim_account", "create", msg_data)
751 elif item == "sdns":
752 msg_data = self.db.get_one(item, {"_id": _id})
753 msg_data.pop("_admin", None)
754 self.msg.write("sdn", "create", msg_data)
755 return _id
756 except ValidationError as e:
757 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
758
759 def new_nslcmop(self, session, nsInstanceId, operation, params):
760 now = time()
761 _id = str(uuid4())
762 nslcmop = {
763 "id": _id,
764 "_id": _id,
765 "operationState": "PROCESSING", # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
766 "statusEnteredTime": now,
767 "nsInstanceId": nsInstanceId,
768 "lcmOperationType": operation,
769 "startTime": now,
770 "isAutomaticInvocation": False,
771 "operationParams": params,
772 "isCancelPending": False,
773 "links": {
774 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id,
775 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsInstanceId,
776 }
777 }
778 return nslcmop
779
780 def ns_operation(self, rollback, session, nsInstanceId, operation, indata, kwargs=None):
781 """
782 Performs a new operation over a ns
783 :param rollback: list where this method appends created items at database in case a rollback may to be done
784 :param session: contains the used login username and working project
785 :param nsInstanceId: _id of the nsr to perform the operation
786 :param operation: it can be: instantiate, terminate, action, TODO: update, heal
787 :param indata: descriptor with the parameters of the operation
788 :param kwargs: used to override the indata descriptor
789 :return: id of the nslcmops
790 """
791 try:
792 # Override descriptor with query string kwargs
793 self._update_descriptor(indata, kwargs)
794 validate_input(indata, "ns_" + operation, new=True)
795 # get ns from nsr_id
796 nsr = self.get_item(session, "nsrs", nsInstanceId)
797 if not nsr["_admin"].get("nsState") or nsr["_admin"]["nsState"] == "NOT_INSTANTIATED":
798 if operation == "terminate" and indata.get("autoremove"):
799 # NSR must be deleted
800 return self.del_item(session, "nsrs", nsInstanceId)
801 if operation != "instantiate":
802 raise EngineException("ns_instance '{}' cannot be '{}' because it is not instantiated".format(
803 nsInstanceId, operation), HTTPStatus.CONFLICT)
804 else:
805 if operation == "instantiate" and not indata.get("force"):
806 raise EngineException("ns_instance '{}' cannot be '{}' because it is already instantiated".format(
807 nsInstanceId, operation), HTTPStatus.CONFLICT)
808 indata["nsInstanceId"] = nsInstanceId
809 self._check_ns_operation(session, nsr, operation, indata)
810 nslcmop = self.new_nslcmop(session, nsInstanceId, operation, indata)
811 self._format_new_data(session, "nslcmops", nslcmop)
812 _id = self.db.create("nslcmops", nslcmop)
813 rollback.insert(0, {"item": "nslcmops", "_id": _id})
814 indata["_id"] = _id
815 self.msg.write("ns", operation, nslcmop)
816 return _id
817 except ValidationError as e:
818 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
819 # except DbException as e:
820 # raise EngineException("Cannot get ns_instance '{}': {}".format(e), HTTPStatus.NOT_FOUND)
821
822 def _add_read_filter(self, session, item, filter):
823 if session["project_id"] == "admin": # allows all
824 return filter
825 if item == "users":
826 filter["username"] = session["username"]
827 elif item in ("vnfds", "nsds", "nsrs"):
828 filter["_admin.projects_read.cont"] = ["ANY", session["project_id"]]
829
830 def _add_delete_filter(self, session, item, filter):
831 if session["project_id"] != "admin" and item in ("users", "projects"):
832 raise EngineException("Only admin users can perform this task", http_code=HTTPStatus.FORBIDDEN)
833 if item == "users":
834 if filter.get("_id") == session["username"] or filter.get("username") == session["username"]:
835 raise EngineException("You cannot delete your own user", http_code=HTTPStatus.CONFLICT)
836 elif item == "project":
837 if filter.get("_id") == session["project_id"]:
838 raise EngineException("You cannot delete your own project", http_code=HTTPStatus.CONFLICT)
839 elif item in ("vnfds", "nsds") and session["project_id"] != "admin":
840 filter["_admin.projects_write.cont"] = ["ANY", session["project_id"]]
841
842 def get_file(self, session, item, _id, path=None, accept_header=None):
843 """
844 Return the file content of a vnfd or nsd
845 :param session: contains the used login username and working project
846 :param item: it can be vnfds or nsds
847 :param _id: Identity of the vnfd, ndsd
848 :param path: artifact path or "$DESCRIPTOR" or None
849 :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
850 :return: opened file or raises an exception
851 """
852 accept_text = accept_zip = False
853 if accept_header:
854 if 'text/plain' in accept_header or '*/*' in accept_header:
855 accept_text = True
856 if 'application/zip' in accept_header or '*/*' in accept_header:
857 accept_zip = True
858 if not accept_text and not accept_zip:
859 raise EngineException("provide request header 'Accept' with 'application/zip' or 'text/plain'",
860 http_code=HTTPStatus.NOT_ACCEPTABLE)
861
862 content = self.get_item(session, item, _id)
863 if content["_admin"]["onboardingState"] != "ONBOARDED":
864 raise EngineException("Cannot get content because this resource is not at 'ONBOARDED' state. "
865 "onboardingState is {}".format(content["_admin"]["onboardingState"]),
866 http_code=HTTPStatus.CONFLICT)
867 storage = content["_admin"]["storage"]
868 if path is not None and path != "$DESCRIPTOR": # artifacts
869 if not storage.get('pkg-dir'):
870 raise EngineException("Packages does not contains artifacts", http_code=HTTPStatus.BAD_REQUEST)
871 if self.fs.file_exists((storage['folder'], storage['pkg-dir'], *path), 'dir'):
872 folder_content = self.fs.dir_ls((storage['folder'], storage['pkg-dir'], *path))
873 return folder_content, "text/plain"
874 # TODO manage folders in http
875 else:
876 return self.fs.file_open((storage['folder'], storage['pkg-dir'], *path), "rb"),\
877 "application/octet-stream"
878
879 # pkgtype accept ZIP TEXT -> result
880 # manyfiles yes X -> zip
881 # no yes -> error
882 # onefile yes no -> zip
883 # X yes -> text
884
885 if accept_text and (not storage.get('pkg-dir') or path == "$DESCRIPTOR"):
886 return self.fs.file_open((storage['folder'], storage['descriptor']), "r"), "text/plain"
887 elif storage.get('pkg-dir') and not accept_zip:
888 raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'"
889 "Accept header", http_code=HTTPStatus.NOT_ACCEPTABLE)
890 else:
891 if not storage.get('zipfile'):
892 # TODO generate zipfile if not present
893 raise EngineException("Only allowed 'text/plain' Accept header for this descriptor. To be solved in "
894 "future versions", http_code=HTTPStatus.NOT_ACCEPTABLE)
895 return self.fs.file_open((storage['folder'], storage['zipfile']), "rb"), "application/zip"
896
897 def get_item_list(self, session, item, filter={}):
898 """
899 Get a list of items
900 :param session: contains the used login username and working project
901 :param item: it can be: users, projects, vnfds, nsds, ...
902 :param filter: filter of data to be applied
903 :return: The list, it can be empty if no one match the filter.
904 """
905 # TODO add admin to filter, validate rights
906 # TODO transform data for SOL005 URL requests. Transform filtering
907 # TODO implement "field-type" query string SOL005
908
909 self._add_read_filter(session, item, filter)
910 return self.db.get_list(item, filter)
911
912 def get_item(self, session, item, _id):
913 """
914 Get complete information on an items
915 :param session: contains the used login username and working project
916 :param item: it can be: users, projects, vnfds, nsds,
917 :param _id: server id of the item
918 :return: dictionary, raise exception if not found.
919 """
920 filter = {"_id": _id}
921 # TODO add admin to filter, validate rights
922 # TODO transform data for SOL005 URL requests
923 self._add_read_filter(session, item, filter)
924 return self.db.get_one(item, filter)
925
926 def del_item_list(self, session, item, filter={}):
927 """
928 Delete a list of items
929 :param session: contains the used login username and working project
930 :param item: it can be: users, projects, vnfds, nsds, ...
931 :param filter: filter of data to be applied
932 :return: The deleted list, it can be empty if no one match the filter.
933 """
934 # TODO add admin to filter, validate rights
935 self._add_read_filter(session, item, filter)
936 return self.db.del_list(item, filter)
937
938 def del_item(self, session, item, _id, force=False):
939 """
940 Delete item by its internal id
941 :param session: contains the used login username and working project
942 :param item: it can be: users, projects, vnfds, nsds, ...
943 :param _id: server id of the item
944 :param force: indicates if deletion must be forced in case of conflict
945 :return: dictionary with deleted item _id. It raises exception if not found.
946 """
947 # TODO add admin to filter, validate rights
948 # data = self.get_item(item, _id)
949 filter = {"_id": _id}
950 self._add_delete_filter(session, item, filter)
951 if item in ("vnfds", "nsds") and not force:
952 descriptor = self.get_item(session, item, _id)
953 descriptor_id = descriptor.get("id")
954 if descriptor_id:
955 self._check_dependencies_on_descriptor(session, item, descriptor_id)
956
957 if item == "nsrs":
958 nsr = self.db.get_one(item, filter)
959 if nsr["_admin"].get("nsState") == "INSTANTIATED" and not force:
960 raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. "
961 "Launch 'terminate' operation first; or force deletion".format(_id),
962 http_code=HTTPStatus.CONFLICT)
963 v = self.db.del_one(item, {"_id": _id})
964 self.db.del_list("nslcmops", {"nsInstanceId": _id})
965 self.db.del_list("vnfrs", {"nsr-id-ref": _id})
966 self.msg.write("ns", "deleted", {"_id": _id})
967 return v
968 if item in ("vim_accounts", "sdns") and not force:
969 self.db.set_one(item, {"_id": _id}, {"_admin.to_delete": True}) # TODO change status
970 if item == "vim_accounts":
971 self.msg.write("vim_account", "delete", {"_id": _id})
972 elif item == "sdns":
973 self.msg.write("sdn", "delete", {"_id": _id})
974 return {"deleted": 1} # TODO indicate an offline operation to return 202 ACCEPTED
975
976 v = self.db.del_one(item, filter)
977 if item in ("vnfds", "nsds"):
978 self.fs.file_delete(_id, ignore_non_exist=True)
979 if item in ("vim_accounts", "sdns", "vnfds", "nsds"):
980 self.msg.write(item[:-1], "deleted", {"_id": _id})
981 return v
982
983 def prune(self):
984 """
985 Prune database not needed content
986 :return: None
987 """
988 return self.db.del_list("nsrs", {"_admin.to_delete": True})
989
990 def create_admin(self):
991 """
992 Creates a new user admin/admin into database if database is empty. Useful for initialization
993 :return: _id identity of the inserted data, or None
994 """
995 users = self.db.get_one("users", fail_on_empty=False, fail_on_more=False)
996 if users:
997 return None
998 # raise EngineException("Unauthorized. Database users is not empty", HTTPStatus.UNAUTHORIZED)
999 indata = {"username": "admin", "password": "admin", "projects": ["admin"]}
1000 fake_session = {"project_id": "admin", "username": "admin"}
1001 self._format_new_data(fake_session, "users", indata)
1002 _id = self.db.create("users", indata)
1003 return _id
1004
1005 def init_db(self, target_version='1.0'):
1006 """
1007 Init database if empty. If not empty it checks that database version is ok.
1008 If empty, it creates a new user admin/admin at 'users' and a new entry at 'version'
1009 :return: None if ok, exception if error or if the version is different.
1010 """
1011 version = self.db.get_one("version", fail_on_empty=False, fail_on_more=False)
1012 if not version:
1013 # create user admin
1014 self.create_admin()
1015 # create database version
1016 version_data = {
1017 "_id": '1.0', # version text
1018 "version": 1000, # version number
1019 "date": "2018-04-12", # version date
1020 "description": "initial design", # changes in this version
1021 'status': 'ENABLED' # ENABLED, DISABLED (migration in process), ERROR,
1022 }
1023 self.db.create("version", version_data)
1024 elif version["_id"] != target_version:
1025 # TODO implement migration process
1026 raise EngineException("Wrong database version '{}'. Expected '{}'".format(
1027 version["_id"], target_version), HTTPStatus.INTERNAL_SERVER_ERROR)
1028 elif version["status"] != 'ENABLED':
1029 raise EngineException("Wrong database status '{}'".format(
1030 version["status"]), HTTPStatus.INTERNAL_SERVER_ERROR)
1031 return
1032
1033 def _edit_item(self, session, item, id, content, indata={}, kwargs=None, force=False):
1034 if indata:
1035 indata = self._remove_envelop(item, indata)
1036
1037 # Override descriptor with query string kwargs
1038 if kwargs:
1039 try:
1040 for k, v in kwargs.items():
1041 update_content = indata
1042 kitem_old = None
1043 klist = k.split(".")
1044 for kitem in klist:
1045 if kitem_old is not None:
1046 update_content = update_content[kitem_old]
1047 if isinstance(update_content, dict):
1048 kitem_old = kitem
1049 elif isinstance(update_content, list):
1050 kitem_old = int(kitem)
1051 else:
1052 raise EngineException(
1053 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k, kitem))
1054 update_content[kitem_old] = v
1055 except KeyError:
1056 raise EngineException(
1057 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k, kitem_old))
1058 except ValueError:
1059 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
1060 k, kitem))
1061 except IndexError:
1062 raise EngineException(
1063 "Invalid query string '{}'. Index '{}' out of range".format(k, kitem_old))
1064 try:
1065 validate_input(indata, item, new=False)
1066 except ValidationError as e:
1067 raise EngineException(e, HTTPStatus.UNPROCESSABLE_ENTITY)
1068
1069 _deep_update(content, indata)
1070 self._validate_new_data(session, item, content, id, force)
1071 # self._format_new_data(session, item, content)
1072 self.db.replace(item, id, content)
1073 if item in ("vim_accounts", "sdns"):
1074 indata.pop("_admin", None)
1075 indata["_id"] = id
1076 if item == "vim_accounts":
1077 self.msg.write("vim_account", "edit", indata)
1078 elif item == "sdns":
1079 self.msg.write("sdn", "edit", indata)
1080 return id
1081
1082 def edit_item(self, session, item, _id, indata={}, kwargs=None, force=False):
1083 """
1084 Update an existing entry at database
1085 :param session: contains the used login username and working project
1086 :param item: it can be: users, projects, vnfds, nsds, ...
1087 :param _id: identifier to be updated
1088 :param indata: data to be inserted
1089 :param kwargs: used to override the indata descriptor
1090 :param force: If True avoid some dependence checks
1091 :return: dictionary, raise exception if not found.
1092 """
1093
1094 content = self.get_item(session, item, _id)
1095 return self._edit_item(session, item, _id, content, indata, kwargs, force)