get_stats.py

#!/usr/bin/env python3

"""
Just for fun

# https://www.youtube.com/watch?v=-jexCDc7n5E
"""

import csv
from collections import Counter
from datetime import datetime as DateTime
from datetime import timedelta as TimeDelta
from enum import Enum
from ipaddress import ip_address


class Kind(Enum):
    CONNECT = "connected"
    DISCONNECT = "disconnected"


def get_stats(file="tarpit.csv"):
    def ts_parser(ts: str) -> DateTime:
        return DateTime.strptime(ts, "%Y-%m-%d %H:%M:%S")

    connected_set = set()
    disconnected_set = set()
    all_data = []

    with open(file, "rt", newline="", encoding="ascii") as fd:
        reader = csv.reader(fd, delimiter=",")
        for row in reader:
            try:
                ts, ip, port, kind = row
            except ValueError:
                if "INFO" in row[0] or "WARNING" in row[0]:
                    continue
                print(f"Incomplete input: {row}")
                continue

            ts = ts_parser(ts)
            ip = ip_address(ip)
            port = int(port)
            try:
                kind = Kind(kind)
            except ValueError:
                print(f"Unknown kind: {kind}")
                continue

            all_data.append((ts, ip, port, kind))
            client = ip, port
            if kind == Kind.CONNECT:
                connected_set.add(client)
            elif kind == Kind.DISCONNECT:
                disconnected_set.add(client)

    active_clients = connected_set - disconnected_set
    active_count = len(active_clients)
    del connected_set, disconnected_set

    active_connections = [row for row in all_data if tuple(row[1:3]) in active_clients]
    del active_clients

    log_duration = all_data[-1][0] - all_data[0][0]
    top10_attacking_ips = Counter(
        (row[1] for row in all_data if row[-1] == Kind.CONNECT)
    ).most_common(10)

    return all_data, active_count, active_connections, log_duration, top10_attacking_ips


if __name__ == "__main__":
    all_data, active_count, active_clients, duration, top10 = get_stats()
    day = TimeDelta(days=1)

    print(f"Logzeitraum: {duration // day} Tage")
    print(f"{active_count} aktive Verbindungen")
    print()
    print("Top 10")

    for ip, anzahl in top10:
        print(f"{ip!s:<16} => {anzahl}")