136 lines
4.3 KiB
Python
136 lines
4.3 KiB
Python
"""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,
|
|
}
|