cased-python

cased-python is the official Python client for Cased, a web service for adding audit trails to any application in minutes. This Python client makes it easy to record and query your Cased audit trail events. The client also includes a robust set of features to mask PII, add client-side resilience, and automatically augment events with more data.

Installation

Install from PyPI:

pip3 install cased

You can install cased-python directly from the repository. It's available for vendoring, self-hosting, or bundling with your application.

git clone https://github.com/cased/cased-python.git
python3 setup.py install

Configuration

The client can be configured using environment variables or initialized programmatically.

To send events, you use a publish key. The Python client will look for an envionment variable CASED_PUBLISH_KEY. This key can be found in your Cased dashboard.

The API key can also be provided programatically, and will take precedence over an env variable. Example environment variable usage:

$ CASED_PUBLISH_KEY="publish_test_c260ffa178db6d5953f11f747ecb7ee3" python app.py

Or programmatically:

import cased
cased.publish_key = "publish_test_c260ffa178db6d5953f11f747ecb7ee3"

You can also send your API key with each request:

import cased

cased.Event.publish({"action": "user.login"}, api_key="publish_test_c260ffa178db6d5953f11f747ecb7ee3")

(Setting an API key on a request takes precedence over both the global setting and the environmental variable setting).

To read events, you use a policy key. Set a default policy key (i.e., a key that will automatically be used unless another is given):

$ CASED_POLICY_KEY="policy_test_f764a5f252aaca986b0526b42a6f7e95"

Or programatically:

cased.policy_key = "policy_test_f764a5f252aaca986b0526b42a6f7e95"

For more advanced configuration of policy keys, see the section Policy Keys below.

How To Guide

Record your first audit trail event using the publish() function on the Event class.

import cased

cased.Event.publish({
  "event": "user.login",
  "location": "130.25.167.191",
  "request_id": "e851b3a7-9a16-4c20-ac7f-cbcdd3a9c183",
  "server": "app.fe1.disney.com",
  "session_id": "e2bd0d0e-165c-4e2a-b40b-8e2997fd7915",
  "user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36"
  "user_id": "User;1",
  "user": "mickeymouse",
})

Fetch an event

All API classes have fetch functions so you can retrieve a JSON representation of any object. For simple usage, pass the object's id. Additionally, you can pass a policy api_key if you haven't already set a default one.

import cased

cased.Event.fetch("e19a2032-f841-426c-8a13-5a938e7934a3", api_key="policy_test_f764a5f252aaca986b0526b42a6f7e95")

and you'll get back the event.

Result lists and pagination

Many objects, such as Event and Policy have list() functions. These functions return ResultsList objects of paginated results, plus pagination metadata and useful functions. For example:

events = cased.Event.list()
events.results # Get the list of the first page of results

By default, list() returns 25 items at a time. This can be adjusted, with a maximum of 50 items.

events = cased.Event.list(limit=10)

You can get metadata:

events.total_count # Total count of objects
events.total_pages # Total number of pages

You can also get urls and page numbers to paginate the results.

events.next_page_url
# > https://api.cased.com.com/api/events?page=3

events.next_page
#> 3

events.last_page_url
# > https://api.cased.com.com/api/events?page=20

events.last_page
# > 20

It is much easier to use the iterators provided by this library.

events = cased.Event.list()
iterator = events.page_iter()
for item in iterator:
  print(item)

page_iter() is generator that yields a results item per iteration. So it can be used to automatically run through paginated results in an efficient way. Like any generator it works with the ordinary for..in, and with next().

page_iter() is available for any resource class that has a list() function.

Control your returned events with policy variables

You can control what events you get back from list() by using the variables parameter. variables is a dict of fields which are then applied to a policy. For example:

events = cased.Event.list(variables={"team_id": "team-123"})

You can set multiple variables:

events = cased.Event.list(variables={"team_id": "team-123", "organization_id": "org-abc"})

Additionally, you can further limit and filter your results by using a search phrase:

events = cased.Event.list(search="user:jill", variables={"team_id": "team-123"})

More for list()

You can use some additional convenience functions:

Event.list_by_actor("jill")

Event.list_by_action("invite.created")

Pass in policy variables as well:

Event.list_by_actor("jill", variables={"team_id": "team-123"})

You can easily build a search query phrase from a Python dictionary, and then pass that to list():

from cased import Query

data = {"actor": "jill", "location": "Austria"}
my_query = Query.make_phrase_from_dict(data)

cased.Event.list(search=my_query)

Policy Keys

You may use multiple policies to read data from Cased. Each policy has its own associated API key. To make this easy, the library provides a cased.policy_keys configuration option, which lets you map arbitrary policy names to keys.

import cased

cased.policies = {
  "primary": "policy_live_1dQpY8bBgEwdpmdpVrrtDzMX4fH",
  "secondary": "policy_live_1dSHQRurWX8JMYMbkRdfzVoo62d"
}

Then use, with, for example, a list() operation.

Event.list(policy="primary")

You can still mix with policy variables, of course:

Event.list(policy="primary", variables={"team_id": "team-123"})

ReliabilityEngine and Backends

A ReliabilityEngine adds extra resilience to the client by writing audit entries to a local datastore for later processing. This can be useful if, for whatever reason, your client is unable to reached Cased. A ReliabilityEngine can queue up events for later sending to Cased.

A ReliabilityEngine has a ReliabilityBackend — right now this library includes a Redis implementation. A backend implements add(), which adds data to a datastore for later processing. You can implement your own by subclassing cased.data.reliability.AbstractReliabilityBackend.

It's very easy to set one up. You can set one globally using either a default string name (just 'redis' currently), or by using a class.

cased.reliability_backend = "redis"

or set a custom class:

cased.reliability_backend = MyCustomClass

All publish events will now also write events to that reliability backend. We recommend using a reliability backend, although it is not required.

You can also set a backend on a per-request basis, by passing a backend keyword arg to Event.publish()

Data Plugins

A DataPlugin enriches your audit events with any arbitrary additional data. Define one easily by subclassing cased.plugins.DataPlugin and implementing an additions function. Then return a dict of additional data and that data will automatically be sent to Cased along with your audit event.

When implementing a plugin, you can access the audit entry itself with self.data, in case your plugin needs to do some processing based on the audit entry data.

cased.add_plugin(MyCustomPlugin)

Here's an example — the default plugin that ships with this library:

class CasedDefaultPlugin(DataPlugin):
    def additions(self):
        return {
            "cased_id": uuid.uuid4().hex,
            "timestamp": str(
                datetime.now(timezone.utc).isoformat(timespec="microseconds")
            ),
        }

Context

You can push data to a library-provided thread-local Context dictionary, allowing you to easily build up events before sending them to Cased.

# Set some contextual information
cased.Context.update({"location": "Austria"})
cased.Context.update({"valid": True})

This information will then be included in any subsequent publish(). For example, now calling:

cased.Event.publish({"username": "blake"})

will publish:

{
  "username": "blake",
  "valid": True,
  "location": "Austria",
  ...
}

You can clear the context with:

cased.Context.clear()

Additionally, you can have cased-python automatically clear the context after every publish() action with the global setting:

cased.clear_context_after_publishing = True

Sensitive Data

You can mark audit entry fields as sensitive to mask PII. Just use the global set:

cased.sensitive_fields = {"username", "address"}

Any field, in any audit event, that matches one of those key name will be marked as sensitive when sent to Cased.

You can also mark patterns in your audit trail dail as sensitive_in order to mask PII. To do so, create a SensitiveDataHandler:

handler = SensitiveDataHandler(label="username", pattern=r"@([A-Za-z0-9_]+)

A sensitive data handler includes a label, which makes it easy to identify what kind of data is being masked. Additionally, it includes a pattern, which is a regular expression matching a pattern you want to mark as sensitive.

Add it globally:

cased.add_handler(handler)

Now any data you send that matches that pattern with will be marked as PII when sent to Cased, and masked in the Cased UI.

You can also redact sensitive data from even being sent to Cased, using the redact_before_publishing setting. When enabled, this setting redacts any configured sensitive data prior to publishing of event. Sensitive data characters will be replaced with X. Set globally:

redact_before_publishing = False

Disable Publishing

You may want to completely prevent events from being published (perhaps for testing purposes). To do so, just set:

cased.disable_publishing == True

Logging

You can enable logging for useful error, info, and debug messages:

import cased
cased.log = 'debug'

Use through Python's native logging:

import logging
logging.basicConfig()
logging.getLogger('cased').setLevel(logging.DEBUG)

Contributing

Contributions to this library are welcomed and a complete test suite is available. Tests can be run locally using the following command:

pytest

Code formatting and linting is provided by Black and Flake8 respectively, so you may want to install them locally.