# /etc/audit/rules.d/sftp-fileop.rules
## SFTP 导出目录文件操作审计
# 监听 openat（读/写打开）、unlink（删除）、rename（重命名）
-w /etc/sftp-exports/ -p wa -k sftp_fileop -f
# 记录所有 sftp 用户的 open/read 操作（auditd 无法区分 open 的 mode，
# 需在 post-process 阶段用 read 事件 + 文件大小推断是否实际外传）
-a always,exit -F arch=b64 -S openat -S read -F key=sftp_read
-a always,exit -F arch=b64 -S unlink -S unlinkat -S rename -S renameat -F key=sftp_mutation

sudo augenrules --load
sudo auditctl -l    # 确认规则已加载
sudo tail -f /var/log/audit/audit.log | grep sftp_fileop   # 验证事件流

# /etc/logrotate.d/audit
/var/log/audit/audit.log {
    daily
    rotate 30
    compress
    missingok
    notifempty
    postrotate
        /usr/bin/systemctl kill -s HUP auditd 2>/dev/null || true
    endpostrotate
}

#!/usr/bin/env python3
"""
sftp_audit_analyzer.py
增量解析 audit.log + auth.log，调用本地 LLM 生成 SFTP 操作审计摘要。
systemd timer 每小时触发。
"""
import json
import os
import re
import subprocess
from datetime import datetime, timedelta
from pathlib import Path

STATE_FILE = Path("/var/lib/sftp-audit/last_state.json")
AUDIT_LOG  = Path("/var/log/audit/audit.log")
AUTH_LOG   = Path("/var/log/auth.log")
REPORT_DIR = Path("/srv/audit-reports")  # Git 仓库根目录
OLLAMA_ENDPOINT = "http://127.0.0.1:11434"

def load_state():
    if STATE_FILE.exists():
        return json.loads(STATE_FILE.read_text())
    return {"audit_inode": 0, "audit_offset": 0, "auth_mtime": 0}

def save_state(state):
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(json.dumps(state, indent=2))

def read_increment(path, state_key_inode, state_key_offset):
    """按 inode+offset 增量读取日志，处理文件轮转（inode 变化）。"""
    st = path.stat()
    if st.st_ino != state[state_key_inode]:
        # 新文件（轮转后），从头读取
        state[state_key_inode] = st.st_ino
        state[state_key_offset] = 0
    if st.st_size < state[state_key_offset]:
        # 日志被截断，重置
        state[state_key_offset] = 0
    with open(path, 'r', errors='replace') as f:
        f.seek(state[state_key_offset])
        data = f.read()
        state[state_key_offset] = f.tell()
    return data

def parse_audit_events(raw: str):
    """
    解析 auditd 格式记录，提取 key=sftp_* 的事件。
    返回: [{ts, uid, pid, comm, syscall, name, success}, ...]
    """
    events = []
    # auditd 记录格式:
    # type=SYSCALL msg=audit(1698307200.123:456): arch=c000003e syscall=2 ...
    #                        item=1 ... exe="/lib/openssh/sftp-server" ...
    #                        key=sftp_fileop
    for line in raw.splitlines():
        if "key=sftp_" not in line:
            continue
        m = re.match(
            r"^\s*type=SYSCALL\s+msg=audit\((\d+\.\d+):(\d+)\):\s+(.*)$",
            line
        )
        if not m:
            continue
        ts = m.group(1)
        detail = m.group(3)
        uid_m    = re.search(r"uid=(\d+)", detail)
        pid_m    = re.search(r"pid=(\d+)", detail)
        comm_m   = re.search(r"comm=\"([^\"]+)\"", detail)
        syscall_m = re.search(r"syscall=(\d+)", detail)
        name_m   = re.search(r"name=\"([^\"]+)\"", detail)
        success_m = re.search(r"success=(\w+)", detail)
        key_m    = re.search(r"key=(\S+)", detail)

        events.append({
            "ts": ts,
            "uid": int(uid_m.group(1)) if uid_m else None,
            "pid": int(pid_m.group(1)) if pid_m else None,
            "comm": comm_m.group(1) if comm_m else "unknown",
            "syscall": int(syscall_m.group(1)) if syscall_m else None,
            "name": name_m.group(1) if name_m else "",
            "success": success_m.group(1).lower() if success_m else "unknown",
            "key": key_m.group(1) if key_m else "sftp_fileop",
        })
    return events

def parse_auth_sessions(raw: str):
    """
    解析 sshd 会话日志，建立 uid → username 映射。
    匹配: Accepted publickey for deploy from 10.0.1.5 port 44123
          sshd[1234]: pam_unix(sshd:session): session opened for user deploy(uid=1001)
    """
    mapping = {}
    for line in raw.splitlines():
        m = re.search(
            r"session opened for user (\S+)\(uid=(\d+)\)", line
        )
        if m:
            mapping[int(m.group(2))] = m.group(1)
    return mapping

def fetch_ollama(prompt: str, model: str = "qwen2.5:7b") -> str:
    """调用本地 Ollama，返回原始文本响应。"""
    payload = json.dumps({
        "model": model,
        "prompt": prompt,
        "stream": False,
        "options": {"temperature": 0.1, "num_predict": 2048}
    })
    result = subprocess.run(
        ["curl", "-s", f"{OLLAMA_ENDPOINT}/api/generate",
         "-d", payload],
        capture_output=True, text=True, timeout=120
    )
    resp = json.loads(result.stdout)
    return resp.get("response", "")

SYSTEM_PROMPT = """\
你是一名 SFTP 文件传输审计专家，负责解析 OpenSSH sftp-server 的
auditd 系统调用日志和 sshd 会话日志，生成结构化合规审计报告。

输出 JSON 格式（不要 markdown 代码块）：
{
  "summary": "一段话概括本批次 SFTP 操作的主要特征",
  "anomalies": [
    {
      "type": "bulk_exfil | unusual_path | mass_delete | permission_change | unknown",
      "severity": "high | medium | low",
      "uid": 1001,
      "username": "deploy",
      "files": ["a.txt", "b.txt"],
      "description": "描述异常行为及判断依据",
      "evidence_refs": ["ts=1698307200.123:456", "ts=1698307210.456:457"]
    }
  ],
  "file_operation_stats": {
    "total_ops": 152,
    "opens": 80,
    "reads": 120,
    "writes": 30,
    "deletes": 3,
    "renames": 5,
    "unique_files": 45,
    "unique_users": 3
  },
  "compliance_timeline": [
    {"ts": "1698307200.123", "user": "deploy", "action": "open", "file": "reports/2024-10.pdf", "result": "success"}
  ],
  "recommendations": ["建议 1", "建议 2"]
}
"""

def build_user_prompt(events, uid_map):
    """将原始事件 + uid 映射拼成 user prompt。"""
    # 按 uid 聚合，减少 token 量
    by_user = {}
    for e in events:
        username = uid_map.get(e["uid"], f"uid:{e['uid']}")
        by_user.setdefault(username, []).append(e)

    lines = []
    lines.append(f"批次时间窗口：{len(events)} 条审计事件，涉及 {len(by_user)} 个用户。")
    lines.append("")
    for user, user_events in by_user.items():
        lines.append(f"=== 用户: {user} ({len(user_events)} 条操作) ===")
        for e in user_events[:50]:  # 每用户最多取 50 条，防止 prompt 爆炸
            lines.append(
                f"  ts={e['ts']} pid={e['pid']} syscall={e['syscall']} "
                f"name={e['name']} success={e['success']} key={e['key']}"
            )
        if len(user_events) > 50:
            lines.append(f"  ... 还有 {len(user_events)-50} 条操作被省略")
    return "\n".join(lines)

def main():
    state = load_state()

    # 1. 增量读取
    audit_raw = read_increment(AUDIT_LOG, "audit_inode", "audit_offset")
    auth_raw  = read_increment(AUTH_LOG,  "auth_inode",  "auth_offset")

    if not audit_raw.strip() and not auth_raw.strip():
        print("No new events. Skipping.")
        save_state(state)
        return

    # 2. 解析
    events = parse_audit_events(audit_raw)
    uid_map = parse_auth_sessions(auth_raw)

    if not events:
        print("No sftp-related audit events found.")
        save_state(state)
        return

    # 3. LLM 分析
    user_prompt = build_user_prompt(events, uid_map)
    raw_response = fetch_ollama(SYSTEM_PROMPT + "\n" + user_prompt)

    # 4. 解析 LLM 输出（容错：LLM 偶尔会包 markdown 代码块）
    clean = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw_response.strip())
    report = json.loads(clean)

    # 5. 生成报告文件
    ts_label = datetime.now().strftime("%Y%m%d_%H%M%S")
    report_file = REPORT_DIR / "reports" / f"sftp-audit-{ts_label}.json"
    report_file.parent.mkdir(parents=True, exist_ok=True)
    report_file.write_text(json.dumps(report, indent=2, ensure_ascii=False))

    # 6. 合规时间线（CSV，便于 Excel/Sheets 直接导入）
    csv_file = REPORT_DIR / "reports" / f"sftp-timeline-{ts_label}.csv"
    with open(csv_file, "w") as f:
        f.write("timestamp,username,action,file,result\n")
        for item in report.get("compliance_timeline", []):
            f.write(
                f"{item['ts']},{item['user']},{item['action']},"
                f"{item['file']},{item['result']}\n"
            )

    # 7. Git 归档
    git_cmds = [
        ["git", "add", "reports/"],
        ["git", "commit", "-m",
         f"audit: SFTP report {ts_label} "
         f"({len(events)} events, {len(report.get('anomalies', []))} anomalies)"],
        ["git", "push", "origin", "main"],
    ]
    for cmd in git_cmds:
        r = subprocess.run(cmd, cwd=REPORT_DIR, capture_output=True, text=True)
        if r.returncode != 0 and cmd[1] != "add":
            print(f"git {cmd[1]} failed: {r.stderr}")
            break

    # 8. 保存状态
    save_state(state)
    print(f"Done: {len(events)} events → {report_file.name} + {csv_file.name}")
    print(f"Anomalies: {len(report.get('anomalies', []))}")

# /etc/systemd/system/sftp-audit-analyzer.service
[Unit]
Description=SFTP LLM Audit Analyzer (hourly)
After=network.target ollama.service
Wants=ollama.service

[Service]
Type=oneshot
User=root
ExecStart=/usr/bin/python3 /opt/sftp-audit/sftp_audit_analyzer.py
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

# /etc/systemd/system/sftp-audit-analyzer.timer
[Unit]
Description=Run SFTP audit analyzer hourly

[Timer]
OnCalendar=*:00:00
Persistent=true
Unit=sftp-audit-analyzer.service

[Install]
WantedBy=timers.target

sudo systemctl daemon-reload
sudo systemctl enable --now sftp-audit-analyzer.timer
systemctl list-timers | grep sftp

def rule_based_guardrails(events, uid_map):
    """
    不依赖 LLM 的硬规则异常检测，作为 LLM 输出的交叉验证。
    返回额外发现的异常列表（LLM 可能漏报的情况）。
    """
    alerts = []
    by_uid = {}
    for e in events:
        by_uid.setdefault(e["uid"], []).append(e)

    for uid, user_events in by_uid.items():
        username = uid_map.get(uid, f"uid:{uid}")
        # 规则 1：单用户 10 分钟内删除 > 20 个文件 → 批量删除嫌疑
        deletes = [e for e in user_events
                   if e["key"] == "sftp_mutation" and "unlink" in e.get("name", "")]
        if len(deletes) > 20:
            alerts.append({
                "type": "mass_delete",
                "severity": "high",
                "uid": uid, "username": username,
                "files": [d["name"] for d in deletes[:10]],
                "description": f"用户 {username} 短时间内删除 {len(deletes)} 个文件",
                "source": "rule_guardrail"
            })
        # 规则 2：只读目录出现 write 操作 → 权限绕过嫌疑
        # 规则 3：非工作时间（02:00-05:00）大量 open → 异常时段活动
    return alerts

all_anomalies = (
    report.get("anomalies", []) +
    rule_based_guardrails(events, uid_map)
)
# 去重（按 uid + type + 时间窗口）
deduped = deduplicate(all_anomalies)
report["anomalies"] = deduped

# 每月归档签名（crontab）
0 3 1 * * cd /srv/audit-reports && \
  git tag -s "audit-$(date +%Y%m)" \
  -m "Monthly audit archive" \
  && git push origin --tags