Move a Prefix to a New Zone and Start a Tcpdump

This example script reads one or more IPv4 or IPv6 host addresses from the command line, then attempts to move them all into a pre-identified Zone and start a tcpdump for all devices associated with that Zone. This can be useful in conjunction with a Zone that is set aside for aggressive filtering as discussed here.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "requests",
# ]
# ///
"""Moves one or more host prefixes to a new zone and start a tcpdump on all devices.

If using this script to respond to network events, it's recommended that the new
zone should be one with an aggressive filtering profile."""
import ipaddress
import requests
import json
import argparse
from urllib.parse import urlencode


# Enter the address of your rWeb
BASE_URL = "https://your.rweb.here.com"

HEADERS = {'Content-Type': 'application/json',
           'Accept': 'application/json'}

# Enter your user credentials. The user must have the 'device' permission.
CREDENTIALS = {'login': 'admin', 'password': 'admin123'}

# The zone ID of the new zone to move the prefix into.
AGGRESSIVE_ZONE_ID = 2031


def url(resource, params=None):
    """Turn a resource into a url.

    Uses BASE_URL and optionally encodes a hash of parameters."""
    querystring = ""
    if params is not None:
        querystring = "?" + urlencode(params)
    return BASE_URL + resource + querystring


def wrap(data):
    """Convert an object to an rWeb suitable json string.

    Wraps the data in the standard rWeb {'data': } object."""
    return json.dumps({'data': data})


def login(session):
    """Log in to a session.

    The authentication cookie is saved to the session"""
    result = session.post(url('/authentication'),
                          wrap(CREDENTIALS))
    assert (result.status_code == requests.codes.created)


def logout(session):
    """Close a session."""
    session.delete(url('/authentication'))


def get_zone_by_id(session, zone_id):
    """Retrieve a zone object by zone ID."""
    result = session.get(url(f'/zones/{zone_id}'))
    assert (result.status_code == requests.codes.ok)
    zone = json.loads(result.content)
    return zone


def trigger_tcpdump(session, zone, ip):
    """Triggers a tcpdump of 10000 packets on all devices associated with the zone."""
    result = session.get(url(zone['zone_devices_resource']))
    assert (result.status_code == requests.codes.ok)
    zone_devices = json.loads(result.content)
    for device in zone_devices['data']:
        result = session.post(url(zone['tcp_dumps_resource']), wrap(
            {'dstip': str(ip),
             'pcap_type': 'all',
             'device_resource': device['device_resource'],
             'snarf': int(1000),
             'count': int(10000)}))


def add_prefix_to_zone(session, zone, prefix):
    """Attempts to add the selected prefix to the zone."""
    result = session.post(url(zone['subnets_resource']), wrap({'subnet': str(prefix)}))
    assert (result.status_code == 201)


def remove_subnet_resource(session, subnet_resource):
    """Attempts to remove the selected subnet resource.

    Deleted subnet resources revert their traffic to the next encompassing zone."""
    result = session.delete(url(subnet_resource))
    assert (result.status_code == requests.codes.ok)


def get_zones_by_prefix(session, prefix):
    """Return the zone that contains the selected prefix."""
    result = session.get(url(f'/zones', {'ip': str(prefix)}))
    return result


def get_subnet_resource(session, zone, prefix):
    """Attempts to return the subnet resource containing the selected prefix."""
    result = json.loads(session.get(url(zone['subnets_resource'],
                                        {'ip': str(prefix)})).content)
    return result['data'][0]['subnet_resource']


def shift_prefix(session, zone, prefix):
    """Attempts to move the prefix to the selected zone.

    If the prefix exists elsewhere it will delete it first
    (the user should have permissions for this)."""
    zones = json.loads(get_zones_by_prefix(session, prefix).content)
    if zones['rows'] == 1:
        delete_zone = zones['data'][0]
        delete_subnet_resource = get_subnet_resource(session, delete_zone, prefix)
        remove_subnet_resource(session, delete_subnet_resource)

    add_prefix_to_zone(session, zone, prefix)


def parse_ip(prefix_string):
    """Parse IPv4 or IPv6 prefix."""
    try:
        address = ipaddress.ip_address(prefix_string)
    except ValueError as e:
        raise argparse.ArgumentTypeError(e)

    return address


def main():
    """Main function. Expects a list of valid IPv4/IPv6 host addresses,
    adds them to a zone and starts tcpdumps."""
    # Handle some command line arguments
    parser = argparse.ArgumentParser(
        description="Move selected prefixes to a Zone and start a tcpdump.")
    parser.add_argument('prefixes', nargs='+', type=parse_ip,
                        help="One or more IPv4/IPv6 host addresses")

    args = parser.parse_args()

    with requests.session() as session:
        session.headers = HEADERS
        # This is insecure but may be needed if you haven't set up an SSL certificate for rWeb yet
        session.verify = False
        login(session)

        zone = get_zone_by_id(session, AGGRESSIVE_ZONE_ID)
        for prefix in args.prefixes:
            shift_prefix(session, zone, prefix)
            trigger_tcpdump(session, zone, prefix)

        logout(session)

if __name__ == '__main__':
    main()