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