Retrieving a List of Top Attackers

This example retrieves the list of Top Attackers for the specified Zone.

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.12"
# dependencies = [
#     "requests",
#     "python-dateutil",
# ]
# ///
"""Retreive Top Attacker list for a given zone"""

import requests
import json
import urllib.parse
import datetime
import dateutil.parser
import dateutil.tz
import argparse

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

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

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 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 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 find_group(session, name):
    """Find group by name

    Search for a group resource by the exact name."""
    result = session.get(url('/groups', {'name':name}))
    assert(result.status_code == requests.codes.ok)
    group_list = json.loads(result.content)
    assert(group_list['rows']==1)
    return group_list['data'][0]

def find_zone(session, group, name):
    """Find zone by group object and name

    Search for a zone resource by the exact name. Expects a group
    map, like one returned by find_group."""
    result = session.get(url(group['zones_resource'],{'name':name}))
    assert(result.status_code == requests.codes.ok)
    zone_list = json.loads(result.content)
    assert(zone_list['rows']==1)
    return zone_list['data'][0]

def get_top_attackers(session, zone, tstamp):
    """Get the top attacker list for a specific zone and timestamp"""

    # Force to zulu format
    params = {"data_timestamp": tstamp.isoformat() + "Z"}

    result = session.get(url(zone['topattackers_resource'], params))
    assert(result.status_code == requests.codes.ok)

    attacker_data = json.loads(result.content)
    return attacker_data['data']

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.message)

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

    # Handle some command line arguments
    parser = argparse.ArgumentParser(description="Retreive sampled attacker list.")
    parser.add_argument('server', help="rWeb server address")
    parser.add_argument('group', help="Group name")
    parser.add_argument('zone', help="Zone name")
    parser.add_argument('time', nargs='?', type=parse_timestamp,
            default=datetime.datetime.now(),
            help="Time to retrieve attackers for. Defaults to the current time. Example: '2014-05-16 20:30:00'")

    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)

        # Find the zone we want to list
        group = find_group(session, args.group)
        zone = find_zone(session, group, args.zone)

        # Retrieve the top attacker list
        attackers = get_top_attackers(session, zone, args.time)

        # For example, render the most basic fields as a tab separated list
        # Could also work with 'ddos_subtype', 'rank', etc.
        print(attackers)
        print("Attacker IP\tVictim IP\tAttack Type\tTime")
        for attacker in attackers:
            print(("{0}\t{1}\t{2}\t{3}".format(
                attacker['src_addr'],
                attacker['dst_net'],
                attacker['ddos_type'],
                attacker['timestamp'])))

        logout(session)

if __name__ == '__main__':
    main()