Compare commits

...

2 Commits

Author SHA1 Message Date
kitos 4e31359fca feat(frontend): add technique/platform/outcome/date filters to Validated Tests
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
The "200" badge on the Validated Tests page was a hardcoded query limit,
not the real total — silently hiding results beyond the 200th once there
were more validated tests than that. Now shows the true total via the new
/tests/count endpoint, and warns when the fetched page doesn't cover all
matches.

Adds filter controls beyond the existing name/technique text search:
technique (MITRE ID or name), platform, attack outcome, detection
outcome, and a validated-date range — all pushed down to the server via
the new GET /tests filters instead of only ever searching the current
page client-side.
2026-07-09 12:45:54 +02:00
kitos 226869d308 feat(tests): richer server-side filters + true total count for GET /tests
GET /tests caps limit at 200, so any page relying on len(results) for a
"total" count (e.g. Validated Tests) silently under-reports once there
are more matches than the page size — the 200 shown was a page cap, not
the real total.

- Adds GET /tests/count, sharing its query-building with list_tests()
  via a new _build_test_query() so the two can't drift apart.
- Adds technique_search (free-text on MITRE ID or name), attack_success,
  detection_result, and validated_from/validated_to filters, so callers
  aren't limited to state/technique_id/platform/creator.
2026-07-09 12:45:05 +02:00
5 changed files with 431 additions and 57 deletions
+57 -21
View File
@@ -118,6 +118,11 @@ from app.services.test_crud_service import (
list_tests as crud_list_tests,
)
# Import from app.services.test_crud_service
from app.services.test_crud_service import (
count_tests as crud_count_tests,
)
# Import from app.services.test_crud_service
from app.services.test_crud_service import (
update_test as crud_update_test,
@@ -203,13 +208,14 @@ def _mask_for_team_blindness(test_out: TestOut, *, viewer_role: str) -> TestOut:
# ---------------------------------------------------------------------------
@router.get("", response_model=list[TestOut])
# Define function list_tests
def list_tests(
def _test_list_filter_params(
# Entry: state
state: Optional[str] = Query(None, description="Filter by test state"),
# Entry: technique_id
technique_id: Optional[uuid.UUID] = Query(None, description="Filter by technique"),
technique_search: Optional[str] = Query(
None, description="Free-text filter on technique MITRE ID or name"
),
# Entry: platform
platform: Optional[str] = Query(None, description="Filter by platform"),
# Entry: created_by
@@ -225,6 +231,52 @@ def list_tests(
not_in_any_campaign: bool = Query(
False, description="Only return tests not linked to any campaign"
),
attack_success: Optional[str] = Query(None, description="Filter by attack success outcome"),
detection_result: Optional[str] = Query(None, description="Filter by detection outcome"),
validated_from: Optional[datetime] = Query(
None, description="Only tests validated on/after this date"
),
validated_to: Optional[datetime] = Query(
None, description="Only tests validated on/before this date"
),
) -> dict:
"""Shared filter params for GET /tests and GET /tests/count, so the two
endpoints can never silently drift out of sync with each other."""
return {
"state": state,
"technique_id": technique_id,
"technique_search": technique_search,
"platform": platform,
"created_by": created_by,
"pending_validation_side": pending_validation_side,
"reviewer_id": reviewer_id,
"not_in_any_campaign": not_in_any_campaign,
"attack_success": attack_success,
"detection_result": detection_result,
"validated_from": validated_from,
"validated_to": validated_to,
}
@router.get("/count")
def count_tests(
filters: dict = Depends(_test_list_filter_params),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
) -> dict:
"""Return the true total of tests matching the given filters.
GET /tests caps `limit` at 200, so pages that show "N results" need
this to report the real total rather than len(page) — which silently
under-reports whenever there are more matches than the page size.
"""
return {"total": crud_count_tests(db, **filters)}
@router.get("", response_model=list[TestOut])
# Define function list_tests
def list_tests(
filters: dict = Depends(_test_list_filter_params),
offset: int = Query(0, ge=0),
# Entry: limit
limit: int = Query(50, ge=1, le=200),
@@ -236,12 +288,7 @@ def list_tests(
"""Return a paginated list of tests, optionally filtered by state, technique, platform or creator.
Args:
state (Optional[str]): Filter by test state (e.g. ``draft``, ``validated``).
technique_id (Optional[uuid.UUID]): Filter tests belonging to a specific technique.
platform (Optional[str]): Filter by target platform (e.g. ``windows``, ``linux``).
created_by (Optional[uuid.UUID]): Filter by the UUID of the creator.
pending_validation_side (Optional[str]): Filter ``in_review`` tests pending validation
on ``'red'`` or ``'blue'`` side.
filters (dict): Shared filter params — see ``_test_list_filter_params``.
offset (int): Number of records to skip for pagination.
limit (int): Maximum number of records to return.
db (Session): SQLAlchemy database session.
@@ -253,18 +300,7 @@ def list_tests(
# Return crud_list_tests(
return crud_list_tests(
db,
# Keyword argument: state
state=state,
# Keyword argument: technique_id
technique_id=technique_id,
# Keyword argument: platform
platform=platform,
# Keyword argument: created_by
created_by=created_by,
# Keyword argument: pending_validation_side
pending_validation_side=pending_validation_side,
reviewer_id=reviewer_id,
not_in_any_campaign=not_in_any_campaign,
**filters,
offset=offset,
# Keyword argument: limit
limit=limit,
+117 -17
View File
@@ -10,8 +10,8 @@ from datetime import datetime
from typing import Any
# Import Session, joinedload from sqlalchemy.orm
from sqlalchemy import or_
from sqlalchemy.orm import Session, joinedload
from sqlalchemy import func, or_
from sqlalchemy.orm import Query, Session, joinedload
# Import from app.domain.errors
from app.domain.errors import (
@@ -49,29 +49,27 @@ def determine_initial_classification(technique: Technique | None) -> str:
return DataClassification.confidential.value
# Define function list_tests
def list_tests(
# Entry: db
def _build_test_query(
db: Session,
*,
# Entry: state
state: str | None = None,
# Entry: technique_id
technique_id: uuid.UUID | None = None,
# Entry: platform
technique_search: str | None = None,
platform: str | None = None,
# Entry: created_by
created_by: uuid.UUID | None = None,
# Entry: pending_validation_side
pending_validation_side: str | None = None,
# Entry: reviewer_id — "my reviews" queue for the red_review/blue_review gate stage
reviewer_id: uuid.UUID | None = None,
not_in_any_campaign: bool = False,
offset: int = 0,
# Entry: limit
limit: int = 50,
) -> list[Test]:
"""Return a paginated list of tests with optional filters."""
attack_success: str | None = None,
detection_result: str | None = None,
validated_from: datetime | None = None,
validated_to: datetime | None = None,
) -> Query:
"""Build the filtered Test query shared by list_tests() and count_tests().
Kept as a single source of truth so the displayed page of results and
the total count it's paginated against can never drift apart.
"""
query = db.query(Test).options(joinedload(Test.technique))
# Check: state
@@ -82,6 +80,16 @@ def list_tests(
if technique_id:
# Assign query = query.filter(Test.technique_id == technique_id)
query = query.filter(Test.technique_id == technique_id)
# Free-text technique filter (MITRE ID or name) — doesn't require the
# caller to already know the technique's UUID like technique_id does.
if technique_search:
pattern = f"%{escape_like(technique_search)}%"
query = query.join(Technique, Test.technique_id == Technique.id).filter(
or_(
Technique.mitre_id.ilike(pattern),
Technique.name.ilike(pattern),
)
)
# Check: platform
if platform:
# Assign query = query.filter(Test.platform.ilike(f"%{escape_like(platform)}%"))
@@ -125,10 +133,102 @@ def list_tests(
linked = db.query(CampaignTest.test_id).distinct().subquery()
query = query.filter(~Test.id.in_(linked))
# Return query.order_by(Test.created_at.desc()).offset(offset).limit(limit)....
if attack_success:
query = query.filter(Test.attack_success == attack_success)
if detection_result:
query = query.filter(Test.detection_result == detection_result)
# Validated-date range: whichever lead validated last (mirrors the
# "Validated" column shown on the Validated Tests page).
if validated_from or validated_to:
validated_at = func.coalesce(Test.blue_validated_at, Test.red_validated_at)
if validated_from:
query = query.filter(validated_at >= validated_from)
if validated_to:
query = query.filter(validated_at <= validated_to)
return query
# Define function list_tests
def list_tests(
db: Session,
*,
state: str | None = None,
technique_id: uuid.UUID | None = None,
technique_search: str | None = None,
platform: str | None = None,
created_by: uuid.UUID | None = None,
pending_validation_side: str | None = None,
# Entry: reviewer_id — "my reviews" queue for the red_review/blue_review gate stage
reviewer_id: uuid.UUID | None = None,
not_in_any_campaign: bool = False,
attack_success: str | None = None,
detection_result: str | None = None,
validated_from: datetime | None = None,
validated_to: datetime | None = None,
offset: int = 0,
# Entry: limit
limit: int = 50,
) -> list[Test]:
"""Return a paginated list of tests with optional filters."""
query = _build_test_query(
db,
state=state,
technique_id=technique_id,
technique_search=technique_search,
platform=platform,
created_by=created_by,
pending_validation_side=pending_validation_side,
reviewer_id=reviewer_id,
not_in_any_campaign=not_in_any_campaign,
attack_success=attack_success,
detection_result=detection_result,
validated_from=validated_from,
validated_to=validated_to,
)
return query.order_by(Test.created_at.desc()).offset(offset).limit(limit).all()
def count_tests(
db: Session,
*,
state: str | None = None,
technique_id: uuid.UUID | None = None,
technique_search: str | None = None,
platform: str | None = None,
created_by: uuid.UUID | None = None,
pending_validation_side: str | None = None,
reviewer_id: uuid.UUID | None = None,
not_in_any_campaign: bool = False,
attack_success: str | None = None,
detection_result: str | None = None,
validated_from: datetime | None = None,
validated_to: datetime | None = None,
) -> int:
"""Return the total count of tests matching the same filters as list_tests().
Lets the frontend show a true total even when the page size caps the
number of rows actually returned (e.g. GET /tests's limit<=200).
"""
query = _build_test_query(
db,
state=state,
technique_id=technique_id,
technique_search=technique_search,
platform=platform,
created_by=created_by,
pending_validation_side=pending_validation_side,
reviewer_id=reviewer_id,
not_in_any_campaign=not_in_any_campaign,
attack_success=attack_success,
detection_result=detection_result,
validated_from=validated_from,
validated_to=validated_to,
)
return query.count()
# Define function create_test
def create_test(
# Entry: db
+85
View File
@@ -0,0 +1,85 @@
"""Tests for the expanded GET /tests filters and the new GET /tests/count.
Block: Validated Tests page couldn't show a true total once there were
more than `limit` (max 200) matching tests, and had no server-side way to
filter by technique/platform/date/outcome beyond a client-side name search.
count_tests() shares its query-building with list_tests() (_build_test_query)
so the two can't silently drift out of sync.
"""
from datetime import datetime, timedelta
from app.models.enums import AttackSuccessResult, TestResult, TestState
from app.models.technique import Technique
from app.models.test import Test
from app.services.test_crud_service import count_tests, list_tests
def _seed(db, owner_id, *, mitre_id, name, technique_name=None, state=TestState.validated, **overrides):
tech = Technique(mitre_id=mitre_id, name=technique_name or f"Technique {mitre_id}", tactic="execution", platforms=["windows"])
db.add(tech)
db.flush()
test = Test(technique_id=tech.id, name=name, state=state, created_by=owner_id, **overrides)
db.add(test)
db.commit()
db.refresh(test)
return test
def test_count_tests_matches_list_tests_total_beyond_page_size(db, red_lead_user):
for i in range(5):
_seed(db, red_lead_user.id, mitre_id=f"T900{i}", name=f"Validated {i}")
total = count_tests(db, state="validated")
page = list_tests(db, state="validated", limit=2)
assert total == 5
assert len(page) == 2
def test_technique_search_matches_mitre_id_or_name(db, red_lead_user):
_seed(db, red_lead_user.id, mitre_id="T1059.001", name="PowerShell test", technique_name="Command Line Interface")
_seed(db, red_lead_user.id, mitre_id="T1055", name="Process Injection test", technique_name="Process Injection")
by_id = list_tests(db, technique_search="T1059")
by_name = list_tests(db, technique_search="injection")
assert [t.name for t in by_id] == ["PowerShell test"]
assert [t.name for t in by_name] == ["Process Injection test"]
def test_attack_success_and_detection_result_filters(db, red_lead_user):
_seed(
db, red_lead_user.id, mitre_id="T9010", name="Successful and detected",
attack_success=AttackSuccessResult.successful, detection_result=TestResult.detected,
)
_seed(
db, red_lead_user.id, mitre_id="T9011", name="Successful and not detected",
attack_success=AttackSuccessResult.successful, detection_result=TestResult.not_detected,
)
results = list_tests(db, attack_success="successful", detection_result="detected")
assert [t.name for t in results] == ["Successful and detected"]
def test_validated_date_range_filter(db, red_lead_user):
old = _seed(db, red_lead_user.id, mitre_id="T9020", name="Old validation")
old.blue_validated_at = datetime.utcnow() - timedelta(days=30)
recent = _seed(db, red_lead_user.id, mitre_id="T9021", name="Recent validation")
recent.blue_validated_at = datetime.utcnow() - timedelta(days=1)
db.commit()
results = list_tests(db, validated_from=datetime.utcnow() - timedelta(days=7))
assert [t.name for t in results] == ["Recent validation"]
def test_count_tests_endpoint_returns_true_total(api, db, red_lead_user, auth_headers):
for i in range(3):
_seed(db, red_lead_user.id, mitre_id=f"T903{i}", name=f"Counted {i}")
resp = api("get", "/api/v1/tests/count?state=validated", auth_headers)
assert resp.status_code == 200
assert resp.json()["total"] == 3
+30 -4
View File
@@ -67,12 +67,19 @@ export interface TestValidatePayload {
export interface TestListFilters {
state?: TestState;
technique_id?: string;
/** Free-text filter on technique MITRE ID or name — doesn't require knowing the technique's UUID. */
technique_search?: string;
platform?: string;
created_by?: string;
pending_validation_side?: "red" | "blue";
/** "My reviews" queue — tests assigned to this reviewer at the red_review/blue_review gate. */
reviewer_id?: string;
not_in_any_campaign?: boolean;
attack_success?: AttackSuccessResult;
detection_result?: TestResult;
/** ISO date/datetime — filters on whichever lead validated last. */
validated_from?: string;
validated_to?: string;
assigned_to_me?: boolean;
unassigned_red?: boolean;
unassigned_blue?: boolean;
@@ -80,18 +87,28 @@ export interface TestListFilters {
limit?: number;
}
// ── CRUD ───────────────────────────────────────────────────────────
/** List tests with optional filters. */
export async function getTests(filters?: TestListFilters): Promise<Test[]> {
function buildTestListParams(filters?: TestListFilters): URLSearchParams {
const params = new URLSearchParams();
if (filters?.state) params.append("state", filters.state);
if (filters?.technique_id) params.append("technique_id", filters.technique_id);
if (filters?.technique_search) params.append("technique_search", filters.technique_search);
if (filters?.platform) params.append("platform", filters.platform);
if (filters?.created_by) params.append("created_by", filters.created_by);
if (filters?.pending_validation_side) params.append("pending_validation_side", filters.pending_validation_side);
if (filters?.reviewer_id) params.append("reviewer_id", filters.reviewer_id);
if (filters?.not_in_any_campaign) params.append("not_in_any_campaign", "true");
if (filters?.attack_success) params.append("attack_success", filters.attack_success);
if (filters?.detection_result) params.append("detection_result", filters.detection_result);
if (filters?.validated_from) params.append("validated_from", filters.validated_from);
if (filters?.validated_to) params.append("validated_to", filters.validated_to);
return params;
}
// ── CRUD ───────────────────────────────────────────────────────────
/** List tests with optional filters. */
export async function getTests(filters?: TestListFilters): Promise<Test[]> {
const params = buildTestListParams(filters);
if (filters?.offset !== undefined) params.append("offset", String(filters.offset));
if (filters?.limit !== undefined) params.append("limit", String(filters.limit));
@@ -101,6 +118,15 @@ export async function getTests(filters?: TestListFilters): Promise<Test[]> {
return data;
}
/** Return the true total of tests matching the given filters (unaffected by pagination limits). */
export async function getTestsCount(filters?: TestListFilters): Promise<number> {
const params = buildTestListParams(filters);
const { data } = await client.get<{ total: number }>(
`/tests/count${params.toString() ? `?${params}` : ""}`,
);
return data.total;
}
/** Create a new test. */
export async function createTest(payload: TestCreatePayload): Promise<Test> {
const { data } = await client.post<Test>("/tests", payload);
+142 -15
View File
@@ -11,8 +11,10 @@ import {
ChevronsUpDown,
ShieldCheck,
Swords,
Filter,
X,
} from "lucide-react";
import { getTests } from "../api/tests";
import { getTests, getTestsCount, type TestListFilters } from "../api/tests";
import type { Test, TestResult, AttackSuccessResult } from "../types/models";
/* ── Result helpers ─────────────────────────────────────────────────── */
@@ -69,6 +71,43 @@ export default function ValidatedTestsPage() {
const [searchText, setSearchText] = useState("");
// Server-side filters — beyond the client-side name/technique search above,
// which only searches within the current page of fetched results.
const [techniqueSearch, setTechniqueSearch] = useState("");
const [platformFilter, setPlatformFilter] = useState("");
const [attackSuccessFilter, setAttackSuccessFilter] = useState<AttackSuccessResult | "">("");
const [detectionResultFilter, setDetectionResultFilter] = useState<TestResult | "">("");
const [validatedFrom, setValidatedFrom] = useState("");
const [validatedTo, setValidatedTo] = useState("");
const hasActiveFilters = Boolean(
techniqueSearch || platformFilter || attackSuccessFilter || detectionResultFilter || validatedFrom || validatedTo
);
const clearFilters = () => {
setTechniqueSearch("");
setPlatformFilter("");
setAttackSuccessFilter("");
setDetectionResultFilter("");
setValidatedFrom("");
setValidatedTo("");
};
const serverFilters: TestListFilters = useMemo(
() => ({
state: "validated",
...(techniqueSearch ? { technique_search: techniqueSearch } : {}),
...(platformFilter ? { platform: platformFilter } : {}),
...(attackSuccessFilter ? { attack_success: attackSuccessFilter } : {}),
...(detectionResultFilter ? { detection_result: detectionResultFilter } : {}),
...(validatedFrom ? { validated_from: validatedFrom } : {}),
// Date-only input parses to midnight — extend to end-of-day so the
// "to" bound includes tests validated any time on that date.
...(validatedTo ? { validated_to: `${validatedTo}T23:59:59` } : {}),
}),
[techniqueSearch, platformFilter, attackSuccessFilter, detectionResultFilter, validatedFrom, validatedTo]
);
type SortKey =
| "name"
| "technique"
@@ -91,8 +130,16 @@ export default function ValidatedTestsPage() {
};
const { data: validatedTests, isLoading, error } = useQuery({
queryKey: ["tests", "validated"],
queryFn: () => getTests({ state: "validated", limit: 200 }),
queryKey: ["tests", "validated", serverFilters],
queryFn: () => getTests({ ...serverFilters, limit: 200 }),
});
// Independent of the page's `limit` cap — lets the count badge (and the
// "showing X of Y" hint below) reflect the true total even when there
// are more matches than the 200-row page size.
const { data: totalCount } = useQuery({
queryKey: ["tests", "validated", "count", serverFilters],
queryFn: () => getTestsCount(serverFilters),
});
const formatDate = (dateStr: string | null | undefined) => {
@@ -211,20 +258,98 @@ export default function ValidatedTestsPage() {
</div>
</div>
<span className="rounded-full border border-green-500/30 bg-green-500/10 px-3 py-1 text-sm font-medium text-green-400">
{validatedTests?.length ?? 0} validated
{totalCount ?? validatedTests?.length ?? 0} validated
</span>
</div>
{/* Search */}
<div className="relative max-w-sm">
<Search className="absolute left-3 top-1/2 h-4 w-4 -translate-y-1/2 text-gray-500" />
<input
type="text"
value={searchText}
onChange={(e) => setSearchText(e.target.value)}
placeholder="Search by name or technique…"
className="w-full rounded-lg border border-gray-700 bg-gray-900 pl-9 pr-3 py-2 text-sm text-gray-300 placeholder-gray-500 focus:border-green-500 focus:outline-none"
/>
{/* Search + filters */}
<div className="space-y-3">
<div className="flex flex-wrap items-center gap-3">
<div className="relative max-w-sm flex-1 min-w-[200px]">
<Search className="absolute left-3 top-1/2 h-4 w-4 -translate-y-1/2 text-gray-500" />
<input
type="text"
value={searchText}
onChange={(e) => setSearchText(e.target.value)}
placeholder="Search by name or technique…"
className="w-full rounded-lg border border-gray-700 bg-gray-900 pl-9 pr-3 py-2 text-sm text-gray-300 placeholder-gray-500 focus:border-green-500 focus:outline-none"
/>
</div>
<div className="relative w-40">
<Filter className="absolute left-3 top-1/2 h-4 w-4 -translate-y-1/2 text-gray-500" />
<input
type="text"
value={techniqueSearch}
onChange={(e) => setTechniqueSearch(e.target.value)}
placeholder="Technique (T1059…)"
className="w-full rounded-lg border border-gray-700 bg-gray-900 pl-9 pr-3 py-2 text-sm text-gray-300 placeholder-gray-500 focus:border-green-500 focus:outline-none"
/>
</div>
<input
type="text"
value={platformFilter}
onChange={(e) => setPlatformFilter(e.target.value)}
placeholder="Platform…"
className="w-32 rounded-lg border border-gray-700 bg-gray-900 px-3 py-2 text-sm text-gray-300 placeholder-gray-500 focus:border-green-500 focus:outline-none"
/>
<select
value={attackSuccessFilter}
onChange={(e) => setAttackSuccessFilter(e.target.value as AttackSuccessResult | "")}
className="rounded-lg border border-gray-700 bg-gray-900 px-3 py-2 text-sm text-gray-300 focus:border-green-500 focus:outline-none"
>
<option value="">Any attack outcome</option>
<option value="successful">Success</option>
<option value="partially_successful">Partial</option>
<option value="not_successful">Blocked</option>
</select>
<select
value={detectionResultFilter}
onChange={(e) => setDetectionResultFilter(e.target.value as TestResult | "")}
className="rounded-lg border border-gray-700 bg-gray-900 px-3 py-2 text-sm text-gray-300 focus:border-green-500 focus:outline-none"
>
<option value="">Any detection</option>
<option value="detected">Detected</option>
<option value="partially_detected">Partial</option>
<option value="not_detected">Not Detected</option>
</select>
<div className="flex items-center gap-1.5 text-xs text-gray-500">
<span>Validated</span>
<input
type="date"
value={validatedFrom}
onChange={(e) => setValidatedFrom(e.target.value)}
className="rounded-lg border border-gray-700 bg-gray-900 px-2 py-2 text-sm text-gray-300 focus:border-green-500 focus:outline-none"
/>
<span></span>
<input
type="date"
value={validatedTo}
onChange={(e) => setValidatedTo(e.target.value)}
className="rounded-lg border border-gray-700 bg-gray-900 px-2 py-2 text-sm text-gray-300 focus:border-green-500 focus:outline-none"
/>
</div>
{hasActiveFilters && (
<button
onClick={clearFilters}
className="flex items-center gap-1 text-xs text-gray-400 hover:text-white transition-colors"
>
<X className="h-3.5 w-3.5" />
Clear filters
</button>
)}
</div>
{totalCount !== undefined && totalCount > tests.length && (
<p className="text-xs text-amber-400/80">
Showing {tests.length} of {totalCount} matching tests narrow the filters above to see the rest.
</p>
)}
</div>
{/* Table */}
@@ -305,7 +430,9 @@ export default function ValidatedTestsPage() {
{tests.length === 0 && (
<div className="py-16 text-center text-gray-500 text-sm">
{searchText ? "No validated tests match your search." : "No validated tests yet."}
{searchText || hasActiveFilters
? "No validated tests match your search/filters."
: "No validated tests yet."}
</div>
)}
</div>