kitos
82ac9c7014
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
- TestTemplate/TemplateSuggestion creation and updates now reject any mitre_technique_id that doesn't match a real, already-synced MITRE ATT&CK technique. Frontend swaps the free-text ID input for a technique picker. - Test Catalog now shows pending template suggestions before the catalog grid, with full field detail (platform, severity, tool, atomic ID, source URL, remediation) instead of just name/procedure. - A manager can edit and directly re-approve a campaign they previously rejected, instead of needing the original lead to resubmit it.
120 lines
4.5 KiB
Python
120 lines
4.5 KiB
Python
"""Operator template-proposal workflow.
|
|
|
|
Leads create test templates directly (see test_template_service.create_template,
|
|
gated to red_lead/blue_lead in the router). An operator (red_tech/blue_tech)
|
|
gets the same "propose a template" action, but nothing is written to the
|
|
live catalog until a lead on their team reviews it — the lead can approve
|
|
as-is, approve with edits, or discard it outright.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.domain.errors import EntityNotFoundError, InvalidOperationError
|
|
from app.models.template_suggestion import TemplateSuggestion
|
|
from app.models.test_template import TestTemplate
|
|
from app.models.user import User
|
|
from app.services.notification_service import notify_role
|
|
from app.services.test_template_service import create_template, validate_mitre_technique_id
|
|
|
|
_TEAM_BY_ROLE = {"red_tech": "red", "blue_tech": "blue"}
|
|
_LEAD_ROLE = {"red": "red_lead", "blue": "blue_lead"}
|
|
|
|
_TEMPLATE_FIELDS = (
|
|
"mitre_technique_id", "name", "description", "source", "source_url",
|
|
"attack_procedure", "expected_detection", "platform", "tool_suggested",
|
|
"severity", "atomic_test_id", "suggested_remediation",
|
|
)
|
|
|
|
|
|
def team_for_submitter(user: User) -> str:
|
|
"""Which team a submission belongs to, based on the submitter's role."""
|
|
team = _TEAM_BY_ROLE.get(user.role)
|
|
if team is None:
|
|
raise InvalidOperationError("Only red_tech/blue_tech operators can propose templates")
|
|
return team
|
|
|
|
|
|
def create_template_suggestion(db: Session, submitter: User, **fields: object) -> TemplateSuggestion:
|
|
"""Create a pending template suggestion and notify the submitter's lead.
|
|
|
|
Does not commit; caller uses UnitOfWork.
|
|
"""
|
|
team = team_for_submitter(submitter)
|
|
validate_mitre_technique_id(db, fields["mitre_technique_id"])
|
|
suggestion = TemplateSuggestion(team=team, submitted_by=submitter.id, **fields)
|
|
db.add(suggestion)
|
|
db.flush()
|
|
|
|
notify_role(
|
|
db,
|
|
role=_LEAD_ROLE[team],
|
|
type="template_suggestion",
|
|
title="New test template proposed for review",
|
|
message=f'A new template, "{suggestion.name}", was proposed and needs your review.',
|
|
entity_type="template_suggestion",
|
|
entity_id=suggestion.id,
|
|
)
|
|
return suggestion
|
|
|
|
|
|
def list_pending_template_suggestions(db: Session, *, team: str | None = None) -> list[TemplateSuggestion]:
|
|
"""List pending template suggestions, optionally filtered to one team."""
|
|
query = db.query(TemplateSuggestion).filter(TemplateSuggestion.status == "pending")
|
|
if team is not None:
|
|
query = query.filter(TemplateSuggestion.team == team)
|
|
return query.order_by(TemplateSuggestion.created_at.asc()).all()
|
|
|
|
|
|
def _get_pending_suggestion_or_raise(db: Session, suggestion_id: uuid.UUID) -> TemplateSuggestion:
|
|
suggestion = db.query(TemplateSuggestion).filter(TemplateSuggestion.id == suggestion_id).first()
|
|
if suggestion is None:
|
|
raise EntityNotFoundError("TemplateSuggestion", str(suggestion_id))
|
|
if suggestion.status != "pending":
|
|
raise InvalidOperationError("This template suggestion has already been reviewed.")
|
|
return suggestion
|
|
|
|
|
|
def approve_template_suggestion(
|
|
db: Session,
|
|
suggestion_id: uuid.UUID,
|
|
user: User,
|
|
overrides: dict | None = None,
|
|
) -> tuple[TemplateSuggestion, TestTemplate]:
|
|
"""Approve a suggestion: create the real TestTemplate and mark it reviewed.
|
|
|
|
*overrides* lets the reviewing lead adjust any field before it goes
|
|
live — unset/None fields fall back to what the operator submitted.
|
|
Does not commit; caller uses UnitOfWork.
|
|
"""
|
|
suggestion = _get_pending_suggestion_or_raise(db, suggestion_id)
|
|
|
|
fields = {name: getattr(suggestion, name) for name in _TEMPLATE_FIELDS}
|
|
for key, value in (overrides or {}).items():
|
|
if value is not None and key in fields:
|
|
fields[key] = value
|
|
|
|
template = create_template(db, **fields)
|
|
db.flush()
|
|
|
|
suggestion.status = "approved"
|
|
suggestion.reviewed_by = user.id
|
|
suggestion.reviewed_at = datetime.utcnow()
|
|
suggestion.created_template_id = template.id
|
|
db.flush()
|
|
return suggestion, template
|
|
|
|
|
|
def reject_template_suggestion(db: Session, suggestion_id: uuid.UUID, user: User) -> TemplateSuggestion:
|
|
"""Discard a suggestion without creating a template. Does not commit."""
|
|
suggestion = _get_pending_suggestion_or_raise(db, suggestion_id)
|
|
suggestion.status = "rejected"
|
|
suggestion.reviewed_by = user.id
|
|
suggestion.reviewed_at = datetime.utcnow()
|
|
db.flush()
|
|
return suggestion
|