import requests,os # Exceptions class InvalidCredentials(Exception): """Raise when auth fails""" pass class AccountNotExist(Exception): """Raise when not account subscription assigned""" pass class UserNotExist(Exception): """Raise when user is'nt already exists""" pass class CustomException(Exception): """Raise when user it`s already exists""" def __init__(self, message, payload=None): self.message = message self.payload = payload # you could add more args def __str__(self): return str(self.message) #Basic Auth USER_NAME= os.getenv("FORTANIX_USER", "yonatan.hernandez@triara.com") PASSWORD= os.getenv("FORTANIX_PASS", "Triara.2023") #USER_NAME= os.getenv("FORTANIX_USER", "yonatan.hernandez@triara.com") #PASSWORD= os.getenv("FORTANIX_PASS", "!2Fqa7wK7YcQyvg*QJgH") #USER_NAME= os.getenv("FORTANIX_USER", "YHALVARA@uninet.com.mx") #PASSWORD= os.getenv("FORTANIX_PASS", "Nale1408*") TOKEN='' HEADERS={} ACCOUNT_ID = os.getenv("ACCOUNT", "43b1eae3-27f8-4a5f-adf6-7afa9e4ddd9e") ACCOUNTS_LIST=dict() ACCOUNT_ROLE = os.getenv("ACCOUNT_ROLE", "83533bca-70a0-4a7d-96e0-f37e4e266e95") # DEFAULTROLE: "ACCOUNTMEMBER" GROUP_ROLE = os.getenv("GROUP_ROLE", "98feb5b2-0dec-4abf-910c-1030c1951907") # DEFAULTROLE: "GROUPADMINISTRATOR" #URL's and Sessions SESSION_AUTH=requests.session() #BASE_URL='https://apps.smartkey.io/' BASE_URL=os.getenv("BASE_URL",'https://dsm.net.telmex.com/') REFRESH_URL=BASE_URL+'sys/v1/session/refresh' AUTH_URL=BASE_URL+'sys/v1/session/auth' GROUPS_URL=BASE_URL+'sys/v1/groups' APPS_URL=BASE_URL+'sys/v1/apps' USERS_URL=BASE_URL+'sys/v1/users' ACCOUNTS_URL=BASE_URL+'sys/v1/accounts' CHILD_AC_URL=BASE_URL+'sys/v1/accounts' SELECT_ACCOUNT=BASE_URL+'sys/v1/session/select_account' ROLES_URL=BASE_URL+'sys/v1/roles' # ACCOUNT_ROLE ACCOUNT_PERMISSIONS = [ # "MANAGE_LOGGING", # "MANAGE_AUTH", # "MANAGE_WORKSPACE_CSE", # "UNWRAP_WORKSPACE_CSE_PRIVILEGED", # "MANAGE_ACCOUNT_CLIENT_CONFIGS", # "CREATE_ACCOUNT_APPROVAL_POLICY", # "SET_APPROVAL_REQUEST_EXPIRY", # "UPDATE_ACCOUNT_CUSTOM_METADATA_ATTRIBUTES", # "MANAGE_ACCOUNT_SUBSCRIPTION", # "MANAGE_ACCOUNT_PROFILE", # "DELETE_ACCOUNT", # "CREATE_ADMIN_APPS", # "UPDATE_ADMIN_APPS", # "DELETE_ADMIN_APPS", # "RETRIEVE_ADMIN_APP_SECRETS", # "MANAGE_ADMIN_APPS", # "CREATE_CUSTOM_ROLES", # "UPDATE_CUSTOM_ROLES", # "DELETE_CUSTOM_ROLES", # "MANAGE_CUSTOM_ROLES", "INVITE_USERS_TO_ACCOUNT", "DELETE_USERS_FROM_ACCOUNT", # "UPDATE_USERS_ACCOUNT_ROLE", # "UPDATE_USERS_ACCOUNT_ENABLED_STATE", "MANAGE_ACCOUNT_USERS", # "CREATE_EXTERNAL_ROLES", # "SYNC_EXTERNAL_ROLES", # "DELETE_EXTERNAL_ROLES", # "MANAGE_EXTERNAL_ROLES", "CREATE_ACCOUNT_SOBJECT_POLICIES", "UPDATE_ACCOUNT_SOBJECT_POLICIES", "DELETE_ACCOUNT_SOBJECT_POLICIES", "MANAGE_ACCOUNT_SOBJECT_POLICIES", # "CREATE_CHILD_ACCOUNTS", # "UPDATE_CHILD_ACCOUNTS", # "DELETE_CHILD_ACCOUNTS", # "CREATE_CHILD_ACCOUNT_USERS", # "GET_CHILD_ACCOUNTS", # "GET_CHILD_ACCOUNT_USERS", # "MANAGE_CHILD_ACCOUNTS", # "CREATE_LOCAL_GROUPS", # "CREATE_EXTERNAL_GROUPS", # "ALLOW_QUORUM_REVIEWER", # "ALLOW_KEY_CUSTODIAN", "GET_ADMIN_APPS", "GET_ALL_APPROVAL_REQUESTS", "GET_CUSTOM_ROLES", "GET_EXTERNAL_ROLES", "GET_ALL_USERS", "GET_ACCOUNT_USAGE" ] #GROUP_ROLE GROUP_PERMISSIONS=[ # "CREATE_GROUP_APPROVAL_POLICY", # "UPDATE_GROUP_EXTERNAL_LINKS", # "MANAGE_GROUP_CLIENT_CONFIGS", # "UPDATE_GROUP_PROFILE", # "DELETE_GROUP", # "MAP_EXTERNAL_ROLES_FOR_APPS", # "MAP_EXTERNAL_ROLES_FOR_USERS", # "MAP_EXTERNAL_ROLES", "ADD_USERS_TO_GROUP", "DELETE_USERS_FROM_GROUP", # "UPDATE_USERS_GROUP_ROLE", "MANAGE_GROUP_USERS", # "MANAGE_GROUP_WRAPPING_KEY", "CREATE_GROUP_SOBJECT_POLICIES", "UPDATE_GROUP_SOBJECT_POLICIES", "DELETE_GROUP_SOBJECT_POLICIES", "MANAGE_GROUP_SOBJECT_POLICIES", # "CREATE_GROUP_CUSTODIAN_POLICY", # "UPDATE_GROUP_CUSTODIAN_POLICY", # "DELETE_GROUP_CUSTODIAN_POLICY", # "MANAGE_GROUP_CUSTODIAN_POLICY", "CREATE_APPS", "UPDATE_APPS", "RETRIEVE_APP_SECRETS", # "DELETE_APPS", # "MANAGE_APPS", "CREATE_PLUGINS", "UPDATE_PLUGINS", # "INVOKE_PLUGINS", # "DELETE_PLUGINS", "MANAGE_PLUGINS", "CREATE_SOBJECTS", "EXPORT_SOBJECTS", "COPY_SOBJECTS", "WRAP_SOBJECTS", "UNWRAP_SOBJECTS", "UPDATE_SOBJECTS_ENABLED_STATE", "ROTATE_SOBJECTS", "DELETE_SOBJECTS", "DESTROY_SOBJECTS", "REVOKE_SOBJECTS", "ACTIVATE_SOBJECTS", "REVERT_SOBJECTS", "DELETE_KEY_MATERIAL", "MOVE_SOBJECTS", "UPDATE_KEY_OPS", "UPDATE_SOBJECT_POLICIES", "UPDATE_SOBJECTS_PROFILE", "SCAN_EXTERNAL_SOBJECTS", "RESTORE_EXTERNAL_SOBJECTS", "DERIVE_SOBJECTS", "TRANSFORM_SOBJECTS", # "WRAP_WORKSPACE_CSE", # "UNWRAP_WORKSPACE_CSE", # "WORKSPACE_CSE", "GET_GROUP", "GET_SOBJECTS", "GET_APPS", "GET_PLUGINS", "GET_GROUP_APPROVAL_REQUESTS", "GET_AUDIT_LOGS" ] #FLAGS ADD_GROUPS="add_groups" DEL_GROUPS="del_groups" # LIMITS OF API QUERY LIMIT=100 # Error def return_error(error:str,type_of_resource:str): return_data=[] tmp_ele=dict() tmp_ele['msg'] = error tmp_ele['id'] = None tmp_ele['name'] = None tmp_ele['created_at'] = None tmp_ele['type_of_resource'] = type_of_resource return_data.append(tmp_ele) return return_data # Auth def authenticate(user:str=USER_NAME,password:str=PASSWORD): global TOKEN,HEADERS,SESSION_AUTH try: response=requests.post(AUTH_URL,auth=(user,password)) if response.status_code != 200: raise InvalidCredentials else: data=response.json() TOKEN=data['access_token'] HEADERS= {"Accept":"application/json", "Authorization": f"Bearer {TOKEN}"} select_account() except InvalidCredentials: print('error on credentials') def select_account(id:str=ACCOUNT_ID): global SELECT_ACCOUNT try: response=SESSION_AUTH.post(SELECT_ACCOUNT,json={'acct_id':id},headers=HEADERS) if response.status_code != 200: raise AccountNotExist except AccountNotExist: print('account not exist') def refresh_session(): try: response=SESSION_AUTH.post(REFRESH_URL,headers=HEADERS) if response.status_code != 204: authenticate() except Exception as error: print(error.message) # Groups"group" def get_groups(acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'group') select_account(id=acct_id) response=SESSION_AUTH.get(GROUPS_URL,headers=HEADERS) return response.json() def get_apps_data_by_group(group_id:str,acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'apps') select_account(id=acct_id) return_data=[];offset=0;has_more=True;limit=LIMIT try: while has_more: params={'group_id':group_id,"offset":offset,"limit":limit} response=SESSION_AUTH.get(APPS_URL,params=params,headers=HEADERS) if response.status_code != 200: raise CustomException(response.text) tmp_data=response.json() for ele in tmp_data: tmp_ele=dict() tmp_ele['msg'] = "done" tmp_ele['id'] = ele['app_id'] tmp_ele['name'] = ele['name'] tmp_ele['created_at'] = ele['created_at'] tmp_ele['type_of_resource'] = "app" return_data.append(tmp_ele) if len(tmp_data) == limit: offset+=limit else: has_more=False return return_data except CustomException as error: print(error.message) def create_group(group_name:str,group_desc:str="",acct_id:str=""): refresh_session() select_account(id=acct_id) return_data = [] try: data={ "name": group_name, "description": group_desc, "acct_id": acct_id } response=SESSION_AUTH.post(GROUPS_URL,json=data,headers=HEADERS) if response.status_code != 201: raise CustomException(response.text) else: print(f"group '{group_name}' created") group_data = response.json() tmp_ele=dict() tmp_ele['msg'] = "done" tmp_ele['id'] = group_data['group_id'] tmp_ele['name'] = group_data['name'] tmp_ele['created_at'] = group_data['created_at'] tmp_ele['type_of_resource'] = "group" return_data.append(tmp_ele) return return_data except CustomException as error: return return_error(error.message, 'group') def delete_group(group_id:str="",group_name:str="",force=False,acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'group') select_account(id=acct_id) return_data=dict() try: if not group_id and not group_name: raise CustomException("Not parameters sended") elif not group_id: group_id = get_group_by_name(group_name=group_name) elif not group_name: group_response = SESSION_AUTH.get(GROUPS_URL+f"/{group_id}",headers=HEADERS) if group_response.status_code != 200: raise CustomException(group_response.text) else: group_data=group_response.json() group_name=group_data['name'] if force: apps=get_apps_data_by_group(group_id=group_id) for app in apps: response=SESSION_AUTH.delete(APPS_URL+f"/{app['id']}",headers=HEADERS) if response.status_code != 204: raise CustomException(response.text) else: print(f"App '{app['name']}' deleted") response=SESSION_AUTH.delete(GROUPS_URL+f"/{group_id}",headers=HEADERS) if response.status_code != 204: raise CustomException(response.text) else: print(f"group '{group_name}' deleted") return_data['msg'] = f'group {group_name} deleted' return_data['done'] = True return return_data except CustomException as error: return return_error(error.message, 'group') def get_group_by_name(group_name:str,acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'group') select_account(id=acct_id) try: groups=get_groups() res = list(filter(lambda groups: groups['name'] == group_name, groups)) if len(res) > 0: return res[0]['group_id'] else: raise CustomException("group not found") except CustomException as error: print(error.message) def create_group_role(role_name:str,role_desc:str,acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'group_role') select_account(id=acct_id) return_data = [] try: data={ "name": role_name, "description": role_desc, "kind": "group", "details": { "exclusive": False, "kind": "group", "permissions": GROUP_PERMISSIONS } } response=SESSION_AUTH.post(ROLES_URL,json=data,headers=HEADERS) if response.status_code != 201: raise CustomException(response.text) else: print(f"group role '{role_name}' created") role_data = response.json() tmp_ele=dict() tmp_ele['msg'] = 'done' tmp_ele['id'] = role_data['role_id'] tmp_ele['name'] = role_data['name'] tmp_ele['created_at'] = role_data['created_at'] tmp_ele['type_of_resource'] = "group_role" return_data.append(tmp_ele) return return_data except CustomException as error: return return_error(error.message, 'group_role') # Users def get_users(acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'user') select_account(id=acct_id) data=[] offset=0 has_more=True limit=LIMIT try: while has_more: params={"offset":offset,"limit":limit} response=SESSION_AUTH.get(USERS_URL,headers=HEADERS) if response.status_code != 200: raise CustomException(response.text) tmp_data=response.json() data+=tmp_data if len(tmp_data) == limit: offset+=limit else: has_more=False return data except CustomException as error: return return_error(error.message, 'user') def create_user(email:str,first_name:str="",last_name:str="",group_ids:set={},account_role:str=ACCOUNT_ROLE,group_role:str=GROUP_ROLE,acct_id:str=ACCOUNT_ID): refresh_session() if not acct_id: return return_error('not account selected', 'user') select_account(id=acct_id) return_data = [] data={ "user_email": email, "account_role": [account_role], } try: if len(group_ids) == 0: raise CustomException("group ids are empty") groups=dict() for gid in group_ids: groups[gid]=[group_role] response=SESSION_AUTH.post(USERS_URL+"/invite",json=data,headers=HEADERS) if response.status_code == 409: raise CustomException("this user already exist") else: user_data=response.json() SESSION_AUTH.patch(USERS_URL+f"/{user_data['user_id']}",params={ 'user_id':user_data['user_id'], 'first_name':first_name, #Not valid, set manually by the user 'last_name':last_name #Not valid, set manually by the user },headers=HEADERS) user_mod_group(user_id=user_data['user_id'],groups=groups,action=ADD_GROUPS,acct_id=acct_id) tmp_ele=dict() tmp_ele['msg'] = 'done' tmp_ele['id'] = user_data['user_id'] tmp_ele['name'] = user_data['user_email'] tmp_ele['created_at'] = None tmp_ele['type_of_resource'] = 'user' return_data.append(tmp_ele) return return_data except CustomException as error: return return_error(error.message, 'user') def delete_user(user_id:str,acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'user') select_account(id=acct_id) select_account(id=acct_id) return_data=dict() try: respose=SESSION_AUTH.get(USERS_URL+f"/{user_id}",headers=HEADERS) if respose.status_code != 200: raise CustomException("user not exist") else: user_data=respose.json() groups= user_data["groups"] if len(groups) > 0: user_mod_group(user_id=user_id,groups=groups,action=DEL_GROUPS) del_usr=SESSION_AUTH.delete(USERS_URL+f"/{user_id}/accounts",headers=HEADERS) if del_usr.status_code != 204: raise CustomException(del_usr.text) else: return_data['msg'] = "user deleted" return_data['done'] = True return return_data except CustomException as error: print(error.message) return return_error(error.message, 'user') def user_mod_group(user_id:str,groups:dict,action:str,acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'user') select_account(id=acct_id) try: response=SESSION_AUTH.patch(USERS_URL+f"/{user_id}",json={action:groups},headers=HEADERS) if response.status_code != 200: raise CustomException(response.text) except CustomException as error: print(error.message) def get_user_by_email(user_email:str,acct_id:str=""): refresh_session() if not acct_id: return return_error('not account selected', 'user') select_account(id=acct_id) try: users=get_users() res = list(filter(lambda users: users['user_email'] == user_email, users)) if len(res) > 0: return res[0]['user_id'] else: raise CustomException("user not found") except CustomException as error: print(error.message) #Accounts def create_account(acct_name:str): refresh_session() select_account() global ACCOUNTS_LIST return_data = [] tmp_elem = dict() #return_data['res']=dict() try: data={ "name": acct_name, } response=SESSION_AUTH.post(CHILD_AC_URL,json=data,headers=HEADERS) if response.status_code != 201: raise CustomException(response.text) else: print(f"account '{acct_name}' created") acct_data = response.json() tmp_elem['msg'] = 'done' tmp_elem['id'] = acct_data['acct_id'] tmp_elem['name'] = acct_data['name'] tmp_elem['created_at'] = acct_data['created_at'] tmp_elem['type_of_resource'] = 'account' return_data.append(tmp_elem) return return_data except CustomException as error: return return_error(error.message, 'account') def delete_accts_by_name(acct_name:str=""): global SESSION_AUTH refresh_session() response=SESSION_AUTH.get(ACCOUNTS_URL,headers=HEADERS) data = response.json() res = list(filter(lambda data: data['name'] == acct_name, data)) for r in res: groups=get_groups(acct_id=r['acct_id']) for g in groups: delete_group(group_id=g['group_id'],acct_id=r['acct_id'],force=True) select_account(id=r['acct_id']) response=SESSION_AUTH.delete(CHILD_AC_URL+f"/{r['acct_id']}",headers=HEADERS) if response.status_code != 204: raise CustomException(response.text) def create_account_role(role_name:str,role_desc:str,acct_id:str=""): refresh_session() return_data =[] if not acct_id: return return_error('not account selected', 'account_role') select_account(id=acct_id) try: data={ "name": role_name, "description": role_desc, "kind": "account", "details": { "exclusive": False, "kind": "account", "permissions": ACCOUNT_PERMISSIONS } } response=SESSION_AUTH.post(ROLES_URL,json=data,headers=HEADERS) if response.status_code != 201: raise CustomException(response.text) else: print(f"account role '{role_name}' created") role_data = response.json() tmp_ele=dict() tmp_ele['msg'] = 'done' tmp_ele['id'] = role_data['role_id'] tmp_ele['name'] = role_data['name'] tmp_ele['created_at'] = role_data['created_at'] tmp_ele['type_of_resource'] = 'account_role' return_data.append(tmp_ele) return return_data except CustomException as error: return return_error(error.message, 'account_role') def delete_app(app_id:str="",acct_id:str=""): refresh_session() return_data =[] if not acct_id: return return_error('not account selected', 'account_role') select_account(id=acct_id) try: response=SESSION_AUTH.delete(APPS_URL+f"/{app_id}",headers=HEADERS) if response.status_code != 204: raise CustomException(response.text) else: tmp_ele=dict() tmp_ele['msg'] = 'deleted' tmp_ele['id'] = app_id tmp_ele['name'] = None tmp_ele['created_at'] = None tmp_ele['type_of_resource'] = 'app' return_data.append(tmp_ele) return return_data except CustomException as error: return return_error(error.message, 'account_role') if __name__ == '__main__': #Auth Process #authenticate() #End Auth Process #print(get_apps_data_by_group(group_id=get_group_by_name("Demo_group_API"))) # account=create_account(acct_name="Bimbo") # ac_role=create_account_role(role_desc="Role API test",role_name="Site Admin") # g_role=create_group_role(role_desc="Role API test",role_name="App creator") # group=create_group(acct_id=ac_role['msg'],group_name="App Container",group_desc="Testing") # create_user(email='YHALVARA@uninet.com.mx',first_name="Yon",last_name="HA",group_ids={group['msg']},account_role=ac_role['msg'],group_role=g_role['msg']) apps=get_apps_data_by_group(acct_id=ACCOUNT_ID,group_id="9b356a2d-c941-443e-aef6-caae65e18e9c") print(apps) #delete_user(user_id="8fc23596-37cf-4201-8f74-e94441712f1c") #delete_group(group_name="Demo_group_API",force=True) #delete_accts_by_name(acct_name="Demo Micro") exit() # data=get_groups() # for item in data: # apps=get_apps_data_by_group(item['group_id']) # if len(apps) != 0: # print('Apps in '+item['name']) # for app in apps: # response=SESSION_AUTH.get(USERS_URL+f'/{item["creator"]["user"]}',headers=HEADERS) # user_data=response.json() # print(f'App: {app["name"]}|Creator: {user_data["user_email"]}') # print(user_data)