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

chore: lint and format all Phase 3 training code

This commit is contained in:
CarterPerez-dev 2026-02-28 09:23:12 -05:00
parent 37738b4d0b
commit d471de31fd
29 changed files with 290 additions and 556 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
PROJECTS/advanced/ai-threat-detection/backend/app/api/deps.py
View File

@ -17,4 +17,3 @@ async def get_session(request: Request) -> AsyncIterator[AsyncSession]:
factory = request.app.state.session_factory
async with factory() as session:
yield session
await session.commit()

42
PROJECTS/advanced/ai-threat-detection/backend/app/api/ingest.py Normal file
View File

@ -0,0 +1,42 @@
"""
©AngelaMos | 2026
ingest.py
"""
import asyncio
from fastapi import APIRouter, Request
from pydantic import BaseModel
router = APIRouter(prefix="/ingest", tags=["ingest"])
class BatchIngestRequest(BaseModel):
"""
Payload for bulk log line ingestion
"""
lines: list[str]
@router.post("/batch", status_code=200)
async def ingest_batch(
body: BatchIngestRequest,
request: Request,
) -> dict[str, int]:
"""
Push a batch of raw log lines into the pipeline queue
"""
pipeline = getattr(request.app.state, "pipeline", None)
if pipeline is None:
return {"queued": 0}
queued = 0
for line in body.lines:
try:
pipeline.raw_queue.put_nowait(line)
queued += 1
except asyncio.QueueFull:
break
return {"queued": queued}

2
PROJECTS/advanced/ai-threat-detection/backend/app/core/detection/ensemble.py
View File

@ -38,7 +38,7 @@ def fuse_scores(
weight_sum += weight
if weight_sum == 0:
return 0.0
return total / weight_sum * weight_sum
return total / weight_sum
def blend_scores(

4
PROJECTS/advanced/ai-threat-detection/backend/app/core/detection/rules.py
View File

@ -11,8 +11,10 @@ from app.core.features.patterns import (
COMMAND_INJECTION,
DOUBLE_ENCODED,
FILE_INCLUSION,
LOG4SHELL,
PATH_TRAVERSAL,
SQLI,
SSRF,
XSS,
)
from app.core.features.signatures import SCANNER_USER_AGENTS
@ -41,10 +43,12 @@ class _ThresholdRule(NamedTuple):
_PATTERN_RULES: list[_PatternRule] = [
_PatternRule("LOG4SHELL", LOG4SHELL, 0.95),
_PatternRule("COMMAND_INJECTION", COMMAND_INJECTION, 0.90),
_PatternRule("SQL_INJECTION", SQLI, 0.85),
_PatternRule("XSS", XSS, 0.80),
_PatternRule("FILE_INCLUSION", FILE_INCLUSION, 0.75),
_PatternRule("SSRF", SSRF, 0.70),
_PatternRule("PATH_TRAVERSAL", PATH_TRAVERSAL, 0.60),
]

2
PROJECTS/advanced/ai-threat-detection/backend/app/core/features/aggregator.py
View File

@ -91,7 +91,7 @@ class WindowAggregator:
results = await pipe.execute()
read_start = 14
read_start = len(keys) * 2
req_count_1m = results[read_start]
req_count_5m = results[read_start + 1]
req_count_10m = results[read_start + 2]

21
PROJECTS/advanced/ai-threat-detection/backend/app/core/features/patterns.py
View File

@ -76,14 +76,31 @@ _FILE_INCLUSION = (r"(?:php://|"
r"phar://|"
r"glob://)")
_SSRF = (
r"(?:169\.254\.169\.254|"
r"metadata\.google\.internal|"
r"169\.254\.170\.2|"
r"100\.100\.100\.200|"
r"(?:localhost|127\.0\.0\.1|0\.0\.0\.0)(?::\d+)?/(?!$)|"
r"file:///(?:etc|proc|sys)|"
r"dict://|"
r"gopher://)"
)
_LOG4SHELL = r"(?:\$\{(?:j(?:ndi|ava)|lower|upper|:-|:\+|#))"
SQLI = re.compile(_SQLI, re.IGNORECASE)
XSS = re.compile(_XSS, re.IGNORECASE)
PATH_TRAVERSAL = re.compile(_PATH_TRAVERSAL, re.IGNORECASE)
COMMAND_INJECTION = re.compile(_COMMAND_INJECTION, re.IGNORECASE)
FILE_INCLUSION = re.compile(_FILE_INCLUSION, re.IGNORECASE)
SSRF = re.compile(_SSRF, re.IGNORECASE)
LOG4SHELL = re.compile(_LOG4SHELL, re.IGNORECASE)
ATTACK_COMBINED = re.compile(
r"|".join(
(_SQLI, _XSS, _PATH_TRAVERSAL, _COMMAND_INJECTION, _FILE_INCLUSION)),
r"|".join((
_SQLI, _XSS, _PATH_TRAVERSAL, _COMMAND_INJECTION,
_FILE_INCLUSION, _SSRF, _LOG4SHELL,
)),
re.IGNORECASE,
)

12
PROJECTS/advanced/ai-threat-detection/backend/app/core/ingestion/tailer.py
View File

@ -56,6 +56,15 @@ class _LogHandler(FileSystemEventHandler):
self._file = None
self._inode = None
def _enqueue(self, line: str) -> None:
"""
Push one line into the queue, logging drops on full queue
"""
try:
self._queue.put_nowait(line)
except asyncio.QueueFull:
logger.warning("Raw queue full — log line dropped")
def _read_new_lines(self) -> None:
"""
Read all new complete lines from the current file position.
@ -66,8 +75,7 @@ class _LogHandler(FileSystemEventHandler):
for line in self._file:
stripped = line.rstrip("\n\r")
if stripped:
self._loop.call_soon_threadsafe(self._queue.put_nowait,
stripped)
self._loop.call_soon_threadsafe(self._enqueue, stripped)
def _handle_rotation(self) -> None:
"""

2
PROJECTS/advanced/ai-threat-detection/backend/app/factory.py
View File

@ -152,12 +152,14 @@ def create_app() -> FastAPI:
app.state.pipeline_running = False
from app.api.health import router as health_router
from app.api.ingest import router as ingest_router
from app.api.models_api import router as models_router
from app.api.stats import router as stats_router
from app.api.threats import router as threats_router
from app.api.websocket import router as ws_router
app.include_router(health_router)
app.include_router(ingest_router)
app.include_router(threats_router)
app.include_router(stats_router)
app.include_router(models_router)

2
PROJECTS/advanced/ai-threat-detection/backend/app/schemas/stats.py
View File

@ -40,7 +40,7 @@ class StatsResponse(BaseModel):
"""
time_range: str
total_requests: int
threats_stored: int
threats_detected: int
severity_breakdown: SeverityBreakdown
top_source_ips: list[IPStatEntry]

2
PROJECTS/advanced/ai-threat-detection/backend/app/services/stats_service.py
View File

@ -67,7 +67,7 @@ async def get_stats(
return StatsResponse(
time_range=time_range,
total_requests=total,
threats_stored=total,
threats_detected=threats_detected,
severity_breakdown=SeverityBreakdown(
high=sev_map.get("HIGH", 0),

1
PROJECTS/advanced/ai-threat-detection/backend/ml/autoencoder.py
View File

@ -45,7 +45,6 @@ class ThreatAutoencoder(nn.Module):
nn.LeakyReLU(0.2),
nn.Dropout(0.2),
nn.Linear(24, input_dim),
nn.Sigmoid(),
)
def encode(self, x: Tensor) -> Tensor:

17
PROJECTS/advanced/ai-threat-detection/backend/ml/data_loader.py
View File

@ -19,15 +19,12 @@ 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*$"
)
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)"
)
_DEFAULT_UA = ("Mozilla/5.0 (compatible; Konqueror/3.5; Linux)"
" KHTML/3.5.8 (like Gecko)")
_WINDOWED_FEATURE_NAMES: list[str] = [
"req_count_1m",
@ -136,9 +133,7 @@ def _parse_request_block(
key, value = line.split(": ", 1)
headers[key] = value
body_lines = [
ln for ln in lines[body_start:] if ln.strip()
]
body_lines = [ln for ln in lines[body_start:] if ln.strip()]
body = "\n".join(body_lines)
return CSICRequest(
@ -161,9 +156,7 @@ def csic_to_parsed_entry(req: CSICRequest) -> ParsedLogEntry:
query = req.query_string
if req.body:
query = (
f"{query}&{req.body}" if query else req.body
)
query = (f"{query}&{req.body}" if query else req.body)
return ParsedLogEntry(
ip=_DEFAULT_IP,

28
PROJECTS/advanced/ai-threat-detection/backend/ml/download_csic.py
View File

@ -13,11 +13,9 @@ logger = logging.getLogger(__name__)
DATASET_DIR = Path("data/datasets/csic2010")
BASE_URL = (
"https://gitlab.fing.edu.uy"
"/gsi/web-application-attacks-datasets"
"/-/raw/master/csic_2010"
)
BASE_URL = ("https://gitlab.fing.edu.uy"
"/gsi/web-application-attacks-datasets"
"/-/raw/master/csic_2010")
FILES = [
"normalTrafficTraining.txt",
@ -57,9 +55,7 @@ def _compute_sha256(path: Path) -> str:
return h.hexdigest()
def download_csic(
output_dir: Path = DATASET_DIR,
) -> None:
def download_csic(output_dir: Path = DATASET_DIR, ) -> None:
"""
Download CSIC 2010 dataset files
"""
@ -69,9 +65,7 @@ def download_csic(
dest = output_dir / filename
if dest.exists() and dest.stat().st_size > MIN_FILE_BYTES:
logger.info(
"Skipping %s (already exists)", filename
)
logger.info("Skipping %s (already exists)", filename)
continue
url = f"{BASE_URL}/{filename}"
@ -92,16 +86,12 @@ def download_csic(
size = dest.stat().st_size
sha = _compute_sha256(dest)
print(
f" Saved: {dest}"
f" ({size:,} bytes, sha256={sha[:12]})"
)
print(f" Saved: {dest}"
f" ({size:,} bytes, sha256={sha[:12]})")
if size < MIN_FILE_BYTES:
print(
f" WARNING: {filename} is suspiciously"
f" small ({size:,} bytes)"
)
print(f" WARNING: {filename} is suspiciously"
f" small ({size:,} bytes)")
print(f"\nDataset directory: {output_dir.resolve()}")

3
PROJECTS/advanced/ai-threat-detection/backend/ml/metadata.py
View File

@ -52,8 +52,7 @@ async def save_model_metadata(
select(ModelMetadata).where(
ModelMetadata.model_type == model_type,
ModelMetadata.is_active == True, # noqa: E712
)
)
))
for old in result.scalars().all():
await session.delete(old)
await session.flush()

62
PROJECTS/advanced/ai-threat-detection/backend/ml/orchestrator.py
View File

@ -88,9 +88,7 @@ class TrainingOrchestrator:
len(split.X_normal_train),
)
with VigilExperiment(
self._experiment_name
) as experiment:
with VigilExperiment(self._experiment_name) as experiment:
experiment.log_params({
"epochs": self._epochs,
"batch_size": self._batch_size,
@ -100,37 +98,26 @@ class TrainingOrchestrator:
"n_features": X.shape[1],
})
ae_result = self._train_ae(
split.X_normal_train
)
ae_result = self._train_ae(split.X_normal_train)
ae_metrics = {
"ae_threshold": ae_result["threshold"],
"ae_final_train_loss": ae_result[
"history"
]["train_loss"][-1],
"ae_final_val_loss": ae_result[
"history"
]["val_loss"][-1],
"ae_final_train_loss": ae_result["history"]["train_loss"][-1],
"ae_final_val_loss": ae_result["history"]["val_loss"][-1],
}
rf_result = self._train_rf(
split.X_train, split.y_train
)
rf_result = self._train_rf(split.X_train, split.y_train)
rf_metrics = rf_result["metrics"]
if_result = self._train_if(
split.X_normal_train
)
if_result = self._train_if(split.X_normal_train)
if_metrics = if_result["metrics"]
self._export_models(
ae_result, rf_result, if_result
)
self._export_models(ae_result, rf_result, if_result)
experiment.log_metrics(ae_metrics)
experiment.log_metrics(
{f"rf_{k}": v for k, v in rf_metrics.items()}
)
experiment.log_metrics({
f"rf_{k}": v
for k, v in rf_metrics.items()
})
try:
ensemble = validate_ensemble(
@ -172,9 +159,7 @@ class TrainingOrchestrator:
mlflow_run_id=run_id,
)
def _train_ae(
self, X_normal: np.ndarray
) -> dict:
def _train_ae(self, X_normal: np.ndarray) -> dict:
"""
Train the autoencoder on normal-only data
"""
@ -189,9 +174,7 @@ class TrainingOrchestrator:
batch_size=self._batch_size,
)
def _train_rf(
self, X: np.ndarray, y: np.ndarray
) -> dict:
def _train_rf(self, X: np.ndarray, y: np.ndarray) -> dict:
"""
Train the random forest classifier
"""
@ -201,9 +184,7 @@ class TrainingOrchestrator:
)
return train_random_forest(X, y)
def _train_if(
self, X_normal: np.ndarray
) -> dict:
def _train_if(self, X_normal: np.ndarray) -> dict:
"""
Train the isolation forest on normal-only data
"""
@ -226,15 +207,10 @@ class TrainingOrchestrator:
ae_result["model"],
self._output_dir / AE_FILENAME,
)
ae_result["scaler"].save_json(
self._output_dir / SCALER_FILENAME
)
threshold_data = {
"threshold": ae_result["threshold"]
}
ae_result["scaler"].save_json(self._output_dir / SCALER_FILENAME)
threshold_data = {"threshold": ae_result["threshold"]}
(self._output_dir / THRESHOLD_FILENAME).write_text(
json.dumps(threshold_data, indent=2)
)
json.dumps(threshold_data, indent=2))
export_random_forest(
rf_result["model"],
@ -248,6 +224,4 @@ class TrainingOrchestrator:
self._output_dir / IF_FILENAME,
)
logger.info(
"Exported models to %s", self._output_dir
)
logger.info("Exported models to %s", self._output_dir)

13
PROJECTS/advanced/ai-threat-detection/backend/ml/splitting.py
View File

@ -39,9 +39,7 @@ def prepare_training_data(
"""
n_classes = len(np.unique(y))
if n_classes < 2:
raise ValueError(
"y must contain at least 2 classes"
)
raise ValueError("y must contain at least 2 classes")
test_size = 1.0 - train_ratio
X_train, X_rem, y_train, y_rem = train_test_split(
@ -67,18 +65,13 @@ def prepare_training_data(
majority_count = class_counts.max()
current_ratio = minority_count / majority_count
if (
minority_count >= smote_k + 1
and current_ratio < smote_strategy
):
if (minority_count >= smote_k + 1 and current_ratio < smote_strategy):
sampler = SMOTE(
sampling_strategy=smote_strategy,
k_neighbors=smote_k,
random_state=random_state,
)
X_train, y_train = sampler.fit_resample(
X_train, y_train
)
X_train, y_train = sampler.fit_resample(X_train, y_train)
return TrainingSplit(
X_train=X_train,

97
PROJECTS/advanced/ai-threat-detection/backend/ml/synthetic.py
View File

@ -253,9 +253,7 @@ def _make_entry(
)
def generate_sqli_requests(
n: int,
) -> list[ParsedLogEntry]:
def generate_sqli_requests(n: int, ) -> list[ParsedLogEntry]:
"""
Generate n requests with SQL injection payloads
"""
@ -273,8 +271,7 @@ def generate_sqli_requests(
response_size=random.randint(0, 5000),
user_agent=random.choice(NORMAL_UAS),
ip=_random_private_ip(),
)
)
))
else:
entries.append(
_make_entry(
@ -285,14 +282,11 @@ def generate_sqli_requests(
response_size=random.randint(0, 2000),
user_agent=random.choice(NORMAL_UAS),
ip=_random_private_ip(),
)
)
))
return entries
def generate_xss_requests(
n: int,
) -> list[ParsedLogEntry]:
def generate_xss_requests(n: int, ) -> list[ParsedLogEntry]:
"""
Generate n requests with XSS payloads
"""
@ -309,14 +303,11 @@ def generate_xss_requests(
response_size=random.randint(500, 5000),
user_agent=random.choice(NORMAL_UAS),
ip=_random_private_ip(),
)
)
))
return entries
def generate_traversal_requests(
n: int,
) -> list[ParsedLogEntry]:
def generate_traversal_requests(n: int, ) -> list[ParsedLogEntry]:
"""
Generate n requests with path traversal payloads
"""
@ -332,14 +323,11 @@ def generate_traversal_requests(
response_size=random.randint(0, 1000),
user_agent=random.choice(NORMAL_UAS),
ip=_random_private_ip(),
)
)
))
return entries
def generate_log4shell_requests(
n: int,
) -> list[ParsedLogEntry]:
def generate_log4shell_requests(n: int, ) -> list[ParsedLogEntry]:
"""
Generate n requests with Log4Shell JNDI payloads
"""
@ -356,14 +344,11 @@ def generate_log4shell_requests(
response_size=random.randint(0, 2000),
user_agent=payload,
ip=_random_private_ip(),
)
)
))
return entries
def generate_ssrf_requests(
n: int,
) -> list[ParsedLogEntry]:
def generate_ssrf_requests(n: int, ) -> list[ParsedLogEntry]:
"""
Generate n requests with SSRF target URLs
"""
@ -380,14 +365,11 @@ def generate_ssrf_requests(
response_size=random.randint(0, 3000),
user_agent=random.choice(NORMAL_UAS),
ip=_random_private_ip(),
)
)
))
return entries
def generate_scanner_requests(
n: int,
) -> list[ParsedLogEntry]:
def generate_scanner_requests(n: int, ) -> list[ParsedLogEntry]:
"""
Generate n requests mimicking vulnerability scanners
"""
@ -396,25 +378,18 @@ def generate_scanner_requests(
path = random.choice(ATTACK_PATHS)
entries.append(
_make_entry(
method=random.choice(
["GET", "HEAD", "OPTIONS"]
),
method=random.choice(["GET", "HEAD", "OPTIONS"]),
path=path,
query_string="",
status_code=random.choice(
[200, 301, 403, 404]
),
status_code=random.choice([200, 301, 403, 404]),
response_size=random.randint(0, 500),
user_agent=random.choice(SCANNER_UAS),
ip=_random_private_ip(),
)
)
))
return entries
def generate_normal_requests(
n: int,
) -> list[ParsedLogEntry]:
def generate_normal_requests(n: int, ) -> list[ParsedLogEntry]:
"""
Generate n realistic benign HTTP requests
"""
@ -422,31 +397,20 @@ def generate_normal_requests(
for _ in range(n):
path = random.choice(NORMAL_PATHS)
has_query = random.random() < 0.3
query = (
f"page={random.randint(1, 20)}"
if has_query
else ""
)
query = (f"page={random.randint(1, 20)}" if has_query else "")
entries.append(
_make_entry(
method=random.choice(
["GET", "GET", "GET", "POST"]
),
method=random.choice(["GET", "GET", "GET", "POST"]),
path=path,
query_string=query,
status_code=random.choice(
[200, 200, 200, 301, 304]
),
status_code=random.choice([200, 200, 200, 301, 304]),
response_size=random.randint(200, 50000),
user_agent=random.choice(NORMAL_UAS),
)
)
))
return entries
def _entries_to_vectors(
entries: list[ParsedLogEntry],
) -> list[list[float]]:
def _entries_to_vectors(entries: list[ParsedLogEntry], ) -> list[list[float]]:
"""
Convert ParsedLogEntry list to 35-dim feature vectors
"""
@ -473,20 +437,12 @@ def generate_mixed_dataset(
remainder = n_attack - (per_type * 6)
attacks: list[ParsedLogEntry] = []
attacks.extend(
generate_sqli_requests(per_type + remainder)
)
attacks.extend(generate_sqli_requests(per_type + remainder))
attacks.extend(generate_xss_requests(per_type))
attacks.extend(
generate_traversal_requests(per_type)
)
attacks.extend(
generate_log4shell_requests(per_type)
)
attacks.extend(generate_traversal_requests(per_type))
attacks.extend(generate_log4shell_requests(per_type))
attacks.extend(generate_ssrf_requests(per_type))
attacks.extend(
generate_scanner_requests(per_type)
)
attacks.extend(generate_scanner_requests(per_type))
attack_vectors = _entries_to_vectors(attacks)
X = np.array(
@ -494,8 +450,7 @@ def generate_mixed_dataset(
dtype=np.float32,
)
y = np.array(
[0] * len(normal_vectors)
+ [1] * len(attack_vectors),
[0] * len(normal_vectors) + [1] * len(attack_vectors),
dtype=np.int32,
)

36
PROJECTS/advanced/ai-threat-detection/backend/ml/validation.py
View File

@ -66,34 +66,20 @@ def validate_ensemble(
engine = InferenceEngine(model_dir=str(model_dir))
if not engine.is_loaded:
raise RuntimeError(
f"Failed to load models from {model_dir}"
)
raise RuntimeError(f"Failed to load models from {model_dir}")
raw_scores = engine.predict(
X_test.astype(np.float32),
)
raw_scores = engine.predict(X_test.astype(np.float32), )
if raw_scores is None:
raise RuntimeError("Inference returned None")
fused = _compute_fused_scores(
raw_scores, engine.threshold, weights
)
fused = _compute_fused_scores(raw_scores, engine.threshold, weights)
y_pred = (fused >= BINARY_THRESHOLD).astype(np.int32)
prec = float(
precision_score(y_test, y_pred, zero_division=0)
)
rec = float(
recall_score(y_test, y_pred, zero_division=0)
)
f1_val = float(
f1_score(y_test, y_pred, zero_division=0)
)
pr_auc_val = float(
average_precision_score(y_test, fused)
)
prec = float(precision_score(y_test, y_pred, zero_division=0))
rec = float(recall_score(y_test, y_pred, zero_division=0))
f1_val = float(f1_score(y_test, y_pred, zero_division=0))
pr_auc_val = float(average_precision_score(y_test, fused))
roc_auc_val = float(roc_auc_score(y_test, fused))
cm = confusion_matrix(y_test, y_pred).tolist()
@ -141,13 +127,9 @@ def _compute_fused_scores(
for i in range(n_samples):
per_model: dict[str, float] = {}
per_model["ae"] = normalize_ae_score(
raw_scores["ae"][i], threshold
)
per_model["ae"] = normalize_ae_score(raw_scores["ae"][i], threshold)
per_model["rf"] = raw_scores["rf"][i]
per_model["if"] = normalize_if_score(
raw_scores["if"][i]
)
per_model["if"] = normalize_if_score(raw_scores["if"][i])
fused[i] = fuse_scores(per_model, weights)

68
PROJECTS/advanced/ai-threat-detection/backend/tests/test_api.py
View File

@ -9,63 +9,24 @@ import uuid
from datetime import datetime, UTC
import pytest
from httpx import ASGITransport, AsyncClient
from app.main import app
from app.models.threat_event import ThreatEvent
@pytest.mark.asyncio
async def test_health_returns_200() -> None:
async def test_health_returns_200(db_client) -> None:
"""
Health endpoint returns 200 with status and uptime.
Health endpoint returns 200 with status, uptime, and pipeline flag.
"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport,
base_url="http://test") as client:
response = await client.get("/health")
response = await db_client.get("/health")
assert response.status_code == 200
data = response.json()
assert data["status"] == "healthy"
assert "uptime_seconds" in data
assert isinstance(data["uptime_seconds"], int | float)
assert "pipeline_running" in data
@pytest.mark.asyncio
async def test_health_returns_pipeline_status() -> None:
"""
Health response includes pipeline_running boolean.
"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport,
base_url="http://test") as client:
response = await client.get("/health")
data = response.json()
assert isinstance(data["pipeline_running"], bool)
@pytest.mark.asyncio
async def test_ready_returns_check_structure() -> None:
"""
Readiness endpoint returns structured component checks.
"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport,
base_url="http://test") as client:
response = await client.get("/ready")
assert response.status_code in (200, 503)
data = response.json()
assert "status" in data
assert "checks" in data
assert "database" in data["checks"]
assert "redis" in data["checks"]
assert "models_loaded" in data["checks"]
@pytest.mark.asyncio
async def test_list_threats_empty(db_client) -> None:
"""
@ -171,7 +132,7 @@ async def test_stats_empty_window(db_client) -> None:
assert response.status_code == 200
data = response.json()
assert data["time_range"] == "24h"
assert data["total_requests"] == 0
assert data["threats_stored"] == 0
assert data["threats_detected"] == 0
assert data["severity_breakdown"]["high"] == 0
assert data["severity_breakdown"]["medium"] == 0
@ -181,30 +142,25 @@ async def test_stats_empty_window(db_client) -> None:
@pytest.mark.asyncio
async def test_model_status() -> None:
async def test_model_status(db_client) -> None:
"""
GET /models/status returns rules detection mode
GET /models/status returns detection mode and active model list.
"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport,
base_url="http://test") as client:
response = await client.get("/models/status")
response = await db_client.get("/models/status")
assert response.status_code == 200
data = response.json()
assert data["detection_mode"] == "rules"
assert data["active_models"] == []
assert "detection_mode" in data
assert "active_models" in data
assert isinstance(data["active_models"], list)
@pytest.mark.asyncio
async def test_retrain_returns_202() -> None:
async def test_retrain_returns_202(db_client) -> None:
"""
POST /models/retrain returns 202 Accepted with a job ID.
"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport,
base_url="http://test") as client:
response = await client.post("/models/retrain")
response = await db_client.post("/models/retrain")
assert response.status_code == 202
data = response.json()

10
PROJECTS/advanced/ai-threat-detection/backend/tests/test_autoencoder.py
View File

@ -42,17 +42,17 @@ class TestAutoencoderArchitecture:
out = model(x)
assert out.shape == (1, 35)
def test_output_values_in_zero_one_range(self) -> None:
def test_output_is_unbounded(self) -> None:
"""
Decoder output is bounded to [0, 1] by the final activation.
Decoder output is unbounded to match RobustScaler-transformed input range.
"""
model = ThreatAutoencoder(input_dim=35)
model.eval()
x = torch.randn(32, 35)
x = torch.randn(64, 35) * 3.0
with torch.no_grad():
out = model(x)
assert out.min() >= 0.0
assert out.max() <= 1.0
assert out.shape == (64, 35)
assert out.min().item() < 0.0 or out.max().item() > 1.0
def test_reconstruction_error_shape(self) -> None:
"""

36
PROJECTS/advanced/ai-threat-detection/backend/tests/test_cli.py
View File

@ -26,35 +26,25 @@ class TestCLICommands:
Test CLI help output and argument validation
"""
def test_train_help_shows_csic_dir(
self,
) -> None:
def test_train_help_shows_csic_dir(self, ) -> None:
"""
train --help shows csic-dir option
"""
result = runner.invoke(
app, ["train", "--help"]
)
result = runner.invoke(app, ["train", "--help"])
assert result.exit_code == 0
assert "csic-dir" in _clean(result.output)
def test_train_help_shows_synthetic_options(
self,
) -> None:
def test_train_help_shows_synthetic_options(self, ) -> None:
"""
train --help shows synthetic normal and attack options
"""
result = runner.invoke(
app, ["train", "--help"]
)
result = runner.invoke(app, ["train", "--help"])
assert result.exit_code == 0
output = _clean(result.output)
assert "synthetic-normal" in output
assert "synthetic-attack" in output
def test_train_invalid_csic_dir_fails(
self,
) -> None:
def test_train_invalid_csic_dir_fails(self, ) -> None:
"""
train with nonexistent csic-dir exits with error
"""
@ -76,9 +66,7 @@ class TestCLICommands:
"""
replay --help exits cleanly and mentions log
"""
result = runner.invoke(
app, ["replay", "--help"]
)
result = runner.invoke(app, ["replay", "--help"])
assert result.exit_code == 0
assert "log" in _clean(result.output)
@ -86,9 +74,7 @@ class TestCLICommands:
"""
serve --help exits cleanly and mentions host
"""
result = runner.invoke(
app, ["serve", "--help"]
)
result = runner.invoke(app, ["serve", "--help"])
assert result.exit_code == 0
assert "host" in _clean(result.output)
@ -96,18 +82,14 @@ class TestCLICommands:
"""
config --help exits cleanly
"""
result = runner.invoke(
app, ["config", "--help"]
)
result = runner.invoke(app, ["config", "--help"])
assert result.exit_code == 0
def test_health_help(self) -> None:
"""
health --help exits cleanly
"""
result = runner.invoke(
app, ["health", "--help"]
)
result = runner.invoke(app, ["health", "--help"])
assert result.exit_code == 0
def test_replay_missing_log_fails(self) -> None:

15
PROJECTS/advanced/ai-threat-detection/backend/tests/test_data_loader.py
View File

@ -155,8 +155,7 @@ class TestParseCSICFile:
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:
def test_parses_attack_file_with_label_1(self, tmp_path: Path) -> None:
"""
Attack file entries get label=1
"""
@ -193,8 +192,7 @@ class TestParseCSICFile:
assert post_req.method == "POST"
assert "OR+1=1" in post_req.body
def test_malformed_blocks_skipped(
self, tmp_path: Path) -> None:
def test_malformed_blocks_skipped(self, tmp_path: Path) -> None:
"""
Malformed/non-HTTP lines are skipped gracefully
"""
@ -207,8 +205,7 @@ class TestParseCSICFile:
assert results[0].method == "GET"
assert results[0].path == "/valid/path"
def test_empty_file_returns_empty_list(
self, tmp_path: Path) -> None:
def test_empty_file_returns_empty_list(self, tmp_path: Path) -> None:
"""
Empty file returns an empty list
"""
@ -324,8 +321,7 @@ class TestLoadCSICDataset:
assert 0 in y
assert 1 in y
def test_label_counts_match_files(
self, tmp_path: Path) -> None:
def test_label_counts_match_files(self, tmp_path: Path) -> None:
"""
Normal and attack counts match the number of requests in each file
"""
@ -339,8 +335,7 @@ class TestLoadCSICDataset:
assert np.sum(y == 0) == 5
assert np.sum(y == 1) == 4
def test_feature_values_are_finite(
self, tmp_path: Path) -> None:
def test_feature_values_are_finite(self, tmp_path: Path) -> None:
"""
All feature values are finite (no NaN or Inf)
"""

18
PROJECTS/advanced/ai-threat-detection/backend/tests/test_features.py
View File

@ -5,9 +5,16 @@ test_features.py
Tests per-request feature extraction, Redis sliding-window aggregation, and feature encoding.
"""
import time
from datetime import datetime, UTC
import fakeredis.aioredis
import pytest
from app.core.features.aggregator import WindowAggregator
from app.core.features.encoder import encode_for_inference
from app.core.features.extractor import extract_request_features
from app.core.features.mappings import FEATURE_ORDER, METHOD_MAP, STATUS_CLASS_MAP
from app.core.ingestion.parsers import ParsedLogEntry
FEATURE_KEYS = {
@ -264,13 +271,6 @@ def test_country_code_passthrough() -> None:
assert features_empty["country_code"] == ""
import time # noqa: E402
import fakeredis.aioredis # noqa: E402
import pytest # noqa: E402
from app.core.features.aggregator import WindowAggregator # noqa: E402
AGGREGATOR_KEYS = {
"req_count_1m",
"req_count_5m",
@ -446,10 +446,6 @@ async def test_aggregator_window_boundary(aggregator) -> None:
assert result["req_count_5m"] == 2
from app.core.features.encoder import encode_for_inference # noqa: E402
from app.core.features.mappings import FEATURE_ORDER, METHOD_MAP, STATUS_CLASS_MAP # noqa: E402
def _full_features() -> dict[str, int | float | bool | str]:
"""
Build a complete 35-key feature dict with realistic values.

66
PROJECTS/advanced/ai-threat-detection/backend/tests/test_metadata.py
View File

@ -35,9 +35,9 @@ async def db_session(tmp_path: Path):
async with engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
factory = async_sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
factory = async_sessionmaker(engine,
class_=AsyncSession,
expire_on_commit=False)
async with factory() as session:
yield session
@ -53,11 +53,11 @@ def model_artifacts(tmp_path: Path) -> Path:
(tmp_path / "rf.onnx").write_bytes(b"rf-model-data-456")
(tmp_path / "if.onnx").write_bytes(b"if-model-data-789")
(tmp_path / "scaler.json").write_text(
json.dumps({"center": [0.0], "scale": [1.0]})
)
(tmp_path / "threshold.json").write_text(
json.dumps({"threshold": 0.05})
)
json.dumps({
"center": [0.0],
"scale": [1.0]
}))
(tmp_path / "threshold.json").write_text(json.dumps({"threshold": 0.05}))
return tmp_path
@ -66,46 +66,31 @@ class TestComputeModelVersion:
Test SHA-256 based model version hashing
"""
def test_returns_12_char_hex(
self, model_artifacts: Path
) -> None:
def test_returns_12_char_hex(self, model_artifacts: Path) -> None:
"""
Version string is a 12-character hex digest
"""
version = compute_model_version(
model_artifacts / "ae.onnx"
)
version = compute_model_version(model_artifacts / "ae.onnx")
assert len(version) == 12
assert all(c in "0123456789abcdef" for c in version)
def test_same_file_same_version(
self, model_artifacts: Path
) -> None:
def test_same_file_same_version(self, model_artifacts: Path) -> None:
"""
Same file produces the same version string
"""
v1 = compute_model_version(
model_artifacts / "ae.onnx"
)
v2 = compute_model_version(
model_artifacts / "ae.onnx"
)
v1 = compute_model_version(model_artifacts / "ae.onnx")
v2 = compute_model_version(model_artifacts / "ae.onnx")
assert v1 == v2
def test_different_files_different_versions(
self, model_artifacts: Path
) -> None:
def test_different_files_different_versions(self,
model_artifacts: Path) -> None:
"""
Different files produce different version strings
"""
v_ae = compute_model_version(
model_artifacts / "ae.onnx"
)
v_rf = compute_model_version(
model_artifacts / "rf.onnx"
)
v_ae = compute_model_version(model_artifacts / "ae.onnx")
v_rf = compute_model_version(model_artifacts / "rf.onnx")
assert v_ae != v_rf
@ -128,7 +113,10 @@ class TestSaveModelMetadata:
db_session,
model_dir=model_artifacts,
training_samples=500,
metrics={"f1": 0.9, "pr_auc": 0.88},
metrics={
"f1": 0.9,
"pr_auc": 0.88
},
)
assert len(rows) == 3
@ -190,9 +178,7 @@ class TestSaveModelMetadata:
metrics={"f1": 0.9},
)
(model_artifacts / "ae.onnx").write_bytes(
b"new-ae-data"
)
(model_artifacts / "ae.onnx").write_bytes(b"new-ae-data")
await save_model_metadata(
db_session,
model_dir=model_artifacts,
@ -200,13 +186,9 @@ class TestSaveModelMetadata:
metrics={"f1": 0.95},
)
result = await db_session.execute(
select(ModelMetadata)
)
result = await db_session.execute(select(ModelMetadata))
all_rows = result.scalars().all()
active_rows = [r for r in all_rows if r.is_active]
assert len(active_rows) == 3
assert all(
r.training_samples == 600 for r in active_rows
)
assert all(r.training_samples == 600 for r in active_rows)

91
PROJECTS/advanced/ai-threat-detection/backend/tests/test_orchestrator.py
View File

@ -25,19 +25,10 @@ def _make_dataset() -> tuple[np.ndarray, np.ndarray]:
Generate a small synthetic dataset for testing
"""
rng = np.random.default_rng(42)
X_normal = rng.standard_normal(
(200, N_FEATURES)
).astype(np.float32)
X_attack = (
rng.standard_normal(
(80, N_FEATURES)
).astype(np.float32)
+ 2.0
)
X_normal = rng.standard_normal((200, N_FEATURES)).astype(np.float32)
X_attack = (rng.standard_normal((80, N_FEATURES)).astype(np.float32) + 2.0)
X = np.vstack([X_normal, X_attack])
y = np.array(
[0] * 200 + [1] * 80, dtype=np.int32
)
y = np.array([0] * 200 + [1] * 80, dtype=np.int32)
return X, y
@ -46,110 +37,76 @@ class TestTrainingOrchestrator:
Test the end-to-end training orchestrator
"""
def test_produces_all_output_files(
self, tmp_path: Path
) -> None:
def test_produces_all_output_files(self, tmp_path: Path) -> None:
"""
Orchestrator produces all 5 expected output files
"""
X, y = _make_dataset()
orch = TrainingOrchestrator(
output_dir=tmp_path, epochs=3
)
orch = TrainingOrchestrator(output_dir=tmp_path, epochs=3)
orch.run(X, y)
for filename in EXPECTED_FILES:
assert (
tmp_path / filename
).exists(), f"Missing {filename}"
assert (tmp_path / filename).exists(), f"Missing {filename}"
def test_returns_training_result(
self, tmp_path: Path
) -> None:
def test_returns_training_result(self, tmp_path: Path) -> None:
"""
Returns a TrainingResult dataclass
"""
X, y = _make_dataset()
orch = TrainingOrchestrator(
output_dir=tmp_path, epochs=3
)
orch = TrainingOrchestrator(output_dir=tmp_path, epochs=3)
result = orch.run(X, y)
assert isinstance(result, TrainingResult)
def test_scaler_json_has_required_keys(
self, tmp_path: Path
) -> None:
def test_scaler_json_has_required_keys(self, tmp_path: Path) -> None:
"""
scaler.json contains center, scale, and n_features
"""
X, y = _make_dataset()
orch = TrainingOrchestrator(
output_dir=tmp_path, epochs=3
)
orch = TrainingOrchestrator(output_dir=tmp_path, epochs=3)
orch.run(X, y)
scaler_data = json.loads(
(tmp_path / "scaler.json").read_text()
)
scaler_data = json.loads((tmp_path / "scaler.json").read_text())
assert "center" in scaler_data
assert "scale" in scaler_data
assert "n_features" in scaler_data
def test_threshold_json_has_float(
self, tmp_path: Path
) -> None:
def test_threshold_json_has_float(self, tmp_path: Path) -> None:
"""
threshold.json contains a float threshold value
"""
X, y = _make_dataset()
orch = TrainingOrchestrator(
output_dir=tmp_path, epochs=3
)
orch = TrainingOrchestrator(output_dir=tmp_path, epochs=3)
orch.run(X, y)
threshold_data = json.loads(
(tmp_path / "threshold.json").read_text()
)
threshold_data = json.loads((tmp_path / "threshold.json").read_text())
assert "threshold" in threshold_data
assert isinstance(
threshold_data["threshold"], float
)
assert isinstance(threshold_data["threshold"], float)
def test_result_has_per_model_metrics(
self, tmp_path: Path
) -> None:
def test_result_has_per_model_metrics(self, tmp_path: Path) -> None:
"""
TrainingResult includes metrics for each model
"""
X, y = _make_dataset()
orch = TrainingOrchestrator(
output_dir=tmp_path, epochs=3
)
orch = TrainingOrchestrator(output_dir=tmp_path, epochs=3)
result = orch.run(X, y)
assert "ae_threshold" in result.ae_metrics
assert "f1" in result.rf_metrics
assert "n_samples" in result.if_metrics
def test_ensemble_metrics_present(
self, tmp_path: Path
) -> None:
def test_ensemble_metrics_present(self, tmp_path: Path) -> None:
"""
Ensemble validation metrics are populated
"""
X, y = _make_dataset()
orch = TrainingOrchestrator(
output_dir=tmp_path, epochs=3
)
orch = TrainingOrchestrator(output_dir=tmp_path, epochs=3)
result = orch.run(X, y)
assert result.ensemble_metrics is not None
assert 0.0 <= result.ensemble_metrics.f1 <= 1.0
def test_mlflow_run_id_set(
self, tmp_path: Path
) -> None:
def test_mlflow_run_id_set(self, tmp_path: Path) -> None:
"""
MLflow run ID is captured in the result
"""
@ -165,16 +122,12 @@ class TestTrainingOrchestrator:
assert result.mlflow_run_id is not None
assert len(result.mlflow_run_id) == 32
def test_passed_gates_is_bool(
self, tmp_path: Path
) -> None:
def test_passed_gates_is_bool(self, tmp_path: Path) -> None:
"""
passed_gates is a boolean value
"""
X, y = _make_dataset()
orch = TrainingOrchestrator(
output_dir=tmp_path, epochs=3
)
orch = TrainingOrchestrator(output_dir=tmp_path, epochs=3)
result = orch.run(X, y)
assert isinstance(result.passed_gates, bool)

74
PROJECTS/advanced/ai-threat-detection/backend/tests/test_splitting.py
View File

@ -24,8 +24,7 @@ def _make_dataset(
"""
rng = np.random.default_rng(42)
X = rng.standard_normal(
(n_normal + n_attack, n_features)
).astype(np.float32)
(n_normal + n_attack, n_features)).astype(np.float32)
y = np.array(
[0] * n_normal + [1] * n_attack,
dtype=np.int32,
@ -55,16 +54,10 @@ class TestPrepareTrainingData:
expected_val = int(TOTAL * 0.15)
expected_test = int(TOTAL * 0.15)
tolerance = int(TOTAL * 0.05)
assert abs(
result.X_val.shape[0] - expected_val
) <= tolerance
assert abs(
result.X_test.shape[0] - expected_test
) <= tolerance
assert abs(result.X_val.shape[0] - expected_val) <= tolerance
assert abs(result.X_test.shape[0] - expected_test) <= tolerance
def test_stratified_class_distribution(
self,
) -> None:
def test_stratified_class_distribution(self, ) -> None:
"""
Splits preserve the original class ratio
"""
@ -74,12 +67,8 @@ class TestPrepareTrainingData:
val_ratio = np.mean(result.y_val == 1)
test_ratio = np.mean(result.y_test == 1)
ratio_tol = 0.10
assert abs(
val_ratio - original_ratio
) < ratio_tol
assert abs(
test_ratio - original_ratio
) < ratio_tol
assert abs(val_ratio - original_ratio) < ratio_tol
assert abs(test_ratio - original_ratio) < ratio_tol
def test_smote_increases_minority(self) -> None:
"""
@ -105,24 +94,12 @@ class TestPrepareTrainingData:
remainder = TOTAL - int(TOTAL * 0.70)
half = remainder // 2
tol = 3
assert abs(
result.X_val.shape[0] - half
) <= tol
assert abs(
result.X_test.shape[0] - half
) <= tol
assert (
result.X_val.shape[0]
== result.y_val.shape[0]
)
assert (
result.X_test.shape[0]
== result.y_test.shape[0]
)
assert abs(result.X_val.shape[0] - half) <= tol
assert abs(result.X_test.shape[0] - half) <= tol
assert (result.X_val.shape[0] == result.y_val.shape[0])
assert (result.X_test.shape[0] == result.y_test.shape[0])
def test_x_normal_train_only_normals(
self,
) -> None:
def test_x_normal_train_only_normals(self, ) -> None:
"""
X_normal_train has only class-0 rows
"""
@ -130,13 +107,8 @@ class TestPrepareTrainingData:
result = prepare_training_data(X, y)
expected = int(N_NORMAL * 0.70)
tol = int(TOTAL * 0.05)
assert abs(
result.X_normal_train.shape[0] - expected
) <= tol
assert (
result.X_normal_train.shape[1]
== N_FEATURES
)
assert abs(result.X_normal_train.shape[0] - expected) <= tol
assert (result.X_normal_train.shape[1] == N_FEATURES)
def test_small_dataset_works(self) -> None:
"""
@ -147,24 +119,18 @@ class TestPrepareTrainingData:
n_attack=10,
n_features=N_FEATURES,
)
result = prepare_training_data(
X, y, smote_k=3
)
result = prepare_training_data(X, y, smote_k=3)
assert isinstance(result, TrainingSplit)
assert result.X_train.shape[0] > 0
assert result.X_val.shape[0] > 0
assert result.X_test.shape[0] > 0
def test_all_one_class_raises_value_error(
self,
) -> None:
def test_all_one_class_raises_value_error(self, ) -> None:
"""
Single-class labels raise ValueError
"""
rng = np.random.default_rng(99)
X = rng.standard_normal(
(100, N_FEATURES)
).astype(np.float32)
X = rng.standard_normal((100, N_FEATURES)).astype(np.float32)
y_zeros = np.zeros(100, dtype=np.int32)
with pytest.raises(ValueError):
prepare_training_data(X, y_zeros)
@ -172,9 +138,7 @@ class TestPrepareTrainingData:
with pytest.raises(ValueError):
prepare_training_data(X, y_ones)
def test_smote_skipped_minority_too_small(
self,
) -> None:
def test_smote_skipped_minority_too_small(self, ) -> None:
"""
Tiny minority skips SMOTE gracefully
"""
@ -183,8 +147,6 @@ class TestPrepareTrainingData:
n_attack=10,
n_features=N_FEATURES,
)
result = prepare_training_data(
X, y, smote_k=50
)
result = prepare_training_data(X, y, smote_k=50)
assert isinstance(result, TrainingSplit)
assert result.X_train.shape[0] > 0

77
PROJECTS/advanced/ai-threat-detection/backend/tests/test_synthetic.py
View File

@ -40,98 +40,74 @@ class TestGenerators:
Test individual attack and normal traffic generators
"""
def test_sqli_returns_correct_count(
self,
) -> None:
def test_sqli_returns_correct_count(self, ) -> None:
"""
generate_sqli_requests returns the requested count
"""
results = generate_sqli_requests(10)
assert len(results) == 10
def test_xss_returns_correct_count(
self,
) -> None:
def test_xss_returns_correct_count(self, ) -> None:
"""
generate_xss_requests returns the requested count
"""
results = generate_xss_requests(10)
assert len(results) == 10
def test_traversal_returns_correct_count(
self,
) -> None:
def test_traversal_returns_correct_count(self, ) -> None:
"""
generate_traversal_requests returns the requested count
"""
results = generate_traversal_requests(10)
assert len(results) == 10
def test_log4shell_returns_correct_count(
self,
) -> None:
def test_log4shell_returns_correct_count(self, ) -> None:
"""
generate_log4shell_requests returns the requested count
"""
results = generate_log4shell_requests(10)
assert len(results) == 10
def test_ssrf_returns_correct_count(
self,
) -> None:
def test_ssrf_returns_correct_count(self, ) -> None:
"""
generate_ssrf_requests returns the requested count
"""
results = generate_ssrf_requests(10)
assert len(results) == 10
def test_scanner_returns_correct_count(
self,
) -> None:
def test_scanner_returns_correct_count(self, ) -> None:
"""
generate_scanner_requests returns the requested count
"""
results = generate_scanner_requests(10)
assert len(results) == 10
def test_normal_returns_correct_count(
self,
) -> None:
def test_normal_returns_correct_count(self, ) -> None:
"""
generate_normal_requests returns the requested count
"""
results = generate_normal_requests(20)
assert len(results) == 20
def test_sqli_has_attack_payloads(
self,
) -> None:
def test_sqli_has_attack_payloads(self, ) -> None:
"""
SQLi entries contain injection patterns in query string
"""
results = generate_sqli_requests(20)
has_sqli = any(
"OR" in e.query_string
or "UNION" in e.query_string
or "DROP" in e.query_string
or "SLEEP" in e.query_string
for e in results
)
has_sqli = any("OR" in e.query_string or "UNION" in e.query_string
or "DROP" in e.query_string or "SLEEP" in e.query_string
for e in results)
assert has_sqli
def test_xss_has_script_patterns(
self,
) -> None:
def test_xss_has_script_patterns(self, ) -> None:
"""
XSS entries contain script-related patterns
"""
results = generate_xss_requests(20)
has_xss = any(
"script" in e.query_string.lower()
or "alert" in e.query_string.lower()
or "onerror" in e.query_string.lower()
for e in results
)
"script" in e.query_string.lower() or "alert" in
e.query_string.lower() or "onerror" in e.query_string.lower()
for e in results)
assert has_xss
def test_traversal_has_dotdot(self) -> None:
@ -139,15 +115,11 @@ class TestGenerators:
Traversal entries contain ../ in path
"""
results = generate_traversal_requests(20)
has_traversal = any(
".." in e.path or "%2e" in e.path.lower()
for e in results
)
has_traversal = any(".." in e.path or "%2e" in e.path.lower()
for e in results)
assert has_traversal
def test_all_entries_are_parsed_log_entry(
self,
) -> None:
def test_all_entries_are_parsed_log_entry(self, ) -> None:
"""
All generators return ParsedLogEntry instances
"""
@ -162,14 +134,9 @@ class TestGenerators:
]
for gen in generators:
results = gen(5)
assert all(
isinstance(e, ParsedLogEntry)
for e in results
)
assert all(isinstance(e, ParsedLogEntry) for e in results)
def test_entries_pass_feature_extraction(
self,
) -> None:
def test_entries_pass_feature_extraction(self, ) -> None:
"""
All generated entries extract and encode without error
"""
@ -184,9 +151,7 @@ class TestGenerators:
]
for gen in generators:
for entry in gen(5):
features = extract_request_features(
entry
)
features = extract_request_features(entry)
for name in _WINDOWED_FEATURE_NAMES:
features[name] = 0.0
vector = encode_for_inference(features)

30
PROJECTS/advanced/ai-threat-detection/backend/tests/test_training_e2e.py
View File

@ -5,7 +5,6 @@ test_training_e2e.py
from pathlib import Path
from app.core.detection.ensemble import (
blend_scores,
fuse_scores,
@ -27,24 +26,19 @@ class TestTrainingE2E:
End-to-end training integration test
"""
def test_full_training_produces_loadable_models(
self, tmp_path: Path
) -> None:
def test_full_training_produces_loadable_models(self,
tmp_path: Path) -> None:
"""
Full pipeline produces models that load and predict
"""
X, y = generate_mixed_dataset(
N_NORMAL, N_ATTACK
)
X, y = generate_mixed_dataset(N_NORMAL, N_ATTACK)
assert X.shape == (
N_NORMAL + N_ATTACK,
N_FEATURES,
)
model_dir = tmp_path / "models"
orch = TrainingOrchestrator(
output_dir=model_dir, epochs=3
)
orch = TrainingOrchestrator(output_dir=model_dir, epochs=3)
result = orch.run(X, y)
expected_files = [
@ -55,9 +49,7 @@ class TestTrainingE2E:
"threshold.json",
]
for filename in expected_files:
assert (
model_dir / filename
).exists(), f"Missing {filename}"
assert (model_dir / filename).exists(), f"Missing {filename}"
engine = InferenceEngine(str(model_dir))
assert engine.is_loaded
@ -72,12 +64,8 @@ class TestTrainingE2E:
threshold = engine.threshold
for i in range(5):
ae_score = normalize_ae_score(
predictions["ae"][i], threshold
)
if_score = normalize_if_score(
predictions["if"][i]
)
ae_score = normalize_ae_score(predictions["ae"][i], threshold)
if_score = normalize_if_score(predictions["if"][i])
rf_score = predictions["rf"][i]
scores = {
@ -85,9 +73,7 @@ class TestTrainingE2E:
"rf": rf_score,
"if": if_score,
}
fused = fuse_scores(
scores, ENSEMBLE_WEIGHTS
)
fused = fuse_scores(scores, ENSEMBLE_WEIGHTS)
assert 0.0 <= fused <= 1.0
blended = blend_scores(fused, 0.0)

14
PROJECTS/advanced/ai-threat-detection/backend/tests/test_validation.py
View File

@ -46,9 +46,7 @@ def trained_model_dir(tmp_path: Path) -> Path:
scaler.fit(X_normal)
scaler.save_json(tmp_path / "scaler.json")
(tmp_path / "threshold.json").write_text(
json.dumps({"threshold": 0.05})
)
(tmp_path / "threshold.json").write_text(json.dumps({"threshold": 0.05}))
return tmp_path
@ -60,9 +58,7 @@ def separable_test_data() -> tuple[np.ndarray, np.ndarray]:
"""
rng = np.random.default_rng(99)
X_normal = rng.standard_normal((50, 35)).astype(np.float32)
X_attack = (
rng.standard_normal((30, 35)).astype(np.float32) + 3.0
)
X_attack = (rng.standard_normal((30, 35)).astype(np.float32) + 3.0)
X = np.vstack([X_normal, X_attack])
y = np.array([0] * 50 + [1] * 30, dtype=np.int32)
return X, y
@ -183,7 +179,11 @@ class TestValidateEnsemble:
trained_model_dir,
X_test,
y_test,
ensemble_weights={"ae": 0.5, "rf": 0.3, "if": 0.2},
ensemble_weights={
"ae": 0.5,
"rf": 0.3,
"if": 0.2
},
)
assert isinstance(result, ValidationResult)