Authentication
The Fuzzball client SDK is designed to interact with the Fuzzball API client after a valid Fuzzball API token has been obtained. To obtain this token, you can either leverage the configuration file generated by the Fuzzball CLI, or perform the entire authentication flow programmatically without relying on a pre-existing config file.
Choose the tab below that best fits your use case.
This method is best for local development and interactive sessions where you have already installed the Fuzzball CLI and logged in.
For more information please see the documentation on installing the Fuzzball CLI.
Once the Fuzzball CLI is set up in your environment, you will need to log into your
context such that the configuration file
~/.config/fuzzball/config.yaml contains a valid API token for your active Fuzzball context.
$ fuzzball context login
Logging into current cluster context...
Using a secure browser, open the link to complete login:
https://auth.ciq.dev/auth/realms/e700253c-8ce8-4e36-a420-1e4cab4cb7ef/device?user_code=IVWJ-ZJAZ
Waiting for login completion... Account "User Account (dgodlove@ciq.co)" in use
The code snippet below initializes an instance of the Fuzzball workflow service API. It does this by:
- reading the configuration file
~/.config/fuzzball/config.yamlinto the variableconfig - initializing a Fuzzball configuration in the variable
api_config - setting the variables
host,api_key,api_key_prefixinapi_configusing the active context in the configuration file~/.config/fuzzball/config.yaml - initializing a Fuzzball API client with the variable
api_config - initializing the Fuzzball workflow service API using the Fuzzball API client initialized in the
previous step as variable
api_instance
Lastly, the code implements a workaround for a bearer authentication bug in Python when using Swagger Codegen.
import fuzzball
import sys
import yaml
import os.path
with open(os.path.expanduser("~/.config/fuzzball/config.yaml")) as stream:
try:
config = yaml.safe_load(stream)
except yaml.YAMLError as exc:
print(exc)
sys.exit(-1)
api_config = fuzzball.Configuration()
activeContext = config['activeContext']
for context in config['contexts']:
if context['name'] == activeContext:
api_config.host = "https://{}/v2".format(context['address'])
api_config.api_key['api_key'] = context['auth']['credentials']['token']
api_config.api_key_prefix = {'Bearer': context['auth']['credentials']['token']}
api_instance = fuzzball.WorkflowServiceApi(fuzzball.ApiClient(api_config))
# Bearer auth is broken for Python in Swagger Codegen
# Workaround from https://github.com/swagger-api/swagger-codegen/issues/10060#issuecomment-819545223
api_instance.api_client.set_default_header("Authorization", "Bearer {}".format(api_config.api_key['api_key']))
This method is ideal for non-interactive environments like CI/CD pipelines, automated scripts, or applications where the Fuzzball CLI is not available or a user cannot log in manually. It performs the full authentication flow from directly within your code.
To achieve this, a HTTP client library like requests will be necessary. If you prefer using another http library like httpx or aiohttp, you can replace the http request code accordingly.
The authentication flow consists of primarily two-stages:
1.Get Keycloak Token: Make a POST request to the Keycloak endpoint with a client_id and
user credentials to receive an initial access token.
2.Get Fuzzball Token: Use that access token to make an authorized GET request to the Fuzzball
API, exchanging it for the final, Fuzzball-scoped API token.
In our example, we will structure the code that does the authentication flow into two functions, get_keycloak_token and get_fuzzball_token. These functions will encapsulate the logic for each stage of the authentication process. Refer to the function docstrings for additional details about their implementation.
Security Best Practice
For security, avoid hardcoding credentials in your scripts. The following example reads the username and password from environment variables called
FUZZBALL_USERNAMEandFUZZBALL_PASSWORD. If you are following along with the examples, make sure these are set in your execution environment.
Here is a complete, runnable script that handles the full authentication flow and then uses the SDK to list workflows for the authenticated account.
import fuzzball
import requests
import sys
import os
def get_keycloak_token(auth_domain, realm_id, client_id, username, password):
"""
Authenticates with Keycloak using the password grant type.
"""
print("1. Authenticating with Keycloak...")
token_url = f"https://{auth_domain}/auth/realms/{realm_id}/protocol/openid-connect/token"
payload = {
'grant_type': 'password',
'client_id': client_id,
'username': username,
'password': password,
}
headers = {'Content-Type': 'application/x-www-form-urlencoded'}
try:
response = requests.post(token_url, data=payload, headers=headers)
response.raise_for_status()
access_token = response.json()['access_token']
print(" -> Success: Obtained Keycloak access token.")
return access_token
except requests.exceptions.RequestException as e:
print(f" -> Error during Keycloak authentication: {e}")
return None
def get_fuzzball_token(api_domain, account_id, keycloak_token):
"""
Exchanges a Keycloak access token for a Fuzzball API token.
"""
print("\n2. Exchanging for a Fuzzball API token...")
token_url = f"https://{api_domain}/v2/accounts/{account_id}/token"
headers = {'Authorization': f"Bearer {keycloak_token}"}
try:
response = requests.get(token_url, headers=headers)
response.raise_for_status()
fuzzball_token = response.json()['token']
print(f" -> Success: Obtained Fuzzball API token for account {account_id}.")
return fuzzball_token
except requests.exceptions.RequestException as e:
print(f" -> Error obtaining Fuzzball API token: {e}")
return None
# --- Configuration: Update these values for your environment ---
AUTH_DOMAIN = "auth.example.fuzzball.dev"
API_DOMAIN = "api.example.fuzzball.dev"
REALM_ID = "your-realm-id-here"
CLIENT_ID = "fuzzball-cli"
ACCOUNT_ID = "your-account-id-here"
# --- End Configuration ---
# Load credentials from environment variables
USERNAME = os.environ.get("FUZZBALL_USERNAME")
PASSWORD = os.environ.get("FUZZBALL_PASSWORD")
if __name__ == "__main__":
if not all([USERNAME, PASSWORD]):
print("Error: Please set the FUZZBALL_USERNAME and FUZZBALL_PASSWORD environment variables.")
sys.exit(1)
# Stages 1 & 2: Get the final Fuzzball token
kc_token = get_keycloak_token(AUTH_DOMAIN, REALM_ID, CLIENT_ID, USERNAME, PASSWORD)
if not kc_token:
sys.exit(1) # Exit if Keycloak auth failed
fb_token = get_fuzzball_token(API_DOMAIN, ACCOUNT_ID, kc_token)
if not fb_token:
sys.exit(1) # Exit if Fuzzball token exchange failed
# Stage 3: Configure and Use the Fuzzball SDK
print("\n3. Configuring Fuzzball SDK instance...")
api_config = fuzzball.Configuration()
api_config.host = f"https://{API_DOMAIN}/v2"
# Initialize the API client and set the authorization header
api_client = fuzzball.ApiClient(api_config)
api_client.set_default_header("Authorization", f"Bearer {fb_token}")
print(" -> SDK configured successfully.")
# You can now use any Fuzzball service API, for example, WorkflowServiceApi
print("\n4. Using the SDK to list workflows...")
workflow_api = fuzzball.WorkflowServiceApi(api_client)
try:
response = workflow_api.list_workflows()
for workflow in response.workflows:
print(f" - ID: {workflow.id}, Status: {workflow.status}")
except fuzzball.ApiException as e:
print(f" -> Error calling Fuzzball API: {e}")
sys.exit(1)
The script is structured into configuration constants, two primary functions for authentication, and a main execution block to tie them together.
1.Configuration Constants: Update the constants at the top of the script (AUTH_DOMAIN,
API_DOMAIN, etc.) to match your specific Fuzzball environment.
2.get_keycloak_token(): This function handles the first stage of authentication. It sends
your username, password, and client ID to the Keycloak server and returns a short-lived access
token.
3.get_fuzzball_token(): This function handles the second stage. It sends the Keycloak token
to the Fuzzball API’s token exchange endpoint and returns the final, Fuzzball-scoped token
needed for all subsequent API calls.
4.SDK Initialization: Once the final fb_token is obtained, it is used to configure the
Fuzzball SDK:
# Configure the Fuzzball host
api_config = fuzzball.Configuration()
api_config.host = f"https://{API_DOMAIN}/v2"
# Initialize the API client and set the authorization header
api_client = fuzzball.ApiClient(api_config)
api_client.set_default_header("Authorization", f"Bearer {fb_token}")
With the api_client configured, you can now initialize any Fuzzball service API (e.g.,
fuzzball.WorkflowServiceApi(api_client)) and begin making calls.
The next sections will detail usage of the initialized Fuzzball workflow service API to perform various Fuzzball workflow operations.