scanner-client

This is the Python SDK for the Scanner API. It is autogenerated from an OpenAPI v3 spec.

You can use the SDK for use cases like these:

• Execute queries on your log files in S3 that have been indexed by Scanner.
• Create, read, and update detection rules
• Manage event sinks, which are destinations for detection alerts.

Documentation

You can view the API documentation for the Scanner API here.

Usage

To install the SDK, run:

pip install scanner-client

Create the client by passing in the API URL and API key, which you can get from Settings > API Keys in the Scanner UI.

import os
from scanner_client import Scanner

scanner = Scanner(
    api_url=os.environ["SCANNER_API_URL"],
    api_key=os.environ["SCANNER_API_KEY"],
)

Synchronously query logs over last 30 days

import os
import time

from datetime import datetime, timezone, timedelta
from scanner_client import Scanner

scanner = Scanner(
    api_url=os.environ["SCANNER_API_URL"],
    api_key=os.environ["SCANNER_API_KEY"],
)

end_time = datetime.now(tz=timezone.utc)
start_time = end_time - timedelta(days=30)

query_text = """
	%ingest.source_type: "aws:cloudtrail"
	eventSource: "s3.amazonaws.com"
	| stats by eventName
"""

# Run blocking query, which runs for up to 60 seconds and returns results.

response = scanner.query.blocking_query(
    query_text=query_text,
    start_time=start_time.isoformat()
    end_time=end_time.isoformat()
)
print(response.results)

# Run non-blocking query and periodically checking for completion.
# Can run for 15 minutes.

qr_id = scanner.query.start_query(
    query_text=query_text,
    start_time=start_time.isoformat()
    end_time=end_time.isoformat()
).qr_id

while True:
    print("Checking query progress")
    query_progress = scanner.query.query_progress(qr_id)
    if query_progress.is_completed:
        print(query_progress.results)
        break

    time.sleep(1)

# Run non-blocking query and return results. This handles checking
# query progress and returns the results once the query is complete.

response = scanner.query.start_query_and_return_results(
    query_text=query_text,
    start_time=start_time.isoformat()
    end_time=end_time.isoformat()
)
print(response.results)

Async Scanner

The AsyncScanner class is also available for use with asyncio. All of the API methods are coroutines and can be awaited.

import asyncio
from scanner_client import AsyncScanner

# ...

scanner = AsyncScanner(
    api_url=os.environ["SCANNER_API_URL"],
    api_key=os.environ["SCANNER_API_KEY"],
)

# ...

response = await scanner.query.blocking_query(
    query_text=query_text,
    start_time=start_time.isoformat()
    end_time=end_time.isoformat()
)
print(response.results)