Python脚本如何操作数据库审计日志

本文目录导读：

1. MySQL审计日志
2. PostgreSQL审计日志
3. 使用PyAudit进行通用审计
4. 实时审计监控系统
5. 审计日志分析工具

我来详细介绍Python操作数据库审计日志的方法。

MySQL审计日志

开启MySQL审计日志

-- 安装审计插件
INSTALL PLUGIN audit_log SONAME 'audit_log.so';
-- 开启审计日志
SET GLOBAL audit_log_policy = 'ALL';
SET GLOBAL audit_log_format = 'JSON';

Python读取MySQL审计日志

import json
import re
from datetime import datetime
import mysql.connector
from pathlib import Path
class MySQLAuditLogReader:
    def __init__(self, log_path="/var/log/mysql/audit.log"):
        self.log_path = Path(log_path)
    def parse_audit_log(self):
        """解析MySQL审计日志"""
        audit_entries = []
        with open(self.log_path, 'r') as f:
            for line in f:
                try:
                    entry = json.loads(line.strip())
                    # 提取关键审计信息
                    audit_info = {
                        'timestamp': entry.get('time', ''),
                        'user': entry.get('user', ''),
                        'host': entry.get('host', ''),
                        'query': entry.get('query', ''),
                        'status': entry.get('status', ''),
                        'rows_examined': entry.get('rows_examined', 0),
                        'thread_id': entry.get('thread_id', '')
                    }
                    audit_entries.append(audit_info)
                except json.JSONDecodeError:
                    continue
        return audit_entries
    def filter_by_user(self, entries, username):
        """按用户筛选审计日志"""
        return [e for e in entries if e['user'] == username]
    def filter_by_time_range(self, entries, start_time, end_time):
        """按时间范围筛选"""
        filtered = []
        for entry in entries:
            try:
                entry_time = datetime.strptime(
                    entry['timestamp'], 
                    '%Y-%m-%d %H:%M:%S'
                )
                if start_time <= entry_time <= end_time:
                    filtered.append(entry)
            except ValueError:
                continue
        return filtered
# 使用示例
reader = MySQLAuditLogReader()
all_logs = reader.parse_audit_log()
# 查询特定用户的日志
user_logs = reader.filter_by_user(all_logs, 'admin')
# 查询时间范围内的日志
from datetime import datetime, timedelta
now = datetime.now()
start = now - timedelta(hours=1)
recent_logs = reader.filter_by_time_range(all_logs, start, now)

PostgreSQL审计日志

配置PostgreSQL审计

-- 修改postgresql.conf
log_destination = 'csvlog'
logging_collector = on
log_statement = 'all'
log_line_prefix = '%t %u %d %p %r '

Python读取PostgreSQL审计日志

import csv
from datetime import datetime
import psycopg2
from pathlib import Path
class PostgreSQLAuditLogReader:
    def __init__(self, log_directory="/var/log/postgresql"):
        self.log_directory = Path(log_directory)
    def parse_csv_logs(self, log_file):
        """解析PostgreSQL CSV格式审计日志"""
        audit_entries = []
        with open(log_file, 'r') as f:
            reader = csv.reader(f)
            for row in reader:
                if len(row) >= 12:
                    entry = {
                        'timestamp': row[0],
                        'user': row[3],
                        'database': row[4],
                        'process_id': row[5],
                        'remote_host': row[7],
                        'session_id': row[8],
                        'command_tag': row[10],
                        'statement': row[11],
                        'error_severity': row[9] if len(row) > 12 else ''
                    }
                    audit_entries.append(entry)
        return audit_entries
    def detect_anomalies(self, entries):
        """检测异常操作"""
        anomalies = []
        for entry in entries:
            statement = entry.get('statement', '').upper()
            # 检测敏感操作
            if any(keyword in statement for keyword in [
                'DROP', 'TRUNCATE', 'DELETE', 'ALTER SYSTEM'
            ]):
                anomalies.append({
                    'type': 'dangerous_operation',
                    'entry': entry
                })
            # 检测批量删除
            if 'DELETE' in statement and 'WHERE' not in statement:
                anomalies.append({
                    'type': 'mass_deletion',
                    'entry': entry
                })
        return anomalies
# 使用示例
pg_reader = PostgreSQLAuditLogReader()
logs = pg_reader.parse_csv_logs('/var/log/postgresql/postgresql-2023-01-01.csv')
anomalies = pg_reader.detect_anomalies(logs)
print(f"发现 {len(anomalies)} 个异常操作")

使用PyAudit进行通用审计

import pyaudit
import logging
from datetime import datetime
class DatabaseAuditor:
    """数据库审计管理器"""
    def __init__(self, db_config):
        self.db_config = db_config
        self.setup_audit_logging()
    def setup_audit_logging(self):
        """设置审计日志系统"""
        # 创建审计日志记录器
        self.audit_logger = logging.getLogger('database_audit')
        self.audit_logger.setLevel(logging.INFO)
        # 文件处理器
        file_handler = logging.FileHandler(
            f'audit_{datetime.now().strftime("%Y%m%d")}.log'
        )
        file_handler.setLevel(logging.INFO)
        # 格式化
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)s - %(message)s'
        )
        file_handler.setFormatter(formatter)
        self.audit_logger.addHandler(file_handler)
    def log_query_audit(self, user, query, status="success"):
        """记录查询审计"""
        audit_entry = {
            'timestamp': datetime.now().isoformat(),
            'user': user,
            'query': query,
            'status': status,
            'ip': self.get_client_ip()
        }
        self.audit_logger.info(json.dumps(audit_entry))
    def get_client_ip(self):
        """获取客户端IP"""
        # 实际实现会根据数据库连接信息获取
        return "192.168.1.100"
    def query_audit_logs(self, start_date=None, end_date=None):
        """查询审计日志"""
        logs = []
        log_file = f'audit_{datetime.now().strftime("%Y%m%d")}.log'
        try:
            with open(log_file, 'r') as f:
                for line in f:
                    log_entry = json.loads(
                        line.split(' - ')[-1].strip()
                    )
                    if start_date and end_date:
                        log_time = datetime.fromisoformat(
                            log_entry['timestamp']
                        )
                        if start_date <= log_time <= end_date:
                            logs.append(log_entry)
                    else:
                        logs.append(log_entry)
        except FileNotFoundError:
            pass
        return logs
# 使用示例
auditor = DatabaseAuditor({
    'host': 'localhost',
    'port': 3306,
    'user': 'admin',
    'password': 'password'
})
# 记录审计
auditor.log_query_audit(
    user='john_doe',
    query='SELECT * FROM users WHERE id = 1'
)
# 查询审计日志
recent_logs = auditor.query_audit_logs(
    start_date=datetime.now().replace(hour=0, minute=0, second=0)
)

实时审计监控系统

import asyncio
import asyncpg
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time
class RealTimeAuditSystem:
    """实时审计监控系统"""
    def __init__(self, log_path, db_config):
        self.log_path = log_path
        self.db_config = db_config
        self.observers = []
    async def start_monitoring(self):
        """开始实时监控"""
        # 设置文件监控
        event_handler = AuditLogHandler(self.process_audit_entry)
        observer = Observer()
        observer.schedule(event_handler, self.log_path, recursive=False)
        observer.start()
        # 启动异步处理
        await self.process_audit_queue()
    async def process_audit_entry(self, entry):
        """处理审计条目"""
        async with asyncpg.create_pool(**self.db_config) as pool:
            async with pool.acquire() as conn:
                await conn.execute(
                    """
                    INSERT INTO audit_logs 
                    (timestamp, user, query, status)
                    VALUES ($1, $2, $3, $4)
                    """,
                    entry['timestamp'],
                    entry['user'],
                    entry['query'],
                    entry['status']
                )
    async def process_audit_queue(self):
        """处理审计队列"""
        queue = asyncio.Queue()
        while True:
            if not queue.empty():
                entry = await queue.get()
                await self.process_audit_entry(entry)
            await asyncio.sleep(0.1)
class AuditLogHandler(FileSystemEventHandler):
    """审计日志文件事件处理器"""
    def __init__(self, callback):
        self.callback = callback
        self.last_position = 0
    def on_modified(self, event):
        if event.src_path.endswith('.log'):
            self.process_new_entries(event.src_path)
    def process_new_entries(self, file_path):
        """处理新写入的日志条目"""
        with open(file_path, 'r') as f:
            f.seek(self.last_position)
            new_lines = f.readlines()
            self.last_position = f.tell()
            for line in new_lines:
                try:
                    entry = json.loads(line.strip())
                    self.callback(entry)
                except json.JSONDecodeError:
                    continue

审计日志分析工具

import pandas as pd
import matplotlib.pyplot as plt
from collections import Counter
import re
class AuditLogAnalyzer:
    """审计日志分析器"""
    def __init__(self, logs):
        self.logs = logs
        self.df = pd.DataFrame(logs)
    def analyze_user_activity(self):
        """分析用户活动"""
        user_activity = self.df['user'].value_counts()
        print("\n用户活动统计：")
        print("-" * 30)
        for user, count in user_activity.items():
            print(f"{user}: {count} 次操作")
        # 可视化
        plt.figure(figsize=(10, 6))
        user_activity.plot(kind='bar')
        plt.title('用户操作频率统计')
        plt.xlabel('用户')
        plt.ylabel('操作次数')
        plt.xticks(rotation=45)
        plt.tight_layout()
        plt.savefig('user_activity.png')
    def analyze_query_types(self):
        """分析查询类型"""
        query_types = Counter()
        for log in self.logs:
            query = log.get('query', '').upper().strip()
            if query.startswith('SELECT'):
                query_types['SELECT'] += 1
            elif query.startswith('INSERT'):
                query_types['INSERT'] += 1
            elif query.startswith('UPDATE'):
                query_types['UPDATE'] += 1
            elif query.startswith('DELETE'):
                query_types['DELETE'] += 1
            elif query.startswith('DROP'):
                query_types['DROP'] += 1
            else:
                query_types['OTHER'] += 1
        print("\n查询类型分布：")
        print("-" * 30)
        for qtype, count in query_types.most_common():
            percentage = (count / len(self.logs)) * 100
            print(f"{qtype}: {count} ({percentage:.1f}%)")
        # 饼图
        plt.figure(figsize=(8, 8))
        plt.pie(query_types.values(), 
                labels=query_types.keys(),
                autopct='%1.1f%%')
        plt.title('查询类型分布')
        plt.savefig('query_types.png')
    def detect_security_risks(self):
        """检测安全风险"""
        risks = []
        for log in self.logs:
            query = log.get('query', '').upper()
            # SQL注入尝试检测
            if re.search(r"'.*OR.*1.*=.*1|--|;.*DROP", query):
                risks.append({
                    'type': 'sql_injection',
                    'user': log.get('user'),
                    'time': log.get('timestamp'),
                    'query': log.get('query')
                })
            # 敏感数据访问检测
            sensitive_tables = ['password', 'credit_card', 'ssn', 'secret']
            for table in sensitive_tables:
                if table.upper() in query:
                    risks.append({
                        'type': 'sensitive_data_access',
                        'user': log.get('user'),
                        'time': log.get('timestamp'),
                        'table': table,
                        'query': log.get('query')
                    })
        return risks
# 使用示例
analyzer = AuditLogAnalyzer(all_logs)
analyzer.analyze_user_activity()
analyzer.analyze_query_types()
risks = analyzer.detect_security_risks()
print(f"发现 {len(risks)} 个安全问题")

这些方法涵盖了：

• 读取不同数据库的审计日志
• 解析和分析审计数据
• 实时监控审计事件
• 异常检测和报警
• 可视化和报表生成

根据你的具体需求选择合适的方案。