Audit | Python SDK

The audit API is designed for recording a trail of application-based user activity in a scalable, tamper-proof log.

download-file

Audit.download_file()

Response Object

Download search results

Audit.download_results(result_id, format)

Get all search results as a compressed (gzip) CSV file.

optional parameters

result_id

ID returned by the search API.

format

Format for the records.

Response Object

URL where search results can be downloaded.

response = audit.download_results(
    result_id="pas_[...]",
    format=DownloadFormat.JSON,
)

Log an entry

Audit.log(message, actor, action, new, old, source, status, target, timestamp, verify, sign_local, verbose, tenant_id)

Create a log entry in the Secure Audit Log.

required parameters

message

str, dict

A message describing a detailed account of what happened.

optional parameters

actor

str

Record who performed the auditable activity.

action

str

The auditable action that occurred.

new

str, dict

The value of a record after it was changed.

old

str, dict

The value of a record before it was changed.

source

str

Used to record the location from where an activity occurred.

status

str

Record whether or not the activity was successful.

target

str

Used to record the specific record that was targeted by the auditable activity.

timestamp

datetime

An optional client-supplied timestamp.

verify

bool

True to verify logs consistency after response.

sign_local

bool

True to sign event with local key.

verbose

bool

True to get a more verbose response.

tenant_id

string

Used to record the tenant associated with this activity.

Response Object

A PangeaResponse where the hash of event data and optional verbose results are returned in the response.result field. Available response fields can be found in our API documentation.

log_response = audit.log(
    message="hello world",
    verbose=True,
)

Log multiple entries

Audit.log_bulk(events, sign_local, verbose)

Create multiple log entries in the Secure Audit Log.

required parameters

events

List[dict[str, Any]]

events to be logged

optional parameters

sign_local

bool

True to sign event with local key.

verbose

bool

True to get a more verbose response.

Response Object

A PangeaResponse where the hash of event data and optional verbose results are returned in the response.result field. Available response fields can be found in our API documentation.

log_response = audit.log_bulk(
    events=[{"message": "hello world"}],
    verbose=True,
)

Log multiple entries asynchronously

Audit.log_bulk_async(events, sign_local, verbose)

Asynchronously create multiple log entries in the Secure Audit Log.

required parameters

events

List[dict[str, Any]]

events to be logged

optional parameters

sign_local

bool

True to sign event with local key.

verbose

bool

True to get a more verbose response.

Response Object

A PangeaResponse where the hash of event data and optional verbose results are returned in the response.result field. Available response fields can be found in our API documentation.

log_response = audit.log_bulk_async(
    events=[{"message": "hello world"}],
    verbose=True,
)

Log an entry

Audit.log_event(event, verify, sign_local, verbose)

Create a log entry in the Secure Audit Log.

required parameters

event

dict[str, Any]

event to be logged

optional parameters

verify

bool

True to verify logs consistency after response.

sign_local

bool

True to sign event with local key.

verbose

bool

True to get a more verbose response.

Response Object

A PangeaResponse where the hash of event data and optional verbose results are returned in the response.result field. Available response fields can be found in our API documentation.

try:
    log_response = audit.log({"message": "hello world"}, verbose=True)
    print(f"Response. Hash: {log_response.result.hash}")
except pe.PangeaAPIException as e:
    print(f"Request Error: {e.response.summary}")
    for err in e.errors:
        print(f"\t{err.detail} \n")

Poll result

Audit.poll_result(exception)

Returns request's result that has been accepted by the server

required parameters

exception

AcceptedRequestException

Exception raise by SDK on the call that is been processed.

Response Object

PangeaResponse

response = service.poll_result(exception)

Results of a search

Audit.results(id, limit, offset, verify_consistency, verify_events)

Fetch paginated results of a previously executed search.

required parameters

id

string

the id of a search action, found in response.result.id

verify_consistency

bool

True to verify logs consistency

verify_events

bool

True to verify hash events and signatures

optional parameters

limit

integer

the maximum number of results to return, default is 20

offset

integer

the position of the first result to return, default is 0

Response Object

response = audit.results(
    id="pas_sqilrhruwu54uggihqj3aie24wrctakr",
    limit=10,
    offset=0,
)

Tamperproof verification

Audit.root(tree_size)

Returns current root hash and consistency proof.

optional parameters

tree_size

int

The size of the tree (the number of records). If None, endpoint will return last tree root.

Response Object

PangeaResponse[RootOutput]

response = audit.root(tree_size=7)

Search the log

Audit.search(query, order, order_by, last, start, end, limit, max_results, search_restriction, verbose, verify_consistency, verify_events)

Search for events that match the provided search criteria.

required parameters

query

str

Natural search string; list of keywords with optional <option>:<value> qualifiers. The following optional qualifiers are supported:

• action
• actor
• message
• new
• old
• status
• target

verify_consistency

bool

True to verify logs consistency

verify_events

bool

True to verify hash events and signatures

optional parameters

order

SearchOrder

Specify the sort order of the response.

order_by

SearchOrderBy, str

Name of column to sort the results by.

last

str

Optional[str] = None,

start

datetime

An RFC-3339 formatted timestamp, or relative time adjustment from the current time.

end

datetime

An RFC-3339 formatted timestamp, or relative time adjustment from the current time.

limit

int

Optional[int] = None,

max_results

int

Maximum number of results to return.

search_restriction

dict

A list of keys to restrict the search results to. Useful for partitioning data available to the query string.

verbose

bool

If true, response include root and membership and consistency proofs.

Response Object

A PangeaResponse[SearchOutput] where the first page of matched events is returned in the response.result field. Available response fields can be found in our API documentation. Pagination can be found in the search results endpoint.

response = audit.search(
    query="message:test",
    search_restriction={'source': ["monitor"]},
    limit=1,
    verify_consistency=True,
    verify_events=True,
)