n2vc/store.py

   1 # Copyright 2021 Canonical Ltd.
   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 implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 import abc
  16 import asyncio
  17 from base64 import b64decode
  18 import re
  19 import typing
  20
  21 from Crypto.Cipher import AES
  22 from motor.motor_asyncio import AsyncIOMotorClient
  23 from n2vc.config import EnvironConfig
  24 from n2vc.vca.connection_data import ConnectionData
  25 from osm_common.dbmongo import DbMongo, DbException
  26
  27 DB_NAME = "osm"
  28
  29
  30 class Store(abc.ABC):
  31     @abc.abstractmethod
  32     async def get_vca_connection_data(self, vca_id: str) -> ConnectionData:
  33         """
  34         Get VCA connection data
  35
  36         :param: vca_id: VCA ID
  37
  38         :returns: ConnectionData with the information of the database
  39         """
  40
  41     @abc.abstractmethod
  42     async def update_vca_endpoints(self, hosts: typing.List[str], vca_id: str):
  43         """
  44         Update VCA endpoints
  45
  46         :param: endpoints: List of endpoints to write in the database
  47         :param: vca_id: VCA ID
  48         """
  49
  50     @abc.abstractmethod
  51     async def get_vca_endpoints(self, vca_id: str = None) -> typing.List[str]:
  52         """
  53         Get list if VCA endpoints
  54
  55         :param: vca_id: VCA ID
  56
  57         :returns: List of endpoints
  58         """
  59
  60     @abc.abstractmethod
  61     async def get_vca_id(self, vim_id: str = None) -> str:
  62         """
  63         Get VCA id for a VIM account
  64
  65         :param: vim_id: Vim account ID
  66         """
  67
  68
  69 class DbMongoStore(Store):
  70     def __init__(self, db: DbMongo):
  71         """
  72         Constructor
  73
  74         :param: db: osm_common.dbmongo.DbMongo object
  75         """
  76         self.db = db
  77
  78     async def get_vca_connection_data(self, vca_id: str) -> ConnectionData:
  79         """
  80         Get VCA connection data
  81
  82         :param: vca_id: VCA ID
  83
  84         :returns: ConnectionData with the information of the database
  85         """
  86         data = self.db.get_one("vca", q_filter={"_id": vca_id})
  87         self.db.encrypt_decrypt_fields(
  88             data,
  89             "decrypt",
  90             ["secret", "cacert"],
  91             schema_version=data["schema_version"],
  92             salt=data["_id"],
  93         )
  94         return ConnectionData(**data)
  95
  96     async def update_vca_endpoints(
  97         self, endpoints: typing.List[str], vca_id: str = None
  98     ):
  99         """
 100         Update VCA endpoints
 101
 102         :param: endpoints: List of endpoints to write in the database
 103         :param: vca_id: VCA ID
 104         """
 105         if vca_id:
 106             data = self.db.get_one("vca", q_filter={"_id": vca_id})
 107             data["endpoints"] = endpoints
 108             self._update("vca", vca_id, data)
 109         else:
 110             # The default VCA. Data for the endpoints is in a different place
 111             juju_info = self._get_juju_info()
 112             # If it doesn't, then create it
 113             if not juju_info:
 114                 try:
 115                     self.db.create(
 116                         "vca",
 117                         {"_id": "juju"},
 118                     )
 119                 except DbException as e:
 120                     # Racing condition: check if another N2VC worker has created it
 121                     juju_info = self._get_juju_info()
 122                     if not juju_info:
 123                         raise e
 124             self.db.set_one(
 125                 "vca",
 126                 {"_id": "juju"},
 127                 {"api_endpoints": endpoints},
 128             )
 129
 130     async def get_vca_endpoints(self, vca_id: str = None) -> typing.List[str]:
 131         """
 132         Get list if VCA endpoints
 133
 134         :param: vca_id: VCA ID
 135
 136         :returns: List of endpoints
 137         """
 138         endpoints = []
 139         if vca_id:
 140             endpoints = self.get_vca_connection_data(vca_id).endpoints
 141         else:
 142             juju_info = self._get_juju_info()
 143             if juju_info and "api_endpoints" in juju_info:
 144                 endpoints = juju_info["api_endpoints"]
 145         return endpoints
 146
 147     async def get_vca_id(self, vim_id: str = None) -> str:
 148         """
 149         Get VCA ID from the database for a given VIM account ID
 150
 151         :param: vim_id: VIM account ID
 152         """
 153         return (
 154             self.db.get_one(
 155                 "vim_accounts",
 156                 q_filter={"_id": vim_id},
 157                 fail_on_empty=False,
 158             ).get("vca")
 159             if vim_id
 160             else None
 161         )
 162
 163     def _update(self, collection: str, id: str, data: dict):
 164         """
 165         Update object in database
 166
 167         :param: collection: Collection name
 168         :param: id: ID of the object
 169         :param: data: Object data
 170         """
 171         self.db.replace(
 172             collection,
 173             id,
 174             data,
 175         )
 176
 177     def _get_juju_info(self):
 178         """Get Juju information (the default VCA) from the admin collection"""
 179         return self.db.get_one(
 180             "vca",
 181             q_filter={"_id": "juju"},
 182             fail_on_empty=False,
 183         )
 184
 185
 186 class MotorStore(Store):
 187     def __init__(self, uri: str, loop=None):
 188         """
 189         Constructor
 190
 191         :param: uri: Connection string to connect to the database.
 192         :param: loop: Asyncio Loop
 193         """
 194         self._client = AsyncIOMotorClient(uri)
 195         self.loop = loop or asyncio.get_event_loop()
 196         self._secret_key = None
 197         self._config = EnvironConfig(prefixes=["OSMLCM_", "OSMMON_"])
 198
 199     @property
 200     def _database(self):
 201         return self._client[DB_NAME]
 202
 203     @property
 204     def _vca_collection(self):
 205         return self._database["vca"]
 206
 207     @property
 208     def _admin_collection(self):
 209         return self._database["admin"]
 210
 211     @property
 212     def _vim_accounts_collection(self):
 213         return self._database["vim_accounts"]
 214
 215     async def get_vca_connection_data(self, vca_id: str) -> ConnectionData:
 216         """
 217         Get VCA connection data
 218
 219         :param: vca_id: VCA ID
 220
 221         :returns: ConnectionData with the information of the database
 222         """
 223         data = await self._vca_collection.find_one({"_id": vca_id})
 224         if not data:
 225             raise Exception("vca with id {} not found".format(vca_id))
 226         await self.decrypt_fields(
 227             data,
 228             ["secret", "cacert"],
 229             schema_version=data["schema_version"],
 230             salt=data["_id"],
 231         )
 232         return ConnectionData(**data)
 233
 234     async def update_vca_endpoints(
 235         self, endpoints: typing.List[str], vca_id: str = None
 236     ):
 237         """
 238         Update VCA endpoints
 239
 240         :param: endpoints: List of endpoints to write in the database
 241         :param: vca_id: VCA ID
 242         """
 243         if vca_id:
 244             data = await self._vca_collection.find_one({"_id": vca_id})
 245             data["endpoints"] = endpoints
 246             await self._vca_collection.replace_one({"_id": vca_id}, data)
 247         else:
 248             # The default VCA. Data for the endpoints is in a different place
 249             juju_info = await self._get_juju_info()
 250             # If it doesn't, then create it
 251             if not juju_info:
 252                 try:
 253                     await self._admin_collection.insert_one({"_id": "juju"})
 254                 except Exception as e:
 255                     # Racing condition: check if another N2VC worker has created it
 256                     juju_info = await self._get_juju_info()
 257                     if not juju_info:
 258                         raise e
 259
 260             await self._admin_collection.replace_one(
 261                 {"_id": "juju"}, {"api_endpoints": endpoints}
 262             )
 263
 264     async def get_vca_endpoints(self, vca_id: str = None) -> typing.List[str]:
 265         """
 266         Get list if VCA endpoints
 267
 268         :param: vca_id: VCA ID
 269
 270         :returns: List of endpoints
 271         """
 272         endpoints = []
 273         if vca_id:
 274             endpoints = (await self.get_vca_connection_data(vca_id)).endpoints
 275         else:
 276             juju_info = await self._get_juju_info()
 277             if juju_info and "api_endpoints" in juju_info:
 278                 endpoints = juju_info["api_endpoints"]
 279         return endpoints
 280
 281     async def get_vca_id(self, vim_id: str = None) -> str:
 282         """
 283         Get VCA ID from the database for a given VIM account ID
 284
 285         :param: vim_id: VIM account ID
 286         """
 287         vca_id = None
 288         if vim_id:
 289             vim_account = await self._vim_accounts_collection.find_one({"_id": vim_id})
 290             if vim_account and "vca" in vim_account:
 291                 vca_id = vim_account["vca"]
 292         return vca_id
 293
 294     async def _get_juju_info(self):
 295         """Get Juju information (the default VCA) from the admin collection"""
 296         return await self._admin_collection.find_one({"_id": "juju"})
 297
 298     # DECRYPT METHODS
 299     async def decrypt_fields(
 300         self,
 301         item: dict,
 302         fields: typing.List[str],
 303         schema_version: str = None,
 304         salt: str = None,
 305     ):
 306         """
 307         Decrypt fields
 308
 309         Decrypt fields from a dictionary. Follows the same logic as in osm_common.
 310
 311         :param: item: Dictionary with the keys to be decrypted
 312         :param: fields: List of keys to decrypt
 313         :param: schema version: Schema version. (i.e. 1.11)
 314         :param: salt: Salt for the decryption
 315         """
 316         flags = re.I
 317
 318         async def process(_item):
 319             if isinstance(_item, list):
 320                 for elem in _item:
 321                     await process(elem)
 322             elif isinstance(_item, dict):
 323                 for key, val in _item.items():
 324                     if isinstance(val, str):
 325                         if any(re.search(f, key, flags) for f in fields):
 326                             _item[key] = await self.decrypt(val, schema_version, salt)
 327                     else:
 328                         await process(val)
 329
 330         await process(item)
 331
 332     async def decrypt(self, value, schema_version=None, salt=None):
 333         """
 334         Decrypt an encrypted value
 335         :param value: value to be decrypted. It is a base64 string
 336         :param schema_version: used for known encryption method used. If None or '1.0' no encryption has been done.
 337                If '1.1' symmetric AES encryption has been done
 338         :param salt: optional salt to be used
 339         :return: Plain content of value
 340         """
 341         if not await self.secret_key or not schema_version or schema_version == "1.0":
 342             return value
 343         else:
 344             secret_key = self._join_secret_key(salt)
 345             encrypted_msg = b64decode(value)
 346             cipher = AES.new(secret_key)
 347             decrypted_msg = cipher.decrypt(encrypted_msg)
 348             try:
 349                 unpadded_private_msg = decrypted_msg.decode().rstrip("\0")
 350             except UnicodeDecodeError:
 351                 raise DbException(
 352                     "Cannot decrypt information. Are you using same COMMONKEY in all OSM components?",
 353                     http_code=500,
 354                 )
 355             return unpadded_private_msg
 356
 357     def _join_secret_key(self, update_key: typing.Any):
 358         """
 359         Join secret key
 360
 361         :param: update_key: str or bytes with the to update
 362         """
 363         if isinstance(update_key, str):
 364             update_key_bytes = update_key.encode()
 365         else:
 366             update_key_bytes = update_key
 367         new_secret_key = (
 368             bytearray(self._secret_key) if self._secret_key else bytearray(32)
 369         )
 370         for i, b in enumerate(update_key_bytes):
 371             new_secret_key[i % 32] ^= b
 372         return bytes(new_secret_key)
 373
 374     @property
 375     async def secret_key(self):
 376         if self._secret_key:
 377             return self._secret_key
 378         else:
 379             if self.database_key:
 380                 self._secret_key = self._join_secret_key(self.database_key)
 381             version_data = await self._admin_collection.find_one({"_id": "version"})
 382             if version_data and version_data.get("serial"):
 383                 self._secret_key = self._join_secret_key(
 384                     b64decode(version_data["serial"])
 385                 )
 386             return self._secret_key
 387
 388     @property
 389     def database_key(self):
 390         return self._config["database_commonkey"]