CN111858251A

CN111858251A - 一种基于大数据计算技术的数据安全审计方法及系统

Info

Publication number: CN111858251A
Application number: CN202010713842.7A
Authority: CN
Inventors: 刘迎风; 冯桂安; 梁满; 冯骏; 何怡; 傅行晓; 周亚美
Original assignee: Shanghai Big Data Center
Current assignee: Shanghai Big Data Center
Priority date: 2020-07-22
Filing date: 2020-07-22
Publication date: 2020-10-30
Anticipated expiration: 2040-07-22
Also published as: CN111858251B

Abstract

本发明公开了一种基于大数据计算技术的数据安全审计方法及系统，属于大数据安全领域，包括采集服务器的日志数据，发送至一流处理平台；接收某一个或多个日志数据，进行解析，将解析后的日志数据发送至至少一个数据目标地；对解析后的日志数据进行归类，判断日志数据是实时数据还是非实时数据，将实时数据发送至流处理平台进行存储；将非实时数据发送至数据中心进行存储；分别对日志数据进行分析处理并得到一分析结果；根据分析结果生成相应的告警信息并输出。本发明的有益效果在于：基于flume实现日志采集及存储，引入任务调度和任务监控，丰富了flume日志采集源和输出源；基于flink实现数据安全审计、告警监控管理及处理、安全风险识别。

Description

一种基于大数据计算技术的数据安全审计方法及系统

技术领域

本发明涉及大数据安全领域，尤其涉及一种基于大数据计算技术的数据安全审计方法及系统。

背景技术

近年来，数据安全审计系统已经愈发重要，数据审计系统主要用于监视并记录对数据服务器的各类操作行为，通过对网络数据的分析，实时地、智能地解析对数据服务器的各种操作，并记入审计数据库中以便日后进行查询、分析、过滤，实现对目标数据审计系统的用户操作的监控和审计，尤其对各行业公共数据资源整合和利用时，急需数据安全审计系统来为数据的安全应用、共享交换、开放提供保证。

在现有的数据流转使用过程中，由于缺乏审计防护措施，工作中针对安全事件的查询方式是借助大量人工手动进行条件筛查，在海量日志库中进行检索，审计效率低，导致结果受人为因素干扰较大，因此存在审计不及时、审计力度不足等问题，不能满足数据安全审计要求，数据再流转使用过程中存在安全风险，而传统的大数据计算方法受限于磁盘读写性能以及网络性能的相关约束，在进行实时数据的查询、计算和存储等各方面都不算是高效的，因此针对以上问题，迫切需要设计出一种基于大数据计算技术的数据安全审计方法及系统，以满足实际使用的需要。

发明内容

为了解决以上技术问题，本发明提供了一种基于大数据计算技术的数据安全审计方法及系统。

本发明所解决的技术问题可以采用以下技术方案实现：

本发明提供一种基于大数据计算技术的数据安全审计方法，包括：

步骤S1，采集服务器的日志数据，并将采集的所述日志数据发送至一流处理平台；

步骤S2，接收所述流处理平台中的某一个或多个所述日志数据，并对所述日志数据进行解析，将解析后的所述日志数据输出并发送至至少一个数据目标地；

步骤S3，对解析后的所述日志数据进行归类，判断所述日志数据是实时数据还是非实时数据：

若是实时数据，则发送至所述流处理平台进行存储；

若是非实时数据，则发送至数据中心进行存储；

步骤S4，根据所述步骤S3中的归类，分别对所述日志数据进行分析处理并得到一分析结果，输出所述分析结果；

步骤S5，根据所述分析结果生成相应的告警信息并输出。

优选地，所述步骤S1，于所述日志数据采集过程中，持续管理和监控所述流处理平台和所述日志数据的采集状况、采集量。

优选地，对所述实时数据进行在线分析处理，以及对所述非实时数据进行离线分析处理；

所述在线分析步骤包括：

步骤A1，对所述实时数据进行分类，并存储于所述流处理平台的集群中，所述集群包括全局事件以及至少一个内部事件；

步骤A2，对所述全局事件和至少一个所述内部事件进行实时关联分析；

步骤A3，判断是否为内部事件：

若是，则转入步骤A4；

若否，则生成所述内部事件，并存储于所述集群的其中一个内部事件中；

步骤A4，于判断需要调试监控时，输出第一分析结果；

所述离线分析步骤包括：

步骤B1，预先存储有离线规则，下发所述离线规则至所述流处理平台；

步骤B2，接收所述离线规则，并根据所述离线规则调取所述数据中心的所述日志数据；

步骤B3，对非实时的所述日志数据进行批量分析，输出第二分析结果并发布至所述流处理平台；

步骤B4，接收所述第二分析结果，并发送至文档数据库。

优选地，所述步骤S2中，至少一个解析节点分别对所述日志数据进行解析，解析步骤如下：

步骤21：对所述日志数据进行初始化处理；

步骤22：从所述日志数据中提取有效的日志信息；

步骤23：对所述日志信息进行处理，得到至少一种数据类型的所述日志数据，并分别发送至至少一个数据目标地。

优选地，所述步骤S1中，通过与日志采集系统进行功能配置的方式控制所述日志采集系统采集所述日志数据，所述功能配置包括采集频率、采集时间段以及任务的开启和关闭。

本发明还提供一种基于大数据计算技术的数据安全审计系统，应用于如上述的基于大数据计算技术的数据安全审计方法，包括：

任务调度模块，连接日志采集系统，用于采集服务器的日志数据，并将采集的所述日志数据发送至一流处理平台；

解析模块，连接所述流处理平台，用于接收所述流处理平台中的某一个或多个所述日志数据，并对所述日志数据进行解析，将解析后的所述日志数据输出并发送至至少一个数据目标地；

审计分析模块，连接所述解析模块，用于对解析后的所述日志数据进行归类，判断所述日志数据是实时数据还是非实时数据：

若是实时数据，则发送至所述流处理平台进行存储；

若是非实时数据，则发送至数据中心进行存储；

所述审计分析模块对所述日志数据进行分析处理并得到一分析结果，输出所述分析结果；

报警模块，连接所述审计分析模块，用于根据所述分析结果生成相应的告警信息并输出。

优选地，所述数据安全审计系统还包括监控模块，分别连接所述日志采集系统和所述流处理平台，用于于所述日志数据采集过程中，持续管理和监控所述流处理平台和所述日志采集系统的采集状况、采集量。

优选地，所述审计分析模块包括：

在线分析引擎，连接所述流处理平台，所述在线分析引擎，用于对所述流处理平台的全局事件和多个内部事件进行实时关联分析输出第一分析结果；

离线分析引擎，连接所述数据中心，用于根据下发的离线规则调取所述数据中心中的所述日志数据，对非实时的所述日志数据进行批量分析，输出第二分析结果。

优选地，所述报警模块包括：

第一报警单元，连接所述在线分析引擎，用于根据所述第一分析结果生成相应的第一告警信息并输出；

第二报警单元，连接所述离线分析引擎，用于根据所述第二分析结果生成相应的第二告警信息并输出。

优选地，所述解析模块包括多个解析节点，每个所述解析节点处对应设置一解析器，用于对所述日志数据进行初始化处理，从所述日志数据中提取有效的日志信息，并根据所述日志信息得到至少一种数据类型的所述日志数据，并分别发送至至少一个数据目标地。

本发明的有益效果在于：

本发明的通过基于开源框架日志采集系统(flume)实现日志采集及存储能力，对日志采集系统进行功能迭代，引入任务调度和任务监控的概念，同时丰富了日志采集系统的日志采集源和输出目标地；通过基于开源组件流处理引擎(flink)建模实现数据安全审计能力、告警监控管理及处理能力、安全风险识别能力接入等功能模块的开发，实现数据安全审计系统的建设，于数据采集、传输、存储、处理、交换以及销毁全生命周期中，进行安全审计，为大数据资源平台提供较为周全的安全管理服务，保证数据的流转使用正常；同时系统能够持续检查、发现并预警业务支撑系统中的各种异常和违规行为，及时发现涉密操作事件，准确快速地对涉密事件的操作人员进行定位，留存可用于追责的相关证据。

附图说明

图1为本发明中，一种基于大数据计算技术的数据安全审计方法的流程图；

图2为本发明中，日志数据解析的流程图；

图3为本发明中，在线分析的流程图；

图4为本发明中，离线分析的流程图；

图5为本发明中，任务调度与监控的结构框图；

图6为本发明中，流处理引擎(Flink)的工作原理图；

图7为本发明中，在线策略的流程框图；

图8为本发明中，离线策略的流程框图；

图9为本发明中，数据安全审计系统的结构框图。

具体实施方式

下面将结合本发明实施例中的附图，对本发明实施例中的技术方案进行清楚、完整地描述，显然，所描述的实施例仅仅是本发明一部分实施例，而不是全部的实施例。基于本发明中的实施例，本领域普通技术人员在没有作出创造性劳动的前提下所获得的所有其他实施例，都属于本发明保护的范围。

需要说明的是，在不冲突的情况下，本发明中的实施例及实施例中的特征可以相互组合。

下面结合附图和具体实施例对本发明作进一步说明，但不作为本发明的限定。

本发明提供一种基于大数据计算技术的数据安全审计方法，属于大数据安全领域，如图1和图5所示，数据安全审计方法步骤如下：

步骤S1，采集服务器的日志数据，并将采集的日志数据发送至一流处理平台；

步骤S2，接收流处理平台中的某一个或多个日志数据，并对日志数据进行解析，将解析后的日志数据输出并发送至至少一个数据目标地；

步骤S3，对解析后的日志数据进行归类，判断日志数据是实时数据还是非实时数据：

若是实时数据，则发送至流处理平台进行存储；

若是非实时数据，则发送至数据中心进行存储；

步骤S4，根据步骤S3中的归类，分别对日志数据进行分析处理并得到一分析结果，输出分析结果；

步骤S5，根据分析结果生成相应的告警信息并输出。

具体的，通过在web端与日志采集系统1进行功能配置，控制日志采集系统1批量聚集日志数据，并调取日志采集系统1采集的日志数据，发送至流处理平台中发布。

进一步的，在本实施例中，本发明提供的日志采集系统1是一个分布式的、高可靠的、高可用的采集系统，该日志采集系统1基于开源框架flume，能够将大批量的不同数据源的日志数据收集、聚合、移动到数据中心(Hadoop Distributed File System)中进行存储。

其中，开源流处理平台是可采用Apache Kafka，由Scala和Java编写。Kafka是一种分布式发布订阅消息系统，具有高吞吐量，Kafka可以处理消费者在网站中的所有动作流数据。Kafka通过Hadoop Distributed File System(简称HDFS)的并行加载机制来统一线上和离线的消息处理，也可通过集群来提供实时的消息。

具体的，通过订阅的方式读取流处理平台中的某一个或批量订阅多个日志数据，并对读取的日志数据分别进行解析，解析方法包括：多层次JSON平展转换、无规则文本正则解析、数据库表字段映射；将解析后的日志数据进行转化输出，并发送至至少一个不同的数据目标地，数据目标地包括开源流处理平台(Kafka)、分布式文件系统(HDFS)、基于Lucene的搜索服务器(Elastic Search)、httpfs，开源数据库(hbase)、file以及关系型数据库等，解决了日志采集系统输出源单一的短板，对flume进行功能迭代，引入任务调度的概念，并通过多个解析节点对日志数据进行解析输出至不同的输出目标地，丰富了日志采集系统的日志采集源和输出源。

具体的，日志审计过程中，对日志数据进行归类，可根据判断日志数据是实时数据还是非实时数据进行归类存储，将实时的日志数据发送至流处理平台进行存储，将非实时的日志数据发送至数据中心进行存储；

通过分别调取流处理平台中的实时数据或调取数据中心中的非实时数据对日志数据进行审计分析，经过分析处理后得到分析结果，输出分析结果，并根据分析结果生成相应的告警信息和/或监控调试信息并输出。

日志审计和安全识别警告均采用基于开源组件Apache Flink的计算引擎。Flink是一种通过java实现的流媒体技术计算引擎。Flink的功能非常强大，既可以处理流数据(stream data)也可以批量处理数据(batch data)，同时还可以具有通用计算引擎(Spark)以及Spark streaming的功能，但Flink与通用计算引擎(Spark)不同，Flink本质上只有stream的概念，batch被认为是special stream。

进一步的优选的实施方式，如图6所示，其中Flink主要包括三个组件：JobClient，JobManager和TaskManager。

用户提交一个Flink程序给JobClient，JobClient将程序发送给JobManager，JobManager接收Job程序后将反馈回JobClient。JobManager策划执行接收的job程序，首先，分配job程序所需的资源，主要就是TaskManagers上要执行的slot；在资源分配之后，JobManager提交单独的Task任务到响应的TaskManager上。TaskManager收到一个任务，产生一给线程用来执行这个任务。当状态改变时，例如开始计算或者完成计算，将被发送回JobManager，定时汇报Task的状态。一旦一个job程序被执行完，JobManager返回任务结果给到JobClient。

本发明基于Flume实现日志采集及存储能力，基于Flink引擎建模实现数据安全审计能力、告警监控管理及处理能力、安全风险识别能力接入；在数据的采集、传输、存储、处理、交换、销毁全生命周期流转过程中，进行安全审计，保证数据的流转使用正常，实现数据安全审计系统的建设，为大数据资源平台提供较为周全的安全管理服务。同时系统能够及时发现涉密操作事件，准确定位事件操作人员；检查、发现并预警业务支撑系统中的各种异常和违规行为，提供可用于追责的相关证据。

作为优选的实施方式，该数据安全审计方法，其中于数据安全审计过程中，于日志数据采集过程中，对日志采集系统1、流处理平台(Kafka)持续进行管理和监控，监控日志数据的采集状况、采集量，对日志采集及存储进行实时监控，以便用户能够实时了解日志数据的采集状况、采集量。

作为优选的实施方式，该数据安全审计方法，对实时数据进行在线分析处理，以及对非实时数据进行离线分析处理；

如图7所示，在线分析步骤包括：

步骤A1，对实时数据进行分类，并存储于流处理平台的集群中，集群包括全局事件以及至少一个内部事件；

步骤A2，对全局事件和至少一个内部事件进行实时关联分析；

步骤A3，判断是否为内部事件：

若是，则转入步骤A4；

若否，则生成内部事件，并存储于集群的其中一个内部事件中；

步骤A4，于判断需要调试监控时，输出第一分析结果；

如图8所示，离线分析步骤包括：

预先存储有多项离线规则，下发离线规则并通过流处理平台发布，接收离线规则，并根据离线规则调取数据中心的日志数据，通过配置名单DB和基线DB的参数，对非实时的原始日志数据进行批量处理分析，输出第二分析结果，将第二分析结果发布至Kafka，订阅后发送给文档数据库(ES)。离线分析可以批量分析过去的日志，根据不同的参数配置，产生不同的告警信息。

作为优选的实施方式，该数据安全审计方法，如图2所示，其中步骤S2中，至少一个解析节点分别对日志数据进行解析，解析步骤如下：

步骤21：对日志数据进行初始化处理；

步骤22：从日志数据中提取有效的日志信息；

步骤23：对日志信息进行处理，得到至少一种数据类型的日志数据，并分别发送至至少一个数据目标地。

具体的，在本实施例中，对原始日志数据进行格式化处理，从文本中提取有效的日志信息，从而降低解析的难度。通过多层次JSON平展转换、无规则文本正则解析或数据库表字段映射的方式对提取的日志信息进行解析，同时解析后对得到的日志进行动态补全，补全内容包括根据IP地址补全地区和国家。

作为优选的实施方式，该数据安全审计方法，其中与日志采集系统1进行配置采集频率、采集时间段以及任务的开启和关闭，通过配置参数来控制日志采集系统1在web端服务器中采集日志数据的状况，可通过配置任务开启和关闭的时间、时间段、开启频率来控制日志采集任务的开启和关闭，配置采集频率、采集时间段、采集量来控制在采集过程中采集的状况。

本发明还提供一种基于大数据计算技术的数据安全审计系统，应用于如上述的基于大数据计算技术的数据安全审计方法，如图9所示，包括：

任务调度模块2，连接日志采集系统1，用于采集服务器的日志数据，并将采集的日志数据发送至一流处理平台；

解析模块3，连接流处理平台，用于接收流处理平台中的某一个或多个日志数据，并对日志数据进行解析，将解析后的日志数据输出并发送至至少一个数据目标地；

审计分析模块5，连接解析模块3，用于对解析后的日志数据进行归类，判断日志数据是实时数据还是非实时数据：

若是实时数据，则发送至流处理平台进行存储；

若是非实时数据，则发送至数据中心进行存储；

审计分析模块5对日志数据进行分析处理并得到一分析结果，输出分析结果；

报警模块4，连接审计分析模块5，用于根据分析结果生成相应的告警信息并输出。

具体的，在本实施例中，数据安全审计系统包括任务调度模块2、解析模块3、审计分析模块5和报警模块4；

任务调度模块2，用于通过与基于flume框架的日志采集系统1进行配置采集频率、采集时间段以及任务的开启和关闭的方式来控制日志采集系统1采集服务器的日志数据，并将采集的日志数据发送至流处理平台，对flume进行功能迭代，引入任务调度的概念；

解析模块3，用于订阅流处理平台中的某一个或多个日志数据，并对日志数据进行解析，将解析后的日志数据进行多源输出，同时丰富了flume日志采集源和输出源。

审计分析模块5，用于对解析后的日志数据进行存储，并判断日志数据是实时数据还是非实时数据：

若是实时数据，则发送至流处理平台中进行存储；

若是非实时数据，则发送至数据中心中进行存储；

审计分析模块5调取存储的日志数据进行审计分析后输出分析结果；

报警模块，用于根据分析结果生成相应的告警信息并输出。

作为优选的实施方式，该数据安全审计系统，其中数据安全审计系统还包括监控模块6，分别连接日志采集系统1和流处理平台，用于于日志数据采集过程中，持续管理和监控流处理平台和日志采集系统的采集状况、采集量。

作为优选的实施方式，该数据安全审计系统，其中审计分析模块5包括：

在线分析引擎51，连接流处理平台，基于Flink框架的在线分析引擎51，用于对流处理平台的全局事件和多个内部事件进行实时关联分析输出第一分析结果；

离线分析引擎52，连接数据中心，基于Flink框架的离线分析引擎52，用于根据下发的离线规则调取数据中心中的原始日志数据，对非实时的原始日志数据进行批量分析，输出第二分析结果。

具体的，在线分析引擎51和离线分析引擎52均基于Flink框架，Flink包括预定义窗口分配器，例如滚动窗口、滑动窗口、会话窗口和全局窗口，实时在线分析引擎可创建窗口，对实时的日志流数据进行窗口化分析，从而实时产生告警信号和监控调试信息，将告警信号和监控调试信息给到操作人员，以便操作人员及时进行处理，降低损失。

作为优选的实施方式，该数据安全审计系统，其中报警模块4包括：

第一报警单元41，连接在线分析引擎51，用于根据第一分析结果生成相应的第一告警信息并输出；

第二报警单元42，连接离线分析引擎52，用于根据第二分析结果生成相应的第二告警信息并输出。

作为优选的实施方式，该数据安全审计系统，其中还包括审计报告模块，分别连接审计分析模块5和报警模块，用于根据分析结果和告警信息生成相应的审计报告。

作为优选的实施方式，该数据安全审计系统，其中解析模块3包括多个解析节点，每个解析节点处对应设置一Parser主机作为解析器，Parser主机，用于对日志数据进行初始化处理，从日志数据中提取有效的日志信息，并根据日志信息得到至少一种数据类型的日志数据，并分别发送至至少一个数据目标地。

具体的，对日志信息进行处理得到不同数据类型的数据，并发送至数据目标地，数据目标地包括Elasticsearch、HBse/HDFS、Druid、CVS等多种类型的输出源。

本发明的有益效果在于：

以上所述仅为本发明较佳的实施例，并非因此限制本发明的实施方式及保护范围，对于本领域技术人员而言，应当能够意识到凡运用本发明说明书及图示内容所作出的等同替换和显而易见的变化所得到的方案，均应当包含在本发明的保护范围内。

Claims

1.一种基于大数据计算技术的数据安全审计方法，其特征在于，包括：

若是实时数据，则发送至所述流处理平台进行存储；

若是非实时数据，则发送至数据中心进行存储；

步骤S5，根据所述分析结果生成相应的告警信息并输出。

2.根据权利要求1所述的基于大数据计算技术的数据安全审计方法，其特征在于，所述步骤S1，于所述日志数据采集过程中，持续管理和监控所述流处理平台和所述日志数据的采集状况、采集量。

3.根据权利要求1所述的基于大数据计算技术的数据安全审计方法，其特征在于，对所述实时数据进行在线分析处理，以及对所述非实时数据进行离线分析处理；

所述在线分析步骤包括：

步骤A3，判断是否为内部事件：

若是，则转入步骤A4；

步骤A4，于判断需要调试监控时，输出第一分析结果；

所述离线分析步骤包括：

步骤B4，接收所述第二分析结果，并发送至文档数据库。

4.根据权利要求1所述的基于大数据计算技术的数据安全审计方法，其特征在于，所述步骤S2中，至少一个解析节点分别对所述日志数据进行解析，解析步骤如下：

步骤21：对所述日志数据进行初始化处理；

步骤22：从所述日志数据中提取有效的日志信息；

5.根据权利要求1所述的基于大数据计算技术的数据安全审计方法，其特征在于，所述步骤S1中，通过与日志采集系统进行功能配置的方式控制所述日志采集系统采集所述日志数据，所述功能配置包括采集频率、采集时间段以及任务的开启和关闭。

6.一种基于大数据计算技术的数据安全审计系统，其特征在于，应用于如权利要求1-5任意一项所述的基于大数据计算技术的数据安全审计方法，包括：

若是实时数据，则发送至所述流处理平台进行存储；

若是非实时数据，则发送至数据中心进行存储；

7.根据权利要求6所述的基于大数据计算技术的数据安全审计系统，其特征在于，所述数据安全审计系统还包括监控模块，分别连接所述日志采集系统和所述流处理平台，用于于所述日志数据采集过程中，持续管理和监控所述流处理平台和所述日志采集系统的采集状况、采集量。

8.根据权利要求6所述的基于大数据计算技术的数据安全审计系统，其特征在于，所述审计分析模块包括：

9.根据权利要求8所述的基于大数据计算技术的数据安全审计系统，其特征在于，所述报警模块包括：

10.根据权利要求6所述的基于大数据计算技术的数据安全审计系统，其特征在于，所述解析模块包括多个解析节点，每个所述解析节点处对应设置一解析器，用于对所述日志数据进行初始化处理，从所述日志数据中提取有效的日志信息，并根据所述日志信息得到至少一种数据类型的所述日志数据，并分别发送至至少一个数据目标地。