e1b51db25f
Addresses spec-reviewer + code-reviewer feedback on the M6 bundle:
Critical:
- frontend/src/lib/missions.ts: add `listPrefix()` so TanStack invalidation
catches every filtered list variant; the previous `list()` returned
`['missions','list',{}]` and only matched the exact empty-filter cache,
leaving filtered tables stale after create/transition/delete.
- backend/app/services/missions.py: acquire the same per-scenario
`pg_advisory_xact_lock` key used by `set_scenario_tests` before
snapshotting; without it a concurrent M5 reorder could freeze a torn
snapshot under READ COMMITTED. Sorted by key to avoid deadlocks with
another snapshotter.
Important:
- backend/app/api/missions.py: `@require_perm("mission.update",
"mission.archive")` on the transition endpoint so users without either
perm get 403 before the body is parsed (no shape leak via 400).
- backend/app/services/missions.py: escape `%` / `_` / `\` in user-typed
`q` / `client` LIKE search; users can no longer trigger wildcard
semantics by typing literal `%`. Added `escape='\\'` arg on every .like().
- backend/app/services/missions.py: filter `MissionTest.deleted_at` and
`MissionScenario.deleted_at` in the list-item and detail counts so M7+
soft-deletes don't drift the totals silently.
Nits:
- backend/app/api/users.py: order `/users/roster` by email for stable
rendering + deterministic e2e selectors.
- frontend/src/pages/MissionDetailPage.tsx: distinct accent per
transition target (cyan/orange/green/teal) matching the status legend.
- e2e/tests/m6-missions.spec.ts: switch fragile `getByRole(name=/In
Progress/i)` to the stable `mission-transition-in_progress` data-testid.
New tests:
- test_create_mission_rejects_soft_deleted_scenario
- test_transition_perm_gate_runs_before_payload_parse
- test_search_treats_wildcards_as_literals
Suite: 106 pytest passing (was 103), 43 Playwright passing.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
499 lines
17 KiB
Python
499 lines
17 KiB
Python
"""Missions API.
|
|
|
|
Per spec §4: a non-admin user can only see (or edit) missions they are a
|
|
member of. The decorator stack here gates the *action type* by permission
|
|
code; the service layer applies the membership filter. Both layers fail
|
|
closed.
|
|
|
|
Status transitions are routed through a single POST endpoint that accepts a
|
|
target status. We accept either `mission.update` or `mission.archive` at the
|
|
gate — archiving requires the dedicated perm if the target is `archived`, and
|
|
the service enforces the lifecycle graph (`_VALID_TRANSITIONS`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from datetime import date
|
|
from typing import Any
|
|
|
|
from flask import Blueprint, abort, g, jsonify, request
|
|
from pydantic import BaseModel, Field, ValidationError
|
|
|
|
from app.core.auth_decorators import AuthenticatedUser, require_auth, require_perm
|
|
from app.services import missions as svc
|
|
|
|
bp = Blueprint("missions", __name__, url_prefix="/missions")
|
|
log = logging.getLogger("metamorph.api.missions")
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Payloads
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
class MemberPayload(BaseModel):
|
|
user_id: uuid.UUID
|
|
role_hint: str = Field(min_length=1, max_length=8)
|
|
|
|
model_config = {"extra": "forbid"}
|
|
|
|
|
|
class CreateMissionPayload(BaseModel):
|
|
name: str = Field(min_length=1, max_length=255)
|
|
client_target: str | None = Field(default=None, max_length=255)
|
|
date_start: date | None = None
|
|
date_end: date | None = None
|
|
description_md: str | None = Field(default=None, max_length=20_000)
|
|
scenario_template_ids: list[uuid.UUID] = Field(default_factory=list, max_length=64)
|
|
members: list[MemberPayload] = Field(default_factory=list, max_length=128)
|
|
|
|
model_config = {"extra": "forbid"}
|
|
|
|
|
|
class UpdateMissionPayload(BaseModel):
|
|
name: str | None = Field(default=None, min_length=1, max_length=255)
|
|
client_target: str | None = Field(default=None, max_length=255)
|
|
date_start: date | None = None
|
|
date_end: date | None = None
|
|
description_md: str | None = Field(default=None, max_length=20_000)
|
|
|
|
model_config = {"extra": "forbid"}
|
|
|
|
|
|
class AddScenariosPayload(BaseModel):
|
|
scenario_template_ids: list[uuid.UUID] = Field(default_factory=list, max_length=64)
|
|
|
|
model_config = {"extra": "forbid"}
|
|
|
|
|
|
class SetMembersPayload(BaseModel):
|
|
members: list[MemberPayload] = Field(default_factory=list, max_length=128)
|
|
|
|
model_config = {"extra": "forbid"}
|
|
|
|
|
|
class TransitionPayload(BaseModel):
|
|
status: str = Field(min_length=1, max_length=16)
|
|
|
|
model_config = {"extra": "forbid"}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Serialisers
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _serialize_member(m: svc.MissionMemberView) -> dict[str, Any]:
|
|
return {
|
|
"user_id": str(m.user_id),
|
|
"user_email": m.user_email,
|
|
"user_display_name": m.user_display_name,
|
|
"role_hint": m.role_hint,
|
|
}
|
|
|
|
|
|
def _serialize_mitre_tag(tag: svc.MissionMitreTagView) -> dict[str, Any]:
|
|
return {
|
|
"kind": tag.kind,
|
|
"external_id": tag.external_id,
|
|
"name": tag.name,
|
|
"url": tag.url,
|
|
}
|
|
|
|
|
|
def _serialize_test(t: svc.MissionTestView) -> dict[str, Any]:
|
|
return {
|
|
"id": str(t.id),
|
|
"position": t.position,
|
|
"snapshot_name": t.snapshot_name,
|
|
"snapshot_description": t.snapshot_description,
|
|
"snapshot_objective": t.snapshot_objective,
|
|
"snapshot_procedure_md": t.snapshot_procedure_md,
|
|
"snapshot_prerequisites_md": t.snapshot_prerequisites_md,
|
|
"snapshot_expected_red_md": t.snapshot_expected_red_md,
|
|
"snapshot_expected_blue_md": t.snapshot_expected_blue_md,
|
|
"snapshot_opsec_level": t.snapshot_opsec_level,
|
|
"snapshot_tags": t.snapshot_tags,
|
|
"snapshot_expected_iocs": t.snapshot_expected_iocs,
|
|
"state": t.state,
|
|
"executed_at": t.executed_at.isoformat() if t.executed_at else None,
|
|
"executed_at_overridden": t.executed_at_overridden,
|
|
"mitre_tags": [_serialize_mitre_tag(tag) for tag in t.mitre_tags],
|
|
"source_test_template_id": (
|
|
str(t.source_test_template_id) if t.source_test_template_id else None
|
|
),
|
|
}
|
|
|
|
|
|
def _serialize_scenario(sc: svc.MissionScenarioView) -> dict[str, Any]:
|
|
return {
|
|
"id": str(sc.id),
|
|
"position": sc.position,
|
|
"snapshot_name": sc.snapshot_name,
|
|
"snapshot_description": sc.snapshot_description,
|
|
"tests": [_serialize_test(t) for t in sc.tests],
|
|
"source_scenario_template_id": (
|
|
str(sc.source_scenario_template_id)
|
|
if sc.source_scenario_template_id
|
|
else None
|
|
),
|
|
}
|
|
|
|
|
|
def _serialize_list_item(m: svc.MissionListItemView) -> dict[str, Any]:
|
|
return {
|
|
"id": str(m.id),
|
|
"name": m.name,
|
|
"client_target": m.client_target,
|
|
"date_start": m.date_start.isoformat() if m.date_start else None,
|
|
"date_end": m.date_end.isoformat() if m.date_end else None,
|
|
"status": m.status,
|
|
"description_md": m.description_md,
|
|
"visibility_mode": m.visibility_mode,
|
|
"scenarios_count": m.scenarios_count,
|
|
"tests_count": m.tests_count,
|
|
"members_count": m.members_count,
|
|
"deleted_at": m.deleted_at.isoformat() if m.deleted_at else None,
|
|
"created_at": m.created_at.isoformat(),
|
|
"updated_at": m.updated_at.isoformat(),
|
|
}
|
|
|
|
|
|
def _serialize_detail(m: svc.MissionView) -> dict[str, Any]:
|
|
base = {
|
|
"id": str(m.id),
|
|
"name": m.name,
|
|
"client_target": m.client_target,
|
|
"date_start": m.date_start.isoformat() if m.date_start else None,
|
|
"date_end": m.date_end.isoformat() if m.date_end else None,
|
|
"status": m.status,
|
|
"description_md": m.description_md,
|
|
"visibility_mode": m.visibility_mode,
|
|
"scenarios_count": m.scenarios_count,
|
|
"tests_count": m.tests_count,
|
|
"members_count": m.members_count,
|
|
"deleted_at": m.deleted_at.isoformat() if m.deleted_at else None,
|
|
"created_at": m.created_at.isoformat(),
|
|
"updated_at": m.updated_at.isoformat(),
|
|
}
|
|
base["scenarios"] = [_serialize_scenario(sc) for sc in m.scenarios]
|
|
base["members"] = [_serialize_member(mb) for mb in m.members]
|
|
return base
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Helpers
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
def _parse_uuid_or_400(raw: str) -> uuid.UUID | None:
|
|
try:
|
|
return uuid.UUID(raw)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _pagination_args() -> tuple[int, int] | tuple[None, tuple[int, str]]:
|
|
try:
|
|
limit = int(request.args.get("limit", "100"))
|
|
offset = int(request.args.get("offset", "0"))
|
|
except ValueError:
|
|
return None, (400, "invalid_pagination")
|
|
return max(1, min(limit, 500)), max(0, offset)
|
|
|
|
|
|
def _current_user() -> AuthenticatedUser:
|
|
user: AuthenticatedUser | None = getattr(g, "current_user", None)
|
|
if user is None:
|
|
abort(401, description="not authenticated")
|
|
assert user is not None # for Pyright; abort raises HTTPException
|
|
return user
|
|
|
|
|
|
def _to_assignments(payload_members: list[MemberPayload]) -> list[svc.MemberAssignment]:
|
|
return [
|
|
svc.MemberAssignment(user_id=m.user_id, role_hint=m.role_hint)
|
|
for m in payload_members
|
|
]
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# Endpoints
|
|
# --------------------------------------------------------------------------- #
|
|
|
|
|
|
@bp.get("")
|
|
@require_auth
|
|
@require_perm("mission.read")
|
|
def list_missions():
|
|
paging = _pagination_args()
|
|
if paging[0] is None:
|
|
return jsonify({"error": paging[1][1]}), paging[1][0]
|
|
limit, offset = paging
|
|
q = request.args.get("q") or None
|
|
status = request.args.get("status") or None
|
|
client = request.args.get("client") or None
|
|
include_deleted = request.args.get("include_deleted", "false").lower() == "true"
|
|
|
|
user = _current_user()
|
|
if include_deleted and not user.is_admin:
|
|
return jsonify({"error": "forbidden"}), 403
|
|
|
|
try:
|
|
items, total = svc.list_missions(
|
|
viewer_id=user.id,
|
|
viewer_is_admin=user.is_admin,
|
|
q=q,
|
|
status=status,
|
|
client=client,
|
|
include_deleted=include_deleted,
|
|
limit=limit,
|
|
offset=offset,
|
|
)
|
|
except ValueError as e:
|
|
return jsonify({"error": "invalid_request", "message": str(e)}), 400
|
|
return jsonify(
|
|
{
|
|
"items": [_serialize_list_item(it) for it in items],
|
|
"total": total,
|
|
"limit": limit,
|
|
"offset": offset,
|
|
}
|
|
)
|
|
|
|
|
|
@bp.get("/<mission_id>")
|
|
@require_auth
|
|
@require_perm("mission.read")
|
|
def get_mission(mission_id: str):
|
|
mid = _parse_uuid_or_400(mission_id)
|
|
if mid is None:
|
|
return jsonify({"error": "invalid_id"}), 400
|
|
include_deleted = request.args.get("include_deleted", "false").lower() == "true"
|
|
user = _current_user()
|
|
if include_deleted and not user.is_admin:
|
|
return jsonify({"error": "forbidden"}), 403
|
|
try:
|
|
view = svc.get_mission(
|
|
mid,
|
|
viewer_id=user.id,
|
|
viewer_is_admin=user.is_admin,
|
|
include_deleted=include_deleted,
|
|
)
|
|
except svc.MissionNotFound:
|
|
return jsonify({"error": "not_found"}), 404
|
|
return jsonify(_serialize_detail(view))
|
|
|
|
|
|
@bp.post("")
|
|
@require_auth
|
|
@require_perm("mission.create")
|
|
def create_mission():
|
|
try:
|
|
payload = CreateMissionPayload.model_validate(request.get_json(silent=True) or {})
|
|
except ValidationError as e:
|
|
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
|
|
user = _current_user()
|
|
try:
|
|
view = svc.create_mission(
|
|
name=payload.name,
|
|
creator_id=user.id,
|
|
creator_is_admin=user.is_admin,
|
|
client_target=payload.client_target,
|
|
date_start=payload.date_start,
|
|
date_end=payload.date_end,
|
|
description_md=payload.description_md,
|
|
scenario_template_ids=list(payload.scenario_template_ids),
|
|
members=_to_assignments(payload.members),
|
|
)
|
|
except svc.UnknownScenarioTemplate as e:
|
|
return jsonify({"error": "unknown_scenario_template", "message": str(e)}), 400
|
|
except svc.UnknownUser as e:
|
|
return jsonify({"error": "unknown_user", "message": str(e)}), 400
|
|
except svc.InvalidMemberPayload as e:
|
|
return jsonify({"error": "invalid_member", "message": str(e)}), 400
|
|
except ValueError as e:
|
|
return jsonify({"error": "invalid_request", "message": str(e)}), 400
|
|
log.info(
|
|
"metamorph.mission.created",
|
|
extra={
|
|
"mission_id": str(view.id),
|
|
"scenarios": view.scenarios_count,
|
|
"tests": view.tests_count,
|
|
"members": view.members_count,
|
|
},
|
|
)
|
|
return jsonify(_serialize_detail(view)), 201
|
|
|
|
|
|
@bp.put("/<mission_id>")
|
|
@require_auth
|
|
@require_perm("mission.update")
|
|
def update_mission(mission_id: str):
|
|
mid = _parse_uuid_or_400(mission_id)
|
|
if mid is None:
|
|
return jsonify({"error": "invalid_id"}), 400
|
|
raw = request.get_json(silent=True) or {}
|
|
try:
|
|
payload = UpdateMissionPayload.model_validate(raw)
|
|
except ValidationError as e:
|
|
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
|
|
# Distinguish "not provided" from "explicitly null" by looking at the raw body.
|
|
kwargs: dict[str, Any] = {}
|
|
if "name" in raw and payload.name is not None:
|
|
kwargs["name"] = payload.name
|
|
if "client_target" in raw:
|
|
kwargs["client_target"] = payload.client_target
|
|
if "date_start" in raw:
|
|
kwargs["date_start"] = payload.date_start
|
|
if "date_end" in raw:
|
|
kwargs["date_end"] = payload.date_end
|
|
if "description_md" in raw:
|
|
kwargs["description_md"] = payload.description_md
|
|
user = _current_user()
|
|
try:
|
|
view = svc.update_mission_metadata(
|
|
mid,
|
|
viewer_id=user.id,
|
|
viewer_is_admin=user.is_admin,
|
|
**kwargs,
|
|
)
|
|
except svc.MissionNotFound:
|
|
return jsonify({"error": "not_found"}), 404
|
|
except ValueError as e:
|
|
return jsonify({"error": "invalid_request", "message": str(e)}), 400
|
|
return jsonify(_serialize_detail(view))
|
|
|
|
|
|
@bp.post("/<mission_id>/scenarios")
|
|
@require_auth
|
|
@require_perm("mission.update")
|
|
def add_scenarios(mission_id: str):
|
|
mid = _parse_uuid_or_400(mission_id)
|
|
if mid is None:
|
|
return jsonify({"error": "invalid_id"}), 400
|
|
try:
|
|
payload = AddScenariosPayload.model_validate(request.get_json(silent=True) or {})
|
|
except ValidationError as e:
|
|
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
|
|
user = _current_user()
|
|
try:
|
|
view = svc.add_scenarios_to_mission(
|
|
mid,
|
|
list(payload.scenario_template_ids),
|
|
viewer_id=user.id,
|
|
viewer_is_admin=user.is_admin,
|
|
)
|
|
except svc.MissionNotFound:
|
|
return jsonify({"error": "not_found"}), 404
|
|
except svc.UnknownScenarioTemplate as e:
|
|
return jsonify({"error": "unknown_scenario_template", "message": str(e)}), 400
|
|
log.info(
|
|
"metamorph.mission.scenarios_added",
|
|
extra={
|
|
"mission_id": str(mid),
|
|
"added": len(payload.scenario_template_ids),
|
|
},
|
|
)
|
|
return jsonify(_serialize_detail(view))
|
|
|
|
|
|
@bp.put("/<mission_id>/members")
|
|
@require_auth
|
|
@require_perm("mission.update")
|
|
def set_members(mission_id: str):
|
|
mid = _parse_uuid_or_400(mission_id)
|
|
if mid is None:
|
|
return jsonify({"error": "invalid_id"}), 400
|
|
try:
|
|
payload = SetMembersPayload.model_validate(request.get_json(silent=True) or {})
|
|
except ValidationError as e:
|
|
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
|
|
user = _current_user()
|
|
try:
|
|
view = svc.set_mission_members(
|
|
mid,
|
|
_to_assignments(payload.members),
|
|
viewer_id=user.id,
|
|
viewer_is_admin=user.is_admin,
|
|
)
|
|
except svc.MissionNotFound:
|
|
return jsonify({"error": "not_found"}), 404
|
|
except svc.UnknownUser as e:
|
|
return jsonify({"error": "unknown_user", "message": str(e)}), 400
|
|
except svc.InvalidMemberPayload as e:
|
|
return jsonify({"error": "invalid_member", "message": str(e)}), 400
|
|
return jsonify(_serialize_detail(view))
|
|
|
|
|
|
@bp.post("/<mission_id>/transition")
|
|
@require_auth
|
|
@require_perm("mission.update", "mission.archive")
|
|
def transition(mission_id: str):
|
|
"""Status transition. The outer decorator gates the endpoint on holding
|
|
EITHER `mission.update` or `mission.archive` — so a request with neither
|
|
perm sees 403 before its body is even parsed (no shape leak via 400).
|
|
The inner refinement then enforces the per-target rule: `mission.archive`
|
|
is required when the target is `archived`; `mission.update` covers the
|
|
other transitions. Admins bypass via the decorator's `is_admin` check.
|
|
"""
|
|
mid = _parse_uuid_or_400(mission_id)
|
|
if mid is None:
|
|
return jsonify({"error": "invalid_id"}), 400
|
|
try:
|
|
payload = TransitionPayload.model_validate(request.get_json(silent=True) or {})
|
|
except ValidationError as e:
|
|
return jsonify({"error": "invalid_request", "details": e.errors()}), 400
|
|
user = _current_user()
|
|
required = "mission.archive" if payload.status == "archived" else "mission.update"
|
|
if not user.is_admin and required not in user.permissions:
|
|
log.info(
|
|
"metamorph.auth.permission_denied",
|
|
extra={
|
|
"user_id": str(user.id),
|
|
"required": [required],
|
|
"had": sorted(user.permissions),
|
|
},
|
|
)
|
|
return jsonify({"error": "forbidden"}), 403
|
|
try:
|
|
view = svc.transition_mission_status(
|
|
mid,
|
|
payload.status,
|
|
viewer_id=user.id,
|
|
viewer_is_admin=user.is_admin,
|
|
)
|
|
except svc.MissionNotFound:
|
|
return jsonify({"error": "not_found"}), 404
|
|
except svc.InvalidTransition as e:
|
|
return jsonify({"error": "invalid_transition", "message": str(e)}), 409
|
|
except ValueError as e:
|
|
return jsonify({"error": "invalid_request", "message": str(e)}), 400
|
|
log.info(
|
|
"metamorph.mission.transitioned",
|
|
extra={"mission_id": str(mid), "status": view.status},
|
|
)
|
|
return jsonify(_serialize_detail(view))
|
|
|
|
|
|
@bp.delete("/<mission_id>")
|
|
@require_auth
|
|
@require_perm("mission.delete")
|
|
def soft_delete_mission(mission_id: str):
|
|
mid = _parse_uuid_or_400(mission_id)
|
|
if mid is None:
|
|
return jsonify({"error": "invalid_id"}), 400
|
|
user = _current_user()
|
|
try:
|
|
svc.soft_delete_mission(
|
|
mid,
|
|
viewer_id=user.id,
|
|
viewer_is_admin=user.is_admin,
|
|
)
|
|
except svc.MissionNotFound:
|
|
return jsonify({"error": "not_found"}), 404
|
|
log.info("metamorph.mission.soft_deleted", extra={"mission_id": str(mid)})
|
|
return jsonify({"ok": True})
|