金融级交易合规监控：构建基于MySQL的高效、安全交易监控体系

    -- 创建审计日志表
    CREATE TABLE IF NOT EXISTS audit_log (
    id INT AUTO_INCREMENT PRIMARY KEY,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
    user_name VARCHAR(255) NOT NULL,
    action_type ENUM('SELECT', 'INSERT', 'UPDATE', 'DELETE') NOT NULL,
    affected_table VARCHAR(255),
    affected_rows INT,
    sql_statement TEXT,
    INDEX idx_user_action (user_name, action_type)
    );
    
    
      
      
      
      
      
      
      
      
      
      
      
    

    DELIMITER
$$
    
    -- 定义通用触发器模板
    CREATE TRIGGER trg_after_insert_general
    AFTER INSERT ON your_protected_table FOR EACH ROW
    BEGIN
    INSERT INTO audit_log (user_name, action_type, affected_table, affected_rows, sql_statement)
    VALUES (USER(), 'INSERT', 'your_protected_table', 1, CONCAT('INSERT INTO ', 'your_protected_table', ' VALUES(...)'));
    END$$
    
    CREATE TRIGGER trg_after_update_general
    AFTER UPDATE ON your_protected_table FOR EACH ROW
    BEGIN
    INSERT INTO audit_log (user_name, action_type, affected_table, affected_rows, sql_statement)
    VALUES (USER(), 'UPDATE', 'your_protected_table', 1, CONCAT('UPDATE ', 'your_protected_table', ' SET ... WHERE ...'));
    END$$
    
    CREATE TRIGGER trg_after_delete_general
    AFTER DELETE ON your_protected_table FOR EACH ROW
    BEGIN
    INSERT INTO audit_log (user_name, action_type, affected_table, affected_rows, sql_statement)
    VALUES (USER(), 'DELETE', 'your_protected_table', 1, CONCAT('DELETE FROM ', 'your_protected_table', ' WHERE ...'));
    END$$
    
    DELIMITER ;
    
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    

    import mysql.connector
    from mysql.connector import errorcode
    import logging
    
    # 配置日志格式
    logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(levelname)s - %(message)s',
                    filename='sql_audit.log')
    
    def connect_to_db():
    try:
        cnx = mysql.connector.connect(user='root',
                                     password='your_password',
                                     host='127.0.0.1',
                                     database='test')
        return cnx
    except mysql.connector.Error as err:
        if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
            print("Something is wrong with your user name or password")
        elif err.errno == errorcode.ER_BAD_DB_ERROR:
            print("Database does not exist")
        else:
            print(err)
    
    def audit_sql(sql):
    """
    对传入的SQL语句进行简单审核，判断是否符合预设的安全策略。
    
    参数:
        sql (str): 待审核的SQL语句
        
    返回:
        bool: 如果通过审核则返回True，否则返回False
    """
    # 示例规则：禁止使用DROP TABLE命令
    if "DROP TABLE" in sql.upper():
        logging.warning(f"Detected DROP TABLE command in SQL: {sql}")
        return False
    
    # 其他规则可以根据实际情况添加...
    
    return True
    
    def execute_sql(cnx, sql):
    cursor = cnx.cursor()
    try:
        if audit_sql(sql):
            cursor.execute(sql)
            cnx.commit()
            logging.info(f"Executed SQL successfully: {sql}")
        else:
            logging.error(f"Blocked potentially dangerous SQL: {sql}")
    except mysql.connector.Error as err:
        logging.error(f"Failed to execute SQL: {sql}. Error: {err}")
    finally:
        cursor.close()
    
    if __name__ == '__main__':
    connection = connect_to_db()
    if connection:
        sample_sql = "SELECT * FROM users LIMIT 1;"
        execute_sql(connection, sample_sql)
        connection.close()
    
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    

    -- 创建交易记录表
    CREATE TABLE IF NOT EXISTS transactions (
    transaction_id INT AUTO_INCREMENT PRIMARY KEY,
    sender_account VARCHAR(255),
    receiver_account VARCHAR(255),
    amount DECIMAL(10, 2),
    transaction_date DATETIME,
    INDEX idx_sender_receiver (sender_account, receiver_account)
    );
    
    -- 插入测试数据
    INSERT INTO transactions (sender_account, receiver_account, amount, transaction_date)
    VALUES ('123456789', '987654321', 1000.00, '2024-12-28 14:00:00'),
       ('123456789', '987654321', 500.00, '2024-12-28 15:00:00');
    
    -- 查询频繁转账给同一账户的行为
    SELECT t.sender_account, t.receiver_account, COUNT(*) AS transfer_count
    FROM transactions t
    WHERE t.transaction_date >= DATE_SUB(NOW(), INTERVAL 1 DAY)
    GROUP BY t.sender_account, t.receiver_account
    HAVING transfer_count > 5;
    
    -- 检测大额转账活动
    SELECT t.*
    FROM transactions t
    WHERE t.amount > 10000;
    
    
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
      
    

    -- 使用AES算法对敏感数据进行加密
    -- 假设存在加密函数EncryptData
    UPDATE CustomerInfo
    SET EncryptedData = EncryptData(SensitiveData, 'encryption_key');
    
    
      
      
      
      
    

    -- 更新加密密钥
    ALTER TABLE CustomerInfo
    MODIFY COLUMN EncryptedData VARBINARY(255);
    
    UPDATE CustomerInfo
    SET EncryptedData = EncryptData(EncryptedData, 'new_encryption_key');
    
    
      
      
      
      
      
      
    

    -- 启用审计插件
    INSTALL PLUGIN audit_log SONAME 'audit_log.so';
    
    -- 设置全局参数以开启审计日志
    SET GLOBAL audit_log_policy = 'ALL';
    
    -- 配置审计日志文件路径和其他选项
    [mysqld]
    audit_log_file = /var/log/mysql/audit.log
    audit_log_format = JSON
    audit_log_policy = ALL