Checking Abnormal V2Ray Users

abnormal.py

#! /usr/bin/python3.10

"""
This module is used for checking excessive use by V2Ray users.
Install Pandas beforehand via: `pip install pandas`

Sample Command:
$ journalctl -u v2ray --since "72 h ago" | grep email | cut -d " " -f 6,7,8,13 | python abnormal.py -

GitHub: https://github.com/Mahyar24/V2Conf
Mahyar@Mahyar24.com, Mon 28 Nov 2022
"""

import ipaddress
import sys

import pandas as pd


def make_df(input_data, subnet: int = 32) -> pd.DataFrame:
    return (
        pd.read_csv(input_data, delimiter=" ", header=None)
        .rename({0: "date", 1: "time", 2: "ip", 3: "username"}, axis=1)
        .assign(
            **{
                "date_time": lambda df_: pd.to_datetime(
                    df_.pop("date") + " " + df_.pop("time")
                ),
                "user": lambda df_: pd.Categorical(
                    df_.pop("username").str.split("@").str[0]
                ),
                "ip": lambda df_: df_.pop("ip")
                .str.removeprefix("tcp:")
                .str.removeprefix("udp:")
                .str.split(":")
                .str[0]
                .map(lambda ip: ipaddress.IPv4Network(f"{ip}/{subnet}", strict=False)),
            }
        )
    )


def analyze_df(df: pd.DataFrame) -> pd.DataFrame:
    return (
        df.groupby([pd.Grouper(key="date_time", freq="30 S"), "user"])[["ip"]]
        .nunique()
        .reset_index()
        .groupby([pd.Grouper(key="date_time", freq="30 Min"), "user"])[["ip"]]
        .sum()
        .query("ip > 60")
        .groupby("user")["ip"]
        .agg(
            **{
                "No": lambda df_: df_.count(),
                "Excessing": lambda df_: (df_.mean() - 60) / 60,
            }
        )
        .sort_values(["No", "Excessing"], ascending=False)
    )


if __name__ == "__main__":
    print(analyze_df(make_df(sys.stdin, subnet=16)))