AI Incident Severity Scoring

"""
AI Incident Severity Framework (AISF) scoring implementation.
 
Produces a composite severity score for AI security incidents
based on seven weighted dimensions.
"""
from dataclasses import dataclass
from enum import Enum
 
 
class SeverityLevel(Enum):
    INFORMATIONAL = "informational"  # 0.0 - 1.9
    LOW = "low"                      # 2.0 - 3.9
    MEDIUM = "medium"                # 4.0 - 5.9
    HIGH = "high"                    # 6.0 - 7.9
    CRITICAL = "critical"            # 8.0 - 10.0
 
 
@dataclass
class AISFDimension:
    """A single dimension of the AISF severity score."""
    name: str
    score: float  # 0.0 to 10.0
    weight: float  # relative weight in composite
    rationale: str  # explanation for the score
 
 
@dataclass
class AISFScore:
    """Complete AISF severity assessment."""
    dimensions: list[AISFDimension]
    composite_score: float
    severity_level: SeverityLevel
    summary: str
 
 
def classify_severity(score: float) -> SeverityLevel:
    if score < 2.0:
        return SeverityLevel.INFORMATIONAL
    if score < 4.0:
        return SeverityLevel.LOW
    if score < 6.0:
        return SeverityLevel.MEDIUM
    if score < 8.0:
        return SeverityLevel.HIGH
    return SeverityLevel.CRITICAL
 
 
def compute_aisf_score(dimensions: list[AISFDimension]) -> AISFScore:
    """
    Compute the composite AISF score from individual dimension scores.
 
    Uses weighted average with normalization.
    """
    total_weight = sum(d.weight for d in dimensions)
    if total_weight == 0:
        raise ValueError("Total weight must be positive")
 
    composite = sum(d.score * d.weight for d in dimensions) / total_weight
    composite = round(min(max(composite, 0.0), 10.0), 1)
 
    return AISFScore(
        dimensions=dimensions,
        composite_score=composite,
        severity_level=classify_severity(composite),
        summary=_generate_summary(dimensions, composite),
    )
 
 
def _generate_summary(dimensions: list[AISFDimension], composite: float) -> str:
    top_dims = sorted(dimensions, key=lambda d: d.score, reverse=True)[:3]
    drivers = ", ".join(f"{d.name} ({d.score})" for d in top_dims)
    level = classify_severity(composite)
    return f"Composite score {composite}/10.0 ({level.value}). Primary drivers: {drivers}."

def score_prompt_injection_exfiltration() -> AISFScore:
    """
    Example: Score a prompt injection incident where an attacker
    used indirect prompt injection via a RAG document to extract
    other users' conversation data from a customer support chatbot.
    """
    dimensions = [
        AISFDimension(
            name="Attack Sophistication",
            score=5.0,
            weight=0.10,
            rationale="Targeted indirect prompt injection requiring knowledge "
                      "of the RAG pipeline and output format",
        ),
        AISFDimension(
            name="Data Sensitivity",
            score=7.0,
            weight=0.20,
            rationale="Customer support conversations contain PII including "
                      "names, email addresses, and account details",
        ),
        AISFDimension(
            name="Model Integrity",
            score=3.0,
            weight=0.20,
            rationale="Model itself was not modified; attack exploited the "
                      "application architecture",
        ),
        AISFDimension(
            name="Blast Radius",
            score=6.0,
            weight=0.15,
            rationale="Approximately 2,000 customer conversations were "
                      "potentially accessible through the vulnerability",
        ),
        AISFDimension(
            name="Reversibility",
            score=8.0,
            weight=0.15,
            rationale="Extracted data cannot be recalled; affected customers "
                      "must be notified",
        ),
        AISFDimension(
            name="Regulatory Exposure",
            score=7.0,
            weight=0.10,
            rationale="PII exposure triggers GDPR Article 33 notification "
                      "requirement (72-hour window)",
        ),
        AISFDimension(
            name="Reputational Impact",
            score=6.0,
            weight=0.10,
            rationale="Customer trust impact; potential media coverage of "
                      "AI chatbot leaking customer data",
        ),
    ]
    return compute_aisf_score(dimensions)
 
 
# Run the example
result = score_prompt_injection_exfiltration()
# Expected composite score: approximately 5.8 (MEDIUM-HIGH)
print(f"Severity: {result.severity_level.value}")
print(f"Score: {result.composite_score}")
print(f"Summary: {result.summary}")

def score_model_backdoor_discovery() -> AISFScore:
    """
    Example: Score the discovery of a backdoor in a third-party
    model that was deployed in a financial fraud detection system.
    The backdoor causes the model to classify certain fraudulent
    transactions as legitimate when a specific pattern is present.
    """
    dimensions = [
        AISFDimension(
            name="Attack Sophistication",
            score=9.0,
            weight=0.10,
            rationale="Training-time backdoor attack requiring access to the "
                      "model training pipeline or supply chain",
        ),
        AISFDimension(
            name="Data Sensitivity",
            score=8.0,
            weight=0.20,
            rationale="Financial transaction data with PII and account details; "
                      "fraudulent transactions enabled by the backdoor",
        ),
        AISFDimension(
            name="Model Integrity",
            score=9.0,
            weight=0.20,
            rationale="Model weights contain a persistent backdoor; the model "
                      "cannot be trusted for any predictions",
        ),
        AISFDimension(
            name="Blast Radius",
            score=8.0,
            weight=0.15,
            rationale="All transactions processed by this model since "
                      "deployment are potentially affected",
        ),
        AISFDimension(
            name="Reversibility",
            score=6.0,
            weight=0.15,
            rationale="Model can be replaced, but all historical predictions "
                      "must be re-evaluated; financial losses may not be recoverable",
        ),
        AISFDimension(
            name="Regulatory Exposure",
            score=8.0,
            weight=0.10,
            rationale="Financial services regulation requires reporting; "
                      "potential regulatory action for inadequate model validation",
        ),
        AISFDimension(
            name="Reputational Impact",
            score=7.0,
            weight=0.10,
            rationale="Trust in AI-based fraud detection undermined; "
                      "customers affected by enabled fraud",
        ),
    ]
    return compute_aisf_score(dimensions)

def automated_initial_score(alert: dict) -> AISFScore:
    """
    Produce an automated preliminary severity score from alert data.
 
    This is intended for initial triage only. Human review is required
    for any incident scoring MEDIUM or above.
 
    Expected alert fields:
    - alert_type: str (e.g., "prompt_injection", "model_drift", "data_leak")
    - users_affected: int
    - data_classification: str ("public", "internal", "confidential", "restricted")
    - model_modified: bool
    - reversible: bool
    - external_visibility: bool
    """
    # Map data classification to sensitivity score
    sensitivity_map = {
        "public": 1.0,
        "internal": 3.0,
        "confidential": 6.0,
        "restricted": 9.0,
    }
 
    # Map user count to blast radius score
    users = alert.get("users_affected", 0)
    if users <= 1:
        blast_score = 2.0
    elif users <= 100:
        blast_score = 4.0
    elif users <= 10000:
        blast_score = 6.0
    elif users <= 1000000:
        blast_score = 8.0
    else:
        blast_score = 10.0
 
    dimensions = [
        AISFDimension("Attack Sophistication", 5.0, 0.10, "Default pending investigation"),
        AISFDimension(
            "Data Sensitivity",
            sensitivity_map.get(alert.get("data_classification", "internal"), 5.0),
            0.20,
            f"Data classification: {alert.get('data_classification', 'unknown')}",
        ),
        AISFDimension(
            "Model Integrity",
            8.0 if alert.get("model_modified") else 3.0,
            0.20,
            "Model modification detected" if alert.get("model_modified") else "No model modification",
        ),
        AISFDimension("Blast Radius", blast_score, 0.15, f"{users} users affected"),
        AISFDimension(
            "Reversibility",
            3.0 if alert.get("reversible") else 7.0,
            0.15,
            "Reversible" if alert.get("reversible") else "Not easily reversible",
        ),
        AISFDimension("Regulatory Exposure", 5.0, 0.10, "Default pending legal review"),
        AISFDimension(
            "Reputational Impact",
            7.0 if alert.get("external_visibility") else 3.0,
            0.10,
            "Externally visible" if alert.get("external_visibility") else "Internal only",
        ),
    ]
 
    return compute_aisf_score(dimensions)

from datetime import datetime
 
 
@dataclass
class ScoreRevision:
    """A point-in-time severity score revision."""
    timestamp: str
    score: AISFScore
    analyst: str
    reason: str
 
 
def track_severity_evolution(
    revisions: list[ScoreRevision],
) -> dict:
    """
    Analyze how the severity score evolved during the investigation.
 
    Significant score changes indicate that initial assumptions
    were revised as new evidence emerged.
    """
    if not revisions:
        return {"error": "No revisions to analyze"}
 
    scores = [r.score.composite_score for r in revisions]
    return {
        "initial_score": scores[0],
        "final_score": scores[-1],
        "peak_score": max(scores),
        "score_delta": scores[-1] - scores[0],
        "revision_count": len(revisions),
        "escalations": sum(
            1 for i in range(1, len(scores)) if scores[i] > scores[i-1]
        ),
        "deescalations": sum(
            1 for i in range(1, len(scores)) if scores[i] < scores[i-1]
        ),
    }