nbi_server = None # instance of Server class
subscription_thread = None # instance of SubscriptionThread class
-
"""
North Bound Interface (O: OSM specific; 5,X: SOL005 not implemented yet; O5: SOL005 implemented)
URL: /osm GET POST PUT DELETE PATCH
/<id> O O O
/sdns O O
/<id> O O O
+ /k8sclusters O O
+ /<id> O O O
+ /k8srepos O O
+ /<id> O O
+ /osmrepos O O
+ /<id> O O
/nst/v1 O O
/netslice_templates_content O O
},
"users": {"METHODS": ("GET", "POST"),
"ROLE_PERMISSION": "users:",
- "<ID>": {"METHODS": ("GET", "POST", "DELETE", "PATCH", "PUT"),
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
"ROLE_PERMISSION": "users:id:"
}
},
"projects": {"METHODS": ("GET", "POST"),
"ROLE_PERMISSION": "projects:",
- "<ID>": {"METHODS": ("GET", "DELETE", "PUT"),
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
"ROLE_PERMISSION": "projects:id:"}
},
"roles": {"METHODS": ("GET", "POST"),
"ROLE_PERMISSION": "roles:",
- "<ID>": {"METHODS": ("GET", "POST", "DELETE", "PUT"),
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
"ROLE_PERMISSION": "roles:id:"
}
},
"vims": {"METHODS": ("GET", "POST"),
"ROLE_PERMISSION": "vims:",
- "<ID>": {"METHODS": ("GET", "DELETE", "PATCH", "PUT"),
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
"ROLE_PERMISSION": "vims:id:"
}
},
"vim_accounts": {"METHODS": ("GET", "POST"),
"ROLE_PERMISSION": "vim_accounts:",
- "<ID>": {"METHODS": ("GET", "DELETE", "PATCH", "PUT"),
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
"ROLE_PERMISSION": "vim_accounts:id:"
}
},
"wim_accounts": {"METHODS": ("GET", "POST"),
"ROLE_PERMISSION": "wim_accounts:",
- "<ID>": {"METHODS": ("GET", "DELETE", "PATCH", "PUT"),
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
"ROLE_PERMISSION": "wim_accounts:id:"
}
},
"sdns": {"METHODS": ("GET", "POST"),
"ROLE_PERMISSION": "sdn_controllers:",
- "<ID>": {"METHODS": ("GET", "DELETE", "PATCH", "PUT"),
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
"ROLE_PERMISSION": "sdn_controllers:id:"
}
},
+ "k8sclusters": {"METHODS": ("GET", "POST"),
+ "ROLE_PERMISSION": "k8sclusters:",
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
+ "ROLE_PERMISSION": "k8sclusters:id:"
+ }
+ },
+ "k8srepos": {"METHODS": ("GET", "POST"),
+ "ROLE_PERMISSION": "k8srepos:",
+ "<ID>": {"METHODS": ("GET", "DELETE"),
+ "ROLE_PERMISSION": "k8srepos:id:"
+ }
+ },
+ "osmrepos": {"METHODS": ("GET", "POST"),
+ "ROLE_PERMISSION": "osmrepos:",
+ "<ID>": {"METHODS": ("GET", "DELETE", "PATCH"),
+ "ROLE_PERMISSION": "osmrepos:id:"
+ }
+ },
+ "domains": {"METHODS": ("GET", ),
+ "ROLE_PERMISSION": "domains:",
+ },
}
},
"pdu": {
"artifacts": {"*": {"METHODS": ("GET", ),
"ROLE_PERMISSION": "vnfds:id:vnfd_artifact:"
}
- }
+ },
+ "action": {"METHODS": ("POST", ),
+ "ROLE_PERMISSION": "vnfds:id:action:"
+ },
}
},
"subscriptions": {"TODO": ("GET", "POST"),
"<ID>": {"TODO": ("GET", "DELETE")}
},
+ "vnfpkg_op_occs": {"METHODS": ("GET", ),
+ "ROLE_PERMISSION": "vnfds:vnfpkgops:",
+ "<ID>": {"METHODS": ("GET", ),
+ "ROLE_PERMISSION": "vnfds:vnfpkgops:id:"
+ }
+ },
}
},
"nslcm": {
def __init__(self):
self.instance += 1
- self.engine = Engine()
self.authenticator = Authenticator(valid_url_methods, valid_query_string)
+ self.engine = Engine(self.authenticator)
def _format_in(self, kwargs):
try:
cherrypy.request.headers.pop("Content-File-MD5", None)
elif "application/yaml" in cherrypy.request.headers["Content-Type"]:
error_text = "Invalid yaml format "
- indata = yaml.load(cherrypy.request.body)
+ indata = yaml.load(cherrypy.request.body, Loader=yaml.SafeLoader)
cherrypy.request.headers.pop("Content-File-MD5", None)
elif "application/binary" in cherrypy.request.headers["Content-Type"] or \
"application/gzip" in cherrypy.request.headers["Content-Type"] or \
# "Only 'Content-Type' of type 'application/json' or
# 'application/yaml' for input format are available")
error_text = "Invalid yaml format "
- indata = yaml.load(cherrypy.request.body)
+ indata = yaml.load(cherrypy.request.body, Loader=yaml.SafeLoader)
cherrypy.request.headers.pop("Content-File-MD5", None)
else:
error_text = "Invalid yaml format "
- indata = yaml.load(cherrypy.request.body)
+ indata = yaml.load(cherrypy.request.body, Loader=yaml.SafeLoader)
cherrypy.request.headers.pop("Content-File-MD5", None)
if not indata:
indata = {}
kwargs[k] = None
elif format_yaml:
try:
- kwargs[k] = yaml.load(v)
+ kwargs[k] = yaml.load(v, Loader=yaml.SafeLoader)
except Exception:
pass
elif k.endswith(".gt") or k.endswith(".lt") or k.endswith(".gte") or k.endswith(".lte"):
v[index] = None
elif format_yaml:
try:
- v[index] = yaml.load(v[index])
+ v[index] = yaml.load(v[index], Loader=yaml.SafeLoader)
except Exception:
pass
}
return self._format_out(problem_details, None)
+ def domain(self):
+ try:
+ domains = {
+ "user_domain_name": cherrypy.tree.apps['/osm'].config["authentication"].get("user_domain_name"),
+ "project_domain_name": cherrypy.tree.apps['/osm'].config["authentication"].get("project_domain_name")}
+ return self._format_out(domains)
+ except NbiException as e:
+ cherrypy.response.status = e.http_code.value
+ problem_details = {
+ "code": e.http_code.name,
+ "status": e.http_code.value,
+ "detail": str(e),
+ }
+ return self._format_out(problem_details, None)
+
@staticmethod
def _format_login(token_info):
"""
@cherrypy.expose
def test(self, *args, **kwargs):
+ if not cherrypy.config.get("server.enable_test") or (isinstance(cherrypy.config["server.enable_test"], str) and
+ cherrypy.config["server.enable_test"].lower() == "false"):
+ cherrypy.response.status = HTTPStatus.METHOD_NOT_ALLOWED.value
+ return "test URL is disabled"
thread_info = None
if args and args[0] == "help":
return "<html><pre>\ninit\nfile/<name> download file\ndb-clear/table\nfs-clear[/folder]\nlogin\nlogin2\n"\
return_text = "<html><pre>{} ->\n".format(main_topic)
try:
if cherrypy.request.method == 'POST':
- to_send = yaml.load(cherrypy.request.body)
+ to_send = yaml.load(cherrypy.request.body, Loader=yaml.SafeLoader)
for k, v in to_send.items():
self.engine.msg.write(main_topic, k, v)
return_text += " {}: {}\n".format(k, v)
elif cherrypy.request.method == 'GET':
for k, v in kwargs.items():
- self.engine.msg.write(main_topic, k, yaml.load(v))
- return_text += " {}: {}\n".format(k, yaml.load(v))
+ v_dict = yaml.load(v, Loader=yaml.SafeLoader)
+ self.engine.msg.write(main_topic, k, v_dict)
+ return_text += " {}: {}\n".format(k, v_dict)
except Exception as e:
return_text += "Error: " + str(e)
return_text += "</pre></html>\n"
method: show, list, delete, write
"""
admin_query = {"force": False, "project_id": (token_info["project_id"], ), "username": token_info["username"],
- "admin": token_info["admin"], "public": None}
+ "admin": token_info["admin"], "public": None,
+ "allow_show_user_project_role": token_info["allow_show_user_project_role"]}
if kwargs:
# FORCE
if "FORCE" in kwargs:
query_string_operations = self._extract_query_string_operations(kwargs, method)
if main_topic == "admin" and topic == "tokens":
return self.token(method, _id, kwargs)
-
- token_info = self.authenticator.authorize(role_permission, query_string_operations)
+ token_info = self.authenticator.authorize(role_permission, query_string_operations, _id)
+ if main_topic == "admin" and topic == "domains":
+ return self.domain()
engine_session = self._manage_admin_query(token_info, kwargs, method, _id)
indata = self._format_in(kwargs)
engine_topic = topic
engine_topic = "nsds"
elif main_topic == "vnfpkgm":
engine_topic = "vnfds"
+ if topic == "vnfpkg_op_occs":
+ engine_topic = "vnfpkgops"
+ if topic == "vnf_packages" and item == "action":
+ engine_topic = "vnfpkgops"
elif main_topic == "nslcm":
engine_topic = "nsrs"
if topic == "ns_lcm_op_occs":
# TODO check that project_id (_id in this context) has permissions
_id = args[0]
outdata = self.engine.get_item(engine_session, engine_topic, _id)
+
elif method == "POST":
cherrypy.response.status = HTTPStatus.CREATED.value
if topic in ("ns_descriptors_content", "vnf_packages_content", "netslice_templates_content"):
indata["netsliceInstanceId"] = _id
nsilcmop_id, _ = self.engine.new_item(rollback, engine_session, "nsilcmops", indata, kwargs)
outdata = {"id": _id, "nsilcmop_id": nsilcmop_id}
-
elif topic == "netslice_instances" and item:
indata["lcmOperationType"] = item
indata["netsliceInstanceId"] = _id
self._set_location_header(main_topic, version, "nsi_lcm_op_occs", _id)
outdata = {"id": _id}
cherrypy.response.status = HTTPStatus.ACCEPTED.value
+ elif topic == "vnf_packages" and item == "action":
+ indata["lcmOperationType"] = item
+ indata["vnfPkgId"] = _id
+ _id, _ = self.engine.new_item(rollback, engine_session, "vnfpkgops", indata, kwargs)
+ self._set_location_header(main_topic, version, "vnfpkg_op_occs", _id)
+ outdata = {"id": _id}
+ cherrypy.response.status = HTTPStatus.ACCEPTED.value
else:
_id, op_id = self.engine.new_item(rollback, engine_session, engine_topic, indata, kwargs,
cherrypy.request.headers)
outdata = self.engine.del_item_list(engine_session, engine_topic, kwargs)
cherrypy.response.status = HTTPStatus.OK.value
else: # len(args) > 1
- delete_in_process = False
+ # for NS NSI generate an operation
+ op_id = None
if topic == "ns_instances_content" and not engine_session["force"]:
nslcmop_desc = {
"lcmOperationType": "terminate",
"nsInstanceId": _id,
"autoremove": True
}
- opp_id, _ = self.engine.new_item(rollback, engine_session, "nslcmops", nslcmop_desc, None)
- if opp_id:
- delete_in_process = True
- outdata = {"_id": opp_id}
- cherrypy.response.status = HTTPStatus.ACCEPTED.value
+ op_id, _ = self.engine.new_item(rollback, engine_session, "nslcmops", nslcmop_desc, kwargs)
+ if op_id:
+ outdata = {"_id": op_id}
elif topic == "netslice_instances_content" and not engine_session["force"]:
nsilcmop_desc = {
"lcmOperationType": "terminate",
"netsliceInstanceId": _id,
"autoremove": True
}
- opp_id, _ = self.engine.new_item(rollback, engine_session, "nsilcmops", nsilcmop_desc, None)
- if opp_id:
- delete_in_process = True
- outdata = {"_id": opp_id}
- cherrypy.response.status = HTTPStatus.ACCEPTED.value
- if not delete_in_process:
- self.engine.del_item(engine_session, engine_topic, _id)
- cherrypy.response.status = HTTPStatus.NO_CONTENT.value
- if engine_topic in ("vim_accounts", "wim_accounts", "sdns"):
- cherrypy.response.status = HTTPStatus.ACCEPTED.value
+ op_id, _ = self.engine.new_item(rollback, engine_session, "nsilcmops", nsilcmop_desc, None)
+ if op_id:
+ outdata = {"_id": op_id}
+ # if there is not any deletion in process, delete
+ if not op_id:
+ op_id = self.engine.del_item(engine_session, engine_topic, _id)
+ if op_id:
+ outdata = {"op_id": op_id}
+ cherrypy.response.status = HTTPStatus.ACCEPTED.value if op_id else HTTPStatus.NO_CONTENT.value
elif method in ("PUT", "PATCH"):
op_id = None
# if Role information changes, it is needed to reload the information of roles
if topic == "roles" and method != "GET":
self.authenticator.load_operation_to_allowed_roles()
+
+ if topic == "projects" and method == "DELETE" \
+ or topic in ["users", "roles"] and method in ["PUT", "PATCH", "DELETE"]:
+ self.authenticator.remove_token_from_cache()
+
return self._format_out(outdata, token_info, _format)
except Exception as e:
if isinstance(e, (NbiException, EngineException, DbException, FsException, MsgException, AuthException,
subscription_thread.start()
# Do not capture except SubscriptionException
- # load and print version. Ignore possible errors, e.g. file not found
- try:
- backend = engine_config["authentication"]["backend"]
- nbi_version
- cherrypy.log.error("Starting OSM NBI Version '{}' with '{}' authentication backend"
- .format(nbi_version + " " + nbi_version_date, backend))
- except Exception:
- pass
+ backend = engine_config["authentication"]["backend"]
+ cherrypy.log.error("Starting OSM NBI Version '{} {}' with '{}' authentication backend"
+ .format(nbi_version, nbi_version_date, backend))
def _stop_service():
projects / osm / NBI.git / blobdiff