This document provides a high-level overview and examples for getting started with a new OAA connector to integrate Veza with SaaS applications, infrastructure systems, custom-built applications, and other systems. These examples use Python and the oaaclient SDK.
When developing a connector, the source systems and customer needs can require changes to code flow for different deployment scenarios. The overall goals, best practices, and flow will apply for most integrations:
Code goals
The sample code was written with the following goals in mind:
Connector should be easy to run from automation platforms and the command line.
Parameters are passed through environment variables as well as command line flags. This makes the code easier to package into containers or serverless stacks and control through outside configuration. Connectors by nature require secrets such as the Veza API key. Managing these through variables affords additional options for secrets management.
Connectors do not require state:
Connectors do not require any persistent data between runs.
There is no special invocation for the first run or subsequent runs.
The connector handles all of the logic for provider and data source creation internally, as needed, or by discovering existing ones.
Data Source name should be unique to the discovered environment. This can be achieved by including a hostname or instance ID in the data source name: discoverable by the code and consistent between runs. This ensures that two instances of the same connector do not interfere with each other if discovering separate instances of the application. When such a property cannot be determined, a command line parameter is an option.
Connector code is importable:
The flexibility to import connector code into another piece of Python code enables a setup wrapper to handle job management tasks such as:
Secrets management
Output logging
Other configurations required by the environment
A separate run function implements connector end-to-end logic. This can be imported and invoked by providing the necessary parameters. The main function that runs when invoked from the command line should only process command line inputs and environment variables for setup.
High-level code flow
The exact flow of the connector can change to meet specific requirements, but the general steps are as follows:
Process and validate configuration parameters. Ensure you have supplied all necessary values such as host names, user names, and API keys.
Call the main run function with all the required and optional parameters.
Initialize the oaaclient connection to the Veza tenant. Initializing the client verifies that the Veza URL and API key are valid before starting any application discovery.
Create an instance of the oaaclient.templates.CustomApplication to populate with the application information.
Connect to the system and perform discovery of required entities:
The discovery order for users, groups, roles, and so on can be changed based on relationships and interface.
Populate the CustomApplication instance with the identity, roles and permissions, resources, and authorization information.
Check if Provider and Data Source exist on Veza. Create them if they do not exist.
Push the application to the Data Source. The SDK creates the JSON payload from the CustomApplication instance.
Process any returned warnings or errors.
Exit.
Implementing from the example
Run the connector to validate the output in Veza.
Automate the connector to run on a regular schedule.
Code walkthrough for custom application
The code below can serve as a template and the comments explain the reasoning beyond the patterns.
oaa-app.py
#!/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()
Update the Provider Name, App Type, and Application Name based on the source system. These values will typically be set to the product name. The data source name must be unique - check that an appropriate distinguishing value such as hostname is included in the data source name. For more information on naming, see
Define any needed in the __init__ method. Properties must be defined on the CustomApplication entities before setting them.
Implement the discovery steps for the discover() function to collect Users, Groups, Roles, and any resources for the application. As the entities are discovered, add them to the CustomApplication object using the appropriate .
