sms-client/python/onapsmsclient/onapsmsclient/__init__.py

   1 # Copyright 2018 Intel Corporation, Inc
   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 requests
  16 import requests.exceptions
  17
  18 try:
  19     from urllib.parse import urljoin
  20 except ImportError:
  21     from urlparse import urljoin
  22
  23 name = "onapsmsclient"
  24
  25
  26 class InvalidRequestException(Exception):
  27     pass
  28
  29
  30 class InternalServerError(Exception):
  31     pass
  32
  33
  34 class UnexpectedError(Exception):
  35     pass
  36
  37
  38 class Client(object):
  39     """Python Client for Secret Management Service"""
  40
  41     def __init__(self, url='http://localhost:10443', timeout=30, cacert=None):
  42         """Creates a new SMS client instance
  43
  44         Args:
  45             url (str): Base URL with port pointing to the SMS service
  46             timeout (int): Number of seconds before aborting the connection
  47             cacert (str): Path to the cacert that will be used to verify
  48                 If this is None, verify will be False and the server cert
  49                 is not verified by the client.
  50         Returns:
  51             A Client object which can be used to interact with SMS service
  52         """
  53
  54         self.base_url = url
  55         self.timeout = timeout
  56         self.cacert = cacert
  57         self.session = requests.Session()
  58
  59         self._base_api_url = '/v1/sms'
  60
  61     def _urlJoin(self, *urls):
  62         """Joins given urls into a single url
  63
  64         Args:
  65             urls (str): url fragments to be combined.
  66
  67         Returns:
  68             str: Joined URL
  69         """
  70
  71         return '/'.join(urls)
  72
  73     def _raiseException(self, statuscode, errors=None):
  74         """ Handles Exception Raising based on statusCode
  75
  76         Args:
  77             statuscode (int): status code returned by the server
  78             errors (str): list of strings containing error messages
  79
  80         Returns:
  81             exception: An exception is raised based on error message
  82         """
  83
  84         if statuscode == 400:
  85             raise InvalidRequestException(errors)
  86         if statuscode == 500:
  87             raise InternalServerError(errors)
  88
  89         raise UnexpectedError(errors)
  90
  91     def _request(self, method, url, headers=None, **kwargs):
  92         """Handles request for all the client methods
  93
  94         Args:
  95             method (str): type of HTTP method (get, post or delete).
  96             url (str): api URL.
  97             headers (dict): custom headers if any.
  98             **kwargs: various args supported by requests library
  99
 100         Returns:
 101             requests.Response: An object containing status_code and returned
 102                                json data is returned here.
 103         """
 104
 105         if headers is None:
 106             headers = {
 107                 'content-type': "application/json",
 108                 'Accept': "application/json"
 109             }
 110
 111         # Verify the server or not based on the cacert argument
 112         if self.cacert is None:
 113             verify = False
 114         else:
 115             verify = self.cacert
 116
 117         url = urljoin(self.base_url, url)
 118         response = self.session.request(method, url, headers=headers,
 119                                         allow_redirects=False, verify=verify,
 120                                         timeout=self.timeout, **kwargs)
 121
 122         errors = None
 123         if response.status_code >= 400 and response.status_code < 600:
 124             # Request Failed. Raise Exception.
 125             errors = response.text
 126             self._raiseException(response.status_code, errors)
 127
 128         return response
 129
 130     def getStatus(self):
 131         """Returns Status of SMS Service
 132
 133         Returns:
 134             bool: True or False depending on if SMS Service is Sealed or not.
 135                   True - Sealed
 136                   False - Unsealed and ready for operations
 137         """
 138
 139         url = self._urlJoin(self._base_api_url, 'quorum', 'status')
 140
 141         response = self._request('get', url)
 142         return response.json()['sealstatus']
 143
 144     def createDomain(self, domainName):
 145         """Creates a Secret Domain
 146
 147         Args:
 148             domainName (str): Name of the secret domain to create
 149
 150         Returns:
 151             string: UUID of the created domain name
 152         """
 153
 154         domainName = domainName.strip()
 155         data = {"name": domainName}
 156         url = self._urlJoin(self._base_api_url, 'domain')
 157
 158         response = self._request('post', url, json=data)
 159         return response.json()['uuid']
 160
 161     def deleteDomain(self, domainUUID):
 162         """Deletes a Secret Domain
 163
 164         Args:
 165             domainUUID (str): UUID of the secret domain to delete
 166
 167         Returns:
 168             bool: True. An exception will be raised if delete failed.
 169         """
 170
 171         domainUUID = domainUUID.strip()
 172         url = self._urlJoin(self._base_api_url, 'domain', domainUUID)
 173
 174         self._request('delete', url)
 175         return True
 176
 177     def getSecretNames(self, domainUUID):
 178         """Get all Secret Names in Domain
 179
 180         Args:
 181             domainUUID (str): UUID of the secret domain
 182
 183         Returns:
 184             string[]: List of strings each corresponding to a
 185                       Secret's Name in this Domain.
 186         """
 187
 188         domainUUID = domainUUID.strip()
 189         url = self._urlJoin(self._base_api_url, 'domain', domainUUID,
 190                             'secret')
 191
 192         response = self._request('get', url)
 193         return response.json()['secretnames']
 194
 195     def storeSecret(self, domainUUID, secretName, values):
 196         """Store a Secret in given Domain
 197
 198         Args:
 199             domainUUID (str): UUID of the secret domain
 200             secretName (str): Name for the Secret
 201             values (dict): A dict containing name-value pairs which
 202                            form the secret
 203
 204         Returns:
 205             bool: True. An exception will be raised if store failed.
 206         """
 207
 208         domainUUID = domainUUID.strip()
 209         secretName = secretName.strip()
 210         url = self._urlJoin(self._base_api_url, 'domain', domainUUID,
 211                             'secret')
 212
 213         if not isinstance(values, dict):
 214             raise TypeError('Input values is not a dictionary')
 215
 216         data = {"name": secretName, "values": values}
 217         self._request('post', url, json=data)
 218         return True
 219
 220     def getSecret(self, domainUUID, secretName):
 221         """Get a particular Secret from Domain.
 222
 223         Args:
 224             domainUUID (str): UUID of the secret domain
 225             secretName (str): Name of the secret
 226
 227         Returns:
 228             dict: dictionary containing the name-value pairs
 229                   which form the secret
 230         """
 231
 232         domainUUID = domainUUID.strip()
 233         secretName = secretName.strip()
 234         url = self._urlJoin(self._base_api_url, 'domain', domainUUID,
 235                             'secret', secretName)
 236
 237         response = self._request('get', url)
 238         return response.json()['values']
 239
 240     def deleteSecret(self, domainUUID, secretName):
 241         """Delete a particular Secret from Domain.
 242
 243         Args:
 244             domainUUID (str): UUID of the secret domain
 245             secretName (str): Name of the secret
 246
 247         Returns:
 248             bool: True. An exception will be raised if delete failed.
 249         """
 250
 251         domainUUID = domainUUID.strip()
 252         secretName = secretName.strip()
 253         url = self._urlJoin(self._base_api_url, 'domain', domainUUID,
 254                             'secret', secretName)
 255
 256         self._request('delete', url)
 257         return True