Data Breach Response

#!/usr/bin/env python3
"""
Breach Severity Calculator

Calculates ENISA breach severity score using the formula SE = (DPC x EI) + CB.
Determines severity verdict (LOW/MEDIUM/HIGH/VERY HIGH) and notification
obligations under GDPR, CCPA, HIPAA, PCI DSS, and NIS2.

Usage:
    python breach_severity_calculator.py --dpc 3 --ei 0.75 --confidentiality 0.5 --malicious
    python breach_severity_calculator.py --dpc 2 --ei 0.5 --confidentiality 0.5 --json
    python breach_severity_calculator.py --dpc 3 --ei 1.0 --confidentiality 0.5 --t0 "2026-04-10T08:00:00" --json
    python breach_severity_calculator.py --template
"""

import argparse
import json
import sys
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional, Tuple


DPC_LABELS: Dict[int, str] = {1: "Simple demographic data", 2: "Behavioral or financial data", 3: "Sensitive personal data", 4: "Special category or highly sensitive data"}
EI_LABELS: Dict[float, str] = {0.25: "Negligible", 0.50: "Limited", 0.75: "Significant", 1.00: "Maximum (direct identifiers)"}
GDPR_HOURS, NIS2_EW_HOURS, NIS2_HOURS, PCI_HOURS = 72, 24, 72, 24


def calculate_cb(
    confidentiality: float,
    integrity: float,
    availability: float,
    malicious: bool,
) -> Tuple[float, Dict[str, float]]:
    """Calculate Circumstances of Breach (CB) component."""
    components: Dict[str, float] = {
        "confidentiality_loss": confidentiality,
        "integrity_loss": integrity,
        "availability_loss": availability,
        "malicious_intent": 0.5 if malicious else 0.0,
    }
    # CB is the sum of loss types minus overlap (max 1.0 practical cap)
    # but ENISA allows it to go negative with encryption offset
    cb = sum(components.values())
    return cb, components


def calculate_severity(
    dpc: int, ei: float, cb: float
) -> Tuple[float, str, str]:
    """Calculate ENISA severity score and return (score, verdict, description)."""
    se = (dpc * ei) + cb
    se = round(se, 2)

    if se < 2:
        verdict = "LOW"
        description = "Internal documentation only. No SA or data subject notification required."
    elif se < 3:
        verdict = "MEDIUM"
        description = "Notify supervisory authority within 72 hours (GDPR Art. 33)."
    elif se < 4:
        verdict = "HIGH"
        description = "Notify SA within 72h + notify individual data subjects without undue delay (GDPR Art. 34)."
    else:
        verdict = "VERY HIGH"
        description = "Notify SA + data subjects + consider public notice. Activate crisis management."

    return se, verdict, description


def get_notification_obligations(verdict: str) -> List[Dict[str, str]]:
    """Return notification obligations based on severity verdict."""
    obligations: List[Dict[str, str]] = []

    # Internal documentation always required
    obligations.append({
        "regulation": "GDPR Art. 33(5)",
        "action": "Document breach in internal breach register",
        "deadline": "Immediately",
        "required": "Always",
    })

    if verdict in ("MEDIUM", "HIGH", "VERY HIGH"):
        obligations.append({
            "regulation": "GDPR Art. 33",
            "action": "Notify supervisory authority",
            "deadline": "72 hours from T0",
            "required": "Yes",
        })

    if verdict in ("HIGH", "VERY HIGH"):
        obligations.append({
            "regulation": "GDPR Art. 34",
            "action": "Notify affected data subjects",
            "deadline": "Without undue delay",
            "required": "Yes — high risk to rights and freedoms",
        })

    if verdict == "VERY HIGH":
        obligations.append({
            "regulation": "GDPR Art. 34 / national law",
            "action": "Consider public communication",
            "deadline": "Without undue delay",
            "required": "When individual notification is disproportionate effort",
        })

    # Cross-regulation obligations (always assess)
    obligations.append({
        "regulation": "CCPA",
        "action": "Notify affected California residents",
        "deadline": "Most expedient time possible",
        "required": "If unencrypted PI of CA residents compromised",
    })
    obligations.append({
        "regulation": "NIS2 Art. 23",
        "action": "Early warning to CSIRT + full notification",
        "deadline": "24h early warning + 72h notification",
        "required": "If essential/important entity with significant incident",
    })
    obligations.append({
        "regulation": "HIPAA",
        "action": "Notify HHS and affected individuals",
        "deadline": "60 days; if >500: media notification",
        "required": "If unsecured PHI compromised",
    })
    obligations.append({
        "regulation": "PCI DSS",
        "action": "Notify card brands",
        "deadline": "24 hours",
        "required": "If cardholder data compromised",
    })

    return obligations


def calculate_countdown(t0_str: str) -> Dict[str, Any]:
    """Calculate time remaining for various deadlines from T0."""
    try:
        t0 = datetime.fromisoformat(t0_str)
    except ValueError:
        return {"error": f"Invalid T0 format: {t0_str}. Use ISO: YYYY-MM-DDTHH:MM:SS"}
    now = datetime.now()
    elapsed_hours = (now - t0).total_seconds() / 3600
    deadlines: Dict[str, Any] = {}
    dl_defs = [("gdpr_72h", GDPR_HOURS, 12), ("nis2_24h_ew", NIS2_EW_HOURS, 6),
               ("nis2_72h", NIS2_HOURS, 12), ("pci_24h", PCI_HOURS, 6)]
    for name, hours, urgent_thresh in dl_defs:
        deadline_time = t0 + timedelta(hours=hours)
        remaining = (deadline_time - now).total_seconds() / 3600
        status = "EXPIRED" if remaining <= 0 else "URGENT" if remaining <= urgent_thresh else "OK"
        deadlines[name] = {"deadline": deadline_time.isoformat(), "remaining_hours": round(remaining, 1),
                           "expired": remaining <= 0, "status": status}
    return {"t0": t0.isoformat(), "current_time": now.isoformat(),
            "elapsed_hours": round(elapsed_hours, 1), "deadlines": deadlines}


def generate_template() -> Dict[str, Any]:
    """Generate input template."""
    return {"dpc": {"value": 0, "options": "1=Simple demographic, 2=Behavioral/financial, 3=Sensitive, 4=Special category"},
            "ei": {"value": 0, "options": "0.25=Negligible, 0.5=Limited, 0.75=Significant, 1.0=Maximum"},
            "confidentiality_loss": 0, "integrity_loss": 0, "availability_loss": 0,
            "malicious_intent": False, "t0": "YYYY-MM-DDTHH:MM:SS"}


def format_human(
    dpc: int, ei: float, cb: float, cb_components: Dict[str, float],
    se: float, verdict: str, description: str,
    obligations: List[Dict[str, str]], countdown: Optional[Dict[str, Any]],
) -> str:
    """Format results for human-readable output."""
    lines: List[str] = ["=" * 65, "DATA BREACH SEVERITY ASSESSMENT", "=" * 65, "",
        f"  DPC: {dpc} ({DPC_LABELS.get(dpc, '?')})  |  EI: {ei} ({EI_LABELS.get(ei, '?')})", ""]
    lines.append("  CB components: " + ", ".join(f"{k.replace('_',' ').title()}: {v}" for k, v in cb_components.items()))
    lines.append(f"\n  SE = ({dpc} x {ei}) + {round(cb, 2)} = {se}")
    lines.append(f"\n  VERDICT: {verdict}\n  {description}\n")
    if countdown and "error" not in countdown:
        lines.append(f"DEADLINES (T0: {countdown['t0']}, elapsed: {countdown['elapsed_hours']}h):")
        for name, dl in countdown["deadlines"].items():
            icon = "!!!" if dl["status"] == "EXPIRED" else ">>>" if dl["status"] == "URGENT" else "   "
            lines.append(f"  {icon} {name}: {dl['remaining_hours']}h [{dl['status']}]")
        lines.append("")
    lines.append("NOTIFICATION OBLIGATIONS:")
    for ob in obligations:
        lines.append(f"  [{ob['regulation']}] {ob['action']} | Deadline: {ob['deadline']} | {ob['required']}")
    lines.append("=" * 65)
    return "\n".join(lines)


def format_json_output(
    dpc: int, ei: float, cb: float, cb_components: Dict[str, float],
    se: float, verdict: str, description: str,
    obligations: List[Dict[str, str]],
    countdown: Optional[Dict[str, Any]],
) -> Dict[str, Any]:
    """Format results as JSON."""
    result: Dict[str, Any] = {
        "calculated": datetime.now().isoformat(),
        "parameters": {
            "dpc": {"value": dpc, "label": DPC_LABELS.get(dpc, "Unknown")},
            "ei": {"value": ei, "label": EI_LABELS.get(ei, "Unknown")},
            "cb": {"value": round(cb, 2), "components": cb_components},
        },
        "formula": f"SE = ({dpc} x {ei}) + {round(cb, 2)} = {se}",
        "severity_score": se,
        "verdict": verdict,
        "description": description,
        "notification_obligations": obligations,
    }
    if countdown:
        result["countdown"] = countdown
    return result


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Breach Severity Calculator — ENISA methodology"
    )
    parser.add_argument("--dpc", type=int, choices=[1, 2, 3, 4],
                        help="Data Processing Context (1-4)")
    parser.add_argument("--ei", type=float,
                        help="Ease of Identification (0.25, 0.5, 0.75, 1.0)")
    parser.add_argument("--confidentiality", type=float, default=0.0,
                        help="Confidentiality loss (0, 0.25, 0.5)")
    parser.add_argument("--integrity", type=float, default=0.0,
                        help="Integrity loss (0, 0.25, 0.5)")
    parser.add_argument("--availability", type=float, default=0.0,
                        help="Availability loss (0, 0.25, 0.5)")
    parser.add_argument("--malicious", action="store_true",
                        help="Breach involved malicious intent")
    parser.add_argument("--t0", type=str,
                        help="T0 timestamp (ISO format) for deadline calculation")
    parser.add_argument("--template", action="store_true",
                        help="Generate input template")
    parser.add_argument("--json", action="store_true",
                        help="Output in JSON format")

    args = parser.parse_args()

    if args.template:
        print(json.dumps(generate_template(), indent=2))
        return

    if args.dpc is None or args.ei is None:
        parser.error("--dpc and --ei are required (or use --template)")

    # Validate EI
    valid_ei = [0.25, 0.5, 0.75, 1.0]
    if args.ei not in valid_ei:
        print(f"Error: --ei must be one of {valid_ei}", file=sys.stderr)
        sys.exit(1)

    # Validate loss values
    valid_loss = [0.0, 0.25, 0.5]
    for name, val in [("confidentiality", args.confidentiality),
                      ("integrity", args.integrity),
                      ("availability", args.availability)]:
        if val not in valid_loss:
            print(f"Error: --{name} must be one of {valid_loss}", file=sys.stderr)
            sys.exit(1)

    # Calculate
    cb, cb_components = calculate_cb(
        args.confidentiality, args.integrity, args.availability, args.malicious
    )
    se, verdict, description = calculate_severity(args.dpc, args.ei, cb)
    obligations = get_notification_obligations(verdict)

    countdown = None
    if args.t0:
        countdown = calculate_countdown(args.t0)

    if args.json:
        result = format_json_output(
            args.dpc, args.ei, cb, cb_components,
            se, verdict, description, obligations, countdown
        )
        print(json.dumps(result, indent=2))
    else:
        output = format_human(
            args.dpc, args.ei, cb, cb_components,
            se, verdict, description, obligations, countdown
        )
        print(output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Breach Timeline Tracker

Tracks data breach response timeline from T0 (moment of awareness).
Records events, calculates time remaining for regulatory deadlines,
and generates status dashboards.

Usage:
    python breach_timeline_tracker.py init --breach-id "BR-2026-001" --t0 "2026-04-10T08:00:00" --description "Unauthorized access"
    python breach_timeline_tracker.py event --timeline breach.json --action "Containment team activated" --category containment
    python breach_timeline_tracker.py status --timeline breach.json
    python breach_timeline_tracker.py deadlines --timeline breach.json --json
"""

import argparse
import json
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List, Optional


DEADLINE_DEFINITIONS: Dict[str, Dict[str, Any]] = {
    "gdpr_72h_sa": {"label": "GDPR Art. 33 — SA Notification", "hours": 72, "description": "Notify SA of personal data breach"},
    "nis2_24h_ew": {"label": "NIS2 Art. 23 — Early Warning", "hours": 24, "description": "Submit early warning to CSIRT"},
    "nis2_72h": {"label": "NIS2 Art. 23 — Full Notification", "hours": 72, "description": "Full incident notification to CSIRT"},
    "pci_24h": {"label": "PCI DSS — Card Brand Notification", "hours": 24, "description": "Notify card brands"},
    "dpa_24h": {"label": "DPA Contractual — 24h Processor", "hours": 24, "description": "Processor notifies controller (24h)"},
    "dpa_48h": {"label": "DPA Contractual — 48h Processor", "hours": 48, "description": "Processor notifies controller (48h)"},
}
VALID_CATEGORIES: List[str] = ["detection", "containment", "assessment", "notification", "remediation", "communication", "documentation", "other"]
RESPONSE_CHECKLIST: List[Dict[str, str]] = [
    {"id": "C1", "action": "Breach detected and confirmed", "cat": "detection"},
    {"id": "C2", "action": "Incident response team activated", "cat": "containment"},
    {"id": "C3", "action": "Containment measures applied", "cat": "containment"},
    {"id": "C4", "action": "Scope and impact assessed", "cat": "assessment"},
    {"id": "C5", "action": "Severity score calculated", "cat": "assessment"},
    {"id": "C6", "action": "Notification obligations determined", "cat": "assessment"},
    {"id": "C7", "action": "Legal counsel engaged", "cat": "assessment"},
    {"id": "C8", "action": "SA notified (if required)", "cat": "notification"},
    {"id": "C9", "action": "Data subjects notified (if required)", "cat": "notification"},
    {"id": "C10", "action": "Root cause analysis completed", "cat": "remediation"},
    {"id": "C11", "action": "Remediation implemented", "cat": "remediation"},
    {"id": "C12", "action": "Breach register updated", "cat": "documentation"},
    {"id": "C13", "action": "Post-incident review conducted", "cat": "documentation"},
    {"id": "C14", "action": "Lessons learned documented", "cat": "documentation"},
]


def load_timeline(filepath: str) -> Dict[str, Any]:
    """Load timeline from file."""
    path = Path(filepath)
    if not path.exists():
        print(f"Error: Timeline file not found: {filepath}", file=sys.stderr)
        sys.exit(1)
    try:
        with open(path, "r") as f:
            return json.load(f)
    except json.JSONDecodeError as e:
        print(f"Error: Invalid JSON: {e}", file=sys.stderr)
        sys.exit(1)


def save_timeline(filepath: str, timeline: Dict[str, Any]) -> None:
    """Save timeline to file."""
    timeline["last_updated"] = datetime.now().isoformat()
    with open(filepath, "w") as f:
        json.dump(timeline, f, indent=2)


def calculate_deadlines(t0_str: str) -> Dict[str, Dict[str, Any]]:
    """Calculate all deadline statuses from T0."""
    try:
        t0 = datetime.fromisoformat(t0_str)
    except ValueError:
        return {}

    now = datetime.now()
    result: Dict[str, Dict[str, Any]] = {}

    for dl_id, dl_def in DEADLINE_DEFINITIONS.items():
        deadline_time = t0 + timedelta(hours=dl_def["hours"])
        remaining = deadline_time - now
        remaining_hours = remaining.total_seconds() / 3600

        if remaining_hours <= 0:
            status = "EXPIRED"
        elif remaining_hours <= 6:
            status = "CRITICAL"
        elif remaining_hours <= 12:
            status = "URGENT"
        elif remaining_hours <= 24:
            status = "WARNING"
        else:
            status = "OK"

        result[dl_id] = {
            "label": dl_def["label"],
            "description": dl_def["description"],
            "deadline": deadline_time.isoformat(),
            "remaining_hours": round(remaining_hours, 1),
            "status": status,
            "expired": remaining_hours <= 0,
        }

    return result


def cmd_init(args: argparse.Namespace) -> None:
    """Initialize a new breach timeline."""
    try:
        t0 = datetime.fromisoformat(args.t0)
    except ValueError:
        print(f"Error: Invalid T0 format '{args.t0}'. Use ISO: YYYY-MM-DDTHH:MM:SS", file=sys.stderr)
        sys.exit(1)

    timeline: Dict[str, Any] = {
        "breach_id": args.breach_id,
        "description": args.description,
        "t0": args.t0,
        "status": "active",
        "severity_score": None,
        "severity_verdict": None,
        "created": datetime.now().isoformat(),
        "last_updated": datetime.now().isoformat(),
        "events": [
            {
                "timestamp": datetime.now().isoformat(),
                "action": "Breach timeline initialized",
                "category": "detection",
                "hours_from_t0": round((datetime.now() - t0).total_seconds() / 3600, 1),
            }
        ],
        "completed_actions": [],
        "notifications_sent": [],
    }

    output = args.output or f"breach_{args.breach_id.lower().replace('-', '_')}.json"
    save_timeline(output, timeline)
    print(f"Breach timeline initialized: {output}")
    print(f"  Breach ID: {args.breach_id}")
    print(f"  T0: {args.t0}")
    print(f"  Description: {args.description}")

    # Show immediate deadlines
    deadlines = calculate_deadlines(args.t0)
    urgent = [dl for dl in deadlines.values() if dl["status"] in ("CRITICAL", "URGENT", "EXPIRED")]
    if urgent:
        print("\n  URGENT DEADLINES:")
        for dl in urgent:
            print(f"    [{dl['status']}] {dl['label']}: {dl['remaining_hours']}h remaining")


def cmd_event(args: argparse.Namespace) -> None:
    """Record an event in the timeline."""
    if args.category not in VALID_CATEGORIES:
        print(f"Error: Invalid category '{args.category}'", file=sys.stderr)
        print(f"Valid: {', '.join(VALID_CATEGORIES)}", file=sys.stderr)
        sys.exit(1)

    timeline = load_timeline(args.timeline)

    try:
        t0 = datetime.fromisoformat(timeline["t0"])
    except (ValueError, KeyError):
        t0 = datetime.now()

    now = datetime.now()
    hours_from_t0 = round((now - t0).total_seconds() / 3600, 1)

    event: Dict[str, Any] = {
        "timestamp": now.isoformat(),
        "action": args.action,
        "category": args.category,
        "hours_from_t0": hours_from_t0,
    }

    timeline["events"].append(event)

    # Auto-complete matching checklist items
    for item in RESPONSE_CHECKLIST:
        if item["cat"] == args.category and item["id"] not in timeline.get("completed_actions", []):
            action_lower = args.action.lower()
            item_lower = item["action"].lower()
            # Simple keyword match
            keywords = item_lower.split()
            matches = sum(1 for kw in keywords if kw in action_lower)
            if matches >= len(keywords) * 0.4:
                timeline.setdefault("completed_actions", []).append(item["id"])

    save_timeline(args.timeline, timeline)

    if args.json:
        print(json.dumps(event, indent=2))
    else:
        print(f"Event recorded at T0+{hours_from_t0}h: {args.action}")


def cmd_status(args: argparse.Namespace) -> None:
    """Display status dashboard."""
    timeline = load_timeline(args.timeline)
    events = timeline.get("events", [])
    t0_str = timeline.get("t0", "")

    try:
        t0 = datetime.fromisoformat(t0_str)
        elapsed_hours = round((datetime.now() - t0).total_seconds() / 3600, 1)
    except ValueError:
        elapsed_hours = 0

    deadlines = calculate_deadlines(t0_str)
    completed = set(timeline.get("completed_actions", []))

    if args.json:
        result = {
            "breach_id": timeline.get("breach_id", "Unknown"),
            "status": timeline.get("status", "active"),
            "t0": t0_str,
            "elapsed_hours": elapsed_hours,
            "severity_score": timeline.get("severity_score"),
            "severity_verdict": timeline.get("severity_verdict"),
            "events_count": len(events),
            "completed_actions": list(completed),
            "pending_actions": [c["id"] for c in RESPONSE_CHECKLIST if c["id"] not in completed],
            "deadlines": deadlines,
        }
        print(json.dumps(result, indent=2))
        return

    lines: List[str] = ["=" * 60, "BREACH RESPONSE STATUS DASHBOARD", "=" * 60, ""]
    lines.append(f"  Breach: {timeline.get('breach_id', 'Unknown')} | Status: {timeline.get('status', 'active').upper()} | Elapsed: {elapsed_hours}h")
    lines.append(f"  T0: {t0_str} | Severity: {timeline.get('severity_verdict', 'N/A')}\n")
    lines.append("DEADLINES:")
    for dl in deadlines.values():
        icon = {"EXPIRED": "!!!", "CRITICAL": "!!!", "URGENT": ">>>", "WARNING": " > "}.get(dl["status"], "   ")
        lines.append(f"  {icon} {dl['label']}: {dl['remaining_hours']}h [{dl['status']}]")
    lines.append("\nCHECKLIST:")
    for item in RESPONSE_CHECKLIST:
        done = "[X]" if item["id"] in completed else "[ ]"
        lines.append(f"  {done} {item['id']}: {item['action']}")
    lines.append(f"  Progress: {len(completed)}/{len(RESPONSE_CHECKLIST)}\n")
    lines.append("RECENT EVENTS:")
    for evt in list(reversed(events))[:10]:
        lines.append(f"  T0+{evt.get('hours_from_t0', 0)}h [{evt.get('category', 'other')}] {evt.get('action', '')}")
    lines.append("=" * 60)
    print("\n".join(lines))


def cmd_deadlines(args: argparse.Namespace) -> None:
    """Check and display deadline status."""
    timeline = load_timeline(args.timeline)
    t0_str = timeline.get("t0", "")
    deadlines = calculate_deadlines(t0_str)

    if args.json:
        print(json.dumps({
            "breach_id": timeline.get("breach_id", "Unknown"),
            "t0": t0_str,
            "deadlines": deadlines,
        }, indent=2))
        return

    print(f"Breach: {timeline.get('breach_id', 'Unknown')}")
    print(f"T0: {t0_str}")
    print()

    for dl in sorted(deadlines.values(), key=lambda x: x["remaining_hours"]):
        icon = "!!!" if dl["expired"] else ">>>" if dl["status"] in ("CRITICAL", "URGENT") else "   "
        print(f"  {icon} {dl['label']}: {dl['remaining_hours']}h [{dl['status']}]")


def main() -> None:
    parser = argparse.ArgumentParser(description="Breach Timeline Tracker")
    sub = parser.add_subparsers(dest="command")
    p = sub.add_parser("init")
    p.add_argument("--breach-id", required=True); p.add_argument("--t0", required=True)
    p.add_argument("--description", required=True); p.add_argument("--output", type=str)
    p.add_argument("--json", action="store_true")
    p = sub.add_parser("event")
    p.add_argument("--timeline", required=True); p.add_argument("--action", required=True)
    p.add_argument("--category", required=True); p.add_argument("--json", action="store_true")
    for name in ["status", "deadlines"]:
        p = sub.add_parser(name)
        p.add_argument("--timeline", required=True); p.add_argument("--json", action="store_true")
    args = parser.parse_args()
    if not args.command:
        parser.print_help(); sys.exit(1)
    cmds = {"init": cmd_init, "event": cmd_event, "status": cmd_status, "deadlines": cmd_deadlines}
    try:
        cmds[args.command](args)
    except KeyError:
        parser.print_help(); sys.exit(1)
    except Exception as e:
        print(f"Error: {e}", file=sys.stderr); sys.exit(1)


if __name__ == "__main__":
    main()