Compare commits

...

2 Commits

Author SHA1 Message Date
kitos c0a88fc489 feat(tests): attack_success 3-value UI, data classification badge, reorder containment options
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
2026-07-06 16:19:50 +02:00
kitos 022c1c756a feat(jira): pause/hold status mapping, RT/BT dates, TTP, attack_success 3-value field, hours 2026-07-06 16:19:39 +02:00
22 changed files with 669 additions and 72 deletions
@@ -0,0 +1,48 @@
"""Convert tests.attack_success from Boolean to a 3-value enum.
Revision ID: b058
Revises: b057
Create Date: 2026-07-06
"""
from alembic import op
import sqlalchemy as sa
revision = "b058"
down_revision = "b057"
branch_labels = None
depends_on = None
_ENUM_NAME = "attacksuccessresult"
def upgrade() -> None:
attack_success_enum = sa.Enum(
"successful", "not_successful", "partially_successful", name=_ENUM_NAME
)
attack_success_enum.create(op.get_bind(), checkfirst=True)
op.add_column("tests", sa.Column("attack_success_new", attack_success_enum, nullable=True))
op.execute(
"UPDATE tests SET attack_success_new = CASE "
"WHEN attack_success = true THEN 'successful' "
"WHEN attack_success = false THEN 'not_successful' "
"ELSE NULL END"
)
op.drop_column("tests", "attack_success")
op.alter_column("tests", "attack_success_new", new_column_name="attack_success")
def downgrade() -> None:
op.add_column("tests", sa.Column("attack_success_bool", sa.Boolean(), nullable=True))
op.execute(
"UPDATE tests SET attack_success_bool = CASE "
"WHEN attack_success = 'successful' THEN true "
"WHEN attack_success = 'not_successful' THEN false "
"ELSE NULL END"
)
op.drop_column("tests", "attack_success")
op.alter_column("tests", "attack_success_bool", new_column_name="attack_success")
sa.Enum(name=_ENUM_NAME).drop(op.get_bind(), checkfirst=True)
+8
View File
@@ -81,6 +81,14 @@ class ContainmentResult(str, enum.Enum):
not_contained = "not_contained"
class AttackSuccessResult(str, enum.Enum):
"""Outcome of a red-team attack from a success perspective."""
successful = "successful"
not_successful = "not_successful"
partially_successful = "partially_successful"
# Define class DataClassification
class DataClassification(str, enum.Enum):
"""Data sensitivity classification levels for compliance and retention policies."""
+1
View File
@@ -7,6 +7,7 @@ working with ``from app.models.enums import ...``.
# Import # noqa: F401 from app.domain.enums
from app.domain.enums import ( # noqa: F401
AttackSuccessResult,
ContainmentResult,
DataClassification,
TeamSide,
+2 -2
View File
@@ -27,7 +27,7 @@ from sqlalchemy.orm import relationship
from app.database import Base
# Import ContainmentResult, TestResult, TestState from app.models.enums
from app.models.enums import ContainmentResult, TestResult, TestState
from app.models.enums import AttackSuccessResult, ContainmentResult, TestResult, TestState
# Define class Test
@@ -69,7 +69,7 @@ class Test(Base):
# ── Red Team fields ─────────────────────────────────────────────
red_summary = Column(Text, nullable=True)
# Assign attack_success = Column(Boolean, nullable=True)
attack_success = Column(Boolean, nullable=True)
attack_success = Column(Enum(AttackSuccessResult, name="attacksuccessresult"), nullable=True)
execution_start_time = Column(DateTime, nullable=True)
execution_end_time = Column(DateTime, nullable=True)
# Assign red_validated_by = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
+2 -2
View File
@@ -49,7 +49,7 @@ from app.domain.unit_of_work import UnitOfWork
# Import limiter from app.limiter
from app.limiter import limiter
from app.models.enums import TestState, TestResult, TeamSide
from app.models.enums import AttackSuccessResult, TestState, TestResult, TeamSide
from app.models.evidence import Evidence
from app.storage import upload_file
from app.models.technique import Technique
@@ -1687,7 +1687,7 @@ class RTEvidenceEntry(BaseModel):
class RTTechniqueEntry(BaseModel):
mitre_id: str
result: str # "detected" | "not_detected" | "partially_detected"
attack_success: bool = True
attack_success: AttackSuccessResult = AttackSuccessResult.successful
platform: Optional[str] = None
notes: Optional[str] = None
evidence: list[RTEvidenceEntry] # REQUIRED — at least one image per technique
+3 -3
View File
@@ -9,7 +9,7 @@ from datetime import datetime
from pydantic import BaseModel, ConfigDict, model_validator
# Import DataClassification from app.domain.enums
from app.domain.enums import ContainmentResult, DataClassification
from app.domain.enums import AttackSuccessResult, ContainmentResult, DataClassification
from app.models.enums import TestResult, TestState
from app.schemas.evidence import EvidenceOut
@@ -79,7 +79,7 @@ class TestRedUpdate(BaseModel):
# Assign tool_used = None
tool_used: str | None = None
# Assign attack_success = None
attack_success: bool | None = None
attack_success: AttackSuccessResult | None = None
# Assign red_summary = None
red_summary: str | None = None
execution_start_time: datetime | None = None
@@ -229,7 +229,7 @@ class TestOut(BaseModel):
# Red Team fields
red_summary: str | None = None
# Assign attack_success = None
attack_success: bool | None = None
attack_success: AttackSuccessResult | None = None
execution_start_time: datetime | None = None
execution_end_time: datetime | None = None
# Assign red_validated_by = None
+8 -3
View File
@@ -35,7 +35,7 @@ from app.database import SessionLocal
from app.models.audit import AuditLog
# Import TeamSide, TechniqueStatus, TestResult, TestState from app.models.enums
from app.models.enums import TeamSide, TechniqueStatus, TestResult, TestState
from app.models.enums import AttackSuccessResult, TeamSide, TechniqueStatus, TestResult, TestState
# Import Evidence from app.models.evidence
from app.models.evidence import Evidence
@@ -354,8 +354,13 @@ def _seed_tests(db: Session, users: list[User], techniques: list[Technique], cou
if state in (TestState.blue_evaluating, TestState.in_review, TestState.validated, TestState.rejected):
# Assign test.red_summary = f"Attack executed successfully using {test.tool_used}."
test.red_summary = f"Attack executed successfully using {test.tool_used}."
# Assign test.attack_success = random.choice([True, True, True, False])
test.attack_success = random.choice([True, True, True, False])
test.attack_success = random.choice([
AttackSuccessResult.successful,
AttackSuccessResult.successful,
AttackSuccessResult.successful,
AttackSuccessResult.partially_successful,
AttackSuccessResult.not_successful,
])
# Check: state in (TestState.in_review, TestState.validated, TestState.rejec...
if state in (TestState.in_review, TestState.validated, TestState.rejected):
+1 -1
View File
@@ -105,7 +105,7 @@ def get_tests_analytics(
# Literal argument value
"tool_used": t.tool_used,
# Literal argument value
"attack_success": t.attack_success,
"attack_success": t.attack_success.value if t.attack_success else None,
# Literal argument value
"remediation_status": t.remediation_status,
}
@@ -35,7 +35,7 @@ from typing import Any
import requests
from sqlalchemy.orm import Session
from app.models.enums import TestState, TestResult
from app.models.enums import AttackSuccessResult, TestState, TestResult
from app.models.evaluation_import import EvaluationImport
from app.models.technique import Technique
from app.models.test import Test
@@ -625,7 +625,7 @@ def import_evaluation_round(
procedure_text=procedure_text,
created_by=current_user.id,
state=TestState.in_review,
attack_success=True,
attack_success=AttackSuccessResult.successful,
red_summary=red_summary,
red_validation_status="approved",
red_validated_by=current_user.id,
@@ -305,7 +305,10 @@ def build_test_results_report(
# Literal argument value
"platform": t.platform,
# Literal argument value
"attack_success": t.attack_success,
"attack_success": (
t.attack_success.value if t.attack_success and hasattr(t.attack_success, "value")
else str(t.attack_success) if t.attack_success else None
),
# Literal argument value
"detection_result": (
t.detection_result.value if t.detection_result and hasattr(t.detection_result, "value")
+179 -8
View File
@@ -65,6 +65,33 @@ from app.models.technique import Technique
from app.models.test import Test
from app.models.user import User
# ---------------------------------------------------------------------------
# Custom field IDs (project-specific — see System Settings → Jira Configuration)
# ---------------------------------------------------------------------------
JIRA_FIELD_RT_START_DATE = "customfield_11803"
JIRA_FIELD_RT_END_DATE = "customfield_11804"
JIRA_FIELD_BT_START_DATE = "customfield_11805"
JIRA_FIELD_BT_END_DATE = "customfield_11806"
JIRA_FIELD_TTP = "customfield_11807"
JIRA_FIELD_SEVERITY = "customfield_10098"
JIRA_FIELD_ATTACK_SUCCESS = "customfield_11815"
JIRA_FIELD_ATTACK_DETECTED = "customfield_11809"
JIRA_FIELD_TIME_TO_DETECT = "customfield_11810"
JIRA_FIELD_TIME_TO_CONTAIN = "customfield_11811"
_ATTACK_SUCCESS_TO_JIRA = {
"successful": "Yes",
"not_successful": "No",
"partially_successful": "Partial",
}
_DETECTION_TO_JIRA = {
"detected": "Yes",
"not_detected": "No",
"partially_detected": "Partial",
}
# Assign logger = logging.getLogger(__name__)
logger = logging.getLogger(__name__)
@@ -388,7 +415,7 @@ def _build_state_comment(
lines += [
"Red Team has finished execution and submitted evidence for Blue Team evaluation.",
"",
f"*Attack Success:* {test.attack_success if test.attack_success is not None else 'N/A'}",
f"*Attack Success:* {_enum_value(test.attack_success) or 'N/A'}",
]
if test.red_summary:
lines += ["", "h4. Red Team Summary", test.red_summary]
@@ -627,6 +654,7 @@ def auto_create_test_issue(
issue_type = settings.JIRA_ISSUE_TYPE_TEST # always Task
poc = test.procedure_text or "N/A"
severity = _technique_severity(technique).capitalize()
fields: dict = {
"project": {"key": project_key},
"summary": f"[Aegis] {mitre_id}{test.name}",
@@ -635,6 +663,8 @@ def auto_create_test_issue(
"labels": ["aegis", "security-test", mitre_id.replace(".", "-")],
# customfield_10309 = Proof of Concept field (required by team's Jira config)
"customfield_10309": f"{{code}}{poc}{{code}}",
JIRA_FIELD_TTP: mitre_id,
JIRA_FIELD_SEVERITY: severity,
}
# Inherit campaign start date if available, otherwise use today
@@ -803,20 +833,67 @@ def push_hold_event(
f"h3. ▶ Test Resumed\n\n"
f"*Resumed by:* {actor.username}\n"
f"*At:* {ts}\n\n"
f"_The test has been taken off hold and is active again._"
f"_The test has been unblocked and is active again._"
)
try:
jira.set_issue_status(link.jira_issue_key, "In Progress")
except Exception as exc_t:
logger.warning("Could not transition %s off hold: %s", link.jira_issue_key, exc_t)
logger.warning("Could not transition %s off blocked: %s", link.jira_issue_key, exc_t)
else:
comment = (
f"h3. Test Placed On Hold\n\n"
f"*Placed on hold by:* {actor.username}\n"
f"h3. 🚫 Test Blocked\n\n"
f"*Blocked by:* {actor.username}\n"
f"*At:* {ts}\n"
f"*Reason:* {reason or 'No reason provided'}\n\n"
f"_This test has been paused. No action required until it is resumed._"
f"_This test is blocked. No action required until it is resumed._"
)
try:
jira.set_issue_status(link.jira_issue_key, "Blocked")
except Exception as exc_t:
logger.warning("Could not transition %s to Blocked: %s", link.jira_issue_key, exc_t)
jira.issue_add_comment(link.jira_issue_key, comment)
link.last_synced_at = datetime.utcnow()
db.flush()
logger.info("Posted hold event to Jira %s (resuming=%s)", link.jira_issue_key, resuming)
except Exception as exc:
logger.warning("Failed to push hold event for test %s: %s", test.id, exc, exc_info=True)
def push_pause_event(db: Session, test, actor, *, resuming: bool = False) -> None:
"""Post a timer pause/resume comment to Jira and toggle 'On Hold' / 'In Progress'.
Distinct from :func:`push_hold_event` — a short timer pause (the
operator stepping away briefly) maps to "On Hold", while an explicit
hold (something external blocking progress) maps to "Blocked".
Non-fatal — any Jira error is logged and swallowed.
"""
if not has_admin_jira_configured(db):
return
link = (
db.query(JiraLink)
.filter(
JiraLink.entity_type == JiraLinkEntityType.test,
JiraLink.entity_id == test.id,
)
.first()
)
if not link:
return
try:
jira = get_admin_jira_client(db)
ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M UTC")
if resuming:
comment = f"h3. ▶ Timer Resumed\n\n*Resumed by:* {actor.username}\n*At:* {ts}"
try:
jira.set_issue_status(link.jira_issue_key, "In Progress")
except Exception as exc_t:
logger.warning("Could not transition %s off On Hold: %s", link.jira_issue_key, exc_t)
else:
comment = f"h3. ⏸ Timer Paused\n\n*Paused by:* {actor.username}\n*At:* {ts}"
try:
jira.set_issue_status(link.jira_issue_key, "On Hold")
except Exception as exc_t:
@@ -825,9 +902,103 @@ def push_hold_event(
jira.issue_add_comment(link.jira_issue_key, comment)
link.last_synced_at = datetime.utcnow()
db.flush()
logger.info("Posted hold event to Jira %s (resuming=%s)", link.jira_issue_key, resuming)
logger.info("Posted pause event to Jira %s (resuming=%s)", link.jira_issue_key, resuming)
except Exception as exc:
logger.warning("Failed to push hold event for test %s: %s", test.id, exc, exc_info=True)
logger.warning("Failed to push pause event for test %s: %s", test.id, exc, exc_info=True)
# ---------------------------------------------------------------------------
# RT/BT date + result custom-field sync
# ---------------------------------------------------------------------------
def _enum_value(v) -> Optional[str]:
"""Return the plain string value of an enum member, or the value itself."""
if v is None:
return None
return v.value if hasattr(v, "value") else str(v)
def _compute_hours(start: Optional[datetime], end: Optional[datetime]) -> Optional[float]:
"""Return the elapsed time between two timestamps in hours, or None if unavailable."""
if not start or not end:
return None
delta_hours = (end - start).total_seconds() / 3600
return round(delta_hours, 2) if delta_hours >= 0 else None
def _update_test_fields(db: Session, test: Test, fields: dict) -> None:
"""Best-effort update of custom fields on the Jira issue linked to *test*.
Non-fatal — any Jira error is logged and swallowed so it never blocks
the test workflow.
"""
if not fields or not has_admin_jira_configured(db):
return
link = (
db.query(JiraLink)
.filter(
JiraLink.entity_type == JiraLinkEntityType.test,
JiraLink.entity_id == test.id,
)
.first()
)
if not link:
return
try:
jira = get_admin_jira_client(db)
jira.update_issue_field(link.jira_issue_key, fields=fields)
link.last_synced_at = datetime.utcnow()
db.flush()
logger.info("Updated Jira fields %s on %s", list(fields.keys()), link.jira_issue_key)
except Exception as exc:
logger.warning(
"Failed to update Jira fields for test %s: %s", test.id, exc, exc_info=True
)
def push_rt_started(db: Session, test: Test) -> None:
"""Set the RT Start Date field — called when the operator hits Start Execution."""
_update_test_fields(db, test, {
JIRA_FIELD_RT_START_DATE: datetime.utcnow().strftime("%Y-%m-%d"),
})
def push_rt_submitted(db: Session, test: Test) -> None:
"""Set the RT End Date + Attack Success fields — called when Red submits for review."""
fields: dict = {JIRA_FIELD_RT_END_DATE: datetime.utcnow().strftime("%Y-%m-%d")}
attack_success = _enum_value(test.attack_success)
if attack_success in _ATTACK_SUCCESS_TO_JIRA:
fields[JIRA_FIELD_ATTACK_SUCCESS] = _ATTACK_SUCCESS_TO_JIRA[attack_success]
_update_test_fields(db, test, fields)
def push_bt_started(db: Session, test: Test) -> None:
"""Set the BT Start Date field — called when Blue picks up the test to evaluate."""
_update_test_fields(db, test, {
JIRA_FIELD_BT_START_DATE: datetime.utcnow().strftime("%Y-%m-%d"),
})
def push_bt_submitted(db: Session, test: Test) -> None:
"""Set BT End Date, Attack Detected, and time-to-detect/contain — Blue submits for review."""
fields: dict = {JIRA_FIELD_BT_END_DATE: datetime.utcnow().strftime("%Y-%m-%d")}
detection_result = _enum_value(test.detection_result)
if detection_result in _DETECTION_TO_JIRA:
fields[JIRA_FIELD_ATTACK_DETECTED] = _DETECTION_TO_JIRA[detection_result]
time_to_detect = _compute_hours(test.execution_end_time, test.detection_time)
if time_to_detect is not None:
fields[JIRA_FIELD_TIME_TO_DETECT] = time_to_detect
time_to_contain = _compute_hours(test.detection_time, test.containment_time)
if time_to_contain is not None:
fields[JIRA_FIELD_TIME_TO_CONTAIN] = time_to_contain
_update_test_fields(db, test, fields)
# ---------------------------------------------------------------------------
@@ -223,6 +223,12 @@ def start_execution(db: Session, test: Test, user: User) -> Test:
except Exception as e:
logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)
try:
from app.services.jira_service import push_rt_started
push_rt_started(db, test)
except Exception as e:
logger.warning("Jira RT start date push failed for test %s: %s", test.id, e, exc_info=True)
return test
@@ -314,6 +320,12 @@ def submit_red_evidence(db: Session, test: Test, user: User) -> Test:
except Exception as e:
logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)
try:
from app.services.jira_service import push_rt_submitted
push_rt_submitted(db, test)
except Exception as e:
logger.warning("Jira RT end date push failed for test %s: %s", test.id, e, exc_info=True)
return test
@@ -410,6 +422,12 @@ def start_blue_work(db: Session, test: Test, user: User) -> Test:
except Exception as e:
logger.warning("Notification failed for test %s: %s", test.id, e, exc_info=True)
try:
from app.services.jira_service import push_bt_started
push_bt_started(db, test)
except Exception as e:
logger.warning("Jira BT start date push failed for test %s: %s", test.id, e, exc_info=True)
return test
@@ -494,6 +512,12 @@ def submit_blue_evidence(db: Session, test: Test, user: User) -> Test:
except Exception as e:
logger.warning("Jira push failed for test %s: %s", test.id, e, exc_info=True)
try:
from app.services.jira_service import push_bt_submitted
push_bt_submitted(db, test)
except Exception as e:
logger.warning("Jira BT end date push failed for test %s: %s", test.id, e, exc_info=True)
return test
@@ -637,6 +661,13 @@ def pause_timer(db: Session, test: Test, user: User) -> Test:
# Keyword argument: details
details={"state": test.state.value},
)
try:
from app.services.jira_service import push_pause_event
push_pause_event(db, test, user, resuming=False)
except Exception as e:
logger.warning("Jira pause push failed for test %s: %s", test.id, e, exc_info=True)
# Return test
return test
@@ -692,6 +723,13 @@ def resume_timer(db: Session, test: Test, user: User) -> Test:
# Keyword argument: details
details={"paused_seconds": paused_seconds, "state": test.state.value},
)
try:
from app.services.jira_service import push_pause_event
push_pause_event(db, test, user, resuming=True)
except Exception as e:
logger.warning("Jira resume push failed for test %s: %s", test.id, e, exc_info=True)
# Return test
return test
+137
View File
@@ -37,6 +37,143 @@ def test_sync_aegis_to_jira_adds_comment(mock_get_client, db):
assert link.last_synced_at is not None
def _make_link(entity_id, issue_key="SEC-100"):
return JiraLink(
entity_type=JiraLinkEntityType.test,
entity_id=entity_id,
jira_issue_key=issue_key,
)
def _make_test(**overrides):
from app.models.test import Test
t = MagicMock(spec=Test)
t.id = uuid4()
t.name = "Test"
t.attack_success = None
t.detection_result = None
t.execution_end_time = None
t.detection_time = None
t.containment_time = None
for k, v in overrides.items():
setattr(t, k, v)
return t
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_pause_event_sets_on_hold(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_pause_event(db, test, MagicMock(username="alice"), resuming=False)
mock_jira.set_issue_status.assert_called_once_with(link.jira_issue_key, "On Hold")
mock_jira.issue_add_comment.assert_called_once()
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_pause_event_resume_sets_in_progress(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_pause_event(db, test, MagicMock(username="alice"), resuming=True)
mock_jira.set_issue_status.assert_called_once_with(link.jira_issue_key, "In Progress")
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_hold_event_sets_blocked(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_hold_event(db, test, MagicMock(username="alice"), resuming=False, reason="waiting on approval")
mock_jira.set_issue_status.assert_called_once_with(link.jira_issue_key, "Blocked")
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_rt_started_sets_date_field(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test()
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_rt_started(db, test)
mock_jira.update_issue_field.assert_called_once()
args, kwargs = mock_jira.update_issue_field.call_args
assert args[0] == link.jira_issue_key
assert jira_service.JIRA_FIELD_RT_START_DATE in kwargs["fields"]
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_rt_submitted_includes_attack_success(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
test = _make_test(attack_success=MagicMock(value="successful"))
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_rt_submitted(db, test)
fields = mock_jira.update_issue_field.call_args[1]["fields"]
assert fields[jira_service.JIRA_FIELD_RT_END_DATE]
assert fields[jira_service.JIRA_FIELD_ATTACK_SUCCESS] == "Yes"
@patch("app.services.jira_service.has_admin_jira_configured", return_value=True)
@patch("app.services.jira_service.get_admin_jira_client")
def test_push_bt_submitted_includes_detection_and_hours(mock_get_client, mock_configured, db):
mock_jira = MagicMock()
mock_get_client.return_value = mock_jira
exec_end = datetime(2026, 1, 1, 10, 0, 0)
detect = datetime(2026, 1, 1, 12, 0, 0)
contain = datetime(2026, 1, 1, 13, 30, 0)
test = _make_test(
detection_result=MagicMock(value="detected"),
execution_end_time=exec_end,
detection_time=detect,
containment_time=contain,
)
link = _make_link(test.id)
db.add(link)
db.commit()
jira_service.push_bt_submitted(db, test)
fields = mock_jira.update_issue_field.call_args[1]["fields"]
assert fields[jira_service.JIRA_FIELD_ATTACK_DETECTED] == "Yes"
assert fields[jira_service.JIRA_FIELD_TIME_TO_DETECT] == 2.0
assert fields[jira_service.JIRA_FIELD_TIME_TO_CONTAIN] == 1.5
def test_update_test_fields_noop_when_not_configured(db):
test = _make_test()
# has_admin_jira_configured defaults to False without mocking — should
# silently no-op rather than raise.
jira_service._update_test_fields(db, test, {"customfield_1": "x"})
@patch("app.services.jira_service.get_jira_client")
def test_search_jira_issues_maps_fields(mock_get_client):
mock_jira = MagicMock()
+2 -1
View File
@@ -1,4 +1,5 @@
import client from "./client";
import type { AttackSuccessResult } from "../types/models";
// ── Types ───────────────────────────────────────────────────────────
@@ -42,7 +43,7 @@ export interface TestResultsReport {
technique_id: string;
state: string;
platform: string | null;
attack_success: boolean | null;
attack_success: AttackSuccessResult | null;
detection_result: string | null;
red_validation_status: string | null;
blue_validation_status: string | null;
+16 -3
View File
@@ -3,6 +3,8 @@ import type {
Test,
TestResult,
ContainmentResult,
AttackSuccessResult,
DataClassification,
TestState,
TestTimelineEntry,
} from "../types/models";
@@ -32,7 +34,7 @@ export interface RedUpdatePayload {
description?: string;
procedure_text?: string;
tool_used?: string;
attack_success?: boolean;
attack_success?: AttackSuccessResult;
red_summary?: string;
execution_start_time?: string;
execution_end_time?: string;
@@ -137,6 +139,17 @@ export async function updateTest(
return data;
}
/** Update the data classification label for a test (admin only). */
export async function updateTestClassification(
testId: string,
dataClassification: DataClassification,
): Promise<Test> {
const { data } = await client.patch<Test>(`/tests/${testId}/classification`, {
data_classification: dataClassification,
});
return data;
}
// ── Red Team ───────────────────────────────────────────────────────
/** Red Team updates their fields (draft, red_executing). */
@@ -365,7 +378,7 @@ export interface RTEvidenceEntry {
export interface RTTechniqueEntry {
mitre_id: string;
result: "detected" | "not_detected" | "partially_detected";
attack_success: boolean;
attack_success: AttackSuccessResult;
platform?: string;
notes?: string;
evidence: RTEvidenceEntry[]; // required — at least 1 image
@@ -382,7 +395,7 @@ export interface RTImportPayload {
export interface RTImportResult {
created: number;
skipped: number;
items: { mitre_id: string; test_name: string; result: string; attack_success: boolean; evidence_attached: number }[];
items: { mitre_id: string; test_name: string; result: string; attack_success: AttackSuccessResult; evidence_attached: number }[];
warnings: { mitre_id: string; reason: string }[];
engagement: string;
}
@@ -0,0 +1,84 @@
import { useState } from "react";
import { ChevronDown, ChevronUp, Loader2, Lock } from "lucide-react";
import type { DataClassification } from "../../types/models";
// ── Options ────────────────────────────────────────────────────────
const CLASSIFICATION_OPTIONS: { value: DataClassification; label: string; color: string }[] = [
{ value: "public", label: "Public", color: "border-gray-500/30 bg-gray-800/50 text-gray-400" },
{ value: "internal", label: "Internal", color: "border-blue-500/30 bg-blue-900/30 text-blue-400" },
{ value: "sensitive", label: "Sensitive", color: "border-amber-500/30 bg-amber-900/30 text-amber-400" },
{ value: "restricted", label: "Restricted", color: "border-red-500/30 bg-red-900/30 text-red-400" },
];
// ── Props ──────────────────────────────────────────────────────────
interface DataClassificationBadgeProps {
value: DataClassification;
canEdit: boolean;
isSaving: boolean;
onChange: (value: DataClassification) => void;
}
// ── Component ──────────────────────────────────────────────────────
export default function DataClassificationBadge({
value,
canEdit,
isSaving,
onChange,
}: DataClassificationBadgeProps) {
const [expanded, setExpanded] = useState(false);
const current = CLASSIFICATION_OPTIONS.find((o) => o.value === value) ?? CLASSIFICATION_OPTIONS[1];
if (!canEdit) {
return (
<span
className={`inline-flex items-center gap-1 rounded-full border px-2 py-0.5 text-[10px] font-medium ${current.color}`}
>
<Lock className="h-2.5 w-2.5" />
{current.label}
</span>
);
}
return (
<div className="relative inline-block">
<button
onClick={() => setExpanded((v) => !v)}
className={`inline-flex items-center gap-1 rounded-full border px-2 py-0.5 text-[10px] font-medium transition-colors hover:opacity-80 ${current.color}`}
>
<Lock className="h-2.5 w-2.5" />
{current.label}
{expanded ? <ChevronUp className="h-2.5 w-2.5" /> : <ChevronDown className="h-2.5 w-2.5" />}
</button>
{expanded && (
<div className="absolute left-0 top-full z-20 mt-1 w-44 rounded-lg border border-gray-700 bg-gray-900 p-1.5 shadow-xl">
{isSaving ? (
<div className="flex items-center justify-center py-2">
<Loader2 className="h-4 w-4 animate-spin text-gray-500" />
</div>
) : (
CLASSIFICATION_OPTIONS.map((opt) => (
<button
key={opt.value}
onClick={() => {
onChange(opt.value);
setExpanded(false);
}}
className={`block w-full rounded px-2 py-1.5 text-left text-xs transition-colors ${
opt.value === value
? "bg-gray-800 text-white"
: "text-gray-400 hover:bg-gray-800 hover:text-white"
}`}
>
{opt.label}
</button>
))
)}
</div>
)}
</div>
);
}
@@ -18,6 +18,7 @@ import type {
Test,
TestResult,
ContainmentResult,
AttackSuccessResult,
TeamSide,
Evidence,
TestTimelineEntry,
@@ -61,11 +62,29 @@ const DETECTION_RESULTS: { value: TestResult; label: string; color: string }[] =
];
// ── Containment result options ─────────────────────────────────────
// Order matches Detection's full/none/partial pattern for consistency.
const CONTAINMENT_RESULTS: { value: ContainmentResult; label: string; color: string }[] = [
{ value: "contained", label: "Contained", color: "border-green-500 bg-green-500/10 text-green-400" },
{ value: "partially_contained", label: "Partially Contained", color: "border-yellow-500 bg-yellow-500/10 text-yellow-400" },
{ value: "not_contained", label: "Not Contained", color: "border-red-500 bg-red-500/10 text-red-400" },
{
value: "partially_contained",
label: "Partially Contained",
color: "border-yellow-500 bg-yellow-500/10 text-yellow-400",
},
];
// ── Attack success options ──────────────────────────────────────────
// Order matches Detection's full/none/partial pattern for consistency.
const ATTACK_SUCCESS_RESULTS: { value: AttackSuccessResult; label: string; color: string }[] = [
{ value: "successful", label: "Yes", color: "border-green-500 bg-green-500/10 text-green-400" },
{ value: "not_successful", label: "No", color: "border-red-500 bg-red-500/10 text-red-400" },
{
value: "partially_successful",
label: "Partial",
color: "border-yellow-500 bg-yellow-500/10 text-yellow-400",
},
];
// ── Props ──────────────────────────────────────────────────────────
@@ -81,7 +100,7 @@ interface TeamTabsProps {
redDraft: {
procedure_text: string;
tool_used: string;
attack_success: boolean;
attack_success: AttackSuccessResult | "";
red_summary: string;
execution_start_time: string;
execution_end_time: string;
@@ -233,28 +252,45 @@ export default function TeamTabs({
)}
</div>
{/* Attack Success toggle */}
{/* Attack Success */}
<div>
<label className="mb-1.5 block text-sm font-medium text-gray-300">
<label className="mb-2 block text-sm font-medium text-gray-300">
Attack Successful?
</label>
{canEditRed ? (
<div className="flex flex-wrap gap-2">
{ATTACK_SUCCESS_RESULTS.map((opt) => (
<button
onClick={() => onRedFieldChange("attack_success", !redDraft.attack_success)}
className={`relative inline-flex h-6 w-11 items-center rounded-full transition-colors ${
redDraft.attack_success ? "bg-green-600" : "bg-gray-600"
key={opt.value}
onClick={() => onRedFieldChange("attack_success", opt.value)}
className={`rounded-lg border px-3 py-2 text-sm font-medium transition-colors ${
redDraft.attack_success === opt.value
? opt.color
: "border-gray-700 bg-gray-800 text-gray-400 hover:border-gray-600"
}`}
>
<span
className={`inline-block h-4 w-4 transform rounded-full bg-white transition-transform ${
redDraft.attack_success ? "translate-x-6" : "translate-x-1"
}`}
/>
{opt.label}
</button>
))}
</div>
) : (
<span className={`text-sm ${test.attack_success ? "text-green-400" : "text-red-400"}`}>
{test.attack_success === null ? "Not set" : test.attack_success ? "Yes" : "No"}
<p className="text-sm">
{test.attack_success ? (
<span
className={`inline-flex rounded-full border px-2.5 py-0.5 text-xs font-medium ${
test.attack_success === "successful"
? "border-green-500/30 bg-green-900/50 text-green-400"
: test.attack_success === "not_successful"
? "border-red-500/30 bg-red-900/50 text-red-400"
: "border-yellow-500/30 bg-yellow-900/50 text-yellow-400"
}`}
>
{ATTACK_SUCCESS_RESULTS.find((o) => o.value === test.attack_success)?.label}
</span>
) : (
<span className="text-gray-500">Not set</span>
)}
</p>
)}
</div>
@@ -660,12 +696,20 @@ export default function TeamTabs({
<div>
<dt className="text-xs font-medium uppercase text-gray-500">Attack Success</dt>
<dd className="mt-0.5 text-sm">
{test.attack_success === null ? (
<span className="text-gray-500">Not set</span>
) : test.attack_success ? (
<span className="text-green-400">Yes</span>
{test.attack_success ? (
<span
className={
test.attack_success === "successful"
? "text-green-400"
: test.attack_success === "not_successful"
? "text-red-400"
: "text-yellow-400"
}
>
{ATTACK_SUCCESS_RESULTS.find((o) => o.value === test.attack_success)?.label}
</span>
) : (
<span className="text-red-400">No</span>
<span className="text-gray-500">Not set</span>
)}
</dd>
</div>
@@ -18,8 +18,9 @@ import {
} from "lucide-react";
import { useMutation } from "@tanstack/react-query";
import { requestDiscussion } from "../../api/tests";
import type { Test, TestState, User } from "../../types/models";
import type { Test, TestState, User, DataClassification } from "../../types/models";
import LiveTimer from "./LiveTimer";
import DataClassificationBadge from "./DataClassificationBadge";
// ── Progress steps ─────────────────────────────────────────────────
@@ -87,6 +88,8 @@ interface TestDetailHeaderProps {
onHold: () => void;
onResume: () => void;
isTogglingHold: boolean;
onUpdateClassification: (value: DataClassification) => void;
isUpdatingClassification: boolean;
}
// ── Component ──────────────────────────────────────────────────────
@@ -109,6 +112,8 @@ export default function TestDetailHeader({
onHold,
onResume,
isTogglingHold,
onUpdateClassification,
isUpdatingClassification,
}: TestDetailHeaderProps) {
const role = user?.role ?? "";
const currentIdx = STATE_INDEX[test.state];
@@ -524,6 +529,12 @@ export default function TestDetailHeader({
>
{getStateLabel(test)}
</span>
<DataClassificationBadge
value={test.data_classification}
canEdit={role === "admin"}
isSaving={isUpdatingClassification}
onChange={onUpdateClassification}
/>
</div>
<p className="mt-1 text-sm text-gray-400">
Created {formatDate(test.created_at)}
+13 -10
View File
@@ -34,7 +34,7 @@ const TEMPLATE_JSON: RTImportPayload = {
{
mitre_id: "T1059.001",
result: "detected",
attack_success: true,
attack_success: "successful",
platform: "windows",
notes: "PowerShell execution caught by Defender for Endpoint within 2 min",
evidence: [
@@ -48,7 +48,7 @@ const TEMPLATE_JSON: RTImportPayload = {
{
mitre_id: "T1078",
result: "not_detected",
attack_success: true,
attack_success: "successful",
platform: "windows",
notes: "Credential reuse via stolen credentials — undetected for 48h",
evidence: [
@@ -67,7 +67,7 @@ const TEMPLATE_JSON: RTImportPayload = {
{
mitre_id: "T1486",
result: "not_detected",
attack_success: false,
attack_success: "not_successful",
platform: "windows",
notes: "Ransomware blocked by AppLocker before execution",
evidence: [
@@ -81,7 +81,7 @@ const TEMPLATE_JSON: RTImportPayload = {
{
mitre_id: "T1190",
result: "partially_detected",
attack_success: true,
attack_success: "partially_successful",
platform: "linux",
notes: "Exploit worked but only a partial alert fired — no full incident created",
evidence: [
@@ -231,7 +231,7 @@ export default function ImportRTPage() {
["name", "string", "Engagement name"],
["techniques[].mitre_id", "string", "e.g. T1059.001"],
["techniques[].result", "enum", "detected | not_detected | partially_detected"],
["techniques[].attack_success", "boolean", "Was the attack successful?"],
["techniques[].attack_success", "enum", "successful | not_successful | partially_successful"],
["techniques[].evidence", "array", "Min 1 image required per technique"],
["techniques[].evidence[].filename", "string", "e.g. screenshot_edr.png"],
["techniques[].evidence[].data", "string", "Base64-encoded image (PNG/JPG/GIF/WebP/BMP, max 10 MB)"],
@@ -299,7 +299,7 @@ export default function ImportRTPage() {
<textarea
value={jsonText}
onChange={(e) => handleTextChange(e.target.value)}
placeholder={`Paste your JSON here or use the template…\n\n{\n "name": "Red Team Q1 2024",\n "techniques": [\n {\n "mitre_id": "T1059.001",\n "result": "detected",\n "attack_success": true\n }\n ]\n}`}
placeholder={`Paste your JSON here or use the template…\n\n{\n "name": "Red Team Q1 2024",\n "techniques": [\n {\n "mitre_id": "T1059.001",\n "result": "detected",\n "attack_success": "successful"\n }\n ]\n}`}
rows={14}
className="w-full rounded-lg border border-gray-700 bg-gray-800 px-3 py-2.5 font-mono text-xs text-gray-200 placeholder-gray-600 focus:border-cyan-500 focus:outline-none resize-none"
/>
@@ -361,10 +361,13 @@ export default function ImportRTPage() {
</span>
</td>
<td className="px-4 py-2.5">
{t.attack_success
? <span className="text-orange-400">Success</span>
: <span className="text-gray-500">Blocked</span>
}
{t.attack_success === "successful" ? (
<span className="text-orange-400">Success</span>
) : t.attack_success === "partially_successful" ? (
<span className="text-yellow-400">Partial</span>
) : (
<span className="text-gray-500">Blocked</span>
)}
</td>
<td className="px-4 py-2.5 text-gray-400 capitalize">{t.platform ?? "—"}</td>
<td className="px-4 py-2.5 text-gray-500 max-w-xs truncate">{t.notes ?? "—"}</td>
+24 -5
View File
@@ -22,12 +22,13 @@ import {
resumeTimer,
holdTest,
resumeTest,
updateTestClassification,
getTestTimeline,
getRetestChain,
} from "../api/tests";
import { uploadEvidence, getEvidence } from "../api/evidence";
import { useAuth } from "../context/AuthContext";
import type { TestResult, ContainmentResult, TeamSide, TestTimelineEntry } from "../types/models";
import type { TestResult, ContainmentResult, AttackSuccessResult, DataClassification, TeamSide, TestTimelineEntry } from "../types/models";
import TestDetailHeader from "../components/test-detail/TestDetailHeader";
import TeamTabs from "../components/test-detail/TeamTabs";
@@ -72,10 +73,17 @@ export default function TestDetailPage() {
const [holdModal, setHoldModal] = useState(false);
const [holdReason, setHoldReason] = useState("");
const [redDraft, setRedDraft] = useState({
const [redDraft, setRedDraft] = useState<{
procedure_text: string;
tool_used: string;
attack_success: AttackSuccessResult | "";
red_summary: string;
execution_start_time: string;
execution_end_time: string;
}>({
procedure_text: "",
tool_used: "",
attack_success: false,
attack_success: "",
red_summary: "",
execution_start_time: "",
execution_end_time: "",
@@ -131,7 +139,7 @@ export default function TestDetailPage() {
setRedDraft({
procedure_text: test.procedure_text || "",
tool_used: test.tool_used || "",
attack_success: test.attack_success ?? false,
attack_success: test.attack_success ?? "",
red_summary: test.red_summary || "",
execution_start_time: isoToDatetimeLocal(test.execution_start_time),
execution_end_time: isoToDatetimeLocal(test.execution_end_time),
@@ -179,7 +187,7 @@ export default function TestDetailPage() {
updateTestRed(testId!, {
procedure_text: redDraft.procedure_text || undefined,
tool_used: redDraft.tool_used || undefined,
attack_success: redDraft.attack_success,
attack_success: redDraft.attack_success || undefined,
red_summary: redDraft.red_summary || undefined,
execution_start_time: redDraft.execution_start_time || undefined,
execution_end_time: redDraft.execution_end_time || undefined,
@@ -351,6 +359,15 @@ export default function TestDetailPage() {
onError: (err: unknown) => showToast(extractError(err), "error"),
});
const updateClassificationMutation = useMutation({
mutationFn: (value: DataClassification) => updateTestClassification(testId!, value),
onSuccess: () => {
invalidateAll();
showToast("Data classification updated", "success");
},
onError: (err: unknown) => showToast(extractError(err), "error"),
});
// Evidence upload
const uploadMutation = useMutation({
mutationFn: ({ file, team }: { file: File; team: TeamSide }) =>
@@ -513,6 +530,8 @@ export default function TestDetailPage() {
onHold={() => setHoldModal(true)}
onResume={() => resumeHoldMutation.mutate()}
isTogglingHold={holdMutation.isPending || resumeHoldMutation.isPending}
onUpdateClassification={(value) => updateClassificationMutation.mutate(value)}
isUpdatingClassification={updateClassificationMutation.isPending}
/>
{/* Content: Tabs + Sidebar */}
+8 -4
View File
@@ -13,7 +13,7 @@ import {
Swords,
} from "lucide-react";
import { getTests } from "../api/tests";
import type { Test, TestResult } from "../types/models";
import type { Test, TestResult, AttackSuccessResult } from "../types/models";
/* ── Result helpers ─────────────────────────────────────────────────── */
@@ -41,19 +41,23 @@ function DetectionBadge({ result }: { result: TestResult | null | undefined }) {
);
}
function AttackBadge({ success }: { success: boolean | null | undefined }) {
function AttackBadge({ success }: { success: AttackSuccessResult | null | undefined }) {
if (success === null || success === undefined)
return <span className="text-gray-600 text-xs"></span>;
const isSuccess = success === "successful";
const isPartial = success === "partially_successful";
return (
<span
className={`inline-flex items-center gap-1 rounded-full border px-2 py-0.5 text-xs font-medium ${
success
isSuccess
? "bg-orange-500/10 text-orange-400 border-orange-500/20"
: isPartial
? "bg-yellow-500/10 text-yellow-400 border-yellow-500/20"
: "bg-gray-500/10 text-gray-400 border-gray-600/20"
}`}
>
<Swords className="h-3 w-3" />
{success ? "Success" : "Blocked"}
{isSuccess ? "Success" : isPartial ? "Partial" : "Blocked"}
</span>
);
}
+8 -1
View File
@@ -60,6 +60,10 @@ export type TestResult = "detected" | "not_detected" | "partially_detected";
export type ContainmentResult = "contained" | "partially_contained" | "not_contained";
export type AttackSuccessResult = "successful" | "not_successful" | "partially_successful";
export type DataClassification = "public" | "internal" | "sensitive" | "restricted";
export type ValidationStatus = "pending" | "approved" | "rejected";
export type TeamSide = "red" | "blue";
@@ -87,7 +91,7 @@ export interface Test {
// Red Team fields
red_summary: string | null;
attack_success: boolean | null;
attack_success: AttackSuccessResult | null;
execution_start_time: string | null;
execution_end_time: string | null;
red_validated_by: string | null;
@@ -145,6 +149,9 @@ export interface Test {
retest_of: string | null;
retest_count: number;
// Data classification
data_classification: DataClassification;
// Technique info (populated in list endpoints)
technique_mitre_id: string | null;
technique_name: string | null;