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