Use Custos Python SDK

Custos Python SDK  is for integration engineers who are seeking for Python clients to integrate with Custos hosted services. Custos SDK is built by wrapping gRPC Python stubs that are generated to connect with Custos services [1] with additional simplicity to deal with the APIs.

   Setting Up Project

•  Register your project with Custos as a Tenant using Custos Admin Portal and obtain Custos Id and Custos Secret.

•  Install    SDK into your python project virtual environment.                                    

  pip install custos-sdk==1.0.2

• Create settings.ini file with the following content and replace keys accordingly

  [CustosServer] SERVER_HOST = custos.scigap.org SERVER_SSL_PORT = 10000 CLIENT_ID = custos-XXXXXX-YYYY CLIENT_SEC = XzvvUuH434lWbCRSAnasasxasxasaxasxas

Next, we will go through some sample use cases to get familiar with SDK 

User Management Use Cases

    Following code snippet shows 

• User registration
• User enabling
• Fetch user
• Update user profile

• Find user  

  import os from custos.clients.user_management_client import UserManagementClient from custos.transport.settings import CustosServerClientSettings import custos.clients.utils.utilities as utl # load root directoty BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) #get settings file path (settings file path reside in configs folder under home directory) settings_path = os.path.join(BASE_DIR, 'configs', "settings.ini") # read settings custos_settings = CustosServerClientSettings(configuration_file_location=settings_path) # create custos user management client user_management_client = UserManagementClient(custos_settings) # obtain base 64 encoded token for tenant b64_encoded_custos_token = utl.get_token(custos_settings=custos_settings) def register_user(): response = user_management_client.register_user(token=b64_encoded_custos_token, username="TestUser4", first_name="Watson", last_name="Christe", password="1234", email="wat@gmail.com", is_temp_password=False) print(response) def enable_user(): response = user_management_client.enable_user(token=b64_encoded_custos_token, username="TestUser4") print(response) def get_user(): response = user_management_client.get_user(token=b64_encoded_custos_token, username="TestUser4") print(response) def update_user(): response = user_management_client.update_user_profile(token=b64_encoded_custos_token, username="TestUser4", first_name="Jimmy", last_name="Jhon", email="wat@gmail.com") print(response) def find_users(): response = user_management_client.find_users(token=b64_encoded_custos_token, offset=0, limit=1, username="TestUser4") print(response) register_user() enable_user() get_user() update_user() find_users()

                              

  There are more advanced use cases available at samples

Identity Management Use Cases

• user login (password grant type, authorization_code)
• user logout
• get client-specific OIDC configuration

• check authentication status of OAuth token

  import os from custos.clients.identity_management_client import IdentityManagementClient from custos.transport.settings import CustosServerClientSettings import custos.clients.utils.utilities as utl BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) settings_path = os.path.join(BASE_DIR, 'configs', "settings.ini") custos_settings = CustosServerClientSettings(configuration_file_location=settings_path) id_client = IdentityManagementClient(custos_settings) b64_encoded_custos_token = utl.get_token(custos_settings=custos_settings) def login(username, password): resp = id_client.authenticate(token=b64_encoded_custos_token, username=username, password=password) print(resp) def obtain_access_token_from_code(redirect_uri, code): resp = id_client.token(token=b64_encoded_custos_token, redirect_uri=redirect_uri, code=code, grant_type="authorization_code") print(resp) def get_oidc_configuration(client_id): response = id_client.get_oidc_configuration(token=b64_encoded_custos_token, client_id=client_id) print(response) def logout(refresh_token): response = id_client.end_user_session(token=b64_encoded_custos_token, refresh_token=refresh_token) print(response)

Group Management Use Cases

• create group
• add user to group
• add child group to parent group
• remove user from group

• remove child group from parent group

  import os from custos.clients.group_management_client import GroupManagementClient from custos.transport.settings import CustosServerClientSettings import custos.clients.utils.utilities as utl # load root directoty BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # get settings file path (settings file path reside in configs folder under home directory) settings_path = os.path.join(BASE_DIR, 'configs', "settings.ini") # read settings custos_settings = CustosServerClientSettings(configuration_file_location=settings_path) # create custos group management client group_management_client = GroupManagementClient(custos_settings) # obtain base 64 encoded token for tenant b64_encoded_custos_token = utl.get_token(custos_settings=custos_settings) def create_group(name, description, owner_id): response = group_management_client.create_groups(token=b64_encoded_custos_token, name=name, description=description, owner_id=owner_id) print(response) return response def add_user_to_group(username, group_id, membership_type): response = group_management_client.add_user_to_group(token=b64_encoded_custos_token, username=username, group_id=group_id, membership_type=membership_type) print(response) def add_child_group_to_parent_group(parent_group_id, child_group_id): response = group_management_client.add_child_group(token=b64_encoded_custos_token, parent_group_id=parent_group_id, child_group_id=child_group_id) print(response) def remove_child_group(parent_group_id, child_group_id): response = group_management_client.add_child_group(token=b64_encoded_custos_token, parent_group_id=parent_group_id, child_group_id=child_group_id) print(response) create_group("Group A", "Paren group", "TestUser4") create_group("Group B", "Child group", "TestUser4") add_user_to_group("Testuser5", "602336d5-e193-41ac-bde6-eb36a73f687e", "Member") add_child_group_to_parent_group("8b0f8241-e995-496e-a4f5-bdbde4235215", "602336d5-e193-41ac-bde6-eb36a73f687e") remove_child_group("8b0f8241-e995-496e-a4f5-bdbde4235215", "602336d5-e193-41ac-bde6-eb36a73f687e")

                                       

 Secret Management Use Cases

• generate SSH Key
• save password kind of raw data
• get SSH Key
• get Password token

• delete credentials

  import os from custos.clients.resource_secret_management_client import ResourceSecretManagementClient from custos.transport.settings import CustosServerClientSettings import custos.clients.utils.utilities as utl # load root directoty BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # get settings file path (settings file path reside in configs folder under home directory) settings_path = os.path.join(BASE_DIR, 'configs', "settings.ini") # read settings custos_settings = CustosServerClientSettings(configuration_file_location=settings_path) # create custos resource secret management client resource_management_client = ResourceSecretManagementClient(custos_settings) # obtain base 64 encoded token for tenant b64_encoded_custos_token = utl.get_token(custos_settings=custos_settings) def generateSSH_key(owner_id, description): response = resource_management_client.add_ssh_credential(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, owner_id=owner_id, description=description) print(response) return response def getSSHKey(ssh_credential_token): response = resource_management_client.get_ssh_credential(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, ssh_credential_token=ssh_credential_token) print(response) return response def addPasswordCredential(owner_id, description, password): response = resource_management_client.add_password_credential(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, owner_id=owner_id, description=description, password=password) print(response) return response def getPasswordCredential(password_credential_token): response = resource_management_client.get_password_credential(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, password_credential_token=password_credential_token) print(response) return response generateSSH_key("TestUser5", "My Gateway SSH Key") addPasswordCredential("TestUser5", "My admin password", "asaxsxasxasx") getSSHKey("655e8845-9afa-4251-bb0e-c1eb42bec2fc") getPasswordCredential("a575726a-3e2a-41d1-ad08-06a97dbae903")

Sharing Management Use Cases

• Share entities with  users under given permissions  

• Share entities with groups under given permissions

  import os from custos.clients.sharing_management_client import SharingManagementClient from custos.transport.settings import CustosServerClientSettings import custos.clients.utils.utilities as utl # load root directoty BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # get settings file path (settings file path reside in configs folder under home directory) settings_path = os.path.join(BASE_DIR, 'configs', "settings.ini") # read settings custos_settings = CustosServerClientSettings(configuration_file_location=settings_path) # create custos user management client sharing_management_client = SharingManagementClient(custos_settings) # obtain base 64 encoded token for tenant b64_encoded_custos_token = utl.get_token(custos_settings=custos_settings) def create_permission_type(id, name, description): response = sharing_management_client.create_permission_type(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, id=id, name=name, description=description) print(response) def create_entity_type(id, name, description): response = sharing_management_client.create_entity_type(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, id=id, name=name, description=description) print(response) def create_entity(id, name, description, owner_id, type): response = sharing_management_client.create_entity(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, id=id, name=name, description=description, owner_id=owner_id, type=type, parent_id='') print(response) def share_entity_with_user(entity_id, permission_type, user_id): response = sharing_management_client.share_entity_with_users(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, entity_id=entity_id, permission_type=permission_type, user_id=user_id) print(response) def share_entity_with_group(entity_id, permission_type, group_id): response = sharing_management_client.share_entity_with_groups(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, entity_id=entity_id, permission_type=permission_type, group_id=group_id) print(response) def check_user_has_access(entity_id, permission_type, user_id): response = sharing_management_client.user_has_access(token=b64_encoded_custos_token, client_id=custos_settings.CUSTOS_CLIENT_ID, entity_id=entity_id, permission_type=permission_type, user_id=user_id) print(response) create_permission_type('RW', 'Read and Write', 'Read write permissions') create_entity_type("CRED_TOKEN", "Credential Token", "This is credential token") create_entity('655e8845-9afa-4251-bb0e-c1eb42bec2fc', 'Password Token', 'This is password token', 'TestUser5', 'CRED_TOKEN') share_entity_with_user('655e8845-9afa-4251-bb0e-c1eb42bec2fc', 'RW', 'TestUser4') share_entity_with_group('655e8845-9afa-4251-bb0e-c1eb42bec2fc', 'RW', '602336d5-e193-41ac-bde6-eb36a73f687e') check_user_has_access('655e8845-9afa-4251-bb0e-c1eb42bec2fc', 'RW', 'TestUser4')

In addition to the above basic use cases, you can try out community accounts creation with agent clients, and access audit logs as well.

[1]https://github.com/apache/airavata-custos/tree/develop/custos-integration-services

[2]https://github.com/apache/airavata-custos/tree/develop/custos-client-sdks/custos-python-sdk