Publicado el 09/04/2023 12:04:00 en Hacking Web.
Author: 84kur10 | Total de votos: 4   Vote

Hola a todos, llevo tiempo sin escribir un artículo y quiero aprovechar la temporada de estos temas de IA para escribir un poco relacionado a esto.

Contexto

La pregunta que queremos resolver en este laboratorio es sobre cómo podríamos entrenar un modelo de machine learning que sea capaz de aprender a detectar un ataque de denegación de servicio a un servidor web nginx. Me he inclinado en que el laboratorio sea sobre nginx debido al uso que tiene en la actualidad, Ya que ha superado a Apache como el servidor web más popular además de que con los diferentes módulos se usa para multiples propositos como crear un balanceador de carga, proxy inverso, waf, incluso en kubernetes se usa como ingress para controlar los puntos de acceso al cluster.

Un poco de datos sobre el uso de los servidores web en la actualidad:

Herramientas a usar

Vamos a usar scikit-learn con el cual podemos entrenar modelos de machine learning para diferentes casos de uso. Como el resultado final que buscamos tener es simplemente una bandera que nos indique si estamos ante la presencia de un DOS o no, entonces me he inclidado por usar la regresión logística para esto, ya que esta usa internamente la función sigmoide que se mueve entre los valores de menos infinito a infinito en el eje x y en el eje y entre 0 y 1, esto hace que sea idea para calcular eventos que tienen que ver con la probabilidad de ocurrencia.



Adicionalmente, este laboratorio será puramente de análisis descriptivo, es decir que no servirá para preveer cuando va a ocurrir un DOS si no para saber si estamos ante la presencia de uno. Y puntualmente si este está ocurriendo por cantidad de solicitudes, en otras palabras, no está pensado en detectar un DOS que sea ocasionado debido a una vulnerabilidad.

Vamos a usar como entrada de datos el access log (Que igualmente existe en apache), en este se encuentran los datos de cada solicitud que se ha realizado, a qué hora se hizo y otras particularidades.

Generando datos para entrenamiento

Para esto, he simulado 2 situaciones, la primera en la cual hay un comportamiento natural de navegación en el servidor web en donde la cantidad de peticiones es natural, y otra en la que se realizan multiples solicitudes, para esto he creado el siguiente script:

import sys
import requests
import time

headers = { }

if len(sys.argv) < 4:
    print("Faltan argumentos. Uso: python script.py cantidad URL intervalo")
    sys.exit()

cantidad = int(sys.argv[1])
url = sys.argv[2]
intervalo = int(sys.argv[3])

if intervalo ==0:
    headers = {
        'User-Agent': 'DOSLAB'
    }
else:
    headers = {
        'User-Agent': 'NODOSLAB'
    }

for i in range(cantidad):
    response = requests.get(url, headers=headers)
    print(f"Peticion {i+1}: {response.status_code}")
    time.sleep(intervalo/1000)

Y podemos llamar el script así:

python request_process.py 100 http://localhost 2000

El primer argumento es la cantidad de peticiones que se desea realizar, el segundo es la URL y el tercero es el intervalo en milisegundos que deseamos tener entre cada petición. Internamente el script define que si el intervalo es cero seteará el user agent en un valor u otro, esto debido a que deseamos primero entrenar nuestro modelo desde el access log con aprendizaje supervisado y esto nos obliga a presentar a nuestro modelo datos etiquetados, en este caso me he aprovechado de esta cabecera para definir si es o no un DOS desde el script, en caso de que el intervalo sea 0, seteamos la cebecera DOSLAB para indicar que si, en caso contrario NODOSLAB para indicar que no.

Para esto he levantado un contenedor con nginx en el puerto 80 así:

docker run -p 80:80 -v "D:/LAB/DETECT_DOS":/var/log/nginx --name nginx_container -d nginx

He ejecutado el código con intervalo de 0 y de 2000 en diferentres ocasiones, un ejemplo de la ejecución:

python request_process.py 100 http://localhost 2000
Peticion 1: 200
Peticion 2: 200
Peticion 3: 200
.
.
.

Ahora que ya tenemos los datos en nuestro access log, he creado el siguiente script que lo genera nuestro dataset. A modo de resumen, esto genera un archivo csv dataset_access_dos.csv que tiene las siguientes columnas:

Tiempo: la hora incluyendo los segundos
Peticiones Cantidad de peticiones en ese segundo
Flag: 1 para indicar que si hay un DOS 0 para indicar lo contrario

import csv
import datetime

filename = 'D:LABSSECDETECT_DOSaccess.log'


# Abre el archivo access.log
with open(filename) as archivo:
    lector = csv.reader(archivo, delimiter=' ')

    peticiones_por_segundo = {}
    flag_degenacion_encontrado = {}

    for linea in lector:
        fecha_str = linea[3].lstrip('[').replace(" ","")
        fecha = datetime.datetime.strptime(fecha_str, '%d/%b/%Y:%H:%M:%S')

        if linea[5].startswith('GET'):
            #timestamp = fecha.replace(second=0, microsecond=0)
            timestamp_str = fecha.strftime('%Y-%m-%d %H:%M:%S')

            if timestamp_str in peticiones_por_segundo:
                peticiones_por_segundo[timestamp_str] += 1
            else:
                peticiones_por_segundo[timestamp_str] = 1

            if linea[9]=="DOSLAB":
                flag_degenacion_encontrado[timestamp_str ] = 1
            else:
                flag_degenacion_encontrado[timestamp_str ] = 0

with open('dataset_access_dos.csv', 'w', newline='') as file:
    writer = csv.writer(file)
    writer.writerow(["tiempo", "peticiones", "flag"])
    for timestamp_str,peticiones in peticiones_por_segundo.items():
        flag = flag_degenacion_encontrado[timestamp_str ]
        writer.writerow([timestamp_str,peticiones , flag])

Ahora tenemos nuestro dataset:

tiempo,peticiones,flag
2023-04-10 00:15:51,173,1
2023-04-10 00:15:52,230,1
2023-04-10 00:15:53,210,1
2023-04-10 00:15:54,230,1
2023-04-10 00:15:55,232,1
2023-04-10 00:15:56,246,1
2023-04-10 00:15:57,211,1
2023-04-10 00:15:58,268,1
2023-04-10 00:15:59,251,1
2023-04-10 00:16:00,234,1
2023-04-10 00:16:01,247,1
2023-04-10 00:16:02,244,1
2023-04-10 00:16:03,259,1

Ahora con esto listo, he creado el siguiente script que permite cargar los datos en nuestro modelo de sklearn para entrenarlo:

import pickle
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.linear_model import LogisticRegression
import pandas as pd
import os
directorio="model_trained"

if not os.path.exists(directorio):
    os.makedirs(directorio)

dosds = pd.read_csv('dataset_access_dos.csv',sep=',')
print(dosds.head(5))

caracteristicas = ["peticiones"]
clasificacion = 'flag'
x = dosds[caracteristicas]
y = dosds[clasificacion]


X_train, X_test, y_train, y_test = train_test_split(x, y, test_size = 20, random_state = 0)

logreg = LogisticRegression(max_iter=10000)
logreg.fit(X_train, y_train)
#logreg.fit(x, y)

print(logreg.score(X_test, y_test))

# Almacennado el modelo
filename = directorio+'/model_dos.sav'
pickle.dump(logreg, open(filename, 'wb'))

Con esto dividimos el dataset para entrenar con el 80% y el 20% lo dejamos para probar.

X_train, X_test, y_train, y_test = train_test_split(x, y, test_size = 20, random_state = 0)

Entenamos el modelo:

logreg.fit(X_train, y_train)

Medimos el score, si se acerca a 1 de forma positiva quiere decir que estamos en buen camino, además usamos datos de test que no son conocidos por el modelo para medir esto.

print(logreg.score(X_test, y_test))

Como verán he usado la librería pickle para dejar en un archivo el modelo entrenado y listo.



El performance del modelo es 1, esto es en apareciencia bueno (Entre más cercano a 1 es mejor), sin embargo este modelo es bastante sencillo de una sola característica y todos los datos con los que se entrenó los simulamos.

He creado el siguiente script para terminar de probarlo, verán que le envío 200,1,10,20 que son cantidades de peticiones en segundos:

import numpy as np
import csv
import pickle
from sklearn.linear_model import LogisticRegression
import pandas as pd

model_path="model_trained/model_dos.sav"
loaded_model = pickle.load(open(model_path, 'rb'))
label=["NORMAL","DOS"]

peticiones = np.array([200,1,10,20]).reshape(-1, 1)

predict = loaded_model.predict(peticiones)
print(str(peticiones[0]) +" - "+label[predict[0]])
print(str(peticiones[1]) +" - "+label[predict[1]])
print(str(peticiones[2]) +" - "+label[predict[2]])
print(str(peticiones[3]) +" - "+label[predict[3]])

Salida:

[200] - DOS
[1] - NORMAL
[10] - NORMAL
[20] - NORMAL

Al parecer para 1 petición en 1 segundo, 20 peticiones en 1 segundo y 20 en un segundo se considera normal, sin embargo 200 peticiones en 1 segundo lo toma como un DOS.

Acá termina el laboratorio 1, imagine poder llevar esto a monitorear cada minuto nuestro access log y automatizar este ciclo de vida con Airflow para saber cuando tenemos un posible DOS, depronto podríamos agregar más características como por ejemplo el uso de CPU del servidor y quizás hacer que aplique algunas políticas automáticamente etc.

Saludos.