mcpField
P
performing-linux-log-forensics-investigation

performing-linux-log-forensics-investigation

github.com/mukul975安全35.2k70 次安装8 周前更新
代码仓库

关于 performing-linux-log-forensics-investigation

面向调查人员和事件响应人员的 Linux 日志取证技能,涵盖认证、系统、审计和 journal 日志,并提供命令行分析方法及 Python 解析器示例。

主要功能

  1. 01分析认证与 SSH 日志
  2. 02导出并筛选 systemd journal
  3. 03搜索 Linux Audit 事件
  4. 04检查 Cron 与软件包活动
  5. 05提供 Python 取证解析器示例

适用场景

  1. 01重建 Linux 用户会话
  2. 02调查疑似未授权访问
  3. 03根据主机日志建立事件时间线

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/mukul975/anthropic-cybersecurity-skills --skill performing-linux-log-forensics-investigation

安装后在 Claude Code 中输入/performing-linux-log-forensics-investigation确认已加载。

包含的文件

除 SKILL.md 外,安装时一起下载的参考资料和脚本。

LICENSE
assets/
template.md
references/
api-reference.md
standards.md
workflows.md
scripts/
agent.py
process.py

详细介绍

执行 Linux 日志取证调查

概述

Linux 系统会维护大量日志,这些日志是取证调查中的主要证据来源。与 Windows 事件日志不同,Linux 日志通常是存储在 /var/log/ 中的纯文本文件,以及由 systemd-journald 管理的二进制日志文件。关键取证日志包括 auth.log(身份验证事件、sudo 使用情况、SSH 会话)、syslog(系统范围的消息)、kern.log(内核事件)以及特定应用程序的日志。Linux Audit 框架(auditd)可提供详细的安全事件日志记录,其功能可与 Windows 安全事件日志相媲美。对这些日志进行取证分析,可帮助调查人员重建用户会话、识别未经授权的访问、检测权限提升、追踪横向移动,并建立完整的事件时间线。

适用场景

  • 开展涉及 Linux 日志取证调查的安全评估时
  • 遵循相关安全事件的事件响应流程时
  • 执行定期安全测试或审计活动时
  • 通过实践测试验证安全控制措施时

先决条件

  • 熟悉数字取证概念和工具
  • 可访问用于安全执行操作的测试或实验环境
  • 已安装所需依赖项的 Python 3.8+
  • 已获得任何测试活动所需的适当授权

关键日志文件及其位置

日志文件路径内容
auth.log / secure/var/log/auth.log(Debian)或 /var/log/secure(RHEL)身份验证、sudo、SSH、PAM
syslog / messages/var/log/syslog(Debian)或 /var/log/messages(RHEL)常规系统消息
kern.log/var/log/kern.log内核消息、USB 事件、驱动程序加载
lastlog/var/log/lastlog每位用户的上次登录记录(二进制)
wtmp/var/log/wtmp登录/注销记录(二进制,使用 last 读取)
btmp/var/log/btmp登录失败尝试(二进制,使用 lastb 读取)
faillog/var/log/faillog登录失败计数器(二进制)
cron.log/var/log/cron 或 /var/log/syslog计划任务执行记录
audit.log/var/log/audit/audit.logLinux Audit 框架事件
journal/var/log/journal/ 或 /run/log/journal/systemd 二进制日志
dpkg.log/var/log/dpkg.log软件包安装/移除记录(Debian)
yum.log/var/log/yum.log软件包安装/移除记录(RHEL)

分析技术

身份验证日志分析
# Find all successful SSH logins
grep "Accepted" /var/log/auth.log

# Find failed SSH login attempts
grep "Failed password" /var/log/auth.log

# Extract unique source IPs from failed logins
grep "Failed password" /var/log/auth.log | grep -oP '\d+\.\d+\.\d+\.\d+' | sort -u

# Find sudo command execution
grep "sudo:" /var/log/auth.log | grep "COMMAND"

# Detect brute force patterns (>10 failures from same IP)
grep "Failed password" /var/log/auth.log | awk '{print $(NF-3)}' | sort | uniq -c | sort -rn | head -20

# Find account creation events
grep "useradd\|adduser" /var/log/auth.log

# Detect SSH key authentication
grep "Accepted publickey" /var/log/auth.log
Systemd 日志分析
# Export journal in JSON format for forensic processing
journalctl --output=json --no-pager > journal_export.json

# Filter by time range
journalctl --since "2025-02-01" --until "2025-02-15" --output=json > timerange.json

# Filter by unit/service
journalctl -u sshd --output=json > sshd_journal.json

# Show kernel messages (USB events, module loads)
journalctl -k --output=json > kernel_journal.json

# Filter by priority (0=emerg to 7=debug)
journalctl -p err --output=json > errors.json

# Boot-specific logs
journalctl -b 0 --output=json > current_boot.json
journalctl --list-boots  # List all recorded boot sessions
Linux Audit 框架分析
# Search audit log for specific event types
ausearch -m USER_AUTH --start today

# Search for file access events
ausearch -f /etc/shadow

# Search for process execution
ausearch -m EXECVE --start "02/01/2025" --end "02/28/2025"

# Generate report of login events
aureport --login --start "02/01/2025"

# Generate summary of failed authentications
aureport --auth --failed

# Search for specific user activity
ausearch -ua 1001  # By UID
ausearch -ua username  # By username
Cron 任务调查
# Check system-wide crontab
cat /etc/crontab

# Check user crontabs
ls -la /var/spool/cron/crontabs/

# Review cron execution logs
grep "CRON" /var/log/syslog

# Check for at/batch jobs
ls -la /var/spool/at/
atq

Python 取证日志解析器

import re
import json
import sys
import os
from datetime import datetime
from collections import defaultdict


class LinuxLogForensicAnalyzer:
    """Analyze Linux system logs for forensic investigation."""

    def __init__(self, log_dir: str, output_dir: str):
        self.log_dir = log_dir
        self.output_dir = output_dir
        os.makedirs(output_dir, exist_ok=True)

    def parse_auth_log(self, auth_log_path: str) -> dict:
        """Parse auth.log for authentication events."""
        events = {
            "successful_logins": [],
            "failed_logins": [],
            "sudo_commands": [],
            "account_changes": [],
            "ssh_sessions": []
        }

        ssh_accepted = re.compile(
            r'(\w+\s+\d+\s+[\d:]+)\s+(\S+)\s+sshd\[\d+\]:\s+Accepted\s+(\S+)\s+for\s+(\S+)\s+from\s+([\d.]+)'
        )
        ssh_failed = re.compile(
            r'(\w+\s+\d+\s+[\d:]+)\s+(\S+)\s+sshd\[\d+\]:\s+Failed\s+password\s+for\s+(\S*)\s+from\s+([\d.]+)'
        )
        sudo_cmd = re.compile(
            r'(\w+\s+\d+\s+[\d:]+)\s+(\S+)\s+sudo:\s+(\S+)\s+:.*COMMAND=(.*)'
        )
        useradd = re.compile(
            r'(\w+\s+\d+\s+[\d:]+)\s+(\S+)\s+useradd\[\d+\]:\s+new user: name=(\S+)'
        )

        with open(auth_log_path, "r", errors="replace") as f:
            for line in f:
                m = ssh_accepted.search(line)
                if m:
                    events["successful_logins"].append({
                        "timestamp": m.group(1), "host": m.group(2),
                        "method": m.group(3), "user": m.group(4), "source_ip": m.group(5)
                    })
                    continue

                m = ssh_failed.search(line)
                if m:
                    events["failed_logins"].append({
                        "timestamp": m.group(1), "host": m.group(2),
                        "user": m.group(3), "source_ip": m.group(4)
                    })
                    continue

                m = sudo_cmd.search(line)
                if m:
                    events["sudo_commands"].append({
                        "timestamp": m.group(1), "host": m.group(2),
                        "user": m.group(3), "command": m.group(4).strip()
                    })
                    continue

                m = useradd.search(line)
                if m:
                    events["account_changes"].append({
                        "timestamp": m.group(1), "host": m.group(2),
                        "new_user": m.group(3)
                    })

        return events

    def detect_brute_force(self, auth_events: dict, threshold: int = 10) -> list:
        """Detect brute force attempts from auth log data."""
        ip_failures = defaultdict(int)
        for event in auth_events.get("failed_logins", []):
            ip_failures[event["source_ip"]] += 1

        brute_force = []
        for ip, count in ip_failures.items():
            if count >= threshold:
                brute_force.append({"source_ip": ip, "failed_attempts": count})

        return sorted(brute_force, key=lambda x: x["failed_attempts"], reverse=True)

    def generate_report(self, auth_log_path: str) -> str:
        """Generate comprehensive forensic analysis report."""
        auth_events = self.parse_auth_log(auth_log_path)
        brute_force = self.detect_brute_force(auth_events)

        report = {
            "analysis_timestamp": datetime.now().isoformat(),
            "log_source": auth_log_path,
            "summary": {
                "successful_logins": len(auth_events["successful_logins"]),
                "failed_logins": len(auth_events["failed_logins"]),
                "sudo_commands": len(auth_events["sudo_commands"]),
                "account_changes": len(auth_events["account_changes"]),
                "brute_force_sources": len(brute_force)
            },
            "brute_force_detected": brute_force,
            "auth_events": auth_events
        }

        report_path = os.path.join(self.output_dir, "linux_log_forensics.json")
        with open(report_path, "w") as f:
            json.dump(report, f, indent=2)

        print(f"[*] Successful logins: {report['summary']['successful_logins']}")
        print(f"[*] Failed logins: {report['summary']['failed_logins']}")
        print(f"[*] Sudo commands: {report['summary']['sudo_commands']}")
        print(f"[*] Brute force sources: {report['summary']['brute_force_sources']}")
        return report_path


def main():
    if len(sys.argv) < 3:
        print("Usage: python process.py <auth_log_path> <output_dir>")
        sys.exit(1)
    analyzer = LinuxLogForensicAnalyzer(os.path.dirname(sys.argv[1]), sys.argv[2])
    analyzer.generate_report(sys.argv[1])


if __name__ == "__main__":
    main()

参考资料

评论

支持 Markdown