Skip to content

tomguyatt/ppa-api

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

80 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Python 3.8 Build Status codecov

Osirium PPA API

A Python package for integrating with Osirium PPA's public API.

Getting Started

Example Scripts

The examples folder contains scripts demonstrating the following operations.

Users

Tasks

Images

Code Snippets:

Client

Users

Tasks

Delayed Tasks

Images

Other

Getting Started

Installation

The package is on pypi and can be installed via pip.

Current Stable Release

pip install ppa-api==2.4.1

Version Requirements

  • PPA Appliance 2.7.1 or later
  • Python 3.8 or later

Authentication & Permissions

You'll need to generate an API key in PPA to use this package.

The API key will have the same permissions as the PPA user it was generated by.

The following snippet requires permission to view users in PPA.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.users()

If the code run without the relevant PPA permissions, it will raise PermissionDenied.

PermissionDenied: Forbidden (403). The user associated with this API key does not have the required users permissions in PPA.

Certificates & Proxies

You can supply a proxy address to the PPA client instance using the proxy keyword argument.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key, proxy="my-https-proxy.net")

To use a custom certificate, supply the verify keyword argument with a path to a CA bundle.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key, verify="/path/to/ca-bundle")

By default no proxy or custom certificate will be used, & the server's TLS certificate will be verified.

To skip certificate verification completely, supply the verify keyword argument as False.

The PPA version is checked during PPAClient instantiation, & any proxy/certificate errors will be raised early.

Code Examples

Client

Create Client Instance

Creating a PPA client instance requires an address & api_key.

If there is no trusted certificate on PPA, you'll need to set the verify keyword argument to False.

Trusted Certificate

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)

Untrusted Certificate

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key, verify=False)

Skipping verification may generate the following log lines whenever a request is made:

InsecureRequestWarning: Unverified HTTPS request is being made to host '1.2.3.4'. Adding certificate verification is strongly advised. See: https://urllib3.readthedocs.io/en/latest/advanced-usage.html#ssl-warnings

You can silence these by importing urllib3 & disabling the warning.

import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

Users

List All Users

This snippet will list all current users in the PPA appliance.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.users()

A list of Users will be returned.

[
    User(
        active=False,
        authenticated_at='2021-02-11T14:28:33.535783+00:00',
        deleted_at=None,
        email='cloud.engineer@domain.com',
        id=3,
        name='cloud engineer',
        username='domain\\cloud.engineer'
    )
]

List Licensed Users

This snippet will list all current users who can start tasks.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.licensed_users()

A list of Users will be returned.

[
    User(
        active=True,
        authenticated_at='2021-02-12T12:03:17.515093+00:00',
        deleted_at=None,
        email='service.operator@domain.com',
        id=3,
        name='service operator',
        username='domain\\service.operator'
    )
]

List Deleted Users

Use the following method to list all deleted users.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.deleted_users()

A list of Users will be returned

Get Specific User

This snippet finds the user with the supplied username.

The search is case insensitive & the domain name is not required.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.user_by_username("service.operator")

A single User is returned if the user is found.

If the user was not found, the method will return None.

Tasks

Auditing Task History

Listing All Tasks

You can view the history of all tasks visible to the API key's associated user.

The following snippet gets all visible tasks.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.tasks()

Listing Tasks Started By Me

The following snippet get all visible tasks started by the API key's associated user.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.tasks_started_by_me()

Both methods return a list of Tasks.

[
    Task(
        author='John Doe',
        cancelled_at=None,
        cancelled_by=None,
        duration='00:01:33.945833',
        exit_code=0,
        exit_message=None,
        id=3,
        image='Add User to Groups',
        image_id=7,
        image_uuid='59787351-16ac-4a85-a812-007b1dcde027',
        is_owner=True,
        is_running=False,
        started_at='2021-02-08T10:12:25.404705+00:00',
        state='success',
        stopped_at='2021-02-08T10:13:59.350538+00:00',
        timed_out=False,
        timeout=15,
        username='domain\\john.doe',
        uuid='b007905b-6c6a-4037-b4cd-b2d4a7fd3c0d'
    )
]

Starting Tasks

This package allows you to start the latest deployed revision of a task.

Tasks can be started either synchronously or asynchronously.

Synchronous

The snippet below starts the latest deployed revision of a task, & waits up to 10 minutes for it to complete.

A Task is returned if it completes within the time limit, otherwise WaitTimeout is raised.

You can supply a custom timeout in seconds using the timeout keyword argument.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
task = ppa.start_task("Domain Admins Audit")

See the asynchronous section below for a non-blocking approach.

Asynchronous

This snippet starts the latest deployed revision of a task & immediately returns a Task.

The state is checked periodically using the Task uuid attribute.

import time

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
task = ppa.start_task_async("Domain Admins Audit")

for i in range(0, 10):
    if ppa.task_running(task.uuid):
        print("The task is still running.")
        time.sleep(30)
        continue
    break
else:
    raise Exception("The task has not completed after 5 minutes.")

print(ppa.get_task_result(task.uuid))

Undeployed Tasks

Starting an undeployed task will raise ImageNotDeployed.

ImageNotDeployed: The task cannot be started as image 'example task' is not deployed.

Checking Task State

The state of a task can be checked a few different ways.

Did a task succeed?

ppa.task_succeeded(task.uuid)

The easiest way to whether a task succeeded.

Returns a bool or raises TaskStillRunning.

Is a task running?

ppa.task_running(task.uuid)

The recommended method to use while waiting for a task that was started asynchronously.

Returns a bool.

Get a Task using its UUID

ppa.task_by_uuid(task.uuid)

Returns a Task reflecting its current state.

Task(
    author='John Doe',
    cancelled_at=None,
    cancelled_by=None,
    deployed=True,
    duration='00:01:33.945833',
    exit_code=0,
    exit_message=None,
    id=3,
    image='Add User to Groups',
    image_id=7,
    image_uuid='59787351-16ac-4a85-a812-007b1dcde027',
    is_owner=True,
    is_running=False,
    started_at='2021-02-08T10:12:25.404705+00:00',
    state='success',
    stopped_at='2021-02-08T10:13:59.350538+00:00',
    timed_out=False,
    timeout=15,
    username='domain\\john.doe',
    uuid='b007905b-6c6a-4037-b4cd-b2d4a7fd3c0d'
)

Get Task Result

This is the only method that returns both the task state & its JSON result.

ppa.get_task_result(task.uuid)

Returns a TaskResult if the task saved a JSON result.

TaskResult(
    exit_code=0,
    exit_message=None,
    result_json={"user_created": True, "group_memberships": None},  # This is saved by the task when it runs.
    state='success'
)

If the task didn't save a JSON result, NoData will be raised.

NoData: No result data was saved by task with UUID '2f35cbe5-75af-4a68-af1b-b3b61c231588'.

Supplying Payloads

You can supply a JSON payload to a task with the payload keyword argument.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
task = ppa.start_task(
    "Audit Untagged EC2 Instances",
    payload={
        "environment": "development",
        "team": "infrastructure"
    }
)

If the supplied payload cannot be converted to JSON, ParameterError will be raised.

ParameterError: The supplied payload cannot be converted to JSON.

Delayed Tasks

The delayed tasks feature will be introduced in PPA version 2.8.0 (May 2021).

Install the latest development candidate of this package (2.1.0-rc2 or later) to use this functionality.

Auditing Delayed Tasks

Listing All Delayed Tasks

The following snippet gets all delayed tasks visible to the API key's associated user.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.delayed_tasks()

Listing Delayed Tasks Created By Me

The following snippet gets all delayed tasks created by the API key's associated user.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.tasks_delayed_by_me()

Listing Pending Delayed Tasks

The following snippet gets all pending delayed tasks visible to the API key's associated user.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.pending_delayed_tasks()

Listing Processed Delayed Tasks

The following snippet gets all processed delayed tasks visible to the API key's associated user.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.processed_delayed_tasks()

All the methods above return a list of Delayed Tasks.

[
    DelayedTask(
        description='Remove John Smith from Remote Desktop Users',
        id=7,
        image='Remove User From Groups',
        image_id=1,
        is_owner=True,
        is_pending=False,
        payload={"user": "john.smith", "groups": ["Remote Desktop Users"]},
        start_time='2021-03-25T10:33:43.244374',
        task_uuid='50d10ddc-be19-464d-8d5f-66b17ac38978',
        timezone='Etc/UTC',
        trigger='api',
        username='example.user'
    )
]

Creating Delayed Tasks

This package allows you to create a delayed task that will run after a certain number of seconds.

Without Payload

The following snippet creates a new delayed task in PPA.

After 10 minutes PPA will start the latest deployed revision of the Domain Admins Audit task.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
delayed_task = ppa.delay_task(
    "Domain Admins Audit",
    description="Delayed Domain Admins audit task",
    delay=600
)

With Payload

You can supply a payload to a delayed task the same way you pass them to a standard task.

The following snippet creates a new delayed task in PPA, along with a payload for the task to use when it runs.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
delayed_task = ppa.delay_task(
    "Remove User From Groups",
    description="Remove John Smith from Remote Desktop Users",
    delay=600,
    payload={"user": "john.smith", "groups": ["Remote Desktop Users"]}
)

Images

The images API endpoint represents the Inventory page in PPA.

You can use the API to list images visible to the API key's associated user.

For an image to be visible to a user, one of the following must be true:

  • The user has permission to see all tasks
  • The image is owned by the user
  • The image has been delegated to a group the user is in

You can check the deployed attribute to determine if an image is deployed or not.

Listing Latest Images

This snippet will list the latest version of each visible image.

In this context latest means one of the following:

  • The current deployed revision (if the image has been deployed)
  • The latest built revision (if the image has not been deployed)
from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.images(latest=True)

A list of Images will be returned.

[
    Image(
        author='John Doe',
        description='Recertify members of your Active Directory groups.',
        deployed=True,
        groups=0,
        id=6,
        name='Group Recertification',
        owner='domain\\john.doe',
        tags=['active directory', 'group', 'recertification'],
        updated_at='2021-01-21T15:28:41.463558+00:00',
        uuid='b2d8d266-03da-49ce-8a4a-836fe2fed6e9'
    )
]

Listing All Images

This snippet will list all revisions of each image visible to the API key's associated user.

from ppa_api.client import PPAClient

ppa = PPAClient(address, api_key=api_key)
ppa.images()

A list of Images will be returned.

Return Types

You can use this package to interact with PPA images, tasks, & users.

Methods that only fetch a single record will return a named tuple.

Methods that can fetch multiple records will return a list of named tuples, even if only 1 item is received.

Named tuples are immutable & their values can be accessed using dot notation.

See Return Fields for the fields available in each named tuple.

Return Fields

The fields available for each type of named tuple are listed below.

User

  • active
  • authenticated_at
  • deleted_at
  • email
  • id
  • name
  • username

Task

  • author
  • cancelled_at
  • cancelled_by
  • deployed
  • duration
  • exit_code
  • exit_message
  • id
  • image
  • image_id
  • image_uuid
  • is_owner
  • is_running
  • result_json
  • started_at
  • state
  • stopped_at
  • timed_out
  • timeout
  • username
  • uuid

DelayedTask

  • description
  • id
  • image
  • image_id
  • is_owner
  • is_pending
  • payload
  • start_time
  • task_uuid
  • timezone
  • trigger
  • username

TaskResult

  • exit_code
  • exit_message
  • result_json
  • state

Image

  • author
  • deployed
  • description
  • groups
  • id
  • name
  • owner
  • tags
  • updated_at
  • uuid

Exceptions

The following custom exceptions can be raised by this package.

CreditsRequired

One of the start task methods was called with no licensed credits remaining.

AuthenticationFailed

Authentication to the PPA Appliance failed using the supplied API key.

VersionError

The target PPA Appliance is older than v2.7.1.

TaskStillRunning

The task_succeeded method was called on a task that is still running.

WaitTimeout

A task is started synchronously & does not finish within the time limit.

ImageNotDeployed

One of the start task methods was called with the name of an undeployed image.

NoData

The get_task_result method was called on a task that has not saved a JSON result.

NotFound

A non-existent endpoint was called or a supplied uuid does not exist.

Usually re-raised as a more meaningful exception.

NoImageFound

An invalid uuid was supplied to an image method.

NoTaskFound

An invalid uuid was supplied to a task method.

PermissionDenied

The API key does not have permission in the PPA Appliance to perform the operation.

ParameterError

Either a uuid or payload was supplied in an invalid format.

ServerError

The PPA appliance responded with status code 500.

RequestError

The PPA appliance responded with status code 400.

UnhandledRequestError

The PPA Appliance responded with an unhandled status code (unlikely).

InvalidResponse

The PPA Appliance responded with invalid JSON (very unlikely).

About

A Python package for integrating with Osirium PPA's public API

Resources

License

Stars

Watchers

Forks

Packages

No packages published