审计日志分析脚本

#!/usr/bin/env python3
# audit_log_analysis.py
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: http://www.fgedu.net.cn

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns
from elasticsearch import Elasticsearch
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

# 配置参数
ELASTICSEARCH_URL = 'http://localhost:9200'
EMAIL_SENDER = 'audit@fgedu.net.cn'
EMAIL_RECEIVER = 'admin@fgedu.net.cn'
EMAIL_SERVER = 'smtp.fgedu.net.cn'
EMAIL_PORT = 587
EMAIL_USERNAME = 'audit@fgedu.net.cn'
EMAIL_PASSWORD = 'Password123'

# 连接Elasticsearchfrom DB视频:www.itpux.com
es = Elasticsearch(ELASTICSEARCH_URL)

# 获取审计日志
def get_audit_logs(start_time, end_time):
    query = {
        "query": {
            "bool": {
                "must": [
                    {
                        "range": {
                            "@timestamp": {
                                "gte": start_time,
                                "lte": end_time
                            }
                        }
                    }
                ]
            }
        },
        "size": 10000
    }
    
    response = es.search(index="opengauss-audit-*", body=query)
    hits = response['hits']['hits']
    
    if not hits:
        return pd.DataFrame()
    
    data = []
    for hit in hits:
        source = hit['_source']
        data.append({
            'timestamp': source.get('@timestamp'),
            'user': source.get('user'),
            'database': source.get('database'),
            'client_ip': source.get('client_ip'),
            'client_port': source.get('client_port'),
            'object_type': source.get('object_type'),
            'object_name': source.get('object_name'),
            'operation_type': source.get('operation_type'),
            'result': source.get('result'),
            'detail': source.get('detail')
        })
    
    return pd.DataFrame(data)

# 分析审计日志
def analyze_audit_logs(logs):
    if logs.empty:
        return {}
    
    analysis = {}
    
    # 登录失败分析
    login_failures = logs[(logs['operation_type'] == 'login') & (logs['result'] == 'fail')]
    analysis['login_failures'] = len(login_failures)
    
    # 权限变更分析
    permission_changes = logs[logs['operation_type'].str.contains('permission', case=False)]
    analysis['permission_changes'] = len(permission_changes)
    
    # 数据修改分析
    data_modifications = logs[logs['operation_type'].isin(['insert', 'update', 'delete'])]
    analysis['data_modifications'] = len(data_modifications)
    
    # 用户活动分析
    user_activity = logs.groupby('user').size().sort_values(ascending=False)
    analysis['top_users'] = user_activity.head(5).to_dict()
    
    # 客户端IP分析
    client_ip_activity = logs.groupby('client_ip').size().sort_values(ascending=False)
    analysis['top_client_ips'] = client_ip_activity.head(5).to_dict()
    
    return analysis

# 发送告警邮件
def send_alert_email(analysis):
    if not analysis:
        return
    
    subject = "openGauss安全审计告警"
    
    body = "
openGauss安全审计分析报告
"
    body += f"
登录失败次数：{analysis.get('login_failures', 0)}
"
    body += f"
权限变更次数：{analysis.get('permission_changes', 0)}
"
    body += f"
数据修改次数：{analysis.get('data_modifications', 0)}
"
    
    if analysis.get('top_users'):
        body += "
活动用户Top 5：
"
        for user, count in analysis['top_users'].items():
            body += f"
{user}: {count}次操作
"
        body += "
"
    
    if analysis.get('top_client_ips'):
        body += "
活动IP Top 5：
"
        for ip, count in analysis['top_client_ips'].items():
            body += f"
{ip}: {count}次操作
"
        body += "
"
    
    msg = MIMEMultipart()
    msg['From'] = EMAIL_SENDER
    msg['To'] = EMAIL_RECEIVER
    msg['Subject'] = subject
    msg.attach(MIMEText(body, 'html'))
    
    try:
        server = smtplib.SMTP(EMAIL_SERVER, EMAIL_PORT)
        server.starttls()
        server.login(EMAIL_USERNAME, EMAIL_PASSWORD)
        server.send_message(msg)
        server.quit()
        print("告警邮件发送成功")
    except Exception as e:
        print(f"告警邮件发送失败: {e}")

# 主函数
def main():
    # 获取最近24小时的审计日志
    import datetime
    end_time = datetime.datetime.now()
    start_time = end_time - datetime.timedelta(hours=24)
    
    logs = get_audit_logs(start_time.isoformat(), end_time.isoformat())
    analysis = analyze_audit_logs(logs)
    
    # 发送告警邮件
    if analysis.get('login_failures', 0) > 5 or analysis.get('permission_changes', 0) > 0:
        send_alert_email(analysis)
    
    # 生成分析报告
    print("安全审计分析报告：")
    print(f"登录失败次数：{analysis.get('login_failures', 0)}")
    print(f"权限变更次数：{analysis.get('permission_changes', 0)}")
    print(f"数据修改次数：{analysis.get('data_modifications', 0)}")
    print("活动用户Top 5：")
    for user, count in analysis.get('top_users', {}).items():
        print(f"  {user}: {count}次操作")
    print("活动IP Top 5：")
    for ip, count in analysis.get('top_client_ips', {}).items():
        print(f"  {ip}: {count}次操作")

if __name__ == '__main__':
    main()

#!/bin/bash
# security_incident_response.sh
# from:www.itpux.com.qq113257174.wx:itpux-com
# web: http://www.fgedu.net.cn

# 定义变量
LOG_FILE="/opengauss/logs/security_incident_response.log"
INCIDENT_DIR="/opengauss/incidents/$(date +'%Y%m%d_%H%M%S')"

# 日志函数
log() {
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1" >> $LOG_FILE
    echo "[$(date +'%Y-%m-%d %H:%M:%S')] $1"
}

# 创建事件目录
create_incident_dir() {
    log "创建事件目录 $INCIDENT_DIR..."
    mkdir -p $INCIDENT_DIR
    log "事件目录创建完成"
}

# 收集证据
collect_evidence() {
    log "收集证据..."
    
    # 收集审计日志
    cp -r /opengauss/logs/audit $INCIDENT_DIR/
    
    # 收集系统日志
    cp -r /var/log $INCIDENT_DIR/
    
    # 收集数据库状态
    gsql -U fgedu -d postgres -c "SELECT * FROM pg_stat_activity; 
" > $INCIDENT_DIR/pg_stat_activity.txt
    gsql -U fgedu -d postgres -c "SELECT * FROM pg_roles; 
" > $INCIDENT_DIR/pg_roles.txt
    gsql -U fgedu -d postgres -c "SELECT * FROM pg_database; 
" > $INCIDENT_DIR/pg_database.txt
    
    log "证据收集完成"
}

# 隔离受影响的系统
isolate_system() {
    log "隔离受影响的系统..."
    
    # 限制网络访问
    firewall-cmd --add-rich-rule='rule family="ipv4" source address="0.0.0.0/0" port protocol="tcp" port="5432" reject' --permanent
    firewall-cmd --reload
    
    # 锁定可疑用户
    gsql -U fgedu -d postgres -c "ALTER ROLE suspicious_user WITH NOLOGIN;"
    
    log "系统隔离完成"
}

# 分析事件
analyze_incident() {
    log "分析事件..."
    
    # 分析审计日志
    python3 /opengauss/scripts/analyze_audit_logs.py $INCIDENT_DIR/audit > $INCIDENT_DIR/analysis_report.txt
    
    # 生成事件报告
    cat > $INCIDENT_DIR/incident_report.txt << EOF
事件报告
=========
事件时间：$(date +'%Y-%m-%d %H:%M:%S')
事件类型：安全事件
事件描述：可疑操作检测
影响范围：openGauss数据库
分析结果：
$(cat $INCIDENT_DIR/analysis_report.txt)
EOF
    
    log "事件分析完成"
}

# 恢复系统
recover_system() {
    log "恢复系统..."
    
    # 恢复网络访问
    firewall-cmd --remove-rich-rule='rule family="ipv4" source address="0.0.0.0/0" port protocol="tcp" port="5432" reject' --permanent
    firewall-cmd --reload
    
    # 恢复用户权限
    gsql -U fgedu -d postgres -c "ALTER ROLE suspicious_user WITH LOGIN;"
    
    log "系统恢复完成"
}

# 通知相关人员
notify_stakeholders() {
    log "通知相关人员..."
    
    # 发送邮件通知
    echo "安全事件报告" | mail -s "openGauss安全事件通知" -a $INCIDENT_DIR/incident_report.txt admin@fgedu.net.cn
    
    log "通知发送完成"
}

# 主流程
log "=== 安全事件响应开始 ==="

create_incident_dir
collect_evidence
isolate_system
analyze_incident
recover_system
notify_stakeholders

log "=== 安全事件响应完成 ==="