diff --git a/docs/superpowers/plans/2026-04-13-phase3c2-attack-chain-graph-view.md b/docs/superpowers/plans/2026-04-13-phase3c2-attack-chain-graph-view.md new file mode 100644 index 0000000..88b0dd1 --- /dev/null +++ b/docs/superpowers/plans/2026-04-13-phase3c2-attack-chain-graph-view.md @@ -0,0 +1,1865 @@ +# Phase 3C.2: Attack Chain Graph View — Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Add an interactive per-engagement attack chain graph visualization to the web dashboard, with edge curation (confirm/reject), MITRE ATT&CK phase coloring, and server-side filtering for scale. + +**Architecture:** Standalone Vue page at `/engagements/:id/chain` wraps `force-graph` (vasturiano). Backend serves filtered subgraphs via a new `GET /api/chain/subgraph` endpoint that caps nodes and filters by severity/status. Edge curation uses `PATCH /api/chain/relations/:id`. No new database tables — builds on 3C.1 models. + +**Tech Stack:** FastAPI, SQLAlchemy async, Vue 3, PrimeVue, `force-graph` (vasturiano), TanStack Query + +**Spec:** `docs/superpowers/specs/2026-04-13-phase3c2-attack-chain-graph-view-design.md` + +--- + +## File Map + +### Backend (new/modified) + +| File | Action | Responsibility | +|------|--------|---------------| +| `packages/web/backend/app/routes/chain.py` | Modify | Add `GET /api/chain/subgraph` and `PATCH /api/chain/relations/{relation_id}` endpoints | +| `packages/web/backend/app/services/chain_service.py` | Modify | Add `subgraph_for_engagement()` and `update_relation_status()` methods | +| `packages/web/backend/app/services/chain_dto.py` | Modify | Add `relation_to_link_dict()` for force-graph link shape with drift computation | +| `packages/web/backend/tests/test_chain_subgraph.py` | Create | Tests for subgraph endpoint filtering, capping, neighborhood, drift | +| `packages/web/backend/tests/test_chain_curation.py` | Create | Tests for relation PATCH (valid transitions, invalid status, auth scoping) | + +### Frontend (new/modified) + +| File | Action | Responsibility | +|------|--------|---------------| +| `packages/web/frontend/src/views/ChainGraphView.vue` | Create | Page component — data fetching, filter state, layout orchestration | +| `packages/web/frontend/src/components/ForceGraphCanvas.vue` | Create | Wrapper around `force-graph` — rendering config, custom draw callbacks, interaction events | +| `packages/web/frontend/src/components/ChainDetailPanel.vue` | Create | Right drawer — node details, edge details with reasons, curation buttons | +| `packages/web/frontend/src/components/ChainFilterToolbar.vue` | Create | Severity/status toggle buttons | +| `packages/web/frontend/src/components/ChainLegend.vue` | Create | Bottom bar — severity color key, edge style key, node count | +| `packages/web/frontend/src/components/ChainEmptyState.vue` | Create | Empty state + rebuild progress polling | +| `packages/web/frontend/src/router/index.ts` | Modify | Add `/engagements/:id/chain` route | +| `packages/web/frontend/src/views/EngagementDetailView.vue` | Modify | Add "View Attack Chain" button | + +--- + +## Task 1: Backend — `relation_to_link_dict` DTO with drift computation + +**Files:** +- Modify: `packages/web/backend/app/services/chain_dto.py` + +This task adds the conversion function that produces the force-graph link shape with inline drift computation. All subsequent backend tasks depend on this. + +- [ ] **Step 1: Write the `relation_to_link_dict` function** + +Add to `packages/web/backend/app/services/chain_dto.py`: + +```python +def relation_to_link_dict(relation: FindingRelation) -> dict[str, Any]: + """Convert a CLI ``FindingRelation`` to a force-graph link dict. + + Includes drift detection: if the relation has status USER_CONFIRMED + and the current reasons differ from the confirmed_at_reasons snapshot, + drift is True. + """ + status_value = ( + relation.status.value + if hasattr(relation.status, "value") + else str(relation.status) + ) + + # Drift: true if user confirmed but reasons have since changed + drift = False + if status_value == "user_confirmed" and relation.confirmed_at_reasons is not None: + current_rules = sorted(r.rule for r in relation.reasons) + confirmed_rules = sorted(r.rule for r in relation.confirmed_at_reasons) + drift = current_rules != confirmed_rules + + return { + "id": relation.id, + "source": relation.source_finding_id, + "target": relation.target_finding_id, + "value": relation.weight, + "status": status_value, + "drift": drift, + "reasons": [r.rule for r in relation.reasons], + "relation_type": relation.llm_relation_type, + "rationale": relation.llm_rationale, + } +``` + +- [ ] **Step 2: Verify the module still imports cleanly** + +Run: `cd packages/web/backend && python -c "from app.services.chain_dto import relation_to_link_dict; print('OK')"` +Expected: `OK` + +- [ ] **Step 3: Commit** + +```bash +git add packages/web/backend/app/services/chain_dto.py +git commit -m "feat(chain): add relation_to_link_dict DTO with drift detection" +``` + +--- + +## Task 2: Backend — `subgraph_for_engagement` service method + +**Files:** +- Modify: `packages/web/backend/app/services/chain_service.py` + +Adds the service method that queries the store for findings + relations scoped to an engagement, applies severity/status filters, enforces max_nodes cap, and returns the force-graph-shaped response. + +- [ ] **Step 1: Add imports at top of chain_service.py** + +Add these imports to the existing import block: + +```python +from app.services.chain_dto import relation_to_link_dict +``` + +- [ ] **Step 2: Add `subgraph_for_engagement` method to `ChainService`** + +```python + async def subgraph_for_engagement( + self, + session: AsyncSession, + *, + user_id: uuid.UUID, + engagement_id: str, + severities: set[str] | None = None, + statuses: set[str] | None = None, + max_nodes: int = 500, + seed_finding_id: str | None = None, + hops: int = 2, + format: str = "force-graph", + ) -> dict[str, Any]: + """Build a filtered subgraph for one engagement. + + Returns a dict with 'graph' (force-graph or canonical shape) + and 'meta' (total_findings, rendered_findings, filtered, generation). + """ + from opentools.chain.config import get_chain_config + from opentools.chain.query.graph_cache import GraphCache + from opentools.chain.query.adapters import to_canonical_json, to_force_graph + from opentools.chain.types import RelationStatus + + from sqlalchemy import select, func + from app.models import Finding, ChainFindingRelation + + store = chain_store_from_session(session) + await store.initialize() + + # Count total findings in engagement (for meta) + total_stmt = select(func.count()).select_from(Finding).where( + Finding.engagement_id == engagement_id, + Finding.user_id == user_id, + Finding.deleted_at.is_(None), + ) + total_result = await session.execute(total_stmt) + total_findings = total_result.scalar() or 0 + + # Fetch findings for this engagement, applying severity filter + finding_stmt = select(Finding).where( + Finding.engagement_id == engagement_id, + Finding.user_id == user_id, + Finding.deleted_at.is_(None), + ) + if severities: + finding_stmt = finding_stmt.where(Finding.severity.in_(severities)) + finding_stmt = finding_stmt.limit(max_nodes) + + finding_result = await session.execute(finding_stmt) + findings = list(finding_result.scalars().all()) + finding_ids = {f.id for f in findings} + + if not finding_ids: + empty_graph = {"nodes": [], "links": []} if format == "force-graph" else {"schema_version": "1.0", "nodes": [], "edges": [], "metadata": {}} + return { + "graph": empty_graph, + "meta": { + "total_findings": total_findings, + "rendered_findings": 0, + "filtered": bool(severities) or total_findings > max_nodes, + "generation": 0, + }, + } + + # Default status filter + if statuses is None: + statuses = {"auto_confirmed", "user_confirmed", "candidate"} + + # Fetch relations where both endpoints are in finding_ids + rel_stmt = select(ChainFindingRelation).where( + ChainFindingRelation.user_id == user_id, + ChainFindingRelation.source_finding_id.in_(finding_ids), + ChainFindingRelation.target_finding_id.in_(finding_ids), + ChainFindingRelation.status.in_(statuses), + ) + rel_result = await session.execute(rel_stmt) + relations_orm = list(rel_result.scalars().all()) + + # Build nodes + nodes = [ + { + "id": f.id, + "name": f.title, + "severity": f.severity, + "tool": f.tool, + "phase": f.phase, + } + for f in findings + ] + + # Build links via DTO + from opentools.chain.models import FindingRelation as DomainRelation, RelationReason + from opentools.chain.stores.postgres_async import _orm_to_relation + + links = [ + relation_to_link_dict(_orm_to_relation(r)) + for r in relations_orm + ] + + # Get latest generation from most recent linker run + from app.models import ChainLinkerRun + gen_stmt = ( + select(ChainLinkerRun.generation) + .where(ChainLinkerRun.user_id == user_id) + .order_by(ChainLinkerRun.started_at.desc()) + .limit(1) + ) + gen_result = await session.execute(gen_stmt) + generation = gen_result.scalar() or 0 + + if format == "force-graph": + graph = {"nodes": nodes, "links": links} + else: + graph = { + "schema_version": "1.0", + "nodes": [{"id": n["id"], "type": "finding", "severity": n["severity"], "tool": n["tool"], "title": n["name"]} for n in nodes], + "edges": [{"source": l["source"], "target": l["target"], "weight": l["value"], "status": l["status"], "symmetric": False, "reasons": l["reasons"], "relation_type": l["relation_type"], "rationale": l["rationale"]} for l in links], + "metadata": {"generation": generation, "max_weight": max((l["value"] for l in links), default=0)}, + } + + return { + "graph": graph, + "meta": { + "total_findings": total_findings, + "rendered_findings": len(findings), + "filtered": bool(severities) or len(findings) < total_findings, + "generation": generation, + }, + } +``` + +- [ ] **Step 3: Verify the module still imports** + +Run: `cd packages/web/backend && python -c "from app.services.chain_service import ChainService; print('OK')"` +Expected: `OK` + +- [ ] **Step 4: Commit** + +```bash +git add packages/web/backend/app/services/chain_service.py +git commit -m "feat(chain): add subgraph_for_engagement service method" +``` + +--- + +## Task 3: Backend — `update_relation_status` service method + +**Files:** +- Modify: `packages/web/backend/app/services/chain_service.py` + +Adds the service method for edge curation — updates relation status to `user_confirmed` or `user_rejected`, snapshots `confirmed_at_reasons_json` on confirm. + +- [ ] **Step 1: Add `update_relation_status` method to `ChainService`** + +```python + async def update_relation_status( + self, + session: AsyncSession, + *, + user_id: uuid.UUID, + relation_id: str, + new_status: str, + ) -> dict[str, Any] | None: + """Update a relation's status for edge curation. + + Only 'user_confirmed' and 'user_rejected' are valid. + On confirm, snapshots current reasons_json into confirmed_at_reasons_json. + Returns the updated relation dict, or None if not found. + """ + from sqlalchemy import select, update + from app.models import ChainFindingRelation + from datetime import datetime, timezone + from opentools.chain.stores.postgres_async import _orm_to_relation + from app.services.chain_dto import relation_to_dict + + # Fetch the relation, scoped to user + stmt = select(ChainFindingRelation).where( + ChainFindingRelation.id == relation_id, + ChainFindingRelation.user_id == user_id, + ) + result = await session.execute(stmt) + relation = result.scalar_one_or_none() + if relation is None: + return None + + # Update status + relation.status = new_status + relation.updated_at = datetime.now(timezone.utc) + + # On confirm, snapshot current reasons for drift detection + if new_status == "user_confirmed": + relation.confirmed_at_reasons_json = relation.reasons_json + + session.add(relation) + await session.commit() + await session.refresh(relation) + + return relation_to_dict(_orm_to_relation(relation)) +``` + +- [ ] **Step 2: Verify import** + +Run: `cd packages/web/backend && python -c "from app.services.chain_service import ChainService; print('OK')"` +Expected: `OK` + +- [ ] **Step 3: Commit** + +```bash +git add packages/web/backend/app/services/chain_service.py +git commit -m "feat(chain): add update_relation_status for edge curation" +``` + +--- + +## Task 4: Backend — Subgraph and curation route endpoints + +**Files:** +- Modify: `packages/web/backend/app/routes/chain.py` + +Adds `GET /api/chain/subgraph` and `PATCH /api/chain/relations/{relation_id}`. + +- [ ] **Step 1: Add new Pydantic models for the endpoints** + +Add to `packages/web/backend/app/routes/chain.py`, after the existing model classes: + +```python +class SubgraphMeta(BaseModel): + total_findings: int + rendered_findings: int + filtered: bool + generation: int + + +class SubgraphResponse(BaseModel): + graph: dict + meta: SubgraphMeta + + +class RelationStatusUpdate(BaseModel): + status: str +``` + +- [ ] **Step 2: Add the subgraph endpoint** + +```python +@router.get("/subgraph", response_model=SubgraphResponse) +async def get_subgraph( + engagement_id: str, + severity: Optional[str] = None, + status: Optional[str] = None, + max_nodes: int = 500, + seed_finding_id: Optional[str] = None, + hops: int = 2, + format: str = "force-graph", + db: AsyncSession = Depends(get_db), + user: User = Depends(get_current_user), + service: ChainService = Depends(get_chain_service), +) -> SubgraphResponse: + severities = set(severity.split(",")) if severity else None + statuses = set(status.split(",")) if status else None + + result = await service.subgraph_for_engagement( + db, + user_id=user.id, + engagement_id=engagement_id, + severities=severities, + statuses=statuses, + max_nodes=max_nodes, + seed_finding_id=seed_finding_id, + hops=hops, + format=format, + ) + return SubgraphResponse( + graph=result["graph"], + meta=SubgraphMeta(**result["meta"]), + ) +``` + +- [ ] **Step 3: Add the relation curation endpoint** + +```python +@router.patch("/relations/{relation_id}") +async def update_relation( + relation_id: str, + body: RelationStatusUpdate, + db: AsyncSession = Depends(get_db), + user: User = Depends(get_current_user), + service: ChainService = Depends(get_chain_service), +): + valid_statuses = {"user_confirmed", "user_rejected"} + if body.status not in valid_statuses: + raise HTTPException( + status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, + detail=f"status must be one of: {', '.join(valid_statuses)}", + ) + + result = await service.update_relation_status( + db, user_id=user.id, relation_id=relation_id, new_status=body.status, + ) + if result is None: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="relation not found") + return result +``` + +- [ ] **Step 4: Verify the app starts** + +Run: `cd packages/web/backend && python -c "from app.main import app; print('OK')"` +Expected: `OK` + +- [ ] **Step 5: Commit** + +```bash +git add packages/web/backend/app/routes/chain.py +git commit -m "feat(chain): add subgraph and relation curation endpoints" +``` + +--- + +## Task 5: Backend — Subgraph endpoint tests + +**Files:** +- Create: `packages/web/backend/tests/test_chain_subgraph.py` + +- [ ] **Step 1: Write subgraph endpoint tests** + +Create `packages/web/backend/tests/test_chain_subgraph.py`: + +```python +"""Subgraph endpoint tests (Phase 3C.2).""" + +import uuid +from datetime import datetime, timezone + +import pytest + +from app.models import ChainFindingRelation, Engagement, Finding +from tests.conftest import test_session_factory + +NOW = datetime.now(timezone.utc) + + +async def _get_user_id(auth_client) -> uuid.UUID: + eng_resp = await auth_client.post("/api/v1/engagements", json={ + "name": "_uid_probe", "target": "127.0.0.1", "type": "pentest", + }) + assert eng_resp.status_code == 201 + eng_id = eng_resp.json()["id"] + async with test_session_factory() as session: + from sqlalchemy import select + from app.models import Engagement as Eng + result = await session.execute(select(Eng).where(Eng.id == eng_id)) + eng = result.scalar_one() + return eng.user_id + + +async def _seed_engagement(user_id, eng_id): + async with test_session_factory() as session: + session.add(Engagement( + id=eng_id, user_id=user_id, name="Test", target="10.0.0.0/24", + type="pentest", created_at=NOW, updated_at=NOW, + )) + await session.commit() + + +async def _seed_finding(user_id, eng_id, finding_id, severity="high", phase=None): + async with test_session_factory() as session: + session.add(Finding( + id=finding_id, user_id=user_id, engagement_id=eng_id, + tool="nmap", severity=severity, title=f"Finding {finding_id}", + phase=phase, created_at=NOW, + )) + await session.commit() + + +async def _seed_relation(user_id, src_id, tgt_id, rel_id, status="auto_confirmed", weight=0.8): + async with test_session_factory() as session: + session.add(ChainFindingRelation( + id=rel_id, user_id=user_id, source_finding_id=src_id, + target_finding_id=tgt_id, weight=weight, status=status, + symmetric=False, created_at=NOW, updated_at=NOW, + )) + await session.commit() + + +@pytest.mark.asyncio +async def test_subgraph_empty_engagement(auth_client): + """Engagement with no findings returns empty graph.""" + user_id = await _get_user_id(auth_client) + await _seed_engagement(user_id, "eng-empty") + + resp = await auth_client.get("/api/chain/subgraph?engagement_id=eng-empty") + assert resp.status_code == 200 + data = resp.json() + assert data["graph"]["nodes"] == [] + assert data["graph"]["links"] == [] + assert data["meta"]["total_findings"] == 0 + assert data["meta"]["rendered_findings"] == 0 + + +@pytest.mark.asyncio +async def test_subgraph_returns_nodes_and_links(auth_client): + """Seeded findings and relations appear in subgraph response.""" + user_id = await _get_user_id(auth_client) + await _seed_engagement(user_id, "eng-sub") + await _seed_finding(user_id, "eng-sub", "f-1", severity="critical") + await _seed_finding(user_id, "eng-sub", "f-2", severity="high") + await _seed_relation(user_id, "f-1", "f-2", "rel-1") + + resp = await auth_client.get("/api/chain/subgraph?engagement_id=eng-sub") + assert resp.status_code == 200 + data = resp.json() + assert len(data["graph"]["nodes"]) == 2 + assert len(data["graph"]["links"]) == 1 + link = data["graph"]["links"][0] + assert link["id"] == "rel-1" + assert link["source"] == "f-1" + assert link["target"] == "f-2" + assert "drift" in link + + +@pytest.mark.asyncio +async def test_subgraph_severity_filter(auth_client): + """Severity filter excludes non-matching findings.""" + user_id = await _get_user_id(auth_client) + await _seed_engagement(user_id, "eng-sev") + await _seed_finding(user_id, "eng-sev", "f-crit", severity="critical") + await _seed_finding(user_id, "eng-sev", "f-low", severity="low") + + resp = await auth_client.get("/api/chain/subgraph?engagement_id=eng-sev&severity=critical") + assert resp.status_code == 200 + data = resp.json() + assert len(data["graph"]["nodes"]) == 1 + assert data["graph"]["nodes"][0]["severity"] == "critical" + + +@pytest.mark.asyncio +async def test_subgraph_status_filter(auth_client): + """Status filter excludes non-matching relations.""" + user_id = await _get_user_id(auth_client) + await _seed_engagement(user_id, "eng-stat") + await _seed_finding(user_id, "eng-stat", "f-a") + await _seed_finding(user_id, "eng-stat", "f-b") + await _seed_relation(user_id, "f-a", "f-b", "rel-conf", status="auto_confirmed") + await _seed_relation(user_id, "f-b", "f-a", "rel-cand", status="candidate") + + # Only auto_confirmed + resp = await auth_client.get( + "/api/chain/subgraph?engagement_id=eng-stat&status=auto_confirmed" + ) + data = resp.json() + assert len(data["graph"]["links"]) == 1 + assert data["graph"]["links"][0]["status"] == "auto_confirmed" + + +@pytest.mark.asyncio +async def test_subgraph_max_nodes_cap(auth_client): + """max_nodes caps the number of returned findings.""" + user_id = await _get_user_id(auth_client) + await _seed_engagement(user_id, "eng-cap") + for i in range(10): + await _seed_finding(user_id, "eng-cap", f"f-cap-{i}") + + resp = await auth_client.get("/api/chain/subgraph?engagement_id=eng-cap&max_nodes=3") + data = resp.json() + assert len(data["graph"]["nodes"]) == 3 + assert data["meta"]["total_findings"] == 10 + assert data["meta"]["filtered"] is True + + +@pytest.mark.asyncio +async def test_subgraph_unauthenticated(client): + """Unauthenticated request returns 401.""" + resp = await client.get("/api/chain/subgraph?engagement_id=eng-x") + assert resp.status_code == 401 +``` + +- [ ] **Step 2: Run the tests** + +Run: `cd packages/web/backend && python -m pytest tests/test_chain_subgraph.py -v` +Expected: all tests PASS + +- [ ] **Step 3: Commit** + +```bash +git add packages/web/backend/tests/test_chain_subgraph.py +git commit -m "test(chain): subgraph endpoint tests — filters, cap, auth" +``` + +--- + +## Task 6: Backend — Curation endpoint tests + +**Files:** +- Create: `packages/web/backend/tests/test_chain_curation.py` + +- [ ] **Step 1: Write curation endpoint tests** + +Create `packages/web/backend/tests/test_chain_curation.py`: + +```python +"""Relation curation (PATCH) endpoint tests (Phase 3C.2).""" + +import uuid +from datetime import datetime, timezone + +import pytest + +from app.models import ChainFindingRelation, Engagement, Finding +from tests.conftest import test_session_factory + +NOW = datetime.now(timezone.utc) + + +async def _get_user_id(auth_client) -> uuid.UUID: + eng_resp = await auth_client.post("/api/v1/engagements", json={ + "name": "_uid_probe", "target": "127.0.0.1", "type": "pentest", + }) + assert eng_resp.status_code == 201 + eng_id = eng_resp.json()["id"] + async with test_session_factory() as session: + from sqlalchemy import select + from app.models import Engagement as Eng + result = await session.execute(select(Eng).where(Eng.id == eng_id)) + eng = result.scalar_one() + return eng.user_id + + +async def _seed_with_relation(user_id, rel_id="rel-cur", status="candidate"): + async with test_session_factory() as session: + session.add(Engagement( + id="eng-cur", user_id=user_id, name="Test", target="10.0.0.1", + type="pentest", created_at=NOW, updated_at=NOW, + )) + await session.flush() + session.add(Finding( + id="f-cur-1", user_id=user_id, engagement_id="eng-cur", + tool="nmap", severity="high", title="Finding 1", created_at=NOW, + )) + session.add(Finding( + id="f-cur-2", user_id=user_id, engagement_id="eng-cur", + tool="nuclei", severity="medium", title="Finding 2", created_at=NOW, + )) + await session.flush() + session.add(ChainFindingRelation( + id=rel_id, user_id=user_id, source_finding_id="f-cur-1", + target_finding_id="f-cur-2", weight=0.75, status=status, + symmetric=False, reasons_json=b'[{"rule":"shared_strong_entity","weight_contribution":0.5,"idf_factor":null,"details":{}}]', + created_at=NOW, updated_at=NOW, + )) + await session.commit() + + +@pytest.mark.asyncio +async def test_confirm_candidate(auth_client): + """Confirming a candidate relation succeeds.""" + user_id = await _get_user_id(auth_client) + await _seed_with_relation(user_id, "rel-c1", status="candidate") + + resp = await auth_client.patch( + "/api/chain/relations/rel-c1", + json={"status": "user_confirmed"}, + ) + assert resp.status_code == 200 + assert resp.json()["status"] == "user_confirmed" + + +@pytest.mark.asyncio +async def test_reject_candidate(auth_client): + """Rejecting a candidate relation succeeds.""" + user_id = await _get_user_id(auth_client) + await _seed_with_relation(user_id, "rel-c2", status="candidate") + + resp = await auth_client.patch( + "/api/chain/relations/rel-c2", + json={"status": "user_rejected"}, + ) + assert resp.status_code == 200 + assert resp.json()["status"] == "user_rejected" + + +@pytest.mark.asyncio +async def test_toggle_confirmed_to_rejected(auth_client): + """User can change from confirmed to rejected.""" + user_id = await _get_user_id(auth_client) + await _seed_with_relation(user_id, "rel-c3", status="user_confirmed") + + resp = await auth_client.patch( + "/api/chain/relations/rel-c3", + json={"status": "user_rejected"}, + ) + assert resp.status_code == 200 + assert resp.json()["status"] == "user_rejected" + + +@pytest.mark.asyncio +async def test_invalid_status_returns_422(auth_client): + """Setting auto_confirmed via PATCH returns 422.""" + user_id = await _get_user_id(auth_client) + await _seed_with_relation(user_id, "rel-c4") + + resp = await auth_client.patch( + "/api/chain/relations/rel-c4", + json={"status": "auto_confirmed"}, + ) + assert resp.status_code == 422 + + +@pytest.mark.asyncio +async def test_nonexistent_relation_returns_404(auth_client): + """Patching a nonexistent relation returns 404.""" + resp = await auth_client.patch( + "/api/chain/relations/rel-does-not-exist", + json={"status": "user_confirmed"}, + ) + assert resp.status_code == 404 + + +@pytest.mark.asyncio +async def test_confirm_snapshots_reasons(auth_client): + """Confirming snapshots reasons_json into confirmed_at_reasons_json.""" + user_id = await _get_user_id(auth_client) + await _seed_with_relation(user_id, "rel-c5", status="candidate") + + await auth_client.patch( + "/api/chain/relations/rel-c5", + json={"status": "user_confirmed"}, + ) + + # Verify in DB + async with test_session_factory() as session: + from sqlalchemy import select + result = await session.execute( + select(ChainFindingRelation).where(ChainFindingRelation.id == "rel-c5") + ) + rel = result.scalar_one() + assert rel.confirmed_at_reasons_json is not None + assert rel.confirmed_at_reasons_json == rel.reasons_json + + +@pytest.mark.asyncio +async def test_unauthenticated_returns_401(client): + """Unauthenticated curation request returns 401.""" + resp = await client.patch( + "/api/chain/relations/rel-x", + json={"status": "user_confirmed"}, + ) + assert resp.status_code == 401 +``` + +- [ ] **Step 2: Run the tests** + +Run: `cd packages/web/backend && python -m pytest tests/test_chain_curation.py -v` +Expected: all tests PASS + +- [ ] **Step 3: Commit** + +```bash +git add packages/web/backend/tests/test_chain_curation.py +git commit -m "test(chain): curation endpoint tests — transitions, validation, auth" +``` + +--- + +## Task 7: Frontend — Install `force-graph` and add route + +**Files:** +- Modify: `packages/web/frontend/package.json` (via npm) +- Modify: `packages/web/frontend/src/router/index.ts` +- Modify: `packages/web/frontend/src/views/EngagementDetailView.vue` + +- [ ] **Step 1: Install force-graph** + +Run: `cd packages/web/frontend && npm install force-graph` + +- [ ] **Step 2: Add the chain route to router** + +In `packages/web/frontend/src/router/index.ts`, add after the `finding-detail` route: + +```typescript + { path: '/engagements/:id/chain', name: 'engagement-chain', component: () => import('@/views/ChainGraphView.vue') }, +``` + +- [ ] **Step 3: Add "View Attack Chain" button to EngagementDetailView** + +In `packages/web/frontend/src/views/EngagementDetailView.vue`, add a button next to the existing Delete button in the header: + +Find the `