1 |
|
# -*- coding: utf-8 -*- |
2 |
|
|
3 |
|
# Licensed under the Apache License, Version 2.0 (the "License"); |
4 |
|
# you may not use this file except in compliance with the License. |
5 |
|
# You may obtain a copy of the License at |
6 |
|
# |
7 |
|
# http://www.apache.org/licenses/LICENSE-2.0 |
8 |
|
# |
9 |
|
# Unless required by applicable law or agreed to in writing, software |
10 |
|
# distributed under the License is distributed on an "AS IS" BASIS, |
11 |
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or |
12 |
|
# implied. |
13 |
|
# See the License for the specific language governing permissions and |
14 |
|
# limitations under the License. |
15 |
|
|
16 |
1 |
import logging |
17 |
|
# import yaml |
18 |
1 |
from osm_common import dbmongo, dbmemory, fslocal, fsmongo, msglocal, msgkafka, version as common_version |
19 |
1 |
from osm_common.dbbase import DbException |
20 |
1 |
from osm_common.fsbase import FsException |
21 |
1 |
from osm_common.msgbase import MsgException |
22 |
1 |
from http import HTTPStatus |
23 |
|
|
24 |
1 |
from osm_nbi.authconn_keystone import AuthconnKeystone |
25 |
1 |
from osm_nbi.authconn_internal import AuthconnInternal |
26 |
1 |
from osm_nbi.authconn_tacacs import AuthconnTacacs |
27 |
1 |
from osm_nbi.base_topic import EngineException, versiontuple |
28 |
1 |
from osm_nbi.admin_topics import VimAccountTopic, WimAccountTopic, SdnTopic |
29 |
1 |
from osm_nbi.admin_topics import K8sClusterTopic, K8sRepoTopic, OsmRepoTopic |
30 |
1 |
from osm_nbi.admin_topics import UserTopicAuth, ProjectTopicAuth, RoleTopicAuth |
31 |
1 |
from osm_nbi.descriptor_topics import VnfdTopic, NsdTopic, PduTopic, NstTopic, VnfPkgOpTopic |
32 |
1 |
from osm_nbi.instance_topics import NsrTopic, VnfrTopic, NsLcmOpTopic, NsiTopic, NsiLcmOpTopic |
33 |
1 |
from osm_nbi.pmjobs_topics import PmJobsTopic |
34 |
1 |
from osm_nbi.subscription_topics import NslcmSubscriptionsTopic |
35 |
1 |
from base64 import b64encode |
36 |
1 |
from os import urandom # , path |
37 |
1 |
from threading import Lock |
38 |
|
|
39 |
1 |
__author__ = "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>" |
40 |
1 |
min_common_version = "0.1.16" |
41 |
|
|
42 |
|
|
43 |
1 |
class Engine(object): |
44 |
1 |
map_from_topic_to_class = { |
45 |
|
"vnfds": VnfdTopic, |
46 |
|
"nsds": NsdTopic, |
47 |
|
"nsts": NstTopic, |
48 |
|
"pdus": PduTopic, |
49 |
|
"nsrs": NsrTopic, |
50 |
|
"vnfrs": VnfrTopic, |
51 |
|
"nslcmops": NsLcmOpTopic, |
52 |
|
"vim_accounts": VimAccountTopic, |
53 |
|
"wim_accounts": WimAccountTopic, |
54 |
|
"sdns": SdnTopic, |
55 |
|
"k8sclusters": K8sClusterTopic, |
56 |
|
"k8srepos": K8sRepoTopic, |
57 |
|
"osmrepos": OsmRepoTopic, |
58 |
|
"users": UserTopicAuth, # Valid for both internal and keystone authentication backends |
59 |
|
"projects": ProjectTopicAuth, # Valid for both internal and keystone authentication backends |
60 |
|
"roles": RoleTopicAuth, # Valid for both internal and keystone authentication backends |
61 |
|
"nsis": NsiTopic, |
62 |
|
"nsilcmops": NsiLcmOpTopic, |
63 |
|
"vnfpkgops": VnfPkgOpTopic, |
64 |
|
"nslcm_subscriptions": NslcmSubscriptionsTopic, |
65 |
|
# [NEW_TOPIC]: add an entry here |
66 |
|
# "pm_jobs": PmJobsTopic will be added manually because it needs other parameters |
67 |
|
} |
68 |
|
|
69 |
1 |
map_target_version_to_int = { |
70 |
|
"1.0": 1000, |
71 |
|
"1.1": 1001, |
72 |
|
"1.2": 1002, |
73 |
|
# Add new versions here |
74 |
|
} |
75 |
|
|
76 |
1 |
def __init__(self, authenticator): |
77 |
0 |
self.db = None |
78 |
0 |
self.fs = None |
79 |
0 |
self.msg = None |
80 |
0 |
self.authconn = None |
81 |
0 |
self.config = None |
82 |
|
# self.operations = None |
83 |
0 |
self.logger = logging.getLogger("nbi.engine") |
84 |
0 |
self.map_topic = {} |
85 |
0 |
self.write_lock = None |
86 |
|
# self.token_cache = token_cache |
87 |
0 |
self.authenticator = authenticator |
88 |
|
|
89 |
1 |
def start(self, config): |
90 |
|
""" |
91 |
|
Connect to database, filesystem storage, and messaging |
92 |
|
:param config: two level dictionary with configuration. Top level should contain 'database', 'storage', |
93 |
|
:return: None |
94 |
|
""" |
95 |
0 |
self.config = config |
96 |
|
# check right version of common |
97 |
0 |
if versiontuple(common_version) < versiontuple(min_common_version): |
98 |
0 |
raise EngineException("Not compatible osm/common version '{}'. Needed '{}' or higher".format( |
99 |
|
common_version, min_common_version)) |
100 |
|
|
101 |
0 |
try: |
102 |
0 |
if not self.db: |
103 |
0 |
if config["database"]["driver"] == "mongo": |
104 |
0 |
self.db = dbmongo.DbMongo() |
105 |
0 |
self.db.db_connect(config["database"]) |
106 |
0 |
elif config["database"]["driver"] == "memory": |
107 |
0 |
self.db = dbmemory.DbMemory() |
108 |
0 |
self.db.db_connect(config["database"]) |
109 |
|
else: |
110 |
0 |
raise EngineException("Invalid configuration param '{}' at '[database]':'driver'".format( |
111 |
|
config["database"]["driver"])) |
112 |
0 |
if not self.fs: |
113 |
0 |
if config["storage"]["driver"] == "local": |
114 |
0 |
self.fs = fslocal.FsLocal() |
115 |
0 |
self.fs.fs_connect(config["storage"]) |
116 |
0 |
elif config["storage"]["driver"] == "mongo": |
117 |
0 |
self.fs = fsmongo.FsMongo() |
118 |
0 |
self.fs.fs_connect(config["storage"]) |
119 |
|
else: |
120 |
0 |
raise EngineException("Invalid configuration param '{}' at '[storage]':'driver'".format( |
121 |
|
config["storage"]["driver"])) |
122 |
0 |
if not self.msg: |
123 |
0 |
if config["message"]["driver"] == "local": |
124 |
0 |
self.msg = msglocal.MsgLocal() |
125 |
0 |
self.msg.connect(config["message"]) |
126 |
0 |
elif config["message"]["driver"] == "kafka": |
127 |
0 |
self.msg = msgkafka.MsgKafka() |
128 |
0 |
self.msg.connect(config["message"]) |
129 |
|
else: |
130 |
0 |
raise EngineException("Invalid configuration param '{}' at '[message]':'driver'".format( |
131 |
|
config["message"]["driver"])) |
132 |
0 |
if not self.authconn: |
133 |
0 |
if config["authentication"]["backend"] == "keystone": |
134 |
0 |
self.authconn = AuthconnKeystone(config["authentication"], self.db, |
135 |
|
self.authenticator.role_permissions) |
136 |
0 |
elif config["authentication"]["backend"] == "tacacs": |
137 |
0 |
self.authconn = AuthconnTacacs(config["authentication"], self.db, |
138 |
|
self.authenticator.role_permissions) |
139 |
|
else: |
140 |
0 |
self.authconn = AuthconnInternal(config["authentication"], self.db, |
141 |
|
self.authenticator.role_permissions) |
142 |
|
# if not self.operations: |
143 |
|
# if "resources_to_operations" in config["rbac"]: |
144 |
|
# resources_to_operations_file = config["rbac"]["resources_to_operations"] |
145 |
|
# else: |
146 |
|
# possible_paths = ( |
147 |
|
# __file__[:__file__.rfind("engine.py")] + "resources_to_operations.yml", |
148 |
|
# "./resources_to_operations.yml" |
149 |
|
# ) |
150 |
|
# for config_file in possible_paths: |
151 |
|
# if path.isfile(config_file): |
152 |
|
# resources_to_operations_file = config_file |
153 |
|
# break |
154 |
|
# if not resources_to_operations_file: |
155 |
|
# raise EngineException("Invalid permission configuration:" |
156 |
|
# "resources_to_operations file missing") |
157 |
|
# |
158 |
|
# with open(resources_to_operations_file, 'r') as f: |
159 |
|
# resources_to_operations = yaml.load(f, Loader=yaml.Loader) |
160 |
|
# |
161 |
|
# self.operations = [] |
162 |
|
# |
163 |
|
# for _, value in resources_to_operations["resources_to_operations"].items(): |
164 |
|
# if value not in self.operations: |
165 |
|
# self.operations += [value] |
166 |
|
|
167 |
0 |
self.write_lock = Lock() |
168 |
|
# create one class per topic |
169 |
0 |
for topic, topic_class in self.map_from_topic_to_class.items(): |
170 |
|
# if self.auth and topic_class in (UserTopicAuth, ProjectTopicAuth): |
171 |
|
# self.map_topic[topic] = topic_class(self.db, self.fs, self.msg, self.auth) |
172 |
0 |
self.map_topic[topic] = topic_class(self.db, self.fs, self.msg, self.authconn) |
173 |
|
|
174 |
0 |
self.map_topic["pm_jobs"] = PmJobsTopic(self.db, config["prometheus"].get("host"), |
175 |
|
config["prometheus"].get("port")) |
176 |
0 |
except (DbException, FsException, MsgException) as e: |
177 |
0 |
raise EngineException(str(e), http_code=e.http_code) |
178 |
|
|
179 |
1 |
def stop(self): |
180 |
0 |
try: |
181 |
0 |
if self.db: |
182 |
0 |
self.db.db_disconnect() |
183 |
0 |
if self.fs: |
184 |
0 |
self.fs.fs_disconnect() |
185 |
0 |
if self.msg: |
186 |
0 |
self.msg.disconnect() |
187 |
0 |
self.write_lock = None |
188 |
0 |
except (DbException, FsException, MsgException) as e: |
189 |
0 |
raise EngineException(str(e), http_code=e.http_code) |
190 |
|
|
191 |
1 |
def new_item(self, rollback, session, topic, indata=None, kwargs=None, headers=None): |
192 |
|
""" |
193 |
|
Creates a new entry into database. For nsds and vnfds it creates an almost empty DISABLED entry, |
194 |
|
that must be completed with a call to method upload_content |
195 |
|
:param rollback: list to append created items at database in case a rollback must to be done |
196 |
|
:param session: contains the used login username and working project, force to avoid checkins, public |
197 |
|
:param topic: it can be: users, projects, vim_accounts, sdns, nsrs, nsds, vnfds |
198 |
|
:param indata: data to be inserted |
199 |
|
:param kwargs: used to override the indata descriptor |
200 |
|
:param headers: http request headers |
201 |
|
:return: _id: identity of the inserted data. |
202 |
|
""" |
203 |
0 |
if topic not in self.map_topic: |
204 |
0 |
raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) |
205 |
0 |
with self.write_lock: |
206 |
0 |
return self.map_topic[topic].new(rollback, session, indata, kwargs, headers) |
207 |
|
|
208 |
1 |
def upload_content(self, session, topic, _id, indata, kwargs, headers): |
209 |
|
""" |
210 |
|
Upload content for an already created entry (_id) |
211 |
|
:param session: contains the used login username and working project |
212 |
|
:param topic: it can be: users, projects, vnfds, nsds, |
213 |
|
:param _id: server id of the item |
214 |
|
:param indata: data to be inserted |
215 |
|
:param kwargs: used to override the indata descriptor |
216 |
|
:param headers: http request headers |
217 |
|
:return: _id: identity of the inserted data. |
218 |
|
""" |
219 |
0 |
if topic not in self.map_topic: |
220 |
0 |
raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) |
221 |
0 |
with self.write_lock: |
222 |
0 |
return self.map_topic[topic].upload_content(session, _id, indata, kwargs, headers) |
223 |
|
|
224 |
1 |
def get_item_list(self, session, topic, filter_q=None, api_req=False): |
225 |
|
""" |
226 |
|
Get a list of items |
227 |
|
:param session: contains the used login username and working project |
228 |
|
:param topic: it can be: users, projects, vnfds, nsds, ... |
229 |
|
:param filter_q: filter of data to be applied |
230 |
|
:param api_req: True if this call is serving an external API request. False if serving internal request. |
231 |
|
:return: The list, it can be empty if no one match the filter_q. |
232 |
|
""" |
233 |
0 |
if topic not in self.map_topic: |
234 |
0 |
raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) |
235 |
0 |
return self.map_topic[topic].list(session, filter_q, api_req) |
236 |
|
|
237 |
1 |
def get_item(self, session, topic, _id, api_req=False): |
238 |
|
""" |
239 |
|
Get complete information on an item |
240 |
|
:param session: contains the used login username and working project |
241 |
|
:param topic: it can be: users, projects, vnfds, nsds, |
242 |
|
:param _id: server id of the item |
243 |
|
:param api_req: True if this call is serving an external API request. False if serving internal request. |
244 |
|
:return: dictionary, raise exception if not found. |
245 |
|
""" |
246 |
0 |
if topic not in self.map_topic: |
247 |
0 |
raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) |
248 |
0 |
return self.map_topic[topic].show(session, _id, api_req) |
249 |
|
|
250 |
1 |
def get_file(self, session, topic, _id, path=None, accept_header=None): |
251 |
|
""" |
252 |
|
Get descriptor package or artifact file content |
253 |
|
:param session: contains the used login username and working project |
254 |
|
:param topic: it can be: users, projects, vnfds, nsds, |
255 |
|
:param _id: server id of the item |
256 |
|
:param path: artifact path or "$DESCRIPTOR" or None |
257 |
|
:param accept_header: Content of Accept header. Must contain applition/zip or/and text/plain |
258 |
|
:return: opened file plus Accept format or raises an exception |
259 |
|
""" |
260 |
0 |
if topic not in self.map_topic: |
261 |
0 |
raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) |
262 |
0 |
return self.map_topic[topic].get_file(session, _id, path, accept_header) |
263 |
|
|
264 |
1 |
def del_item_list(self, session, topic, _filter=None): |
265 |
|
""" |
266 |
|
Delete a list of items |
267 |
|
:param session: contains the used login username and working project |
268 |
|
:param topic: it can be: users, projects, vnfds, nsds, ... |
269 |
|
:param _filter: filter of data to be applied |
270 |
|
:return: The deleted list, it can be empty if no one match the _filter. |
271 |
|
""" |
272 |
0 |
if topic not in self.map_topic: |
273 |
0 |
raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) |
274 |
0 |
with self.write_lock: |
275 |
0 |
return self.map_topic[topic].delete_list(session, _filter) |
276 |
|
|
277 |
1 |
def del_item(self, session, topic, _id, not_send_msg=None): |
278 |
|
""" |
279 |
|
Delete item by its internal id |
280 |
|
:param session: contains the used login username and working project |
281 |
|
:param topic: it can be: users, projects, vnfds, nsds, ... |
282 |
|
:param _id: server id of the item |
283 |
|
:param not_send_msg: If False, message will not be sent to kafka. |
284 |
|
If a list, message is not sent, but content is stored in this variable so that the caller can send this |
285 |
|
message using its own loop. If None, message is sent |
286 |
|
:return: dictionary with deleted item _id. It raises exception if not found. |
287 |
|
""" |
288 |
0 |
if topic not in self.map_topic: |
289 |
0 |
raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) |
290 |
0 |
with self.write_lock: |
291 |
0 |
return self.map_topic[topic].delete(session, _id, not_send_msg=not_send_msg) |
292 |
|
|
293 |
1 |
def edit_item(self, session, topic, _id, indata=None, kwargs=None): |
294 |
|
""" |
295 |
|
Update an existing entry at database |
296 |
|
:param session: contains the used login username and working project |
297 |
|
:param topic: it can be: users, projects, vnfds, nsds, ... |
298 |
|
:param _id: identifier to be updated |
299 |
|
:param indata: data to be inserted |
300 |
|
:param kwargs: used to override the indata descriptor |
301 |
|
:return: dictionary with edited item _id, raise exception if not found. |
302 |
|
""" |
303 |
0 |
if topic not in self.map_topic: |
304 |
0 |
raise EngineException("Unknown topic {}!!!".format(topic), HTTPStatus.INTERNAL_SERVER_ERROR) |
305 |
0 |
with self.write_lock: |
306 |
0 |
return self.map_topic[topic].edit(session, _id, indata, kwargs) |
307 |
|
|
308 |
1 |
def upgrade_db(self, current_version, target_version): |
309 |
0 |
if target_version not in self.map_target_version_to_int.keys(): |
310 |
0 |
raise EngineException("Cannot upgrade to version '{}' with this version of code".format(target_version), |
311 |
|
http_code=HTTPStatus.INTERNAL_SERVER_ERROR) |
312 |
|
|
313 |
0 |
if current_version == target_version: |
314 |
0 |
return |
315 |
|
|
316 |
0 |
target_version_int = self.map_target_version_to_int[target_version] |
317 |
|
|
318 |
0 |
if not current_version: |
319 |
|
# create database version |
320 |
0 |
serial = urandom(32) |
321 |
0 |
version_data = { |
322 |
|
"_id": "version", # Always "version" |
323 |
|
"version_int": 1000, # version number |
324 |
|
"version": "1.0", # version text |
325 |
|
"date": "2018-10-25", # version date |
326 |
|
"description": "added serial", # changes in this version |
327 |
|
'status': "ENABLED", # ENABLED, DISABLED (migration in process), ERROR, |
328 |
|
'serial': b64encode(serial) |
329 |
|
} |
330 |
0 |
self.db.create("admin", version_data) |
331 |
0 |
self.db.set_secret_key(serial) |
332 |
0 |
current_version = "1.0" |
333 |
|
|
334 |
0 |
if current_version in ("1.0", "1.1") and target_version_int >= self.map_target_version_to_int["1.2"]: |
335 |
0 |
if self.config['authentication']['backend'] == "internal": |
336 |
0 |
self.db.del_list("roles") |
337 |
|
|
338 |
0 |
version_data = { |
339 |
|
"_id": "version", |
340 |
|
"version_int": 1002, |
341 |
|
"version": "1.2", |
342 |
|
"date": "2019-06-11", |
343 |
|
"description": "set new format for roles_operations" |
344 |
|
} |
345 |
|
|
346 |
0 |
self.db.set_one("admin", {"_id": "version"}, version_data) |
347 |
0 |
current_version = "1.2" |
348 |
|
# TODO add future migrations here |
349 |
|
|
350 |
1 |
def init_db(self, target_version='1.0'): |
351 |
|
""" |
352 |
|
Init database if empty. If not empty it checks that database version and migrates if needed |
353 |
|
If empty, it creates a new user admin/admin at 'users' and a new entry at 'version' |
354 |
|
:param target_version: check desired database version. Migrate to it if possible or raises exception |
355 |
|
:return: None if ok, exception if error or if the version is different. |
356 |
|
""" |
357 |
|
|
358 |
0 |
version_data = self.db.get_one("admin", {"_id": "version"}, fail_on_empty=False, fail_on_more=True) |
359 |
|
# check database status is ok |
360 |
0 |
if version_data and version_data.get("status") != 'ENABLED': |
361 |
0 |
raise EngineException("Wrong database status '{}'".format( |
362 |
|
version_data["status"]), HTTPStatus.INTERNAL_SERVER_ERROR) |
363 |
|
|
364 |
|
# check version |
365 |
0 |
db_version = None if not version_data else version_data.get("version") |
366 |
0 |
if db_version != target_version: |
367 |
0 |
self.upgrade_db(db_version, target_version) |
368 |
|
|
369 |
0 |
return |