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 time
  23 import getpass
  24 import requests
  25 import warnings
  26 from jproperties import Properties
  27 from typing import List
  28 warnings.filterwarnings('ignore', message='Unverified HTTPS request')
  29 # global configurations
  30
  31
  32 def get_environment_variable(name):
  33     configs = Properties()
  34     env_file = os.path.dirname(os.path.abspath(__file__)) + '/' + '../' + '.env'
  35     with open(env_file, "rb") as read_prop:
  36         configs.load(read_prop)
  37     return configs.get(name).data
  38
  39
  40 def load_arguments(args: List[str]) -> tuple:
  41     realm_file = os.path.dirname(os.path.abspath(
  42         __file__)) + '/o-ran-sc-realm.json'
  43     auth_file = os.path.dirname(os.path.abspath(
  44         __file__)) + '/authentication.json'
  45     ready_timeout = 180
  46     args.pop(0)
  47     while len(args) > 0:
  48         arg = args.pop(0)
  49         if arg == '--auth' and len(args) > 0:
  50             auth_file = args.pop(0)
  51             print('overwriting auth file: {}'.format(auth_file))
  52         elif arg == '--realm' and len(args) > 0:
  53             realm_file = args.pop(0)
  54             print('overwriting realm file: {}'.format(realm_file))
  55         elif arg == '--timeout' and len(args) > 0:
  56             ready_timeout = int(args.pop(0))
  57             print('waiting for ready {} seconds'.format(ready_timeout))
  58
  59     return (realm_file, auth_file, ready_timeout)
  60
  61
  62 def isReady(timeoutSeconds=180):
  63     url = getBaseUrl()
  64     while timeoutSeconds > 0:
  65         try:
  66             response = requests.get(url, verify=False, headers={})
  67         except:
  68             pass
  69         if response is not None and response.status_code == 200:
  70             return True
  71         time.sleep(1)
  72         timeoutSeconds -= 1
  73     return False
  74
  75
  76 def getBaseUrl():
  77     return get_environment_variable("IDENTITY_PROVIDER_URL")
  78
  79 # Request a token for further communication
  80 def getToken():
  81     url = base + '/realms/master/protocol/openid-connect/token'
  82     headers = {
  83         'content-type': 'application/x-www-form-urlencoded',
  84         'accept': 'application/json'
  85     }
  86     body = {
  87         'client_id': 'admin-cli',
  88         'grant_type': 'password',
  89         'username': username,
  90         'password': password
  91     }
  92     try:
  93         response = requests.post(url, verify=False, auth=(
  94             username, password), data=body, headers=headers)
  95     except requests.exceptions.Timeout:
  96         sys.exit('HTTP request failed, please check you internet connection.')
  97     except requests.exceptions.TooManyRedirects:
  98         sys.exit('HTTP request failed, please check your proxy settings.')
  99     except requests.exceptions.RequestException as e:
 100         # catastrophic error. bail.
 101         raise SystemExit(e)
 102
 103     if response.status_code >= 200 and response.status_code < 300:
 104         print('Got token!')
 105         return response.json()['access_token']
 106     else:
 107         sys.exit('Getting token failed.')
 108
 109 # create the default realm from file
 110
 111
 112 def createRealm(token, realm):
 113     url = base + '/admin/realms'
 114     auth = 'bearer ' + token
 115     headers = {
 116         'content-type': 'application/json',
 117         'accept': 'application/json',
 118         'authorization': auth
 119     }
 120     try:
 121         response = requests.post(
 122             url, verify=False, json=realm, headers=headers)
 123     except requests.exceptions.Timeout:
 124         sys.exit('HTTP request failed, please check you internet connection.')
 125     except requests.exceptions.TooManyRedirects:
 126         sys.exit('HTTP request failed, please check your proxy settings.')
 127     except requests.exceptions.RequestException as e:
 128         # catastrophic error. bail.
 129         raise SystemExit(e)
 130
 131     return response.status_code >= 200 and response.status_code < 300
 132
 133 # Check if default realm exists
 134
 135
 136 def checkRealmExists(token, realmId):
 137     url = base + '/admin/realms/' + realmId
 138     auth = 'bearer ' + token
 139     headers = {
 140         'accept': 'application/json',
 141         'authorization': auth
 142     }
 143     try:
 144         response = requests.get(url, verify=False, headers=headers)
 145     except requests.exceptions.Timeout:
 146         sys.exit('HTTP request failed, please check you internet connection.')
 147     except requests.exceptions.TooManyRedirects:
 148         sys.exit('HTTP request failed, please check your proxy settings.')
 149     except requests.exceptions.RequestException as e:
 150         # catastrophic error. bail.
 151         raise SystemExit(e)
 152
 153     if response.status_code >= 200 and response.status_code < 300:
 154         return realmId == response.json()['id']
 155     else:
 156         # sys.exit('Getting realm failed.')
 157         return False
 158
 159 # create a user in default realm
 160
 161
 162 def createUser(token, realmConfig, user):
 163     realmId = realmConfig['id']
 164     url = base + '/admin/realms/' + realmId + '/users'
 165     auth = 'bearer ' + token
 166     headers = {
 167         'accept': 'application/json',
 168         'authorization': auth
 169     }
 170     try:
 171         response = requests.post(url, verify=False, json=user, headers=headers)
 172     except requests.exceptions.Timeout:
 173         sys.exit('HTTP request failed, please check you internet connection.')
 174     except requests.exceptions.TooManyRedirects:
 175         sys.exit('HTTP request failed, please check your proxy settings.')
 176     except requests.exceptions.RequestException as e:
 177         # catastrophic error. bail.
 178         raise SystemExit(e)
 179
 180     if response.status_code >= 200 and response.status_code < 300:
 181         print('User', user['username'], 'created!')
 182     else:
 183         print('User creation', user['username'], 'failed!\n', response.text)
 184
 185 # creates User accounts in realm based a file
 186
 187
 188 def createUsers(token, realmConfig, authConfig):
 189     for user in authConfig['users']:
 190         createUser(token, realmConfig, user)
 191
 192     # create a user based on system user
 193     systemUser = {
 194         "firstName": getpass.getuser(),
 195         "lastName": "",
 196         "email": getpass.getuser() + "@sdnr.onap.org",
 197         "enabled": "true",
 198         "username": getpass.getuser(),
 199         "credentials": [
 200             {
 201                 "type": "password",
 202                 "value": password,
 203                 "temporary": False
 204             }
 205         ]
 206     }
 207     createUser(token, realmConfig, systemUser)
 208
 209 # Grants a role to a user
 210
 211
 212 def addUserRole(user: dict, role: dict, options: dict):
 213     url = options['url'] + '/' + user['id'] + '/role-mappings/realm'
 214     try:
 215         response = requests.post(url, verify=False, json=[
 216                                  {'id': role['id'], 'name':role['name']}], headers=options['headers'])
 217     except requests.exceptions.Timeout:
 218         sys.exit('HTTP request failed, please check you internet connection.')
 219     except requests.exceptions.TooManyRedirects:
 220         sys.exit('HTTP request failed, please check your proxy settings.')
 221     except requests.exceptions.RequestException as e:
 222         # catastrophic error. bail.
 223         raise SystemExit(e)
 224
 225     if response.status_code >= 200 and response.status_code < 300:
 226         print('User role', user['username'], role['name'], 'created!')
 227     else:
 228         print('Creation of user role',
 229               user['username'], role['name'], 'failed!\n', response.text)
 230
 231 # searches for the role of a given user
 232
 233
 234 def findRole(username: str, authConfig: dict, realmConfig: dict) -> dict:
 235     roleName = 'administration'
 236     for grant in authConfig['grants']:
 237         if grant['username'] == username:
 238             roleName = grant['role']
 239     for role in realmConfig['roles']['realm']:
 240         if role['name'] == roleName:
 241             return role
 242     return None
 243
 244 # adds roles to users
 245
 246
 247 def addUserRoles(token, realmConfig, authConfig):
 248     realmId = realmConfig['id']
 249     url = base + '/admin/realms/' + realmId + '/users'
 250     auth = 'bearer ' + token
 251     headers = {
 252         'content-type': 'application/json',
 253         'accept': 'application/json',
 254         'authorization': auth
 255     }
 256     try:
 257         response = requests.get(url, verify=False, headers=headers)
 258     except requests.exceptions.Timeout:
 259         sys.exit('HTTP request failed, please check you internet connection.')
 260     except requests.exceptions.TooManyRedirects:
 261         sys.exit('HTTP request failed, please check your proxy settings.')
 262     except requests.exceptions.RequestException as e:
 263         # catastrophic error. bail.
 264         raise SystemExit(e)
 265
 266     if response.status_code >= 200 and response.status_code < 300:
 267         users = response.json()
 268         options = {
 269             "url": url,
 270             "auth": auth,
 271             "headers": headers
 272         }
 273         for user in users:
 274             role = findRole(user['username'], authConfig, realmConfig)
 275             addUserRole(user, role, options)
 276     else:
 277         sys.exit('Getting users failed.')
 278
 279 # main
 280
 281
 282 (realmFile, authFile, readyTimeout) = load_arguments(sys.argv)
 283 username = get_environment_variable('ADMIN_USERNAME')
 284 password = get_environment_variable('ADMIN_PASSWORD')
 285 base = getBaseUrl()
 286 isReady(readyTimeout)
 287 token = getToken()
 288 if token:
 289     with open(realmFile) as file:
 290         realmConfig = json.load(file)
 291         if not checkRealmExists(token, realmConfig['id']):
 292             createRealm(token, realmConfig)
 293
 294         with open(authFile) as authConfig:
 295             authConfig = json.load(authConfig)
 296             createUsers(token, realmConfig, authConfig)
 297         addUserRoles(token, realmConfig, authConfig)
 298     exit(0)
 299 exit(1)