[数据采集层]
  主机日志 ─┐
  网络流量 ─┼─→ [归一化/特征工程] ─→ [AI 关联分析] ─→ [攻击链还原]
  内存镜像 ─┘                                        │
                                                      ↓
                                            [IOC 提取与情报关联]
                                                      │
                                                      ↓
                                      [入口点定位 + 攻击者画像/归因]
                                                      │
                                                      ↓
                                            [报告 + 处置 + 加固建议]

各主机/服务
   │ (filebeat / fluent-bit / vector)
   ▼
消息队列 (Kafka / Redis Streams)
   │
   ▼
解析与归一化服务 (本文代码)
   │
   ▼
存储 (Elasticsearch / OpenSearch / ClickHouse / 对象存储)
   │
   ▼
关联分析引擎 (规则 + ML + LLM)  ──→  告警/攻击时间线

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
log_normalizer.py
多源日志归一化框架：把不同格式的日志统一为 OCSF 风格的事件 dict。
仅用于授权环境下的日志分析与取证。
"""
import re
import json
import sys
from datetime import datetime, timezone
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any


# ---------- 统一事件模型 ----------
def make_event(**kwargs) -> Dict[str, Any]:
    """构造一个统一事件，补齐默认值。"""
    base = {
        "event_time": None,
        "event_source": "unknown",
        "event_type": "unknown",
        "actor": None,
        "src_ip": None,
        "dst_ip": None,
        "dst_port": None,
        "action": "unknown",
        "outcome": "unknown",
        "host": "unknown",
        "raw": "",
        "extra": {},
    }
    base.update(kwargs)
    return base


# ---------- 解析器基类 ----------
class BaseParser(ABC):
    @abstractmethod
    def match(self, line: str) -> bool:
        """能否处理这一行。"""
        ...

    @abstractmethod
    def parse(self, line: str) -> Optional[Dict[str, Any]]:
        """解析为统一事件；无法解析返回 None。"""
        ...


# ---------- Nginx combined 日志解析器 ----------
class NginxCombinedParser(BaseParser):
    # 示例: 127.0.0.1 - alice [10/Oct/2024:13:55:36 +0000] "GET /api/user HTTP/1.1" 200 2326 "-"
    _re = re.compile(
        r'(?P<src_ip>\S+) \S+ (?P<actor>\S+) '
        r'$$(?P<ts>[^$$]+)\] "(?P<method>\S+) (?P<path>\S+) (?P<proto>\S+)" '
        r'(?P<status>\d{3}) (?P<size>\d+|-)'
    )

    def match(self, line: str) -> bool:
        return bool(self._re.match(line))

    def parse(self, line: str) -> Optional[Dict[str, Any]]:
        m = self._re.match(line)
        if not m:
            return None
        d = m.groupdict()
        # 时间格式: 10/Oct/2024:13:55:36 +0000
        try:
            dt = datetime.strptime(d["ts"], "%d/%b/%Y:%H:%M:%S %z")
            event_time = dt.astimezone(timezone.utc).isoformat()
        except ValueError:
            event_time = None
        return make_event(
            event_time=event_time,
            event_source="nginx",
            event_type="web",
            actor=d["actor"] if d["actor"] != "-" else None,
            src_ip=d["src_ip"],
            action="http_" + d["method"].lower(),
            outcome="success" if d["status"][0] in "23" else "failure",
            host="web-01",
            raw=line,
            extra={"path": d["path"], "status": int(d["status"])},
        )


# ---------- SSH 认证日志解析器 (syslog 风格) ----------
class SSHAuthParser(BaseParser):
    _re_fail = re.compile(
        r'(?P<ts>\w{3}\s+\d+\s+\d{2}:\d{2}:\d{2}) .*?'
        r'sshd$$\d+$$: Failed password for (?:invalid user )?(?P<user>\S+) '
        r'from (?P<src_ip>\S+) port \d+'
    )
    _re_ok = re.compile(
        r'(?P<ts>\w{3}\s+\d+\s+\d{2}:\d{2}:\d{2}) .*?'
        r'sshd$$\d+$$: Accepted password for (?P<user>\S+) from (?P<src_ip>\S+)'
    )
    _month = {m: i + 1 for i, m in enumerate(
        ["Jan", "Feb", "Mar", "Apr", "May", "Jun",
         "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"])}

    def match(self, line: str) -> bool:
        return "sshd" in line and ("Failed password" in line or "Accepted password" in line)

    def _ts(self, ts: str, year: int = 2024) -> Optional[str]:
        try:
            mon, day, hhmmss = ts.split()
            dt = datetime(year, self._month[mon], int(day),
                          *map(int, hhmmss.split(":")))
            return dt.replace(tzinfo=timezone.utc).isoformat()
        except Exception:
            return None

    def parse(self, line: str) -> Optional[Dict[str, Any]]:
        m = self._re_fail.match(line) or self._re_ok.match(line)
        if not m:
            return None
        d = m.groupdict()
        ok = "Accepted" in line
        return make_event(
            event_time=self._ts(d["ts"]),
            event_source="ssh",
            event_type="auth",
            actor=d["user"],
            src_ip=d["src_ip"],
            action="login",
            outcome="success" if ok else "failure",
            host="bastion-01",
            raw=line,
        )


# ---------- 解析器注册表 ----------
PARSERS = [NginxCombinedParser(), SSHAuthParser()]


def normalize_line(line: str) -> Optional[Dict[str, Any]]:
    line = line.rstrip("\n")
    if not line.strip():
        return None
    for p in PARSERS:
        if p.match(line):
            return p.parse(line)
    return None


def main():
    """从 stdin 读原始日志，输出 JSON Lines 归一化事件。"""
    out = sys.stdout
    for line in sys.stdin:
        ev = normalize_line(line)
        if ev:
            out.write(json.dumps(ev, ensure_ascii=False) + "\n")


if __name__ == "__main__":
    main()

# 把多种日志合并后归一化
cat /var/log/nginx/access.log /var/log/auth.log | python3 log_normalizer.py > events.jsonl

# 仅统计认证失败来源 IP  Top 10（后续关联用）
python3 log_normalizer.py < /var/log/auth.log \
  | jq -c 'select(.event_type=="auth" and .outcome=="failure") | .src_ip' \
  | sort | uniq -c | sort -rn | head

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
log_anomaly.py
对归一化日志事件做特征工程 + 无监督异常检测 (IsolationForest)。
输出每个 (src_ip) 的异常分数，供分析师优先审查。
仅用于授权环境的异常发现。
"""
import json
from collections import defaultdict
from datetime import datetime
from urllib.parse import urlparse

import numpy as np
from sklearn.ensemble import IsolationForest


def load_events(path: str):
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                yield json.loads(line)


def build_features(events):
    """按 src_ip 在全局时间窗口内聚合出行为特征。"""
    # 每个 IP 的聚合器
    agg = defaultdict(lambda: {
        "total": 0, "fail_auth": 0, "success_auth": 0,
        "paths": set(), "status_4xx": 0, "status_5xx": 0,
        "methods": defaultdict(int), "first_seen": None, "last_seen": None,
    })
    for ev in events:
        ip = ev.get("src_ip")
        if not ip:
            continue
        a = agg[ip]
        a["total"] += 1
        if ev.get("event_type") == "auth":
            if ev.get("outcome") == "failure":
                a["fail_auth"] += 1
            elif ev.get("outcome") == "success":
                a["success_auth"] += 1
        st = (ev.get("extra") or {}).get("status")
        if st:
            if 400 <= st < 500:
                a["status_4xx"] += 1
            elif 500 <= st < 600:
                a["status_5xx"] += 1
        p = (ev.get("extra") or {}).get("path")
        if p:
            a["paths"].add(urlparse(p).path)
        m = ev.get("action")
        if m:
            a["methods"][m] += 1
        t = ev.get("event_time")
        if t:
            dt = datetime.fromisoformat(t)
            if a["first_seen"] is None or dt < a["first_seen"]:
                a["first_seen"] = dt
            if a["last_seen"] is None or dt > a["last_seen"]:
                a["last_seen"] = dt

    rows = []
    ips = []
    for ip, a in agg.items():
        span = 0.0
        if a["first_seen"] and a["last_seen"]:
            span = (a["last_seen"] - a["first_seen"]).total_seconds()
        distinct_paths = len(a["paths"])
        fail_ratio = a["fail_auth"] / a["total"] if a["total"] else 0
        # 特征向量：请求总数、失败认证数、成功认证数、不同路径数、
        # 4xx数、5xx数、时间跨度(分)、失败率
        feat = [
            a["total"], a["fail_auth"], a["success_auth"], distinct_paths,
            a["status_4xx"], a["status_5xx"], span / 60.0, fail_ratio,
        ]
        rows.append(feat)
        ips.append(ip)
    return np.array(rows, dtype=float), ips


def detect(rows, ips, contamination: float = 0.02):
    """IsolationForest 无监督异常检测。"""
    if rows.shape[0] == 0:
        return [], []
    # 对数变换抑制长尾
    X = np.log1p(rows)
    model = IsolationForest(n_estimators=200, contamination=contamination,
                            random_state=42)
    pred = model.fit_predict(X)
    scores = -model.score_samples(X)  # 越大越异常
    anomalies = []
    for ip, s, p in zip(ips, scores, pred):
        if p == -1:  # 异常
            anomalies.append((ip, round(float(s), 4)))
    anomalies.sort(key=lambda x: x[1], reverse=True)
    return anomalies, model


def main():
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else "events.jsonl"
    events = list(load_events(path))
    print(f"[+] 载入事件 {len(events)} 条")
    rows, ips = build_features(events)
    print(f"[+] 聚合出 {len(ips)} 个独立源 IP")
    anomalies, _ = detect(rows, ips)
    print(f"[+] 检出异常 IP {len(anomalies)} 个，按异常分排序（优先审查）：")
    for ip, score in anomalies:
        print(f"    {ip:20s}  anomaly_score={score}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
log_correlate.py
基于规则的日志关联引擎：识别常见攻击模式并生成攻击时间线。
仅用于授权环境的攻击链还原。
"""
import json
from collections import defaultdict
from datetime import datetime


def load_events(path):
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if line:
                yield json.loads(line)


def by_ip(events):
    d = defaultdict(list)
    for ev in events:
        ip = ev.get("src_ip")
        if ip:
            d[ip].append(ev)
    for ip in d:
        d[ip].sort(key=lambda e: e.get("event_time") or "")
    return d


def detect_bruteforce_then_success(ip, evs, fail_threshold=10):
    """模式：N 次认证失败 + 随后一次成功（同源 IP）。"""
    fails = [e for e in evs if e["event_type"] == "auth" and e["outcome"] == "failure"]
    succ = [e for e in evs if e["event_type"] == "auth" and e["outcome"] == "success"]
    if len(fails) >= fail_threshold and succ:
        last_fail = fails[-1]
        first_succ = min(succ, key=lambda e: e["event_time"])
        if first_succ["event_time"] >= last_fail["event_time"]:
            return {
                "pattern": "bruteforce_then_success",
                "src_ip": ip,
                "fail_count": len(fails),
                "compromised_actor": first_succ.get("actor"),
                "window": f"{last_fail['event_time']} -> {first_succ['event_time']}",
                "evidence": [last_fail["raw"], first_succ["raw"]],
            }
    return None


def detect_web_scan(ip, evs, path_threshold=20):
    """模式：同一 IP 短时间内访问大量不同路径（目录爆破/扫描）。"""
    paths = {(e.get("extra") or {}).get("path") for e in evs
             if e["event_type"] == "web" and (e.get("extra") or {}).get("path")}
    if len(paths) >= path_threshold:
        return {
            "pattern": "web_path_scan",
            "src_ip": ip,
            "distinct_paths": len(paths),
            "sample": list(paths)[:10],
        }
    return None


def main():
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else "events.jsonl"
    events = list(load_events(path))
    grouped = by_ip(events)
    print(f"[+] 开始关联分析，涉及 {len(grouped)} 个源 IP\n")
    findings = []
    for ip, evs in grouped.items():
        for fn in (detect_bruteforce_then_success, detect_web_scan):
            r = fn(ip, evs)
            if r:
                findings.append(r)
    for f in findings:
        print("="*60)
        print(json.dumps(f, ensure_ascii=False, indent=2))
    print(f"\n[+] 共识别疑似攻击序列 {len(findings)} 条")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
llm_log_analyst.py
用大语言模型对归一化后的高危事件做语义归纳与假设生成。
安全约束：默认不发送任何明文凭证/PII；仅发送已脱敏的事件摘要。
需自行配置模型端点（兼容 OpenAI 接口）。仅用于授权环境。
"""
import json
import os
import re

# —— 脱敏：在把日志送出去之前，先抹掉明显的敏感字段 ——
SENSITIVE = [
    (re.compile(r'password["\']?\s*[:=]\s*["\']?[^"\',\s]+', re.I), 'password=***'),
    (re.compile(r'token["\']?\s*[:=]\s*["\']?[^"\',\s]+', re.I), 'token=***'),
    (re.compile(r'Authorization:\s*Bearer\s+\S+', re.I), 'Authorization: Bearer ***'),
    (re.compile(r'\b\d{6,}\b'), '[NUM]'),  # 粗略抹除长数字（身份证/手机号占位）
]

def redact(text: str) -> str:
    for pat, repl in SENSITIVE:
        text = pat.sub(repl, text)
    return text


def call_llm(messages, model="gpt-4o-mini", base_url=None, api_key=None):
    """极简 OpenAI 兼容调用（不依赖 sdk，便于审计）。"""
    import urllib.request
    api_key = api_key or os.getenv("LLM_API_KEY")
    base_url = base_url or os.getenv("LLM_BASE_URL", "https://api.openai.com/v1")
    body = json.dumps({"model": model, "messages": messages,
                       "temperature": 0.1}).encode("utf-8")
    req = urllib.request.Request(
        base_url.rstrip("/") + "/chat/completions",
        data=body, headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=60) as r:
        return json.loads(r.read().decode("utf-8"))["choices"][0]["message"]["content"]


def summarize_events(events, max_events=200):
    """把高危事件脱敏后交给模型，产出攻击概述 + 下一步建议。"""
    sample = events[:max_events]
    cleaned = [redact(json.dumps(e, ensure_ascii=False)) for e in sample]
    prompt = (
        "你是资深蓝队分析师。下面是某系统归一化后的高危安全事件（已脱敏）。"
        "请完成三件事：\n"
        "1) 用 3-5 句话概述攻击者可能做了什么；\n"
        "2) 列出 3 条最该优先核实的假设；\n"
        "3) 给出 3 条具体的下一步取证动作（查什么日志/镜像/流量）。\n"
        "只输出分析，不要执行任何操作。\n\n"
        "事件样本：\n" + "\n".join(cleaned)
    )
    return call_llm([
        {"role": "system", "content": "你是严谨的网络安全事件响应专家，仅做防御性分析。"},
        {"role": "user", "content": prompt},
    ])


def main():
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else "events.jsonl"
    # 仅选高危事件（异常分高或规则命中的）送模型，控制成本与泄露面
    events = []
    with open(path, "r", encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            e = json.loads(line)
            if e.get("event_type") in ("auth", "web") and e.get("outcome") == "failure":
                events.append(e)
    print("[+] 向模型提交高危事件数：", len(events))
    result = summarize_events(events)
    print("\n===== LLM 分析摘要 =====\n")
    print(result)


if __name__ == "__main__":
    main()

# 采集后对镜像计算 SHA256，写入证据登记表
sha256sum memory_dump.raw > evidence/memory_dump.raw.sha256
# 记录：采集时间、采集人、采集工具版本、主机名、序列号

python3 -m venv venv && source venv/bin/activate
pip install volatility3
# 或源码安装（便于开发自定义插件）
git clone https://github.com/volatilityfoundation/volatility3.git
pip install -e ./volatility3

# 1) 识别镜像基本信息（系统版本、内存布局）
vol -f memory_dump.raw windows.info

# 2) 进程列表（找隐藏/异常进程，对比 EDR 进程树）
vol -f memory_dump.raw windows.pslist
vol -f memory_dump.raw windows.psscan      # 扫描已卸载/隐藏进程
vol -f memory_dump.raw windows.pstree      # 进程树，看清父子关系

# 3) 网络连接（定位 C2、横向移动）
vol -f memory_dump.raw windows.netscan
vol -f memory_dump.raw windows.netstat

# 4) 已加载模块 / 驱动（Rootkit、恶意驱动）
vol -f memory_dump.raw windows.modules
vol -f memory_dump.raw windows.driverscan

# 5) 命令行历史（攻击者敲过什么）
vol -f memory_dump.raw windows.cmdline
vol -f memory_dump.raw windows.consoles

# 6) 注册表（提取自动启动、RDP 配置等）
vol -f memory_dump.raw windows.registry.printkey --key "Software\Microsoft\Windows\CurrentVersion\Run"

# 7) 凭证（明文/哈希，需对应插件与系统版本）
vol -f memory_dump.raw windows.hashdump
vol -f memory_dump.raw windows.lsadump    # 提取 LSA secrets、明文密码（Win2008 前或 Wdigest 开启时）

# 8) 注入代码检测（早期恶意 Shellcode）
vol -f memory_dump.raw windows.malfind

# 9) 异常内存区域、未链接模块
vol -f memory_dump.raw windows.apihooks
vol -f memory_dump.raw windows.callbacks   # 系统回调（可能藏持久化）

pslist/pstree  →  发现一个父进程为 explorer.exe、却不在系统白名单的进程
      │
      ▼
cmdline/consoles  →  读它的命令行，看是否伪装成 svchost / 执行了 powershell -enc ...
      │
      ▼
netscan  →  该进程是否对外连了陌生 IP:端口（疑似 C2）
      │
      ▼
malfind --pid <pid> --dump  →  导出注入内存段
      │
      ▼
file  / sha256sum / yarascan  →  判定是否为已知恶意样本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
vol_runner.py
自动化批量运行 Volatility 3 插件，结构化收集证据并初筛高危信号。
仅用于授权环境的内存取证。
"""
import json
import subprocess
import re
from pathlib import Path

VOL = "vol"  # 确保 vol 在 PATH 中（Volatility 3）
IMAGE = "memory_dump.raw"
OUTDIR = Path("vol_output")
OUTDIR.mkdir(exist_ok=True)

# 要运行的 (插件名, 输出文件名)
PLUGINS = [
    ("windows.info", "info.txt"),
    ("windows.pstree", "pstree.txt"),
    ("windows.netscan", "netscan.txt"),
    ("windows.cmdline", "cmdline.txt"),
    ("windows.malfind", "malfind.txt"),
    ("windows.hashdump", "hashdump.txt"),
    ("windows.lsadump", "lsadump.txt"),
    ("windows.modules", "modules.txt"),
]


def run_plugin(plugin: str, outfile: str) -> str:
    cmd = [VOL, "-f", IMAGE, plugin]
    print(f"[*] 运行 {plugin} ...")
    try:
        res = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
        text = res.stdout + res.stderr
    except subprocess.TimeoutExpired:
        text = f"TIMEOUT running {plugin}"
    (OUTDIR / outfile).write_text(text, encoding="utf-8", errors="ignore")
    return text


def flag_suspicious(results: dict):
    """基于简单启发式，标出值得人工优先看的信号。"""
    flags = []

    # 1) netscan: 找 ESTABLISHED 的外连（非内网 RFC1918）
    netscan = results.get("netscan.txt", "")
    for line in netscan.splitlines():
        if "ESTABLISHED" in line:
            m = re.search(r'(\d+\.\d+\.\d+\.\d+):(\d+)', line)
            if m:
                ip = m.group(1)
                if not ip.startswith(("10.", "192.168.", "172.16.", "127.")):
                    flags.append(f"[外连] {line.strip()}")

    # 2) cmdline: 找 powershell -enc / base64 / 可疑下载
    cmdline = results.get("cmdline.txt", "")
    for line in cmdline.splitlines():
        low = line.lower()
        if any(k in low for k in ("powershell -enc", "iex ", "downloadstring",
                                   "invoke-webrequest", "certutil", "mimikatz")):
            flags.append(f"[可疑命令行] {line.strip()}")

    # 3) malfind: 存在条目即高优先
    malfind = results.get("malfind.txt", "")
    if "0x" in malfind and "PAGE_EXECUTE_READWRITE" in malfind:
        cnt = malfind.count("PAGE_EXECUTE_READWRITE")
        flags.append(f"[注入内存段] 发现 {cnt} 处 RWX 可执行可写内存")

    return flags


def main():
    results = {}
    for plugin, outfile in PLUGINS:
        results[outfile] = run_plugin(plugin, outfile)
    flags = flag_suspicious(results)
    report = {"image": IMAGE, "flags": flags}
    (OUTDIR / "flags.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print("\n" + "="*60)
    print("初步高危信号：")
    for f in flags:
        print("  ", f)
    print(f"\n[+] 详细输出见 {OUTDIR}/")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
自定义 Volatility 3 插件：mem_ioc_scan.py
扫描各进程可读取内存，提取 IP/域名/URL 作为内存侧 IOC 线索。
放置位置：volatility3/volatility3/framework/plugins/windows/mem_ioc_scan.py
仅用于授权环境取证。
"""
import re
from volatility3.framework import interfaces, renderers
from volatility3.framework.configuration import requirements
from volatility3.framework.renderers import format_hints

# 简单但够用的正则
IP_RE = re.compile(rb'\b(?:\d{1,3}\.){3}\d{1,3}\b')
DOMAIN_RE = re.compile(rb'\b(?:[a-z0-9-]+\.)+[a-z]{2,}\b', re.I)
URL_RE = re.compile(rb'https?://[^\s"\x00]+', re.I)


class MemIOCScan(interfaces.plugins.PluginInterface):
    """从进程内存提取潜在 IOC（IP/域名/URL）。"""

    _required_framework_version = (2, 0, 0)
    _version = (1, 0, 0)

    @classmethod
    def get_requirements(cls):
        return [
            requirements.ModuleRequirement(
                name="kernel", description="Windows 内核模块",
                architectures=["Intel32", "Intel64"]),
            requirements.IntRequirement(
                name="pid", description="仅扫描指定 PID（可选）", optional=True),
        ]

    def _scan_process(self, proc):
        import volatility3.framework.symbols as symbols  # noqa
        # 读取进程内存（简化：通过 proc.get_process_memory()）
        try:
            mem = proc.get_process_memory()
        except Exception:
            return []
        hits = set()
        try:
            for chunk in mem.read(0, 0x1000000):  # 最多读 16MB 示意
                for pat in (IP_RE, DOMAIN_RE, URL_RE):
                    for m in pat.findall(chunk):
                        hits.add(m.decode("utf-8", "ignore"))
        except Exception:
            pass
        return hits

    def run(self):
        kernel = self.context.modules[self.config["kernel"]]
        procs = list(kernel.object("nt!_EPROCESS", offset=0, native_layer_name="")
                     .cast("nt!_EPROCESS").traverse()) if False else []
        # 用框架标准方式枚举进程
        from volatility3.plugins.windows import pslist
        proc_gen = pslist.PsList.list_processes(self.context, kernel.layer_name,
                                                kernel.symbol_table_name)
        for proc in proc_gen:
            pid = proc.UniqueProcessId
            if self.config.get("pid") and pid != self.config["pid"]:
                continue
            name = proc.ImageFileName.cast("string", max_length=15)
            hits = self._scan_process(proc)
            for h in sorted(hits):
                yield (0, (format_hints.Hex(pid), name, h))


# 渲染器（框架要求）
MemIOCScan.parallelism = interfaces.plugins.PluginInterface.Parallelism.NoneType

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
decode_powershell.py
解码 powershell -enc 的 base64 命令并本地做初步语义标注；
可选地把「脱敏后的行为摘要」送 LLM 做进一步解释。仅用于授权取证。
"""
import base64
import re
import sys


def decode_enc(b64: str) -> str:
    # PowerShell -enc 使用 UTF-16LE
    try:
        raw = base64.b64decode(b64.strip())
        return raw.decode("utf-16-le", errors="ignore")
    except Exception as e:
        return f"[解码失败] {e}"


def local_annotate(plain: str) -> list:
    """本地启发式标注危险行为，避免先依赖外部模型。"""
    notes = []
    low = plain.lower()
    signals = {
        "downloadstring": "从远程下载并执行脚本（典型恶意拉取）",
        "invoke-webrequest": "发起 HTTP 请求（可能下载载荷）",
        "iex": "在内存中执行脚本（不落盘，规避文件扫描）",
        "certutil": "可能用 certutil 下载/解码恶意文件",
        "net user": "可能创建/修改账户",
        "schtasks": "可能创建计划任务（持久化）",
        "regsvr32": "可能通过 regsvr32 执行 COM 脚本（绕过）",
        "wmic": "可能通过 WMI 执行命令",
        "encodedcommand": "自身仍被编码（嵌套混淆）",
    }
    for k, v in signals.items():
        if k in low:
            notes.append(v)
    return notes


def main():
    if len(sys.argv) < 2:
        print("用法: python3 decode_powershell.py '<base64>'")
        return
    b64 = sys.argv[1]
    plain = decode_enc(b64)
    print("===== 解码后明文 =====")
    print(plain[:4000])
    print("\n===== 本地行为标注 =====")
    for n in local_annotate(plain):
        print("  -", n)


if __name__ == "__main__":
    main()

# 在受影响主机上抓包（示意，需授权）
tcpdump -i eth0 -w capture.pcap -G 3600 -W 1 'host 1.2.3.4 or port 445'

# 用 tshark 快速看协议分布
tshark -r capture.pcap -q -z io,phs

# 1) 抽取所有 TCP 会话（五元组 + 字节数）—— 看清谁和谁聊了多久
tshark -r capture.pcap -T fields \
  -e frame.time_epoch -e ip.src -e tcp.srcport -e ip.dst -e tcp.dstport \
  -e tcp.len -e _ws.col.Protocol \
  -Y "tcp" -E separator=, -E quote=d > conns.csv

# 2) 抽取 HTTP 请求（URL、Host、User-Agent、 referrer）
tshark -r capture.pcap -Y "http.request" -T fields \
  -e frame.time_epoch -e ip.src -e ip.dst -e http.host -e http.request.method \
  -e http.request.uri -e http.user_agent -e http.referer \
  -E separator=, -E quote=d > http_reqs.csv

# 3) 抽取 DNS 查询（找可疑域名、C2 域名生成算法 DGA 痕迹）
tshark -r capture.pcap -Y "dns.qry.name" -T fields \
  -e frame.time_epoch -e ip.src -e dns.qry.name -e dns.resp.addr \
  -E separator=, -E quote=d > dns.csv

# 4) 抽取 TLS 握手信息（SNI、JA3/JA4 指纹）—— 即使加密也能看到"和谁握手"
tshark -r capture.pcap -Y "tls.handshake" -T fields \
  -e frame.time_epoch -e ip.src -e ip.dst -e tls.handshake.extensions_server_name \
  -e tls.handshake.type -E separator=, -E quote=d > tls.csv

# 5) 导出所有 HTTP 对象（图片/脚本/上传文件）到大概率含载荷的目录
tshark -r capture.pcap --export-objects "http,http_objects" -q

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
pcap_correlate.py
读取 tshark 导出的连接 CSV，重建会话、识别攻击阶段、生成时间线。
仅用于授权环境的流量回溯分析。
"""
import csv
from collections import defaultdict
from datetime import datetime


def load_conns(path):
    rows = []
    with open(path, newline="", encoding="utf-8") as f:
        r = csv.DictReader(f)
        for row in r:
            try:
                row["epoch"] = float(row["frame.time_epoch"])
            except (ValueError, KeyError):
                row["epoch"] = 0.0
            rows.append(row)
    rows.sort(key=lambda x: x["epoch"])
    return rows


def identify_stage(conn):
    """对单条连接做粗粒度阶段判定（可扩展）。"""
    proto = (conn.get("_ws.col.Protocol") or "").upper()
    src, dst = conn.get("ip.src"), conn.get("ip.dst")
    dport = conn.get("tcp.dstport")
    # 扫描：连接大量不同目的端口
    # （此处仅单条占位，真正的扫描识别在聚合层做）
    return proto, dport


def aggregate(rows):
    """按 (src,dst) 聚合，统计端口分布与字节。"""
    sess = defaultdict(lambda: {
        "ports": defaultdict(int), "bytes": 0, "count": 0,
        "first": None, "last": None,
    })
    for c in rows:
        key = (c.get("ip.src"), c.get("ip.dst"))
        s = sess[key]
        dp = c.get("tcp.dstport")
        if dp:
            s["ports"][dp] += 1
        try:
            s["bytes"] += int(c.get("tcp.len") or 0)
        except ValueError:
            pass
        s["count"] += 1
        if s["first"] is None:
            s["first"] = c["epoch"]
        s["last"] = c["epoch"]
    return sess


def detect_scan(sess, port_threshold=15):
    """同源 IP 对单一目的 IP 连接超过 N 个不同端口 → 疑似端口扫描。"""
    findings = []
    for (src, dst), s in sess.items():
        if len(s["ports"]) >= port_threshold:
            findings.append({
                "type": "port_scan",
                "src": src, "dst": dst,
                "distinct_ports": len(s["ports"]),
                "window": f"{s['first']}->{s['last']}",
                "sample_ports": sorted(s["ports"].keys(), key=lambda x: int(x))[:20],
            })
    return findings


def detect_exfil(sess, byte_threshold=10_000_000):
    """单会话出站字节异常大 → 疑似数据外传（需结合业务基线判断）。"""
    findings = []
    for (src, dst), s in sess.items():
        if s["bytes"] >= byte_threshold:
            findings.append({
                "type": "possible_exfil",
                "src": src, "dst": dst,
                "bytes": s["bytes"],
                "window": f"{s['first']}->{s['last']}",
            })
    return findings


def main():
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else "conns.csv"
    rows = load_conns(path)
    print(f"[+] 载入连接记录 {len(rows)} 条")
    sess = aggregate(rows)
    print(f"[+] 重建会话 {len(sess)} 个")
    findings = detect_scan(sess) + detect_exfil(sess)
    for f in findings:
        print("="*50)
        for k, v in f.items():
            print(f"  {k}: {v}")
    print(f"\n[+] 识别疑似事件 {len(findings)} 条")


if __name__ == "__main__":
    main()

# 导出所有 HTTP 传输的对象到 ./http_objects
tshark -r capture.pcap --export-objects "http,http_objects" -q

# 导出 SMTP 附件
tshark -r capture.pcap --export-objects "smtp,eml_export" -q

# 导出 SMB 文件（横向移动/共享投毒）
tshark -r capture.pcap --export-objects "smb,smb_objects" -q

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
stream_reassemble.py
用 scapy 按四元组重组 TCP 流，并从 HTTP 响应中导出正文保存为文件。
仅用于授权环境的流量取证与样本提取。
"""
from scapy.all import rdpcap, TCP
from collections import defaultdict
import re


def reassemble(path):
    pkts = rdpcap(path)
    streams = defaultdict(dict)  # (src,dst,sport,dport) -> {seq: payload}
    for p in pkts:
        if TCP in p and p[TCP].payload:
            key = (p[TCP].sport, p[TCP].dport,
                   p["IP"].src, p["IP"].dst)
            seq = p[TCP].seq
            streams[key][seq] = bytes(p[TCP].payload)
    return streams


def stream_to_bytes(stream):
    """按 seq 顺序拼装（简化：忽略重传/乱序纠错）。"""
    ordered = sorted(stream.items())
    return b"".join(data for _, data in ordered)


def extract_http_body_from_server(streams, outdir="extracted"):
    """对每个服务端→客户端的流，尝试抽 HTTP body。"""
    import os
    os.makedirs(outdir, exist_ok=True)
    saved = 0
    for key, stream in streams.items():
        data = stream_to_bytes(stream)
        if b"HTTP/" in data and b"\r\n\r\n" in data:
            header, _, body = data.partition(b"\r\n\r\n")
            # 仅保存看起来像二进制/脚本的内容
            if len(body) > 200:
                fn = f"{key[2]}_{key[0]}_{key[1]}_{key[3]}.bin"
                with open(f"{outdir}/{fn}", "wb") as f:
                    f.write(body)
                saved += 1
    return saved


def main():
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else "capture.pcap"
    print("[*] 重组 TCP 流 ...")
    streams = reassemble(path)
    print(f"[*] 共 {len(streams)} 条流")
    n = extract_http_body_from_server(streams)
    print(f"[+] 提取疑似文件 {n} 个，存于 ./extracted")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
beacon_detector.py
基于连接时间戳的周期性与字节熵，识别疑似 C2 心跳流量。
仅用于授权环境的流量分析。
"""
import csv
import math
from collections import defaultdict


def shannon_entropy(data: bytes) -> float:
    if not data:
        return 0.0
    freq = defaultdict(int)
    for b in data:
        freq[b] += 1
    n = len(data)
    return -sum((c / n) * math.log2(c / n) for c in freq.values())


def load_timestamps(path, src_ip=None):
    """读取 tshark 导出的连接 CSV，按 (src,dst) 收集时间戳。"""
    series = defaultdict(list)
    with open(path, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            try:
                t = float(row["frame.time_epoch"])
            except (ValueError, KeyError):
                continue
            s, d = row.get("ip.src"), row.get("ip.dst")
            if src_ip and s != src_ip:
                continue
            series[(s, d)].append(t)
    return series


def period_score(times, tol=2.0):
    """计算时间序列的周期稳定性（0~1，越高越像固定心跳）。"""
    if len(times) < 5:
        return 0.0, 0.0
    times = sorted(times)
    gaps = [times[i + 1] - times[i] for i in range(len(times) - 1)]
    mean = sum(gaps) / len(gaps)
    if mean == 0:
        return 0.0, 0.0
    # 方差越小越规律
    var = sum((g - mean) ** 2 for g in gaps) / len(gaps)
    cv = math.sqrt(var) / mean  # 变异系数
    score = max(0.0, 1.0 - cv / tol)
    return score, mean


def main():
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else "conns.csv"
    series = load_timestamps(path)
    print(f"[+] 分析 {len(series)} 条流的时间周期性\n")
    for (s, d), ts in series.items():
        score, mean_gap = period_score(ts)
        if score > 0.6:
            print(f"  [疑似心跳] {s} -> {d}  周期分数={score:.2f} "
                  f"平均间隔={mean_gap:.1f}s  连接数={len(ts)}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ioc_extractor.py
从文本证据中自动提取 IOC（IP/域名/URL/邮箱/hash/比特币地址）。
支持白名单与私有网段过滤，输出结构化 JSON。
仅用于授权环境的 IOC 收集。
"""
import re
import json
import ipaddress
from pathlib import Path


IPV4_RE = re.compile(r'\b(?:(?:25[0-5]|2[0-4]\d|1?\d?\d)\.){3}(?:25[0-5]|2[0-4]\d|1?\d?\d)\b')
DOMAIN_RE = re.compile(r'\b(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,24}\b', re.I)
URL_RE = re.compile(r'https?://[^\s"\x27<>]+', re.I)
EMAIL_RE = re.compile(r'[a-z0-9._%+-]+@[a-z0-9.-]+\.[a-z]{2,}', re.I)
BTC_RE = re.compile(r'\b(?:bc1|[13])[a-zA-HJ-NP-Z0-9]{25,39}\b')
HASH_RE = re.compile(r'\b(?:[a-fA-F0-9]{32}|[a-fA-F0-9]{40}|[a-fA-F0-9]{64})\b')
B64_RE = re.compile(r'\b(?:[A-Za-z0-9+/]{40,}={0,2})\b')


# 常见白名单域名（自家业务/可信 CDN，按需扩充）
WHITELIST_DOMAINS = {
    "microsoft.com", "windows.com", "office.com", "google.com",
    "gstatic.com", "googleapis.com", "cloudflare.com", "w3.org",
    "trustedsite.example", "yourcorp.com",
}


def is_private_ip(ip: str) -> bool:
    try:
        return ipaddress.ip_address(ip).is_private
    except ValueError:
        return False


def extract(text: str, enable_b64=False):
    iocs = {"ip": set(), "domain": set(), "url": set(),
            "email": set(), "btc": set(), "hash": set(), "b64": set()}
    for m in IPV4_RE.findall(text):
        if not is_private_ip(m):
            iocs["ip"].add(m)
    for m in DOMAIN_RE.findall(text):
        dom = m.lower()
        # 过滤白名单及白名单子域
        if any(dom == w or dom.endswith("." + w) for w in WHITELIST_DOMAINS):
            continue
        # 过滤纯数字 TLD 误报
        if re.match(r'^\d', dom.split('.')[-1]):
            continue
        iocs["domain"].add(dom)
    for m in URL_RE.findall(text):
        iocs["url"].add(m)
    for m in EMAIL_RE.findall(text):
        iocs["email"].add(m.lower())
    for m in BTC_RE.findall(text):
        iocs["btc"].add(m)
    for m in HASH_RE.findall(text):
        iocs["hash"].add(m.lower())
    if enable_b64:
        for m in B64_RE.findall(text):
            iocs["b64"].add(m)
    # 把 set 转 list 便于序列化
    return {k: sorted(v) for k, v in iocs.items()}


def main():
    import sys
    if len(sys.argv) < 2:
        print("用法: python3 ioc_extractor.py <evidence.txt> [--b64]")
        return
    txt = Path(sys.argv[1]).read_text(encoding="utf-8", errors="ignore")
    enable_b64 = "--b64" in sys.argv
    result = extract(txt, enable_b64)
    print(json.dumps(result, ensure_ascii=False, indent=2))
    # 统计
    total = sum(len(v) for v in result.values())
    print(f"\n[+] 共提取 IOC {total} 条")


if __name__ == "__main__":
    main()

# 从内存 netscan + cmdline 文本里挖 IOC
cat vol_output/netscan.txt vol_output/cmdline.txt > evidence_mem.txt
python3 ioc_extractor.py evidence_mem.txt > ioc_from_mem.json

# 从解码后的 powershell 明文里挖
python3 ioc_extractor.py decoded_ps.txt --b64 > ioc_from_ps.json

# 从 HTTP 请求 CSV 里挖域名/URL
python3 ioc_extractor.py http_reqs.csv > ioc_from_http.json

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
sample_hasher.py
对目录下的可疑样本批量计算多算法哈希，并输出 IOC JSON。
仅用于授权环境的样本指纹化。
"""
import hashlib
import json
import sys
from pathlib import Path


def hash_file(path: Path):
    md5, sha1, sha256 = hashlib.md5(), hashlib.sha1(), hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(1 << 16), b""):
            md5.update(chunk); sha1.update(chunk); sha256.update(chunk)
    return {
        "filename": path.name,
        "size": path.stat().st_size,
        "md5": md5.hexdigest(),
        "sha1": sha1.hexdigest(),
        "sha256": sha256.hexdigest(),
    }


def main():
    d = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("extracted")
    out = []
    for p in d.rglob("*"):
        if p.is_file():
            out.append(hash_file(p))
    print(json.dumps(out, indent=2))
    print(f"\n[+] 已指纹化 {len(out)} 个样本")


if __name__ == "__main__":
    main()

rule APT_CobaltStrike_Beacon_Generic
{
    meta:
        author = "BlueTeam-IR"
        description = "通用型 Cobalt Strike Beacon 内存/样本特征（用于检测，非攻击）"
        reference = "https://www.cobaltstrike.com/"

    strings:
        $a1 = "*** BATMAN ***" ascii   // 早期 CS 配置标记
        $a2 = "ReflectiveLoader" ascii
        $a3 = "beacon.x64.dll" ascii
        $a4 = { 56 57 53 48 83 EC 28 48 8B 05 }  // 典型 prologue 片段（示意）
        $u1 = "POST /activity" ascii wide
        $u2 = "C:\\Windows\\System32\\" ascii

    condition:
        uint16(0) == 0x5a4d and (3 of ($a*) or 2 of ($u*))
}

# 对本机/样本目录跑 YARA
yara64 -r cobalt_strike.yar extracted/ > yara_hits.txt

# 在 Volatility 导出的内存段上跑（配合 yarascan 插件）
vol -f memory_dump.raw windows.yarascan.Yarascan --yara-rules cobalt_strike.yar

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ioc_enrich.py
把提取的 IOC 送威胁情报平台做碰撞/富化，返回置信与标签。
安全约束：仅对接你拥有合法使用权的情报源；不硬编码密钥。
仅用于授权环境的 IOC 富化。
"""
import json
import os
import urllib.request
import urllib.parse


def query_ip(ip: str, api_base: str, api_key: str) -> dict:
    """查询 IP 情报（此处以通用 REST 风格示意，对接你的合规源）。"""
    url = f"{api_base.rstrip('/')}/ip/{ip}"
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {api_key}"})
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            return json.loads(r.read().decode("utf-8"))
    except Exception as e:
        return {"error": str(e)[:120]}


def enrich_iocs(ioc_file: str, api_base: str, api_key: str):
    with open(ioc_file, encoding="utf-8") as f:
        iocs = json.load(f)
    results = {"ip": {}}
    for ip in iocs.get("ip", []):
        results["ip"][ip] = query_ip(ip, api_base, api_key)
    # 落盘富化结果
    out = ioc_file.replace(".json", "_enriched.json")
    with open(out, "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)
    print(f"[+] 富化完成，结果见 {out}")
    return results


def main():
    import sys
    if len(sys.argv) < 2:
        print("用法: python3 ioc_enrich.py <ioc.json>")
        print("环境变量 TI_BASE_URL / TI_API_KEY 需指向你的合规情报源")
        return
    base = os.getenv("TI_BASE_URL", "")
    key = os.getenv("TI_API_KEY", "")
    if not base or not key:
        print("[!] 未配置 TI_BASE_URL / TI_API_KEY，跳过富化（仅演示）")
        return
    enrich_iocs(sys.argv[1], base, key)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
stix_export.py
把 IOC 集合导出为 STIX 2.1 风格的 JSON（可被 MISP/TAXII 消费）。
仅用于授权环境的 IOC 共享。
"""
import json
import uuid
import sys
from datetime import datetime, timezone


def make_bundle(iocs: dict):
    now = datetime.now(timezone.utc).isoformat()
    objects = []
    for ip in iocs.get("ip", []):
        objects.append({
            "type": "ipv4-addr", "spec_version": "2.1",
            "id": f"ipv4-addr--{uuid.uuid4()}", "value": ip,
            "created": now, "modified": now,
            "labels": ["malicious-activity"],
        })
    for d in iocs.get("domain", []):
        objects.append({
            "type": "domain-name", "spec_version": "2.1",
            "id": f"domain-name--{uuid.uuid4()}", "value": d,
            "created": now, "modified": now,
            "labels": ["malicious-activity"],
        })
    for h in iocs.get("hash", []):
        objects.append({
            "type": "file", "spec_version": "2.1",
            "id": f"file--{uuid.uuid4()}",
            "hashes": {"SHA-256": h}, "created": now, "modified": now,
            "labels": ["malicious-activity"],
        })
    return {"type": "bundle", "id": f"bundle--{uuid.uuid4()}", "objects": objects}


def main():
    path = sys.argv[1] if len(sys.argv) > 1 else "ioc.json"
    iocs = json.loads(open(path, encoding="utf-8").read())
    bundle = make_bundle(iocs)
    out = path.replace(".json", "_stix.json")
    json.dump(bundle, open(out, "w", encoding="utf-8"), ensure_ascii=False, indent=2)
    print(f"[+] 已导出 STIX 包：{out}（{len(bundle['objects'])} 个对象）")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
timeline_builder.py
合并多源事件（日志/内存/流量导出的 JSONL/CSV）为统一时间线，
并标注入口点候选。仅用于授权环境的攻击链还原。
"""
import json
import csv
import sys
from datetime import datetime
from collections import defaultdict


def parse_iso(ts):
    try:
        return datetime.fromisoformat(ts)
    except Exception:
        return None


def load_jsonl(path):
    out = []
    with open(path, encoding="utf-8") as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                e = json.loads(line)
                out.append({"time": e.get("event_time"), "src": e.get("src_ip"),
                            "type": e.get("event_type"), "detail": e.get("raw", "")})
            except json.JSONDecodeError:
                continue
    return out


def load_ioc_csv(path, kind):
    """把 IOC 提取结果（csv）并入时间线（时间字段为空则标记为无时间戳 IOC）。"""
    out = []
    with open(path, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            out.append({"time": row.get("time", ""),
                        "src": row.get("ip") or row.get("src"),
                        "type": kind, "detail": json.dumps(row, ensure_ascii=False)})
    return out


def build_timeline(sources: list):
    events = []
    for s in sources:
        events.extend(s)
    # 按时间排序（无时间戳的排最后）
    def keyf(e):
        t = parse_iso(e.get("time") or "")
        return t or datetime.max
    events.sort(key=keyf)
    return events


def find_entry_candidates(events, external_filter=None):
    """入口点候选：来自外部 IP 的最早高危事件。"""
    cands = []
    for e in events:
        src = e.get("src")
        if not src:
            continue
        if external_filter and not external_filter(src):
            continue
        # 高危类型优先
        if e.get("type") in ("auth_failure", "web_attack", "external_conn", "phish"):
            cands.append(e)
    return cands[:10]


def is_external(ip):
    if not ip:
        return False
    import ipaddress
    try:
        return not (ipaddress.ip_address(ip).is_private
                    or ipaddress.ip_address(ip).is_loopback)
    except ValueError:
        return False


def main():
    print("[*] 加载多源事件 ...")
    sources = []
    for arg in sys.argv[1:]:
        if arg.endswith(".jsonl"):
            sources.append(load_jsonl(arg))
        elif arg.endswith(".csv"):
            sources.append(load_ioc_csv(arg, "ioc"))
    timeline = build_timeline(sources)
    print(f"[+] 时间线共 {len(timeline)} 个事件\n")
    print("="*70)
    for e in timeline[:50]:
        print(f"  {e.get('time') or 'NO_TS':22s} | {str(e.get('type')):14s} | "
              f"{str(e.get('src')):16s} | {str(e.get('detail'))[:50]}")
    print("\n--- 入口点候选（外部 IP 的最早高危事件）---")
    for c in find_entry_candidates(timeline, is_external):
        print(f"  {c.get('time')} | {c.get('src')} | {c.get('detail')[:60]}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
attacker_profile.py
基于已收集的 IOC 与阶段事件，生成结构化「对手行为画像草案」。
仅用于授权环境的防御性归因分析。
"""
import json
import sys
from collections import Counter


# 简化版 ATT&CK 技法映射（按需扩充）
TTP_MAP = {
    "port_scan": "TA0043 Reconnaissance / T1595 Active Scanning",
    "bruteforce_then_success": "TA0006 Credential Access / T1110 Brute Force",
    "web_path_scan": "TA0043 Reconnaissance / T1595",
    "possible_exfil": "TA0010 Exfiltration / T1041 Exfil over C2",
    "powershell_enc": "TA0002 Execution / T1059.001 PowerShell",
    "cobaltstrike": "TA0011 Command and Control / T1071 Application Layer Protocol",
    "mimikatz": "TA0006 Credential Access / T1003 OS Credential Dumping",
    "schtasks": "TA0003 Persistence / T1053 Scheduled Task",
}


def build_profile(ioc_file, findings_file=None):
    iocs = json.loads(open(ioc_file, encoding="utf-8").read())
    findings = []
    if findings_file:
        findings = json.loads(open(findings_file, encoding="utf-8").read())

    ttp_counter = Counter()
    infra = {"ip": iocs.get("ip", []), "domain": iocs.get("domain", []),
             "url": iocs.get("url", []), "hash": iocs.get("hash", [])}
    for f in findings:
        pat = f.get("pattern") or f.get("type")
        if pat in TTP_MAP:
            ttp_counter[TTP_MAP[pat]] += 1

    profile = {
        "observed_ttps": dict(ttp_counter),
        "infrastructure": {k: len(v) for k, v in infra.items()},
        "sample_iocs": {k: v[:5] for k, v in infra.items()},
        "confidence_note": "行为画像基于证据，不代表已确认身份；"
                           "身份归因需结合威胁情报与执法/合规流程。",
    }
    return profile


def main():
    if len(sys.argv) < 2:
        print("用法: python3 attacker_profile.py <ioc.json> [findings.json]")
        return
    profile = build_profile(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else None)
    print(json.dumps(profile, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
infra_pivot.py
针对 C2 域名/IP 做被动情报反查（PDNS / CT / RDAP）。
安全约束：仅对接合规、已授权的情报源；不硬编码密钥。
仅用于授权环境的防御性归因。
"""
import json
import os
import urllib.request
import sys


def api_get(url, api_key):
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {api_key}"})
    try:
        with urllib.request.urlopen(req, timeout=15) as r:
            return json.loads(r.read().decode("utf-8"))
    except Exception as e:
        return {"error": str(e)[:120]}


def pivot_domain(domain, base, key):
    """查该域名的 Passive DNS 历史解析。"""
    url = f"{base.rstrip('/')}/pdns/{domain}"
    return api_get(url, key)


def pivot_ip(ip, base, key):
    """查该 IP 上出现过的其他域名（反向 PDNS）。"""
    url = f"{base.rstrip('/')}/pdns/ip/{ip}"
    return api_get(url, key)


def main():
    if len(sys.argv) < 3:
        print("用法: python3 infra_pivot.py <domain|ip> <value>")
        print("需配置环境变量 TI_BASE_URL / TI_API_KEY（你的合规源）")
        return
    kind, value = sys.argv[1], sys.argv[2]
    base = os.getenv("TI_BASE_URL", "")
    key = os.getenv("TI_API_KEY", "")
    if not base or not key:
        print("[!] 未配置情报源，仅演示流程")
        return
    if kind == "domain":
        print(json.dumps(pivot_domain(value, base, key), ensure_ascii=False, indent=2))
    else:
        print(json.dumps(pivot_ip(value, base, key), ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
geoip_lookup.py
对 IOC 中的 IP 做 GeoIP + ASN 查询，辅助画像（非身份认定）。
需本地放置 GeoLite2-City.mmdb / GeoLite2-ASN.mmdb（合规获取）。
仅用于授权环境的防御性分析。
"""
import json
import sys


def lookup(ip, city_db, asn_db):
    import geoip2.database
    out = {"ip": ip}
    try:
        with geoip2.database.Reader(city_db) as r:
            c = r.city(ip)
            out["country"] = c.country.iso_code
            out["city"] = c.city.name
            out["lat"] = c.location.latitude
            out["lon"] = c.location.longitude
    except Exception as e:
        out["geo_error"] = str(e)[:80]
    try:
        with geoip2.database.Reader(asn_db) as r:
            a = r.asn(ip)
            out["asn"] = a.autonomous_system_number
            out["asn_org"] = a.autonomous_system_organization
    except Exception as e:
        out["asn_error"] = str(e)[:80]
    return out


def main():
    if len(sys.argv) < 3:
        print("用法: python3 geoip_lookup.py <ioc.json> <city_db> <asn_db>")
        return
    iocs = json.loads(open(sys.argv[1], encoding="utf-8").read())
    city_db, asn_db = sys.argv[2], sys.argv[3]
    results = [lookup(ip, city_db, asn_db) for ip in iocs.get("ip", [])]
    print(json.dumps(results, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    main()

# 1) 归一化
cat nginx_access.log auth.log | python3 log_normalizer.py > events.jsonl

# 2) 异常检测：找出最可疑的源 IP
python3 log_anomaly.py events.jsonl
# 输出：203.0.113.45  anomaly_score=0.91  ← 明显异常

# 3) 关联规则：发现"爆破后成功"
python3 log_correlate.py events.jsonl
# 输出 pattern=bruteforce_then_success, src=203.0.113.45,
#        compromised_actor=svc_backup,  fail_count=312

# 自动化运行插件
python3 vol_runner.py
# flags.json 显示：
#   [外连] 203.0.113.45:443 ESTABLISHED  (PID 4488, 伪装为 svchost)
#   [可疑命令行] powershell -enc JABjAD0ATgBlAHcALg...(长 base64)
#   [注入内存段] 发现 3 处 RWX 可执行可写内存

# 解码那条 powershell
python3 decode_powershell.py "<上面那段base64>"
# 解码后：从 http://203.0.113.45/loader.bin 下载并执行 Beacon

# 把 Beacon 配置里的 C2 域名从内存扫出来
vol -f dev07.raw windows.yarascan.Yarascan --yara-rules cs.yar

# 抽取 HTTP 对象，拿回 loader.bin
tshark -r border.pcap --export-objects "http,http_objects" -q

# 周期性检测：确认 C2 心跳
python3 beacon_detector.py conns.csv
# 输出：[疑似心跳] DEV-07 -> 203.0.113.45  周期分数=0.87 平均间隔=60.0s

# 扫描：发现该主机对内网 445 端口的大面积连接
python3 pcap_correlate.py conns.csv
# 输出 type=port_scan, src=DEV-07, distinct_ports=38  → 内网横向扫描

# 汇总所有文本证据，提取 IOC
cat vol_output/cmdline.txt vol_output/netscan.txt http_reqs.csv \
  > evidence_all.txt
python3 ioc_extractor.py evidence_all.txt > ioc.json
# ioc.json: ip=[203.0.113.45], domain=[evil-c2.example, updater.example],
#           url=[http://203.0.113.45/loader.bin], hash=[...]

# 样本指纹化
python3 sample_hasher.py http_objects > samples.json
# 拿到 loader.bin 的 sha256

# 导出 STIX 供 SIEM/MISP 消费
python3 stix_export.py ioc.json

# 统一时间线
python3 timeline_builder.py events.jsonl ioc.csv > timeline.txt
# 入口点候选：203.0.113.45 对 svc_backup 的爆破-成功事件（最早）

# 画像
python3 attacker_profile.py ioc.json findings.json > profile.json
# observed_ttps: T1110 Brute Force, T1059.001 PowerShell,
#                T1071 App Layer C2, T1021.002 SMB 横向

# 基础设施反查（对接合规源）
python3 infra_pivot.py domain evil-c2.example
# 发现同证书下还有 3 个未曝光域名 → 扩大封禁范围

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ir_pipeline.py
端到端事件响应编排：日志→内存→流量→IOC→时间线→画像。
所有子模块均为本文前述脚本；仅用于授权环境。
"""
import subprocess
import sys
import json
from pathlib import Path

STEPS = [
    ("归一化日志", ["python3", "log_normalizer.py"],
     ["cat", "nginx_access.log", "auth.log"], "events.jsonl"),
    ("异常检测", ["python3", "log_anomaly.py", "events.jsonl"], None, None),
    ("关联规则", ["python3", "log_correlate.py", "events.jsonl"], None, None),
    ("内存取证", ["python3", "vol_runner.py"], None, None),
    ("命令解码", ["python3", "decode_powershell.py", "<B64>"], None, None),
    ("流量抽取", ["tshark", "-r", "border.pcap", "--export-objects",
                  "http,http_objects", "-q"], None, None),
    ("C2 心跳", ["python3", "beacon_detector.py", "conns.csv"], None, None),
    ("IOC 提取", ["python3", "ioc_extractor.py", "evidence_all.txt"], None, "ioc.json"),
    ("画像", ["python3", "attacker_profile.py", "ioc.json"], None, "profile.json"),
    ("时间线", ["python3", "timeline_builder.py", "events.jsonl", "ioc.csv"],
     None, "timeline.txt"),
]


def run(cmd, stdin_file=None, out_file=None):
    print(f"\n>>> 执行: {' '.join(cmd)}")
    inp = None
    if stdin_file and Path(stdin_file).exists():
        inp = open(stdin_file, "rb").read()
    res = subprocess.run(cmd, input=inp, capture_output=True, text=True)
    if out_file:
        Path(out_file).write_text(res.stdout + res.stderr, encoding="utf-8")
    else:
        print(res.stdout[:2000])
    return res.returncode


def main():
    print("="*70)
    print("  端到端事件响应流水线（教学演示，需替换为真实证据路径）")
    print("="*70)
    for name, cmd, stdin_f, out_f in STEPS:
        print(f"\n### 步骤：{name}")
        run(cmd, stdin_f, out_f)
    print("\n[+] 流水线执行完毕。请人工复核各阶段输出并生成归因草案。")


if __name__ == "__main__":
    main()

[采集层]  EDR / 防火墙镜像 / 日志转发
    │
    ▼
[分析层]  ┌─ 日志归一化+异常 (log_normalizer / log_anomaly)
          ├─ 内存自动化 (vol_runner + 自定义插件)
          ├─ 流量回溯 (tshark + pcap_correlate + beacon_detector)
          └─ AI 辅助 (本地 LLM 摘要 / 聚类去噪)
    │
    ▼
[IOC 层]  ioc_extractor + sample_hasher + YARA + 情报富化 + STIX
    │
    ▼
[关联层]  timeline_builder + attacker_profile + infra_pivot
    │
    ▼
[交付层]  TheHive(案件) / MISP(IOC) / SIEM(告警规则) / 报告生成

# —— 日志 ——
python3 log_normalizer.py < raw.log > events.jsonl
python3 log_anomaly.py events.jsonl
python3 log_correlate.py events.jsonl

# —— 内存（Volatility 3）——
vol -f mem.raw windows.pstree
vol -f mem.raw windows.netscan
vol -f mem.raw windows.cmdline
vol -f mem.raw windows.malfind --dump
vol -f mem.raw windows.hashdump
vol -f mem.raw windows.lsadump
vol -f mem.raw windows.yarascan.Yarascan --yara-rules cs.yar
python3 vol_runner.py
python3 decode_powershell.py "<b64>"

# —— 流量 ——
tshark -r cap.pcap -q -z io,phs
tshark -r cap.pcap -Y "http.request" -T fields -e http.host -e http.request.uri
tshark -r cap.pcap --export-objects "http,http_objects" -q
tshark -r cap.pcap -Y "dns.qry.name" -T fields -e dns.qry.name
python3 pcap_correlate.py conns.csv
python3 beacon_detector.py conns.csv
python3 stream_reassemble.py cap.pcap

# —— IOC ——
python3 ioc_extractor.py evidence.txt > ioc.json
python3 sample_hasher.py extracted/ > samples.json
yara64 -r cs.yar extracted/
python3 ioc_enrich.py ioc.json
python3 stix_export.py ioc.json

# —— 归因 ——
python3 timeline_builder.py events.jsonl ioc.csv > timeline.txt
python3 attacker_profile.py ioc.json findings.json
python3 infra_pivot.py domain evil.example
python3 geoip_lookup.py ioc.json GeoLite2-City.mmdb GeoLite2-ASN.mmdb

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ewma_anomaly.py
基于指数加权移动平均的轻量时序异常检测，适合流式日志。
仅用于授权环境的异常发现。
"""
from collections import defaultdict
import math


class EWMA:
    def __init__(self, alpha=0.3, threshold=3.0):
        self.alpha = alpha          # 平滑系数，越大越敏感
        self.threshold = threshold  # 多少倍标准差算异常
        self.mean = None
        self.var = None

    def update(self, x):
        if self.mean is None:
            self.mean = x
            self.var = 0.0
            return 0.0
        prev_mean = self.mean
        self.mean = self.alpha * x + (1 - self.alpha) * prev_mean
        self.var = self.alpha * (x - prev_mean) ** 2 + (1 - self.alpha) * self.var
        std = math.sqrt(self.var) if self.var > 0 else 0.0
        if std == 0:
            return 0.0
        return (x - self.mean) / std  # z-score


def main():
    # 示例：模拟某 IP 的请求数时序（前段平稳，中段突增）
    series = ([20] * 30) + ([20, 20, 220, 240, 200]) + ([20] * 10)
    detectors = defaultdict(lambda: EWMA(alpha=0.3, threshold=3.0))
    for t, val in enumerate(series):
        z = detectors["ip_203_0_113_45"].update(val)
        if abs(z) >= 3.0:
            print(f"t={t} 异常! 值={val} z={z:.2f}")


if __name__ == "__main__":
    main()

# Linux：先准备带符号的内核 profile（需要目标系统的 System.map / vmlinux）
vol -f linux_mem.raw linux.pslist
vol -f linux_mem.raw linux.netstat
vol -f linux_mem.raw linux.bash_history   # 读 bash 历史（即使文件被删）
vol -f linux_mem.raw linux.check_syscall  # 检查 syscall 表钩子

# macOS
vol -f mac_mem.raw mac.pslist
vol -f mac_mem.raw mac.netstat
vol -f mac_mem.raw mac.check_mig_table

# tshark 直接输出 JA3（需编译时支持）
tshark -r cap.pcap -Y "tls" -T fields \
  -e ip.src -e ip.dst -e tcp.dstport -e ja3.hash -e ja3s.hash

# 或用 zeek 的 ssl.log（自动带 ja3）
zeek -r cap.pcap  # 生成 ssl.log，含 ja3 字段

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
dns_tunnel_detect.py
基于子域名长度、熵、频率的 DNS 隧道启发式检测。
仅用于授权环境的流量分析。
"""
import math
import csv
from collections import defaultdict


def entropy(s: str) -> float:
    if not s:
        return 0.0
    freq = defaultdict(int)
    for c in s:
        freq[c] += 1
    n = len(s)
    return -sum((c / n) * math.log2(c / n) for c in freq.values())


def load_dns(path):
    q = defaultdict(list)
    with open(path, newline="", encoding="utf-8") as f:
        for row in csv.DictReader(f):
            dom = row.get("dns.qry.name", "")
            q[dom].append(row)
    return q


def score(dom, count):
    labels = dom.split(".")
    max_label = max((len(l) for l in labels), default=0)
    e = entropy(dom)
    flags = []
    if max_label > 20:
        flags.append(f"超长标签({max_label})")
    if e > 3.5:
        flags.append(f"高熵({e:.2f})")
    if count > 200:
        flags.append(f"高频({count})")
    return flags


def main():
    import sys
    path = sys.argv[1] if len(sys.argv) > 1 else "dns.csv"
    q = load_dns(path)
    print("[+] 检测 DNS 隧道疑似域名：")
    for dom, rows in q.items():
        flags = score(dom, len(rows))
        if flags:
            print(f"  {dom}  ->  {', '.join(flags)}")


if __name__ == "__main__":
    main()

title: 暴力破解成功后异常登录
status: experimental
description: 检测同一源 IP 在短时窗内大量认证失败后成功登录
author: BlueTeam-IR
logsource:
  category: authentication
detection:
  selection_fail:
    outcome: failure
  selection_success:
    outcome: success
  group_by_src:
    src_ip: "*"
  condition: selection_fail and selection_success by src_ip
  threshold:
    fail_count: 10
    window: 10m
falsepositives:
  - 用户忘记密码后重试
level: high
tags:
  - attack.t1110
  - attack.t1078

【归因草案】
置信度总评：中（基于行为归因，身份归因待情报/执法确认）

支持假设 H1（疑似与已知勒索团伙 X 相关）的证据：
  - 行为：使用 Cobalt Strike + 固定 60s 心跳 + SMB 横向（与 X 公开 TTP 一致）；
  - 基础设施：C2 证书与 X 历史域名同源（PDNS 反查）；
  - 时间戳：活动集中于 UTC+8 工作时间（弱证据）。
削弱 H1 的证据：
  - 使用公开商业工具，任何团伙均可获取；
  - GeoIP 指向云主机，可能为租用，不构成国籍证据。
无法排除的替代假设：
  - H2：机会主义犯罪团伙复用公开工具与泄露基础设施；
  - H3：内部人员借外部 IP 实施。
下一步建议：
  - 将样本/IOC 提交合规情报平台做团伙匹配；
  - 申请 CERT 协调，分享证据供进一步归因。

  威胁猎杀（主动找）  ──发现线索──▶  事件响应（被动处置）
        ▲                                  │
        │                                  ▼
   沉淀检测规则 ◀──复盘改进── 复盘/加固