1 # -*- coding: utf-8 -*-
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
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
20 from copy
import deepcopy
, copy
21 from validation
import validate_input
, ValidationError
23 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
26 class EngineException(Exception):
28 def __init__(self
, message
, http_code
=HTTPStatus
.BAD_REQUEST
):
29 self
.http_code
= http_code
30 Exception.__init
__(self
, message
)
33 def _deep_update(dict_to_change
, dict_reference
):
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
40 for k
in dict_reference
:
41 if dict_reference
[k
] is None: # None->Anything
42 if k
in dict_to_change
:
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
])
52 dict_to_change
[k
] = deepcopy(dict_reference
[k
])
53 _deep_update(dict_to_change
[k
], dict_reference
[k
])
56 def get_iterable(input):
58 Returns an iterable, in case input is None it just returns an empty tuple
75 self
.logger
= logging
.getLogger("nbi.engine")
77 def start(self
, config
):
79 Connect to database, filesystem storage, and messaging
80 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
86 if config
["database"]["driver"] == "mongo":
87 self
.db
= dbmongo
.DbMongo()
88 self
.db
.db_connect(config
["database"])
89 elif config
["database"]["driver"] == "memory":
90 self
.db
= dbmemory
.DbMemory()
91 self
.db
.db_connect(config
["database"])
93 raise EngineException("Invalid configuration param '{}' at '[database]':'driver'".format(
94 config
["database"]["driver"]))
96 if config
["storage"]["driver"] == "local":
97 self
.fs
= fslocal
.FsLocal()
98 self
.fs
.fs_connect(config
["storage"])
100 raise EngineException("Invalid configuration param '{}' at '[storage]':'driver'".format(
101 config
["storage"]["driver"]))
103 if config
["message"]["driver"] == "local":
104 self
.msg
= msglocal
.MsgLocal()
105 self
.msg
.connect(config
["message"])
106 elif config
["message"]["driver"] == "kafka":
107 self
.msg
= msgkafka
.MsgKafka()
108 self
.msg
.connect(config
["message"])
110 raise EngineException("Invalid configuration param '{}' at '[message]':'driver'".format(
111 config
["storage"]["driver"]))
112 except (DbException
, FsException
, MsgException
) as e
:
113 raise EngineException(str(e
), http_code
=e
.http_code
)
118 self
.db
.db_disconnect()
120 self
.fs
.fs_disconnect()
122 self
.fs
.fs_disconnect()
123 except (DbException
, FsException
, MsgException
) as e
:
124 raise EngineException(str(e
), http_code
=e
.http_code
)
126 def authorize(self
, token
):
129 raise EngineException("Needed a token or Authorization http header",
130 http_code
=HTTPStatus
.UNAUTHORIZED
)
131 if token
not in self
.tokens
:
132 raise EngineException("Invalid token or Authorization http header",
133 http_code
=HTTPStatus
.UNAUTHORIZED
)
134 session
= self
.tokens
[token
]
136 if session
["expires"] < now
:
137 del self
.tokens
[token
]
138 raise EngineException("Expired Token or Authorization http header",
139 http_code
=HTTPStatus
.UNAUTHORIZED
)
141 except EngineException
:
142 if self
.config
["global"].get("test.user_not_authorized"):
143 return {"id": "fake-token-id-for-test",
144 "project_id": self
.config
["global"].get("test.project_not_authorized", "admin"),
145 "username": self
.config
["global"]["test.user_not_authorized"]}
149 def new_token(self
, session
, indata
, remote
):
153 # Try using username/password
154 if indata
.get("username"):
155 user_rows
= self
.db
.get_list("users", {"username": indata
.get("username")})
158 user_content
= user_rows
[0]
159 salt
= user_content
["_admin"]["salt"]
160 shadow_password
= sha256(indata
.get("password", "").encode('utf-8') + salt
.encode('utf-8')).hexdigest()
161 if shadow_password
!= user_content
["password"]:
164 raise EngineException("Invalid username/password", http_code
=HTTPStatus
.UNAUTHORIZED
)
166 user_rows
= self
.db
.get_list("users", {"username": session
["username"]})
168 user_content
= user_rows
[0]
170 raise EngineException("Invalid token", http_code
=HTTPStatus
.UNAUTHORIZED
)
172 raise EngineException("Provide credentials: username/password or Authorization Bearer token",
173 http_code
=HTTPStatus
.UNAUTHORIZED
)
175 token_id
= ''.join(random_choice('abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789')
176 for _
in range(0, 32))
177 if indata
.get("project_id"):
178 project_id
= indata
.get("project_id")
179 if project_id
not in user_content
["projects"]:
180 raise EngineException("project {} not allowed for this user".format(project_id
),
181 http_code
=HTTPStatus
.UNAUTHORIZED
)
183 project_id
= user_content
["projects"][0]
184 if project_id
== "admin":
187 project
= self
.db
.get_one("projects", {"_id": project_id
})
188 session_admin
= project
.get("admin", False)
189 new_session
= {"issued_at": now
, "expires": now
+3600,
190 "_id": token_id
, "id": token_id
, "project_id": project_id
, "username": user_content
["username"],
191 "remote_port": remote
.port
, "admin": session_admin
}
193 new_session
["remote_host"] = remote
.name
195 new_session
["remote_host"] = remote
.ip
197 self
.tokens
[token_id
] = new_session
198 return deepcopy(new_session
)
200 def get_token_list(self
, session
):
202 for token_id
, token_value
in self
.tokens
.items():
203 if token_value
["username"] == session
["username"]:
204 token_list
.append(deepcopy(token_value
))
207 def get_token(self
, session
, token_id
):
208 token_value
= self
.tokens
.get(token_id
)
210 raise EngineException("token not found", http_code
=HTTPStatus
.NOT_FOUND
)
211 if token_value
["username"] != session
["username"] and not session
["admin"]:
212 raise EngineException("needed admin privileges", http_code
=HTTPStatus
.UNAUTHORIZED
)
215 def del_token(self
, token_id
):
217 del self
.tokens
[token_id
]
218 return "token '{}' deleted".format(token_id
)
220 raise EngineException("Token '{}' not found".format(token_id
), http_code
=HTTPStatus
.NOT_FOUND
)
223 def _remove_envelop(item
, indata
=None):
225 Obtain the useful data removing the envelop. It goes throw the vnfd or nsd catalog and returns the
227 :param item: can be vnfds, nsds, users, projects, userDefinedData (initial content of a vnfds, nsds
228 :param indata: Content to be inspected
229 :return: the useful part of indata
231 clean_indata
= indata
235 if clean_indata
.get('vnfd:vnfd-catalog'):
236 clean_indata
= clean_indata
['vnfd:vnfd-catalog']
237 elif clean_indata
.get('vnfd-catalog'):
238 clean_indata
= clean_indata
['vnfd-catalog']
239 if clean_indata
.get('vnfd'):
240 if not isinstance(clean_indata
['vnfd'], list) or len(clean_indata
['vnfd']) != 1:
241 raise EngineException("'vnfd' must be a list only one element")
242 clean_indata
= clean_indata
['vnfd'][0]
244 if clean_indata
.get('nsd:nsd-catalog'):
245 clean_indata
= clean_indata
['nsd:nsd-catalog']
246 elif clean_indata
.get('nsd-catalog'):
247 clean_indata
= clean_indata
['nsd-catalog']
248 if clean_indata
.get('nsd'):
249 if not isinstance(clean_indata
['nsd'], list) or len(clean_indata
['nsd']) != 1:
250 raise EngineException("'nsd' must be a list only one element")
251 clean_indata
= clean_indata
['nsd'][0]
252 elif item
== "userDefinedData":
253 if "userDefinedData" in indata
:
254 clean_indata
= clean_indata
['userDefinedData']
257 def _check_dependencies_on_descriptor(self
, session
, item
, descriptor_id
, _id
):
259 Check that the descriptor to be deleded is not a dependency of others
260 :param session: client session information
261 :param item: can be vnfds, nsds
262 :param descriptor_id: id (provided by client) of descriptor to be deleted
263 :param _id: internal id of descriptor to be deleted
264 :return: None or raises exception
267 _filter
= {"constituent-vnfd.ANYINDEX.vnfd-id-ref": descriptor_id
}
268 if self
.get_item_list(session
, "nsds", _filter
):
269 raise EngineException("There are nsd that depends on this VNFD", http_code
=HTTPStatus
.CONFLICT
)
270 if self
.get_item_list(session
, "vnfrs", {"vnfd-id": _id
}):
271 raise EngineException("There are vnfr that depends on this VNFD", http_code
=HTTPStatus
.CONFLICT
)
273 _filter
= {"nsdId": _id
}
274 if self
.get_item_list(session
, "nsrs", _filter
):
275 raise EngineException("There are nsr that depends on this NSD", http_code
=HTTPStatus
.CONFLICT
)
277 def _check_descriptor_dependencies(self
, session
, item
, descriptor
):
279 Check that the dependent descriptors exist on a new descriptor or edition
280 :param session: client session information
281 :param item: can be nsds, nsrs
282 :param descriptor: descriptor to be inserted or edit
283 :return: None or raises exception
286 if not descriptor
.get("constituent-vnfd"):
288 for vnf
in descriptor
["constituent-vnfd"]:
289 vnfd_id
= vnf
["vnfd-id-ref"]
290 if not self
.get_item_list(session
, "vnfds", {"id": vnfd_id
}):
291 raise EngineException("Descriptor error at 'constituent-vnfd':'vnfd-id-ref'='{}' references a non "
292 "existing vnfd".format(vnfd_id
), http_code
=HTTPStatus
.CONFLICT
)
294 if not descriptor
.get("nsdId"):
296 nsd_id
= descriptor
["nsdId"]
297 if not self
.get_item_list(session
, "nsds", {"id": nsd_id
}):
298 raise EngineException("Descriptor error at nsdId='{}' references a non exist nsd".format(nsd_id
),
299 http_code
=HTTPStatus
.CONFLICT
)
301 def _validate_new_data(self
, session
, item
, indata
, id=None, force
=False):
303 if not indata
.get("username"):
304 raise EngineException("missing 'username'", HTTPStatus
.UNPROCESSABLE_ENTITY
)
305 if not indata
.get("password"):
306 raise EngineException("missing 'password'", HTTPStatus
.UNPROCESSABLE_ENTITY
)
307 if not indata
.get("projects"):
308 raise EngineException("missing 'projects'", HTTPStatus
.UNPROCESSABLE_ENTITY
)
309 # check username not exists
310 if self
.db
.get_one(item
, {"username": indata
.get("username")}, fail_on_empty
=False, fail_on_more
=False):
311 raise EngineException("username '{}' exists".format(indata
["username"]), HTTPStatus
.CONFLICT
)
312 elif item
== "projects":
313 if not indata
.get("name"):
314 raise EngineException("missing 'name'")
315 # check name not exists
316 if self
.db
.get_one(item
, {"name": indata
.get("name")}, fail_on_empty
=False, fail_on_more
=False):
317 raise EngineException("name '{}' exists".format(indata
["name"]), HTTPStatus
.CONFLICT
)
318 elif item
in ("vnfds", "nsds"):
319 filter = {"id": indata
["id"]}
321 filter["_id.neq"] = id
322 # TODO add admin to filter, validate rights
323 self
._add
_read
_filter
(session
, item
, filter)
324 if self
.db
.get_one(item
, filter, fail_on_empty
=False):
325 raise EngineException("{} with id '{}' already exists for this tenant".format(item
[:-1], indata
["id"]),
327 # TODO validate with pyangbind. Load and dumps to convert data types
329 # transform constituent-vnfd:member-vnf-index to string
330 if indata
.get("constituent-vnfd"):
331 for constituent_vnfd
in indata
["constituent-vnfd"]:
332 if "member-vnf-index" in constituent_vnfd
:
333 constituent_vnfd
["member-vnf-index"] = str(constituent_vnfd
["member-vnf-index"])
335 if item
== "nsds" and not force
:
336 self
._check
_descriptor
_dependencies
(session
, "nsds", indata
)
337 elif item
== "userDefinedData":
338 # TODO validate userDefinedData is a keypair values
343 elif item
== "vim_accounts" or item
== "sdns":
344 filter = {"name": indata
.get("name")}
346 filter["_id.neq"] = id
347 if self
.db
.get_one(item
, filter, fail_on_empty
=False, fail_on_more
=False):
348 raise EngineException("name '{}' already exists for {}".format(indata
["name"], item
),
351 def _check_ns_operation(self
, session
, nsr
, operation
, indata
):
353 Check that user has enter right parameters for the operation
355 :param operation: it can be: instantiate, terminate, action, TODO: update, heal
356 :param indata: descriptor with the parameters of the operation
363 def check_valid_vnf_member_index(member_vnf_index
):
364 for vnf
in nsd
["constituent-vnfd"]:
365 if member_vnf_index
== vnf
["member-vnf-index"]:
366 vnfd_id
= vnf
["vnfd-id-ref"]
367 if vnfd_id
not in vnfds
:
368 vnfds
[vnfd_id
] = self
.db
.get_one("vnfds", {"id": vnfd_id
})
369 return vnfds
[vnfd_id
]
371 raise EngineException("Invalid parameter member_vnf_index='{}' is not one of the "
372 "nsd:constituent-vnfd".format(member_vnf_index
))
374 def check_valid_vim_account(vim_account
):
375 if vim_account
in vim_accounts
:
378 self
.db
.get_one("vim_accounts", {"_id": vim_account
})
380 raise EngineException("Invalid vimAccountId='{}' not present".format(vim_account
))
381 vim_accounts
.append(vim_account
)
383 if operation
== "action":
384 # check vnf_member_index
385 if indata
.get("vnf_member_index"):
386 indata
["member_vnf_index"] = indata
.pop("vnf_member_index") # for backward compatibility
387 if not indata
.get("member_vnf_index"):
388 raise EngineException("Missing 'member_vnf_index' parameter")
389 vnfd
= check_valid_vnf_member_index(indata
["member_vnf_index"])
391 for config_primitive
in get_iterable(vnfd
.get("vnf-configuration", {}).get("config-primitive")):
392 if indata
["primitive"] == config_primitive
["name"]:
393 # check needed primitive_params are provided
394 if indata
.get("primitive_params"):
395 in_primitive_params_copy
= copy(indata
["primitive_params"])
397 in_primitive_params_copy
= {}
398 for paramd
in get_iterable(config_primitive
.get("parameter")):
399 if paramd
["name"] in in_primitive_params_copy
:
400 del in_primitive_params_copy
[paramd
["name"]]
401 elif not paramd
.get("default-value"):
402 raise EngineException("Needed parameter {} not provided for primitive '{}'".format(
403 paramd
["name"], indata
["primitive"]))
404 # check no extra primitive params are provided
405 if in_primitive_params_copy
:
406 raise EngineException("parameter/s '{}' not present at vnfd for primitive '{}'".format(
407 list(in_primitive_params_copy
.keys()), indata
["primitive"]))
410 raise EngineException("Invalid primitive '{}' is not present at vnfd".format(indata
["primitive"]))
411 if operation
== "scale":
412 vnfd
= check_valid_vnf_member_index(indata
["scaleVnfData"]["scaleByStepData"]["member-vnf-index"])
413 for scaling_group
in get_iterable(vnfd
.get("scaling-group-descriptor")):
414 if indata
["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"] == scaling_group
["name"]:
417 raise EngineException("Invalid scaleVnfData:scaleByStepData:scaling-group-descriptor '{}' is not "
418 "present at vnfd:scaling-group-descriptor".format(
419 indata
["scaleVnfData"]["scaleByStepData"]["scaling-group-descriptor"]))
420 if operation
== "instantiate":
422 check_valid_vim_account(indata
["vimAccountId"])
423 for in_vnf
in get_iterable(indata
.get("vnf")):
424 vnfd
= check_valid_vnf_member_index(in_vnf
["member-vnf-index"])
425 if in_vnf
.get("vimAccountId"):
426 check_valid_vim_account(in_vnf
["vimAccountId"])
427 for in_vdu
in get_iterable(in_vnf
.get("vdu")):
428 for vdud
in get_iterable(vnfd
.get("vdu")):
429 if vdud
["id"] == in_vdu
["id"]:
430 for volume
in get_iterable(in_vdu
.get("volume")):
431 for volumed
in get_iterable(vdud
.get("volumes")):
432 if volumed
["name"] == volume
["name"]:
435 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu[id='{}']:"
436 "volume:name='{}' is not present at vnfd:vdu:volumes list".
437 format(in_vnf
["member-vnf-index"], in_vdu
["id"],
441 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:vdu:id='{}' is not "
442 "present at vnfd".format(in_vnf
["member-vnf-index"], in_vdu
["id"]))
444 for in_internal_vld
in get_iterable(in_vnf
.get("internal-vld")):
445 for internal_vldd
in get_iterable(vnfd
.get("internal-vld")):
446 if in_internal_vld
["name"] == internal_vldd
["name"] or \
447 in_internal_vld
["name"] == internal_vldd
["id"]:
450 raise EngineException("Invalid parameter vnf[member-vnf-index='{}']:internal-vld:name='{}'"
451 " is not present at vnfd '{}'".format(in_vnf
["member-vnf-index"],
452 in_internal_vld
["name"],
454 for in_vld
in get_iterable(indata
.get("vld")):
455 for vldd
in get_iterable(nsd
.get("vld")):
456 if in_vld
["name"] == vldd
["name"] or in_vld
["name"] == vldd
["id"]:
459 raise EngineException("Invalid parameter vld:name='{}' is not present at nsd:vld".format(
462 def _format_new_data(self
, session
, item
, indata
):
464 if "_admin" not in indata
:
465 indata
["_admin"] = {}
466 indata
["_admin"]["created"] = now
467 indata
["_admin"]["modified"] = now
469 indata
["_id"] = indata
["username"]
471 indata
["_admin"]["salt"] = salt
472 indata
["password"] = sha256(indata
["password"].encode('utf-8') + salt
.encode('utf-8')).hexdigest()
473 elif item
== "projects":
474 indata
["_id"] = indata
["name"]
476 if not indata
.get("_id"):
477 indata
["_id"] = str(uuid4())
478 if item
in ("vnfds", "nsds", "nsrs", "vnfrs"):
479 if not indata
["_admin"].get("projects_read"):
480 indata
["_admin"]["projects_read"] = [session
["project_id"]]
481 if not indata
["_admin"].get("projects_write"):
482 indata
["_admin"]["projects_write"] = [session
["project_id"]]
483 if item
in ("vnfds", "nsds"):
484 indata
["_admin"]["onboardingState"] = "CREATED"
485 indata
["_admin"]["operationalState"] = "DISABLED"
486 indata
["_admin"]["usageSate"] = "NOT_IN_USE"
488 indata
["_admin"]["nsState"] = "NOT_INSTANTIATED"
489 if item
in ("vim_accounts", "sdns"):
490 indata
["_admin"]["operationalState"] = "PROCESSING"
492 def upload_content(self
, session
, item
, _id
, indata
, kwargs
, headers
):
494 Used for receiving content by chunks (with a transaction_id header and/or gzip file. It will store and extract)
495 :param session: session
496 :param item: can be nsds or vnfds
497 :param _id : the nsd,vnfd is already created, this is the id
498 :param indata: http body request
499 :param kwargs: user query string to override parameters. NOT USED
500 :param headers: http request headers
501 :return: True package has is completely uploaded or False if partial content has been uplodaed.
502 Raise exception on error
504 # Check that _id exists and it is valid
505 current_desc
= self
.get_item(session
, item
, _id
)
507 content_range_text
= headers
.get("Content-Range")
508 expected_md5
= headers
.get("Content-File-MD5")
510 content_type
= headers
.get("Content-Type")
511 if content_type
and "application/gzip" in content_type
or "application/x-gzip" in content_type
or \
512 "application/zip" in content_type
:
514 filename
= headers
.get("Content-Filename")
516 filename
= "package.tar.gz" if compressed
else "package"
517 # TODO change to Content-Disposition filename https://tools.ietf.org/html/rfc6266
521 if content_range_text
:
522 content_range
= content_range_text
.replace("-", " ").replace("/", " ").split()
523 if content_range
[0] != "bytes": # TODO check x<y not negative < total....
525 start
= int(content_range
[1])
526 end
= int(content_range
[2]) + 1
527 total
= int(content_range
[3])
532 if not self
.fs
.file_exists(_id
, 'dir'):
533 raise EngineException("invalid Transaction-Id header", HTTPStatus
.NOT_FOUND
)
535 self
.fs
.file_delete(_id
, ignore_non_exist
=True)
538 storage
= self
.fs
.get_params()
539 storage
["folder"] = _id
541 file_path
= (_id
, filename
)
542 if self
.fs
.file_exists(file_path
, 'file'):
543 file_size
= self
.fs
.file_size(file_path
)
546 if file_size
!= start
:
547 raise EngineException("invalid Content-Range start sequence, expected '{}' but received '{}'".format(
548 file_size
, start
), HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
549 file_pkg
= self
.fs
.file_open(file_path
, 'a+b')
550 if isinstance(indata
, dict):
551 indata_text
= yaml
.safe_dump(indata
, indent
=4, default_flow_style
=False)
552 file_pkg
.write(indata_text
.encode(encoding
="utf-8"))
556 indata_text
= indata
.read(4096)
557 indata_len
+= len(indata_text
)
560 file_pkg
.write(indata_text
)
561 if content_range_text
:
562 if indata_len
!= end
-start
:
563 raise EngineException("Mismatch between Content-Range header {}-{} and body length of {}".format(
564 start
, end
-1, indata_len
), HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
566 # TODO update to UPLOADING
573 chunk_data
= file_pkg
.read(1024)
575 file_md5
.update(chunk_data
)
576 chunk_data
= file_pkg
.read(1024)
577 if expected_md5
!= file_md5
.hexdigest():
578 raise EngineException("Error, MD5 mismatch", HTTPStatus
.CONFLICT
)
580 if compressed
== "gzip":
581 tar
= tarfile
.open(mode
='r', fileobj
=file_pkg
)
582 descriptor_file_name
= None
584 tarname
= tarinfo
.name
585 tarname_path
= tarname
.split("/")
586 if not tarname_path
[0] or ".." in tarname_path
: # if start with "/" means absolute path
587 raise EngineException("Absolute path or '..' are not allowed for package descriptor tar.gz")
588 if len(tarname_path
) == 1 and not tarinfo
.isdir():
589 raise EngineException("All files must be inside a dir for package descriptor tar.gz")
590 if tarname
.endswith(".yaml") or tarname
.endswith(".json") or tarname
.endswith(".yml"):
591 storage
["pkg-dir"] = tarname_path
[0]
592 if len(tarname_path
) == 2:
593 if descriptor_file_name
:
594 raise EngineException(
595 "Found more than one descriptor file at package descriptor tar.gz")
596 descriptor_file_name
= tarname
597 if not descriptor_file_name
:
598 raise EngineException("Not found any descriptor file at package descriptor tar.gz")
599 storage
["descriptor"] = descriptor_file_name
600 storage
["zipfile"] = filename
601 self
.fs
.file_extract(tar
, _id
)
602 with self
.fs
.file_open((_id
, descriptor_file_name
), "r") as descriptor_file
:
603 content
= descriptor_file
.read()
605 content
= file_pkg
.read()
606 storage
["descriptor"] = descriptor_file_name
= filename
608 if descriptor_file_name
.endswith(".json"):
609 error_text
= "Invalid json format "
610 indata
= json
.load(content
)
612 error_text
= "Invalid yaml format "
613 indata
= yaml
.load(content
)
615 current_desc
["_admin"]["storage"] = storage
616 current_desc
["_admin"]["onboardingState"] = "ONBOARDED"
617 current_desc
["_admin"]["operationalState"] = "ENABLED"
619 self
._edit
_item
(session
, item
, _id
, current_desc
, indata
, kwargs
)
620 # TODO if descriptor has changed because kwargs update content and remove cached zip
621 # TODO if zip is not present creates one
624 except EngineException
:
627 raise EngineException("invalid Content-Range header format. Expected 'bytes start-end/total'",
628 HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
630 raise EngineException("invalid upload transaction sequence: '{}'".format(e
), HTTPStatus
.BAD_REQUEST
)
631 except tarfile
.ReadError
as e
:
632 raise EngineException("invalid file content {}".format(e
), HTTPStatus
.BAD_REQUEST
)
633 except (ValueError, yaml
.YAMLError
) as e
:
634 raise EngineException(error_text
+ str(e
))
639 def new_nsr(self
, rollback
, session
, ns_request
):
641 Creates a new nsr into database. It also creates needed vnfrs
642 :param rollback: list where this method appends created items at database in case a rollback may to be done
643 :param session: contains the used login username and working project
644 :param ns_request: params to be used for the nsr
645 :return: the _id of nsr descriptor stored at database
647 rollback_index
= len(rollback
)
651 step
= "getting nsd id='{}' from database".format(ns_request
.get("nsdId"))
652 nsd
= self
.get_item(session
, "nsds", ns_request
["nsdId"])
653 nsr_id
= str(uuid4())
655 step
= "filling nsr from input data"
657 "name": ns_request
["nsName"],
658 "name-ref": ns_request
["nsName"],
659 "short-name": ns_request
["nsName"],
660 "admin-status": "ENABLED",
662 "datacenter": ns_request
["vimAccountId"],
663 "resource-orchestrator": "osmopenmano",
664 "description": ns_request
.get("nsDescription", ""),
665 "constituent-vnfr-ref": [],
667 "operational-status": "init", # typedef ns-operational-
668 "config-status": "init", # typedef config-states
669 "detailed-status": "scheduled",
671 "orchestration-progress": {},
672 # {"networks": {"active": 0, "total": 0}, "vms": {"active": 0, "total": 0}},
675 "nsd-name-ref": nsd
["name"],
676 "operational-events": [], # "id", "timestamp", "description", "event",
677 "nsd-ref": nsd
["id"],
679 "instantiate_params": ns_request
,
680 "ns-instance-config-ref": nsr_id
,
683 # "input-parameter": xpath, value,
684 "ssh-authorized-key": ns_request
.get("key-pair-ref"),
686 ns_request
["nsr_id"] = nsr_id
690 for member_vnf
in nsd
["constituent-vnfd"]:
691 vnfd_id
= member_vnf
["vnfd-id-ref"]
692 step
= "getting vnfd id='{}' constituent-vnfd='{}' from database".format(
693 member_vnf
["vnfd-id-ref"], member_vnf
["member-vnf-index"])
694 if vnfd_id
not in needed_vnfds
:
696 vnf_filter
= {"id": vnfd_id
}
697 self
._add
_read
_filter
(session
, "vnfds", vnf_filter
)
698 vnfd
= self
.db
.get_one("vnfds", vnf_filter
)
700 needed_vnfds
[vnfd_id
] = vnfd
702 vnfd
= needed_vnfds
[vnfd_id
]
703 step
= "filling vnfr vnfd-id='{}' constituent-vnfd='{}'".format(
704 member_vnf
["vnfd-id-ref"], member_vnf
["member-vnf-index"])
705 vnfr_id
= str(uuid4())
709 "nsr-id-ref": nsr_id
,
710 "member-vnf-index-ref": member_vnf
["member-vnf-index"],
712 # "vnfd": vnfd, # at OSM model.but removed to avoid data duplication TODO: revise
714 "vnfd-id": vnfd
["_id"], # not at OSM model, but useful
715 "vim-account-id": None,
717 "connection-point": [],
718 "ip-address": None, # mgmt-interface filled by LCM
720 for cp
in vnfd
.get("connection-point", ()):
723 "connection-point-id": cp
.get("id"),
725 # "ip-address", "mac-address" # filled by LCM
726 # vim-id # TODO it would be nice having a vim port id
728 vnfr_descriptor
["connection-point"].append(vnf_cp
)
729 for vdu
in vnfd
["vdu"]:
730 vdur_id
= str(uuid4())
733 "vdu-id-ref": vdu
["id"],
734 # TODO "name": "" Name of the VDU in the VIM
735 "ip-address": None, # mgmt-interface filled by LCM
736 # "vim-id", "flavor-id", "image-id", "management-ip" # filled by LCM
737 "internal-connection-point": [],
740 # TODO volumes: name, volume-id
741 for icp
in vdu
.get("internal-connection-point", ()):
744 "connection-point-id": icp
["id"],
745 "name": icp
.get("name"),
746 # "ip-address", "mac-address" # filled by LCM
747 # vim-id # TODO it would be nice having a vim port id
749 vdur
["internal-connection-point"].append(vdu_icp
)
750 for iface
in vdu
.get("interface", ()):
752 "name": iface
.get("name"),
753 # "ip-address", "mac-address" # filled by LCM
754 # vim-id # TODO it would be nice having a vim port id
756 vdur
["interfaces"].append(vdu_iface
)
757 vnfr_descriptor
["vdur"].append(vdur
)
759 step
= "creating vnfr vnfd-id='{}' constituent-vnfd='{}' at database".format(
760 member_vnf
["vnfd-id-ref"], member_vnf
["member-vnf-index"])
761 self
._format
_new
_data
(session
, "vnfrs", vnfr_descriptor
)
762 self
.db
.create("vnfrs", vnfr_descriptor
)
763 rollback
.insert(0, {"item": "vnfrs", "_id": vnfr_id
})
764 nsr_descriptor
["constituent-vnfr-ref"].append(vnfr_id
)
766 step
= "creating nsr at database"
767 self
._format
_new
_data
(session
, "nsrs", nsr_descriptor
)
768 self
.db
.create("nsrs", nsr_descriptor
)
769 rollback
.insert(rollback_index
, {"item": "nsrs", "_id": nsr_id
})
771 except Exception as e
:
772 raise EngineException("Error {}: {}".format(step
, e
))
775 def _update_descriptor(desc
, kwargs
):
777 Update descriptor with the kwargs. It contains dot separated keys
778 :param desc: dictionary to be updated
779 :param kwargs: plain dictionary to be used for updating.
785 for k
, v
in kwargs
.items():
786 update_content
= desc
790 if kitem_old
is not None:
791 update_content
= update_content
[kitem_old
]
792 if isinstance(update_content
, dict):
794 elif isinstance(update_content
, list):
795 kitem_old
= int(kitem
)
797 raise EngineException(
798 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k
, kitem
))
799 update_content
[kitem_old
] = v
801 raise EngineException(
802 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k
, kitem_old
))
804 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
807 raise EngineException(
808 "Invalid query string '{}'. Index '{}' out of range".format(k
, kitem_old
))
810 def new_item(self
, rollback
, session
, item
, indata
={}, kwargs
=None, headers
={}, force
=False):
812 Creates a new entry into database. For nsds and vnfds it creates an almost empty DISABLED entry,
813 that must be completed with a call to method upload_content
814 :param rollback: list where this method appends created items at database in case a rollback may to be done
815 :param session: contains the used login username and working project
816 :param item: it can be: users, projects, vim_accounts, sdns, nsrs, nsds, vnfds
817 :param indata: data to be inserted
818 :param kwargs: used to override the indata descriptor
819 :param headers: http request headers
820 :param force: If True avoid some dependence checks
821 :return: _id: identity of the inserted data.
826 if item
in ("nsds", "vnfds"):
827 item_envelop
= "userDefinedData"
828 content
= self
._remove
_envelop
(item_envelop
, indata
)
830 # Override descriptor with query string kwargs
831 self
._update
_descriptor
(content
, kwargs
)
832 if not indata
and item
not in ("nsds", "vnfds"):
833 raise EngineException("Empty payload")
835 validate_input(content
, item
, new
=True)
838 # in this case the input descriptor is not the data to be stored
839 return self
.new_nsr(rollback
, session
, ns_request
=content
)
841 self
._validate
_new
_data
(session
, item_envelop
, content
, force
)
842 if item
in ("nsds", "vnfds"):
843 content
= {"_admin": {"userDefinedData": content
}}
844 self
._format
_new
_data
(session
, item
, content
)
845 _id
= self
.db
.create(item
, content
)
846 rollback
.insert(0, {"item": item
, "_id": _id
})
848 if item
== "vim_accounts":
849 msg_data
= self
.db
.get_one(item
, {"_id": _id
})
850 msg_data
.pop("_admin", None)
851 self
.msg
.write("vim_account", "create", msg_data
)
853 msg_data
= self
.db
.get_one(item
, {"_id": _id
})
854 msg_data
.pop("_admin", None)
855 self
.msg
.write("sdn", "create", msg_data
)
857 except ValidationError
as e
:
858 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
860 def new_nslcmop(self
, session
, nsInstanceId
, operation
, params
):
866 "operationState": "PROCESSING", # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
867 "statusEnteredTime": now
,
868 "nsInstanceId": nsInstanceId
,
869 "lcmOperationType": operation
,
871 "isAutomaticInvocation": False,
872 "operationParams": params
,
873 "isCancelPending": False,
875 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
876 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsInstanceId
,
881 def ns_operation(self
, rollback
, session
, nsInstanceId
, operation
, indata
, kwargs
=None):
883 Performs a new operation over a ns
884 :param rollback: list where this method appends created items at database in case a rollback may to be done
885 :param session: contains the used login username and working project
886 :param nsInstanceId: _id of the nsr to perform the operation
887 :param operation: it can be: instantiate, terminate, action, TODO: update, heal
888 :param indata: descriptor with the parameters of the operation
889 :param kwargs: used to override the indata descriptor
890 :return: id of the nslcmops
893 # Override descriptor with query string kwargs
894 self
._update
_descriptor
(indata
, kwargs
)
895 validate_input(indata
, "ns_" + operation
, new
=True)
897 nsr
= self
.get_item(session
, "nsrs", nsInstanceId
)
898 if not nsr
["_admin"].get("nsState") or nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
899 if operation
== "terminate" and indata
.get("autoremove"):
900 # NSR must be deleted
901 return self
.del_item(session
, "nsrs", nsInstanceId
)
902 if operation
!= "instantiate":
903 raise EngineException("ns_instance '{}' cannot be '{}' because it is not instantiated".format(
904 nsInstanceId
, operation
), HTTPStatus
.CONFLICT
)
906 if operation
== "instantiate" and not indata
.get("force"):
907 raise EngineException("ns_instance '{}' cannot be '{}' because it is already instantiated".format(
908 nsInstanceId
, operation
), HTTPStatus
.CONFLICT
)
909 indata
["nsInstanceId"] = nsInstanceId
910 self
._check
_ns
_operation
(session
, nsr
, operation
, indata
)
911 nslcmop
= self
.new_nslcmop(session
, nsInstanceId
, operation
, indata
)
912 self
._format
_new
_data
(session
, "nslcmops", nslcmop
)
913 _id
= self
.db
.create("nslcmops", nslcmop
)
914 rollback
.insert(0, {"item": "nslcmops", "_id": _id
})
916 self
.msg
.write("ns", operation
, nslcmop
)
918 except ValidationError
as e
:
919 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
920 # except DbException as e:
921 # raise EngineException("Cannot get ns_instance '{}': {}".format(e), HTTPStatus.NOT_FOUND)
923 def _add_read_filter(self
, session
, item
, filter):
924 if session
["project_id"] == "admin": # allows all
927 filter["username"] = session
["username"]
928 elif item
in ("vnfds", "nsds", "nsrs", "vnfrs"):
929 filter["_admin.projects_read.cont"] = ["ANY", session
["project_id"]]
931 def _add_delete_filter(self
, session
, item
, filter):
932 if session
["project_id"] != "admin" and item
in ("users", "projects"):
933 raise EngineException("Only admin users can perform this task", http_code
=HTTPStatus
.FORBIDDEN
)
935 if filter.get("_id") == session
["username"] or filter.get("username") == session
["username"]:
936 raise EngineException("You cannot delete your own user", http_code
=HTTPStatus
.CONFLICT
)
937 elif item
== "project":
938 if filter.get("_id") == session
["project_id"]:
939 raise EngineException("You cannot delete your own project", http_code
=HTTPStatus
.CONFLICT
)
940 elif item
in ("vnfds", "nsds") and session
["project_id"] != "admin":
941 filter["_admin.projects_write.cont"] = ["ANY", session
["project_id"]]
943 def get_file(self
, session
, item
, _id
, path
=None, accept_header
=None):
945 Return the file content of a vnfd or nsd
946 :param session: contains the used login username and working project
947 :param item: it can be vnfds or nsds
948 :param _id: Identity of the vnfd, ndsd
949 :param path: artifact path or "$DESCRIPTOR" or None
950 :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
951 :return: opened file or raises an exception
953 accept_text
= accept_zip
= False
955 if 'text/plain' in accept_header
or '*/*' in accept_header
:
957 if 'application/zip' in accept_header
or '*/*' in accept_header
:
959 if not accept_text
and not accept_zip
:
960 raise EngineException("provide request header 'Accept' with 'application/zip' or 'text/plain'",
961 http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
963 content
= self
.get_item(session
, item
, _id
)
964 if content
["_admin"]["onboardingState"] != "ONBOARDED":
965 raise EngineException("Cannot get content because this resource is not at 'ONBOARDED' state. "
966 "onboardingState is {}".format(content
["_admin"]["onboardingState"]),
967 http_code
=HTTPStatus
.CONFLICT
)
968 storage
= content
["_admin"]["storage"]
969 if path
is not None and path
!= "$DESCRIPTOR": # artifacts
970 if not storage
.get('pkg-dir'):
971 raise EngineException("Packages does not contains artifacts", http_code
=HTTPStatus
.BAD_REQUEST
)
972 if self
.fs
.file_exists((storage
['folder'], storage
['pkg-dir'], *path
), 'dir'):
973 folder_content
= self
.fs
.dir_ls((storage
['folder'], storage
['pkg-dir'], *path
))
974 return folder_content
, "text/plain"
975 # TODO manage folders in http
977 return self
.fs
.file_open((storage
['folder'], storage
['pkg-dir'], *path
), "rb"),\
978 "application/octet-stream"
980 # pkgtype accept ZIP TEXT -> result
981 # manyfiles yes X -> zip
983 # onefile yes no -> zip
986 if accept_text
and (not storage
.get('pkg-dir') or path
== "$DESCRIPTOR"):
987 return self
.fs
.file_open((storage
['folder'], storage
['descriptor']), "r"), "text/plain"
988 elif storage
.get('pkg-dir') and not accept_zip
:
989 raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'"
990 "Accept header", http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
992 if not storage
.get('zipfile'):
993 # TODO generate zipfile if not present
994 raise EngineException("Only allowed 'text/plain' Accept header for this descriptor. To be solved in "
995 "future versions", http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
996 return self
.fs
.file_open((storage
['folder'], storage
['zipfile']), "rb"), "application/zip"
998 def get_item_list(self
, session
, item
, filter={}):
1001 :param session: contains the used login username and working project
1002 :param item: it can be: users, projects, vnfds, nsds, ...
1003 :param filter: filter of data to be applied
1004 :return: The list, it can be empty if no one match the filter.
1006 # TODO add admin to filter, validate rights
1007 # TODO transform data for SOL005 URL requests. Transform filtering
1008 # TODO implement "field-type" query string SOL005
1010 self
._add
_read
_filter
(session
, item
, filter)
1011 return self
.db
.get_list(item
, filter)
1013 def get_item(self
, session
, item
, _id
):
1015 Get complete information on an items
1016 :param session: contains the used login username and working project
1017 :param item: it can be: users, projects, vnfds, nsds,
1018 :param _id: server id of the item
1019 :return: dictionary, raise exception if not found.
1021 filter = {"_id": _id
}
1022 # TODO add admin to filter, validate rights
1023 # TODO transform data for SOL005 URL requests
1024 self
._add
_read
_filter
(session
, item
, filter)
1025 return self
.db
.get_one(item
, filter)
1027 def del_item_list(self
, session
, item
, filter={}):
1029 Delete a list of items
1030 :param session: contains the used login username and working project
1031 :param item: it can be: users, projects, vnfds, nsds, ...
1032 :param filter: filter of data to be applied
1033 :return: The deleted list, it can be empty if no one match the filter.
1035 # TODO add admin to filter, validate rights
1036 self
._add
_read
_filter
(session
, item
, filter)
1037 return self
.db
.del_list(item
, filter)
1039 def del_item(self
, session
, item
, _id
, force
=False):
1041 Delete item by its internal id
1042 :param session: contains the used login username and working project
1043 :param item: it can be: users, projects, vnfds, nsds, ...
1044 :param _id: server id of the item
1045 :param force: indicates if deletion must be forced in case of conflict
1046 :return: dictionary with deleted item _id. It raises exception if not found.
1048 # TODO add admin to filter, validate rights
1049 # data = self.get_item(item, _id)
1050 filter = {"_id": _id
}
1051 self
._add
_delete
_filter
(session
, item
, filter)
1052 if item
in ("vnfds", "nsds") and not force
:
1053 descriptor
= self
.get_item(session
, item
, _id
)
1054 descriptor_id
= descriptor
.get("id")
1056 self
._check
_dependencies
_on
_descriptor
(session
, item
, descriptor_id
, _id
)
1059 nsr
= self
.db
.get_one(item
, filter)
1060 if nsr
["_admin"].get("nsState") == "INSTANTIATED" and not force
:
1061 raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. "
1062 "Launch 'terminate' operation first; or force deletion".format(_id
),
1063 http_code
=HTTPStatus
.CONFLICT
)
1064 v
= self
.db
.del_one(item
, {"_id": _id
})
1065 self
.db
.del_list("nslcmops", {"nsInstanceId": _id
})
1066 self
.db
.del_list("vnfrs", {"nsr-id-ref": _id
})
1067 self
.msg
.write("ns", "deleted", {"_id": _id
})
1069 if item
in ("vim_accounts", "sdns") and not force
:
1070 self
.db
.set_one(item
, {"_id": _id
}, {"_admin.to_delete": True}) # TODO change status
1071 if item
== "vim_accounts":
1072 self
.msg
.write("vim_account", "delete", {"_id": _id
})
1073 elif item
== "sdns":
1074 self
.msg
.write("sdn", "delete", {"_id": _id
})
1075 return {"deleted": 1} # TODO indicate an offline operation to return 202 ACCEPTED
1077 v
= self
.db
.del_one(item
, filter)
1078 if item
in ("vnfds", "nsds"):
1079 self
.fs
.file_delete(_id
, ignore_non_exist
=True)
1080 if item
in ("vim_accounts", "sdns", "vnfds", "nsds"):
1081 self
.msg
.write(item
[:-1], "deleted", {"_id": _id
})
1086 Prune database not needed content
1089 return self
.db
.del_list("nsrs", {"_admin.to_delete": True})
1091 def create_admin(self
):
1093 Creates a new user admin/admin into database if database is empty. Useful for initialization
1094 :return: _id identity of the inserted data, or None
1096 users
= self
.db
.get_one("users", fail_on_empty
=False, fail_on_more
=False)
1099 # raise EngineException("Unauthorized. Database users is not empty", HTTPStatus.UNAUTHORIZED)
1100 indata
= {"username": "admin", "password": "admin", "projects": ["admin"]}
1101 fake_session
= {"project_id": "admin", "username": "admin"}
1102 self
._format
_new
_data
(fake_session
, "users", indata
)
1103 _id
= self
.db
.create("users", indata
)
1106 def init_db(self
, target_version
='1.0'):
1108 Init database if empty. If not empty it checks that database version is ok.
1109 If empty, it creates a new user admin/admin at 'users' and a new entry at 'version'
1110 :return: None if ok, exception if error or if the version is different.
1112 version
= self
.db
.get_one("version", fail_on_empty
=False, fail_on_more
=False)
1116 # create database version
1118 "_id": '1.0', # version text
1119 "version": 1000, # version number
1120 "date": "2018-04-12", # version date
1121 "description": "initial design", # changes in this version
1122 'status': 'ENABLED' # ENABLED, DISABLED (migration in process), ERROR,
1124 self
.db
.create("version", version_data
)
1125 elif version
["_id"] != target_version
:
1126 # TODO implement migration process
1127 raise EngineException("Wrong database version '{}'. Expected '{}'".format(
1128 version
["_id"], target_version
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
1129 elif version
["status"] != 'ENABLED':
1130 raise EngineException("Wrong database status '{}'".format(
1131 version
["status"]), HTTPStatus
.INTERNAL_SERVER_ERROR
)
1134 def _edit_item(self
, session
, item
, id, content
, indata
={}, kwargs
=None, force
=False):
1136 indata
= self
._remove
_envelop
(item
, indata
)
1138 # Override descriptor with query string kwargs
1141 for k
, v
in kwargs
.items():
1142 update_content
= indata
1144 klist
= k
.split(".")
1146 if kitem_old
is not None:
1147 update_content
= update_content
[kitem_old
]
1148 if isinstance(update_content
, dict):
1150 elif isinstance(update_content
, list):
1151 kitem_old
= int(kitem
)
1153 raise EngineException(
1154 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k
, kitem
))
1155 update_content
[kitem_old
] = v
1157 raise EngineException(
1158 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k
, kitem_old
))
1160 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
1163 raise EngineException(
1164 "Invalid query string '{}'. Index '{}' out of range".format(k
, kitem_old
))
1166 validate_input(indata
, item
, new
=False)
1167 except ValidationError
as e
:
1168 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
1170 _deep_update(content
, indata
)
1171 self
._validate
_new
_data
(session
, item
, content
, id, force
)
1172 # self._format_new_data(session, item, content)
1173 self
.db
.replace(item
, id, content
)
1174 if item
in ("vim_accounts", "sdns"):
1175 indata
.pop("_admin", None)
1177 if item
== "vim_accounts":
1178 self
.msg
.write("vim_account", "edit", indata
)
1179 elif item
== "sdns":
1180 self
.msg
.write("sdn", "edit", indata
)
1183 def edit_item(self
, session
, item
, _id
, indata
={}, kwargs
=None, force
=False):
1185 Update an existing entry at database
1186 :param session: contains the used login username and working project
1187 :param item: it can be: users, projects, vnfds, nsds, ...
1188 :param _id: identifier to be updated
1189 :param indata: data to be inserted
1190 :param kwargs: used to override the indata descriptor
1191 :param force: If True avoid some dependence checks
1192 :return: dictionary, raise exception if not found.
1195 content
= self
.get_item(session
, item
, _id
)
1196 return self
._edit
_item
(session
, item
, _id
, content
, indata
, kwargs
, force
)