本文目录导读：
我来详细介绍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)} 个安全问题")
这些方法涵盖了：
- 读取不同数据库的审计日志
- 解析和分析审计数据
- 实时监控审计事件
- 异常检测和报警
- 可视化和报表生成
根据你的具体需求选择合适的方案。
