"""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 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: 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/") @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/") @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 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: 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/") @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