Sample Apps

Usage examples for the oaaclient Python SDK

The examples on this page can be used as a starting point for building new connectors for custom applications. You should also refer to the current list of Veza-supported connectors for real-world examples, and to check if an integration already exists for your use case.

To run the code, you will need to export environment variables for the Veza URL, user and API keys, for example:

export OAA_TOKEN="xxxxxxx"
export VEZA_URL="https://myveza.vezacloud.com"
./sample-app.py

• Sample Application

• Sample Identity Provider

• Sample IdP from CSV file

Sample Application

The most common OAA use case is modeling local identities, data sources, and permissions within an application containing sensitive information such as a database, ticket desk, or SCM platform. This sample app uses the CustomApplication class to create an OAA payload for a typical application that includes users, groups, roles and resources.

sample-app.py

#!env python3
"""
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.

Example of using the `CustomApplication` class to model a typical application where users and groups are assigned
permissions to the application or resources.

If you want to run the code you will need to export environment variables for the Veza URL, user and API keys.

```
export OAA_TOKEN="xxxxxxx"
export VEZA_URL="https://myveza.vezacloud.com"
./sample-app.py
```

"""

from oaaclient.client import OAAClient, OAAClientError
from oaaclient.templates import CustomApplication, OAAPermission
import os
import sys


def main():

    # OAA requires an API token, which you can generate from your Veza user profile
    # Export the API token, and Veza URL as environment variables
    # Making them available to your connector in this way keeps credentials out of the source code
    veza_api_key = os.getenv('VEZA_API_KEY')
    veza_url = os.getenv('VEZA_URL')
    if None in (veza_url, veza_api_key):
        print("Unable to local all environment variables")
        sys.exit(1)

    # Instantiates a client connection. The client will confirm the credentials and Veza URL are valid
    # Checking this early helps prevents connection failures in the final stage
    veza_con = OAAClient(url=veza_url, api_key=veza_api_key)

    # Create an instance of the OAA CustomApplication class, modeling the application name and type
    # `name` will be displayed in the Veza UI
    # `application_type` should be a short key reflecting the source application authorization is being modeled for
    # You can use the same type for multiple applications
    custom_app = CustomApplication(name="Sample App", application_type="sample")

    # In the OAA payload, each permission native to the custom app is mapped to the Veza effective permission (data/nondata C/R/U/D).
    # Permissions must be defined before they can be referenced, as they are discovered or ahead of time.
    # For each custom application permission, bind them to the Veza permissions using the `OAAPermission` enum:
    custom_app.add_custom_permission("admin", [OAAPermission.DataRead, OAAPermission.DataWrite])
    custom_app.add_custom_permission("operator", [OAAPermission.DataRead])

    # Create resources and sub-resource to model the entities in the application
    # To Veza, an application can be a single entity or can contain resources and sub-resources
    # Utilizing resources enables tracking of authorizations to specific components of the system being modeled
    # Setting a `resource_type` can help group entities of the same type for reporting/queries
    entity1 = custom_app.add_resource(name="Entity1", resource_type="thing", description="Some entity in the application")
    entity2 = custom_app.add_resource(name="Entity2", resource_type="thing", description="Another entity in the application")
    other = custom_app.add_resource(name="Other", resource_type="other", description="Something totally different")

    # Sub-resources can be added to any resource (including other sub-resources)
    child1 = entity1.add_sub_resource(name="Child 1", resource_type="child", description="My information about resource")
    child1.add_sub_resource(name="Grandchild 1", resource_type="grandchild", description="My information about resource")

    # Any users and groups local to the application can be defined.
    # IdP users can be mapped directly without defining them in the OAA application (see below)
    custom_app.add_local_user("bob")
    # A local user can be associated with an IdP user by adding an identity to the user:
    custom_app.local_users["bob"].add_identity("[email protected]")
    # Identities, groups and roles can be assigned to local users at creation or after (groups and roles must exist):
    jane = custom_app.add_local_user("jane", identities="[email protected]")

    # when adding a user the new user is returned and can updated if needed, there are multiple built in properties that can be set for a user
    jane.is_active = False
    jane.created_at = "2022-01-26T20:48:12.460Z"

    # Define a local group and add a user to it
    custom_app.add_local_group("admins")
    custom_app.local_users["jane"].add_group("admins")

    # adding local users and groups requires that the name not already exist
    if "yan" not in custom_app.local_users:
        custom_app.add_local_user("yan")

    # For each Identity (user, group, IdP) assign permissions to the application or resource.
    # The identities (users, groups) permissions and resources must already be defined

    # To add a permission directly to the application use `apply_to_application=True`
    custom_app.local_users["bob"].add_permission(permission="operator", apply_to_application=True)
    custom_app.local_groups["admins"].add_permission(permission="admin",  apply_to_application=True)

    # You can describe specific permissions to individual resources or sub-resources:
    custom_app.local_users["yan"].add_permission(permission="operator", resources=[entity1, child1])

    # Authorization can also be created directly for an IdP identity
    custom_app.add_idp_identity("[email protected]")
    # resources can also be referenced by name from the application model
    custom_app.idp_identities["[email protected]"].add_permission(permission="admin", resources=[custom_app.resources['Entity1']])

    # Once all authorizations have been mapped, the final step is to publish the app to Veza
    # Connect to the API to Push to Veza, define the provider and create if necessary:

    provider_name = "Sample"
    provider = veza_con.get_provider(provider_name)
    if provider:
        print("-- Found existing provider")
    else:
        print(f"++ Creating Provider {provider_name}")
        provider = veza_con.create_provider(provider_name, "application")
    print(f"-- Provider: {provider['name']} ({provider['id']})")

    # Push the metadata payload:

    try:
        response = veza_con.push_application(provider_name,
                                               data_source_name=f"{custom_app.name} ({custom_app.application_type})",
                                               application_object=custom_app,
                                               save_json=False
                                               )
        if response.get("warnings", None):
            # Veza may return warnings on a successful uploads. These are informational warnings that did not stop the processing
            # of the OAA data but may be important. Specifically identities that cannot be resolved will be returned here.
            print("-- Push succeeded with warnings:")
            for e in response["warnings"]:
                print(f"  - {e}")
    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(f"  -- {d}", file=sys.stderr)
    return


if __name__ == '__main__':
    main()

Sample Identity Provider

Customers that implement a custom Identity Provider (or one that doesn't have a native Veza integration) can use the OAA Custom IdP template to describe federated users and groups. This sample app generates a custom payload containing users, groups, and identity metadata, using the CustomIdPProvider class.

The example also demonstrates OAA support for AWS role entitlements, for scenarios where users can assume AWS roles in ways that can't be discovered by Veza's native AWS integration.

sample-idp.py

#!env python3
"""
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.

Example of using the `CustomIdPProvider` class to model an identity provider as a source of users.

If you want to run the code you will need to export environment variables for the Veza URL, user and API keys.

```
export VEZA_API_KEY="xxxxxxx"
export VEZA_URL="https://myveza.vezacloud.com"
./sample-idp.py
```

Since the example includes fake AWS ARNs that Veza will not have discovered the expected output will
contain warning like "Cannot find IAM role by names ..."
"""

from oaaclient.client import OAAClient, OAAClientError
from oaaclient.templates import CustomIdPProvider, OAAPropertyType
import os
import sys


def main():

    # OAA requires a Veza API key, which you can generate from Administration > API Keys
    # Export the API key, and Veza URL as environment variables
    # Making them available to your connector in this way keeps credentials out of the source code
    veza_api_key = os.getenv("VEZA_API_KEY")
    veza_url = os.getenv("VEZA_URL")
    if not veza_api_key or not veza_url:
        print("Unable to load VEZA_API_KEY and VEZA_URL from OS")
        sys.exit(1)

    # Instantiates a client connection. The client will confirm the credentials and Veza URL are valid
    # Checking this early helps prevents connection failures in the final stage
    veza_con = OAAClient(url=veza_url, api_key=veza_api_key)

    # create a CustomIdPProvider to represent your IdP. This can be named generically or specific to the environment if you have
    # multiple namespaces to model. idp_type will typically be the technology/vendor for the provider.
    idp = CustomIdPProvider("My IdP", domain="example.com", idp_type="custom_idp")

    # add users to the idp, properties for users can be set during creation or updated after
    idp.add_user("mrichardson", full_name="Michelle Richardson", email="[email protected]")

    evargas_user = idp.add_user("evargas")
    evargas_user.full_name = "Elizabeth Vargas"
    evargas_user.email = "[email protected]"

    # users and groups can have optional identity property. The identity serves as the unique reference identifier across
    # Veza. If omitted CustomIdPProvider will automatically populate the identity with the name
    idp.add_user("willis", email="[email protected]", identity="cwilliams")

    # OAA can support custom properties for users to track additional metadata unique to the environment
    # to use custom properties the property must first be defined and given a type, then can be set for the individual entity
    idp.property_definitions.define_user_property("region", OAAPropertyType.STRING)
    idp.property_definitions.define_user_property("is_contractor", OAAPropertyType.BOOLEAN)

    idp.users['willis'].set_property("region", "NorthAmerica")
    idp.users['willis'].set_property("is_contractor", True)

    # Create Groups
    idp.add_group("developers")
    idp.add_group("sec-ops")
    idp.add_group("everyone", full_name="All Company Employees")

    # users can be added to groups using the add_group function, users can be added to multiple groups in a single call
    for username in idp.users:
        idp.users[username].add_groups(["everyone"])

    evargas_user.add_groups(["developers", "sec-ops"])
    idp.users["mrichardson"].add_groups(["developers"])

    # Veza CustomIdP supports tracking the AWS Roles a user can assume. For users who can assume roles Veza can calculate
    # their effective permissions to AWS resources based on the role(s)
    # roles are added by ARN
    idp.users["mrichardson"].add_assumed_role_arns(["arn:aws:iam::123456789012:role/role001", "arn:aws:iam::123456789012:role/role002"])

    # After adding users and groups, the IdP information is pushed to Veza using the OAA API
    provider_name = "Sample-IdP"
    provider = veza_con.get_provider(provider_name)
    if provider:
        print("-- Found existing provider")
    else:
        print(f"++ Creating Provider {provider_name}")
        provider = veza_con.create_provider(provider_name, "identity_provider")
    print(f"-- Provider: {provider['name']} ({provider['id']})")

    # Push the metadata payload:
    try:
        response = veza_con.push_application(provider_name,
                                               data_source_name=f"{idp.name} ({idp.idp_type})",
                                               application_object=idp,
                                               save_json=True
                                               )
        if response.get("warnings", None):
            # Veza may return warnings on a successful uploads. These are informational warnings that did not stop the processing
            # of the OAA data but may be important, for example: AWS role ARNs assigned to users that Veza has not discovered
            print("-- Push succeeded with warnings:")
            for e in response["warnings"]:
                print(f"  - {e}")
    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(f"  -- {d}", file=sys.stderr)


if __name__ == '__main__':
    main()

Sample IdP from CSV input

Sample app for importing Custom Identity Provider users from a CSV file.

This example can be used as a simple starting point for an OAA data source importer. Demonstrates use of CustomIdPProvider to create IdP users and assign properties using an input file with with the column headings:

identity,name,full_name,is_active,is_guest,manager_id

IdPImportCSV.py

#!env python3
"""Example of using `oaaclient` to import users from a CSV file.

Reads user properties from the input file, populates an OAA payload,
and creates a new OAA identity provider containing the imported users.

Expected CSV headers are `identity,name,full_name,is_active,is_guest,manager_id`
Can be updated to match custom column headings or apply custom properties.

Example:
    ```
    export VEZA_URL=<Veza URL>
    export VEZA_API_KEY=<Veza API key>
    ./idp-importer-csv.py --provider MyIdpProvider --datasource MyDatasource  ./my-users.csv

    ```

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.

"""

from oaaclient.client import OAAClient, OAAClientError
from oaaclient.templates import CustomIdPProvider
import click
import csv
import logging
import os
import sys


logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
log = logging.getLogger()


def load_users(idp: CustomIdPProvider, source: str) -> None:
    """Populate the idp with user from the source csv file"""

    log.info(f"Loading users from {source}")
    with open(source) as f:
        for r in csv.DictReader(f):
            # get the identity, name, full_name and email
            try:
                identity = r["identity"]
                name = r["name"]
                full_name = r["full_name"]
                email = r["email"]
            except KeyError as e:
                log.error(f"Incorrect CSV column headers, missing column {e}")
                sys.exit(1)

            # create a new IDP user
            new_user = idp.add_user(name=name, full_name=full_name, email=email, identity=identity)

            # set the user to active and guest or not, look for strings like true/false or yes/no
            if r.get("active"):
                active_string = r["active"].lower()
                if active_string in ["true", "yes"]:
                    new_user.is_active = True
                elif active_string in ["false", "no"]:
                    new_user.is_active = False

            if r.get("is_guest"):
                guest_string = r["is_guest"].lower()
                if guest_string in ["true", "yes"]:
                    new_user.is_guest = True
                elif guest_string in ["false", "no"]:
                    new_user.is_guest = False

            # if the manager id column is filled in, set the new users manager
            if r.get("manager_id"):
                new_user.manager_id = r.get("manager_id")

    return

@click.command()
@click.option("--provider", required=True)
@click.option("--datasource", required=True)
@click.option("--save-json", is_flag=True)
@click.argument("file", required=True)
def main(provider, datasource, save_json, file):

    # load the Veza URL and API key from the environment
    veza_url = os.getenv("VEZA_URL")
    veza_api_key = os.getenv("VEZA_API_KEY")

    if not (veza_url or veza_api_key):
        log.error("Must set VEZA_URL and VEZA_API_KEY")
        sys.exit(1)

    try:
        log.info("Testing Veza credentials")
        veza_con = OAAClient(url=veza_url, api_key=veza_api_key)
    except OAAClientError as e:
        log.error("Unable to connect to Veza API")
        log.error(e)
        sys.exit(1)

    # create a new provider if the provider doesn't already exist
    if not veza_con.get_provider(provider):
        log.info(f"Creating new provider {provider}")
        veza_con.create_provider(provider, "identity_provider")

    # Configure the Custom IdP
    idp = CustomIdPProvider("Custom IdP", idp_type="custom_idp", domain="example.com", description=None)

    # load the user from
    load_users(idp, file)

    try:
        # push the IdP to Veza, log and errors or warnings
        log.info("Sending to Veza")
        response = veza_con.push_application(provider, datasource, idp, save_json=save_json)
        if response.get("warnings", None):
            log.warning("Push succeeded with warnings")
            for w in response['warnings']:
                log.warning(w)
        log.info("Success")
    except OAAClientError as e:
        log.error(f"Error during push {e.message} {e.status_code}")
        if e.details:
            log.error(e.details)
        sys.exit(1)

    return

if __name__ == "__main__":
    main()