implementing-vulnerability-sla-breach-alerting

实现漏洞 SLA 违约告警

概述

漏洞修复 SLA 定义了根据严重程度处理安全发现的最长期限。本技能涵盖构建自动化告警系统，用于跟踪修复时间线、检测 SLA 违约、发送升级通知并生成合规报告。行业标准 SLA 目标为：严重（24-48 小时）、高（15-30 天）、中（60 天）、低（90 天）。

何时使用

• 在环境中部署或配置实现漏洞 SLA 违约告警能力时
• 在建立符合合规要求的安全控制时
• 在为此领域构建或改进安全架构时
• 在进行需要此实现的安全评估时

前置条件

• Python 3.9+ 及 requests、pandas、jinja2、smtplib 库
• 具有 API 访问权限的漏洞管理平台（DefectDojo、Qualys、Tenable）
• SMTP 服务器或 webhook 端点（Slack、Microsoft Teams、PagerDuty）
• 用于 SLA 跟踪的数据库（PostgreSQL 或 SQLite）

SLA 策略定义

标准 SLA 层级

严重程度	修复 SLA	宽限期	升级级别
严重（CVSS 9.0-10.0）	48 小时	12 小时	工程 VP + CISO
高（CVSS 7.0-8.9）	15 天	5 天	工程总监
中（CVSS 4.0-6.9）	60 天	14 天	团队负责人
低（CVSS 0.1-3.9）	90 天	30 天	资产负责人

SLA 配置文件

# sla_policy.yaml
sla_tiers:
  critical:
    cvss_min: 9.0
    cvss_max: 10.0
    remediation_days: 2
    grace_period_days: 0.5
    escalation_contacts:
      - [email protected]
      - [email protected]
    pagerduty_severity: critical
  high:
    cvss_min: 7.0
    cvss_max: 8.9
    remediation_days: 15
    grace_period_days: 5
    escalation_contacts:
      - [email protected]
    pagerduty_severity: high
  medium:
    cvss_min: 4.0
    cvss_max: 6.9
    remediation_days: 60
    grace_period_days: 14
    escalation_contacts:
      - [email protected]
    pagerduty_severity: warning
  low:
    cvss_min: 0.1
    cvss_max: 3.9
    remediation_days: 90
    grace_period_days: 30
    escalation_contacts:
      - [email protected]
    pagerduty_severity: info

notification_channels:
  slack:
    webhook_url: "${SLACK_WEBHOOK_URL}"
    channel: "#vulnerability-alerts"
  email:
    smtp_host: smtp.company.com
    smtp_port: 587
    from_address: [email protected]
  pagerduty:
    api_key: "${PAGERDUTY_API_KEY}"
    service_id: "${PAGERDUTY_SERVICE_ID}"

alert_schedules:
  approaching_breach:
    percentage_elapsed: 80
    frequency_hours: 24
  at_breach:
    notification: immediate
    escalation: true
  post_breach:
    frequency_hours: 12
    escalation_increase: true

工作流

步骤 1：用于 SLA 跟踪的数据库架构

CREATE TABLE vulnerability_sla (
    id SERIAL PRIMARY KEY,
    cve_id VARCHAR(20) NOT NULL,
    finding_id VARCHAR(100) NOT NULL,
    asset_hostname VARCHAR(255),
    severity VARCHAR(20) NOT NULL,
    cvss_score DECIMAL(3,1),
    discovered_at TIMESTAMP NOT NULL,
    sla_deadline TIMESTAMP NOT NULL,
    remediated_at TIMESTAMP,
    status VARCHAR(20) DEFAULT 'open',
    owner_email VARCHAR(255),
    escalation_level INTEGER DEFAULT 0,
    last_alert_sent TIMESTAMP,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE INDEX idx_sla_status ON vulnerability_sla(status);
CREATE INDEX idx_sla_deadline ON vulnerability_sla(sla_deadline);
CREATE INDEX idx_sla_severity ON vulnerability_sla(severity);

步骤 2：SLA 违约检测逻辑

from datetime import datetime, timedelta, timezone
import yaml

def load_sla_policy(policy_path="sla_policy.yaml"):
    with open(policy_path, "r") as f:
        return yaml.safe_load(f)

def get_sla_tier(cvss_score, policy):
    for tier_name, tier in policy["sla_tiers"].items():
        if tier["cvss_min"] <= cvss_score <= tier["cvss_max"]:
            return tier_name, tier
    return "low", policy["sla_tiers"]["low"]

def calculate_sla_deadline(discovered_at, cvss_score, policy):
    tier_name, tier = get_sla_tier(cvss_score, policy)
    deadline = discovered_at + timedelta(days=tier["remediation_days"])
    return deadline, tier_name

def check_sla_status(discovered_at, sla_deadline, remediated_at=None):
    now = datetime.now(timezone.utc)
    if remediated_at:
        if remediated_at <= sla_deadline:
            return "remediated_within_sla"
        return "remediated_breach"
    if now > sla_deadline:
        overdue_days = (now - sla_deadline).days
        return f"breached_{overdue_days}d_overdue"
    remaining = sla_deadline - now
    total_sla = sla_deadline - discovered_at
    pct_elapsed = ((total_sla - remaining) / total_sla) * 100
    if pct_elapsed >= 80:
        return "approaching_breach"
    return "within_sla"

步骤 3：通知发送

import requests
import json
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

def send_slack_alert(webhook_url, vuln_data, sla_status):
    color = {"breached": "#FF0000", "approaching_breach": "#FFA500", "within_sla": "#36A64F"}
    status_color = color.get("breached" if "breached" in sla_status else sla_status, "#808080")
    payload = {
        "attachments": [{
            "color": status_color,
            "title": f"Vulnerability SLA Alert: {vuln_data['cve_id']}",
            "fields": [
                {"title": "Severity", "value": vuln_data["severity"], "short": True},
                {"title": "CVSS", "value": str(vuln_data["cvss_score"]), "short": True},
                {"title": "Asset", "value": vuln_data["asset_hostname"], "short": True},
                {"title": "SLA Status", "value": sla_status, "short": True},
                {"title": "Deadline", "value": vuln_data["sla_deadline"].strftime("%Y-%m-%d %H:%M UTC"), "short": True},
                {"title": "Owner", "value": vuln_data.get("owner_email", "Unassigned"), "short": True},
            ],
        }]
    }
    requests.post(webhook_url, json=payload, timeout=10)

def send_pagerduty_alert(api_key, service_id, vuln_data, severity):
    payload = {
        "routing_key": api_key,
        "event_action": "trigger",
        "payload": {
            "summary": f"SLA Breach: {vuln_data['cve_id']} on {vuln_data['asset_hostname']}",
            "severity": severity,
            "source": vuln_data["asset_hostname"],
            "custom_details": {
                "cve_id": vuln_data["cve_id"],
                "cvss_score": vuln_data["cvss_score"],
                "sla_deadline": vuln_data["sla_deadline"].isoformat(),
            }
        }
    }
    requests.post(
        "https://events.pagerduty.com/v2/enqueue",
        json=payload, timeout=10
    )

def send_email_alert(smtp_config, to_addresses, vuln_data, sla_status):
    msg = MIMEMultipart("alternative")
    msg["Subject"] = f"[SLA {sla_status.upper()}] {vuln_data['cve_id']} - {vuln_data['severity']}"
    msg["From"] = smtp_config["from_address"]
    msg["To"] = ", ".join(to_addresses)
    body = f"""
    Vulnerability SLA Alert

    CVE: {vuln_data['cve_id']}
    Severity: {vuln_data['severity']} (CVSS {vuln_data['cvss_score']})
    Asset: {vuln_data['asset_hostname']}
    SLA Deadline: {vuln_data['sla_deadline'].strftime('%Y-%m-%d %H:%M UTC')}
    Status: {sla_status}
    Owner: {vuln_data.get('owner_email', 'Unassigned')}

    Please take immediate action to remediate this vulnerability.
    """
    msg.attach(MIMEText(body, "plain"))
    with smtplib.SMTP(smtp_config["smtp_host"], smtp_config["smtp_port"]) as server:
        server.starttls()
        server.send_message(msg)

步骤 4：定时 SLA 检查运行器

# Run SLA breach check every hour via cron
echo "0 * * * * cd /opt/vuln-sla && python3 scripts/process.py --check-sla" | crontab -

# Manual check
python3 scripts/process.py --check-sla --policy sla_policy.yaml

# Generate SLA compliance report
python3 scripts/process.py --report --period monthly --output sla_report.html

SLA 指标仪表板

关键绩效指标

def calculate_sla_metrics(db_connection, period_start, period_end):
    metrics = {
        "total_findings": 0,
        "remediated_within_sla": 0,
        "sla_breach_count": 0,
        "mean_time_to_remediate": {},
        "sla_compliance_rate": 0.0,
        "current_overdue": 0,
    }
    # Query findings in period grouped by severity
    query = """
        SELECT severity, COUNT(*) as total,
               SUM(CASE WHEN remediated_at <= sla_deadline THEN 1 ELSE 0 END) as within_sla,
               AVG(EXTRACT(EPOCH FROM (COALESCE(remediated_at, NOW()) - discovered_at))/86400) as avg_days
        FROM vulnerability_sla
        WHERE discovered_at BETWEEN %s AND %s
        GROUP BY severity
    """
    return metrics

参考资料

• 漏洞管理 SLA 指南
• NIST SP 800-40 Rev 4 - 补丁管理
• PagerDuty Events API v2
• Slack Incoming Webhooks