Retrieving New Alerts

You can use the WebOrion Monitor API to check for new alerts once a minute and pull their full details, so they can go anywhere you need them: a SIEM, a logging platform, a chat channel or a ticketing system. This guide shows you how to do this by hand with curl, then how to automate it with a small script on a Linux VM.

How it works

Two API calls are all you need:

List alerts (GET /api/v1/alerts) with start_id set to one more than the last alert ID you have seen. Only alerts newer than that are returned.
Get an alert (GET /api/v1/alerts/{id}) for each new alert, to get the alert reason, GenAI Triage result, screenshots and the changes detected.

After handling the new alerts, remember the highest alert ID. Next time, start from the one after it.

Before you begin

You will need:

An API token. Generate one in the portal under Account → Manage Credentials.
A Linux VM with Python 3.9 or later and outbound HTTPS (port 443) to api.monitor.weborion.io. Python 3 comes preinstalled on Ubuntu 24.04 and Amazon Linux 2023. The VM used for Scheduled Rebaseline works fine.

Try it by hand

Set your token for this terminal session:

export WEBORION_API_TOKEN=paste-your-token-here

1. List recent alerts

The first time, there is no last alert ID yet, so list alerts from a point in time instead. This lists alerts from the last 60 minutes. Times are in UTC.

SINCE=$(date -u -d '60 minutes ago' +%Y-%m-%dT%H:%M:%S)

curl -sS -G -H "Authorization: Bearer $WEBORION_API_TOKEN" \
  --data-urlencode "start_date=$SINCE" -d items_per_page=100 \
  https://api.monitor.weborion.io/api/v1/alerts | python3 -m json.tool

The response is paginated. Each entry in data gives the alert ID, the webpage and when the change was detected:

{
  "current_page": 1,
  "data": [
    {
      "id": 1048214,
      "url": { "id": 5521, "name": "Agency home page", "url": "https://www.example.gov.sg/" },
      "time_detected": "2026-10-01 08:51:02"
    },
    {
      "id": 1048187,
      "url": { "id": 5530, "name": "Contact us", "url": "https://www.example.gov.sg/contact" },
      "time_detected": "2026-10-01 08:22:47"
    }
  ],
  "last_page": 1,
  "per_page": 100,
  "total": 2
}

If last_page is more than 1, repeat the request with -d page=2, -d page=3 and so on until you reach it.

2. List only new alerts

Note the highest id you received, 1048214 in the example above. To get only alerts newer than that, pass one more than it as start_id:

curl -sS -G -H "Authorization: Bearer $WEBORION_API_TOKEN" \
  -d start_id=1048215 -d items_per_page=100 \
  https://api.monitor.weborion.io/api/v1/alerts | python3 -m json.tool

An empty data list means there are no new alerts.

3. Get the details of an alert

For each new alert, request its details by ID:

curl -sS -H "Authorization: Bearer $WEBORION_API_TOKEN" \
  https://api.monitor.weborion.io/api/v1/alerts/1048214 | python3 -m json.tool

The response includes everything shown in the alert email and the portal. A shortened example:

{
  "id": 1048214,
  "url": { "id": 5521, "name": "Agency home page", "url": "https://www.example.gov.sg/" },
  "time_detected": "2026-10-01 08:51:02",
  "alert_reason": "changes_detected_from_baseline",
  "download_difference_url": "https://...",
  "result": {
    "summary": "New external script added",
    "network_error": false,
    "screenshot": { "before": "https://...", "after": "https://...", "difference": "https://..." },
    "ai_triage_result": {
      "severity": "HIGH",
      "description": "A script from an unknown domain was injected into the page.",
      "reasons": ["Script source is not on the whitelisted domains"],
      "suggested_action": "Review the page source and restore from a known good version."
    },
    "changes_detected": { "content_engine": {}, "integrity_engine": {}, "image_engine": [] }
  }
}

Field	What it tells you
`alert_reason`	`changes_detected_from_baseline`, `changes_detected_from_last_alert` or `webpage_unreachable`.
`result.summary`	A short description of what changed.
`result.ai_triage_result`	The GenAI Triage severity (`LOW`, `MEDIUM` or `HIGH`), description, reasons and suggested action. Present when GenAI Triage is turned on.
`result.changes_detected`	The detailed changes found by each monitoring engine.
`result.screenshot`, `download_difference_url`	Links to the screenshots and the difference file. These links can be time-limited, so download what you need when you receive the alert.

An alert can appear in the list a little before its details are ready. If Get an alert does not return the details yet, try again a minute later.

Automate it

The script below runs the same steps every minute. It remembers the last alert ID it handled and passes each new alert to a function called handle_alert(). By default, that function adds the alert to a file as one line of JSON. The script also keeps things gentle on the API and safe for whatever receives the alerts:

One list request per minute when there is nothing new, and a detail request only for each new alert.
At most 100 alerts per run, with a short pause between requests. If more are waiting, for example after the VM was offline, they are picked up over the next few runs, oldest first.
Each alert is handled once. The script saves its progress after every run, including runs that stop part way.
No alert is dropped. Alerts whose details aren’t ready are retried each minute for 30 minutes. After that, they are passed on with the basic information and details_available set to false.
Errors stop the run quietly. If the API rate limit is reached or the network is down, the script stops and continues on the next run.

1. Prepare the VM

Create a service user and the folders the script needs:

id weborion >/dev/null 2>&1 || sudo useradd --system --create-home --shell /usr/sbin/nologin weborion
sudo mkdir -p /opt/weborion /etc/weborion
sudo install -d -o weborion -g weborion -m 750 /var/lib/weborion-alerts
sudo install -d -o weborion -g weborion -m 755 /var/log/weborion

On Amazon Linux 2023, cron is not installed by default. Run sudo dnf install -y cronie and then sudo systemctl enable --now crond.

2. Store the API token

sudo tee /etc/weborion/alerts.env >/dev/null <<'EOF'
WEBORION_API_TOKEN=paste-your-token-here
EOF
sudo chown root:weborion /etc/weborion/alerts.env
sudo chmod 640 /etc/weborion/alerts.env

You can also add these optional settings to the same file:

Setting	Default	What it does
`WEBORION_FIRST_RUN_LOOKBACK_MINUTES`	`60`	How far back to look on the very first run.
`WEBORION_MAX_ALERTS_PER_RUN`	`100`	The most alert details fetched in one run.
`WEBORION_RETRY_MINUTES`	`30`	How long to keep retrying an alert whose details aren’t ready.
`WEBORION_OUTPUT_FILE`	`/var/log/weborion/alerts.ndjson`	The file new alerts are added to.

3. Save the script

Paste this whole block into your terminal. It saves the script to /opt/weborion/check_alerts.py as root, makes it owned by root so the service user can run it but not change it, and checks that the file was pasted cleanly:

sudo tee /opt/weborion/check_alerts.py >/dev/null <<'EOF'
#!/usr/bin/env python3
"""
Check WebOrion Monitor for new alerts and get their details.

Run once a minute. Each new alert is passed to handle_alert(), which by default
appends it to a file as one JSON line. Uses only the Python 3 standard library.
"""
import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone


def load_config(path):
    cfg = {}
    try:
        with open(path, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    key, value = line.split("=", 1)
                    cfg[key.strip()] = value.strip().strip("\"'")
    except FileNotFoundError:
        pass
    # Environment variables override the file
    cfg.update({k: v for k, v in os.environ.items() if k.startswith("WEBORION_")})
    return cfg


CFG = load_config(os.environ.get("WEBORION_CONFIG", "/etc/weborion/alerts.env"))
API_BASE = CFG.get("WEBORION_API_BASE", "https://api.monitor.weborion.io/api/v1").rstrip("/")
TOKEN = CFG.get("WEBORION_API_TOKEN", "")
OUTPUT_FILE = CFG.get("WEBORION_OUTPUT_FILE", "/var/log/weborion/alerts.ndjson")
STATE_FILE = CFG.get("WEBORION_STATE_FILE", "/var/lib/weborion-alerts/state.json")
FIRST_RUN_LOOKBACK_MINUTES = int(CFG.get("WEBORION_FIRST_RUN_LOOKBACK_MINUTES", "60"))
MAX_ALERTS_PER_RUN = int(CFG.get("WEBORION_MAX_ALERTS_PER_RUN", "100"))
RETRY_MINUTES = int(CFG.get("WEBORION_RETRY_MINUTES", "30"))


def handle_alert(alert):
    """Do something with one new alert. By default, append it to OUTPUT_FILE as one JSON line."""
    with open(OUTPUT_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(alert, ensure_ascii=False) + "\n")


class ApiError(Exception):
    def __init__(self, status):
        super().__init__(f"HTTP {status}")
        self.status = status


def api_get(path, params=None):
    url = API_BASE + path + ("?" + urllib.parse.urlencode(params) if params else "")
    request = urllib.request.Request(url, headers={
        "Authorization": f"Bearer {TOKEN}",
        "Accept": "application/json",
    })
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            return json.loads(response.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        raise ApiError(e.code)
    except (OSError, ValueError):
        raise ApiError(0)


def list_alerts(params):
    """Return {alert_id: list entry} for every alert matching params, across all pages."""
    alerts, page = {}, 1
    while True:
        response = api_get("/alerts", dict(params, page=page, items_per_page=100))
        for entry in response.get("data") or []:
            alerts[int(entry["id"])] = entry
        if page >= int(response.get("last_page") or 1):
            return alerts
        page += 1


def load_state():
    try:
        with open(STATE_FILE, encoding="utf-8") as f:
            return json.load(f)
    except FileNotFoundError:
        return {}


def save_state(state):
    tmp = STATE_FILE + ".tmp"
    with open(tmp, "w", encoding="utf-8") as f:
        json.dump(state, f)
    os.replace(tmp, STATE_FILE)


def main():
    if not TOKEN:
        print("WEBORION_API_TOKEN is not set")
        return 1

    state = load_state()
    pending = state.setdefault("pending", {})  # alerts whose details weren't ready yet
    now = time.time()
    fetched = handled = 0
    exit_code = 0

    def fetch_and_handle(alert_id):
        """Get one alert's details and handle it. Returns False if the details aren't ready yet."""
        nonlocal fetched, handled
        fetched += 1
        try:
            alert = api_get(f"/alerts/{alert_id}")
        except ApiError as e:
            if e.status in (401, 403, 429):
                raise
            return False
        finally:
            time.sleep(0.2)  # space out requests to the API
        handle_alert(dict(alert, details_available=True))
        handled += 1
        return True

    try:
        # 1. Retry alerts whose details weren't ready on an earlier run
        for key in sorted(pending, key=int):
            if fetched >= MAX_ALERTS_PER_RUN:
                break
            if fetch_and_handle(key):
                del pending[key]
            elif now - pending[key]["first_seen"] > RETRY_MINUTES * 60:
                # Never drop an alert: pass on what the list call returned
                handle_alert(dict(pending[key]["entry"], details_available=False))
                handled += 1
                del pending[key]

        # 2. List alerts newer than the last one handled
        if state.get("last_id") is None:
            since = state.setdefault("first_run_since", (
                datetime.now(timezone.utc) - timedelta(minutes=FIRST_RUN_LOOKBACK_MINUTES)
            ).strftime("%Y-%m-%dT%H:%M:%S"))
            new_alerts = list_alerts({"start_date": since})
        else:
            new_alerts = list_alerts({"start_id": state["last_id"] + 1})

        # 3. Get the details of each new alert, oldest first
        for alert_id in sorted(new_alerts):
            if state.get("last_id") is not None and alert_id <= state["last_id"]:
                continue
            if fetched >= MAX_ALERTS_PER_RUN:
                break  # the rest are picked up on the next run
            if not fetch_and_handle(alert_id):
                pending[str(alert_id)] = {"first_seen": now, "entry": new_alerts[alert_id]}
            state["last_id"] = alert_id

    except ApiError as e:
        if e.status in (401, 403):
            print("The API rejected the token. Check WEBORION_API_TOKEN.")
            exit_code = 1
        elif e.status == 429:
            print("The API rate limit was reached. Continuing on the next run.")
        else:
            print(f"An API request failed ({e}). Retrying on the next run.")
    finally:
        # Save progress even if a run stops part way, so handled alerts aren't repeated
        save_state(state)

    if handled or pending:
        print(f"{datetime.now():%Y-%m-%d %H:%M:%S} handled {handled} alerts, "
              f"{len(pending)} waiting for details, last alert ID {state.get('last_id')}")
    return exit_code


if __name__ == "__main__":
    sys.exit(main())
EOF
sudo chown root:root /opt/weborion/check_alerts.py
sudo chmod 755 /opt/weborion/check_alerts.py
python3 -c "import ast,sys; ast.parse(open(sys.argv[1]).read())" /opt/weborion/check_alerts.py \
  && grep -q "sys.exit(main())" /opt/weborion/check_alerts.py \
  && echo "check_alerts.py OK"

If the last line prints check_alerts.py OK, the whole script was pasted correctly. If you see an error or nothing at all, paste the block again. Then do a test run as the service user:

sudo -u weborion /usr/bin/python3 /opt/weborion/check_alerts.py

The first run picks up alerts from the last 60 minutes. It prints a line such as handled 3 alerts, 0 waiting for details, last alert ID 1048214, or nothing if there were no alerts.

4. Run it every minute

Add a cron entry. flock makes sure a new run never starts while the previous one is still going.

sudo tee /etc/cron.d/weborion-alerts >/dev/null <<'EOF'
* * * * * weborion flock -n /tmp/weborion-alerts.lock /usr/bin/python3 /opt/weborion/check_alerts.py >> /var/log/weborion/check_alerts.log 2>&1
EOF
sudo chmod 644 /etc/cron.d/weborion-alerts

The script only writes to check_alerts.log when it handled an alert or something went wrong, so the log stays small.

5. Check that it works

After a few minutes, look at the log and the latest alert:

tail -n 20 /var/log/weborion/check_alerts.log
tail -n 1 /var/log/weborion/alerts.ndjson | python3 -m json.tool

To keep the files from growing forever, rotate them with logrotate:

sudo tee /etc/logrotate.d/weborion-alerts >/dev/null <<'EOF'
/var/log/weborion/alerts.ndjson /var/log/weborion/check_alerts.log {
    su weborion weborion
    daily
    rotate 14
    compress
    delaycompress
    missingok
    notifempty
    create 0644 weborion weborion
}
EOF

Changing what happens to each alert

handle_alert() is the only part of the script you need to change. It receives each new alert in the same format as the Get an alert response, plus a details_available field. The script is owned by root, so open it with sudo, for example sudo vim /opt/weborion/check_alerts.py. For example, to keep the file and also post HIGH severity alerts to a Slack incoming webhook, add WEBORION_WEBHOOK_URL=https://hooks.slack.com/services/... to /etc/weborion/alerts.env and replace handle_alert() with:

WEBHOOK_URL = CFG.get("WEBORION_WEBHOOK_URL")


def handle_alert(alert):
    with open(OUTPUT_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(alert, ensure_ascii=False) + "\n")

    triage = (alert.get("result") or {}).get("ai_triage_result") or {}
    if WEBHOOK_URL and triage.get("severity") == "HIGH":
        page = alert.get("url") or {}
        text = f"HIGH WebOrion alert {alert['id']} on {page.get('url')}: {triage.get('description')}"
        request = urllib.request.Request(
            WEBHOOK_URL,
            data=json.dumps({"text": text}).encode("utf-8"),
            headers={"Content-Type": "application/json"},
        )
        try:
            urllib.request.urlopen(request, timeout=10)
        except OSError as e:
            # Don't let a chat outage hold up the other alerts
            print(f"Could not post alert {alert['id']} to the webhook: {e}")

Send only the alerts people need to act on, such as HIGH severity, to chat or paging tools, and keep the full record in the file. To reduce noisy alerts at the source, see Reducing False Positives and Auto Finetune.

Use cases

Once new alerts land in a file, or pass through handle_alert(), you can send them wherever your team works.

Use case	How to set it up
SIEM, such as Splunk, Elastic, Microsoft Sentinel, QRadar or Google SecOps	Point your SIEM’s log collection agent at `/var/log/weborion/alerts.ndjson`. Each line is one complete JSON alert. Use the alert `id` to remove duplicates.
Central logging	Have rsyslog, Fluent Bit or Vector read the same file and forward it to your logging platform.
Chat notifications	Post alerts to Slack, Microsoft Teams or similar from `handle_alert()`, as in the example above.
Ticketing	Create an incident in Jira from`handle_alert()`. Grouping by webpage avoids one ticket per alert.
On-call paging	Send `HIGH` severity alerts to PagerDuty, using the alert `id` as the dedupe key.
Long-term archive	Copy the rotated files to object storage, such as Amazon S3, for audit and reporting.

If your log collection agent runs as its own user, give it read access to the file, for example sudo usermod -aG weborion splunkfwd, and restart the agent.

Troubleshooting

Symptom	Likely cause
`The API rejected the token`	The token is wrong, expired or revoked. Generate a new one under Account → Manage Credentials and update `/etc/weborion/alerts.env`.
`The API rate limit was reached`	Can happen while catching up on many alerts. The script continues on the next run. If it happens often, lower `WEBORION_MAX_ALERTS_PER_RUN`.
Alerts with `details_available` set to `false`	The alert details weren’t ready within the retry window. Open the alert in the portal, or raise `WEBORION_RETRY_MINUTES`.
Nothing happens at all	Check that cron is running (`systemctl status cron`, or `crond` on Amazon Linux) and that `/var/lib/weborion-alerts` is owned by `weborion`.

To start again from scratch, delete /var/lib/weborion-alerts/state.json. The next run picks up the last 60 minutes of alerts again, so some may be handled twice. For anything else, contact CloudsineAI support.

Getting Started

Configuring Monitoring

Managing Notifications

Managing Changes

Configuring Traffic

API Guide

Getting Support

Retrieving New Alerts

How it works

Before you begin

Try it by hand

1. List recent alerts

2. List only new alerts

3. Get the details of an alert

Automate it

1. Prepare the VM

2. Store the API token

3. Save the script

4. Run it every minute

5. Check that it works

Changing what happens to each alert

Use cases

Troubleshooting

​How it works

​Before you begin

​Try it by hand

​1. List recent alerts

​2. List only new alerts

​3. Get the details of an alert

​Automate it

​1. Prepare the VM

​2. Store the API token

​3. Save the script

​4. Run it every minute

​5. Check that it works

​Changing what happens to each alert

​Use cases

​Troubleshooting

How it works

Before you begin

Try it by hand

1. List recent alerts

2. List only new alerts

3. Get the details of an alert

Automate it

1. Prepare the VM

2. Store the API token

3. Save the script

4. Run it every minute

5. Check that it works

Changing what happens to each alert

Use cases

Troubleshooting