Building blocks for your custom OAA integration
from oaaclient.client import OAAClient
from oaaclient.templates import CustomApplication, OAAPermission
# creates a connection class to communicate with Veza
veza_con = OAAClient(url=veza_url, token=veza_api_key)
# creates a new Custom Application model
custom_app = CustomApplication(name="Sample App", application_type="sample")# define a permission
custom_app.add_custom_permission("owner", [OAAPermission.DataRead, OAAPermission.DataWrite])
# create a local user
jsmith = custom_app.add_local_user(unique_id="jsmith", name="Jane Smith", identities=["jane@example.com"])
# create a resource
resource1 = custom_app.add_resource(name="Resource 1", resource_type="Thing")
# assign a user to a resource
jsmith.add_permission(permission="owner", resources=[resource1])veza_con.push_application(provider, data_source_name, application_object=custom_app) try:
response = veza_con.push_application(provider_name=provider_name,
data_source_name=data_source_name,
application_object=custom_app,
)
if response.get("warnings"):
print("Push succeeded with warnings:")
for w in response["warnings"]:
print(w)
except OAAClientError as e:
print(f"Error: {e.error}: {e.message} ({e.status_code})", file=sys.stderr)
if hasattr(e, "details"):
for d in e.details:
print(d, file=sys.stderr)VEZA_URLreport_builder_entrypoint() → Nonemain()__init__(
error: 'str',
message: 'str',
status_code: 'int' = None,
details: 'list' = None
) → None__init__(*args, **kwargs)__init__(
error: 'str',
message: 'str',
status_code: 'int' = None,
details: 'list' = None
) → None__init__(
url: 'str' = None,
api_key: 'str' = None,
username: 'str' = None,
token: 'str' = None
)add_query_report(report_id: 'str', query_id: 'str') → dictapi_delete(api_path: 'str', params: 'dict' = None) → dictapi_get(api_path: 'str', params: 'dict' = None) → list | dictapi_patch(api_path: 'str', data: 'dict', params: 'dict' = None) → dictapi_post(api_path: 'str', data: 'dict', params: 'dict' = None) → list | dictapi_put(api_path: 'str', data: 'dict', params: 'dict' = None) → list | dictcreate_data_source(
name: 'str',
provider_id: 'str',
options: 'dict | None' = None
) → dictcreate_datasource(name, provider_id)create_provider(
name: 'str',
custom_template: 'str',
base64_icon: 'str' = '',
options: 'dict | None' = None
) → dictcreate_query(query: 'dict') → dictcreate_report(report: 'dict') → dictdelete_data_source(data_source_id: 'str', provider_id: 'str') → dictdelete_provider(provider_id: 'str') → dictdelete_query(id: 'str', force: 'bool' = False) → dictdelete_report(id: 'str') → dictget_data_source(name: 'str', provider_id: 'str') → dictget_data_sources(provider_id: 'str') → list[dict]get_provider(name: 'str') → dictget_provider_by_id(provider_id: 'str') → dictget_provider_list() → list[dict]get_queries(include_inactive_queries: 'bool' = True) → list[dict]get_query_by_id(id: 'str') → dictget_report_by_id(id: 'str', include_inactive_queries: 'bool' = True) → dictget_reports(
include_inactive_reports: 'bool' = True,
include_inactive_queries: 'bool' = True
) → list[dict]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
) → dictpush_metadata(
provider_name: 'str',
data_source_name: 'str',
metadata: 'dict',
save_json: 'bool' = False,
options: 'dict | None' = None
) → dictupdate_provider_icon(provider_id: 'str', base64_icon: 'str') → Noneupdate_report(report_id: 'str', report: 'dict') → dictupdate_user_agent(extra: 'str' = '') → None__init__(backoff_max=30, **kwargs) → Noneload_json_from_file(json_path: str) → dictencode_icon_file(icon_path: str) → strexists_in_query_array(value_to_find, input_array) → boolbuild_report(veza_con, report_definition: dict) → dicttruncate_string(source_str: str, length: int = 256) → str__init__(data=None, **kwargs) → Noneunique_id (str, optional): Unique identifier for role for reference by IDignore_noneignore_noneignore_noneFalseunique_strs(input: 'list') → list__init__(message)__init__(name, custom_template)serialize()__init__(name, application_type, description=None)__init__(name: 'str', application_type: 'str', description: 'str' = None) → Noneadd_access(identity, identity_type, permission, resource=None)
---
### <kbd>method</kbd> `add_access_cred`
```python
add_access_cred(unique_id: 'str', name: 'str') → AccessCredadd_custom_permission(
name: 'str',
permissions: 'List[OAAPermission]',
apply_to_sub_resources: 'bool' = False,
resource_types: 'List[str]' = None
) → CustomPermissionadd_idp_identity(name: 'str') → IdPIdentityadd_local_group(
name: 'str',
identities: 'List[str]' = None,
unique_id: 'str' = None
) → LocalGroupadd_local_role(
name: 'str',
permissions: 'List[str]' = None,
unique_id: 'str' = None
) → LocalRoleadd_local_user(
name: 'str',
identities: 'List[str]' = None,
groups: 'List[str]' = None,
unique_id: 'str' = None
) → LocalUseradd_resource(
name: 'str',
resource_type: 'str',
description: 'str' = None,
unique_id: 'str' = None
) → CustomResourceadd_tag(key: 'str', value: 'str' = '') → Noneapp_dict() → dictdefine_custom_permission(
custom_permission: 'CustomPermission'
) → CustomPermission
**Args:**
- <b>`custom_permission`</b> (CustomPermission): CustomPermission class
**Raises:**
- <b>`Exception`</b>: Duplicate Keys
**Returns:**
- <b>`CustomPermission`</b>: The defined custom Permission
---
### <kbd>method</kbd> `get_identity_to_permissions`
```python
get_identity_to_permissions() → listget_payload() → dictpermissions_dict() → dictset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → None__init__(
name: 'str',
resource_type: 'str',
description: 'str',
application_name: 'str',
resource_key: 'str' = None,
property_definitions: 'ApplicationPropertyDefinitions' = None,
unique_id: 'str' = None
) → Noneadd_access(identity, identity_type, permission)add_resource_connection(id: 'str', node_type: 'str') → Noneadd_sub_resource(
name: 'str',
resource_type: 'str',
description: 'str' = None,
unique_id: 'str' = None
) → CustomResourceadd_tag(key: 'str', value: 'str' = '') → Noneset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__(
name: 'str',
identity_type: 'OAAIdentityType',
unique_id: 'str' = None,
property_definitions: 'ApplicationPropertyDefinitions' = None
) → Noneadd_permission(
permission: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'bool' = False
) → Noneadd_role(
role: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'Optional[bool]' = None,
assignment_properties: 'Optional[dict]' = None
) → Noneadd_tag(key: 'str', value: 'str' = '') → Noneget_identity_to_permissions(application_name: 'str') → dictset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → None__init__(
name: 'str',
identities: 'List[str]' = None,
groups: 'List[str]' = None,
unique_id: 'str' = None,
property_definitions: 'ApplicationPropertyDefinitions' = None
) → Noneadd_access_cred(access_cred: 'str') → Noneadd_group(group: 'str') → Noneadd_identities(identities: 'List[str]') → Noneadd_identity(identity: 'str') → Noneadd_permission(
permission: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'bool' = False
) → Noneadd_role(
role: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'Optional[bool]' = None,
assignment_properties: 'Optional[dict]' = None
) → Noneadd_tag(key: 'str', value: 'str' = '') → Noneget_identity_to_permissions(application_name: 'str') → dictset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__(
name,
identities=None,
unique_id: 'str' = None,
property_definitions: 'ApplicationPropertyDefinitions' = None
)add_group(group: 'str') → Noneadd_identity(identity: 'str') → Noneadd_permission(
permission: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'bool' = False
) → Noneadd_role(
role: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'Optional[bool]' = None,
assignment_properties: 'Optional[dict]' = None
) → Noneadd_tag(key: 'str', value: 'str' = '') → Noneget_identity_to_permissions(application_name: 'str') → dictset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__(name: 'str') → Noneadd_permission(
permission: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'bool' = False
) → Noneadd_role(
role: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'Optional[bool]' = None,
assignment_properties: 'Optional[dict]' = None
) → Noneadd_tag(key: 'str', value: 'str' = '') → Noneget_identity_to_permissions(application_name: 'str') → dictset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → None__init__(
unique_id: 'str',
name: 'str',
property_definitions: 'ApplicationPropertyDefinitions' = None
) → Noneadd_permission(
permission: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'bool' = False
) → Noneadd_role(
role: 'str',
resources: 'List[CustomResource]' = None,
apply_to_application: 'Optional[bool]' = None,
assignment_properties: 'Optional[dict]' = None
) → Noneadd_tag(key: 'str', value: 'str' = '') → Noneget_identity_to_permissions(application_name: 'str') → dictset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__(
name: 'str',
permissions: 'List[str]' = None,
unique_id: 'str' = None,
property_definitions: 'ApplicationPropertyDefinitions' = None
) → Noneadd_permissions(permissions: 'List[str]') → Noneadd_role(role: 'str') → Noneadd_tag(key: 'str', value: 'str' = '') → Noneset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__(
name: 'str',
permissions: 'List[OAAPermission]',
apply_to_sub_resources: 'bool' = False,
resource_types: 'list' = None
) → Noneadd_resource_type(resource_type: 'str') → Noneto_dict() → dict__init__(application_type: 'str') → Nonedefine_access_cred_property(
name: 'str',
property_type: 'OAAPropertyType'
) → Nonedefine_application_property(
name: 'str',
property_type: 'OAAPropertyType'
) → Nonedefine_local_group_property(
name: 'str',
property_type: 'OAAPropertyType'
) → Nonedefine_local_role_property(name: 'str', property_type: 'OAAPropertyType') → Nonedefine_local_user_property(name: 'str', property_type: 'OAAPropertyType') → Nonedefine_resource_property(
resource_type: 'str',
name: 'str',
property_type: 'OAAPropertyType'
) → Nonedefine_role_assignment_property(
name: 'str',
property_type: 'OAAPropertyType'
) → Noneto_dict() → dictvalidate_name(name: 'str') → Nonevalidate_property_name(
property_name: 'str',
entity_type: 'str',
resource_type: 'str' = None
) → bool__init__(
name: 'str',
idp_type: 'str',
domain: 'str',
description: 'str' = None
) → Noneadd_app(id: 'str', name: 'str') → CustomIdPAppadd_group(
name: 'str',
full_name: 'str' = None,
identity: 'str' = None
) → CustomIdPGroupadd_user(
name: 'str',
full_name: 'str' = None,
email: 'str' = None,
identity: 'str' = None
) → CustomIdPUserget_payload() → dict__init__(
name: 'str',
property_definitions: 'IdPPropertyDefinitions' = None
) → Noneadd_tag(key: 'str', value: 'str' = '') → Noneset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__(
name: 'str',
email: 'str' = None,
full_name: 'str' = None,
identity: 'str' = None,
property_definitions: 'IdPPropertyDefinitions' = None
) → Noneadd_app_assignment(
id: 'str',
name: 'str',
app_id: 'str',
assignment_properties: 'Optional[dict]' = None
) → Noneadd_assumed_role_arns(arns: 'List[str]') → Noneadd_groups(group_identities: 'List[str]') → Noneadd_tag(key: 'str', value: 'str' = '') → Noneset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneset_source_identity(identity: 'str', provider_type: 'IdPProviderType') → Noneto_dict() → dict__init__(
name: 'str',
full_name: 'str' = None,
identity: 'str' = None,
property_definitions: 'IdPPropertyDefinitions' = None
) → Noneadd_app_assignment(
id: 'str',
name: 'str',
app_id: 'str',
assignment_properties: 'Optional[dict]' = None
) → Noneadd_assumed_role_arns(arns: 'List[str]') → Noneadd_groups(group_identities: 'List[str]') → Noneadd_tag(key: 'str', value: 'str' = '') → Noneset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → None__init__(
id: 'str',
name: 'str',
property_definitions: 'IdPPropertyDefinitions' = None
) → Noneadd_assumed_role_arns(arns: 'List[str]') → Noneadd_tag(key: 'str', value: 'str' = '') → Noneset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__() → Nonedefine_app_assignment_property(
name: 'str',
property_type: 'OAAPropertyType'
) → Nonedefine_app_property(name: 'str', property_type: 'OAAPropertyType') → Nonedefine_domain_property(name: 'str', property_type: 'OAAPropertyType') → Nonedefine_group_property(name: 'str', property_type: 'OAAPropertyType') → Nonedefine_user_property(name: 'str', property_type: 'OAAPropertyType') → Noneto_dict() → dictvalidate_property_name(property_name: 'str', entity_type: 'str') → None__init__(key: 'str', value: 'str' = '') → None__init__(name: 'str', hris_type: 'str', url: 'str')add_employee(
unique_id: 'str',
name: 'str',
employee_number: 'str',
first_name: 'str',
last_name: 'str',
is_active: 'bool',
employment_status: 'str'
) → HRISEmployeeadd_group(unique_id: 'str', name: 'str', group_type: 'str') → HRISGroupget_payload() → dict__init__(name: 'str', url: 'str' = '')add_idp_type(provider_type: 'IdPProviderType') → list[IdPProviderType]to_dict() → dict__init__(
unique_id: 'str',
name: 'str',
employee_number: 'str',
first_name: 'str',
last_name: 'str',
is_active: 'bool',
employment_status: 'str'
)add_group(group_id: 'str') → Noneadd_manager(manager_id: 'str') → Noneset_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__(unique_id: 'str', name: 'str', group_type: 'str')set_property(
property_name: 'str',
property_value: 'any',
ignore_none: 'bool' = False
) → Noneto_dict() → dict__init__()define_employee_property(name: 'str', property_type: 'OAAPropertyType') → Nonedefine_group_property(name: 'str', property_type: 'OAAPropertyType') → Nonedefine_system_property(name: 'str', property_type: 'OAAPropertyType') → Noneto_dict() → dictvalidate_name(name: 'str') → Nonediscover()CustomApplication#!/usr/bin/env python3
from __future__ import annotations
import argparse
import json
import logging
import os
import sys
import requests
from requests.exceptions import HTTPError, RequestException
from oaaclient.client import OAAClient, OAAClientError
from oaaclient.templates import CustomApplication, OAAPermission, OAAPropertyType, LocalUser
import oaaclient.utils as oaautils
"""
Logging in connectors is very important. Establishing a local logging function here
allows the connector log even if its imported into another block of code
"""
logging.basicConfig(
format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
log = logging.getLogger(__name__)
"""
The main logic for connecting to the destination application.
Authorization data collection and OAA template population should be structured into a class.
The class can store the authentication tokens making connections through an API or SDK easier.
The class should also instantiate the OAA CustomApplication or CustomIdP class
to manage populating all the identity, resource and authorization data.
"""
class OAAConnector():
APP_TYPE="SampleApp"
def __init__(self, auth_token: str) -> None:
self.auth_token = auth_token
# Set the application name and type. The type will generally be the vendor or product.
# The App Name can be the same, or contain additional context like the instance name.
app_name = "My App - West"
self.app = CustomApplication(app_name, application_type=self.APP_TYPE)
# declaring custom properties as part of the `__init__` keeps them together
self.app.property_definitions.define_local_user_property("email", OAAPropertyType.STRING)
self.app.property_definitions.define_local_user_property("has_mfa", OAAPropertyType.BOOLEAN)
"""
A `discover` method that starts the discovery cycle.
The discovery cycle should be invoked separately from the init.
More complex connectors may have additional setup steps between init and discovery
"""
def discover(self) -> None:
"""Discovery method"""
log.info(
"Start App discovery") # Start and stop log messages provide progress as discovery proceeds
self._discover_users()
self._discover_roles()
log.info("Finished App discovery")
return
"""
Smaller functions to perform portions of the discovery like users, groups, roles
should be private `_` functions to imply that they should not be run alone,
unless care is taken to ensure no dependencies.
For example, discovery of roles may assume that users are already discovered.
"""
def _discover_users(self) -> None:
log.info("Start user discovery")
# perform user discovery, for example process each user returned from an API call
for user in self._api_get("/Users"):
local_user = self.app.add_local_user(id=user["id"], name=user["name"])
local_user.is_active = user["active"]
local_user.set_property("has_mfa", user["mfa_enabled"])
log.info("Finished user discovery")
return
def _discover_roles(self) -> None:
log.info("Start role discovery")
# perform user discovery
log.info("Finished role discovery")
return
"""
Any required methods to interface with the application should be defined
as part of the class. Not all connectors need these methods, as they may use
other SDKs to interface with the application.
"""
def _api_get(self, path: str) -> dict:
# implement logic to make API call, process results, handle errors, retries ect.
# Could be done with a vendor SDK, SQL, or any method supported by the application
return {}
"""
A run function that is separate from `main` makes it easy to import the connector
into another piece of Python code. This may be useful to call the connector from
code that retrieves secrets or manages the job in other ways.
All necessary parameters should be taken in through the `run` function
"""
def run(veza_url: str, veza_api_key: str, app_key: str, **config_args) -> None:
# Process any configuration arguments
if config_args.get("debug"):
log.setLevel(logging.DEBUG)
logging.getLogger("urllib3").setLevel(logging.INFO)
log.info("Enabling debug logging")
else:
log.setLevel(logging.INFO)
save_json = config_args.get("save_json", False)
if not isinstance(save_json, bool):
raise TypeError("save_json argument must be boolean")
# Connect to the Veza instance before discovery to validate that the credentials are valid
try:
conn = OAAClient(url = veza_url, api_key = veza_api_key)
except OAAClientError as error:
log.error(f"Unable to connect to Veza {veza_url}")
log.error(error.message)
# run function should raise any exception so that they can be handled by the parent code, never exit
raise error
# Initialize the connector class and run discovery
try:
app = OAAConnector(auth_token=app_key)
app.discover()
except RequestException as e:
# process possible exceptions from the app discovery
log.error("Error during discovery")
log.error(f"{e} - {e.response.status_code} {e.response.text}")
raise e
# After discovery is complete, set up the Provider and Data Source to push the data to
# Provider name should be consistent with the vendor and application
provider_name = "My App"
provider = conn.get_provider(provider_name)
if provider:
log.info("found existing provider")
else:
log.info(f"creating provider {provider_name}")
provider = conn.create_provider(provider_name, "application", base64_icon=APP_SVG_B64)
log.info(f"provider: {provider['name']} ({provider['id']})")
# Data Source name should be unique to the instance of the app that is discovered but consistent.
# For example the hostname of the application or deployment name. Do not use something that will change.
data_source_name = f"App - {app.unique_identifier}"
try:
log.info("uploading application data")
response = conn.push_application(provider_name,
data_source_name = data_source_name,
application_object = app.app,
save_json = save_json
)
# An OAA Push can succeed with warnings, you can log out the warnings
if response.get("warnings", None):
log.warning("Push succeeded with warnings:")
for e in response["warnings"]:
log.warning(e)
log.info("success")
except OAAClientError as error:
# if there is an issue with the OAA payload the error details should contain useful information to help resolve the problem
log.error(f"{error.error}: {error.message} ({error.status_code})")
if hasattr(error, "details"):
for detail in error.details:
log.error(detail)
raise error
return
"""
Setting an application icon helps visually identify the app in the Veza UI.
A Base64 encoding of an SVG or PNG in the app code is an option.
You can also import an icon from a file with `oaaclient.utils.encode_icon_file`.
"""
APP_SVG_B64 = """
PD94bWwgdmVyc2lvbj0iMS4wIiBlbmNvZGluZz0iVVRGLTgiIHN0YW5kYWxvbmU9InllcyI/Pg0KPCFET0NUWVBFIHN2ZyBQVUJMSUMgIi0vL1czQy8vRFRE
IFNWRyAxLjEvL0VOIiAiaHR0cDovL3d3dy53My5vcmcvR3JhcGhpY3MvU1ZHLzEuMS9EVEQvc3ZnMTEuZHRkIj4NCjxzdmcgd2lkdGg9IjQwMCIgaGVpZ2h0
PSI0MDAiIHZpZXdCb3g9IjAgMCAyMDAgMjAwIiB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciPg0KICA8cGF0aCBmaWxsPSJyZWQiIHN0cm9r
ZT0iIzAwMCIgc3Ryb2tlLXdpZHRoPSIyLjUiIGQ9Ik0xNjggMTY4SDMyVjMyaDEzNnoiLz4NCjwvc3ZnPg0K
"""
"""
The main entry point should only deal with processing parameters and invoking
the run function. No OAA or application discovery should happen during the main
function. All required parameters should be configurable through the OS
environment. It should be possible to run the connector from the command line
with no arguments.
"""
def main():
"""
process command line and OS environment variables, then call `run`
"""
parser = argparse.ArgumentParser(description = "OAA Connector")
# using the `default=os.getenv()` pattern makes it easier to get parameters from command line or OS environment
parser.add_argument("--veza-url", default=os.getenv("VEZA_URL"), help="the URL of the Veza instance")
parser.add_argument("--debug", action="store_true", help="Set the log level to debug")
parser.add_argument("--save-json", action="store_true", help="Save OAA JSON payload to file")
args = parser.parse_args()
# Secrets should only be passed in through ENV
veza_api_key = os.getenv("VEZA_API_KEY")
veza_url = args.veza_url
save_json = args.save_json
if not veza_api_key:
oaautils.log_arg_error(log, None, "VEZA_API_KEY")
if not veza_url:
oaautils.log_arg_error(log, "--veza-url", "VEZA_URL")
# ensure required variables are provided
if None in [veza_api_key, veza_url]:
log.error(f"missing one or more required parameters")
sys.exit(1)
try:
run( veza_url=veza_url, veza_api_key=veza_api_key, save_json=save_json, debug=args.debug)
except (OAAClientError, RequestException):
log.error("Exiting with error")
sys.exit(1)
if __name__ == "__main__":
# replace the log with the root logger if running as main
log = logging.getLogger()
logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
main()