1 # -*- coding: utf-8 -*-
12 from random
import choice
as random_choice
13 from uuid
import uuid4
14 from hashlib
import sha256
, md5
15 from dbbase
import DbException
16 from fsbase
import FsException
17 from msgbase
import MsgException
18 from http
import HTTPStatus
20 from copy
import deepcopy
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
41 for k
in dict_reference
:
42 if dict_reference
[k
] is None: # None->Anything
43 if k
in dict_to_change
:
45 elif not isinstance(dict_reference
[k
], dict): # NotDict->Anything
46 dict_to_change
[k
] = dict_reference
[k
]
47 elif k
not in dict_to_change
: # Dict->Empty
48 dict_to_change
[k
] = deepcopy(dict_reference
[k
])
49 _deep_update(dict_to_change
[k
], dict_reference
[k
])
50 elif isinstance(dict_to_change
[k
], dict): # Dict->Dict
51 _deep_update(dict_to_change
[k
], dict_reference
[k
])
53 dict_to_change
[k
] = deepcopy(dict_reference
[k
])
54 _deep_update(dict_to_change
[k
], dict_reference
[k
])
64 self
.logger
= logging
.getLogger("nbi.engine")
66 def start(self
, config
):
68 Connect to database, filesystem storage, and messaging
69 :param config: two level dictionary with configuration. Top level should contain 'database', 'storage',
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"])
82 raise EngineException("Invalid configuration param '{}' at '[database]':'driver'".format(
83 config
["database"]["driver"]))
85 if config
["storage"]["driver"] == "local":
86 self
.fs
= fslocal
.FsLocal()
87 self
.fs
.fs_connect(config
["storage"])
89 raise EngineException("Invalid configuration param '{}' at '[storage]':'driver'".format(
90 config
["storage"]["driver"]))
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"])
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
)
107 self
.db
.db_disconnect()
109 self
.fs
.fs_disconnect()
111 self
.fs
.fs_disconnect()
112 except (DbException
, FsException
, MsgException
) as e
:
113 raise EngineException(str(e
), http_code
=e
.http_code
)
115 def authorize(self
, 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
]
125 if session
["expires"] < now
:
126 del self
.tokens
[token
]
127 raise EngineException("Expired Token or Authorization http header",
128 http_code
=HTTPStatus
.UNAUTHORIZED
)
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"]}
138 def new_token(self
, session
, indata
, remote
):
142 # Try using username/password
143 if indata
.get("username"):
144 user_rows
= self
.db
.get_list("users", {"username": indata
.get("username")})
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"]:
153 raise EngineException("Invalid username/password", http_code
=HTTPStatus
.UNAUTHORIZED
)
155 user_rows
= self
.db
.get_list("users", {"username": session
["username"]})
157 user_content
= user_rows
[0]
159 raise EngineException("Invalid token", http_code
=HTTPStatus
.UNAUTHORIZED
)
161 raise EngineException("Provide credentials: username/password or Authorization Bearer token",
162 http_code
=HTTPStatus
.UNAUTHORIZED
)
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
)
172 project_id
= user_content
["projects"][0]
173 if project_id
== "admin":
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
}
182 new_session
["remote_host"] = remote
.name
184 new_session
["remote_host"] = remote
.ip
186 self
.tokens
[token_id
] = new_session
187 return deepcopy(new_session
)
189 def get_token_list(self
, session
):
191 for token_id
, token_value
in self
.tokens
.items():
192 if token_value
["username"] == session
["username"]:
193 token_list
.append(deepcopy(token_value
))
196 def get_token(self
, session
, token_id
):
197 token_value
= self
.tokens
.get(token_id
)
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
)
204 def del_token(self
, token_id
):
206 del self
.tokens
[token_id
]
207 return "token '{}' deleted".format(token_id
)
209 raise EngineException("Token '{}' not found".format(token_id
), http_code
=HTTPStatus
.NOT_FOUND
)
212 def _remove_envelop(item
, indata
=None):
214 Obtain the useful data removing the envelop. It goes throw the vnfd or nsd catalog and returns the
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
220 clean_indata
= indata
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]
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']
246 def _validate_new_data(self
, session
, item
, indata
, id=None):
248 if not indata
.get("username"):
249 raise EngineException("missing 'username'", HTTPStatus
.UNPROCESSABLE_ENTITY
)
250 if not indata
.get("password"):
251 raise EngineException("missing 'password'", HTTPStatus
.UNPROCESSABLE_ENTITY
)
252 if not indata
.get("projects"):
253 raise EngineException("missing 'projects'", HTTPStatus
.UNPROCESSABLE_ENTITY
)
254 # check username not exist
255 if self
.db
.get_one(item
, {"username": indata
.get("username")}, fail_on_empty
=False, fail_on_more
=False):
256 raise EngineException("username '{}' exist".format(indata
["username"]), HTTPStatus
.CONFLICT
)
257 elif item
== "projects":
258 if not indata
.get("name"):
259 raise EngineException("missing 'name'")
260 # check name not exist
261 if self
.db
.get_one(item
, {"name": indata
.get("name")}, fail_on_empty
=False, fail_on_more
=False):
262 raise EngineException("name '{}' exist".format(indata
["name"]), HTTPStatus
.CONFLICT
)
263 elif item
in ("vnfds", "nsds"):
264 filter = {"id": indata
["id"]}
266 filter["_id.neq"] = id
267 # TODO add admin to filter, validate rights
268 self
._add
_read
_filter
(session
, item
, filter)
269 if self
.db
.get_one(item
, filter, fail_on_empty
=False):
270 raise EngineException("{} with id '{}' already exist for this tenant".format(item
[:-1], indata
["id"]),
273 # TODO validate with pyangbind
274 elif item
== "userDefinedData":
275 # TODO validate userDefinedData is a keypair values
280 elif item
== "vim_accounts" or item
== "sdns":
281 if self
.db
.get_one(item
, {"name": indata
.get("name")}, fail_on_empty
=False, fail_on_more
=False):
282 raise EngineException("name '{}' already exist for {}".format(indata
["name"], item
),
285 def _format_new_data(self
, session
, item
, indata
):
287 if not "_admin" in indata
:
288 indata
["_admin"] = {}
289 indata
["_admin"]["created"] = now
290 indata
["_admin"]["modified"] = now
292 indata
["_id"] = indata
["username"]
294 indata
["_admin"]["salt"] = salt
295 indata
["password"] = sha256(indata
["password"].encode('utf-8') + salt
.encode('utf-8')).hexdigest()
296 elif item
== "projects":
297 indata
["_id"] = indata
["name"]
299 if not indata
.get("_id"):
300 indata
["_id"] = str(uuid4())
301 if item
in ("vnfds", "nsds", "nsrs"):
302 if not indata
["_admin"].get("projects_read"):
303 indata
["_admin"]["projects_read"] = [session
["project_id"]]
304 if not indata
["_admin"].get("projects_write"):
305 indata
["_admin"]["projects_write"] = [session
["project_id"]]
306 if item
in ("vnfds", "nsds"):
307 indata
["_admin"]["onboardingState"] = "CREATED"
308 indata
["_admin"]["operationalState"] = "DISABLED"
309 indata
["_admin"]["usageSate"] = "NOT_IN_USE"
311 indata
["_admin"]["nsState"] = "NOT_INSTANTIATED"
312 if item
in ("vim_accounts", "sdns"):
313 indata
["_admin"]["operationalState"] = "PROCESSING"
315 def upload_content(self
, session
, item
, _id
, indata
, kwargs
, headers
):
317 Used for receiving content by chunks (with a transaction_id header and/or gzip file. It will store and extract)
318 :param session: session
319 :param item: can be nsds or vnfds
320 :param _id : the nsd,vnfd is already created, this is the id
321 :param indata: http body request
322 :param kwargs: user query string to override parameters. NOT USED
323 :param headers: http request headers
324 :return: True package has is completely uploaded or False if partial content has been uplodaed.
325 Raise exception on error
327 # Check that _id exist and it is valid
328 current_desc
= self
.get_item(session
, item
, _id
)
330 content_range_text
= headers
.get("Content-Range")
331 expected_md5
= headers
.get("Content-File-MD5")
333 content_type
= headers
.get("Content-Type")
334 if content_type
and "application/gzip" in content_type
or "application/x-gzip" in content_type
or \
335 "application/zip" in content_type
:
337 filename
= headers
.get("Content-Filename")
339 filename
= "package.tar.gz" if compressed
else "package"
340 # TODO change to Content-Disposition filename https://tools.ietf.org/html/rfc6266
344 if content_range_text
:
345 content_range
= content_range_text
.replace("-", " ").replace("/", " ").split()
346 if content_range
[0] != "bytes": # TODO check x<y not negative < total....
348 start
= int(content_range
[1])
349 end
= int(content_range
[2]) + 1
350 total
= int(content_range
[3])
355 if not self
.fs
.file_exists(_id
, 'dir'):
356 raise EngineException("invalid Transaction-Id header", HTTPStatus
.NOT_FOUND
)
358 self
.fs
.file_delete(_id
, ignore_non_exist
=True)
361 storage
= self
.fs
.get_params()
362 storage
["folder"] = _id
364 file_path
= (_id
, filename
)
365 if self
.fs
.file_exists(file_path
, 'file'):
366 file_size
= self
.fs
.file_size(file_path
)
369 if file_size
!= start
:
370 raise EngineException("invalid Content-Range start sequence, expected '{}' but received '{}'".format(
371 file_size
, start
), HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
372 file_pkg
= self
.fs
.file_open(file_path
, 'a+b')
373 if isinstance(indata
, dict):
374 indata_text
= yaml
.safe_dump(indata
, indent
=4, default_flow_style
=False)
375 file_pkg
.write(indata_text
.encode(encoding
="utf-8"))
379 indata_text
= indata
.read(4096)
380 indata_len
+= len(indata_text
)
383 file_pkg
.write(indata_text
)
384 if content_range_text
:
385 if indata_len
!= end
-start
:
386 raise EngineException("Mismatch between Content-Range header {}-{} and body length of {}".format(
387 start
, end
-1, indata_len
), HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
389 # TODO update to UPLOADING
396 chunk_data
= file_pkg
.read(1024)
398 file_md5
.update(chunk_data
)
399 chunk_data
= file_pkg
.read(1024)
400 if expected_md5
!= file_md5
.hexdigest():
401 raise EngineException("Error, MD5 mismatch", HTTPStatus
.CONFLICT
)
403 if compressed
== "gzip":
404 tar
= tarfile
.open(mode
='r', fileobj
=file_pkg
)
405 descriptor_file_name
= None
407 tarname
= tarinfo
.name
408 tarname_path
= tarname
.split("/")
409 if not tarname_path
[0] or ".." in tarname_path
: # if start with "/" means absolute path
410 raise EngineException("Absolute path or '..' are not allowed for package descriptor tar.gz")
411 if len(tarname_path
) == 1 and not tarinfo
.isdir():
412 raise EngineException("All files must be inside a dir for package descriptor tar.gz")
413 if tarname
.endswith(".yaml") or tarname
.endswith(".json") or tarname
.endswith(".yml"):
414 storage
["pkg-dir"] = tarname_path
[0]
415 if len(tarname_path
) == 2:
416 if descriptor_file_name
:
417 raise EngineException("Found more than one descriptor file at package descriptor tar.gz")
418 descriptor_file_name
= tarname
419 if not descriptor_file_name
:
420 raise EngineException("Not found any descriptor file at package descriptor tar.gz")
421 storage
["descriptor"] = descriptor_file_name
422 storage
["zipfile"] = filename
423 self
.fs
.file_extract(tar
, _id
)
424 with self
.fs
.file_open((_id
, descriptor_file_name
), "r") as descriptor_file
:
425 content
= descriptor_file
.read()
427 content
= file_pkg
.read()
428 storage
["descriptor"] = descriptor_file_name
= filename
430 if descriptor_file_name
.endswith(".json"):
431 error_text
= "Invalid json format "
432 indata
= json
.load(content
)
434 error_text
= "Invalid yaml format "
435 indata
= yaml
.load(content
)
437 current_desc
["_admin"]["storage"] = storage
438 current_desc
["_admin"]["onboardingState"] = "ONBOARDED"
439 current_desc
["_admin"]["operationalState"] = "ENABLED"
441 self
._edit
_item
(session
, item
, _id
, current_desc
, indata
, kwargs
)
442 # TODO if descriptor has changed because kwargs update content and remove cached zip
443 # TODO if zip is not present creates one
446 except EngineException
:
449 raise EngineException("invalid Content-Range header format. Expected 'bytes start-end/total'",
450 HTTPStatus
.REQUESTED_RANGE_NOT_SATISFIABLE
)
452 raise EngineException("invalid upload transaction sequence: '{}'".format(e
), HTTPStatus
.BAD_REQUEST
)
453 except tarfile
.ReadError
as e
:
454 raise EngineException("invalid file content {}".format(e
), HTTPStatus
.BAD_REQUEST
)
455 except (ValueError, yaml
.YAMLError
) as e
:
456 raise EngineException(error_text
+ str(e
))
461 def new_nsr(self
, session
, ns_request
):
463 Creates a new nsr into database
464 :param session: contains the used login username and working project
465 :param ns_request: params to be used for the nsr
466 :return: nsr descriptor to be stored at database and the _id
470 nsd
= self
.get_item(session
, "nsds", ns_request
["nsdId"])
473 "name": ns_request
["nsName"],
474 "name-ref": ns_request
["nsName"],
475 "short-name": ns_request
["nsName"],
476 "admin-status": "ENABLED",
478 "datacenter": ns_request
["vimAccountId"],
479 "resource-orchestrator": "osmopenmano",
480 "description": ns_request
.get("nsDescription", ""),
481 "constituent-vnfr-ref": ["TODO datacenter-id, vnfr-id"],
483 "operational-status": "init", # typedef ns-operational-
484 "config-status": "init", # typedef config-states
485 "detailed-status": "scheduled",
487 "orchestration-progress": {}, # {"networks": {"active": 0, "total": 0}, "vms": {"active": 0, "total": 0}},
489 "crete-time": time(),
490 "nsd-name-ref": nsd
["name"],
491 "operational-events": [], # "id", "timestamp", "description", "event",
492 "nsd-ref": nsd
["id"],
493 "instantiate_params": ns_request
,
494 "ns-instance-config-ref": _id
,
498 # "input-parameter": xpath, value,
499 "ssh-authorized-key": ns_request
.get("key-pair-ref"),
501 ns_request
["nsr_id"] = _id
502 return nsr_descriptor
505 def _update_descriptor(desc
, kwargs
):
507 Update descriptor with the kwargs
514 for k
, v
in kwargs
.items():
515 update_content
= content
519 if kitem_old
is not None:
520 update_content
= update_content
[kitem_old
]
521 if isinstance(update_content
, dict):
523 elif isinstance(update_content
, list):
524 kitem_old
= int(kitem
)
526 raise EngineException(
527 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k
, kitem
))
528 update_content
[kitem_old
] = v
530 raise EngineException(
531 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k
, kitem_old
))
533 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
536 raise EngineException(
537 "Invalid query string '{}'. Index '{}' out of range".format(k
, kitem_old
))
539 def new_item(self
, session
, item
, indata
={}, kwargs
=None, headers
={}):
541 Creates a new entry into database. For nsds and vnfds it creates an almost empty DISABLED entry,
542 that must be completed with a call to method upload_content
543 :param session: contains the used login username and working project
544 :param item: it can be: users, projects, vim_accounts, sdns, nsrs, nsds, vnfds
545 :param indata: data to be inserted
546 :param kwargs: used to override the indata descriptor
547 :param headers: http request headers
548 :return: _id, transaction_id: identity of the inserted data. or transaction_id if Content-Range is used
553 if item
in ("nsds", "vnfds"):
554 item_envelop
= "userDefinedData"
555 content
= self
._remove
_envelop
(item_envelop
, indata
)
557 # Override descriptor with query string kwargs
558 self
._update
_descriptor
(content
, kwargs
)
559 if not indata
and item
not in ("nsds", "vnfds"):
560 raise EngineException("Empty payload")
562 validate_input(content
, item
, new
=True)
565 # in this case the imput descriptor is not the data to be stored
567 content
= self
.new_nsr(session
, ns_request
)
569 self
._validate
_new
_data
(session
, item_envelop
, content
)
570 if item
in ("nsds", "vnfds"):
571 content
= {"_admin": {"userDefinedData": content
}}
572 self
._format
_new
_data
(session
, item
, content
)
573 _id
= self
.db
.create(item
, content
)
576 # self.msg.write("ns", "created", _id) # sending just for information.
577 elif item
== "vim_accounts":
578 msg_data
= self
.db
.get_one(item
, {"_id": _id
})
579 msg_data
.pop("_admin", None)
580 self
.msg
.write("vim_account", "create", msg_data
)
582 msg_data
= self
.db
.get_one(item
, {"_id": _id
})
583 msg_data
.pop("_admin", None)
584 self
.msg
.write("sdn", "create", msg_data
)
586 except ValidationError
as e
:
587 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
589 def new_nslcmop(self
, session
, nsInstanceId
, action
, params
):
595 "operationState": "PROCESSING", # COMPLETED,PARTIALLY_COMPLETED,FAILED_TEMP,FAILED,ROLLING_BACK,ROLLED_BACK
596 "statusEnteredTime": now
,
597 "nsInstanceId": nsInstanceId
,
598 "lcmOperationType": action
,
600 "isAutomaticInvocation": False,
601 "operationParams": params
,
602 "isCancelPending": False,
604 "self": "/osm/nslcm/v1/ns_lcm_op_occs/" + _id
,
605 "nsInstance": "/osm/nslcm/v1/ns_instances/" + nsInstanceId
,
610 def ns_action(self
, session
, nsInstanceId
, action
, indata
, kwargs
=None):
612 Performs a new action over a ns
613 :param session: contains the used login username and working project
614 :param nsInstanceId: _id of the nsr to perform the action
615 :param action: it can be: instantiate, terminate, action, TODO: update, heal
616 :param indata: descriptor with the parameters of the action
617 :param kwargs: used to override the indata descriptor
618 :return: id of the nslcmops
621 # Override descriptor with query string kwargs
622 self
._update
_descriptor
(indata
, kwargs
)
623 validate_input(indata
, "ns_" + action
, new
=True)
625 nsr
= self
.get_item(session
, "nsrs", nsInstanceId
)
626 if not nsr
["_admin"].get("nsState") or nsr
["_admin"]["nsState"] == "NOT_INSTANTIATED":
627 if action
== "terminate" and indata
.get("autoremove"):
628 # NSR must be deleted
629 return self
.del_item(session
, "nsrs", nsInstanceId
)
630 if action
!= "instantiate":
631 raise EngineException("ns_instance '{}' cannot be '{}' because it is not instantiated".format(
632 nsInstanceId
, action
), HTTPStatus
.CONFLICT
)
634 if action
== "instantiate" and not indata
.get("force"):
635 raise EngineException("ns_instance '{}' cannot be '{}' because it is already instantiated".format(
636 nsInstanceId
, action
), HTTPStatus
.CONFLICT
)
637 indata
["nsInstanceId"] = nsInstanceId
639 nslcmop
= self
.new_nslcmop(session
, nsInstanceId
, action
, indata
)
640 self
._format
_new
_data
(session
, "nslcmops", nslcmop
)
641 _id
= self
.db
.create("nslcmops", nslcmop
)
643 self
.msg
.write("ns", action
, nslcmop
)
645 except ValidationError
as e
:
646 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
647 # except DbException as e:
648 # raise EngineException("Cannot get ns_instance '{}': {}".format(e), HTTPStatus.NOT_FOUND)
650 def _add_read_filter(self
, session
, item
, filter):
651 if session
["project_id"] == "admin": # allows all
654 filter["username"] = session
["username"]
655 elif item
in ("vnfds", "nsds", "nsrs"):
656 filter["_admin.projects_read.cont"] = ["ANY", session
["project_id"]]
658 def _add_delete_filter(self
, session
, item
, filter):
659 if session
["project_id"] != "admin" and item
in ("users", "projects"):
660 raise EngineException("Only admin users can perform this task", http_code
=HTTPStatus
.FORBIDDEN
)
662 if filter.get("_id") == session
["username"] or filter.get("username") == session
["username"]:
663 raise EngineException("You cannot delete your own user", http_code
=HTTPStatus
.CONFLICT
)
664 elif item
== "project":
665 if filter.get("_id") == session
["project_id"]:
666 raise EngineException("You cannot delete your own project", http_code
=HTTPStatus
.CONFLICT
)
667 elif item
in ("vnfds", "nsds") and session
["project_id"] != "admin":
668 filter["_admin.projects_write.cont"] = ["ANY", session
["project_id"]]
670 def get_file(self
, session
, item
, _id
, path
=None, accept_header
=None):
672 Return the file content of a vnfd or nsd
673 :param session: contains the used login username and working project
674 :param item: it can be vnfds or nsds
675 :param _id: Identity of the vnfd, ndsd
676 :param path: artifact path or "$DESCRIPTOR" or None
677 :param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain
678 :return: opened file or raises an exception
680 accept_text
= accept_zip
= False
682 if 'text/plain' in accept_header
or '*/*' in accept_header
:
684 if 'application/zip' in accept_header
or '*/*' in accept_header
:
686 if not accept_text
and not accept_zip
:
687 raise EngineException("provide request header 'Accept' with 'application/zip' or 'text/plain'",
688 http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
690 content
= self
.get_item(session
, item
, _id
)
691 if content
["_admin"]["onboardingState"] != "ONBOARDED":
692 raise EngineException("Cannot get content because this resource is not at 'ONBOARDED' state. "
693 "onboardingState is {}".format(content
["_admin"]["onboardingState"]),
694 http_code
=HTTPStatus
.CONFLICT
)
695 storage
= content
["_admin"]["storage"]
696 if path
is not None and path
!= "$DESCRIPTOR": # artifacts
697 if not storage
.get('pkg-dir'):
698 raise EngineException("Packages does not contains artifacts", http_code
=HTTPStatus
.BAD_REQUEST
)
699 if self
.fs
.file_exists((storage
['folder'], storage
['pkg-dir'], *path
), 'dir'):
700 folder_content
= self
.fs
.dir_ls((storage
['folder'], storage
['pkg-dir'], *path
))
701 return folder_content
, "text/plain"
702 # TODO manage folders in http
704 return self
.fs
.file_open((storage
['folder'], storage
['pkg-dir'], *path
), "rb"), \
705 "application/octet-stream"
707 # pkgtype accept ZIP TEXT -> result
708 # manyfiles yes X -> zip
710 # onefile yes no -> zip
713 if accept_text
and (not storage
.get('pkg-dir') or path
== "$DESCRIPTOR"):
714 return self
.fs
.file_open((storage
['folder'], storage
['descriptor']), "r"), "text/plain"
715 elif storage
.get('pkg-dir') and not accept_zip
:
716 raise EngineException("Packages that contains several files need to be retrieved with 'application/zip'"
717 "Accept header", http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
719 if not storage
.get('zipfile'):
720 # TODO generate zipfile if not present
721 raise EngineException("Only allowed 'text/plain' Accept header for this descriptor. To be solved in future versions"
722 "", http_code
=HTTPStatus
.NOT_ACCEPTABLE
)
723 return self
.fs
.file_open((storage
['folder'], storage
['zipfile']), "rb"), "application/zip"
725 def get_item_list(self
, session
, item
, filter={}):
728 :param session: contains the used login username and working project
729 :param item: it can be: users, projects, vnfds, nsds, ...
730 :param filter: filter of data to be applied
731 :return: The list, it can be empty if no one match the filter.
733 # TODO add admin to filter, validate rights
734 # TODO transform data for SOL005 URL requests. Transform filtering
735 # TODO implement "field-type" query string SOL005
737 self
._add
_read
_filter
(session
, item
, filter)
738 return self
.db
.get_list(item
, filter)
740 def get_item(self
, session
, item
, _id
):
742 Get complete information on an items
743 :param session: contains the used login username and working project
744 :param item: it can be: users, projects, vnfds, nsds,
745 :param _id: server id of the item
746 :return: dictionary, raise exception if not found.
749 filter = {"_id": _id
}
750 # TODO add admin to filter, validate rights
751 # TODO transform data for SOL005 URL requests
752 self
._add
_read
_filter
(session
, item
, filter)
753 return self
.db
.get_one(item
, filter)
755 def del_item_list(self
, session
, item
, filter={}):
757 Delete a list of items
758 :param session: contains the used login username and working project
759 :param item: it can be: users, projects, vnfds, nsds, ...
760 :param filter: filter of data to be applied
761 :return: The deleted list, it can be empty if no one match the filter.
763 # TODO add admin to filter, validate rights
764 self
._add
_read
_filter
(session
, item
, filter)
765 return self
.db
.del_list(item
, filter)
767 def del_item(self
, session
, item
, _id
, force
=False):
769 Get complete information on an items
770 :param session: contains the used login username and working project
771 :param item: it can be: users, projects, vnfds, nsds, ...
772 :param _id: server id of the item
773 :param force: indicates if deletion must be forced in case of conflict
774 :return: dictionary with deleted item _id. It raises exception if not found.
776 # TODO add admin to filter, validate rights
777 # data = self.get_item(item, _id)
778 filter = {"_id": _id
}
779 self
._add
_delete
_filter
(session
, item
, filter)
782 nsr
= self
.db
.get_one(item
, filter)
783 if nsr
["_admin"]["nsState"] == "INSTANTIATED" and not force
:
784 raise EngineException("nsr '{}' cannot be deleted because it is in 'INSTANTIATED' state. "
785 "Launch 'terminate' action first; or force deletion".format(_id
),
786 http_code
=HTTPStatus
.CONFLICT
)
787 v
= self
.db
.del_one(item
, {"_id": _id
})
788 self
.db
.del_list("nslcmops", {"nsInstanceId": _id
})
789 self
.msg
.write("ns", "deleted", {"_id": _id
})
791 if item
in ("vim_accounts", "sdns"):
792 desc
= self
.db
.get_one(item
, filter)
793 desc
["_admin"]["to_delete"] = True
794 self
.db
.replace(item
, _id
, desc
) # TODO change to set_one
795 if item
== "vim_accounts":
796 self
.msg
.write("vim_account", "delete", {"_id": _id
})
798 self
.msg
.write("sdn", "delete", {"_id": _id
})
799 return {"deleted": 1} # TODO indicate an offline operation to return 202 ACCEPTED
801 v
= self
.db
.del_one(item
, filter)
802 self
.fs
.file_delete(_id
, ignore_non_exist
=True)
807 Prune database not needed content
810 return self
.db
.del_list("nsrs", {"_admin.to_delete": True})
812 def create_admin(self
):
814 Creates a new user admin/admin into database if database is empty. Useful for initialization
815 :return: _id identity of the inserted data, or None
817 users
= self
.db
.get_one("users", fail_on_empty
=False, fail_on_more
=False)
820 # raise EngineException("Unauthorized. Database users is not empty", HTTPStatus.UNAUTHORIZED)
821 indata
= {"username": "admin", "password": "admin", "projects": ["admin"]}
822 fake_session
= {"project_id": "admin", "username": "admin"}
823 self
._format
_new
_data
(fake_session
, "users", indata
)
824 _id
= self
.db
.create("users", indata
)
827 def init_db(self
, target_version
='1.0'):
829 Init database if empty. If not empty it checks that database version is ok.
830 If empty, it creates a new user admin/admin at 'users' and a new entry at 'version'
831 :return: None if ok, exception if error or if the version is different.
833 version
= self
.db
.get_one("version", fail_on_empty
=False, fail_on_more
=False)
837 # create database version
839 "_id": '1.0', # version text
840 "version": 1000, # version number
841 "date": "2018-04-12", # version date
842 "description": "initial design", # changes in this version
843 'status': 'ENABLED' # ENABLED, DISABLED (migration in process), ERROR,
845 self
.db
.create("version", version_data
)
846 elif version
["_id"] != target_version
:
847 # TODO implement migration process
848 raise EngineException("Wrong database version '{}'. Expected '{}'".format(
849 version
["_id"], target_version
), HTTPStatus
.INTERNAL_SERVER_ERROR
)
850 elif version
["status"] != 'ENABLED':
851 raise EngineException("Wrong database status '{}'".format(
852 version
["status"]), HTTPStatus
.INTERNAL_SERVER_ERROR
)
855 def _edit_item(self
, session
, item
, id, content
, indata
={}, kwargs
=None):
857 indata
= self
._remove
_envelop
(item
, indata
)
859 # Override descriptor with query string kwargs
862 for k
, v
in kwargs
.items():
863 update_content
= indata
867 if kitem_old
is not None:
868 update_content
= update_content
[kitem_old
]
869 if isinstance(update_content
, dict):
871 elif isinstance(update_content
, list):
872 kitem_old
= int(kitem
)
874 raise EngineException(
875 "Invalid query string '{}'. Descriptor is not a list nor dict at '{}'".format(k
, kitem
))
876 update_content
[kitem_old
] = v
878 raise EngineException(
879 "Invalid query string '{}'. Descriptor does not contain '{}'".format(k
, kitem_old
))
881 raise EngineException("Invalid query string '{}'. Expected integer index list instead of '{}'".format(
884 raise EngineException(
885 "Invalid query string '{}'. Index '{}' out of range".format(k
, kitem_old
))
887 validate_input(content
, item
, new
=False)
888 except ValidationError
as e
:
889 raise EngineException(e
, HTTPStatus
.UNPROCESSABLE_ENTITY
)
891 _deep_update(content
, indata
)
892 self
._validate
_new
_data
(session
, item
, content
, id)
893 # self._format_new_data(session, item, content)
894 self
.db
.replace(item
, id, content
)
895 if item
in ("vim_accounts", "sdns"):
896 indata
.pop("_admin", None)
898 if item
== "vim_accounts":
899 self
.msg
.write("vim_account", "edit", indata
)
901 self
.msg
.write("sdn", "edit", indata
)
904 def edit_item(self
, session
, item
, _id
, indata
={}, kwargs
=None):
906 Update an existing entry at database
907 :param session: contains the used login username and working project
908 :param item: it can be: users, projects, vnfds, nsds, ...
909 :param _id: identifier to be updated
910 :param indata: data to be inserted
911 :param kwargs: used to override the indata descriptor
912 :return: dictionary, raise exception if not found.
915 content
= self
.get_item(session
, item
, _id
)
916 return self
._edit
_item
(session
, item
, _id
, content
, indata
, kwargs
)