🔵Применение в детектирование аномалий в сетевом трафике

Ну и теперь пример из реальной жизни, вообще это пример реализуем к любому сетевому трафику, мы может говорить об трафике на S-GW от пользователя к eNODEb, или о торговых данных(спорно, так как вопрос целостности и наличия аномалий здесь вопрос вторичный) и так далее.

Сетевой трафик можно представить как потоком многомерных наблюдений: размер пакета, межпакетный интервал, флаги TCP, размер окна, статистики source/destination port. Нормальная динамика трафика концентрируется на низкоразмерном многообразии в пространстве признаков; атаки типа port scan, DDoS, exfiltration проявляются как точки или подмножества, нарушающие топологическую структуру.

Вот offline-pipeline для проверки концепции на дампах трафика:

import numpy as np
import gudhi as gd
from gudhi.wasserstein import wasserstein_distance

def extract_features(packets):
    return np.column_stack([
        packets['size'],
        packets['iat'],
        packets['tcp_flags'],
        packets['window'],
    ])

def reference_diagram(baseline_features, landmarks):
    wc = gd.EuclideanWitnessComplex(landmarks=landmarks, witnesses=baseline_features)
    st = wc.create_simplex_tree(max_alpha_square=1.0, limit_dimension=2)
    diag = st.persistence()
    return np.array([(b, d) for dim, (b, d) in diag
                     if dim == 1 and d != float('inf')])

def score_window(window_features, landmarks, ref_diag):
    wc = gd.EuclideanWitnessComplex(landmarks=landmarks, witnesses=window_features)
    st = wc.create_simplex_tree(max_alpha_square=1.0, limit_dimension=2)
    diag = st.persistence()
    test = np.array([(b, d) for dim, (b, d) in diag
                     if dim == 1 and d != float('inf')])
    if len(test) == 0 or len(ref_diag) == 0:
        return float('inf')
    return wasserstein_distance(ref_diag, test, order=2)

def detect(stream, landmarks, ref_diag, window_size, threshold):
    buffer = []
    for pkt in stream:
        buffer.append(pkt)
        if len(buffer) >= window_size:
            features = extract_features(buffer[-window_size:])
            score = score_window(features, landmarks, ref_diag)
            if score > threshold:
                yield buffer[-1]['timestamp'], score

Бюджет задержки для inline IDS - около десятков микросекунд на пакет. Вычисление Vietoris-Rips на окне 10^5 точек в этот бюджет явно не укладывается; witness complex с m = 256 landmarks обеспечивает требуемую пропускную способность на устройствах типа Xilinx Alveo U280 при тактовой частоте 300 МГц. В продакшене этот пайплайн нужно заменить потоковым обновлением через StreamingWitnessComplex и аппаратным persistence engine.

В открытом доступе есть несколько реализаций комплексов, в частности в библиотеке CGAL(а именно в модуле TDA) + в том же GUHDI.