RomanNum3ral
/
Cybersecurity-Projects
mirror of https://github.com/CarterPerez-dev/Cybersecurity-Projects.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat(ml): add CSIC 2010 data loader with feature extraction 

Parses multi-line CSIC 2010 HTTP request files into CSICRequest
dataclasses, converts to ParsedLogEntry with synthesized defaults,
and produces 35-dim feature vectors via the existing extraction pipeline.
This commit is contained in:
CarterPerez-dev 2026-02-28 05:39:51 -05:00
parent cf92b157f3
commit 2475d1603f
2 changed files with 574 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

220
PROJECTS/advanced/ai-threat-detection/backend/ml/data_loader.py Normal file
View File

@ -0,0 +1,220 @@
"""
©AngelaMos | 2026
data_loader.py
"""
import logging
import re
from dataclasses import dataclass
from datetime import datetime, UTC
from pathlib import Path
import numpy as np
from app.core.features.encoder import encode_for_inference
from app.core.features.extractor import extract_request_features
from app.core.ingestion.parsers import ParsedLogEntry
logger = logging.getLogger(__name__)
_REQUEST_LINE_RE = re.compile(
r"^(GET|POST|PUT|DELETE|HEAD|OPTIONS|PATCH|TRACE)"
r"\s+(\S+)\s+(HTTP/\d\.\d)\s*$"
)
_DEFAULT_IP = "192.168.1.100"
_DEFAULT_UA = (
"Mozilla/5.0 (compatible; Konqueror/3.5; Linux)"
" KHTML/3.5.8 (like Gecko)"
)
_WINDOWED_FEATURE_NAMES: list[str] = [
"req_count_1m",
"req_count_5m",
"req_count_10m",
"error_rate_5m",
"unique_paths_5m",
"unique_uas_10m",
"method_entropy_5m",
"avg_response_size_5m",
"status_diversity_5m",
"path_depth_variance_5m",
"inter_request_time_mean",
"inter_request_time_std",
]
@dataclass
class CSICRequest:
"""
Single HTTP request parsed from CSIC 2010 dataset format
"""
method: str
path: str
query_string: str
protocol: str
headers: dict[str, str]
body: str
label: int
def parse_csic_file(
path: Path,
label: int,
) -> list[CSICRequest]:
"""
Parse a CSIC 2010 dataset file into a list of CSICRequest objects
"""
text = path.read_text(encoding="utf-8", errors="replace")
lines = text.splitlines()
blocks: list[list[str]] = []
current: list[str] = []
for line in lines:
match = _REQUEST_LINE_RE.match(line)
if match and current:
blocks.append(current)
current = [line]
elif match:
current = [line]
elif current:
current.append(line)
if current:
blocks.append(current)
results: list[CSICRequest] = []
for block in blocks:
req = _parse_request_block(block, label)
if req is not None:
results.append(req)
logger.info(
"Parsed %d requests from %s (label=%d)",
len(results),
path.name,
label,
)
return results
def _parse_request_block(
lines: list[str],
label: int,
) -> CSICRequest | None:
"""
Parse a single request block into a CSICRequest
"""
if not lines:
return None
match = _REQUEST_LINE_RE.match(lines[0])
if not match:
return None
method = match.group(1)
full_uri = match.group(2)
protocol = match.group(3)
if "?" in full_uri:
path, query_string = full_uri.split("?", 1)
else:
path = full_uri
query_string = ""
headers: dict[str, str] = {}
body_start = len(lines)
for i, line in enumerate(lines[1:], 1):
if not line.strip():
body_start = i + 1
break
if ": " in line:
key, value = line.split(": ", 1)
headers[key] = value
body_lines = [
ln for ln in lines[body_start:] if ln.strip()
]
body = "\n".join(body_lines)
return CSICRequest(
method=method,
path=path,
query_string=query_string,
protocol=protocol,
headers=headers,
body=body,
label=label,
)
def csic_to_parsed_entry(req: CSICRequest) -> ParsedLogEntry:
"""
Convert a CSICRequest to a ParsedLogEntry with synthesized defaults
for fields not present in the CSIC dataset
"""
ua = req.headers.get("User-Agent", _DEFAULT_UA)
query = req.query_string
if req.body:
query = (
f"{query}&{req.body}" if query else req.body
)
return ParsedLogEntry(
ip=_DEFAULT_IP,
timestamp=datetime.now(UTC),
method=req.method,
path=req.path,
query_string=query,
status_code=200,
response_size=0,
referer="",
user_agent=ua,
raw_line="",
)
def load_csic_dataset(
normal_path: Path,
attack_path: Path,
) -> tuple[np.ndarray, np.ndarray]:
"""
Load CSIC 2010 normal and attack files, extract features,
and return (X, y) arrays ready for model training
"""
normal_reqs = parse_csic_file(normal_path, label=0)
attack_reqs = parse_csic_file(attack_path, label=1)
all_reqs = normal_reqs + attack_reqs
vectors: list[list[float]] = []
labels: list[int] = []
for req in all_reqs:
entry = csic_to_parsed_entry(req)
features = extract_request_features(entry)
for name in _WINDOWED_FEATURE_NAMES:
features[name] = 0.0
vector = encode_for_inference(features)
vectors.append(vector)
labels.append(req.label)
X = np.array(vectors, dtype=np.float32)
y = np.array(labels, dtype=np.int32)
logger.info(
"Dataset loaded: X=%s, y=%s (normal=%d, attack=%d)",
X.shape,
y.shape,
np.sum(y == 0),
np.sum(y == 1),
)
return X, y

354
PROJECTS/advanced/ai-threat-detection/backend/tests/test_data_loader.py Normal file
View File

@ -0,0 +1,354 @@
"""
©AngelaMos | 2026
test_data_loader.py
"""
from pathlib import Path
import numpy as np
from ml.data_loader import (
CSICRequest,
csic_to_parsed_entry,
load_csic_dataset,
parse_csic_file,
)
NORMAL_GET_FIXTURE = """\
GET /tienda1/publico/anadir.jsp?id=2&nombre=Jam%F3n HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Accept: text/xml,application/xml,application/xhtml+xml,text/html;q=0.9,text/plain;q=0.8,image/png,*/*;q=0.5
Accept-Language: en
Accept-Charset: iso-8859-1,*,utf-8
Accept-Encoding: x-gzip, x-deflate, gzip, deflate
Connection: close
GET /tienda1/publico/pagar.jsp HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Accept: text/xml,application/xml
Accept-Language: en
Connection: close
GET /tienda1/publico/entrar.jsp HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Accept: text/xml
Connection: close
POST /tienda1/publico/registro.jsp HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Accept: text/xml,application/xml
Content-Type: application/x-www-form-urlencoded
Content-Length: 64
Connection: close
nombre=Juan&apellidos=Garcia&email=juan@example.com&submit=Enviar
GET /tienda1/publico/vac498.jsp?manufacturer=Dell HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Accept: text/xml
Connection: close
"""
ATTACK_FIXTURE = """\
GET /tienda1/publico/anadir.jsp?id=2&nombre=Jam%F3n'+OR+1=1-- HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Accept: text/xml
Connection: close
POST /tienda1/publico/autenticar.jsp HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Content-Type: application/x-www-form-urlencoded
Content-Length: 68
Connection: close
usuario=admin'+OR+1=1--&contrasenya=pass&B1=Enviar
GET /tienda1/publico/../../../etc/passwd HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Accept: text/xml
Connection: close
GET /tienda1/publico/anadir.jsp?id=<script>alert(1)</script> HTTP/1.1
Host: localhost:8080
User-Agent: Mozilla/5.0 (compatible; Konqueror/3.5; Linux) KHTML/3.5.8 (like Gecko)
Accept: text/xml
Connection: close
"""
MALFORMED_FIXTURE = """\
THIS IS NOT HTTP
GET /valid/path HTTP/1.1
Host: localhost:8080
User-Agent: TestBot
Connection: close
just some random garbage here
and more garbage
"""
class TestParseCSICFile:
"""
Test CSIC 2010 file parser
"""
def test_parses_normal_get_requests(self, tmp_path: Path) -> None:
"""
Normal GET requests are parsed into CSICRequest dataclass instances
"""
f = tmp_path / "normalTrafficTraining.txt"
f.write_text(NORMAL_GET_FIXTURE, encoding="utf-8")
results = parse_csic_file(f, label=0)
assert len(results) == 5
assert all(isinstance(r, CSICRequest) for r in results)
assert all(r.label == 0 for r in results)
def test_parses_get_method_and_path(self, tmp_path: Path) -> None:
"""
First GET request has correct method, path, and query string
"""
f = tmp_path / "normal.txt"
f.write_text(NORMAL_GET_FIXTURE, encoding="utf-8")
results = parse_csic_file(f, label=0)
first = results[0]
assert first.method == "GET"
assert first.path == "/tienda1/publico/anadir.jsp"
assert first.query_string == "id=2&nombre=Jam%F3n"
assert first.protocol == "HTTP/1.1"
def test_parses_headers(self, tmp_path: Path) -> None:
"""
Headers are captured as a dict
"""
f = tmp_path / "normal.txt"
f.write_text(NORMAL_GET_FIXTURE, encoding="utf-8")
results = parse_csic_file(f, label=0)
first = results[0]
assert first.headers["Host"] == "localhost:8080"
assert "Konqueror" in first.headers["User-Agent"]
def test_parses_post_with_body(self, tmp_path: Path) -> None:
"""
POST request body is captured
"""
f = tmp_path / "normal.txt"
f.write_text(NORMAL_GET_FIXTURE, encoding="utf-8")
results = parse_csic_file(f, label=0)
post_req = results[3]
assert post_req.method == "POST"
assert "nombre=Juan" in post_req.body
def test_parses_attack_file_with_label_1(
self, tmp_path: Path) -> None:
"""
Attack file entries get label=1
"""
f = tmp_path / "anomalous.txt"
f.write_text(ATTACK_FIXTURE, encoding="utf-8")
results = parse_csic_file(f, label=1)
assert len(results) == 4
assert all(r.label == 1 for r in results)
def test_attack_sqli_in_query(self, tmp_path: Path) -> None:
"""
SQLi payload appears in query string of attack GET
"""
f = tmp_path / "anomalous.txt"
f.write_text(ATTACK_FIXTURE, encoding="utf-8")
results = parse_csic_file(f, label=1)
first = results[0]
assert "OR+1=1" in first.query_string
def test_attack_sqli_in_body(self, tmp_path: Path) -> None:
"""
SQLi payload appears in POST body of attack
"""
f = tmp_path / "anomalous.txt"
f.write_text(ATTACK_FIXTURE, encoding="utf-8")
results = parse_csic_file(f, label=1)
post_req = results[1]
assert post_req.method == "POST"
assert "OR+1=1" in post_req.body
def test_malformed_blocks_skipped(
self, tmp_path: Path) -> None:
"""
Malformed/non-HTTP lines are skipped gracefully
"""
f = tmp_path / "malformed.txt"
f.write_text(MALFORMED_FIXTURE, encoding="utf-8")
results = parse_csic_file(f, label=0)
assert len(results) == 1
assert results[0].method == "GET"
assert results[0].path == "/valid/path"
def test_empty_file_returns_empty_list(
self, tmp_path: Path) -> None:
"""
Empty file returns an empty list
"""
f = tmp_path / "empty.txt"
f.write_text("", encoding="utf-8")
results = parse_csic_file(f, label=0)
assert results == []
class TestCSICToParsedEntry:
"""
Test conversion from CSICRequest to ParsedLogEntry
"""
def test_converts_get_request(self) -> None:
"""
GET CSICRequest converts to ParsedLogEntry with synthesized fields
"""
req = CSICRequest(
method="GET",
path="/tienda1/publico/anadir.jsp",
query_string="id=2&nombre=test",
protocol="HTTP/1.1",
headers={
"Host": "localhost:8080",
"User-Agent": "Mozilla/5.0 (compatible; Konqueror/3.5)",
},
body="",
label=0,
)
entry = csic_to_parsed_entry(req)
assert entry.method == "GET"
assert entry.path == "/tienda1/publico/anadir.jsp"
assert entry.query_string == "id=2&nombre=test"
assert entry.status_code == 200
assert entry.response_size == 0
assert entry.user_agent == "Mozilla/5.0 (compatible; Konqueror/3.5)"
assert entry.ip != ""
assert entry.timestamp is not None
def test_converts_post_with_body_in_query(self) -> None:
"""
POST body is appended to query_string for feature extraction
"""
req = CSICRequest(
method="POST",
path="/login",
query_string="",
protocol="HTTP/1.1",
headers={"User-Agent": "TestBot"},
body="user=admin'+OR+1=1--&pass=x",
label=1,
)
entry = csic_to_parsed_entry(req)
assert entry.method == "POST"
assert "OR+1=1" in entry.query_string
def test_missing_ua_gets_default(self) -> None:
"""
CSICRequest without User-Agent header gets a default user agent
"""
req = CSICRequest(
method="GET",
path="/test",
query_string="",
protocol="HTTP/1.1",
headers={"Host": "localhost"},
body="",
label=0,
)
entry = csic_to_parsed_entry(req)
assert len(entry.user_agent) > 0
class TestLoadCSICDataset:
"""
Test end-to-end dataset loading with feature extraction
"""
def test_returns_correct_shape(self, tmp_path: Path) -> None:
"""
load_csic_dataset returns X with 35 columns and matching y
"""
normal = tmp_path / "normal.txt"
attack = tmp_path / "attack.txt"
normal.write_text(NORMAL_GET_FIXTURE, encoding="utf-8")
attack.write_text(ATTACK_FIXTURE, encoding="utf-8")
X, y = load_csic_dataset(normal, attack)
assert X.shape[1] == 35
assert X.shape[0] == y.shape[0]
def test_contains_both_labels(self, tmp_path: Path) -> None:
"""
y array contains both 0 (normal) and 1 (attack) labels
"""
normal = tmp_path / "normal.txt"
attack = tmp_path / "attack.txt"
normal.write_text(NORMAL_GET_FIXTURE, encoding="utf-8")
attack.write_text(ATTACK_FIXTURE, encoding="utf-8")
_, y = load_csic_dataset(normal, attack)
assert 0 in y
assert 1 in y
def test_label_counts_match_files(
self, tmp_path: Path) -> None:
"""
Normal and attack counts match the number of requests in each file
"""
normal = tmp_path / "normal.txt"
attack = tmp_path / "attack.txt"
normal.write_text(NORMAL_GET_FIXTURE, encoding="utf-8")
attack.write_text(ATTACK_FIXTURE, encoding="utf-8")
_, y = load_csic_dataset(normal, attack)
assert np.sum(y == 0) == 5
assert np.sum(y == 1) == 4
def test_feature_values_are_finite(
self, tmp_path: Path) -> None:
"""
All feature values are finite (no NaN or Inf)
"""
normal = tmp_path / "normal.txt"
attack = tmp_path / "attack.txt"
normal.write_text(NORMAL_GET_FIXTURE, encoding="utf-8")
attack.write_text(ATTACK_FIXTURE, encoding="utf-8")
X, _ = load_csic_dataset(normal, attack)
assert np.all(np.isfinite(X))