Getting Zones with Historical Attacks Which Exceed a Threshold

This sample shows how to retrieve a list of incidents broken down by group/zone overlapping a specified time window which exceed a specified throughput.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "requests",
#     "python-dateutil",
# ]
# ///
import requests
import json
import urllib.parse
import datetime
import dateutil.parser
import argparse

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

# Put in your user credentials here
CREDENTIALS = {'login':'admin', 'password':'admin123'}

# Convenience functions
def parse_timestamp(string):
    """Parse simple timestamp."""
    try:
        time = dateutil.parser.parse(string)
        if time.tzinfo:
            return time.astimezone(dateutil.tz.tzutc()).replace(tzinfo=None)
        else:
            return time
    except ValueError as e:
        raise argparse.ArgumentTypeError(e)

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

    Uses base_url and optionally encodes a hash of parameters"""
    querystring = ""
    if params != None:
        querystring = "?"+urllib.parse.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})

# API functions
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 get_victims(session, start, end, threshold):
    """Retrieve a list of victims which overlap start to end time window, and which exceed the threshold"""
    result = session.get(url('/zones'))
    assert(result.status_code == requests.codes.ok)
    zones = json.loads(result.content)["data"]
    victims = []
    for zone in zones:
        result = session.get(url(zone["victims_resource"],
                                 {"start_time":start,
                                  "end_time":end,
                                  "sort_name":"starttime"}))
        assert(result.status_code == requests.codes.ok)
        res = json.loads(result.content)
        for victim in res["data"]:
            victims += [(zone, victim)]
    return [(z,v) for z, v in victims if v["max_pps"] >= threshold]

def logout(session):
    """Close a session"""
    result = session.delete(url('/authentication'))
    assert(result.status_code == requests.codes.ok)

def main():
    """Main CLI interface"""
    global BASE_URL

    # Handle some command line arguments
    parser = argparse.ArgumentParser(description="Retrieve victims with a peak packet rate over a specified threshold.")
    parser.add_argument('server', help="rWeb server address")
    parser.add_argument('start', nargs='?', type=parse_timestamp,
            default=datetime.datetime.now() - datetime.timedelta(minutes=5),
            help="Start of window to look for victims. Defaults to five minutes ago. Example: '2014-05-16 20:30:00'")
    parser.add_argument('end', nargs='?', type=parse_timestamp,
            default=datetime.datetime.now(),
            help="End of window to look for victims. Defaults to the current time. Example: '2014-05-16 20:30:00'")
    parser.add_argument("-t", "--threshold", nargs='?', default=10000, type=int,
                        help="Peak packet rate threshold, in pps. Defaults to 10000", )

    args = parser.parse_args()

    BASE_URL = "https://{0}".format(args.server) #change to specific rWeb instance

    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)
        victims = get_victims(session, args.start, args.end, args.threshold)
        last = ""
        index = 1
        for zone, victim in victims:
            if zone["name"] != last:
                last = zone["name"]
                index = 1
                print(f"Group: {zone["group_name"]}")
                print(f"Zone: {zone["name"]}")
            print(f"\tIncident {index}")
            print("\t-----------")
            print(f"\tVictim: {victim["ip_addr"]}")
            print(f"\tAttack Type: {victim["attacktype"]}")
            print(f"\tPeak Attack Rate: {victim["max_pps"]} pps")
            print(f"\tAttack Start Time: {victim["starttime"]}")
            print(f"\tAttack End Time: {victim["endtime"]}")
            print()
            index += 1
        logout(session)

if __name__ == '__main__':
    main()