寻找国际石油公司不知道表名或字段标签

分享这篇文章

有一个漏洞!你是一个信息安全事件响应器得到调查。你出现,并开始要求人们对网络流量日志和遥测数据。人们开始与你分享tb的数据,指出你在云存储不同的位置。你编译的数以百计的IP地址和域名列表作为妥协(国际石油公司)的指标。首先,你想要检查是否这些国际石油公司出现在日志和遥测数据给你。你快速的外观和意识到日志数据从所有不同的系统,安全设备和云提供商的组织使用很多不同的模式和格式。你如何分析这些数据?你不能完全的数据下载到笔记本电脑和执行grep。你不能完全把这变成一个安全信息和事件管理(SIEM)系统——它将繁琐的设置,过于昂贵,可能太慢。 How would you query for IOCs over the different schemas? You imagine spending days figuring out the different schemas.

现在想象一下,如果你有砖Lakehousebob体育客户端下载平台。所有的日志和遥测数据从组织的系统,出口安全传感器和云提供商可以直接到砖Lakehouse摄取三角洲表(也存储在廉价的云存储)。三角洲表也促进高绩效分析和AI /毫升。因为砖可以在多个云,不需要将数据合并到一个云数据驻留在多个云。您可以过滤砖在多个云中的数据并获取结果通过三角洲铺文件共享协议。因此你只付出口费用查询结果没有数据摄入!想象的各种查询、分析和AI /毫升模型等网络安全平台可以运行在你深入地了解一个事件反应(IR)的调查。bob体育客户端下载想象是多么容易寻找那些国际石油公司。

在这个博客中,我们将

在网络安全操作,解释国际奥委会匹配问题
向您展示如何执行国际奥委会匹配日志和遥测数据存储在数据砖Lakehouse平台不知道表名或字段名,bob体育客户端下载
向您展示如何扩展连续或增量监测特定查询,和
向您展示如何创建概要结构增加时间覆盖率和加快历史奥委会搜索。

年底这个博客,你将能够把相关的笔记本和样本数据,尝试在自己的砖工作区。

国际奥委会匹配为什么重要?

检测

匹配的原子国际石油公司的一个基本构建块检测使用的规则或模型检测系统,如端点检测和响应(EDR)系统和入侵检测系统(IDS)。原子奥委会可以一个IP地址,一个完全限定域名(FQDN),文件哈希(MD5、SHA1, SHA256等等。),TLS指纹,注册表键或文件名与潜在的入侵或恶意活动。检测系统通常使用(原子)国际奥委会匹配与其他谓词检测网络威胁和生成一个安全事件预警与相对较高的信心。

例如,考虑国际奥委会的FQDN恶意指挥控制(C2)服务器。检测规则需要找到一个域名系统(DNS)请求相匹配的FQDN日志,确认请求成功,确认发送请求试图连接到主机的IP地址与之前的FQDN生成警报。从检测系统警报通常聚合,优先和修复安全操作中心(SOC)的一个组织。大多数警报是假阳性,但当真正发现积极的提醒,一个事件响应触发工作流。

事件反应(IR)

当违约,如SolarWinds黑客,怀疑,第一个任务事件反应者要做的就是构建一个列表相关的国际石油公司和扫描所有日志和遥测数据的国际石油公司。扫描的结果,或国际奥委会匹配,是国际奥委会的列表(有时也称为线索或低忠诚警报)。这些国际奥委会击中受到事件反应者和一个更深的法医调查进行受影响的系统。法医调查的目的是建立时间表和违约的范围。

威胁狩猎

威胁狩猎是一个积极主动的寻找逃避现有的检测系统的安全威胁。威胁狩猎通常始于一个奥委会搜索所有可用的日志和遥测数据。国际石油公司用于狩猎的列表通常是策划的具体组织的威胁在过去,发现公共新闻/博客,威胁情报来源。我们可以进一步威胁情报来源分解成付费订阅(VirusTotal等等),开源(Facebook ThreatExchange)和执法(联邦调查局,国土安全部,网络司令部)。bob下载地址

在红外和威胁狩猎用例,这一事件响应者或威胁猎人(今后“分析师”)将执行国际奥委会匹配获得国际奥委会的列表。这些点击分组的设备(主机、服务器、笔记本电脑、等)与国际奥委会相关事件的发生。对于每一个组,分析师将和查询额外数据之前事件的时间戳。这些数据包括流程执行、文件下载、用户账户和有时富含威胁情报(如。检查文件对VirusTotal散列)。如果触发事件被认为是恶意的,修复分离或隔离设备可能会采取行动。注意,国际石油公司威胁情报订阅的限制,他们是有限的(如“公共”指标。公共IP地址),一些恶意演员劫持受害者的基础设施,因此操作私有IP地址,难以检测。在任何情况下,调查过程是由国际奥委会匹配操作,因此它在网络安全操作的重要性。

为什么是国际奥委会匹配困难的?

考虑国际奥委会为红外用例匹配。在最好的情况下的所有数据坐在安全信息和事件管理(SIEM)系统,因此可以很容易地查询;然而,SIEM通常有一个相对较短的停留时间(通常少于90天热存储)由于成本和一些威胁可能在组织的运营环境,只要一年。的Solarwinds或阳光违反2020年是一个很好的例子:最早的活动可以追溯到2020年2月即使违反2020年11月才发现。此外,即使你可以保持一年的数据SIEM,大多数遗留SIEM无法查询了这么多数据交互速度和许多分析师最终“组块”查询分成多个查询,短时间内(如一个星期)。在其他情况下,数据可能坐跨多个孤立的数据存储与不再保留,但需要巨大的努力来执行国际奥委会匹配不同的数据存储。

即使一个外部网络安全厂商从事红外,供应商红外团队经常会想把客户数据回中央数据存储与国际奥委会执行匹配和分析的功能。困难之一将变化的数据模式数据回落,需要一些努力来处理模式变化使用schema-on-write或schema-on-read技术。另一个困难的报道将搜索的时间或数据保留。给定一个红外的紧迫性,只有最近的时间窗口的数据被审查,因为它往往是困难或不可行的收购涉及的数据量很长的停留时间。注意,数据采集的红外可以非常复杂:客户可能没有一个安全、long-retention,和防篡改记录设施;源系统上的日志可能滚由于有限的存储;演员的威胁可能会篡改日志来掩盖自己的踪迹。

威胁狩猎用例面临类似的数据工程挑战,但有一个额外的困难,国际石油公司匹配的列表可以在数百或数千人。在执行单奥委会匹配可能仍然是可接受的红外使用情况的国际石油公司数以千万计,这种方法不可能威胁打猎。国际奥委会匹配等威胁狩猎需要治疗一个数据库连接操作和利用各种高性能加入算法开发的数据库社区。

国际奥委会匹配使用砖Lakehouse平台bob体育客户端下载

现在回到事件反应(IR)场景中,我们开始这个博客,你会怎么做国际奥委会在你所有的日志和遥测数据匹配?

如果你有一个SIEM,您可以运行一个查询匹配的一个或一组国际石油公司,但我们已经提到过的局限性。也许你构建自己的解决方案通过发送和储存你所有的日志和遥测等廉价的云存储的AWS S3(或湖Azure数据存储(ADLS)或谷歌云存储)。你仍然需要一个查询引擎像转眼间/ Trino AWS雅典娜,AWS红移,Azure数据浏览,或谷歌大查询来查询数据。然后你还需要一个用户界面(UI)支持协作所需的大多数红外和威胁狩猎用例。将这些碎片粘合起来成有功能的解决方案仍需要重大工程工作。

好消息是,砖Lakehouse平台是一个统一的数据平台:bob体育客户端下载

让你摄取所有的日志和遥测的原始形式在云存储到三角洲表(云存储)使用三角洲湖框架(三角洲湖框架使用开放格式,支持快速、可靠和可伸缩的查询处理);
支持分析和AI /毫升的工作负载在同一三角洲表——不需要复制或ETL;
支持协作工作流通过笔记本UI以及通过一组丰富的api和与其他工具和系统集成;
同时支持流媒体和批处理相同的运行时环境。

砖的红外场景

让我们深入的红外场景假设所有数据已经摄取到砖Lakehouse平台。bob体育客户端下载你想要检查是否发生任何的国际石油公司你编译的日志和遥测数据在过去的12个月。现在你面临以下问题:

数据库、表和列包含相关数据,应该检查国际石油公司?
我们如何提取指标(IP地址,fqdn)相关列?
我们如何表达国际奥委会匹配查询作为一个有效的连接查询?

第一个问题本质上是一个模式发现任务。你可以自由使用第三方模式发现的工具,但它也直接查询砖metastore与数据库相关联的元数据,表和列。下面的代码片段,并将结果进行进一步的过滤到一个临时视图。

db_list = [x [0]为x在spark.sql (“显示数据库”).collect ())excluded_tables = [“test01.ioc”,“test01.iochits”]# =完整列表模式、表、列类型full_list = []为我在db_list:试一试:tb_df = spark.sql (f "显示表{我}”)除了异常作为x:打印(f”无法显示表{我}…跳过”)继续为(db、table_name is_temp)在tb_df.collect ():full_table_name = db +“。”+ table_name如果is_temp或full_table_name在excluded_tables:继续试一试:cols_df = spark.sql (f”描述{full_table_name}”)除了异常作为x:#最可能的例外是拒绝访问,因为桌子是不可见的用户帐户打印(f”无法形容{full_table_name}…跳过”)继续为(col_type col_name评论)在cols_df.collect ():如果不col_type或col_name [:5)= =“部分”:继续full_list。追加([db, table_name、col_name col_type])火花。createDataFrame (full_list模式= [“数据库”,“表”,“columnName”,“colType”]).createOrReplaceTempView (“allColumns”)显示器(spark.sql (“从allColumns SELECT *”))

然后写一个SQL查询临时视图上找到相关的列在where子句中使用简单的启发式方法。

metadata_sql_str =”“”选择数据库、表collect_set (columnName)过滤器(columnName我喜欢' %源自% '或columnName我喜欢' % resp % '或columnName我喜欢“% dest %”或columnName我喜欢' % dst % '或columnName我喜欢' % src % '或columnName我喜欢' % ipaddr % '或columnName(“查询”、“主机”,“推荐人”)作为ipv4_col_list),collect_set (columnName)过滤器(在columnName(“查询”、“推荐人”)作为fqdn_col_list从allColumns在colType = '字符串'通过数据库,表名”“”
              显示器(spark.sql (metadata_sql_str))

对于第二个问题,您可以使用SQL内置函数regexp_extract_all ()从列中提取指标使用正则表达式。例如,下面的SQL查询,

选择regexp_extract_all (“这个https://d.test.com粘贴到浏览器”,”((? !)[A-Za-z0-9 -] {63} (? !) \ \) + [A-Za-z] {2,6} ',0)作为提取联盟所有选择regexp_extract_all (萍为1.2.3.4然后ssh 10.0.0.1和类型表示的,”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0)作为提取

将返回结果:

提取

(“d.test.com”)

["为1.2.3.4”、“10.0.0.1”表示)

对于第三个问题,让我们考虑情况下,采取单表域名系统(DNS)表为例。从网络中提取的DNS表包含DNS请求数据包捕获文件。DNS表,您将运行下面的查询执行国际奥委会对指标提取匹配相关的列。

选择/ * +广播(ioc) * /现在()作为detection_ts,“test01.dns”作为src, aug.rawioc.ioc_value作为matched_ioc,ioc.ioc_type从(选择exp.raw extracted_obs从(选择to_json (struct (d。*))作为生,concat (regexp_extract_all (d.query”(\ \ d + \ \ \ d + \。\ \ d + \ \ \ d +)。”,0),regexp_extract_all (d.id_orig_h”(\ \ d + \ \ \ d + \。\ \ d + \ \ \ d +)。”,0),regexp_extract_all (d.id_resp_h”(\ \ d + \ \ \ d + \。\ \ d + \ \ \ d +)。”,0),regexp_extract_all (d.query”((? !)[A-Za-z0-9 -] {63} (? !) \ \) + [A-Za-z] {2,6} ',0))作为extracted_obslist从test01.dns作为d)作为exp侧视爆炸(exp.extracted_obslist)作为extracted_obs)作为8月内连接test01.ioc作为国际奥委会在aug.extracted_obs = ioc.ioc_value

注意,可选的优化器指令“广播(ioc)”。告诉砖查询优化器选择的查询执行计划,广播较小的“国际奥委会”表包含国际石油公司所有职工的列表节点处理连接操作符。还请注意,正则表达式提供了简化的例子(考虑使用的正则表达式msticpy生产)。现在,你只需要使用上面的查询作为模板并生成相应的SQL查询所有表可能包含有关列指标。您可以查看的Python代码中提供的笔记本。

注意,需要的时间将取决于运行这些奥委会匹配查询的数据量和可用的计算资源的笔记本:考虑到相同体积的数据,计算资源越多,越并行处理,更快的处理时间。

砖的威胁的狩猎场景

狩猎场景的威胁呢?狩猎,威胁你的安全团队通常维护一个策划的国际石油公司并执行列表周期国际奥委会的匹配对国际石油公司的列表。

几句关于维护一个策划的国际石油公司列表。根据您的组织的网络安全实践的成熟,国际石油公司的策划列表可能只是恶意IP地址的集合,fqdn,散列等获得组织的威胁情报订阅所策划的猎人的威胁与您的组织或行业之间的关系。在一些组织中,猎人的威胁可能发挥更积极的作用为包容,寻找国际石油公司测试的发病率统计国际石油公司,以确保假阳性利率管理,和即将到期的国际石油公司当他们不再相关。

当执行国际奥委会匹配使用策划国际石油公司的列表,您可以选择跳过模式发现步骤如果数据库、表和列是众所周知的和缓慢的变化。为了提高效率,您也不希望运行国际奥委会匹配操作每次都从头开始,因为大多数的数据在上一次运行检查。砖三角洲生活表(DLT)提供一个非常方便的方式将国际奥委会匹配查询转化为管道运行逐步只在底层表的更新。

创建流表iochits生活作为现在选择()作为detection_ts,“test01.dns”作为src, aug.raw ioc.ioc_value作为matched_ioc, ioc.ioc_type从(选择exp.raw extracted_obs从(选择to_json (struct (d。*))作为生,concat (regexp_extract_all (d.query”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),regexp_extract_all (d.id_orig_h”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),regexp_extract_all (d.id_resp_h”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),regexp_extract_all (d.query”((? !)[A-Za-z0-9 -] {63} (? !) \ \) + [A-Za-z] {2,6} ',0))作为extracted_obslist从流(test01.dns)作为d)作为exp侧视爆炸(exp.extracted_obslist)作为extracted_obs)作为8月内连接test01.ioc作为国际奥委会在aug.extracted_obs = ioc.ioc_valueUNION ALL现在选择()作为detection_ts,“test01.http”作为src,aug.raw,ioc.ioc_value作为matched_ioc,ioc.ioc_type从(选择exp.raw extracted_obs从(选择to_json (struct (d。*))作为生,concat (regexp_extract_all (d.origin”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),regexp_extract_all (d.referrer”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),regexp_extract_all (d.id_orig_h”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),regexp_extract_all (d.host”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),regexp_extract_all (d.id_resp_h”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),regexp_extract_all (d.referrer”((? !)[A-Za-z0-9 -] {63} (? !) \ \) + [A-Za-z] {2,6} ',0))作为extracted_obslist从流(test01.http)作为d)作为exp侧视爆炸(exp.extracted_obslist)作为extracted_obs)作为8月内连接test01.ioc作为国际奥委会在aug.extracted_obs = ioc.ioc_value;

事实上,你已经完全控制增量处理的程度:你可以在计划执行增量处理连续或间隔。

当你开始执行一个狩猎查询定期或连续,这不是在检测领域而不是打猎?确实这是一个细线。狩猎的查询往往更少和更低的忠诚而检测规则在成千误判率高是不接受的。由此产生的加工特点也不同。缩放的问题检测处理成千上万的检测规则在大量的木材和遥测数据非常适合并行化自检测规则几乎都是独立的。流和micro-batching方法可用于增量执行检测处理新日志和遥测数据的到来。狩猎往往是一些查询,但每个查询试图覆盖更多的数据,因此需要更多的资源来处理。

狩猎和调查,更多特别的在自然界中?有什么方法可以让这些查询在互动工作速度?这是一种常见的请求从威胁猎人等大量的网络日志数据尤其是DNS数据。

一个优雅的和有效的方法是保持highly-aggregated总结结构作为物化视图。例如,DNS数据,总结结构只会持有DNS记录聚合使用桶为每一个独特的价值(日期、sourceTable indicator_value、sourceIP destinationIP)。威胁猎人首先查询这个总结的结构,然后使用字段汇总记录源表查询详情。砖DLT的特性又提供了一种方便的方法来为维护这些概要创建管道结构和下面显示了SQL的DNS汇总表。

创建流表ioc_summary_dns生活作为选择ts_day、obs_value src_data、src_ip dst_ip, count (*)作为问从(选择“test01.dns”作为src_data,extracted_obs作为obs_value,date_trunc (“天”,时间戳(exp.ts))作为ts_day,exp.id_orig_h作为src_ip,exp.id_resp_h作为dst_ip从(选择d。*,concat (regexp_extract_all (d.query”(\ \ d + \ \ \ \ d + \ \。\ \ d + \ \ \ \ d +)。”,0),数组(d.id_orig_h),数组(d.id_resp_h),regexp_extract_all (d.query”((? !)[A-Za-z0-9 -] {63} (? !) \ \) + [A-Za-z] {2,6} ',0))作为extracted_obslist从流(test01.dns)作为d)作为exp侧视爆炸(exp.extracted_obslist)作为extracted_obs)作为8月GROUP BY ts_day、obs_value src_data、src_ip dst_ip;

你会为每一个源表创建单独的DLT管道,然后创建一个视图联盟所有汇总表成一个单一的视图查询了以下SQL。

创建视图test01.ioc_summary_all作为选择*从test01.ioc_summary_dns联盟所有选择*从test01.ioc_summary_http

汇总表帮助如何?

考虑到DNS表,回想一下,总结结构只会持有DNS记录聚合为每一个独特的价值(日期、sourceTable indicator_value、sourceIP destinationIP)。相同的源目的地IP组之间,可能会有成千上万的DNS请求一天同样的FQDN。聚合汇总表,只有一条记录代表潜在的,成千上万的DNS请求相同的FQDN之间相同的源目的地地址。因此汇总表是原始表的有损压缩的压缩比至少10倍。查询规模小得多的汇总表因此更多性能和互动。此外的压缩性质总结结构意味着它可以涵盖更长的停留时间与原始数据相比。你失去与聚合时间分辨率,但至少它给了你在调查过程中急需的可见性。想想如何总结结构能告诉你你是否在的影响的阳光威胁即使威胁被发现9个月后第一个可疑活动。