Features used by Dynatrace - SAL - LIT.AI.JKU in the NAD 2021 challenge

dt_nad2021_features.py

# Copyright 2021
#    Dynatrace Research
#    SAL Silicon Austria Labs
#    LIT Artificial Intelligence Lab
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import pandas as pd

import feature_ip_port as ipp
import feature_ratios as rat
import feature_segments as seg

df = pd.read_csv("example.csv", parse_dates=["time"]).drop("label", axis="columns")

features = pd.concat([
    ipp.get_ip_area(df, columns=["src", "dst"], areas=["global", "link_local", "unspecified"]),
    ipp.get_ip_flow_type(df, src="src", dst="dst"),
    ipp.get_ip_binary_match(df, src="src", dst="dst"),
    ipp.get_ip_one_bits(df, columns=["src", "dst"]),
    ipp.get_ip_parts(df, columns=["src", "dst"]),
    ipp.get_port_area(df, columns=["spt", "dpt"]),
    ipp.get_port_match(df, spt="spt", dpt="dpt"),
    rat.get_cnt_ratios(df, unsafe=False, fill=0),
    rat.get_in_out_ratios(df, fill=0),
    rat.get_cnt_distances(df),
    seg.get_segment_features(df, ipp.get_ip_parts(df, columns=["src", "dst"]),
                             groups=["src", "dst_3", "dst_2", "dst_1"], lag=1, parallel=False)
], axis="columns")

# take a glimpse at the data
print(features.head().T)

dt_nad2021_features_nb.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "opponent-exhaust",
   "metadata": {},
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import pandas as pd\n",
    "\n",
    "import feature_ip_port as ipp\n",
    "import feature_ratios as rat\n",
    "import feature_segments as seg"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "italic-opposition",
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.read_csv(\"example.csv\", parse_dates=[\"time\"]).drop(\"label\", axis=\"columns\")\n",
    "\n",
    "features = pd.concat([\n",
    "    ipp.get_ip_area(df, columns=[\"src\", \"dst\"], areas=[\"global\", \"link_local\", \"unspecified\"]),\n",
    "    ipp.get_ip_flow_type(df, src=\"src\", dst=\"dst\"),\n",
    "    ipp.get_ip_binary_match(df, src=\"src\", dst=\"dst\"),\n",
    "    ipp.get_ip_one_bits(df, columns=[\"src\", \"dst\"]),\n",
    "    ipp.get_ip_parts(df, columns=[\"src\", \"dst\"]),\n",
    "    ipp.get_port_area(df, columns=[\"spt\", \"dpt\"]),\n",
    "    ipp.get_port_match(df, spt=\"spt\", dpt=\"dpt\"),\n",
    "    rat.get_cnt_ratios(df, unsafe=False, fill=0),\n",
    "    rat.get_in_out_ratios(df, fill=0),\n",
    "    rat.get_cnt_distances(df),\n",
    "    seg.get_segment_features(df, ipp.get_ip_parts(df, columns=[\"src\", \"dst\"]),\n",
    "                             groups=[\"src\", \"dst_3\", \"dst_2\", \"dst_1\"], lag=1, parallel=False)\n",
    "], axis=\"columns\")\n",
    "\n",
    "# take a glimpse at the data\n",
    "features.T"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}

example.csv

feature_ip_port.py

# Copyright 2021
#    Dynatrace Research
#    SAL Silicon Austria Labs
#    LIT Artificial Intelligence Lab
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from typing import List
from ipaddress import IPv4Address

import numpy as np
import pandas as pd


def get_ip_area(df: pd.DataFrame, columns: List[str], areas: List[str]) -> pd.DataFrame:
    """
    One-hot encode to which ip address block a certain address belongs to

    :param df: original cleaned data frame
    :param columns: the columns to process (most likely `["src", "dst"]`)
    :param areas: the areas to process, one or more of
        `multicast, private, global, unspecified, reserved, loopback, link_local`
    """
    result = pd.DataFrame(index=df.index)

    for column in columns:
        ipv4 = df[column].map(IPv4Address)
        for area in areas:
            result[f"{column}_{area}"] = ipv4.apply(lambda ip: getattr(ip, f"is_{area}")).astype(bool)

    return result


# noinspection PyUnresolvedReferences
def get_ip_flow_type(df: pd.DataFrame, src: str = "src", dst: str = "dst") -> pd.DataFrame:
    """
    One-hot encoding of source and destination traffic type

    :param df: original cleaned data frame
    :param src: the column name of the source address
    :param dst: the column name of the destination address
    """
    result = pd.DataFrame(index=df.index)
    src_global = df[src].apply(lambda e: IPv4Address(e).is_global)
    dst_global = df[dst].apply(lambda e: IPv4Address(e).is_global)

    result["is_inter"] = np.logical_and(src_global, dst_global).astype(bool)
    result["is_ingress"] = np.logical_and(src_global, ~dst_global).astype(bool)
    result["is_egress"] = np.logical_and(~src_global, dst_global).astype(bool)
    result["is_intra"] = np.logical_and(~src_global, ~dst_global).astype(bool)

    # ensure one-hot encoding
    assert np.all(np.sum(result, axis=1) == 1)

    return result


def get_ip_binary_match(df: pd.DataFrame, src: str = "src", dst: str = "dst") -> pd.DataFrame:
    """
    Compute the number of bits that are equal in src and dst,
    from left to right, stopping at the first mismatch,
    which helps identifying how similar two addresses are

    :param df: original cleaned data frame
    :param src: the column name of the source address
    :param dst: the column name of the destination address
    """
    ips = df[[src, dst]].to_numpy()

    # a bitwise XOR will reveal the bit position of the first mismatch
    # which we use to check the necessary bit representation length
    match = np.bitwise_xor(ips[:, 0], ips[:, 1])
    bit_length = 32 - np.ceil(np.log2(match + 1)).astype("uint8")

    result = pd.DataFrame(index=df.index)
    result["ip_match"] = bit_length
    return result


def get_ip_one_bits(df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
    """
    Transform every number (preferably, ip address) to its number of set bits

    :param df: original cleaned data frame
    :param columns: the columns to process (most likely `["src", "dst"]`)
    """
    result = df.loc[:, columns]
    result[:] = _hamming_weight(result.to_numpy())
    result = result.add_suffix("_one_bits")

    result = result.astype("uint8")
    return result


def get_ip_parts(df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
    """
    Get the four individual parts of an ip address, indexed from lowest to highest,
    e.g. `192.168.0.1` becomes `{"_0": 1, "_1": 0, "_2": 168, "_3": 192}`

    :param df: original cleaned data frame
    :param columns: the columns to process (most likely `["src", "dst"]`)
    """
    result = pd.DataFrame(index=df.index)

    for column in columns:
        arr = df[column].to_numpy()
        result[f"{column}_0"] = arr & 0xFF
        result[f"{column}_1"] = (arr >> 8) & 0xFF
        result[f"{column}_2"] = (arr >> 16) & 0xFF
        result[f"{column}_3"] = (arr >> 24) & 0xFF

    result = result.astype("uint8")
    return result


def get_port_area(df: pd.DataFrame, columns: List[str]) -> pd.DataFrame:
    """
    One-hot encode port numbers to their respective area

    ```
    common:         0 to  1023
    registered:  1024 to 49151
    ephemeral:  49152 to 65535
    ```

    :param df: original cleaned data frame
    :param columns: the columns to process (most likely `["spt", "dpt"]`)
    """
    result = pd.DataFrame(index=df.index)

    for column in columns:
        port = df[column]
        result[f"{column}_common"] = (port <= 1023).astype(bool)
        result[f"{column}_registered"] = ((port >= 1024) & (port <= 49151)).astype(bool)
        result[f"{column}_ephemeral"] = (port >= 49152).astype(bool)

    return result


# noinspection PyUnresolvedReferences
def get_port_match(df: pd.DataFrame, spt: str = "spt", dpt: str = "dpt") -> pd.DataFrame:
    """
    A column that is one if the src and dst port match

    :param df: original cleaned data frame
    :param spt: the column name of the source port
    :param dpt: the column name of the destination port
    """
    result = pd.DataFrame(index=df.index)
    result["port_match"] = (df[spt] == df[dpt]).astype(bool)
    return result


HAM_M1 = np.uint64(0x5555555555555555)
HAM_M2 = np.uint64(0x3333333333333333)
HAM_M4 = np.uint64(0x0f0f0f0f0f0f0f0f)
HAM_H01 = np.uint64(0x0101010101010101)


def _hamming_weight(x):
    """
    efficient implementation of the hamming weight to
    find the number of one bits of a number
    (c) https://en.wikipedia.org/wiki/Hamming_weight#Efficient_implementation
    """
    x = x - ((x >> 1) & HAM_M1)
    x = (x & HAM_M2) + ((x >> 2) & HAM_M2)
    x = (x + (x >> 4)) & HAM_M4
    x = (x * HAM_H01) >> 56
    return x

feature_ratios.py

# Copyright 2021
#    Dynatrace Research
#    SAL Silicon Austria Labs
#    LIT Artificial Intelligence Lab
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import numpy as np
import pandas as pd


def get_cnt_ratios(df: pd.DataFrame, unsafe: bool = False, fill: float = 0) -> pd.DataFrame:
    """
    Get various ratios between cnt values.
    This function expects the following column names to be present:
        cnt_src, cnt_src_slow, cnt_src_conn,
        cnt_dst, cnt_dst_slow, cnt_dst_conn,
        cnt_serv_src, cnt_serv_src_slow, cnt_serv_src_conn,
        cnt_serv_dst, cnt_serv_dst_slow, cnt_serv_dst_conn

    :param df: the original cleaned data frame
    :param unsafe: also compute ratios for columns where the divisor can be zero
    :param fill: fill invalid divisions with this number
    """
    res = pd.DataFrame(index=df.index)

    # ratios relative to slow and conn
    ##################################

    assert np.all(df["cnt_src_slow"] > 0)
    assert np.all(df["cnt_src_conn"] > 0)

    res["relative_cnt_src_to_slow"] = df["cnt_src"] / df["cnt_src_slow"]
    res["relative_cnt_src_to_conn"] = df["cnt_src"] / df["cnt_src_conn"]

    assert np.all(df["cnt_dst_slow"] > 0)
    assert np.all(df["cnt_dst_conn"] > 0)

    res["relative_cnt_dst_to_slow"] = df["cnt_dst"] / df["cnt_dst_slow"]
    res["relative_cnt_dst_to_conn"] = df["cnt_dst"] / df["cnt_dst_conn"]

    # assert np.all(df["cnt_serv_src_slow"] > 0)  # FALSE
    # assert np.all(df["cnt_serv_src_conn"] > 0)  # FALSE

    if unsafe:
        res["relative_cnt_serv_src_to_slow"] = _finite_divide(df["cnt_serv_src"], df["cnt_serv_src_slow"], fill)
        res["relative_cnt_serv_src_to_conn"] = _finite_divide(df["cnt_serv_src"], df["cnt_serv_src_conn"], fill)

    # assert np.all(df["cnt_serv_dst_slow"] > 0)  # FALSE
    # assert np.all(df["cnt_serv_dst_conn"] > 0)  # FALSE

    if unsafe:
        res["relative_cnt_serv_dst_to_slow"] = _finite_divide(df["cnt_serv_dst"], df["cnt_serv_dst_slow"], fill)
        res["relative_cnt_serv_dst_to_conn"] = _finite_divide(df["cnt_serv_dst"], df["cnt_serv_dst_conn"], fill)

    # src / dst ratios
    ##################

    assert np.all(df["cnt_dst"] > 0)
    assert np.all(df["cnt_dst_slow"] > 0)
    assert np.all(df["cnt_dst_slow"] > 0)

    res["ratio_cnt_src_dst"] = df["cnt_src"] / df["cnt_dst"]
    res["ratio_cnt_src_dst_slow"] = df["cnt_src_slow"] / df["cnt_dst_slow"]
    res["ratio_cnt_src_dst_conn"] = df["cnt_src_conn"] / df["cnt_dst_conn"]

    # assert np.all(df["cnt_serv_dst"] > 0)  # FALSE
    # assert np.all(df["cnt_serv_dst_slow"] > 0)  # FALSE
    # assert np.all(df["cnt_serv_dst_conn"] > 0)  # FALSE

    if unsafe:
        res["ratio_cnt_serv_src_dst"] = _finite_divide(df["cnt_serv_src"], df["cnt_serv_dst"], fill)
        res["ratio_cnt_serv_src_dst_slow"] = _finite_divide(df["cnt_serv_src_slow"], df["cnt_serv_dst_slow"], fill)
        res["ratio_cnt_serv_src_dst_conn"] = _finite_divide(df["cnt_serv_src_conn"], df["cnt_serv_dst_conn"], fill)

    # conn to slow ratios
    #####################

    assert np.all(df["cnt_dst_slow"] > 0)
    assert np.all(df["cnt_src_slow"] > 0)

    res["ratio_cnt_dst_conn_slow"] = df["cnt_dst_conn"] / df["cnt_dst_slow"]
    res["ratio_cnt_src_conn_slow"] = df["cnt_src_conn"] / df["cnt_src_slow"]

    # assert np.all(df["cnt_serv_src_slow"] > 0)  # FALSE
    # assert np.all(df["cnt_serv_dst_slow"] > 0)  # FALSE

    if unsafe:
        res["ratio_cnt_serv_src_conn_slow"] = _finite_divide(df["cnt_serv_src_conn"], df["cnt_serv_src_slow"], fill)
        res["ratio_cnt_serv_dst_conn_slow"] = _finite_divide(df["cnt_serv_dst_conn"], df["cnt_serv_dst_slow"], fill)

    res = res.astype("float32")  # save some gpu memory

    # make sure that our values are finite and not too big
    assert np.all(np.isfinite(res)) and np.all(res < 1e+12)
    return res


def get_in_out_ratios(df: pd.DataFrame, fill: float = 0) -> pd.DataFrame:
    """
    Get ratios between in and out traffic counts.
    This function expects the following column names to be present:
        in (bytes), out (bytes), duration

    :param df: the original cleaned data frame
    :param fill: fill invalid divisions with this number
    """
    res = pd.DataFrame(index=df.index)

    res["in_bytes_per_duration"] = _finite_divide(df["in (bytes)"], df["duration"], fill)
    res["out_bytes_per_duration"] = _finite_divide(df["out (bytes)"], df["duration"], fill)

    res["ratio_in_out_bytes"] = _finite_divide(df["in (bytes)"], df["out (bytes)"], fill)

    res = res.astype("float32")  # save some gpu memory

    # make sure that our values are finite and not too big
    assert np.all(np.isfinite(res)) and np.all(res < 1e+12)
    return res


def get_cnt_distances(df: pd.DataFrame) -> pd.DataFrame:
    """
    Get various distances between cnt values.
    This function expects the following column names to be present:
        cnt_src, cnt_src_slow, cnt_src_conn,
        cnt_dst, cnt_dst_slow, cnt_dst_conn,
        cnt_serv_src, cnt_serv_src_slow, cnt_serv_src_conn,
        cnt_serv_dst, cnt_serv_dst_slow, cnt_serv_dst_conn

    :param df: the original cleaned data frame
    """
    df = df.astype("int64")
    res = pd.DataFrame(index=df.index)

    # src dst differences
    #####################

    res["diff_cnt_src_dst"] = df["cnt_src"] - df["cnt_dst"]
    res["diff_cnt_src_dst_slow"] = df["cnt_src_slow"] - df["cnt_dst_slow"]
    res["diff_cnt_src_dst_conn"] = df["cnt_src_conn"] - df["cnt_dst_conn"]

    res["diff_cnt_serv_src_dst"] = df["cnt_serv_src"] - df["cnt_serv_dst"]
    res["diff_cnt_serv_src_dst_slow"] = df["cnt_serv_src_slow"] - df["cnt_serv_dst_slow"]
    res["diff_cnt_serv_src_dst_conn"] = df["cnt_serv_src_conn"] - df["cnt_serv_dst_conn"]

    # conn slow differences
    #######################

    res["diff_dst_conn_slow"] = df["cnt_dst_conn"] - df["cnt_dst_slow"]
    res["diff_src_conn_slow"] = df["cnt_src_conn"] - df["cnt_src_slow"]
    res["diff_serv_src_conn_slow"] = df["cnt_serv_src_conn"] - df["cnt_serv_src_slow"]
    res["diff_serv_dst_conn_slow"] = df["cnt_serv_dst_conn"] - df["cnt_serv_dst_slow"]

    return res


def _finite_divide(a: np.ndarray, b: np.ndarray, fill: float = 0) -> np.ndarray:
    """Divides `a / b` but will fix `0 / 0` and `1 / 0` to `fill` (default: 0)"""
    with np.errstate(divide="ignore", invalid="ignore"):
        c = np.true_divide(a, b)
        c[c == np.inf] = fill
        c = np.nan_to_num(c, nan=fill)

    return c

feature_segments.py

# Copyright 2021
#    Dynatrace Research
#    SAL Silicon Austria Labs
#    LIT Artificial Intelligence Lab
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from typing import List
from functools import partial
from multiprocessing import Pool, cpu_count

import numpy as np
import pandas as pd


def get_segment_features(df_clean: pd.DataFrame, df_ip: pd.DataFrame, groups: List[str], lag: int,
                         parallel: bool = True) -> pd.DataFrame:
    """
    Group the data frame according to the specified `groups`, then, within each group,
    cluster rows together that are consecutive to one another (according to the `lag` param),
    and assign the segment length, and the number of unique spt, dpt, app per segment.

    Example: For `times = [1, 3, 4, 5, 7, 8, 15]` we would create the segments
    `[[1], [3, 4, 5], [7, 8], [15]]` with segment lengths `[1, 3, 2, 1]` with `lag = 1`
    and the number of unique spt, dpt, app per segment.

    Larger values for `lag` will also cluster points farther apart into one segment, e.g.
    with `lag = 2` we would create the segments `[[1, 3, 4, 5, 7, 8], [15]]`

    :param df_clean: the original cleaned data frame
    :param df_ip: the "ip_parts" features data frame
    :param groups: a list of columns to group by
    :param lag: the minimum difference between observations to form a segment
    :param parallel: whether to process groups in parallel or not
    """

    df = pd.concat([df_clean, df_ip], axis="columns")

    # transform and sort time column
    df["time"] = df["time"].astype(np.int64) // 10 ** 9
    df = df.sort_values("time")

    groups = [group for _, group in df.groupby(groups, as_index=False)]
    apply_function = partial(_assign_segment_metrics, lag=lag)  # fix the lag parameter

    if not parallel:
        segments = map(apply_function, groups)
    else:
        with Pool(cpu_count()) as pool:
            segments = pool.map(apply_function, groups)

    return pd.concat(segments).sort_index()


def _assign_segment_metrics(group: pd.DataFrame, lag: int = 1):
    assert lag >= 1
    assert len(group) > 0
    assert group["time"].is_monotonic_increasing

    # append new columns with some default values
    columns = ["segment_length", "nunique_dst", "nunique_spt", "nunique_dpt", "nunique_app"]
    result = pd.DataFrame(1, dtype=int, index=group.index, columns=columns)

    # get the time segment splits
    time = group["time"].to_numpy()
    splits = np.flatnonzero(np.diff(time) > lag) + 1
    splits = np.insert(splits, 0, 0)

    dim, nsegments = len(time), len(splits)

    # pre-cache column locations to make the subsequent loop faster
    loc_segment_length = result.columns.get_loc("segment_length")
    loc_nunique_dst = result.columns.get_loc("nunique_dst")
    loc_nunique_spt = result.columns.get_loc("nunique_spt")
    loc_nunique_dpt = result.columns.get_loc("nunique_dpt")
    loc_nunique_app = result.columns.get_loc("nunique_app")

    loc_dst = group.columns.get_loc("dst")
    loc_spt = group.columns.get_loc("spt")
    loc_dpt = group.columns.get_loc("dpt")
    loc_app = group.columns.get_loc("app")

    for i in range(nsegments):
        lo = splits[i]
        hi = splits[i + 1] if i + 1 < nsegments else dim
        slc = slice(lo, hi - 1)  # segment slice

        # assign the segment length for the entire group
        result.iloc[slc, loc_segment_length] = hi - lo

        # compute number of unique spt, dpt, app fields for this segment
        result.iloc[slc, loc_nunique_dst] = group.iloc[slc, loc_dst].nunique()
        result.iloc[slc, loc_nunique_spt] = group.iloc[slc, loc_spt].nunique()
        result.iloc[slc, loc_nunique_dpt] = group.iloc[slc, loc_dpt].nunique()
        result.iloc[slc, loc_nunique_app] = group.iloc[slc, loc_app].nunique()

    return result

requirements.txt

numpy>=1.19
pandas>=1.2