"""D3FEND endpoints — defensive technique listings, mappings, and import trigger.""" import logging from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.orm import Session from app.database import get_db from app.dependencies.auth import get_current_user, require_role from app.models.user import User from app.models.technique import Technique from app.models.defensive_technique import DefensiveTechnique, DefensiveTechniqueMapping from app.services.d3fend_import_service import ( import_d3fend_techniques, import_d3fend_mappings, get_defenses_for_technique, ) logger = logging.getLogger(__name__) router = APIRouter(prefix="/d3fend", tags=["d3fend"]) # --------------------------------------------------------------------------- # GET /d3fend — List all defensive techniques # --------------------------------------------------------------------------- @router.get("") def list_defensive_techniques( tactic: Optional[str] = Query(None), search: Optional[str] = Query(None), offset: int = Query(0, ge=0), limit: int = Query(50, ge=1, le=200), db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """List all D3FEND defensive techniques with optional filters.""" query = db.query(DefensiveTechnique) if tactic: query = query.filter(DefensiveTechnique.tactic == tactic) if search: pattern = f"%{search}%" query = query.filter( DefensiveTechnique.name.ilike(pattern) | DefensiveTechnique.d3fend_id.ilike(pattern) ) total = query.count() items = query.order_by(DefensiveTechnique.d3fend_id).offset(offset).limit(limit).all() return { "total": total, "offset": offset, "limit": limit, "items": [ { "id": str(dt.id), "d3fend_id": dt.d3fend_id, "name": dt.name, "description": dt.description, "tactic": dt.tactic, "d3fend_url": dt.d3fend_url, } for dt in items ], } # --------------------------------------------------------------------------- # GET /d3fend/tactics — List all D3FEND tactics # --------------------------------------------------------------------------- @router.get("/tactics") def list_d3fend_tactics( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Return a list of all D3FEND tactics with counts.""" from sqlalchemy import func rows = ( db.query(DefensiveTechnique.tactic, func.count(DefensiveTechnique.id)) .group_by(DefensiveTechnique.tactic) .order_by(DefensiveTechnique.tactic) .all() ) return [{"tactic": tactic or "Unknown", "count": count} for tactic, count in rows] # --------------------------------------------------------------------------- # GET /d3fend/for-technique/{mitre_id} — Defenses for a technique # --------------------------------------------------------------------------- @router.get("/for-technique/{mitre_id}") def get_defenses_for_attack_technique( mitre_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Get all D3FEND defensive techniques mapped to a given ATT&CK technique.""" technique = db.query(Technique).filter(Technique.mitre_id == mitre_id).first() if not technique: raise HTTPException(status_code=404, detail=f"Technique {mitre_id} not found") defenses = get_defenses_for_technique(db, technique.id) return { "mitre_id": mitre_id, "technique_name": technique.name, "defenses": defenses, "total": len(defenses), } # --------------------------------------------------------------------------- # POST /d3fend/import — Trigger D3FEND import (admin only) # --------------------------------------------------------------------------- @router.post("/import") def trigger_d3fend_import( db: Session = Depends(get_db), current_user: User = Depends(require_role("admin")), ): """Import D3FEND techniques and ATT&CK mappings. Admin only.""" tech_result = import_d3fend_techniques(db) mapping_result = import_d3fend_mappings(db) return { "techniques": tech_result, "mappings": mapping_result, }