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









  • No labels