VPS 自动化漏洞响应：从 CVE 情报到无人值守热修补

#!/usr/bin/env python3
"""VPS CVE auto-triage + 4-stage hot patch pipeline.
Stages: dry-run -> sandbox smoke -> prod smoke -> atomic commit w/ 30s rollback."""
import json, subprocess, time, sqlite3, os, hashlib, urllib.request, sys

OLLAMA = "http://127.0.0.1:11434"
MODEL  = "qwen2.5:7b"
AUDIT  = "/var/lib/vps-ops/audit.db"
ROLLBACK_WINDOW = 30          # seconds
CONFIDENCE_THRESHOLD = 0.70   # below this -> human approval, no auto-patch

def vps_fingerprint():
    """OS/kernel/installed packages/ports -> context for LLM relevance check."""
    os_ver  = open("/etc/os-release").read().split("\n")[0]
    kernel  = os.uname().release
    pkgs    = subprocess.check_output(
        "dpkg -l | awk '{print $2\" \"$3}' | grep -v 'ii' || rpm -qa",
        shell=True).decode(errors="ignore")
    ports   = subprocess.check_output("ss -tlnp | awk '{print $4}'",
        shell=True).decode(errors="ignore")
    return {"os": os_ver, "kernel": kernel, "packages": pkgs.strip(),
            "listen_ports": ports.strip()}

def llm_judge(cve_entry, fingerprint):
    """Local LLM: relevance + severity + patch strategy, constrained JSON out."""
    prompt = (
        "You are a VPS security triage engine. Given the VPS fingerprint and "
        "CVE below, respond ONLY with JSON: {cve, affected, severity, strategy, "
        "target_services, rollback_window_s, confidence}. strategy one of: "
        "hot-reload | rolling-restart | full-restart | no-action."
        "\nFingerprint: " + json.dumps(fingerprint) +
        "\nCVE: " + json.dumps(cve_entry)
    )
    r = urllib.request.urlopen(OLLAMA + "/api/generate",
        data=json.dumps({"model": MODEL, "prompt": prompt,
                         "stream": False}).encode())
    out = json.loads(r.read())["response"].strip()
    return json.loads(out)          # caller validates against schema

def run_stage(cmd, label, timeout=300):
    """Run one pipeline stage; return (ok, output)."""
    p = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
    ok = p.returncode == 0
    audit(f"{label} ok={ok} cmd={cmd}")
    return ok, p.stdout + p.stderr

def audit(msg, stage="unknown"):
    """Hash-chained SQLite audit log (tamper-evident)."""
    conn = sqlite3.connect(AUDIT)
    prev = conn.execute(
        "SELECT hash FROM audit ORDER BY ts DESC, id DESC LIMIT 1").fetchone()
    prev_hash = prev[0] if prev else "0" * 64
    payload = f"{time.time()}|{stage}|{msg}|{prev_hash}"
    h = hashlib.sha256(payload.encode()).hexdigest()
    conn.execute("INSERT INTO audit (ts, stage, msg, prev_hash, hash) "
                 "VALUES (?,?,?,?,?)", (time.time(), stage, msg, prev_hash, h))
    conn.commit(); conn.close()

def execute_patch(decision):
    """4-stage pipeline. Returns 'patched' | 'rolled-back' | 'pending-approval'."""
    cve, strategy = decision["cve"], decision["strategy"]
    audit(f"start {cve} strategy={strategy}", "plan")

    if decision.get("confidence", 0) < CONFIDENCE_THRESHOLD:
        return "pending-approval"      # push to human via Telegram/email

    # Stage 1: dry-run (zero side effects)
    ok, out = run_stage(
        "vps-patch-tool dry-run --cve " + cve + " --strategy " + strategy,
        "stage1-dryrun")
    if not ok:
        audit("dry-run failed, aborting", "stage1"); return "rolled-back"

    # Stage 2: sandbox smoke (isolated instance, prod untouched)
    ok, out = run_stage("vps-patch-tool sandbox --cve " + cve, "stage2-sandbox")
    if not ok:
        audit("sandbox failed, prod untouched, aborting", "stage2")
        return "rolled-back"

    # Stage 3: prod apply + immediate smoke
    ok, out = run_stage("vps-patch-tool apply --cve " + cve
                        + " --strategy " + strategy, "stage3-prod")
    ok2, _ = run_stage("curl -fsS http://127.0.0.1:8080/health && "
                       "vps-patch-tool smoke --cve " + cve, "stage3-smoke")
    if not (ok and ok2):
        subprocess.run("vps-patch-tool rollback --cve " + cve, shell=True)
        audit("prod smoke failed -> auto rollback", "stage3")
        return "rolled-back"

    # Stage 4: rollback window, then atomic commit
    time.sleep(ROLLBACK_WINDOW)
    ok3, _ = run_stage("vps-monitor check --since "
        + str(int(time.time()) - ROLLBACK_WINDOW), "stage4-verify")
    if not ok3:
        subprocess.run("vps-patch-tool rollback --cve " + cve, shell=True)
        audit("post-window anomaly -> rollback", "stage4")
        return "rolled-back"

    audit("committed", "stage4")
    return "patched"

def main():
    cves = json.load(open(sys.argv[1]))            # fetched CVE list
    fp = vps_fingerprint()
    for cve in cves:
        decision = llm_judge(cve, fp)
        result = execute_patch(decision)
        print(f"{decision['cve']}: {result} "
              f"(sev={decision.get('severity')}, "
              f"conf={decision.get('confidence')})")

if __name__ == "__main__":
    main()

# /etc/systemd/system/vps-cve-triage.service
[Unit]
Description=VPS CVE auto-triage & hot patch pipeline
After=network-online.target ollama.service

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /opt/vps-ops/cve_triage_patch.py \
    /var/lib/vps-ops/latest_cves.json
Environment=OLLAMA=127.0.0.1:11434
TimeoutStartSec=900
Restart=on-failure

[Install]
WantedBy=multi-user.target

# /etc/systemd/system/vps-cve-triage.timer
[Unit]
Description=Run VPS CVE triage daily

[Timer]
OnCalendar=*-*-* 02:00:00
Persistent=true

[Install]
WantedBy=timers.target