Client

Classes for calling Veza APIs and managing OAA providers and data sources.

Copyright 2022 Veza Technologies Inc.

Use of this source code is governed by the MIT license that can be found in the LICENSE file or at https://opensource.org/licenses/MIT.

Global Variables

---

• OAACLIENT_VERSION

• PROVIDER_ICON_MAX_SIZE

---

function report_builder_entrypoint

report_builder_entrypoint() → None

Entry point for oaaclient-report-builder command

Reads a JSON file and passes it to the oaaclient.utils.build_report method

---

function main

main()

---

class OAAClientError

Error raised by OAAClient.

Raised for issues connecting to the OAA API and when the API returns an error.

Args:

• error (str): short string for error message

• message (str): detailed error message

• status_code (int, optional): status code for HTTP related errors. Defaults to None.

• details (list, optional): list of additional details for error. Defaults to None.

method __init__

__init__(
    error: 'str',
    message: 'str',
    status_code: 'int' = None,
    details: 'list' = None
) → None

---

class OAAResponseError

Error returned from API Call

method __init__

__init__(*args, **kwargs)

---

class OAAConnectionError

Error with API Connection

method __init__

__init__(
    error: 'str',
    message: 'str',
    status_code: 'int' = None,
    details: 'list' = None
) → None

---

class OAAClient

Class for OAA API Connection and Management.

Utilities for OAA-related operations with Veza API calls. Manages custom providers and data sources, and can push OAA payloads from JSON or template objects.

Connection url and API key can be automatically loaded from OS environment values if set. To utilize environment variables initialize OAAClient without providing a URL or API key value and set the VEZA_URL and VEZA_API_KEY OS environment variables.

Args:

• url (str, optional): URL for Veza instance.

• api_key (str, optional): Veza API key.

• username (str, optional): Not used (legacy). Defaults to None.

• token (str, optional): Legacy parameter name for API key. Defaults to None.

Attributes:

• url (str): URL of the Veza instance to connect to

• api_key (str): Veza API key

• enable_compression (bool): Enable or disable compression of the OAA payload during push, defaults to enabled (True)

Raises:

• OAAClientError: For errors connecting to API and if API returns errors

method __init__

__init__(
    url: 'str' = None,
    api_key: 'str' = None,
    username: 'str' = None,
    token: 'str' = None
)

---

method add_query_report

add_query_report(report_id: 'str', query_id: 'str') → dict

Add a Query to a Report

Adds a Query to an existing Report by ID

Args:

• report_id (str): Report UUID to add Query to

• query_id (str): Query UUID to add

Returns:

• dict: API Response

---

method api_delete

api_delete(api_path: 'str', params: 'dict' = None) → dict

Perform REST API DELETE operation.

Args:

• api_path (str): API Path API path relative to Veza URL

Raises:

• OAAResponseError: API returned an error

• OAAConnectionError: Connection error during HTTP operation

Returns:

• dict: API response from call

---

method api_get

api_get(api_path: 'str', params: 'dict' = None) → list | dict

Perform a Veza API GET operation.

Makes the GET API call to the given path and processes the API response. Returns the value or values result returned from the API.

• For API endpoints that return a list like /api/v1/providers/custom function will return a list of entities or an empty list if the API returns no results.

• For API endpoints that are a specific ID such as /api/v1/providers/custom/<uuid> function will return the dictionary result of the JSON returned by the API.

Args:

• api_path (str): API path relative to Veza URL (example /api/v1/providers).

Raises:

• OAAResponseError: API returned an error

• OAAConnectionError: Connection error during HTTP operation

Returns:

• list|dict: Returns list or dict based on API destination

---

method api_patch

api_patch(api_path: 'str', data: 'dict', params: 'dict' = None) → dict

Perform REST API PATCH operation.

Args:

• api_path (str): API Path API path relative to Veza URL

Raises:

• OAAResponseError: API returned an error

• OAAConnectionError: Connection error during HTTP operation

Returns:

• dict: API response from call

---

method api_post

api_post(api_path: 'str', data: 'dict', params: 'dict' = None) → list | dict

Perform a Veza API POST operation.

Call POST on the supplied Veza instance API path, submitting a data payload.

Returns value or values response from API result. Paginated responses are automatically processed to collect all responses a single list.

Args:

• api_path (str): API path relative to Veza URL example /api/v1/providers

• data (dict): dictionary object included as JSON in body of POST operation

• params (dict, optional): Optional HTTP query parameters. Defaults to empty dictionary.

Raises:

• OAAResponseError: API returned an error

• OAAConnectionError: Connection error during HTTP operation

Returns:

• dict: API response as dictionary

---

method api_put

api_put(api_path: 'str', data: 'dict', params: 'dict' = None) → list | dict

Perform Veza API PUT operation.

Call PUT on the supplied Veza instance API path, including the data payload.

Returns value or values response from API result. Paginated responses are automatically processed to collect all responses a single list.

Args:

• api_path (str): API path relative to Veza URL example /api/v1/providers

• data (dict): dictionary object included as JSON in body of PUT operation

• params (dict, optional): Optional HTTP query parameters. Defaults to empty dictionary.

Raises:

• OAAResponseError: API returned an error

• OAAConnectionError: Connection error during HTTP operation

Returns:

• dict: API response as dictionary

---

method create_data_source

create_data_source(
    name: 'str',
    provider_id: 'str',
    options: 'dict | None' = None
) → dict

Create a new Data Source for the given Provider ID.

Args:

• name (str): Name for new Data Source

• provider_id (str): Unique identifier for the Provider

• options: (dict, optional): Additional arguments to be included with data source create call to Veza. Defaults to None.

Raises:

• ValueError: Data source name contains invalid characters

Returns:

• dict: dictionary of new Data Source

---

method create_datasource

create_datasource(name, provider_id)

Deprecated Legacy function for backward-compatibility.

---

method create_provider

create_provider(
    name: 'str',
    custom_template: 'str',
    base64_icon: 'str' = '',
    options: 'dict | None' = None
) → dict

Create a new Provider.

Creates a new Provider with the given name. An error will be raised for naming conflicts.

Args:

• name (str): new Provider name

• custom_template (str): the OAA template to use for the Provider ("application" or "identity_provider")

• base64_icon (str, optional): Base64 encoded string of icon to set for Provider. Defaults to "".

• options: (dict, optional): Additional arguments to be included with provider create call to Veza. Defaults to None.

Raises:

• ValueError: Provider name contains invalid characters

Returns:

• dict: dictionary representing the created Provider

---

method create_query

create_query(query: 'dict') → dict

Create a new Assessment Query

For details on how to define an Assessment Query see the Veza docs.

Args:

• query (dict): Query definition

Returns:

• dict: API response including ID of created Query

---

method create_report

create_report(report: 'dict') → dict

Create a new Report

For details on how to define a new Report see the Veza docs.

Args:

• report (dict): Report definition

Returns:

• dict: API response including ID of created Report

---

method delete_data_source

delete_data_source(data_source_id: 'str', provider_id: 'str') → dict

Delete an existing Data Source by ID.

Removes a Data Source and all entity data.

Args:

• data_source_id (str): ID of Data Source to delete

• provider_id (str): ID of Data Source Provider

Returns:

• dict: API response

---

method delete_provider

delete_provider(provider_id: 'str') → dict

Delete an existing provider by ID.

Deleting a provider removes all datasources and historical data. Fully deleting the provider is a background operation that will complete after API response is returned.

Args:

• provider_id (str): ID of provider to delete

Returns:

• dict: API response

---

method delete_query

delete_query(id: 'str', force: 'bool' = False) → dict

Delete an Assessment Query by ID

Args:

• id (str): UUID of Query to delete

• force (bool): Force deletion of query that may be part of a report. Defaults to False

Returns:

• dict: API response from delete

---

method delete_report

delete_report(id: 'str') → dict

Delete Report by ID

Args:

• id (str): UUID of Report to delete

Returns:

• dict: API response

---

method get_data_source

get_data_source(name: 'str', provider_id: 'str') → dict

Get Provider's Data Source by name.

Find a Data Source from a specific provider based on the name of the Data Source

Args:

• name (str): Data Source name

• provider_id (str): Provider unique ID

Returns:

• dict: Data Source as dict or None

---

method get_data_sources

get_data_sources(provider_id: 'str') → list[dict]

Get Data Sources for Provider by ID.

Get the list of existing Data Sources, filtered by Provider UUID.

Args:

• provider_id (str): ID of Provider

Returns:

• list[dict]: List of Data Sources as dictionaries

---

method get_provider

get_provider(name: 'str') → dict

Get Provider by name.

Args:

• name (str): name of Provider

Returns:

• dict: dictionary representing Provider or None

---

method get_provider_by_id

get_provider_by_id(provider_id: 'str') → dict

Get Provider by UUID identifier.

Args:

• provider_id (str): Unique global identifier for provider

Returns:

• dict: dictionary representation of Provider or None

---

method get_provider_list

get_provider_list() → list[dict]

Return list of Providers.

Returns:

• list[dict]: Returns a list of existing Providers as dictionaries

---

method get_queries

get_queries(include_inactive_queries: 'bool' = True) → list[dict]

Get all saved Assessment Queries

Veza can filter out queries that include inactive entity types (e.g. Okta Users on a system without Okta configured). To only retrieve queries that include active entity types set include_inactive_queries to False.

Args:

• include_inactive_queries (bool): Set False to exclude inactive queries from result. Defaults to True.

Returns:

• list: List of assessment Queries as dictionaries

---

method get_query_by_id

get_query_by_id(id: 'str') → dict

Get Assessment Query by ID

Args:

• id (str): UUID identifier for Query

Returns:

• dict: Query definition

---

method get_report_by_id

get_report_by_id(id: 'str', include_inactive_queries: 'bool' = True) → dict

Get Report by ID

Veza can filter out queries from reports that only contain entity types that are not configured (e.g. Okta Users on a system without Okta configured). To only return queries configured on the report that match entity types configured on the system set include_inactive_queries to False

Args:

• id (str): UUID of Report to get

• include_inactive_queries (bool): Set True to include inactive queries. Default True.

Returns:

• dict: Report definition

---

method get_reports

get_reports(
    include_inactive_reports: 'bool' = True,
    include_inactive_queries: 'bool' = True
) → list[dict]

Get all Reports

Gets Reports created on the system. To get all reports include_inactive_reports and include_inactive_queries must be set to True.

Args:

• include_inactive_reports (bool, Optional): Set to True to include reports that contain no active providers, defaults to True.

• include_inactive_queries (bool, Optional): Set to True to include reports that contain only inactive queries, defaults to True.

Returns:

• list[dict]: List of Reports as dictionary objects

---

method push_application

push_application(
    provider_name: 'str',
    data_source_name: 'str',
    application_object: 'CustomApplication | CustomIdPProvider',
    save_json: 'bool' = False,
    create_provider: 'bool' = False,
    options: 'dict | None' = None
) → dict

Push an OAA Application Object (such as CustomApplication).

Extracts the OAA JSON payload from the supplied OAA class (e.g. CustomApplication, CustomIdPProvider, etc) and push to the supplied Data Source.

The Provider with provider_name must be a valid existing Provider or create_provider must be set to True. A new data source will be created automatically by default if it does not already exist.

For logging, and debugging, the optional save_json flag writes the payload to a local file (before push). Output file name is formatted with a timestamp: {data source name}-{%Y%m%d-%H%M%S}.json

Args:

• provider_name (str): Name of an existing Provider.

• data_source_name (str): Name for Data Source (will be created if it doesn't exist).

• application_object (Class): OAA object to extract the payload from

• save_json (bool, optional): Save the JSON payload to a local file before push. Defaults to False.

• create_provider (bool, optional): Create a new Provider if Provider does not already exists. Defaults to False.

• options (dict, optional): Additional dictionary of key/values to be included in push API call. Defaults to None.

Raises:

• OAAClientError: If any API call returns an error (including errors processing the OAA payload).

Returns:

• dict: API response from push, including any warnings that are returned.

---

method push_metadata

push_metadata(
    provider_name: 'str',
    data_source_name: 'str',
    metadata: 'dict',
    save_json: 'bool' = False,
    options: 'dict | None' = None
) → dict

Push an OAA payload dictionary to Veza.

Publishes the supplied metadata dictionary representing an OAA payload to the specified provider and data source. The function will create a new data source if it does not already exist, but requires the Provider be created ahead of time.

Args:

• provider_name (str): Name of existing Provider

• data_source_name (str): Name for Data Source, will be created if doesn't exist.

• metadata (dict): Dictionary of OAA payload to push.

• save_json (bool, optional): Save the OAA JSON payload to a local file before push. Defaults to False.

• options (dict, optional): Additional dictionary of key/values to be included in push API call. Defaults to None.

Raises:

• OAAClientError: If any API call returns an error including errors processing the OAA payload.

Returns:

• dict: API response to the push request (including any warnings).

---

method update_provider_icon

update_provider_icon(provider_id: 'str', base64_icon: 'str') → None

Update an existing custom provider icon from base64 encoded string.

To load an icon from file, use utils.encode_icon_file to get the base64 encoding of the file first

Args:

• provider_id (str): unique ID of existing provider

• base64_icon (str): base64 encoded string of new icon

Raises:

• ValueError: If icon size exceeds maximum allowed size

---

method update_report

update_report(report_id: 'str', report: 'dict') → dict

Update an existing report

Args:

• report_id (str): UUID of Report to updated

• report (dict): Updated Report definition

Returns:

• dict: API response

---

method update_user_agent

update_user_agent(extra: 'str' = '') → None

Updates the User-Agent string passed with all API calls

Generates a User-Agent with the oaaclient version, Python version and platform information.

The optional extra string will be appended if provided.

Args:

• extra (str, optional): Additional information to append to User-Agent string. Defaults to "".

---

class OAARetry

Super class for urllib3.util.retry

Super class to allow modifying the default max backoff time from 120 seconds to our own value

Args:

• Retry (type): description

method __init__

__init__(backoff_max=30, **kwargs) → None

---