solution/integration/smo/common/identity/config.py

   1 #!/usr/bin/env python
   2 ################################################################################
   3 # Copyright 2021 highstreet technologies GmbH
   4 #
   5 # Licensed under the Apache License, Version 2.0 (the 'License');
   6 # you may not use this file except in compliance with the License.
   7 # You may obtain a copy of the License at
   8 #
   9 #     http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 # Unless required by applicable law or agreed to in writing, software
  12 # distributed under the License is distributed on an 'AS IS' BASIS,
  13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 # See the License for the specific language governing permissions and
  15 # limitations under the License.
  16 #
  17
  18 # importing the sys, json, requests library
  19 import os
  20 import sys
  21 import json
  22 import requests
  23 import getpass
  24
  25 # global configurations
  26 # TODO: read from ../.env
  27 base = 'http://localhost:8081'
  28 username = 'admin'
  29 password = 'Kp8bJ4SXszM0WXlhak3eHlcse2gAw84vaoGGmJvUy2U'
  30 realmFile = os.path.dirname(os.path.abspath(__file__)) + '/o-ran-sc-realm.json'
  31 authFile = os.path.dirname(os.path.abspath(__file__)) + '/authentication.json'
  32
  33 # Request a token for futher communication
  34 def getToken():
  35     url = base + '/auth/realms/master/protocol/openid-connect/token'
  36     headers = {
  37         'content-type': 'application/x-www-form-urlencoded',
  38         'accept': 'application/json'
  39     }
  40     body = {
  41       'client_id':'admin-cli',
  42       'grant_type': 'password',
  43       'username': username,
  44       'password': password
  45     }
  46     try:
  47       response = requests.post(url, verify=False, auth=(username, password), data=body, headers=headers)
  48     except requests.exceptions.Timeout:
  49       sys.exit('HTTP request failed, please check you internet connection.')
  50     except requests.exceptions.TooManyRedirects:
  51       sys.exit('HTTP request failed, please check your proxy settings.')
  52     except requests.exceptions.RequestException as e:
  53       # catastrophic error. bail.
  54       raise SystemExit(e)
  55
  56     if response.status_code >= 200 and response.status_code < 300:
  57       print('Got tocken!')
  58       return response.json()['access_token']
  59     else:
  60       sys.exit('Getting token failed.')
  61
  62 # create the default realm from file
  63 def createRealm(token, realm):
  64   url = base + '/auth/admin/realms'
  65   auth = 'bearer ' + token
  66   headers = {
  67       'content-type': 'application/json',
  68       'accept': 'application/json',
  69       'authorization': auth
  70   }
  71   try:
  72     response = requests.post(url, verify=False, json=realm, headers=headers)
  73   except requests.exceptions.Timeout:
  74     sys.exit('HTTP request failed, please check you internet connection.')
  75   except requests.exceptions.TooManyRedirects:
  76     sys.exit('HTTP request failed, please check your proxy settings.')
  77   except requests.exceptions.RequestException as e:
  78     # catastrophic error. bail.
  79     raise SystemExit(e)
  80
  81   return response.status_code >= 200 and response.status_code < 300
  82
  83 # Check if default realm exists
  84 def checkRealmExists(token, realmId):
  85   url = base + '/auth/admin/realms/' + realmId
  86   auth = 'bearer ' + token
  87   headers = {
  88       'accept': 'application/json',
  89       'authorization': auth
  90   }
  91   try:
  92       response = requests.get(url, verify=False, headers=headers)
  93   except requests.exceptions.Timeout:
  94       sys.exit('HTTP request failed, please check you internet connection.')
  95   except requests.exceptions.TooManyRedirects:
  96       sys.exit('HTTP request failed, please check your proxy settings.')
  97   except requests.exceptions.RequestException as e:
  98       # catastrophic error. bail.
  99       raise SystemExit(e)
 100
 101   if response.status_code >= 200 and response.status_code < 300:
 102     return realmId == response.json()['id']
 103   else:
 104     # sys.exit('Getting realm failed.')
 105     return False
 106
 107 # create a user in default realm
 108 def createUser(token, realmId, user):
 109   url = base + '/auth/admin/realms/' + realmId + '/users'
 110   auth = 'bearer ' + token
 111   headers = {
 112       'accept': 'application/json',
 113       'authorization': auth
 114   }
 115   try:
 116     response = requests.post(url, verify=False, json=user, headers=headers)
 117   except requests.exceptions.Timeout:
 118     sys.exit('HTTP request failed, please check you internet connection.')
 119   except requests.exceptions.TooManyRedirects:
 120     sys.exit('HTTP request failed, please check your proxy settings.')
 121   except requests.exceptions.RequestException as e:
 122     # catastrophic error. bail.
 123     raise SystemExit(e)
 124
 125   if response.status_code >= 200 and response.status_code < 300:
 126     print('User', user['username'], 'created!')
 127   else:
 128     print('User creation', user['username'], 'failed!\n', response.text)
 129
 130 # creates User accounts in realm based a file
 131 def createUsers(token, realm, authConfig):
 132   for user in authConfig['users']:
 133     createUser(token, realm, user)
 134
 135   # create a user based on system user
 136   systemUser = {
 137     "firstName": getpass.getuser(),
 138     "lastName": "",
 139     "email": getpass.getuser() + "@sdnr.onap.org",
 140     "enabled": "true",
 141     "username": getpass.getuser(),
 142     "credentials": [
 143       {
 144         "type": "password",
 145         "value": password,
 146         "temporary": False
 147       }
 148     ]
 149   }
 150   createUser(token, realm, systemUser)
 151
 152 # Grants a role to a user
 153 def addUserRole(user, role, options):
 154   url = options['url'] + '/' + user['id'] + '/role-mappings/realm'
 155   try:
 156     response = requests.post(url, verify=False, json=role, headers=options['headers'])
 157   except requests.exceptions.Timeout:
 158     sys.exit('HTTP request failed, please check you internet connection.')
 159   except requests.exceptions.TooManyRedirects:
 160     sys.exit('HTTP request failed, please check your proxy settings.')
 161   except requests.exceptions.RequestException as e:
 162     # catastrophic error. bail.
 163     raise SystemExit(e)
 164
 165   if response.status_code >= 200 and response.status_code < 300:
 166     print('User role', user['username'], role[0]['name'], 'created!')
 167   else:
 168     print('Creation of user role', user['username'], role[0]['name'], 'failed!\n', response.text)
 169
 170 # searches for the role of a given user
 171 def findRole(user, authConfig):
 172   roleName='administration'
 173   for grant in authConfig['grants']:
 174     if grant['username'] == user:
 175       roleName= grant['role']
 176   role=authConfig['roles'][roleName]
 177   return role
 178
 179 # adds roles to users
 180 def addUserRoles(token, realmId, authConfig):
 181   url = base + '/auth/admin/realms/' + realmId + '/users'
 182   auth = 'bearer ' + token
 183   headers = {
 184       'content-type': 'application/json',
 185       'accept': 'application/json',
 186       'authorization': auth
 187   }
 188   try:
 189       response = requests.get(url, verify=False, headers=headers)
 190   except requests.exceptions.Timeout:
 191       sys.exit('HTTP request failed, please check you internet connection.')
 192   except requests.exceptions.TooManyRedirects:
 193       sys.exit('HTTP request failed, please check your proxy settings.')
 194   except requests.exceptions.RequestException as e:
 195       # catastrophic error. bail.
 196       raise SystemExit(e)
 197
 198   if response.status_code >= 200 and response.status_code < 300:
 199     users = response.json()
 200     options = {
 201       "url": url,
 202       "auth": auth,
 203       "headers": headers
 204     }
 205     for user in users:
 206       role=findRole(user['username'], authConfig)
 207       addUserRole(user, role, options)
 208   else:
 209     sys.exit('Getting users failed.')
 210
 211 # main
 212 token = getToken()
 213 if token:
 214   with open(realmFile) as file:
 215     realm = json.load(file)
 216     if not checkRealmExists(token, realm['id']):
 217       createRealm(token, realm)
 218
 219     with open(authFile) as authConfig:
 220       auth = json.load(authConfig)
 221       createUsers(token, realm['id'], auth);
 222     addUserRoles(token, realm['id'], auth)