kitos/Aegis
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: D3FEND expandable cards, System page cleanup, and multi-source improvements

Browse Source 
- Make D3FEND defense cards clickable with expandable details and external link
- Fix D3FEND URLs to use PascalCase technique names matching the ontology
- Remove duplicate Import Atomic Red Team from System page (use Data Sources)
- Add bulk Activate All / Deactivate All buttons with confirmation modal
- Fix template admin list to show both active and inactive templates
- Add PATCH /test-templates/bulk-activate backend endpoint
- Auto-seed data sources on container startup via entrypoint.sh
- Fix SigmaHQ, CALDERA, GTFOBins import issues
- Register D3FEND sync handler in data sources router
- Add CIS Controls v8 compliance framework import
- Expand Test Catalog source filters (CALDERA, LOLBAS, GTFOBins)
- Campaign Generate from Threat Actor now opens actor selector modal
- Add coverage snapshot creation button to Comparison page
- Update README with accurate data source and feature documentation
This commit is contained in:
Kitos
2026-02-10 13:22:23 +01:00
parent 8032b67fab
commit c2e9c687f4
19 changed files with 778 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
backend/app/routers/compliance.py
View File

@@ -24,7 +24,10 @@ from app.models.technique import Technique
from app.models.test_template import TestTemplate
from app.models.threat_actor import ThreatActorTechnique
from app.services.scoring_service import calculate_technique_score
from app.services.compliance_import_service import import_nist_800_53_mappings
from app.services.compliance_import_service import (
import_nist_800_53_mappings,
import_cis_controls_v8_mappings,
)
router = APIRouter(prefix="/compliance", tags=["compliance"])
@@ -378,3 +381,13 @@ def import_nist(
"""Import NIST 800-53 Rev 5 mappings (admin only)."""
result = import_nist_800_53_mappings(db)
return result
@router.post("/import/cis-controls-v8")
def import_cis(
db: Session = Depends(get_db),
current_user: User = Depends(require_role("admin")),
):
"""Import CIS Controls v8 mappings (admin only)."""
result = import_cis_controls_v8_mappings(db)
return result

2
backend/app/routers/data_sources.py
View File

@@ -40,7 +40,7 @@ def _get_sync_handler(source_name: str):
"caldera": ("app.services.caldera_import_service", "sync"),
"elastic_rules": ("app.services.elastic_import_service", "sync"),
"mitre_cti": ("app.services.threat_actor_import_service", "sync"),
# d3fend added in later phases
"d3fend": ("app.services.d3fend_import_service", "sync"),
}
if source_name not in handlers:

40
backend/app/routers/test_templates.py
View File

@@ -55,13 +55,16 @@ def list_templates(
severity: Optional[str] = Query(None, description="Filter by severity (low, medium, high, critical)"),
mitre_technique_id: Optional[str] = Query(None, description="Filter by MITRE technique ID"),
search: Optional[str] = Query(None, description="Search in name and description"),
is_active: Optional[bool] = Query(None, description="Filter by active status (true/false). Omit to return all."),
offset: int = Query(0, ge=0),
limit: int = Query(50, ge=1, le=200),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Return a paginated, filterable list of test templates."""
query = db.query(TestTemplate).filter(TestTemplate.is_active == True) # noqa: E712
query = db.query(TestTemplate)
if is_active is not None:
query = query.filter(TestTemplate.is_active == is_active) # noqa: E712
if source:
query = query.filter(TestTemplate.source == source)
@@ -137,6 +140,41 @@ def template_stats(
}
# ---------------------------------------------------------------------------
# PATCH /test-templates/bulk-activate — activate/deactivate all (admin)
# ---------------------------------------------------------------------------
@router.patch("/bulk-activate")
def bulk_activate_templates(
activate: bool = Query(True, description="True to activate all, False to deactivate all"),
db: Session = Depends(get_db),
current_user: User = Depends(require_role("admin")),
):
"""Set all templates to active or inactive. Admin only."""
count = (
db.query(TestTemplate)
.filter(TestTemplate.is_active != activate)
.update({TestTemplate.is_active: activate})
)
db.commit()
log_action(
db,
user_id=current_user.id,
action="bulk_activate_templates" if activate else "bulk_deactivate_templates",
entity_type="test_template",
entity_id=None,
details={"affected": count, "is_active": activate},
)
return {
"detail": f"{'Activated' if activate else 'Deactivated'} {count} templates",
"affected": count,
"is_active": activate,
}
# ---------------------------------------------------------------------------
# GET /test-templates/by-technique/{mitre_id}
# ---------------------------------------------------------------------------

13
backend/app/seed_data_sources.py
View File

@@ -41,8 +41,8 @@ INITIAL_SOURCES = [
"3 000+ rules with MITRE ATT&CK mappings.",
"sync_frequency": "weekly",
"config": {
"zip_url": "https://github.com/SigmaHQ/sigma/archive/refs/heads/main.zip",
"root_prefix": "sigma-main",
"zip_url": "https://github.com/SigmaHQ/sigma/archive/refs/heads/master.zip",
"root_prefix": "sigma-master",
"rules_dir": "rules",
},
},
@@ -78,13 +78,14 @@ INITIAL_SOURCES = [
"name": "caldera",
"display_name": "MITRE CALDERA",
"type": "attack_procedure",
"url": "https://github.com/mitre/caldera",
"url": "https://github.com/mitre/stockpile",
"description": "Automated adversary emulation platform by MITRE. "
"400+ abilities (executable actions) mapped to ATT&CK.",
"400+ abilities (executable actions) mapped to ATT&CK "
"(via the Stockpile plugin).",
"sync_frequency": "monthly",
"config": {
"zip_url": "https://github.com/mitre/caldera/archive/refs/heads/master.zip",
"root_prefix": "caldera-master",
"zip_url": "https://github.com/mitre/stockpile/archive/refs/heads/master.zip",
"root_prefix": "stockpile-master",
"abilities_dir": "data/abilities",
},
},

17
backend/app/services/caldera_import_service.py
View File

@@ -44,12 +44,12 @@ logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
CALDERA_ZIP_URL = (
"https://github.com/mitre/caldera"
"https://github.com/mitre/stockpile"
"/archive/refs/heads/master.zip"
)
_DOWNLOAD_TIMEOUT = 300
_ZIP_ROOT_PREFIX = "caldera-master"
_ZIP_ROOT_PREFIX = "stockpile-master"
# ---------------------------------------------------------------------------
@@ -144,10 +144,17 @@ def _parse_abilities(abilities_dir: Path) -> list[dict]:
logger.debug("Failed to parse %s: %s", yaml_path, exc)
continue
# Stockpile YAML files may contain YAML lists of abilities
# (e.g. [- id: ..., - id: ...]) or single-document dicts.
# Flatten everything into individual ability dicts.
abilities: list[dict] = []
for data in data_list:
if not isinstance(data, dict):
continue
if isinstance(data, dict):
abilities.append(data)
elif isinstance(data, list):
abilities.extend(d for d in data if isinstance(d, dict))
for data in abilities:
ability_id = data.get("id", "")
if not ability_id:
continue
@@ -193,7 +200,7 @@ def _parse_abilities(abilities_dir: Path) -> list[dict]:
"tool_suggested": executor_str,
"attack_procedure": commands[:4000] if commands else None,
"atomic_test_id": f"caldera:{ability_id}",
"source_url": f"https://github.com/mitre/caldera/tree/master/data/abilities/{tactic}",
"source_url": f"https://github.com/mitre/stockpile/tree/master/data/abilities/{tactic}",
})
logger.info("Parsed %d CALDERA abilities total", len(results))

153
backend/app/services/compliance_import_service.py
View File

@@ -329,6 +329,159 @@ def _import_sample_nist_mappings(db: Session, framework: ComplianceFramework) ->
}
def import_cis_controls_v8_mappings(db: Session) -> dict:
"""Import CIS Controls v8 with ATT&CK technique mappings.
Uses a curated set of CIS Controls mapped to MITRE ATT&CK techniques
based on the CIS Controls Navigator and official documentation.
Returns a summary dict with counts.
"""
# ── 1. Create or get framework ────────────────────────────────
framework = (
db.query(ComplianceFramework)
.filter(ComplianceFramework.name == "CIS Controls v8")
.first()
)
if not framework:
framework = ComplianceFramework(
name="CIS Controls v8",
version="8",
description="Center for Internet Security Critical Security Controls Version 8 — "
"a prioritized set of 18 security safeguards organized by Implementation Groups (IG1, IG2, IG3).",
url="https://www.cisecurity.org/controls/v8",
is_active=True,
)
db.add(framework)
db.flush()
logger.info("Created CIS Controls v8 framework")
else:
logger.info("CIS Controls v8 framework already exists")
# ── 2. Control definitions with ATT&CK mappings ───────────────
CIS_CONTROLS = [
{"control_id": "CIS-1", "title": "Inventory and Control of Enterprise Assets",
"category": "IG1 — Basic",
"techniques": ["T1595", "T1590", "T1018", "T1082"]},
{"control_id": "CIS-2", "title": "Inventory and Control of Software Assets",
"category": "IG1 — Basic",
"techniques": ["T1518", "T1072", "T1195"]},
{"control_id": "CIS-3", "title": "Data Protection",
"category": "IG1 — Basic",
"techniques": ["T1005", "T1114", "T1560", "T1048", "T1041"]},
{"control_id": "CIS-4", "title": "Secure Configuration of Enterprise Assets and Software",
"category": "IG1 — Basic",
"techniques": ["T1574", "T1546", "T1112", "T1543"]},
{"control_id": "CIS-5", "title": "Account Management",
"category": "IG1 — Basic",
"techniques": ["T1078", "T1136", "T1098", "T1087"]},
{"control_id": "CIS-6", "title": "Access Control Management",
"category": "IG1 — Basic",
"techniques": ["T1078", "T1548", "T1134", "T1021"]},
{"control_id": "CIS-7", "title": "Continuous Vulnerability Management",
"category": "IG2 — Foundational",
"techniques": ["T1190", "T1203", "T1068", "T1210"]},
{"control_id": "CIS-8", "title": "Audit Log Management",
"category": "IG2 — Foundational",
"techniques": ["T1562", "T1070", "T1059"]},
{"control_id": "CIS-9", "title": "Email and Web Browser Protections",
"category": "IG2 — Foundational",
"techniques": ["T1566", "T1204", "T1189", "T1598"]},
{"control_id": "CIS-10", "title": "Malware Defenses",
"category": "IG2 — Foundational",
"techniques": ["T1059", "T1204", "T1027", "T1140", "T1497"]},
{"control_id": "CIS-11", "title": "Data Recovery",
"category": "IG1 — Basic",
"techniques": ["T1486", "T1490", "T1561"]},
{"control_id": "CIS-12", "title": "Network Infrastructure Management",
"category": "IG2 — Foundational",
"techniques": ["T1557", "T1071", "T1572", "T1571"]},
{"control_id": "CIS-13", "title": "Network Monitoring and Defense",
"category": "IG2 — Foundational",
"techniques": ["T1071", "T1048", "T1041", "T1105", "T1572"]},
{"control_id": "CIS-14", "title": "Security Awareness and Skills Training",
"category": "IG1 — Basic",
"techniques": ["T1566", "T1204", "T1598"]},
{"control_id": "CIS-15", "title": "Service Provider Management",
"category": "IG2 — Foundational",
"techniques": ["T1199", "T1195"]},
{"control_id": "CIS-16", "title": "Application Software Security",
"category": "IG2 — Foundational",
"techniques": ["T1190", "T1059", "T1203"]},
{"control_id": "CIS-17", "title": "Incident Response Management",
"category": "IG2 — Foundational",
"techniques": ["T1059", "T1547", "T1053"]},
{"control_id": "CIS-18", "title": "Penetration Testing",
"category": "IG3 — Organizational",
"techniques": ["T1595", "T1046", "T1190", "T1059"]},
]
# Build technique lookup
all_techniques = {t.mitre_id: t for t in db.query(Technique).all()}
existing_controls = {
c.control_id: c
for c in db.query(ComplianceControl)
.filter(ComplianceControl.framework_id == framework.id)
.all()
}
existing_mappings = set()
for m in (
db.query(ComplianceControlMapping)
.join(ComplianceControl)
.filter(ComplianceControl.framework_id == framework.id)
.all()
):
existing_mappings.add((str(m.compliance_control_id), str(m.technique_id)))
controls_created = 0
mappings_created = 0
for item in CIS_CONTROLS:
if item["control_id"] in existing_controls:
control = existing_controls[item["control_id"]]
else:
control = ComplianceControl(
framework_id=framework.id,
control_id=item["control_id"],
title=item["title"],
category=item["category"],
)
db.add(control)
db.flush()
existing_controls[item["control_id"]] = control
controls_created += 1
for mitre_id in item["techniques"]:
technique = all_techniques.get(mitre_id)
if not technique:
continue
key = (str(control.id), str(technique.id))
if key in existing_mappings:
continue
mapping = ComplianceControlMapping(
compliance_control_id=control.id,
technique_id=technique.id,
)
db.add(mapping)
existing_mappings.add(key)
mappings_created += 1
db.commit()
summary = {
"framework": framework.name,
"controls_created": controls_created,
"controls_existing": len(existing_controls) - controls_created,
"mappings_created": mappings_created,
"total_controls": len(existing_controls),
}
logger.info(f"CIS Controls v8 import complete: {summary}")
return summary
def _get_nist_category(family_code: str) -> str:
"""Map NIST 800-53 family code to category name."""
categories = {

41
backend/app/services/d3fend_import_service.py
View File

@@ -20,7 +20,7 @@ logger = logging.getLogger(__name__)
D3FEND_ALL_URL = "https://d3fend.mitre.org/api/technique/api-all.json"
D3FEND_MAPPING_URL = "https://d3fend.mitre.org/api/offensive-technique/{attack_id}.json"
D3FEND_BASE_URL = "https://d3fend.mitre.org/technique/d3f:{d3fend_id}"
D3FEND_BASE_URL = "https://d3fend.mitre.org/technique/d3f:{technique_name}"
# ── Tactic extraction helpers ────────────────────────────────────────
@@ -139,7 +139,8 @@ def import_d3fend_techniques(db: Session) -> dict[str, int]:
.filter(DefensiveTechnique.d3fend_id == tech_data["d3fend_id"])
.first()
)
d3fend_url = D3FEND_BASE_URL.format(d3fend_id=tech_data["d3fend_id"])
technique_name = tech_data["name"].replace(" ", "")
d3fend_url = D3FEND_BASE_URL.format(technique_name=technique_name)
if existing:
existing.name = tech_data["name"]
@@ -416,7 +417,8 @@ def _import_d3fend_fallback(db: Session) -> dict[str, int]:
.filter(DefensiveTechnique.d3fend_id == d3fend_id)
.first()
)
d3fend_url = D3FEND_BASE_URL.format(d3fend_id=d3fend_id)
technique_name = tech_data["name"].replace(" ", "")
d3fend_url = D3FEND_BASE_URL.format(technique_name=technique_name)
if existing:
existing.name = tech_data["name"]
@@ -605,6 +607,39 @@ def import_d3fend_mappings(db: Session) -> dict[str, int]:
return {"created": created, "skipped": skipped, "total": total}
def sync(db: Session) -> dict:
"""Sync D3FEND techniques and ATT&CK mappings.
Called by the Data Sources router when the user clicks Sync for D3FEND.
Returns a flat summary dict suitable for ``last_sync_stats``.
"""
from app.models.data_source import DataSource
from datetime import datetime
tech_result = import_d3fend_techniques(db)
mapping_result = import_d3fend_mappings(db)
summary = {
"techniques_created": tech_result.get("created", 0),
"techniques_updated": tech_result.get("updated", 0),
"techniques_total": tech_result.get("total", 0),
"mappings_created": mapping_result.get("created", 0),
"mappings_skipped": mapping_result.get("skipped", 0),
"mappings_total": mapping_result.get("total", 0),
}
# Update DataSource record
ds = db.query(DataSource).filter(DataSource.name == "d3fend").first()
if ds:
ds.last_sync_at = datetime.utcnow()
ds.last_sync_status = "success"
ds.last_sync_stats = summary
db.commit()
logger.info("D3FEND sync complete — %s", summary)
return summary
def get_defenses_for_technique(db: Session, technique_id) -> list[dict]:
"""Get all D3FEND defensive techniques mapped to a given ATT&CK technique."""
mappings = (

17
backend/app/services/lolbas_import_service.py
View File

@@ -68,6 +68,8 @@ _GTFOBINS_FUNCTION_MAP: dict[str, str] = {
"non-interactive-bind-shell": "T1059",
"file-upload": "T1105",
"file-download": "T1105",
"upload": "T1105",
"download": "T1105",
"file-write": "T1105",
"file-read": "T1005",
"library-load": "T1129",
@@ -201,8 +203,11 @@ def _parse_gtfobins(root_dir: Path) -> list[dict]:
logger.warning("GTFOBins directory not found at %s", gtfobins_root)
return results
md_files = sorted(gtfobins_root.glob("*.md"))
logger.info("GTFOBins: Found %d markdown files", len(md_files))
md_files = sorted(
f for f in gtfobins_root.iterdir()
if f.is_file() and f.suffix in (".md", "")
)
logger.info("GTFOBins: Found %d files", len(md_files))
for md_path in md_files:
binary_name = md_path.stem # e.g. "awk"
@@ -259,8 +264,12 @@ def _parse_gtfobins(root_dir: Path) -> list[dict]:
def _extract_front_matter(content: str) -> dict | None:
"""Extract YAML front-matter from a markdown file."""
match = re.match(r"^---\s*\n(.*?)\n---", content, re.DOTALL)
"""Extract YAML front-matter from a markdown/GTFOBins file.
Supports both ``---/---`` (standard front-matter) and ``---/...``
(YAML document-end marker used by GTFOBins).
"""
match = re.match(r"^---\s*\n(.*?)\n(?:---|\.\.\.)", content, re.DOTALL)
if not match:
return None
try:

6
backend/app/services/sigma_import_service.py
View File

@@ -46,11 +46,11 @@ logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
SIGMA_ZIP_URL = (
"https://github.com/SigmaHQ/sigma/archive/refs/heads/main.zip"
"https://github.com/SigmaHQ/sigma/archive/refs/heads/master.zip"
)
_DOWNLOAD_TIMEOUT = 300
_ZIP_ROOT_PREFIX = "sigma-main"
_ZIP_ROOT_PREFIX = "sigma-master"
# Regex to extract MITRE ATT&CK technique IDs from Sigma tags
# e.g. "attack.t1059.001" → "T1059.001"
@@ -170,7 +170,7 @@ def _parse_sigma_rules(rules_dir: Path) -> list[dict]:
# Create one entry per technique
for tech_id in technique_ids:
source_url = (
f"https://github.com/SigmaHQ/sigma/blob/main/"
f"https://github.com/SigmaHQ/sigma/blob/master/"
f"{relative_path.replace(chr(92), '/')}"
)
results.append({

3
backend/entrypoint.sh
View File

@@ -7,5 +7,8 @@ alembic upgrade head
echo "=== Seeding admin user ==="
python -m app.seed
echo "=== Seeding data sources ==="
python -m app.seed_data_sources
echo "=== Starting uvicorn ==="
exec uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload