Fuzzer de aplicações web baseado em CURL automatizado via BASH e Python

Após ter usado por um bom tempo o FFUF percebi algumas limitações da ferramenta para a minha maneira de coletar informações em aplicações web. Resolvi que deveria desenvolver meu próprio fuzzer, após alguns meses de pesquisa cheguei a versão atual que segue em teste.

Utilizando curl-impersonate como forma de bypass de rate limit e bloqueio baseado em fingerprint. Atualmente o código precisa ser polido e ter mais incrementos de funcionalidades como identificação de códigos 429, conexão instável e over head.

Em desenvolvimento!

Esse é o motor de fuzzing recebendo os dados para realizar a requisição e retornando os dados da resposta.

#!/bin/bash

curl_temp=$(mktemp)
headers_out_temp=$(mktemp)

fuzz() {
  get_data() {
    headers=()
    if [ -n "$headers_file" ]; then
      echo "{{{" >> "$headers_in"
      while IFS= read -r h; do
        headers+=(-H "$h")
        echo "$h" >> "$headers_in"
      done < "$headers_file"
      echo "}}}" >> "$headers_in" 
    fi
    if [ -n "$headers_in_temp" ]; then
      echo "{{{" >> "$headers_in"
      while IFS= read -r h; do
        headers+=(-H "$h")
        echo "$h" >> "$headers_in"
      done <<< "$headers_in_temp"
      echo "}}}" >> "$headers_in" 
    fi
    fuzz_engine() {
      "$curl_type" -s -L -X "$method" \
      --max-time 5 --connect-timeout 3 \
      --max-redirs 5 --limit-rate 100k \
      --path-as-is -o $curl_temp \
      "${headers[@]}" \
      -D "$headers_out_temp" \
      -w \
"%{response_code}\n\
%{url}\n\
%{url_effective}\n\
%{method}\n\
%{scheme}\n\
%{num_redirects}\n\
%{redirect_url}\n\
%{num_headers}\n\
%{size_header}\n\
%{size_request}\n\
%{size_download}\n\
%{size_upload}\n\
%{content_type}\n\
%{remote_ip}\n\
%{remote_port}\n\
%{local_ip}\n\
%{local_port}\n\
%{http_version}\n\
%{num_connects}\n\
%{conn_id}\n\
%{time_namelookup}\n\
%{time_connect}\n\
%{time_appconnect}\n\
%{time_pretransfer}\n\
%{time_starttransfer}\n\
%{time_redirect}\n\
%{time_total}\n\
%{ssl_verify_result}\n\
%{exitcode}\n" \
      "$url"
  }
  fuzz_engine
  }
  get_data $curl_type $method $url $headers_file
   
  output_hash=$(head -c 256 $curl_temp | md5sum | awk '{print $1}')
  echo $output_hash
  echo "$payload"
  echo "{{{" >> $headers_out 
  cat "$headers_out_temp" | tr -d '\r' >> $headers_out
  echo "}}}" >> $headers_out 
  rm $curl_temp
  rm $headers_out_temp
}
  curl_type=$1
  method=$2
  url=$3
  headers_file=$4
  headers_in_temp=$5
  payload=$6
  headers_in=$8
  headers_out=$9
echo "{{{" >> "$7"
fuzz >> $7
echo "}}}" >> "$7"

exit 0

Há também um script que valida o arquivo de resultados do scan e importa tudo para o banco de dados SQL. Essa é a nova versão utilizando Bash e Python a versão anterior toda desenvolvida em Bash era extremamente lenta .

#!/bin/bash

echo "Arquivo atual: $1"

verificador () {
  epoch_start=$(head -n1 $1)
  epoch_end=$(tail -n1 $1)

  if date -d "@$epoch_start" >/dev/null 2>&1; then
    echo "Epoch start [ OK ] Valor: $epoch_start"
  else
    echo "Epoch start [ ERRO ] Valor: $epoch_start"
    echo "Arquivo: $1"
    exit 1
#    sleep 1
  fi
  if date -d "@$epoch_end" >/dev/null 2>&1; then
    echo "Epoch end [ OK ] Valor: $epoch_end"
  else
    echo "Epoch end [ ERRO ] Valor: $epoch_end"
    echo "Arquivo: $1"
    exit 1
#    sleep 1
  fi

  mapfile -t linhas < "$1"

  line_object_start=1
  line_object_end=33

  while true; do

      open_object="${linhas[line_object_start]}"
      close_object="${linhas[line_object_end]}"

      if [[ $open_object != "{{{" ]]; then
          echo "Objeto inválido"
          exit 1
      fi

      if [[ $close_object != "}}}" ]]; then
          echo "Objeto inválido"
          exit 1
      fi

      verify_end_line=$((line_object_end + 1))
      get_end_line="${linhas[verify_end_line]}"

      if [[ $get_end_line == "$epoch_end" ]]; then
          echo "Arquivo OK"
          return
      fi

      ((line_object_start += 33))
      ((line_object_end += 33))

  done
}

get_index_id () {
  id_old=$(sqlite3 "$1" "SELECT fuzz_index_id FROM metadata LIMIT 1;")
  id_new=$id_old
  ((id_new += 1))
  echo "$id_old index id atual"
  echo "$id_new index id novo do banco"
}

extract_data () {
  python3 ~/ethical-hacking/scripts/fuzzer2/data-importer.py "$2" "$id_new" "$epoch_start" "$epoch_end" < "$1"

}

runner () {
  request=1 # conta o request atual
  line_start=2 # linha inicial dos dados do fuzzing
  line_to_stop=$(( $(wc -l < "$2") - 33 )) # pega o total de linhas menos 2
  {
  echo "BEGIN;" 
  while true; do
  if [[ $line_start == $line_to_stop ]]; then
    break
  fi
  ((line_start += 33))
  ((request += 1)) 
  done
  echo "COMMIT;"
} | sqlite3 "$1" 
}

verificador $1
get_index_id $2 # pega o index_id do argumento do banco
extract_data $1 $2
runner $2 $1
sqlite3 "$2" "
  UPDATE metadata
  SET fuzz_index_id = $id_new;
  "
exit 0

Motor de importação todo em Python

import sys
import time
import sqlite3

db = sys.argv[1]
id = sys.argv[2]
epoch_start = sys.argv[3]
epoch_end = sys.argv[4]

SQL = """
INSERT INTO fuzz_out (
    id,
    epoch_start,
    epoch_end,
    request,
    response_code,
    url,
    url_effective,
    method,
    scheme,
    num_redirects,
    redirect_url,
    num_headers,
    size_header,
    size_request,
    size_download,
    size_upload,
    content_type,
    remote_ip,
    remote_port,
    local_ip,
    local_port,
    http_version,
    num_connects,
    conn_id,
    time_namelookup,
    time_connect,
    time_appconnect,
    time_pretransfer,
    time_starttransfer,
    time_redirect,
    time_total,
    ssl_verify_result,
    exitcode,
    output_hash,
    payload
)
VALUES (
    ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
    ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
    ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
    ?, ?, ?, ?, ?
)
"""

request=1
data = sys.stdin.read().splitlines()
line_start=2
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute("BEGIN")

print(db, id, epoch_start, epoch_end)

while True:
    line_start += 33
    obj = data[line_start:line_start + 31]
    if len(obj) != 31:
        break
#    print(obj)
    cursor.execute(
            SQL,
            (
            id,
            epoch_start,
            epoch_end,
            request,
            *obj
            )
        )

#    time.sleep(1) 
    request += 1
conn.commit()
conn.close()

Modelo inicial de analise de dados via estatística e machine learning

import sqlite3
import pandas as pd
import sys
from sklearn.ensemble import IsolationForest

data_base = sys.argv[1]
dominio = sys.argv[2] 
metodo = sys.argv[3]

conn = sqlite3.connect(data_base)

df = pd.read_sql("""
SELECT
    id,
    epoch_start,
    epoch_end,
    request,
    response_code,
    url,
    url_effective,
    method,
    scheme,
    num_redirects,
    redirect_url,
    num_headers,
    size_header,
    size_request,
    size_download,
    size_upload,
    content_type,
    remote_ip,
    remote_port,
    local_ip,
    local_port,
    http_version,
    num_connects,
    conn_id,
    time_namelookup,
    time_connect,
    time_appconnect,
    time_pretransfer,
    time_starttransfer,
    time_redirect,
    time_total,
    ssl_verify_result,
    exitcode,
    output_hash,
    payload
FROM fuzz_out
WHERE url LIKE ?
AND method = ?
""", conn, params=(f"%{dominio}%", metodo))

FEATURE_WEIGHTS = {
    "num_redirects":      5.0,
    "num_headers":        5.0,
    "size_header":        5.0,
    "size_download":      4.0,
    "size_upload":        0.1,
    "num_connects":       1.0,
    "time_namelookup":    0.1,
    "time_connect":       0.1,
    "time_appconnect":    0.1,
    "time_pretransfer":   0.1,
    "time_starttransfer": 0.1,
    "time_redirect":      0.1,
    "time_total":         0.1,
}
for feature in FEATURE_WEIGHTS:
    df[feature] = pd.to_numeric(df[feature], errors="coerce")

X = df[list(FEATURE_WEIGHTS.keys())].copy()

# Aplica os pesos
for feature, weight in FEATURE_WEIGHTS.items():
    X[feature] *= weight

model = IsolationForest(
    n_estimators=100,
    contamination="auto",
    random_state=42
)

# Treina
model.fit(X)

ml_score = model.decision_function(X)

ml_score = ml_score.max() - ml_score
ml_score = ml_score / ml_score.max()
ml_score = (ml_score * 1000).astype(int)

df["ml_score"] = ml_score

def modelo_estatistico(df):

    # Campos avaliados por frequência
    colunas_freq = [
        "response_code",
        "output_hash",
        "content_type",
        "remote_ip",
        "http_version"
    ]
    total = len(df)
    for col in colunas_freq:

        freq = df[col].value_counts(dropna=False)

        df[f"rarity_{col}"] = (
            1 - df[col].map(freq).fillna(0) / total
        ) * 1000

        df[f"rarity_{col}"] = (
            df[f"rarity_{col}"]
            .clip(0, 1000)
            .astype(int)
        )
    RARITY_WEIGHTS = {
        "response_code":        5.0,
        "output_hash":          2.0,
        "content_type":         2.0,
        "remote_ip":            1.0,
        "http_version":         1.0,
    }

    df["rarity_score"] = 0

    for campo, peso in RARITY_WEIGHTS.items():

        df["rarity_score"] += (
            df[f"rarity_{campo}"] * peso
        )
        df["score_final"] = df["ml_score"] * df["rarity_score"]

    print(
        df[df["response_code"] != "000"]
        .sort_values(by="score_final", ascending=False)
        .head(50)
        [["id", "epoch_start", "epoch_end", "request", "score_final"]]
        .to_csv(index=False, header=False)
    )

O objetivo com esse sistema todo é reunir o máximo de informações processar todos os dados filtrando apenas anomalias entre os endpoints e então realizar a investigação manual usando ZAPROXY nos candidatos que possuem maior nível de entropia.