Kitos
9d7832c571
T-109: Rewrite tests router with full Red/Blue workflow endpoints - list with filters, create from template, Red/Blue team updates with state guards, start-execution, submit-red, submit-blue, validate-red, validate-blue, reopen, and timeline. All using workflow service from Phase 11. T-110: Rewrite evidence router with Red/Blue separation - upload with team field, list with team filter, delete with state-based permissions. Red Team edits in draft/red_executing, Blue Team in blue_evaluating, admin bypasses all. T-111: Create test_templates router with full CRUD - paginated list with source/platform/severity/search filters, by-technique lookup, admin-only create/update, and soft delete. Registered in main.py. T-112: Add POST /system/import-atomic-tests endpoint to system router - admin-only trigger for Atomic Red Team import with error handling and statistics response. Includes validation tests for all four tasks (35 checks total).
117 lines
3.2 KiB
Python
117 lines
3.2 KiB
Python
"""System-level endpoints (admin only).
|
|
|
|
Provides manual triggers for background operations such as the MITRE
|
|
ATT&CK synchronisation, intel scanning, Atomic Red Team import, and
|
|
scheduler health introspection.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.dependencies.auth import require_role
|
|
from app.models.user import User
|
|
from app.services.mitre_sync_service import sync_mitre
|
|
from app.services.intel_service import scan_intel
|
|
from app.services.atomic_import_service import import_atomic_red_team
|
|
from app.jobs.mitre_sync_job import scheduler
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/system", tags=["system"])
|
|
|
|
|
|
@router.post("/sync-mitre")
|
|
def trigger_mitre_sync(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Manually trigger a MITRE ATT&CK synchronisation.
|
|
|
|
**Requires** the ``admin`` role.
|
|
|
|
Returns a JSON object with the sync summary including the count of
|
|
new and updated techniques.
|
|
"""
|
|
summary = sync_mitre(db)
|
|
return {
|
|
"message": "MITRE sync completed",
|
|
"new": summary["created"],
|
|
"updated": summary["updated"],
|
|
}
|
|
|
|
|
|
@router.post("/run-intel-scan")
|
|
def trigger_intel_scan(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Manually trigger a threat-intelligence scan.
|
|
|
|
**Requires** the ``admin`` role.
|
|
|
|
Returns a JSON object with the scan summary including the count of
|
|
new intel items found.
|
|
"""
|
|
summary = scan_intel(db)
|
|
return {
|
|
"message": "Intel scan completed",
|
|
"new_items": summary["new_items"],
|
|
}
|
|
|
|
|
|
@router.post("/import-atomic-tests")
|
|
def trigger_atomic_import(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Trigger an import of Atomic Red Team tests as TestTemplates.
|
|
|
|
**Requires** the ``admin`` role.
|
|
|
|
Downloads the Atomic Red Team repository ZIP from GitHub, parses the
|
|
YAML files, and creates/updates TestTemplate records. Running this
|
|
endpoint multiple times is idempotent — duplicates are skipped.
|
|
|
|
Returns a JSON object with import statistics.
|
|
"""
|
|
try:
|
|
summary = import_atomic_red_team(db)
|
|
except Exception as exc:
|
|
logger.error("Atomic Red Team import failed: %s", exc)
|
|
return {
|
|
"message": "Import failed",
|
|
"error": str(exc),
|
|
}
|
|
|
|
return {
|
|
"message": "Import completed",
|
|
"imported": summary["created"],
|
|
"skipped": summary["skipped_existing"],
|
|
"total_parsed": summary["total_tests_parsed"],
|
|
}
|
|
|
|
|
|
@router.get("/scheduler-status")
|
|
def scheduler_status(
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
"""Return the current state of the background scheduler.
|
|
|
|
**Requires** the ``admin`` role.
|
|
"""
|
|
jobs = scheduler.get_jobs()
|
|
return {
|
|
"running": scheduler.running,
|
|
"jobs": [
|
|
{
|
|
"id": job.id,
|
|
"name": job.name,
|
|
"next_run_time": str(job.next_run_time) if job.next_run_time else None,
|
|
}
|
|
for job in jobs
|
|
],
|
|
}
|