python3 update_dynamic_ip_whitelist.py --list

python3 update_dynamic_ip_whitelist.py --dry-run

#!/usr/bin/env python3
"""
CyberGate Customer API example: keep the SIP IP white-list in sync with a dynamic public IP address.

Use case
--------
Your intercoms connect to CyberGate from a site whose internet provider (ISP) assigns a dynamic
public IP address. CyberGate only accepts SIP traffic from white-listed IP addresses, so each time
the ISP hands out a new address, the white-list has to be updated.

Run this script periodically on a machine inside that site (for example every 5 minutes). It:
  1. Determines the current public IP address of the site.
  2. Reads the SIP white-list through the CyberGate Customer API and looks up the entry with the
     configured ID.
  3. Does nothing when the IP address has not changed.
  4. Updates the entry to the new IP address when it has changed.
     The update replaces the old address with the new one in a single call, so there is no moment
     where both or neither of the addresses are white-listed.

The script only updates an existing entry, it never creates or deletes entries.

Requirements
------------
- Python 3.8 or newer. Only the Python standard library is used, nothing needs to be installed.
- A Customer API secret, created in the CyberGate portal.

One-time setup
--------------
1. In the CyberGate portal, go to the network settings and add a white-list entry for the site,
   with the current public IP address as both the start and the end address.
2. Look up the ID of that entry. The portal does not show IDs, so let this script list them:

       python3 update_dynamic_ip_whitelist.py --list

3. Enter that ID as WHITELIST_ENTRY_ID in the configuration section below.

Usage
-----
    Linux / macOS:
        export CYBERGATE_API_SECRET="<your API secret>"
        python3 update_dynamic_ip_whitelist.py

    Windows (PowerShell):
        $env:CYBERGATE_API_SECRET = "<your API secret>"
        python update_dynamic_ip_whitelist.py

    Options:
        --list      show all white-list entries with their ID and exit
        --dry-run   show what would change, without changing anything

Scheduling
----------
    Linux / macOS (cron), every 5 minutes, with the output appended to a log file:
        */5 * * * * CYBERGATE_API_SECRET=<secret> python3 /opt/cybergate/update_dynamic_ip_whitelist.py \
            >> /var/log/cybergate-ip.log 2>&1

    (In crontab, write this on a single line, without the backslash.)

    Windows: create a task in Task Scheduler that runs python.exe with this script as argument,
    repeated every 5 minutes, and set CYBERGATE_API_SECRET as a system environment variable.

Keep the interval at a few minutes or more: the Customer API is rate limited per IP address.
"""

import argparse
import ipaddress
import json
import os
import sys
import urllib.error
import urllib.request

# ---------------------------------------------------------------------------------------------
# Configuration: adjust these values to your situation.
# ---------------------------------------------------------------------------------------------

# Base URL of the CyberGate Customer API.
API_BASE_URL = "https://admin.cybergate.cybertwice.com/api/customer"

# The API secret is read from an environment variable, so it does not end up in this file.
API_SECRET = os.environ.get("CYBERGATE_API_SECRET", "")

# ID of the white-list entry this script keeps up to date. Find it with: --list
WHITELIST_ENTRY_ID = 0

# Service that returns the public IPv4 address of the caller as plain text.
# Alternatives: https://ipv4.icanhazip.com or https://checkip.amazonaws.com
PUBLIC_IP_SERVICE_URL = "https://api.ipify.org"

# Timeout in seconds for every HTTP request.
HTTP_TIMEOUT_SECONDS = 15


# ---------------------------------------------------------------------------------------------
# Helper functions
# ---------------------------------------------------------------------------------------------

def get_public_ip_address():
    """
    Asks an external service which public IP address this site uses on the internet.

    The service sees the address our request comes from, which is the address the ISP assigned
    to the site router. That is also the address the intercoms use to connect to CyberGate.
    The result is validated, so an error page from the service is never written to the white-list.
    """
    request = urllib.request.Request(PUBLIC_IP_SERVICE_URL, headers={"User-Agent": "CyberGate-IP-Updater"})
    with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT_SECONDS) as response:
        text = response.read().decode("utf-8").strip()

    # Raises ValueError when the response is not a valid IPv4 address.
    return str(ipaddress.IPv4Address(text))


def call_customer_api(method, path, body=None):
    """
    Sends a request to the CyberGate Customer API and returns the JSON response (or None).

    - method: "GET", "POST", "PATCH" or "DELETE"
    - path:   the part after the base URL, for example "/whitelist" or "/whitelist/12"
    - body:   optional dictionary, sent as JSON

    Authentication: the API secret is sent as a Bearer token in the Authorization header.
    On an error response, an exception is raised that includes the message returned by the API.
    """
    headers = {
        "Authorization": f"Bearer {API_SECRET}",
        "Accept": "application/json",
        "User-Agent": "CyberGate-IP-Updater",
    }

    data = None
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        headers["Content-Type"] = "application/json"

    request = urllib.request.Request(API_BASE_URL + path, data=data, headers=headers, method=method)

    try:
        with urllib.request.urlopen(request, timeout=HTTP_TIMEOUT_SECONDS) as response:
            content = response.read().decode("utf-8")
            return json.loads(content) if content else None
    except urllib.error.HTTPError as error:
        details = error.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"{method} {path} failed with HTTP {error.code}: {details}") from error


def get_whitelist():
    """
    GET /whitelist: returns all SIP white-list entries of your tenant.

    Each entry looks like:
        {"id": 12, "enabled": true, "ipAddressStart": "203.0.113.10", "ipAddressEnd": "203.0.113.10",
         "description": "Main office", "location": "Amsterdam"}
    """
    return call_customer_api("GET", "/whitelist")


def print_whitelist(whitelist):
    """Prints all white-list entries with their ID, to find the value for WHITELIST_ENTRY_ID."""
    print(f"{'ID':>6}  {'Enabled':<8} {'IP start':<16} {'IP end':<16} {'Description':<30} Location")
    for entry in whitelist:
        print(f"{entry['id']:>6}  {entry['enabled']!s:<8} {entry['ipAddressStart']:<16} {entry['ipAddressEnd']:<16} "
              f"{entry.get('description') or '':<30} {entry.get('location') or ''}")


def find_entry_by_id(whitelist, entry_id):
    """Returns the white-list entry with the given ID, or None when it does not exist."""
    for entry in whitelist:
        if entry["id"] == entry_id:
            return entry
    return None


def update_whitelist_entry(entry_id, ip_address):
    """
    PATCH /whitelist/{id}: changes the IP address of an existing white-list entry.

    A single IP address is a range where the start and end address are the same.
    Only the properties in the request body are changed; description, location and the
    enabled state of the entry are left as they are. Because the old address is replaced by the
    new one in a single call, the number of white-listed addresses of your tenant stays the same.
    """
    call_customer_api("PATCH", f"/whitelist/{entry_id}", {
        "ipAddressStart": ip_address,
        "ipAddressEnd": ip_address,
    })


# ---------------------------------------------------------------------------------------------
# Main flow
# ---------------------------------------------------------------------------------------------

def main():
    """Checks the public IP address and updates the white-list entry when the address has changed."""
    parser = argparse.ArgumentParser(
        description="Keep the CyberGate SIP white-list in sync with a dynamic public IP address.")
    parser.add_argument("--list", action="store_true", help="show all white-list entries with their ID and exit")
    parser.add_argument("--dry-run", action="store_true", help="show what would change, without changing anything")
    args = parser.parse_args()

    # sys.exit() with a text prints that text as error and stops the script with exit code 1.
    if not API_SECRET:
        sys.exit("ERROR: environment variable CYBERGATE_API_SECRET is not set.")

    # Read the white-list once; it is needed both for --list and for the update.
    whitelist = get_whitelist()

    if args.list:
        print_whitelist(whitelist)
        return

    if WHITELIST_ENTRY_ID <= 0:
        sys.exit("ERROR: WHITELIST_ENTRY_ID is not configured. Run this script with --list to find the ID.")

    # Step 1: determine the current public IP address of this site.
    current_ip = get_public_ip_address()
    print(f"Current public IP address: {current_ip}")

    # Step 2: look up the configured entry. It is never created by this script, only updated.
    entry = find_entry_by_id(whitelist, WHITELIST_ENTRY_ID)
    if entry is None:
        sys.exit(f"ERROR: white-list entry {WHITELIST_ENTRY_ID} does not exist. "
                 "Run this script with --list to check the ID.")

    # Safety check: this script manages a single IP address. When the entry is an IP range, the ID
    # probably points to the wrong entry, and updating it would remove the rest of the range.
    if entry["ipAddressStart"] != entry["ipAddressEnd"]:
        sys.exit(f"ERROR: white-list entry {WHITELIST_ENTRY_ID} is an IP range "
                 f"({entry['ipAddressStart']} - {entry['ipAddressEnd']}), not a single IP address. "
                 "Check WHITELIST_ENTRY_ID.")

    if not entry["enabled"]:
        print("WARNING: this white-list entry is disabled in CyberGate, so the address is not allowed.")

    old_ip = entry["ipAddressStart"]

    # Step 3: the IP address has not changed, nothing to do.
    if old_ip == current_ip:
        print(f"IP address unchanged (entry {WHITELIST_ENTRY_ID}), nothing to do.")
        return

    # Step 4: the IP address has changed, replace the old address by the new one.
    print(f"IP address changed from {old_ip} to {current_ip}, updating entry {WHITELIST_ENTRY_ID}.")
    if not args.dry_run:
        update_whitelist_entry(WHITELIST_ENTRY_ID, current_ip)


if __name__ == "__main__":
    try:
        main()
    except (OSError, ValueError, KeyError, RuntimeError) as error:
        # OSError:      network problems, such as no connection or a timeout.
        # ValueError:   unexpected response, such as no valid IP address or no valid JSON.
        # KeyError:     a field is missing in the response of the API.
        # RuntimeError: the API returned an error, see call_customer_api().
        # The error is reported on one line; exit code 1 lets a scheduler detect the failure.
        sys.exit(f"ERROR: {error}")

export CYBERGATE_API_SECRET="<your API secret>"
python3 update_dynamic_ip_whitelist.py

$env:CYBERGATE_API_SECRET = "<your API secret>"
python update_dynamic_ip_whitelist.py

*/5 * * * * CYBERGATE_API_SECRET=<secret> python3 /opt/cybergate/update_dynamic_ip_whitelist.py >> /var/log/cybergate-ip.log 2>&1