grep -c "Accepted publickey" /var/log/auth.log
ls -la /home/*/

# 安装
apt-get install -y auditd audispd-plugins
systemctl enable --now auditd

# /etc/audit/rules.d/vps_ssh_audit.rules
# 捕获 SSH 登录/登出（PAM auth 成功事件）
-w /var/log/auth.log -p wa -k ssh_auth

# 捕获所有 Shell 命令执行（execve）
-a always,exit -F arch=b64 -S execve -F a0!=/usr/bin/env -F a0!=/usr/bin/script -F exit=0 -k vps_cmd

# 捕获文件写入（仅关注关键目录）
-w /root/.ssh -w /etc -w /opt -w /srv -w /var/www -p wa -k vps_file

# 捕获出站连接（排除本地回环）
-a always,exit -F arch=b64 -S connect -F exit=0 -F success=0 -k vps_net

# 加载规则
auditctl -R /etc/audit/rules.d/vps_ssh_audit.rules

# /opt/ssh_audit/pam_login.sh
#!/bin/bash
echo "$(date -Iseconds) LOGIN user=$PAM_USER ip=$PAM_RHOST uid=$PAM_UID" >> /var/log/ssh_session.log

#!/usr/bin/env python3
"""
collector.py — SSH 会话采集守护进程
监听 auditd 事件流，解析后写入 SQLite
"""

import os
import re
import sqlite3
import subprocess
import json
import time
import logging
from datetime import datetime, timedelta
from pathlib import Path

DB_DIR = Path("/opt/ssh_audit/data")
DB_DIR.mkdir(parents=True, exist_ok=True)

# 按天分区，文件名如 ssh_audit_20261011.db
def get_db_path() -> str:
    today = datetime.now().strftime("%Y%m%d")
    return str(DB_DIR / f"ssh_audit_{today}.db")


def init_schema(db_path: str):
    conn = sqlite3.connect(db_path)
    conn.executescript("""
    CREATE TABLE IF NOT EXISTS sessions (
        session_id    TEXT PRIMARY KEY,
        user          TEXT,
        src_ip        TEXT,
        login_time    TEXT,
        logout_time   TEXT,
        duration_sec  INTEGER DEFAULT 0,
        cmd_count     INTEGER DEFAULT 0
    );
    CREATE TABLE IF NOT EXISTS commands (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id    TEXT REFERENCES sessions(session_id),
        ts            TEXT,
        user          TEXT,
        cmd           TEXT,
        exit_code     INTEGER
    );
    CREATE TABLE IF NOT EXISTS file_writes (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id    TEXT,
        ts            TEXT,
        user          TEXT,
        filepath      TEXT,
        operation     TEXT   -- 'create' / 'write' / 'delete' / 'rename'
    );
    CREATE TABLE IF NOT EXISTS net_outbound (
        id            INTEGER PRIMARY KEY AUTOINCREMENT,
        session_id    TEXT,
        ts            TEXT,
        user          TEXT,
        dst_ip        TEXT,
        dst_port      INTEGER,
        proto         TEXT   -- 'tcp' / 'udp'
    );
    CREATE INDEX IF NOT EXISTS idx_cmd_session ON commands(session_id);
    CREATE INDEX IF NOT EXISTS idx_fw_session  ON file_writes(session_id);
    CREATE INDEX IF NOT EXISTS idx_net_session ON net_outbound(session_id);
    """)
    conn.commit()
    conn.close()


# ── 解析 auditd 原始日志 ──────────────────────────────────────────────

def parse_audit_line(line: str):
    """解析单行 auditd 日志，返回 (event_type, payload) 或 None"""
    # 格式: type=SYSCALL msg=audit(1728...): ... comm="bash" exe="/bin/bash" ...
    m = re.match(
        r"^\s*type=(\w+)\s+msg=audit\((\d+)\.\d+\):.*?a\d+=\(?(\d+)\"?\s*(.*)$",
        line,
        re.DOTALL,
    )
    if not m:
        return None
    etype = m.group(1)
    return etype, m.group(0)


def extract_execve(payload: str) -> dict | None:
    """从 execve 系统调用记录提取命令"""
    m = re.search(r'a0=(\d+).*?a1=(\d+).*?cmdline="([^"]*)"', payload)
    return {"cmd": m.group(3)} if m else None


def extract_connect(payload: str) -> dict | None:
    """从 connect 系统调用提取出站目标"""
    m = re.search(r'addr=(\d+):\d+', payload)
    if not m:
        return None
    dst = m.group(1)
    # auditd 中 IP 为小端十六进制，转换为点分十进制
    ip = ".".join(
        str(int(dst[i : i + 2], 16)) for i in range(6, -2, -2)
    )
    return {"dst_ip": ip, "dst_port": 0}  # port 需结合 sa_family 进一步解析


# ── 主采集循环 ────────────────────────────────────────────────────────

SESSIONS = {}  # user → session_id

def main():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
    )
    db = get_db_path()
    init_schema(db)
    conn = sqlite3.connect(db)

    logging.info(f"Collector started, DB={db}")

    # 启动 auditctl 监听（后台）
    # 实际部署中建议用 ausearch -f 或 auread 替代 tail
    tail_proc = subprocess.Popen(
        ["tail", "-F", "/var/log/audit/audit.log"],
        stdout=subprocess.PIPE,
        stderr=subprocess.DEVNULL,
        text=True,
    )

    for line in tail_proc.stdout:
        line = line.strip()
        if not line:
            continue
        etype, _ = parse_audit_line(line)
        if etype is None:
            continue

        # 提取 session 上下文（user + src_ip 来自 pam 日志）
        # 简化：从 line 中的 ppid 关联，实际生产建议用 audit 的 auid 字段
        # 此处为演示目的，直接从 pam 日志 tail 维护 SESSIONS 字典

        if "pam_login.sh" in line:
            m = re.search(r"user=(\S+)\s+ip=(\S+)", line)
            if m:
                user, ip = m.group(1), m.group(2)
                sid = f"{user}_{int(time.time())}"
                SESSIONS[user] = {"session_id": sid, "src_ip": ip}
                conn.execute(
                    "INSERT OR REPLACE INTO sessions VALUES (?,?,?,?,?,0,0)",
                    (sid, user, ip, datetime.now().isoformat(), None),
                )
                conn.commit()

        elif "execve" in line:
            ev = extract_execve(line)
            if ev:
                conn.execute(
                    "INSERT INTO commands VALUES (?,?,?,?,?,?)",
                    (None, SESSIONS.get("_", {}).get("session_id", "unknown"),
                     datetime.now().isoformat(), "_", ev["cmd"], 0),
                )
                conn.commit()

        # 网络出站 / 文件写入 同理，此处省略

    tail_proc.terminate()
    conn.close()


if __name__ == "__main__":
    main()

# baseline.py — 每 15 分钟滑动窗口聚合

import sqlite3
import statistics
from datetime import datetime, timedelta
from pathlib import Path

DB_DIR = Path("/opt/ssh_audit/data")


def compute_user_baseline(user: str, window_days: int = 14) -> dict:
    """
    计算某用户最近 window_days 天的行为基线
    返回 4 个维度的统计量
    """
    db = get_db_path_for_range(window_days)
    conn = sqlite3.connect(db)

    # ① 命令熵：去重命令数 / 总命令数（高熵 = 操作多样，低熵 = 重复性操作）
    row = conn.execute(
        "SELECT COUNT(*), COUNT(DISTINCT cmd) FROM commands WHERE user=?",
        (user,),
    ).fetchone()
    total, distinct = row
    cmd_entropy = distinct / total if total > 0 else 0.0

    # ② 文件写频率：每日平均写入文件数
    row = conn.execute(
        "SELECT COUNT(*) FROM file_writes WHERE user=?",
        (user,),
    ).fetchone()
    file_write_rate = row[0] / window_days

    # ③ 网络出站目标集合（去重 IP）
    row = conn.execute(
        "SELECT DISTINCT dst_ip FROM net_outbound WHERE user=?",
        (user,),
    ).fetchall()
    net_targets = [r[0] for r in row]

    # ④ 工作时段：登录时间戳的 2 小时分箱直方图
    rows = conn.execute(
        "SELECT login_time FROM sessions WHERE user=?",
        (user,),
    ).fetchall()
    hour_bins = [0] * 24
    for r in rows:
        try:
            h = int(r[0][11:13])
            hour_bins[h] += 1
        except (IndexError, ValueError):
            pass

    conn.close()
    return {
        "user": user,
        "cmd_entropy": round(cmd_entropy, 4),
        "file_write_rate": round(file_write_rate, 2),
        "net_target_count": len(net_targets),
        "net_targets": net_targets[:20],  # 限制长度，控制 LLM 输入
        "hour_distribution": hour_bins,
        "window_days": window_days,
    }

# 基线存入独立 SQLite，避免与流水表混在一起
BASELINE_DB = Path("/opt/ssh_audit/baseline.db")

def save_baseline(baseline: dict):
    conn = sqlite3.connect(BASELINE_DB)
    conn.execute(
        """CREATE TABLE IF NOT EXISTS user_baselines (
            user TEXT, computed_at TEXT,
            cmd_entropy REAL, file_write_rate REAL,
            net_target_count INTEGER, net_targets TEXT,
            hour_distribution TEXT, window_days INTEGER,
            PRIMARY KEY (user, computed_at)
        )"""
    )
    conn.execute(
        "INSERT INTO user_baselines VALUES (?,?,?,?,?,?,?,?)",
        (
            baseline["user"],
            datetime.now().isoformat(),
            baseline["cmd_entropy"],
            baseline["file_write_rate"],
            baseline["net_target_count"],
            json.dumps(baseline["net_targets"]),
            json.dumps(baseline["hour_distribution"]),
            baseline["window_days"],
        ),
    )
    conn.commit()
    conn.close()

# risk_rules.py — 静态规则引擎，无 LLM 依赖

import re
import ipaddress
from dataclasses import dataclass, field


@dataclass
class RiskEvent:
    user: str
    cmd: str
    ts: str
    risk_level: str   # 'HIGH' / 'MEDIUM' / 'LOW'
    reason: str


def evaluate_risk(user: str, cmd: str, ts: str) -> list[RiskEvent]:
    """
    对单条命令执行所有规则，返回命中的风险事件列表
    （一条命令可能命中多条规则）
    """
    events: list[RiskEvent] = []
    c = cmd.lower()

    # ── 高危：凭证 / 密钥操作 ──────────────────────────────────────
    CRED_PATTERNS = [
        (r"(?<=\s)cat\s+/etc/shadow", "读取 /etc/shadow（密码哈希文件）"),
        (r"cp\s+\S*\.ssh/", "复制 SSH 密钥目录"),
        (r"rsync.*\.env\b", "同步 .env 文件（含 API 密钥）"),
        (r"git\s+push.*--force", "强制推送（可能覆盖他人提交）"),
        (r"(rm|sudo\s+rm)\s+-rf\s+/", "递归删除根目录"),
        (r"chmod\s+777\s+/\S+/", "危险的全开放权限"),
    ]
    for pattern, reason in CRED_PATTERNS:
        if re.search(pattern, c):
            events.append(RiskEvent(user, cmd, ts, "HIGH", reason))

    # ── 中危：数据外泄类 ───────────────────────────────────────────
    EXFIL_PATTERNS = [
        (r"(scp|rsync|curl|wget|ftp)\s+\S+\s+@\S+", "向外部目标传输文件"),
        (r"tar\s+-czf\s+/tmp/\S+\s+/var", "打包敏感目录到 /tmp（可被外传）"),
        (r"nc\s+\S+\s+\d+", "netcat 出站连接（非常规通道）"),
    ]
    for pattern, reason in EXFIL_PATTERNS:
        if re.search(pattern, c):
            events.append(RiskEvent(user, cmd, ts, "MEDIUM", reason))

    # ── 低危：可疑但常见 ──────────────────────────────────────────
    if re.search(r"history\s*-c|shred\s+", c):
        events.append(RiskEvent(user, cmd, ts, "LOW", "清除 Shell 历史 / 安全擦除"))

    return events

def on_risk_event(ev: RiskEvent, baseline: dict):
    msg = (
        f"🚨 SSH 审计告警 [{ev.risk_level}]\n"
        f"用户: {ev.user}\n"
        f"时间: {ev.ts}\n"
        f"命令: {ev.cmd[:120]}\n"
        f"原因: {ev.reason}\n"
        f"基线偏离: {format_baseline_deviation(ev.user, baseline)}"
    )
    send_telegram(msg)  # 实时推送

# 最小 2GB 内存 VPS 推荐 Qwen2.5-7B（INT4 量化）
ollama pull qwen2.5:7b

# 验证
curl -s http://localhost:11434/api/tags | python3 -c "
import sys,json; d=json.load(sys.stdin)
print([m['name'] for m in d['models']])
"

# llm_analyzer.py

import json
import requests
import logging

OLLAMA_URL = "http://localhost:11434/api/generate"
LOG = logging.getLogger(__name__)


def analyze_session_batch(
    session_id: str,
    commands: list[dict],
    baseline: dict,
    top_risk_events: list[dict],
) -> dict:
    """
    将一批未分类的 SSH 命令 + 基线 + 硬规则命中事件
    提交给 Ollama，获取语义风险评分与自然语言摘要
    """
    # 限制命令数量，控制 token
    cmds_text = "\n".join(
        f"[{c['ts']}] {c['cmd'][:80]}" for c in commands[:50]
    )

    prompt = f"""你是一名 Linux 安全审计专家。
以下是一位 VPS 用户（{baseline['user']}）最近一小时的 SSH 会话命令记录。

【用户行为基线】
- 命令熵: {baseline['cmd_entropy']}（0=高度重复, 1=完全随机）
- 日均文件写入: {baseline['file_write_rate']} 次
- 已知网络出站目标: {baseline['net_target_count']} 个
- 工作时段分布: 活跃时段为 {active_hours(baseline)}

【硬规则命中】
{json.dumps(top_risk_events, ensure_ascii=False, indent=2) if top_risk_events else '（无）'}

【命令记录】
{cmds_text}

请输出 JSON（不要其他内容）：
{{
  "risk_score": <0-100 整数，0=完全安全, 100=极高风险>,
  "risk_level": "<low|medium|high|critical>",
  "reasoning": "<3 句以内，中文，说明为什么给这个分数>",
  "anomalies": [
    {{"cmd": "<具体命令>", "issue": "<一句话说明异常原因>"}}
  ],
  "suggestion": "<1-2 句，给运维人员的下一步建议>"
}}"""

    try:
        resp = requests.post(
            OLLAMA_URL,
            json={
                "model": "qwen2.5:7b",
                "prompt": prompt,
                "stream": False,
                "options": {"temperature": 0.1, "num_predict": 1024},
            },
            timeout=120,
        )
        raw = resp.json().get("response", "")
        # 提取 JSON 块
        json_match = re.search(r"\{[\s\S]*\}", raw)
        return json.loads(json_match.group(0)) if json_match else {}
    except Exception as e:
        LOG.error(f"LLM analysis failed: {e}")
        return {"risk_score": 50, "risk_level": "unknown", "reasoning": str(e)}


def active_hours(baseline: dict) -> str:
    bins = baseline.get("hour_distribution", [0] * 24)
    active = [h for h, v in enumerate(bins) if v > 0]
    if not active:
        return "未知"
    # 找连续活跃区间
    ranges = []
    start = active[0]
    for h in active[1:]:
        if h != start + 1:
            ranges.append((start, h - 1))
            start = h
    ranges.append((start, active[-1]))
    return ", ".join(f"{a:02d}:00-{b:02d}:59" for a, b in ranges)

# scheduler.py（伪代码）

def scheduler_loop():
    while True:
        # ① 高危 session 优先
        critical = get_unanalyzed_critical_sessions()
        for s in critical:
            result = analyze_session_batch(
                s["session_id"], s["commands"], s["baseline"], s["risk_events"]
            )
            save_llm_result(s["session_id"], result)
            if result.get("risk_level") in ("high", "critical"):
                send_telegram(format_llm_alert(s, result))

        # ② 批量常规分析（每 5 分钟最多 10 个 session）
        batch = get_unanalyzed_sessions(limit=10)
        for s in batch:
            result = analyze_session_batch(
                s["session_id"], s["commands"], s["baseline"], []
            )
            save_llm_result(s["session_id"], result)

        time.sleep(300)

def generate_daily_report(date_str: str) -> str:
    conn = get_db_connection_for(date_str)

    total_sessions = count_sessions(conn)
    total_cmds     = count_commands(conn)
    high_risk      = count_risk_events(conn, "HIGH")
    llm_avg_score  = avg_llm_risk_score(conn)
    top_users      = top_active_users(conn, limit=5)

    return f"""📋 VPS SSH 审计日报 — {date_str}

📊 总览
  会话数: {total_sessions}  |  命令总数: {total_cmds}
  高危事件: {high_risk}  |  LLM 平均风险分: {llm_avg_score}/100

👤 活跃用户 TOP 5
{format_user_list(top_users)}

⚠️ 高危事件摘要
{format_high_risk_events(get_high_risk_events(conn, limit=10))}

🤖 LLM 发现（Top 3 异常）
{format_llm_findings(get_top_llm_findings(conn, limit=3))}

———
本日报由本地 LLM 自动生成，数据未离开服务器。
"""

# telegram_alert.py

import requests
import os
import logging

LOG = logging.getLogger(__name__)
BOT_TOKEN  = os.environ["TELEGRAM_BOT_TOKEN"]
CHAT_ID    = os.environ["TELEGRAM_CHAT_ID"]
API_URL    = f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage"


def send_telegram(text: str, parse_mode: str = "HTML"):
    resp = requests.post(
        API_URL,
        json={
            "chat_id": CHAT_ID,
            "text": text,
            "parse_mode": parse_mode,
            "disable_web_page_preview": True,
        },
        timeout=15,
    )
    if resp.status_code != 200:
        LOG.error(f"Telegram send failed: {resp.status_code} {resp.text}")
    else:
        LOG.info("Telegram alert sent")

#!/bin/bash
set -e

echo "=== 1. 安装依赖 ==="
apt-get update -qq
apt-get install -y -qq python3-pip sqlite3 auditd

pip3 install requests

echo "=== 2. 拉取 Ollama 模型 ==="
if ! command -v ollama &>/dev/null; then
    curl -fsSL https://ollama.com/install.sh | sh
    systemctl enable --now ollama
fi
ollama pull qwen2.5:7b
echo "   模型就绪"

echo "=== 3. 创建目录与配置 ==="
mkdir -p /opt/ssh_audit/{data,scripts}

# 复制脚本（此处省略，实际部署时用 scp 或 git clone）
# cp collector.py baseline.py risk_rules.py llm_analyzer.py \
#    telegram_alert.py scheduler.py /opt/ssh_audit/scripts/

echo "=== 4. 配置 auditd 规则 ==="
cat > /etc/audit/rules.d/vps_ssh_audit.rules << 'EOF'
-w /var/log/auth.log -p wa -k ssh_auth
-a always,exit -F arch=b64 -S execve -F exit=0 -k vps_cmd
-w /root/.ssh -w /etc -w /opt -w /srv -w /var/www -p wa -k vps_file
-a always,exit -F arch=b64 -S connect -F exit=0 -k vps_net
EOF

# PAM hook
cat > /opt/ssh_audit/scripts/pam_login.sh << 'EOF'
#!/bin/bash
echo "$(date -Iseconds) LOGIN user=$PAM_USER ip=$PAM_RHOST" >> /var/log/ssh_session.log
EOF
chmod +x /opt/ssh_audit/scripts/pam_login.sh

echo "=== 5. 启用 systemd 服务 ==="
cat > /etc/systemd/system/ssh-audit.service << 'EOF'
[Unit]
Description=SSH Session Audit Daemon
After=network.target auditd.service
Requires=auditd.service

[Service]
ExecStart=/usr/bin/python3 /opt/ssh_audit/scripts/collector.py
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target
EOF

cat > /etc/systemd/system/ssh-audit-scheduler.service << 'EOF'
[Unit]
Description=SSH Audit LLM Scheduler
After=ssh-audit.service ollama.service
Requires=ollama.service

[Service]
ExecStart=/usr/bin/python3 /opt/ssh_audit/scripts/scheduler.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

systemctl daemon-reload
systemctl enable --now ssh-audit.service
systemctl enable --now ssh-audit-scheduler.service

echo "=== 6. 配置 Telegram 环境变量 ==="
cat > /etc/ssh_audit_env << EOF
TELEGRAM_BOT_TOKEN=你的_BOT_TOKEN
TELEGRAM_CHAT_ID=你的_CHAT_ID
EOF
chmod 600 /etc/ssh_audit_env

echo "=== 部署完成 ==="
systemctl status ssh-audit.service --no-pager
echo "验证: journalctl -u ssh-audit -f"