"""Scoring endpoints — technique, tactic, threat actor, and organization scores. Provides granular scoring with breakdowns and configurable weights. """ from typing import Optional from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel 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.threat_actor import ThreatActor from app.config import settings from app.services.scoring_service import ( calculate_technique_score, calculate_tactic_score, calculate_actor_coverage_score, calculate_organization_score, get_score_history, ) router = APIRouter(prefix="/scores", tags=["scores"]) # ── GET /scores/technique/{mitre_id} ───────────────────────────────── @router.get("/technique/{mitre_id}") def score_technique( mitre_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Get detailed score with breakdown for a specific technique.""" technique = ( db.query(Technique) .filter(Technique.mitre_id == mitre_id) .first() ) if not technique: raise HTTPException(status_code=404, detail="Technique not found") result = calculate_technique_score(technique, db) return { "mitre_id": technique.mitre_id, "name": technique.name, "tactic": technique.tactic, "status_global": technique.status_global.value if technique.status_global else None, **result, } # ── GET /scores/tactic/{tactic} ────────────────────────────────────── @router.get("/tactic/{tactic}") def score_tactic( tactic: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Get average score for a tactic.""" return calculate_tactic_score(tactic, db) # ── GET /scores/threat-actor/{id} ──────────────────────────────────── @router.get("/threat-actor/{actor_id}") def score_threat_actor( actor_id: str, db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Get coverage score against a specific threat actor.""" actor = db.query(ThreatActor).filter(ThreatActor.id == actor_id).first() if not actor: raise HTTPException(status_code=404, detail="Threat actor not found") return calculate_actor_coverage_score(actor_id, db) # ── GET /scores/organization ───────────────────────────────────────── @router.get("/organization") def score_organization( db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Get the overall organization security score (cached for 5 min).""" from app.services.score_cache import get_organization_score_cached return get_organization_score_cached(db) # ── GET /scores/history ────────────────────────────────────────────── @router.get("/history") def score_history( period: str = Query("90d", pattern="^(30d|90d|1y)$"), db: Session = Depends(get_db), current_user: User = Depends(get_current_user), ): """Get historical score data points (weekly).""" return get_score_history(db, period) # ── GET /scores/config ─────────────────────────────────────────────── @router.get("/config") def get_scoring_config( current_user: User = Depends(require_role("admin")), ): """Get current scoring weights (admin only).""" return { "weights": { "tests": settings.SCORING_WEIGHT_TESTS, "detection_rules": settings.SCORING_WEIGHT_DETECTION_RULES, "d3fend": settings.SCORING_WEIGHT_D3FEND, "freshness": settings.SCORING_WEIGHT_FRESHNESS, "platform_diversity": settings.SCORING_WEIGHT_PLATFORM_DIVERSITY, }, "total": ( settings.SCORING_WEIGHT_TESTS + settings.SCORING_WEIGHT_DETECTION_RULES + settings.SCORING_WEIGHT_D3FEND + settings.SCORING_WEIGHT_FRESHNESS + settings.SCORING_WEIGHT_PLATFORM_DIVERSITY ), } # ── PATCH /scores/config ───────────────────────────────────────────── class ScoringConfigUpdate(BaseModel): tests: Optional[int] = None detection_rules: Optional[int] = None d3fend: Optional[int] = None freshness: Optional[int] = None platform_diversity: Optional[int] = None @router.patch("/config") def update_scoring_config( payload: ScoringConfigUpdate, current_user: User = Depends(require_role("admin")), ): """Update scoring weights (admin only). Note: Since we're using Opcion A (env vars / Settings), changes are applied at runtime but won't persist across restarts unless the .env file is also updated. For production, consider migrating to Option B (database table). """ if payload.tests is not None: settings.SCORING_WEIGHT_TESTS = payload.tests if payload.detection_rules is not None: settings.SCORING_WEIGHT_DETECTION_RULES = payload.detection_rules if payload.d3fend is not None: settings.SCORING_WEIGHT_D3FEND = payload.d3fend if payload.freshness is not None: settings.SCORING_WEIGHT_FRESHNESS = payload.freshness if payload.platform_diversity is not None: settings.SCORING_WEIGHT_PLATFORM_DIVERSITY = payload.platform_diversity # Weights changed — bust the score cache from app.services.score_cache import invalidate invalidate() return { "message": "Scoring config updated", "weights": { "tests": settings.SCORING_WEIGHT_TESTS, "detection_rules": settings.SCORING_WEIGHT_DETECTION_RULES, "d3fend": settings.SCORING_WEIGHT_D3FEND, "freshness": settings.SCORING_WEIGHT_FRESHNESS, "platform_diversity": settings.SCORING_WEIGHT_PLATFORM_DIVERSITY, }, "total": ( settings.SCORING_WEIGHT_TESTS + settings.SCORING_WEIGHT_DETECTION_RULES + settings.SCORING_WEIGHT_D3FEND + settings.SCORING_WEIGHT_FRESHNESS + settings.SCORING_WEIGHT_PLATFORM_DIVERSITY ), }