094_数字取证进阶技术：数据库取证与结构化数据分析实战指南——从SQL注入到事务日志的全面证据提取方法

-- MySQL数据库备份示例
mysqldump --all-databases --single-transaction --routines --triggers --events > full_backup.sql

-- PostgreSQL数据库备份示例
pg_dumpall > full_backup.sql

-- SQL Server数据库备份示例（使用T-SQL）
BACKUP DATABASE [DatabaseName] TO DISK = 'D:\\forensics\\DatabaseName.bak' WITH COMPRESSION, INIT;

-- Oracle数据库备份示例（使用RMAN）
RMAN> BACKUP DATABASE;

# 使用Python脚本获取进程内存（Windows环境）
import ctypes
import os
import struct

def dump_process_memory(pid, output_file):
    # 打开目标进程
    PROCESS_ALL_ACCESS = 0x1F0FFF
    kernel32 = ctypes.WinDLL('kernel32', use_last_error=True)
    handle = kernel32.OpenProcess(PROCESS_ALL_ACCESS, False, pid)
    
    if not handle:
        raise ctypes.WinError(ctypes.get_last_error())
    
    try:
        # 获取系统信息以确定内存页大小
        class SYSTEM_INFO(ctypes.Structure):
            _fields_ = [
                ("wProcessorArchitecture", ctypes.c_ushort),
                ("wReserved", ctypes.c_ushort),
                ("dwPageSize", ctypes.c_uint),
                ("lpMinimumApplicationAddress", ctypes.c_void_p),
                ("lpMaximumApplicationAddress", ctypes.c_void_p),
                ("dwActiveProcessorMask", ctypes.c_void_p),
                ("dwNumberOfProcessors", ctypes.c_uint),
                ("dwProcessorType", ctypes.c_uint),
                ("dwAllocationGranularity", ctypes.c_uint),
                ("wProcessorLevel", ctypes.c_ushort),
                ("wProcessorRevision", ctypes.c_ushort),
            ]
        
        system_info = SYSTEM_INFO()
        kernel32.GetSystemInfo(ctypes.byref(system_info))
        
        # 初始化输出文件
        with open(output_file, 'wb') as f:
            # 遍历进程地址空间
            addr = system_info.lpMinimumApplicationAddress
            while addr < system_info.lpMaximumApplicationAddress:
                # 尝试读取内存页
                buffer = ctypes.create_string_buffer(system_info.dwPageSize)
                bytes_read = ctypes.c_size_t(0)
                
                # 检查内存页是否可访问
                if kernel32.ReadProcessMemory(
                    handle, 
                    addr, 
                    buffer, 
                    system_info.dwPageSize, 
                    ctypes.byref(bytes_read)
                ):
                    # 写入内存数据和地址信息
                    f.write(struct.pack('<Q', addr))  # 写入地址
                    f.write(struct.pack('<I', bytes_read.value))  # 写入数据大小
                    f.write(buffer[:bytes_read.value])  # 写入数据
                
                # 移动到下一页
                addr = ctypes.cast(
                    ctypes.c_void_p(ctypes.cast(addr, ctypes.c_void_p).value + system_info.dwPageSize),
                    ctypes.c_void_p
                )
    finally:
        kernel32.CloseHandle(handle)

# 使用示例：转储MySQL进程内存
try:
    # 假设MySQL进程ID为1234
    dump_process_memory(1234, "mysql_memory_dump.raw")
    print("进程内存转储成功")
except Exception as e:
    print(f"错误: {e}")

# MySQL日志文件位置
- 错误日志: /var/log/mysql/error.log 或 datadir/mysqld.log
- 查询日志: 默认禁用，可通过设置general_log = 1启用
- 慢查询日志: datadir/slow_query.log
- 二进制日志: datadir/mysql-bin.*

# PostgreSQL日志文件位置
- 错误日志: /var/log/postgresql/postgresql-<version>-main.log
- 查询日志: 需配置log_statement参数

# SQL Server日志文件位置
- 错误日志: C:\\Program Files\\Microsoft SQL Server\\MSSQL<version>.<instance>\\MSSQL\\Log\\ERRORLOG
- SQL Agent日志: SQL Server Management Studio中的SQL Server Agent节点
- 事务日志: 与数据库文件位于同一目录，扩展名为.ldf

# MySQL二进制日志分析脚本
import subprocess
import re
import json

def analyze_mysql_binlogs(binlog_dir, output_file):
    results = {
        'events': [],
        'summary': {
            'total_events': 0,
            'event_types': {}
        }
    }
    
    # 获取二进制日志文件列表
    try:
        # 使用mysqlbinlog工具列出所有二进制日志文件
        list_cmd = ['mysqlbinlog', '--list', '--base64-output=decode-rows', '--verbose', binlog_dir + 'mysql-bin.000001']
        list_output = subprocess.run(list_cmd, capture_output=True, text=True, check=True)
        
        # 分析每个二进制日志文件
        binlog_files = re.findall(r'mysql-bin\.\d+', list_output.stdout)
        for binlog_file in binlog_files:
            print(f"分析二进制日志文件: {binlog_file}")
            
            # 使用mysqlbinlog工具解析二进制日志
            cmd = [
                'mysqlbinlog', 
                '--base64-output=decode-rows', 
                '--verbose', 
                f'{binlog_dir}{binlog_file}'
            ]
            
            try:
                output = subprocess.run(cmd, capture_output=True, text=True, check=True)
                
                # 解析输出以提取事件
                events = re.split(r'# at \d+\n#', output.stdout)
                
                for event in events[1:]:  # 跳过第一个空条目
                    event_data = {}
                    
                    # 提取时间戳
                    timestamp_match = re.search(r'Timestamp: (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})', event)
                    if timestamp_match:
                        event_data['timestamp'] = timestamp_match.group(1)
                    
                    # 提取事件类型
                    event_type_match = re.search(r'\n# (\w+)', event)
                    if event_type_match:
                        event_type = event_type_match.group(1)
                        event_data['type'] = event_type
                        
                        # 更新摘要统计
                        if event_type in results['summary']['event_types']:
                            results['summary']['event_types'][event_type] += 1
                        else:
                            results['summary']['event_types'][event_type] = 1
                    
                    # 提取服务器ID
                    server_id_match = re.search(r'server id \d+', event)
                    if server_id_match:
                        event_data['server_id'] = server_id_match.group(0)
                    
                    # 提取SQL语句（如果有）
                    sql_match = re.search(r'(?<=\n).*?(?=\nDELIMITER|$)', event, re.DOTALL)
                    if sql_match:
                        sql = sql_match.group(0).strip()
                        event_data['sql'] = sql
                        
                        # 检查是否包含敏感操作
                        sensitive_commands = ['DROP', 'DELETE', 'TRUNCATE', 'ALTER', 'INSERT', 'UPDATE']
                        for cmd in sensitive_commands:
                            if re.search(rf'^{cmd}\s+', sql, re.IGNORECASE):
                                event_data['is_sensitive'] = True
                                break
                    
                    results['events'].append(event_data)
                    results['summary']['total_events'] += 1
                    
            except subprocess.CalledProcessError as e:
                print(f"分析二进制日志文件 {binlog_file} 失败: {e}")
                continue
    
    except Exception as e:
        print(f"获取二进制日志文件列表失败: {e}")
    
    # 保存结果到JSON文件
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(results, f, indent=2, ensure_ascii=False)
    
    print(f"分析完成，共处理 {results['summary']['total_events']} 个事件")
    print(f"事件类型统计: {results['summary']['event_types']}")
    print(f"结果已保存到 {output_file}")
    
    return results

# 使用示例
# analyze_mysql_binlogs('/var/lib/mysql/', 'mysql_binlog_analysis.json')

# InnoDB存储引擎文件
- .frm文件: 表结构定义文件
- .ibd文件: 独立表空间文件（存储表数据和索引）
- ibdata1文件: 共享表空间文件
- ib_logfile0/ib_logfile1: 重做日志文件
- ib_buffer_pool: 缓冲池信息

# MyISAM存储引擎文件
- .frm文件: 表结构定义文件
- .MYD文件: 数据文件
- .MYI文件: 索引文件

# 其他重要文件
- mysql-bin.*: 二进制日志文件
- mysql.err: 错误日志文件
- mysql.pid: 进程ID文件

# 分析MySQL配置文件的Python脚本
import re
import json

def analyze_mycnf(config_path):
    results = {
        'general': {},
        'security': {
            'weak_settings': [],
            'recommended_settings': []
        },
        'performance': {},
        'logging': {},
        'storage_engines': {}
    }
    
    try:
        with open(config_path, 'r') as f:
            config_content = f.read()
        
        # 提取基本配置
        basedir_match = re.search(r'basedir\s*=\s*(.+)', config_content)
        if basedir_match:
            results['general']['basedir'] = basedir_match.group(1).strip()
        
        datadir_match = re.search(r'datadir\s*=\s*(.+)', config_content)
        if datadir_match:
            results['general']['datadir'] = datadir_match.group(1).strip()
        
        port_match = re.search(r'port\s*=\s*(\d+)', config_content)
        if port_match:
            results['general']['port'] = port_match.group(1).strip()
        
        # 提取安全相关配置
        if not re.search(r'bind-address\s*=\s*127\.0\.0\.1', config_content):
            results['security']['weak_settings'].append('bind-address not set to 127.0.0.1, potentially exposing MySQL to external network')
        
        if not re.search(r'skip-networking', config_content) and not re.search(r'bind-address\s*=\s*127\.0\.0\.1', config_content):
            results['security']['weak_settings'].append('Networking not disabled and not bound to localhost only')
        
        if not re.search(r'local-infile\s*=\s*0', config_content):
            results['security']['weak_settings'].append('local-infile not disabled, may allow LOAD DATA attacks')
        
        if not re.search(r'ssl-ca|ssl-cert|ssl-key', config_content):
            results['security']['weak_settings'].append('SSL not configured, connections may be unencrypted')
        
        # 提取日志配置
        log_error_match = re.search(r'log-error\s*=\s*(.+)', config_content)
        if log_error_match:
            results['logging']['error_log'] = log_error_match.group(1).strip()
        
        general_log_match = re.search(r'general_log\s*=\s*(\w+)', config_content)
        if general_log_match:
            results['logging']['general_log'] = general_log_match.group(1).strip()
            if general_log_match.group(1).strip().lower() == 'on':
                log_file_match = re.search(r'general_log_file\s*=\s*(.+)', config_content)
                if log_file_match:
                    results['logging']['general_log_file'] = log_file_match.group(1).strip()
        
        slow_query_log_match = re.search(r'slow_query_log\s*=\s*(\w+)', config_content)
        if slow_query_log_match:
            results['logging']['slow_query_log'] = slow_query_log_match.group(1).strip()
            if slow_query_log_match.group(1).strip().lower() == 'on':
                slow_query_log_file_match = re.search(r'slow_query_log_file\s*=\s*(.+)', config_content)
                if slow_query_log_file_match:
                    results['logging']['slow_query_log_file'] = slow_query_log_file_match.group(1).strip()
        
        # 提取存储引擎配置
        default_storage_engine_match = re.search(r'default-storage-engine\s*=\s*(\w+)', config_content)
        if default_storage_engine_match:
            results['storage_engines']['default'] = default_storage_engine_match.group(1).strip()
        
        # 提取二进制日志配置
        log_bin_match = re.search(r'log-bin\s*=\s*(.+)', config_content)
        if log_bin_match:
            results['logging']['binary_log'] = log_bin_match.group(1).strip()
            results['security']['recommended_settings'].append('Binary logging enabled')
        else:
            results['security']['weak_settings'].append('Binary logging not enabled, limited audit capabilities')
        
        # 添加安全建议
        results['security']['recommendations'] = [
            'Set bind-address to 127.0.0.1 to restrict connections',
            'Disable local-infile to prevent data theft',
            'Configure SSL for encrypted connections',
            'Enable binary logging for audit purposes',
            'Set appropriate expire_logs_days to manage log file size',
            'Use strong passwords for all database accounts',
            'Limit user privileges to minimum required'
        ]
        
    except Exception as e:
        print(f"分析MySQL配置文件失败: {e}")
        return None
    
    return results

# 使用示例
# result = analyze_mycnf('/etc/mysql/my.cnf')
# print(json.dumps(result, indent=2))

# 提取MySQL二进制日志中的事务信息
import subprocess
import re
import json
from datetime import datetime

def extract_mysql_transactions(binlog_file):
    transactions = []
    current_transaction = None
    
    # 使用mysqlbinlog工具解析二进制日志
    cmd = [
        'mysqlbinlog',
        '--base64-output=decode-rows',
        '--verbose',
        binlog_file
    ]
    
    try:
        output = subprocess.run(cmd, capture_output=True, text=True, check=True)
        lines = output.stdout.split('\n')
        
        # 事务开始和结束标记
        tx_start_pattern = re.compile(r'START TRANSACTION')
        tx_commit_pattern = re.compile(r'COMMIT')
        tx_rollback_pattern = re.compile(r'ROLLBACK')
        timestamp_pattern = re.compile(r'Timestamp: (\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})')
        sql_pattern = re.compile(r'^([^#].*)$')
        
        for line in lines:
            # 提取时间戳
            timestamp_match = timestamp_pattern.search(line)
            if timestamp_match:
                current_time = timestamp_match.group(1)
            
            # 检查事务开始
            if tx_start_pattern.search(line):
                current_transaction = {
                    'start_time': current_time,
                    'end_time': None,
                    'status': 'active',
                    'queries': [],
                    'affected_tables': set()
                }
                transactions.append(current_transaction)
            
            # 检查SQL语句
            sql_match = sql_pattern.match(line)
            if sql_match and current_transaction:
                sql = sql_match.group(1).strip()
                if sql:
                    current_transaction['queries'].append(sql)
                    
                    # 尝试提取受影响的表名
                    table_match = re.search(r'FROM\s+(\w+)', sql, re.IGNORECASE)
                    if not table_match:
                        table_match = re.search(r'INTO\s+(\w+)', sql, re.IGNORECASE)
                    if not table_match:
                        table_match = re.search(r'UPDATE\s+(\w+)', sql, re.IGNORECASE)
                    if not table_match:
                        table_match = re.search(r'DELETE\s+FROM\s+(\w+)', sql, re.IGNORECASE)
                    if not table_match:
                        table_match = re.search(r'CREATE\s+TABLE\s+(\w+)', sql, re.IGNORECASE)
                    
                    if table_match:
                        current_transaction['affected_tables'].add(table_match.group(1))
            
            # 检查事务提交
            if tx_commit_pattern.search(line) and current_transaction:
                current_transaction['end_time'] = current_time
                current_transaction['status'] = 'committed'
                current_transaction['affected_tables'] = list(current_transaction['affected_tables'])
            
            # 检查事务回滚
            if tx_rollback_pattern.search(line) and current_transaction:
                current_transaction['end_time'] = current_time
                current_transaction['status'] = 'rolled_back'
                current_transaction['affected_tables'] = list(current_transaction['affected_tables'])
    
    except subprocess.CalledProcessError as e:
        print(f"解析二进制日志失败: {e}")
        return None
    
    # 统计信息
    summary = {
        'total_transactions': len(transactions),
        'committed_transactions': sum(1 for tx in transactions if tx['status'] == 'committed'),
        'rolled_back_transactions': sum(1 for tx in transactions if tx['status'] == 'rolled_back'),
        'active_transactions': sum(1 for tx in transactions if tx['status'] == 'active'),
        'unique_tables_affected': len(set(table for tx in transactions for table in tx['affected_tables']))
    }
    
    return {
        'summary': summary,
        'transactions': transactions
    }

# 使用示例
# tx_data = extract_mysql_transactions('/var/lib/mysql/mysql-bin.000001')
# if tx_data:
#     print(f"总事务数: {tx_data['summary']['total_transactions']}")
#     print(f"已提交事务: {tx_data['summary']['committed_transactions']}")
#     print(f"已回滚事务: {tx_data['summary']['rolled_back_transactions']}")
#     print(f"受影响的唯一表数: {tx_data['summary']['unique_tables_affected']}")

-- 列出所有用户账户
SELECT user, host FROM mysql.user;

-- 检查用户权限
SELECT user, host, select_priv, insert_priv, update_priv, delete_priv, create_priv, drop_priv 
FROM mysql.user;

-- 检查特定用户的所有权限
SHOW GRANTS FOR 'username'@'host';

-- 检查近期登录失败记录（如果开启了审计）
SELECT * FROM mysql.general_log 
WHERE command_type = 'Connect' AND argument LIKE '%Failed%' 
ORDER BY event_time DESC;

-- 检查正在执行的查询（当前活动）
SHOW PROCESSLIST;

-- 检查慢查询
SELECT * FROM mysql.slow_log ORDER BY start_time DESC LIMIT 10;

# 分析MySQL用户活动日志
import re
from collections import defaultdict
import json

def analyze_mysql_user_activity(log_file):
    results = {
        'login_attempts': {
            'successful': [],
            'failed': []
        },
        'user_activity': defaultdict(list),
        'ip_addresses': defaultdict(int),
        'summary': {
            'total_logins': 0,
            'failed_logins': 0,
            'unique_users': 0,
            'unique_ips': 0,
            'suspicious_activities': []
        }
    }
    
    try:
        with open(log_file, 'r') as f:
            lines = f.readlines()
        
        # 模式匹配
        login_success_pattern = re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*?Access denied for user.*?\((Using password: YES)\)')
        login_failure_pattern = re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*?Access denied for user.*?\((Using password: (?:YES|NO))\)')
        connection_pattern = re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*?Connected.*?user=(\w+).*?host=(.*?)\)')
        query_pattern = re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*?Query.*?user=(\w+).*?@(.*?)\s(.*)')
        admin_action_pattern = re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}).*?Query.*?(CREATE USER|ALTER USER|DROP USER|GRANT|REVOKE|FLUSH PRIVILEGES)', re.IGNORECASE)
        
        # 敏感操作模式
        sensitive_commands = ['DROP TABLE', 'TRUNCATE TABLE', 'DELETE FROM', 'UPDATE.*WHERE', 'ALTER TABLE']
        sensitive_patterns = [re.compile(cmd, re.IGNORECASE) for cmd in sensitive_commands]
        
        for line in lines:
            # 检查登录成功
            success_match = connection_pattern.search(line)
            if success_match:
                timestamp = success_match.group(1)
                username = success_match.group(2)
                ip = success_match.group(3)
                
                login_record = {
                    'timestamp': timestamp,
                    'user': username,
                    'ip': ip
                }
                
                results['login_attempts']['successful'].append(login_record)
                results['user_activity'][username].append({
                    'timestamp': timestamp,
                    'type': 'login',
                    'status': 'success',
                    'ip': ip
                })
                results['ip_addresses'][ip] += 1
                results['summary']['total_logins'] += 1
            
            # 检查登录失败
            failure_match = login_failure_pattern.search(line)
            if failure_match:
                timestamp = failure_match.group(1)
                
                # 尝试提取用户名和IP
                user_match = re.search(r'user=(\'[^\']+\')', line)
                ip_match = re.search(r'host=(\'[^\']+\')', line)
                
                username = user_match.group(1) if user_match else 'unknown'
                ip = ip_match.group(1) if ip_match else 'unknown'
                
                login_record = {
                    'timestamp': timestamp,
                    'user': username,
                    'ip': ip
                }
                
                results['login_attempts']['failed'].append(login_record)
                results['ip_addresses'][ip] += 1
                results['summary']['failed_logins'] += 1
                
                # 记录可疑活动
                if username != 'unknown' and ip != 'unknown':
                    results['summary']['suspicious_activities'].append({
                        'timestamp': timestamp,
                        'type': 'failed_login',
                        'user': username,
                        'ip': ip
                    })
            
            # 检查查询活动
            query_match = query_pattern.search(line)
            if query_match:
                timestamp = query_match.group(1)
                username = query_match.group(2)
                ip = query_match.group(3)
                query = query_match.group(4)
                
                activity_type = 'query'
                
                # 检查是否为敏感操作
                for i, pattern in enumerate(sensitive_patterns):
                    if pattern.search(query):
                        activity_type = 'sensitive_query'
                        results['summary']['suspicious_activities'].append({
                            'timestamp': timestamp,
                            'type': 'sensitive_operation',
                            'user': username,
                            'ip': ip,
                            'query': query
                        })
                        break
                
                results['user_activity'][username].append({
                    'timestamp': timestamp,
                    'type': activity_type,
                    'query': query,
                    'ip': ip
                })
            
            # 检查管理员操作
            admin_match = admin_action_pattern.search(line)
            if admin_match:
                timestamp = admin_match.group(1)
                action = admin_match.group(2)
                
                # 尝试提取用户名和IP
                user_match = re.search(r'user=(\w+)', line)
                ip_match = re.search(r'@(\S+)', line)
                
                username = user_match.group(1) if user_match else 'unknown'
                ip = ip_match.group(1) if ip_match else 'unknown'
                
                results['summary']['suspicious_activities'].append({
                    'timestamp': timestamp,
                    'type': 'admin_action',
                    'user': username,
                    'ip': ip,
                    'action': action
                })
        
        # 更新摘要统计
        results['summary']['unique_users'] = len(results['user_activity'])
        results['summary']['unique_ips'] = len(results['ip_addresses'])
        
        # 识别可疑IP（失败登录次数过多）
        suspicious_ips = [ip for ip, count in results['ip_addresses'].items() if count > 10]
        if suspicious_ips:
            results['summary']['suspicious_ips'] = suspicious_ips
        
    except Exception as e:
        print(f"分析MySQL日志文件失败: {e}")
        return None
    
    return results

# 使用示例
# activity_data = analyze_mysql_user_activity('/var/log/mysql/error.log')
# if activity_data:
#     print(f"总登录次数: {activity_data['summary']['total_logins']}")
#     print(f"失败登录次数: {activity_data['summary']['failed_logins']}")
#     print(f"唯一用户数: {activity_data['summary']['unique_users']}")
#     print(f"可疑活动数: {len(activity_data['summary']['suspicious_activities'])}")

-- 从MySQL一般查询日志中查找可能的SQL注入尝试
SELECT * FROM mysql.general_log 
WHERE argument LIKE '%--%' 
   OR argument LIKE '%/*%*/%' 
   OR argument LIKE '%1=1%' 
   OR argument LIKE '%UNION SELECT%' 
   OR argument LIKE '%SELECT * FROM%' 
   OR argument LIKE '%OR 1=%' 
   OR argument LIKE '%EXEC sp_%' 
   OR argument LIKE '%xp_cmdshell%' 
ORDER BY event_time DESC;

# 检测日志文件中的SQL注入尝试
import re
from collections import defaultdict
import json

def detect_sql_injection(log_file):
    injection_patterns = [
        # 基本注入模式
        r'\bOR\s+\d+=\d+\b',
        r'\bAND\s+\d+=\d+\b',
        r'\bUNION\s+(ALL\s+)?SELECT\b',
        r'\bSELECT\s+\*\s+FROM\b',
        r'\bINSERT\s+INTO\b',
        r'\bUPDATE\s+.*?SET\s+.*?=.*?WHERE\b',
        r'\bDELETE\s+FROM\b',
        r'\bDROP\s+(TABLE|DATABASE)\b',
        
        # 注释模式
        r'--[^\r\n]*$',
        r'/\*.*?\*/',
        r';[^\s]*--[^\r\n]*$',
        
        # 特殊字符序列
        r'\bCHAR\s*\(.*?\)',
        r'\bCONCAT\s*\(.*?\)',
        r'\bGROUP_CONCAT\s*\(.*?\)',
        r'\bEXEC\s+\(',
        r'\bEXEC\s+sp_',
        r'\bEXECUTE\s+sp_',
        r'\bxp_cmdshell\b',
        
        # 绕过技术
        r'\bOR\s+\'[^\']*\'\s*=\s*\'[^\']*\'',
        r'\bOR\s+\d+\s*(=|LIKE)\s*\d+',
        r'\b--\+',
        r'\b--\s*#',
        r'\b;\s*--',
        r'\bUNION\s+DISTINCT\s+SELECT\b',
        
        # 数据泄露尝试
        r'\bversion\s*\(\)',
        r'\bdatabase\s*\(\)',
        r'\buser\s*\(\)',
        r'\bsystem_user\s*\(\)',
        r'\bcurrent_user\s*\(\)',
        r'\bload_file\s*\(\)',
        r'\binto\s+outfile\b',
        r'\binto\s+dumpfile\b'
    ]
    
    results = {
        'injection_attempts': [],
        'attackers': defaultdict(lambda: {'ip': '', 'attempts': 0, 'patterns': set()}),
        'affected_endpoints': defaultdict(int),
        'summary': {
            'total_attempts': 0,
            'unique_attackers': 0,
            'unique_patterns': 0,
            'affected_pages': 0
        }
    }
    
    unique_patterns_set = set()
    
    try:
        with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
            lines = f.readlines()
        
        for line in lines:
            # 提取常见日志格式的IP地址和请求路径
            ip_match = re.search(r'\b(?:\d{1,3}\.){3}\d{1,3}\b', line)
            url_match = re.search(r'GET\s+(.*?)\s+HTTP|POST\s+(.*?)\s+HTTP', line)
            timestamp_match = re.search(r'(\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2})', line)
            
            ip = ip_match.group(0) if ip_match else 'unknown'
            url = url_match.group(1) or url_match.group(2) if url_match else 'unknown'
            timestamp = timestamp_match.group(1) if timestamp_match else 'unknown'
            
            # 检查是否包含SQL注入模式
            for pattern in injection_patterns:
                if re.search(pattern, line, re.IGNORECASE):
                    # 记录攻击尝试
                    injection_record = {
                        'timestamp': timestamp,
                        'ip': ip,
                        'url': url,
                        'pattern': pattern,
                        'raw_log': line.strip()[:500]  # 截断长日志行
                    }
                    
                    results['injection_attempts'].append(injection_record)
                    unique_patterns_set.add(pattern)
                    
                    # 更新攻击者信息
                    if ip != 'unknown':
                        results['attackers'][ip]['ip'] = ip
                        results['attackers'][ip]['attempts'] += 1
                        results['attackers'][ip]['patterns'].add(pattern)
                    
                    # 更新受影响的端点
                    if url != 'unknown':
                        results['affected_endpoints'][url] += 1
        
        # 更新摘要统计
        results['summary']['total_attempts'] = len(results['injection_attempts'])
        results['summary']['unique_attackers'] = len(results['attackers'])
        results['summary']['unique_patterns'] = len(unique_patterns_set)
        results['summary']['affected_pages'] = len(results['affected_endpoints'])
        
        # 识别最活跃的攻击者
        if results['attackers']:
            top_attackers = sorted(
                results['attackers'].items(), 
                key=lambda x: x[1]['attempts'], 
                reverse=True
            )[:5]
            
            results['summary']['top_attackers'] = [
                {
                    'ip': ip,
                    'attempts': data['attempts'],
                    'unique_patterns': len(data['patterns'])
                }
                for ip, data in top_attackers
            ]
        
        # 识别最常见的攻击模式
        pattern_counts = defaultdict(int)
        for attempt in results['injection_attempts']:
            pattern_counts[attempt['pattern']] += 1
        
        if pattern_counts:
            top_patterns = sorted(
                pattern_counts.items(), 
                key=lambda x: x[1], 
                reverse=True
            )[:5]
            
            results['summary']['top_patterns'] = [
                {'pattern': pattern, 'count': count}
                for pattern, count in top_patterns
            ]
        
    except Exception as e:
        print(f"分析日志文件失败: {e}")
        return None
    
    return results

# 使用示例
# injection_data = detect_sql_injection('/var/log/apache2/access.log')
# if injection_data:
#     print(f"检测到 {injection_data['summary']['total_attempts']} 次SQL注入尝试")
#     print(f"来自 {injection_data['summary']['unique_attackers']} 个唯一IP地址")
#     print(f"使用了 {injection_data['summary']['unique_patterns']} 种不同的攻击模式")

-- 从备份恢复数据库
RESTORE DATABASE [AffectedDatabase] FROM DISK = 'path_to_backup.bak' WITH REPLACE;

-- 修复受损的表结构
CHECK TABLE table_name;
REPAIR TABLE table_name;

-- 重新创建受损的存储过程
DELIMITER //
CREATE PROCEDURE safe_procedure(IN param VARCHAR(50))
BEGIN
    -- 使用参数化查询防止SQL注入
    SELECT * FROM users WHERE username = param;
END //
DELIMITER ;

-- 检查并移除恶意创建的用户
SELECT user, host FROM mysql.user WHERE user NOT IN ('root', 'mysql.sys', 'mysql.session', 'mysql.infoschema');
DROP USER 'suspicious_user'@'host';
FLUSH PRIVILEGES;

# 检查数据库完整性的Python脚本
import pymysql
import json

def check_database_integrity(host, user, password, database):
    results = {
        'tables': {},
        'suspicious_objects': [],
        'security_issues': [],
        'summary': {
            'total_tables': 0,
            'corrupted_tables': 0,
            'suspicious_users': 0,
            'weak_permissions': 0
        }
    }
    
    try:
        # 连接数据库
        conn = pymysql.connect(
            host=host,
            user=user,
            password=password,
            database=database
        )
        cursor = conn.cursor(pymysql.cursors.DictCursor)
        
        # 检查所有表
        cursor.execute("SHOW TABLES")
        tables = cursor.fetchall()
        results['summary']['total_tables'] = len(tables)
        
        for table in tables:
            table_name = list(table.values())[0]
            
            # 检查表结构
            cursor.execute(f"DESCRIBE {table_name}")
            columns = cursor.fetchall()
            
            # 检查是否有可疑列（例如包含password但没有加密相关名称的列）
            for column in columns:
                col_name = column['Field'].lower()
                col_type = column['Type'].lower()
                
                if 'password' in col_name and not any(x in col_type for x in ['hash', 'md5', 'sha']):
                    results['security_issues'].append({
                        'type': 'weak_password_storage',
                        'table': table_name,
                        'column': column['Field'],
                        'description': f"列 {column['Field']} 可能存储未加密密码"
                    })
                    results['summary']['weak_permissions'] += 1
            
            # 检查表索引
            cursor.execute(f"SHOW INDEX FROM {table_name}")
            indexes = cursor.fetchall()
            
            # 检查是否有主键
            has_primary_key = any(index['Key_name'] == 'PRIMARY' for index in indexes)
            if not has_primary_key and table_name not in ['sessions', 'temp_data']:
                results['security_issues'].append({
                    'type': 'missing_primary_key',
                    'table': table_name,
                    'description': "表缺少主键，可能导致数据完整性问题"
                })
            
            # 检查表数据完整性
            try:
                cursor.execute(f"CHECK TABLE {table_name}")
                check_result = cursor.fetchone()
                if check_result and check_result['Msg_text'] != 'OK':
                    results['summary']['corrupted_tables'] += 1
                    results['tables'][table_name] = {
                        'status': 'corrupted',
                        'message': check_result['Msg_text']
                    }
                else:
                    results['tables'][table_name] = {
                        'status': 'ok',
                        'column_count': len(columns),
                        'has_primary_key': has_primary_key
                    }
            except Exception as e:
                results['summary']['corrupted_tables'] += 1
                results['tables'][table_name] = {
                    'status': 'error',
                    'message': str(e)
                }
        
        # 检查可疑用户和权限
        cursor.execute("SELECT user, host, Select_priv, Insert_priv, Update_priv, Delete_priv, Create_priv, Drop_priv FROM mysql.user")
        users = cursor.fetchall()
        
        for user in users:
            # 检查可疑用户名
            username = user['user']
            if any(x in username.lower() for x in ['test', 'admin', 'root'] if username not in ['mysql.sys', 'mysql.session', 'mysql.infoschema']):
                results['suspicious_objects'].append({
                    'type': 'suspicious_user',
                    'name': username,
                    'host': user['host'],
                    'description': f"可疑用户名: {username}@"{user['host']}"
                })
                results['summary']['suspicious_users'] += 1
            
            # 检查过度权限
            privs = ['Select_priv', 'Insert_priv', 'Update_priv', 'Delete_priv', 'Create_priv', 'Drop_priv']
            user_privs = [p for p in privs if user[p] == 'Y']
            
            if len(user_privs) >= 5 and username not in ['root']:
                results['security_issues'].append({
                    'type': 'excessive_privileges',
                    'user': username,
                    'host': user['host'],
                    'privileges': user_privs,
                    'description': f"用户 {username}@"{user['host']}" 拥有过多权限"
                })
        
        # 检查存储过程和函数
        cursor.execute("SHOW PROCEDURE STATUS WHERE Db = %s", (database,))
        procedures = cursor.fetchall()
        
        for proc in procedures:
            proc_name = proc['Name']
            # 检查可疑的存储过程名称
            if any(x in proc_name.lower() for x in ['exec', 'shell', 'system', 'dump', 'load']):
                cursor.execute(f"SHOW CREATE PROCEDURE {proc_name}")
                create_proc = cursor.fetchone()
                proc_definition = create_proc['Create Procedure'] if create_proc else ''
                
                results['suspicious_objects'].append({
                    'type': 'suspicious_procedure',
                    'name': proc_name,
                    'definition': proc_definition[:500],  # 截断长定义
                    'description': f"可疑存储过程: {proc_name}"
                })
        
        # 检查触发器
        cursor.execute("SHOW TRIGGERS LIKE '%'")
        triggers = cursor.fetchall()
        
        for trigger in triggers:
            cursor.execute(f"SHOW CREATE TRIGGER {trigger['Trigger']}")
            create_trigger = cursor.fetchone()
            trigger_definition = create_trigger['SQL Original Statement'] if create_trigger else ''
            
            # 检查触发器中是否包含可疑操作
            if any(x in trigger_definition.lower() for x in ['drop', 'delete', 'truncate', 'insert', 'update']):
                results['suspicious_objects'].append({
                    'type': 'suspicious_trigger',
                    'name': trigger['Trigger'],
                    'table': trigger['Table'],
                    'definition': trigger_definition[:500],
                    'description': f"可疑触发器: {trigger['Trigger']} 作用于表 {trigger['Table']}"
                })
        
    except Exception as e:
        print(f"数据库完整性检查失败: {e}")
        return None
    finally:
        if 'conn' in locals():
            conn.close()
    
    return results

# 使用示例
# integrity_data = check_database_integrity('localhost', 'admin', 'password', 'testdb')
# if integrity_data:
#     print(json.dumps(integrity_data['summary'], indent=2))

# MongoDB数据文件
- .wt文件: WiredTiger存储引擎的数据文件
- collection-*.wt: 集合数据文件
- index-*.wt: 索引文件
- _mdb_catalog.wt: 目录文件
- sizeStorer.wt: 大小存储文件
- WiredTiger: WiredTiger引擎元数据目录
- WiredTiger.wt: WiredTiger引擎核心文件
- WiredTiger.turtle: WiredTiger引擎配置文件

# MongoDB日志分析脚本
import re
from collections import defaultdict
import json
from datetime import datetime

def analyze_mongodb_log(log_file):
    results = {
        'connections': {
            'successful': [],
            'failed': []
        },
        'operations': defaultdict(list),
        'errors': [],
        'warnings': [],
        'authentication': [],
        'summary': {
            'total_connections': 0,
            'failed_connections': 0,
            'unique_ips': 0,
            'error_count': 0,
            'warning_count': 0,
            'auth_failures': 0
        }
    }
    
    ip_addresses = set()
    
    try:
        with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
            lines = f.readlines()
        
        # MongoDB日志格式的正则表达式
        # 典型格式: 2023-05-10T14:25:36.789+0000 I NETWORK  [listener] connection accepted from 192.168.1.100:54321 #123 (1 connection now open)
        timestamp_pattern = re.compile(r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}\+\d{4})')
        connection_pattern = re.compile(r'connection accepted from ([\d\.]+):(\d+)')
        disconnection_pattern = re.compile(r'end connection ([\d\.]+):(\d+)')
        error_pattern = re.compile(r'E\s+(\w+)\s+\[([^\]]+)\]\s+(.+)')
        warning_pattern = re.compile(r'W\s+(\w+)\s+\[([^\]]+)\]\s+(.+)')
        auth_success_pattern = re.compile(r'authenticated as (\w+)')
        auth_failure_pattern = re.compile(r'Authentication failed for (\w+)')
        operation_pattern = re.compile(r'I\s+(\w+)\s+\[([^\]]+)\]\s+command\s+([\w\.]+)\.\$cmd\s+command: (.+)')
        
        for line in lines:
            # 提取时间戳
            timestamp_match = timestamp_pattern.search(line)
            timestamp = timestamp_match.group(1) if timestamp_match else 'unknown'
            
            # 检查连接
            conn_match = connection_pattern.search(line)
            if conn_match:
                ip = conn_match.group(1)
                port = conn_match.group(2)
                ip_addresses.add(ip)
                
                results['connections']['successful'].append({
                    'timestamp': timestamp,
                    'ip': ip,
                    'port': port,
                    'raw_log': line.strip()
                })
                results['summary']['total_connections'] += 1
            
            # 检查断开连接
            disconn_match = disconnection_pattern.search(line)
            if disconn_match:
                ip = disconn_match.group(1)
                port = disconn_match.group(2)
                
                results['operations']['disconnections'].append({
                    'timestamp': timestamp,
                    'ip': ip,
                    'port': port,
                    'raw_log': line.strip()
                })
            
            # 检查错误
            error_match = error_pattern.search(line)
            if error_match:
                error_type = error_match.group(1)
                component = error_match.group(2)
                message = error_match.group(3)
                
                results['errors'].append({
                    'timestamp': timestamp,
                    'type': error_type,
                    'component': component,
                    'message': message,
                    'raw_log': line.strip()
                })
                results['summary']['error_count'] += 1
                
                # 特别关注认证失败
                auth_fail_match = auth_failure_pattern.search(message)
                if auth_fail_match:
                    username = auth_fail_match.group(1)
                    results['authentication'].append({
                        'timestamp': timestamp,
                        'type': 'failure',
                        'username': username,
                        'raw_log': line.strip()
                    })
                    results['summary']['auth_failures'] += 1
            
            # 检查警告
            warning_match = warning_pattern.search(line)
            if warning_match:
                warning_type = warning_match.group(1)
                component = warning_match.group(2)
                message = warning_match.group(3)
                
                results['warnings'].append({
                    'timestamp': timestamp,
                    'type': warning_type,
                    'component': component,
                    'message': message,
                    'raw_log': line.strip()
                })
                results['summary']['warning_count'] += 1
            
            # 检查认证成功
            auth_success_match = auth_success_pattern.search(line)
            if auth_success_match and 'I ACCESS' in line:
                username = auth_success_match.group(1)
                
                results['authentication'].append({
                    'timestamp': timestamp,
                    'type': 'success',
                    'username': username,
                    'raw_log': line.strip()
                })
            
            # 检查数据库操作
            op_match = operation_pattern.search(line)
            if op_match:
                op_type = op_match.group(1)
                component = op_match.group(2)
                database = op_match.group(3)
                command = op_match.group(4)
                
                # 提取操作类型
                operation = 'unknown'
                if 'find' in command:
                    operation = 'find'
                elif 'insert' in command:
                    operation = 'insert'
                elif 'update' in command:
                    operation = 'update'
                elif 'delete' in command:
                    operation = 'delete'
                elif 'createUser' in command:
                    operation = 'createUser'
                elif 'dropUser' in command:
                    operation = 'dropUser'
                elif 'grantRolesToUser' in command:
                    operation = 'grantRolesToUser'
                
                results['operations'][operation].append({
                    'timestamp': timestamp,
                    'database': database,
                    'command': command[:500],  # 截断长命令
                    'raw_log': line.strip()
                })
                
                # 检查敏感操作
                sensitive_operations = ['createUser', 'dropUser', 'grantRolesToUser', 'dropDatabase']
                if any(so in command for so in sensitive_operations):
                    results['warnings'].append({
                        'timestamp': timestamp,
                        'type': 'SENSITIVE_OPERATION',
                        'database': database,
                        'operation': operation,
                        'message': f"检测到敏感操作: {operation} 在数据库 {database}",
                        'raw_log': line.strip()
                    })
        
        # 更新摘要统计
        results['summary']['unique_ips'] = len(ip_addresses)
        
        # 识别可疑IP（有多次连接或认证失败）
        ip_connection_count = defaultdict(int)
        ip_auth_fail_count = defaultdict(int)
        
        for conn in results['connections']['successful']:
            ip_connection_count[conn['ip']] += 1
        
        for auth in results['authentication']:
            if auth['type'] == 'failure':
                # 尝试从日志中提取IP
                ip_match = re.search(r'from ([\d\.]+):(\d+)', auth['raw_log'])
                if ip_match:
                    ip = ip_match.group(1)
                    ip_auth_fail_count[ip] += 1
        
        suspicious_ips = []
        for ip, count in ip_connection_count.items():
            if count > 50 or ip_auth_fail_count.get(ip, 0) > 3:
                suspicious_ips.append({
                    'ip': ip,
                    'connection_count': count,
                    'auth_failures': ip_auth_fail_count.get(ip, 0)
                })
        
        if suspicious_ips:
            results['summary']['suspicious_ips'] = suspicious_ips
        
    except Exception as e:
        print(f"分析MongoDB日志文件失败: {e}")
        return None
    
    return results

# 使用示例
# mongodb_log_data = analyze_mongodb_log('/var/log/mongodb/mongod.log')
# if mongodb_log_data:
#     print(f"总连接数: {mongodb_log_data['summary']['total_connections']}")
#     print(f"错误数: {mongodb_log_data['summary']['error_count']}")
#     print(f"认证失败: {mongodb_log_data['summary']['auth_failures']}")
#     print(f"唯一IP数: {mongodb_log_data['summary']['unique_ips']}")

# Redis日志文件通常位于以下位置
# Linux: /var/log/redis/redis-server.log
# Windows: 安装目录下的logs文件夹

# Redis持久化文件
# RDB文件: 默认名为dump.rdb
# AOF文件: 默认名为appendonly.aof

# Redis日志分析脚本
import re
from collections import defaultdict
import json

def analyze_redis_log(log_file):
    results = {
        'connections': [],
        'commands': defaultdict(int),
        'errors': [],
        'warnings': [],
        'authentication': [],
        'keyspace': [],
        'summary': {
            'total_connections': 0,
            'unique_clients': 0,
            'error_count': 0,
            'warning_count': 0,
            'auth_failures': 0,
            'top_commands': []
        }
    }
    
    clients = set()
    
    try:
        with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
            lines = f.readlines()
        
        # Redis日志格式的正则表达式
        # 典型格式: [30245] 10 May 14:25:36.789 # Connection accepted from 192.168.1.100:54321
        timestamp_pattern = re.compile(r'\[(\d+)\]\s+(\d+ \w+ \d{2}:\d{2}:\d{2}\.\d{3})')
        connection_pattern = re.compile(r'Connection accepted from ([\d\.]+):(\d+)')
        disconnection_pattern = re.compile(r'Client \d+ (?:closed connection|disconnected)')
        error_pattern = re.compile(r'#\s+(ERR|ERROR|Failed)\s+(.+)')
        warning_pattern = re.compile(r'#\s+WARNING: (.+)')
        auth_success_pattern = re.compile(r'Authentication succeeded')
        auth_failure_pattern = re.compile(r'Authentication failed')
        command_pattern = re.compile(r'\*\d+\r\n\$\d+\r\n([A-Za-z]+)')
        keyspace_pattern = re.compile(r'\*\d+\r\n\$\d+\r\n([A-Za-z]+)\r\n\$\d+\r\n([^\r\n]+)')
        slowlog_pattern = re.compile(r'\*\d+\r\n\$\d+\r\nSLOWLOG\r\n\$\d+\r\nGET\r\n\$\d+\r\n(\d+)')
        
        for line in lines:
            # 提取时间戳和进程ID
            timestamp_match = timestamp_pattern.search(line)
            if timestamp_match:
                pid = timestamp_match.group(1)
                timestamp = timestamp_match.group(2)
            else:
                timestamp = 'unknown'
                pid = 'unknown'
            
            # 检查连接
            conn_match = connection_pattern.search(line)
            if conn_match:
                ip = conn_match.group(1)
                port = conn_match.group(2)
                clients.add(f"{ip}:{port}")
                
                results['connections'].append({
                    'timestamp': timestamp,
                    'pid': pid,
                    'ip': ip,
                    'port': port,
                    'raw_log': line.strip()
                })
                results['summary']['total_connections'] += 1
            
            # 检查错误
            error_match = error_pattern.search(line)
            if error_match:
                error_type = error_match.group(1)
                message = error_match.group(2)
                
                results['errors'].append({
                    'timestamp': timestamp,
                    'type': error_type,
                    'message': message,
                    'raw_log': line.strip()
                })
                results['summary']['error_count'] += 1
                
                # 特别关注认证失败
                if 'Authentication failed' in message:
                    # 尝试提取IP
                    ip_match = re.search(r'from ([\d\.]+):(\d+)', line)
                    ip = ip_match.group(1) if ip_match else 'unknown'
                    
                    results['authentication'].append({
                        'timestamp': timestamp,
                        'type': 'failure',
                        'ip': ip,
                        'raw_log': line.strip()
                    })
                    results['summary']['auth_failures'] += 1
            
            # 检查警告
            warning_match = warning_pattern.search(line)
            if warning_match:
                message = warning_match.group(1)
                
                results['warnings'].append({
                    'timestamp': timestamp,
                    'message': message,
                    'raw_log': line.strip()
                })
                results['summary']['warning_count'] += 1
            
            # 检查认证成功
            if auth_success_pattern.search(line):
                # 尝试提取IP
                ip_match = re.search(r'from ([\d\.]+):(\d+)', line)
                ip = ip_match.group(1) if ip_match else 'unknown'
                
                results['authentication'].append({
                    'timestamp': timestamp,
                    'type': 'success',
                    'ip': ip,
                    'raw_log': line.strip()
                })
            
            # 检查命令（如果日志包含详细的命令信息）
            command_match = command_pattern.search(line)
            if command_match:
                command = command_match.group(1).upper()
                results['commands'][command] += 1
                
                # 检查敏感命令
                sensitive_commands = ['FLUSHALL', 'FLUSHDB', 'CONFIG', 'DEBUG', 'SHUTDOWN', 'SLAVEOF', 'REPLICAOF']
                if command in sensitive_commands:
                    results['warnings'].append({
                        'timestamp': timestamp,
                        'type': 'SENSITIVE_COMMAND',
                        'command': command,
                        'message': f"检测到敏感命令: {command}",
                        'raw_log': line.strip()
                    })
            
            # 检查键空间操作
            keyspace_match = keyspace_pattern.search(line)
            if keyspace_match:
                command = keyspace_match.group(1).upper()
                key = keyspace_match.group(2)
                
                results['keyspace'].append({
                    'timestamp': timestamp,
                    'command': command,
                    'key': key,
                    'raw_log': line.strip()
                })
            
            # 检查慢查询日志
            slowlog_match = slowlog_pattern.search(line)
            if slowlog_match:
                execution_time = slowlog_match.group(1)
                
                results['warnings'].append({
                    'timestamp': timestamp,
                    'type': 'SLOW_QUERY',
                    'execution_time': execution_time,
                    'message': f"检测到慢查询，执行时间: {execution_time}ms",
                    'raw_log': line.strip()
                })
        
        # 更新摘要统计
        results['summary']['unique_clients'] = len(clients)
        
        # 识别最常用的命令
        sorted_commands = sorted(results['commands'].items(), key=lambda x: x[1], reverse=True)[:10]
        results['summary']['top_commands'] = [
            {'command': cmd, 'count': count}
            for cmd, count in sorted_commands
        ]
        
        # 识别可疑IP（有多次认证失败）
        ip_auth_fail_count = defaultdict(int)
        for auth in results['authentication']:
            if auth['type'] == 'failure' and auth['ip'] != 'unknown':
                ip_auth_fail_count[auth['ip']] += 1
        
        suspicious_ips = [
            {'ip': ip, 'auth_failures': count}
            for ip, count in ip_auth_fail_count.items()
            if count >= 3
        ]
        
        if suspicious_ips:
            results['summary']['suspicious_ips'] = suspicious_ips
        
    except Exception as e:
        print(f"分析Redis日志文件失败: {e}")
        return None
    
    return results

# 使用示例
# redis_log_data = analyze_redis_log('/var/log/redis/redis-server.log')
# if redis_log_data:
#     print(f"总连接数: {redis_log_data['summary']['total_connections']}")
#     print(f"唯一客户端数: {redis_log_data['summary']['unique_clients']}")
#     print(f"认证失败: {redis_log_data['summary']['auth_failures']}")
#     print(f"最常用命令: {redis_log_data['summary']['top_commands'][:5]}")

# 数据库内存取证分析脚本
import volatility.plugins.malware.malfind as malfind
import volatility.plugins.malware.psxview as psxview
import volatility.plugins.dumpfiles as dumpfiles
import volatility.plugins.modscan as modscan
import volatility.plugins.proc_maps as proc_maps
import volatility.plugins.malware.dlllist as dlllist
from volatility.framework import contexts
from volatility.framework.configuration import requirements
from volatility.framework.renderers import treegrid
from volatility.plugins.windows import pslist, dlllist, memmap, vadinfo

# 这是一个概念性脚本，实际使用需要根据Volatility 3的API进行调整
def analyze_database_memory(dump_file, db_process_names=None):
    """
    分析数据库进程的内存转储文件
    
    Args:
        dump_file: 内存转储文件路径
        db_process_names: 数据库进程名称列表，如['mysqld', 'postgres', 'oracle', 'mongod', 'redis-server']
    """
    if db_process_names is None:
        db_process_names = ['mysqld', 'postgres', 'oracle', 'mongod', 'redis-server', 'sqlservr']
    
    results = {
        'processes': [],
        'memory_regions': [],
        'injected_code': [],
        'suspicious_dlls': [],
        'summary': {
            'total_db_processes': 0,
            'suspicious_regions': 0,
            'potential_injections': 0
        }
    }
    
    try:
        # 创建上下文
        context = contexts.Context()
        
        # 设置内存镜像
        # 这里需要根据实际的Volatility 3 API进行调整
        # 加载内存镜像并配置适当的profile
        
        # 枚举所有进程
        print("枚举数据库进程...")
        # process_list = pslist.PsList.list_processes(context)
        
        # 对每个进程进行分析
        # for process in process_list:
        #     proc_name = process.ImageFileName.lower()
        #     
        #     # 检查是否为数据库进程
        #     if proc_name in [name.lower() for name in db_process_names]:
        #         results['summary']['total_db_processes'] += 1
        #         
        #         proc_info = {
        #             'pid': process.UniqueProcessId,
        #             'name': proc_name,
        #             'parent_pid': process.InheritedFromUniqueProcessId,
        #             'image_base': process.ImageBaseAddress
        #         }
        #         
        #         # 获取内存映射
        #         # mem_regions = memmap.Memmap.list_mappings(context, process)
        #         # suspicious_regions = []
        #         # 
        #         # for region in mem_regions:
        #         #     # 检查可疑内存区域（可执行、可写但非可执行文件区域）
        #         #     if region.Protection & 0x40 and region.Protection & 0x04 and not (region.Protection & 0x20):
        #         #         suspicious_regions.append({
        #         #             'base': region.BaseAddress,
        #         #             'size': region.RegionSize,
        #         #             'protection': region.Protection,
        #         #             'state': region.State,
        #         #             'type': region.Type
        #         #         })
        #         #         results['summary']['suspicious_regions'] += 1
        #         
        #         # proc_info['suspicious_regions'] = suspicious_regions
        #         
        #         # 获取加载的DLL
        #         # dlls = dlllist.Dlllist.list_dlls(context, process)
        #         # proc_info['dlls'] = [dll.FullDllName for dll in dlls if hasattr(dll, 'FullDllName')]
        #         
        #         # 检查注入代码
        #         # injected = malfind.Malfind.find_injections(context, process)
        #         # if injected:
        #         #     proc_info['injected_code'] = True
        #         #     results['summary']['potential_injections'] += 1
        #         
        #         results['processes'].append(proc_info)
        # 
        # # 扫描隐藏进程
        # print("扫描隐藏进程...")
        # # hidden_procs = psxview.PsXview.find_hidden_processes(context)
        # # if hidden_procs:
        # #     results['hidden_processes'] = [
        # #         {'pid': proc.UniqueProcessId, 'name': proc.ImageFileName}
        # #         for proc in hidden_procs
        # #     ]
        # 
        # # 扫描可疑驱动程序或模块
        # print("扫描可疑模块...")
        # # modules = modscan.ModScan.list_modules(context)
        # # results['modules'] = [
        # #     {'name': mod.Name, 'base': mod.DllBase, 'size': mod.SizeOfImage}
        # #     for mod in modules[:20]  # 限制数量
        # # ]
    
    except Exception as e:
        print(f"内存取证分析失败: {e}")
        return None
    
    return results

# 使用示例
# result = analyze_database_memory('memory_dump.raw')
# if result:
#     print(json.dumps(result['summary'], indent=2))

# 数据库活动时间线分析脚本
import re
from datetime import datetime
from collections import defaultdict
import json

def create_database_timeline(log_files):
    """
    从多个日志文件创建数据库活动时间线
    
    Args:
        log_files: 日志文件路径列表
    """
    timeline_events = []
    
    # 不同类型日志的时间戳格式和事件模式
    log_patterns = {
        'mysql_error': {
            'timestamp': re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})'),
            'event_type': re.compile(r'\[(\w+)\]'),
            'message': re.compile(r'\] (.+)')
        },
        'mysql_general': {
            'timestamp': re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})'),
            'command_type': re.compile(r'\t(\w+)\t'),
            'user': re.compile(r'\t(\w+)@'),
            'query': re.compile(r'\t\t(.+)$')
        },
        'postgres': {
            'timestamp': re.compile(r'(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}.\d+)'),
            'user': re.compile(r'user=(\w+)'),
            'database': re.compile(r'db=(\w+)'),
            'action': re.compile(r'\] (.+)')
        },
        'mongodb': {
            'timestamp': re.compile(r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}\+\d{4})'),
            'event_type': re.compile(r'(I|E|W)\s+(\w+)'),
            'component': re.compile(r'\[([^\]]+)\]'),
            'message': re.compile(r'\] (.+)')
        }
    }
    
    # 事件严重程度分类
    severity_map = {
        'ERROR': 'high',
        'CRITICAL': 'high',
        'FATAL': 'high',
        'WARNING': 'medium',
        'WARN': 'medium',
        'INFO': 'low',
        'DEBUG': 'low'
    }
    
    # 敏感操作列表
    sensitive_operations = [
        'DROP TABLE', 'DROP DATABASE', 'TRUNCATE TABLE', 'DELETE FROM',
        'ALTER TABLE', 'CREATE USER', 'DROP USER', 'GRANT', 'REVOKE',
        'FLUSH PRIVILEGES', 'SHUTDOWN', 'RESTART', 'CONFIG SET', 'LOAD DATA'
    ]
    
    for log_file in log_files:
        try:
            with open(log_file, 'r', encoding='utf-8', errors='ignore') as f:
                lines = f.readlines()
            
            # 根据文件扩展名或名称确定日志类型
            log_type = 'unknown'
            if 'mysql' in log_file and 'error' in log_file:
                log_type = 'mysql_error'
            elif 'mysql' in log_file and 'general' in log_file:
                log_type = 'mysql_general'
            elif 'postgres' in log_file:
                log_type = 'postgres'
            elif 'mongo' in log_file:
                log_type = 'mongodb'
            
            for line in lines:
                event = {
                    'source_file': log_file,
                    'severity': 'medium',
                    'is_sensitive': False
                }
                
                # 提取通用信息
                if log_type in log_patterns:
                    patterns = log_patterns[log_type]
                    
                    # 提取时间戳
                    if 'timestamp' in patterns:
                        ts_match = patterns['timestamp'].search(line)
                        if ts_match:
                            event['timestamp'] = ts_match.group(1)
                        else:
                            continue  # 跳过没有时间戳的行
                    
                    # 根据日志类型提取特定信息
                    if log_type == 'mysql_error':
                        if 'event_type' in patterns:
                            et_match = patterns['event_type'].search(line)
                            if et_match:
                                event_type = et_match.group(1)
                                event['event_type'] = event_type
                                event['severity'] = severity_map.get(event_type, 'medium')
                        
                        if 'message' in patterns:
                            msg_match = patterns['message'].search(line)
                            if msg_match:
                                message = msg_match.group(1)
                                event['message'] = message
                                
                                # 检查是否为认证失败
                                if 'Access denied' in message:
                                    event['category'] = 'authentication_failure'
                                    event['severity'] = 'high'
                                elif 'Connection' in message:
                                    event['category'] = 'connection'
                    
                    elif log_type == 'mysql_general':
                        if 'command_type' in patterns:
                            ct_match = patterns['command_type'].search(line)
                            if ct_match:
                                command_type = ct_match.group(1)
                                event['command_type'] = command_type
                                event['category'] = command_type.lower()
                        
                        if 'user' in patterns:
                            u_match = patterns['user'].search(line)
                            if u_match:
                                event['user'] = u_match.group(1)
                        
                        if 'query' in patterns:
                            q_match = patterns['query'].search(line)
                            if q_match:
                                query = q_match.group(1)
                                event['query'] = query
                                
                                # 检查敏感操作
                                for op in sensitive_operations:
                                    if op.lower() in query.lower():
                                        event['is_sensitive'] = True
                                        event['severity'] = 'high'
                                        break
                    
                    # 添加更多日志类型的处理...
                
                timeline_events.append(event)
                
        except Exception as e:
            print(f"处理日志文件 {log_file} 失败: {e}")
            continue
    
    # 按时间戳排序事件
    timeline_events.sort(key=lambda x: x.get('timestamp', '9999-99-99 99:99:99'))
    
    # 生成摘要统计
    summary = {
        'total_events': len(timeline_events),
        'by_severity': defaultdict(int),
        'by_category': defaultdict(int),
        'sensitive_operations': 0,
        'time_range': {
            'start': timeline_events[0]['timestamp'] if timeline_events else None,
            'end': timeline_events[-1]['timestamp'] if timeline_events else None
        }
    }
    
    for event in timeline_events:
        summary['by_severity'][event['severity']] += 1
        if 'category' in event:
            summary['by_category'][event['category']] += 1
        if event['is_sensitive']:
            summary['sensitive_operations'] += 1
    
    return {
        'summary': summary,
        'events': timeline_events
    }

# 使用示例
# log_files = ['/var/log/mysql/error.log', '/var/log/mysql/mysql.log', '/var/log/postgresql/postgresql.log']
# timeline = create_database_timeline(log_files)
# if timeline:
#     print(f"总事件数: {timeline['summary']['total_events']}")
#     print(f"敏感操作数: {timeline['summary']['sensitive_operations']}")
#     print(f"时间范围: {timeline['summary']['time_range']['start']} - {timeline['summary']['time_range']['end']}")

# 数据库取证调查报告

## 1. 执行摘要
- 调查目的和范围
- 主要发现
- 结论和建议

## 2. 调查背景
- 事件描述
- 时间线
- 涉及系统和数据库

## 3. 证据收集方法
- 收集的证据类型
- 使用的工具和技术
- 证据保全措施

## 4. 分析过程
- 数据库结构分析
- 日志分析
- 内存分析
- SQL注入攻击分析

## 5. 发现
- 未授权访问证据
- 数据泄露范围
- 攻击方法和技术
- 攻击者特征

## 6. 结论
- 事件性质
- 影响评估
- 根本原因分析

## 7. 建议
- 立即补救措施
- 长期安全改进
- 预防类似事件的策略

## 8. 附录
- 技术细节
- 原始日志样本
- 工具版本信息

-- MySQL启用审计插件示例
INSTALL PLUGIN audit_log SONAME 'audit_log.so';
SET GLOBAL audit_log_policy = 'ALL';
SET GLOBAL audit_log_format = 'JSON';

-- PostgreSQL启用审计日志示例
-- 在postgresql.conf中设置
-- log_statement = 'all'          # none, ddl, mod, all
-- log_min_duration_statement = 0  # 记录所有语句
-- log_line_prefix = '%t [%p]: [%l-1] user=%u,db=%d '  # 包含用户和数据库信息

-- SQL Server启用审计示例
USE [master];
CREATE SERVER AUDIT [DatabaseAudit] TO FILE (FILEPATH = 'D:\AuditLogs\');
ALTER SERVER AUDIT [DatabaseAudit] WITH (STATE = ON);

CREATE SERVER AUDIT SPECIFICATION [ServerAuditSpecification] FOR SERVER AUDIT [DatabaseAudit]
ADD (FAILED_LOGIN_GROUP),
ADD (SUCCESSFUL_LOGIN_GROUP),
ADD (DATABASE_CHANGE_GROUP)
WITH (STATE = ON);