Overview

Use this guide to build a scheduled job that copies Cloud Secure Edge (CSE) events out of the Command Center and into a system you control, such as a SIEM, a log platform, or archive storage. The Command Center retains events for 14 days, so an external copy is required for longer retention.

The guide covers:

• Creating an API key with read-only access
• Confirming API access from your host
• How the Events API pages through results so that no event is duplicated or missed
• A ready-to-run Python script that you configure by filling in a few variables
• Scheduling the script and loading its output into your destination

If your destination is the ELK Stack, use the Filebeat integration instead. It implements the same approach without custom code.

Prerequisites

• A CSE admin account that can create API keys
• A Linux, macOS, or Windows host that can reach your Command Center over HTTPS and that runs on a schedule (a server, a container, or a serverless function)
• Python 3.8 or later on that host. The script uses only the Python standard library.
• curl for the connectivity test in Step 2

Step 1: Create a read-only API key

1.1 In the Command Center, navigate to Settings > API Keys and select Add API Key.

1.2 Configure the key:

• Name: Event Export
• Description: Used by the event export script
• Scope: ReadOnly

1.3 Copy the generated API Key Secret and store it in your secrets manager or a file readable only by the account that will run the script. The secret is shown once.

For more detail on API keys and scopes, see Authentication with API Key and API Key Privilege Levels.

Step 2: Confirm API access

2.1 On the host that will run the script, export the secret as an environment variable:

export CSE_API_KEY='PASTE_YOUR_API_KEY_SECRET_HERE'

2.2 Request the single most recent event:

curl -s -H "Authorization: Bearer $CSE_API_KEY" \
  "https://net.banyanops.com/api/v1/events?limit=1&order=DESC"

If your organization is provisioned on the European (EUCC) Command Center, replace net.banyanops.com with eucc.console.banyanops.com here and in the script.

A successful response is a JSON object with a data array containing one event. An HTTP 401 response means the key was not accepted; confirm that you copied the full secret and that the key has not been deleted.

Step 3: Understand how the Events API pages results

The Events API is documented under the event tag in the API Specifications. The parameters that matter for incremental retrieval are:

For example, to request only Access and Identity events at severity INFO or higher:

?type=Access|Identity&severity=INFO&order=ASC&limit=1000

The response contains a data array of events. Each event has a unique id and a created_at timestamp in epoch milliseconds. The response does not include a total count, so a page shorter than limit marks the end of the window.

To retrieve events incrementally without duplicates, each run of your job should:

1. Read the checkpoint: the created_at value of the newest event written by the previous run. On the very first run, choose a starting point such as 24 hours ago.
2. Fix the time window for this run: after is the checkpoint and before is the current time, less a short safety lag so that events still being written are picked up next time.
3. Page through the window with order=ASC and limit=1000, increasing skip by 1000 each request, until a page comes back shorter than limit. Keep after and before unchanged for every page of the run.
4. Write the events to your destination.
5. Save the new checkpoint: the highest created_at seen, together with the id values of the events at that exact millisecond so that they can be skipped if the boundary is returned again.

Deduplicating on the event id in your destination system is a recommended safety net in addition to the checkpoint.

Step 4: Install and configure the script

The following script implements the pattern above. It appends events to a newline-delimited JSON (NDJSON) file, which most SIEM forwarders and log shippers can read directly, and keeps its checkpoint in a small JSON file next to it.

4.1 Save the script as pull_cse_events.py on the host, for example in /opt/cse/.

4.2 Edit the values in the Fill in these values block:

#!/usr/bin/env python3
"""
pull_cse_events.py - Retrieve Cloud Secure Edge (CSE) events incrementally.

Each run pulls every event created since the previous run, appends the events
to an NDJSON file (one JSON object per line), and records a checkpoint so the
next run continues where this one stopped. Run it on a schedule (for example,
every 15 minutes with cron) to keep a complete, duplicate-free copy of your
CSE events outside the Command Center.
Requires Python 3.8 or later. No third-party packages are needed.
"""
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
---------------------------------------------------------------------------
Fill in these values
---------------------------------------------------------------------------
Command Center URL. Use https://eucc.console.banyanops.com for the European
(EUCC) Command Center.
COMMAND_CENTER = "https://net.banyanops.com"
API key secret. Prefer the CSE_API_KEY environment variable so the secret is
not stored in this file; the literal value below is used only as a fallback.
API_KEY = os.environ.get("CSE_API_KEY", "PASTE_YOUR_API_KEY_SECRET_HERE")
Where to append events (NDJSON) and where to store the checkpoint.
OUTPUT_FILE = "cse-events.ndjson"
CHECKPOINT_FILE = "cse-events.checkpoint.json"
Lowest severity to collect: DEBUG, INFO, WARN, or ERROR.
INFO returns INFO, WARN, and ERROR events.
SEVERITY = "INFO"
First run only: how many hours of history to start with. The Command Center
retains events for 14 days, so values above 336 have no effect.
INITIAL_LOOKBACK_HOURS = 24
---------------------------------------------------------------------------
Advanced settings (the defaults suit most deployments)
---------------------------------------------------------------------------
PAGE_SIZE = 1000         # events requested per API call
SAFETY_LAG_SECONDS = 60  # leave the most recent minute for the next run so
                         # that events still being written are not missed
MAX_RETRIES = 5          # retries for HTTP 429 and 5xx responses
REQUEST_TIMEOUT = 60     # seconds
def now_ms():
    return int(time.time() * 1000)
def load_checkpoint():
    """Return (created_at of the newest event written, ids written at that millisecond)."""
    try:
        with open(CHECKPOINT_FILE) as f:
            data = json.load(f)
        return int(data["created_at"]), set(data.get("ids", []))
    except (FileNotFoundError, ValueError, KeyError):
        return now_ms() - INITIAL_LOOKBACK_HOURS * 3600 * 1000, set()
def save_checkpoint(created_at, ids):
    tmp = CHECKPOINT_FILE + ".tmp"
    with open(tmp, "w") as f:
        json.dump({"created_at": created_at, "ids": sorted(ids)}, f)
    os.replace(tmp, CHECKPOINT_FILE)
def fetch_page(after, before, skip):
    params = {
        "after": after,
        "before": before,
        "order": "ASC",
        "limit": PAGE_SIZE,
        "skip": skip,
        "severity": SEVERITY,
    }
    url = f"{COMMAND_CENTER}/api/v1/events?{urllib.parse.urlencode(params)}"
    request = urllib.request.Request(url, headers={"Authorization": f"Bearer {API_KEY}"})
    for attempt in range(MAX_RETRIES):
        try:
            with urllib.request.urlopen(request, timeout=REQUEST_TIMEOUT) as response:
                return json.load(response).get("data") or []
        except urllib.error.HTTPError as error:
            retryable = error.code == 429 or error.code >= 500
            if not retryable or attempt == MAX_RETRIES - 1:
                raise
        except urllib.error.URLError:
            if attempt == MAX_RETRIES - 1:
                raise
        time.sleep(2 ** attempt)
def main():
    if not API_KEY or API_KEY.startswith("PASTE_"):
        sys.exit("Set the CSE_API_KEY environment variable or fill in API_KEY.")
after, ids_at_checkpoint = load_checkpoint()
before = now_ms() - SAFETY_LAG_SECONDS * 1000
if before <= after:
    print("Nothing to do yet; run again later.")
    return

newest = after
newest_ids = set(ids_at_checkpoint)
written = 0
skip = 0

with open(OUTPUT_FILE, "a") as out:
    while True:
        page = fetch_page(after, before, skip)
        for event in page:
            created_at = int(event["created_at"])
            if created_at == after and event["id"] in ids_at_checkpoint:
                continue  # already written by the previous run
            out.write(json.dumps(event) + "\n")
            written += 1
            if created_at > newest:
                newest, newest_ids = created_at, set()
            if created_at == newest:
                newest_ids.add(event["id"])
        if len(page) < PAGE_SIZE:
            break
        skip += PAGE_SIZE

save_checkpoint(newest, newest_ids)
print(f"Wrote {written} events to {OUTPUT_FILE}; checkpoint is now {newest}.")

if name == "main":
    main()

Step 5: Run the script once and verify

5.1 Run the script manually with the API key in the environment:

CSE_API_KEY='PASTE_YOUR_API_KEY_SECRET_HERE' python3 /opt/cse/pull_cse_events.py

The script prints how many events it wrote and the new checkpoint value:

Wrote 1834 events to cse-events.ndjson; checkpoint is now 1758150000000.

5.2 Inspect the output. Each line is one complete event:

head -n 1 cse-events.ndjson | python3 -m json.tool

5.3 Run the script a second time. It should report zero or only a handful of new events, which confirms that the checkpoint is working.

Step 6: Schedule the script

Choose an interval between 5 and 15 minutes. Shorter intervals keep the destination current; any interval under a few hours stays well inside the retention window.

Linux or macOS with cron

6.1 Store the secret in a file readable only by the account that runs the job:

sudo install -m 600 /dev/null /etc/cse-events.env
echo "export CSE_API_KEY='PASTE_YOUR_API_KEY_SECRET_HERE'" | sudo tee /etc/cse-events.env > /dev/null

6.2 Add a crontab entry (crontab -e) that runs every 15 minutes:

*/15 * * * * . /etc/cse-events.env && cd /opt/cse && /usr/bin/python3 pull_cse_events.py >> /var/log/cse-events.log 2>&1

Windows with Task Scheduler

Create a basic task that runs python.exe C:\cse\pull_cse_events.py on a repeating 15-minute trigger, with Start in set to C:\cse. Set the CSE_API_KEY environment variable for the account that runs the task, or paste the secret into API_KEY and restrict the file's permissions to that account.

Step 7: Load the events into your destination

The NDJSON file is the hand-off point to your pipeline. Common approaches:

• Universal forwarders and agents (Splunk Universal Forwarder, the Azure Monitor Agent, Fluent Bit, Vector, and similar) can monitor the file and ship each new line as one event. Configure the source type as JSON and map created_at to the event timestamp.
• Object storage or a data lake: rotate OUTPUT_FILE daily by including the date in its name, and upload the closed file with your cloud provider's CLI on the same schedule.
• Direct HTTP ingestion: replace the out.write(...) line in the script with a call to your SIEM's HTTP event collector.

Whichever destination you use, configure it to deduplicate on the event id field. This makes the pipeline tolerant of a re-run after a failure.

Troubleshooting

Next steps

• Review Event Properties and Definitions for the meaning of each field.
• Filter what you export with the type and severity parameters, for example to send only Access and Identity events to a high-cost SIEM tier and everything to archive storage.
• Explore every parameter of the Events endpoint in the API Specifications.