如何编写审计日志脚本

wen 实用脚本 2026-06-29 206

本文目录导读：

如何编写审计日志脚本

基础Linux系统审计日志脚本
Python高级审计日志脚本
日志监控和告警脚本
定时任务配置 (crontab)
使用说明

我来帮你编写一个全面的审计日志脚本,下面提供几种不同场景的审计日志脚本方案：

基础Linux系统审计日志脚本

#!/bin/bash
# 审计日志脚本 - Linux系统审计
# 文件名: audit_log.sh
# 配置
LOG_DIR="/var/log/audit"
CURRENT_DATE=$(date +"%Y%m%d_%H%M%S")
LOG_FILE="${LOG_DIR}/audit_${CURRENT_DATE}.log"
# 确保日志目录存在
mkdir -p "$LOG_DIR"
# 日志函数
write_log() {
    local timestamp=$(date +"%Y-%m-%d %H:%M:%S")
    local user=$(whoami)
    local ip=$(echo $SSH_CLIENT | awk '{print $1}')
    echo "[${timestamp}] [${user}@${ip}] $1" >> "$LOG_FILE"
}
# 开始审计
echo "==========================================" >> "$LOG_FILE"
write_log "开始系统审计日志记录"
# 1. 记录登录信息
write_log "=== 当前登录用户信息 ==="
who >> "$LOG_FILE"
write_log "=== 历史登录记录 ==="
last -10 >> "$LOG_FILE"
# 2. 记录系统信息
write_log "=== 系统运行时间 ==="
uptime >> "$LOG_FILE"
# 3. 记录进程信息
write_log "=== 关键进程列表 ==="
ps aux --sort=-%mem | head -20 >> "$LOG_FILE"
# 4. 记录网络连接
write_log "=== 当前网络连接 ==="
netstat -tunap 2>/dev/null >> "$LOG_FILE"
# 5. 记录文件系统使用
write_log "=== 磁盘使用情况 ==="
df -h >> "$LOG_FILE"
# 6. 记录最后修改的配置文件
write_log "=== 最近修改的配置文件 ==="
find /etc -type f -mmin -1440 -ls >> "$LOG_FILE"
echo "==========================================" >> "$LOG_FILE"
write_log "审计完成"
# 设置日志权限
chmod 640 "$LOG_FILE"

Python高级审计日志脚本

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
高级审计日志脚本
功能：监控文件变化、用户操作、系统事件
"""
import os
import sys
import json
import hashlib
import logging
import platform
from datetime import datetime
from logging.handlers import RotatingFileHandler
import psutil
import subprocess
class AuditLogger:
    def __init__(self, log_path='/var/log/audit'):
        self.log_path = log_path
        self.setup_logging()
    def setup_logging(self):
        """配置日志系统"""
        if not os.path.exists(self.log_path):
            os.makedirs(self.log_path)
        # 创建日志文件
        log_file = os.path.join(self.log_path, 
                                f"audit_{datetime.now().strftime('%Y%m%d')}.log")
        # 配置日志格式
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - [%(process)d] - %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        )
        # 文件处理器 (按大小轮转)
        file_handler = RotatingFileHandler(
            log_file, 
            maxBytes=10*1024*1024,  # 10MB
            backupCount=5
        )
        file_handler.setFormatter(formatter)
        # 控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setFormatter(formatter)
        # 配置日志器
        self.logger = logging.getLogger('AuditLogger')
        self.logger.setLevel(logging.INFO)
        self.logger.addHandler(file_handler)
        self.logger.addHandler(console_handler)
    def audit_user_login(self):
        """记录用户登录信息"""
        self.logger.info("=== 用户登录审计 ===")
        # 当前登录用户
        current_users = subprocess.run(['who'], capture_output=True, text=True)
        self.logger.info(f"当前登录用户:\n{current_users.stdout}")
        # 登录历史
        last_logins = subprocess.run(['last', '-5'], capture_output=True, text=True)
        self.logger.info(f"最近登录记录:\n{last_logins.stdout}")
    def audit_system_resources(self):
        """审计系统资源使用情况"""
        self.logger.info("=== 系统资源审计 ===")
        # CPU信息
        cpu_percent = psutil.cpu_percent(interval=1)
        self.logger.info(f"CPU使用率: {cpu_percent}%")
        # 内存信息
        memory = psutil.virtual_memory()
        self.logger.info(f"内存使用: {memory.percent}% (总:{memory.total/1024/1024:.0f}MB)")
        # 磁盘信息
        disk = psutil.disk_usage('/')
        self.logger.info(f"磁盘使用: {disk.percent}% (总:{disk.total/1024/1024/1024:.1f}GB)")
    def audit_network_connections(self):
        """审计网络连接"""
        self.logger.info("=== 网络连接审计 ===")
        connections = psutil.net_connections()
        for conn in connections[:20]:  # 限制显示前20个
            if conn.status == 'LISTEN':
                self.logger.info(f"监听端口: {conn.laddr.port} - PID: {conn.pid}")
    def audit_file_changes(self, watched_dirs=['/etc', '/root']):
        """审计文件变化"""
        self.logger.info("=== 文件变化审计 ===")
        file_hashes = {}
        for directory in watched_dirs:
            if os.path.exists(directory):
                for root, dirs, files in os.walk(directory):
                    for file in files:
                        file_path = os.path.join(root, file)
                        try:
                            # 计算文件哈希
                            with open(file_path, 'rb') as f:
                                file_hash = hashlib.sha256(f.read()).hexdigest()
                            # 检查文件权限
                            permissions = oct(os.stat(file_path).st_mode)[-3:]
                            file_info = {
                                'path': file_path,
                                'hash': file_hash,
                                'permissions': permissions,
                                'modified': datetime.fromtimestamp(
                                    os.path.getmtime(file_path)
                                ).isoformat()
                            }
                            self.logger.debug(f"监控文件: {json.dumps(file_info)}")
                        except (PermissionError, FileNotFoundError):
                            continue
    def audit_processes(self):
        """审计运行进程"""
        self.logger.info("=== 进程审计 ===")
        suspicious_processes = {
            'nmap', 'wireshark', 'tcpdump', 'hydra',
            'john', 'aircrack', 'ettercap'
        }
        for proc in psutil.process_iter(['pid', 'name', 'cpu_percent', 'memory_percent']):
            try:
                process_name = proc.info['name'].lower()
                # 检测可疑进程
                if process_name in suspicious_processes:
                    self.logger.warning(
                        f"发现可疑进程: {process_name} (PID: {proc.info['pid']})"
                    )
                # 检测高资源消耗进程
                if proc.info['cpu_percent'] > 80 or proc.info['memory_percent'] > 80:
                    self.logger.warning(
                        f"高资源消耗进程: {process_name} - "
                        f"CPU: {proc.info['cpu_percent']}%, "
                        f"内存: {proc.info['memory_percent']:.1f}%"
                    )
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
    def generate_report(self):
        """生成审计报告"""
        report = {
            'timestamp': datetime.now().isoformat(),
            'hostname': platform.node(),
            'os': f"{platform.system()} {platform.release()}",
            'audit_results': '完成'
        }
        report_file = os.path.join(
            self.log_path,
            f"report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
        )
        with open(report_file, 'w') as f:
            json.dump(report, f, indent=2)
        self.logger.info(f"审计报告已生成: {report_file}")
def main():
    """主函数"""
    audit = AuditLogger()
    print("=" * 50)
    print(f"开始系统审计 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    print("=" * 50)
    # 执行审计任务
    audit.audit_user_login()
    audit.audit_system_resources()
    audit.audit_network_connections()
    audit.audit_file_changes()
    audit.audit_processes()
    audit.generate_report()
    print("=" * 50)
    print("审计完成！")
    print("=" * 50)
if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        print("\n审计被用户中断")
        sys.exit(1)
    except Exception as e:
        print(f"审计出错: {e}")
        sys.exit(1)

日志监控和告警脚本

#!/bin/bash
# 实时日志监控和告警系统
# 文件名: log_monitor.sh
# 配置
LOG_FILE="/var/log/audit/audit_*.log"
ALERT_EMAIL="admin@example.com"
THRESHOLD_FAILED_LOGIN=5
MONITOR_INTERVAL=60  # 秒
# 初始化计数器
declare -A FAILED_LOGIN_COUNT
# 日志解析函数
parse_log_line() {
    local line="$1"
    local timestamp user action
    # 解析不同的日志格式
    if [[ $line =~ ^\[([0-9-]+\ [0-9:]+)\] ]]; then
        timestamp="${BASH_REMATCH[1]}"
    fi
    if [[ $line =~ \[([^@]+)@([^\]]+)\] ]]; then
        user="${BASH_REMATCH[1]}"
    fi
    echo "$timestamp $user $line"
}
# 告警函数
send_alert() {
    local subject="$1"
    local message="$2"
    echo "$message" | mail -s "审计告警: $subject" "$ALERT_EMAIL"
    logger "审计告警: $subject"
}
# 监控函数
monitor_logs() {
    echo "开始监控审计日志..."
    tail -f $LOG_FILE | while read line; do
        # 解析日志行
        parsed=$(parse_log_line "$line")
        # 检测失败登录
        if echo "$line" | grep -qi "failed\|denied\|error"; then
            echo "检测到异常: $line"
            # 统计失败次数
            user=$(echo "$line" | grep -oP '\@[^\]]+' | cut -d@ -f2)
            if [ ! -z "$user" ]; then
                ((FAILED_LOGIN_COUNT[$user]++))
                if [ ${FAILED_LOGIN_COUNT[$user]} -ge $THRESHOLD_FAILED_LOGIN ]; then
                    send_alert "检测到多次登录失败" "用户 $user 登录失败次数超过阈值"
                    FAILED_LOGIN_COUNT[$user]=0
                fi
            fi
        fi
        # 检测敏感操作
        if echo "$line" | grep -qi "sudo\|su\|root"; then
            echo "检测到提权操作: $line"
        fi
        # 检测文件操作
        if echo "$line" | grep -qi "delete\|remove\|chmod\|chown"; then
            echo "检测到敏感文件操作: $line"
        fi
    done
}
# 清理函数
cleanup() {
    echo "停止日志监控..."
    exit 0
}
# 注册清理函数
trap cleanup SIGINT SIGTERM
# 主函数
main() {
    # 检查权限
    if [ "$(id -u)" != "0" ]; then
        echo "需要root权限运行此脚本"
        exit 1
    fi
    # 检查日志文件
    if ! ls $LOG_FILE 1>/dev/null 2>&1; then
        echo "未找到审计日志文件"
        exit 1
    fi
    # 开始监控
    monitor_logs
}
main

定时任务配置 (crontab)

# 每天凌晨2点执行审计
0 2 * * * /path/to/audit_log.sh
# 每小时检查日志
0 * * * * /path/to/log_monitor.sh
# 每周生成汇总报告
0 3 * * 1 /path/to/generate_report.sh

使用说明

安装依赖 (Python版本需要)
```
pip install psutil
```

设置权限

chmod +x audit_log.sh log_monitor.sh
chmod 640 /var/log/audit/*.log

配置告警 (如需邮件通知)

# 安装邮件服务
apt-get install mailutils  # Debian/Ubuntu
yum install mailx          # CentOS/RHEL

定期查看日志
```
tail -f /var/log/audit/audit_*.log
```

这些脚本提供了完整的审计日志解决方案,可以根据实际需求进行调整。

本文地址： https://www.bmjup.com/post/11872.html