Audit | Python SDK
The audit API is designed for recording a trail of application-based user activity in a scalable, tamper-proof log.
Audit client
Audit(token, config, private_key_file, public_key_info, tenant_id, logger_name, config_id)Initializes a new Audit client.
config = PangeaConfig(domain="pangea_domain")
audit = Audit(token="pangea_token", config=config)download-file
Audit.download_file()Download search results
Audit.download_results(result_id, format, request_id, return_context)Get all search results as a compressed (gzip) CSV file.
response = audit.download_results(
result_id="pas_[...]",
format=DownloadFormat.JSON,
)Export from the audit log
Audit.export(format, start, end, order, order_by, verbose)Bulk export of data from the Secure Audit Log, with optional filtering.
export_res = audit.export(verbose=False)
# Export may take several dozens of minutes, so polling for the result
# should be done in a loop. That is omitted here for brevity's sake.
try:
audit.poll_result(request_id=export_res.request_id)
except AcceptedRequestException:
# Retry later.
# Download the result when it's ready.
download_res = audit.download_results(request_id=export_res.request_id)
download_res.result.dest_url
# => https://pangea-runtime.s3.amazonaws.com/audit/xxxxx/search_results_[...]fix-consistency-proofs
Audit.fix_consistency_proofs()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.
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.
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.
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.
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")Log streaming endpoint
Audit.log_stream(data)This API allows 3rd party vendors (like Auth0) to stream events to this endpoint where the structure of the payload varies across different vendors.
data = {
"logs": [
{
"log_id": "some log ID",
"data": {
"date": "2024-03-29T17:26:50.193Z",
"type": "sapi",
"description": "Create a log stream",
"client_id": "some client ID",
"ip": "127.0.0.1",
"user_agent": "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0",
"user_id": "some user ID",
},
}
# ...
]
}
response = audit.log_stream(data)Poll result
Audit.poll_result(exception)Returns request's result that has been accepted by the server
response = service.poll_result(exception)Results of a search
Audit.results(id, limit, offset, assert_search_restriction, verify_consistency, verify_events, return_context)Fetch paginated results of a previously executed search.
response = audit.results(
id="pas_sqilrhruwu54uggihqj3aie24wrctakr",
limit=10,
offset=0,
)Tamperproof verification
Audit.root(tree_size)Returns current root hash and consistency proof.
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, return_context)Search for events that match the provided search criteria.
response = audit.search(
query="message:test",
search_restriction={'source': ["monitor"]},
limit=1,
verify_consistency=True,
verify_events=True,
)