安全监控与分析：从日志采集到威胁狩猎的工程化实践

import json, hashlib
from dataclasses import dataclass, asdict, field
from datetime import datetime, timezone

@dataclass
class SecEvent:
    ts: str
    source: str          # firewall / edr / waf / auth
    host: str
    user: str
    action: str          # login / connect / alert
    src_ip: str
    dst_ip: str
    severity: str        # info / low / medium / high / critical
    raw: dict = field(default_factory=dict)

    def fingerprint(self) -> str:
        raw = f"{self.source}|{self.host}|{self.src_ip}|{self.ts[:19]}"
        return hashlib.sha256(raw.encode()).hexdigest()[:16]

def normalize(source: str, record: dict) -> SecEvent:
    """不同源映射到统一字段"""
    mappers = {
        "firewall": lambda r: dict(
            ts=r["time"], host=r["devname"], user="-", action=r["action"],
            src_ip=r["srcip"], dst_ip=r["dstip"], severity="info"),
        "auth": lambda r: dict(
            ts=r["timestamp"], host=r["hostname"], user=r["user"],
            action=r["event"], src_ip=r.get("ip", "-"), dst_ip="-",
            severity="medium" if r["event"] == "failed" else "info"),
    }
    fields = mappers[source](record)
    return SecEvent(source=source, raw=record, **fields)

# rules/brute_force.yml
title: 多次登录失败后成功
id: auth-brute-001
level: high
source: auth
condition:
  group_by: [user, src_ip]
  window: 300
  sequence:
    - match: {action: failed, count: ">=5"}
    - match: {action: success}

import yaml, time
from collections import defaultdict

class RuleEngine:
    def __init__(self, rule_path):
        with open(rule_path) as f:
            self.rule = yaml.safe_load(f)
        self.buckets = defaultdict(list)

    def feed(self, ev: SecEvent):
        if ev.source != self.rule["source"]:
            return None
        cond = self.rule["condition"]
        key = tuple(getattr(ev, k) for k in cond["group_by"])
        now = time.time()
        self.buckets[key].append((now, ev.action, ev))

        # 清理窗口外数据
        window = cond["window"]
        self.buckets[key] = [(t, a, e) for t, a, e in self.buckets[key]
                             if now - t <= window]

        seq = cond["sequence"]
        actions = [a for _, a, _ in self.buckets[key]]
        return self._match_sequence(seq, actions)

    def _match_sequence(self, seq, actions):
        """简化版：按顺序检查每步是否满足"""
        i = 0
        for step in seq:
            rule = step["match"]
            count = int(rule.get("count", ">=1").replace(">=", ""))
            hits = sum(1 for a in actions[i:] if a == rule["action"])
            if hits < count:
                return None
            i += hits
        return {"rule": self.rule["title"], "level": self.rule["level"]}

class Correlator:
    def __init__(self, window=300, threshold=3):
        self.window = window
        self.threshold = threshold
        self.buf = defaultdict(list)

    def add(self, entity: str, alert: dict, ts: float = None):
        ts = ts or time.time()
        self.buf[entity].append((ts, alert))
        self.buf[entity] = [(t, a) for t, a in self.buf[entity]
                            if ts - t <= self.window]

    def incidents(self) -> list[dict]:
        out = []
        for entity, items in self.buf.items():
            if len(items) < self.threshold:
                continue
            levels = [a["level"] for _, a in items]
            level = ("critical" if "critical" in levels else
                     "high" if "high" in levels else "medium")
            out.append({"entity": entity, "count": len(items),
                        "level": level,
                        "rules": sorted({a["rule"] for _, a in items})})
        return out

import numpy as np
from collections import deque

class BaselineDetector:
    def __init__(self, window=100, z_threshold=3.5):
        self.history = deque(maxlen=window)
        self.z = z_threshold

    def update(self, value: float) -> float | None:
        """返回 z-score，超过阈值即异常"""
        if len(self.history) < 20:
            self.history.append(value)
            return None
        med = np.median(self.history)
        mad = np.median(np.abs(np.array(self.history) - med)) or 1e-9
        z = 0.6745 * (value - med) / mad
        self.history.append(value)
        return z if abs(z) > self.z else None

import httpx
from functools import lru_cache

@lru_cache(maxsize=4096)
def ip_reputation(ip: str) -> dict:
    # 示例：接入任意威胁情报API
    try:
        r = httpx.get(f"https://api.example.com/ip/{ip}", timeout=3)
        return r.json()
    except Exception:
        return {"score": 0, "tags": []}

def enrich(alert: dict) -> dict:
    ip = alert.get("src_ip")
    if ip:
        rep = ip_reputation(ip)
        alert["intel"] = rep
        if rep.get("score", 0) > 80:
            alert["level"] = "critical"
    return alert

class Playbook:
    def __init__(self, name, steps):
        self.name = name
        self.steps = steps       # [(action_name, fn, requires_approval)]

    def run(self, alert, auto=True):
        log = []
        for name, fn, need_approval in self.steps:
            if need_approval and not auto:
                log.append(f"[待审批] {name}")
                break
            try:
                result = fn(alert)
                log.append(f"[完成] {name}: {result}")
            except Exception as e:
                log.append(f"[失败] {name}: {e}")
                break
        return log

def block_ip(alert):
    # 实际调用防火墙API封禁
    return f"已封禁 {alert['src_ip']}"

def notify(alert):
    return "已通知值班"

playbook = Playbook("brute_force_response", [
    ("封禁源IP", block_ip, True),      # 需审批
    ("通知值班", notify, False),        # 自动
])