mimic/backend/app/api/templates.py

"""SimulationTemplate CRUD endpoints — admin and redteam only."""
from __future__ import annotations

from datetime import UTC, datetime

import sqlalchemy.exc
from flask import Blueprint, g, jsonify, request

from backend.app.auth import role_required
from backend.app.extensions import db
from backend.app.models.simulation_template import SimulationTemplate
from backend.app.serializers import serialize_template
from backend.app.services import mitre as mitre_svc
from backend.app.services.simulation_workflow import (
    _resolve_tactic_ids,
    _resolve_technique_ids,
)

templates_bp = Blueprint("templates", __name__)

_MUTABLE_FIELDS = {"name", "description", "commands", "prerequisites", "technique_ids", "tactic_ids"}


@templates_bp.get("/api/templates")
@role_required("admin", "redteam")
def list_templates():
    items = SimulationTemplate.query.order_by(SimulationTemplate.name).all()
    return jsonify([serialize_template(t) for t in items]), 200


@templates_bp.post("/api/templates")
@role_required("admin", "redteam")
def create_template():
    data = request.get_json(silent=True) or {}
    name = (data.get("name") or "").strip()
    if not name:
        return jsonify({"error": "name is required"}), 400

    techniques: list[dict] = []
    tactic_ids_val: list[str] = []

    if "technique_ids" in data:
        if not isinstance(data["technique_ids"], list):
            return jsonify({"error": "technique_ids must be a list"}), 400
        if not mitre_svc.mitre_loaded:
            return jsonify({"error": "mitre bundle not loaded"}), 503
        resolved, err = _resolve_technique_ids(data["technique_ids"])
        if err is not None:
            return err
        techniques = resolved or []

    if "tactic_ids" in data:
        if not isinstance(data["tactic_ids"], list):
            return jsonify({"error": "tactic_ids must be a list"}), 400
        resolved_ta, err = _resolve_tactic_ids(data["tactic_ids"])
        if err is not None:
            return err
        tactic_ids_val = resolved_ta or []

    tmpl = SimulationTemplate(
        name=name,
        description=data.get("description"),
        commands=data.get("commands"),
        prerequisites=data.get("prerequisites"),
        techniques=techniques,
        tactic_ids=tactic_ids_val,
        created_at=datetime.now(UTC),
        created_by_id=g.current_user.id,
    )
    db.session.add(tmpl)
    try:
        db.session.commit()
    except sqlalchemy.exc.IntegrityError:
        db.session.rollback()
        return jsonify({"error": "template name already exists"}), 409

    return jsonify(serialize_template(tmpl)), 201


@templates_bp.get("/api/templates/<int:tid>")
@role_required("admin", "redteam")
def get_template(tid: int):
    tmpl = db.session.get(SimulationTemplate, tid)
    if tmpl is None:
        return jsonify({"error": "Template not found"}), 404
    return jsonify(serialize_template(tmpl)), 200


@templates_bp.patch("/api/templates/<int:tid>")
@role_required("admin", "redteam")
def update_template(tid: int):
    tmpl = db.session.get(SimulationTemplate, tid)
    if tmpl is None:
        return jsonify({"error": "Template not found"}), 404

    data = request.get_json(silent=True) or {}
    unknown = set(data.keys()) - _MUTABLE_FIELDS
    if unknown:
        return jsonify({"error": f"unknown fields: {sorted(unknown)}"}), 400

    if not data:
        return jsonify(serialize_template(tmpl)), 200

    if "name" in data:
        name = (data["name"] or "").strip()
        if not name:
            return jsonify({"error": "name cannot be empty"}), 400
        tmpl.name = name

    for field in ("description", "commands", "prerequisites"):
        if field in data:
            setattr(tmpl, field, data[field])

    if "technique_ids" in data:
        if not isinstance(data["technique_ids"], list):
            return jsonify({"error": "technique_ids must be a list"}), 400
        if not mitre_svc.mitre_loaded:
            return jsonify({"error": "mitre bundle not loaded"}), 503
        resolved, err = _resolve_technique_ids(data["technique_ids"])
        if err is not None:
            return err
        tmpl.techniques = resolved

    if "tactic_ids" in data:
        if not isinstance(data["tactic_ids"], list):
            return jsonify({"error": "tactic_ids must be a list"}), 400
        resolved_ta, err = _resolve_tactic_ids(data["tactic_ids"])
        if err is not None:
            return err
        tmpl.tactic_ids = resolved_ta

    tmpl.updated_at = datetime.now(UTC)

    try:
        db.session.commit()
    except sqlalchemy.exc.IntegrityError:
        db.session.rollback()
        return jsonify({"error": "template name already exists"}), 409

    return jsonify(serialize_template(tmpl)), 200


@templates_bp.delete("/api/templates/<int:tid>")
@role_required("admin", "redteam")
def delete_template(tid: int):
    tmpl = db.session.get(SimulationTemplate, tid)
    if tmpl is None:
        return jsonify({"error": "Template not found"}), 404
    db.session.delete(tmpl)
    db.session.commit()
    return "", 204