Page History

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.

Compiling Python SDK

You need to have Python 3.6 or later installed on your OS.
Clone Python SDK source code [2]

Follow "Generating Distribution Archives" section in ReadMe
After successfully following the steps given in the ReadMe, you should be able to generate 'custos-python-sdk-1.0.0.tar.gz' file.

Setting Up Project

Register your project with Custos as a Tenant using Custos Admin Portal and obtain Custos Id and Custos Secret.
Install "custos-python-sdk-1.0.0.tar.gz" Install SDK into your python project virtual environment.
pip install install custos-python-sdk-==1.0.0.tar.gz2
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

...

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")

...

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.

...

Page tree

Versions Compared

Old Version 16

New Version Current

Key

Compiling Python SDK

Setting Up Project

Group Management Use Cases

Sharing Management Use Cases