Elements API Cookbook

from json import *
from base64 import b64encode

import requests

API_URL = 'https://api.connect.withsecure.com' 

def authenticate(client_id, client_secret):
    auth_header = b64encode(bytes(client_id + ':' + client_secret, 'utf-8')).decode('utf-8')
    headers = {
        'Content-type': 'application/x-www-form-urlencoded;charset=UTF-8',
        'Accept': 'application/json',
        'Authorization': 'Basic ' + auth_header,
        'User-Agent': 'my-script' # each request must contain User-Agent header
    }
    scopes = ['connect.api.read', 'connect.api.write'] # authenticated client can read
                                                       # and write data
    
    data = {'grant_type': 'client_credentials', 'scope': ' '.join(scopes)}
    
    response = requests.post(API_URL + '/as/token.oauth2', data=data, headers=headers)
    if response.ok:
        res_body = response.json() 
        return res_body['access_token']
    else:
        print('Response=' + response.text)
        print('Headers=' + str(response.headers))

        # Each response contains header `X-Transaction`. It can be send to Elements API
        # support team in order to help investigation of possible errors
        print('Transaction-id=' + response.headers.get('X-Transaction'))
        raise Exception('Authentication failed')

import requests
from authentication import authenticate

API_URL = "https://api.connect.withsecure.com"
DEVICES_PATH = "/devices/v1/devices"

def get_devices(auth_token,
                organization_id=None, next_page=None):
    headers = {
        "Accept": "application/json", # always use that header if you expect JSON response
        "Authorization": "Bearer " + auth_token,
        "User-Agent": "my-script"
    }

    params = {"limit": 10} # use `limit` to define size of single page
    
    params["protectionStatusOverview"] = "critical"
    # with parameter `protectionStatusOverview` we can read select devices with given
    # protection status.

    if organization_id:
        # `organization_id` parameter is optional. If authenticated client belongs to
        # partner organization, it can use `organization_id` to select devices, that
        # belong to given organization
        params["organizationId"] = organization_id
        
    if next_page:
        params["anchor"] = next_page # add `anchor` to query parameters to read next page
                                     # as value use property `nextAnchor` from last response
                                     # if `nextAnchor` is not present in response it means
                                     # that query returned last page

    response = requests.get(API_URL + DEVICES_PATH, params=params, headers=headers)

    if not response.ok:
        print("Error", response.text)
        sys.exit(0)

    body = response.json()
    items = body["items"] # if any device matching query is found then `items`
                              # list is empty

    for d in items:
        print("Device name={}, status={}, company={}".format(d["name"], d["protectionStatusOverview"],
                                                             d["company"]["name"]))    
    if "nextAnchor" in body:
        # if `nextAnchor` is present in response, client can use it to read next page
        # of query result. For example by calling function
        # get_devices_with_status(client_id, client_secret, next_page=response["nextAnchor"])
        print("Link to next page=", body["nextAnchor"])
        return body["nextAnchor"]
    else:
        # if `nextAnchor` is missing in response it means, that client received last page
        # of query results
        print("Last page")
        return None

def list_all_devices(client_id, client_secret, organization_id=None):
    auth_token = authenticate(client_id, client_secret)
    next_page = True
    anchor = None
    while next_page:
        anchor = get_devices(auth_token, organization_id, anchor)
        next_page = anchor is not None

from datetime import datetime, timezone
from time import sleep

import requests

from authentication import authenticate

API_URL = 'https://api.connect.withsecure.com' 
EVENTS_PATH = "/security-events/v1/security-events"

def get_events_page(auth_token, min_date, org_id=None, next_page=None):
    headers = {
        "Accept": "application/json", # always use that header if you expect JSON response
        "Authorization": "Bearer " + auth_token,
        "User-Agent": "my-script",
        "Content-type": "application/x-www-form-urlencoded"
    }
    
    params = {"limit": 200,
              "engineGroup": ["epp", "edr"],
              "persistenceTimestampStart": min_date, # read all events created AFTER `min_date`
              "order": "asc", # ascending order
    }

    # `min_date` represents timestamp of last event in previous page
    # That event should not be included in next response. When `exclusiveStart=true` is set
    # then `persistenceTimestampStart` is used as exclusive start of time range.
    params["exclusiveStart"] = "true"

    if next_page:
        params['anchor'] = next_page

    if org_id:
        params["organizationId"] = org_id

    response = requests.post(API_URL + EVENTS_PATH, data=params, headers=headers)

    if not response.ok:
        print("Error", response.text)
        raise Exception("Request error")

    return response.json()
    
def get_events_after(auth_token, last_date, org_id):
    # Elements API accepts timestamps in [RFC 3339](https://datatracker.ietf.org/doc/html/rfc3339#section-5.6)
    # format: YYYY-MM-DDThh:mm:ss.SSS
    last_date_str = last_date.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    next_page = None
    fetch_page = True
    print("Reading events created after {}".format(last_date_str))
    while fetch_page:
        page = get_events_page(auth_token, last_date_str, org_id, next_page)
        next_page = page.get("nextAnchor")

        # If `nextAnchor` is present we should fetch next page in next iteration
        fetch_page = next_page is not None
        for event in page["items"]:
            print("EventId={}, EventTs={}".format(event["id"],
                                                  event["persistenceTimestamp"]))
            # updating `last_date` with time when event was persisted
            # when iteration ends, `last_date` will be equal to `persistenceTimestamp`
            # of last received event
            last_date_str = event["persistenceTimestamp"]

    # Parse timestamp of last received event after reading all
    # events, that were created after `min_dt`
    return datetime.strptime(last_date_str, "%Y-%m-%dT%H:%M:%S.%fZ")

def poll_security_events(client_id, client_secret, poll_interval, org_id=None):
    # initialize start date with current timestamp
    last_date = datetime.now(timezone.utc)

    # start infinite loop
    while True:
        # obtain authentication token 
        auth_token = authenticate(client_id, client_secret)

        # read all events created after `last_date`. After reading all events
        # that variable is updated with timestamp of last events. `last_date` will be
        # used in next iteration
        last_date = get_events_after(auth_token, last_date, org_id)
        print("Last date", last_date)

        # execute next iteration every `poll_interval`
        sleep(poll_interval)

from datetime import datetime, timezone, timedelta
import requests
from authentication import authenticate

API_URL = "https://api.connect.withsecure.com"
PATH = "/security-events/v1/security-events"

def count_by_engine_last_week(client_id, client_secret, org_id):
    end = datetime.now(tz=timezone.utc)
    start = end - timedelta(days=7)
    stats = count_by_engine(client_id, client_secret, start, end, org_id)
    fmt = "|{}|{}|"
    print(fmt.format("Engine", "Events count"))
    for row in stats["items"]:
        print(fmt.format(row["engine"], row["count"]))

def count_by_engine(client_id, client_secret, start_dt, end_dt, org_id=None):
    auth_token = authenticate(client_id, client_secret)

    headers = {
        # when `Accept` header has value `application/vnd.withsecure.aggr+json`
        # then API calculates statistics for events matching query parameters
        "Accept": "application/vnd.withsecure.aggr+json",
        "Authorization": "Bearer " + auth_token,
        "User-Agent": "my-script",
        "Content-Type": "application/x-www-form-urlencoded",
    }

    param_start_dt = start_dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    param_end_dt = end_dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ")

    # API selects all security events from `epp` engine group that were persisted
    # between `start_dt` and `end_dt`, group items by value of `engine` property
    # and return number of items in each group
    params = {
        "engineGroup": ["epp"],
        "persistenceTimestampStart": param_start_dt,
        "persistenceTimestampEnd": param_end_dt,
        "count": "engine",  # property `engine` is used to group
    }

    if org_id:
        params["organizationId"] = org_id

    response = requests.post(API_URL + PATH, data=params, headers=headers)
    if not response.ok:
        print("Error", response.text)
        raise Exception("Request error")
    return response.json()

from datetime import datetime, timezone
from time import sleep
import requests
from authentication import authenticate

API_URL = 'https://api.connect.withsecure.com' 
INCIDENTS_PATH = "/incidents/v1/incidents"
DETECTIONS_PATH = "/incidents/v1/detections"

def get_updated_incidents(auth_token, min_date, org_id=None, next_page=None):
    headers = {
        "Accept": "application/json", # always use that header if you expect JSON response
        "Authorization": "Bearer " + auth_token,
        "User-Agent": "my-script"
    }
    
    params = {
        "limit": 50,
        "archived": False, # read only not archived incidents
        "updatedTimestampStart": min_date, # read all incidents updated AFTER `min_date`
        "order": "asc", # ascending order by `updateTimestampStart`
    }

    # `min_date` represents timestamp of last event in previous page
    # That event should not be included in next response. When `exclusiveStart=true` is set
    # then `updatedTimestampStart` is used as exclusive start of time range.
    params["exclusiveStart"] = "true"

    if next_page:
        params['anchor'] = next_page

    if org_id:
        params["organizationId"] = org_id

    response = requests.get(API_URL + INCIDENTS_PATH, params=params, headers=headers)

    if not response.ok:
        print("Error", response.text)
        raise Exception("Request error")

    return response.json()

def get_new_detections_page(auth_token, incident_id, min_date, next_page=None):
    headers = {
        "Accept": "application/json",
        "Authorization": "Bearer " + auth_token,
        "User-Agent": "my-script"
    }
    
    params = {"limit": 100,
              "incidentId": incident_id 
    }

    if next_page:
        params['anchor'] = next_page
        
    response = requests.get(API_URL + DETECTIONS_PATH, params=params, headers=headers)

    if not response.ok:
        print("Error", response.text)
        raise Exception("Request error")

    return response.json()

def print_all_detections(auth_token, incident_id, min_date):
    next_page = None
    fetch_page = True
    while fetch_page:
        page = get_new_detections_page(auth_token, incident_id, min_date, next_page)
        next_page = page.get("nextAnchor")
        fetch_page = next_page is not None
        for det in page["items"]:
            print("DetectionId={}, IncidentId={}, Created={}".format(det["detectionId"],
                                                       det["incidentId"], det["createdTimestamp"]))

def get_incidents_updated_after(auth_token, min_dt, org_id):
    # Elements API accepts timestamps in [RFC 3339](https://datatracker.ietf.org/doc/html/rfc3339#section-5.6)
    # format: YYYY-MM-DDThh:mm:ss.fff
    last_date = min_dt.strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    next_page = None
    fetch_page = True
    print("Reading incidents updated after {}".format(last_date))
    updated_incidents = []
    while fetch_page:
        page = get_updated_incidents(auth_token, last_date, org_id, next_page)
        next_page = page.get("nextAnchor")

        # If `nextAnchor` is present we should fetch next page in next iteration
        fetch_page = next_page is not None
        for incident in page["items"]:
            print("IncidentId={}, Updated Ts={}".format(incident["incidentId"],
                                                  incident["updatedTimestamp"]))
            # updating `last_date` with time when incident was updated
            # when iteration ends, `last_date` will be equal to `updatedTimestamp`
            # of last received incident
            last_date = incident["updatedTimestamp"]
            updated_incidents.append((incident["incidentId"], last_date))

    # Parse timestamp of last received incident after reading all
    # events, that were created after `min_dt`
    return {
        "last_date": datetime.strptime(last_date, "%Y-%m-%dT%H:%M:%S.%fZ"),
        "updated_incidents": updated_incidents
    }
        
def poll_edr_detections(client_id, client_secret, poll_interval, org_id=None):
    # initialize start date with current timestamp
    last_date = datetime.now(timezone.utc)

    # start infinite loop
    while True:
        # obtain authentication token 
        auth_token = authenticate(client_id, client_secret)

        # read all incidents created after `last_date`. After reading all items
        # that variable is updated with timestamp of last incident. `last_date` will be
        # used in next iteration
        updates = get_incidents_updated_after(auth_token, last_date, org_id)
        last_date = updates["last_date"]
        for incident in updates["updated_incidents"]:
            print_all_detections(auth_token, incident[0], incident[1])
        print("Last date", last_date)
        # execute next iteration every `poll_interval`
        sleep(poll_interval)