详解EMBER数据集中对PE文件提取ByteEntropyHistogram特征

1. 引入

在我们对PE文件提取特征时，经常会在PE特征工程的项目中，看到如下这段代码

class ByteEntropyHistogram(FeatureType):''' 2d byte/entropy histogram based loosely on (Saxe and Berlin, 2015).This roughly approximates the joint probability of byte value and local entropy.See Section 2.1.1 in https://arxiv.org/pdf/1508.03096.pdf for more info.'''name = 'byteentropy'dim = 256def __init__(self, step=1024, window=2048):super(FeatureType, self).__init__()self.window = windowself.step = stepdef _entropy_bin_counts(self, block):# coarse histogram, 16 bytes per binc = np.bincount(block >> 4, minlength=16)  # 16-bin histogramp = c.astype(np.float32) / self.windowwh = np.where(c)[0]H = np.sum(-p[wh] * np.log2(p[wh])) * 2  # * x2 b.c. we reduced information by half: 256 bins (8 bits) to 16 bins (4 bits)Hbin = int(H * 2)  # up to 16 bins (max entropy is 8 bits)if Hbin == 16:  # handle entropy = 8.0 bitsHbin = 15return Hbin, cdef raw_features(self, bytez, lief_binary):output = np.zeros((16, 16), dtype=np.int)a = np.frombuffer(bytez, dtype=np.uint8)if a.shape[0] < self.window:Hbin, c = self._entropy_bin_counts(a)output[Hbin, :] += celse:# strided trick from here: http://www.rigtorp.se/2011/01/01/rolling-statistics-numpy.htmlshape = a.shape[:-1] + (a.shape[-1] - self.window + 1, self.window)strides = a.strides + (a.strides[-1],)blocks = np.lib.stride_tricks.as_strided(a, shape=shape, strides=strides)[::self.step, :]# from the blocks, compute histogramfor block in blocks:Hbin, c = self._entropy_bin_counts(block)output[Hbin, :] += creturn output.flatten().tolist()def process_raw_features(self, raw_obj):counts = np.array(raw_obj, dtype=np.float32)sum = counts.sum()normalized = counts / sumreturn normalized

这段代码是来自知名的项目EMBER（参考1），EMBER对PE文件提取了很多特征，EMBER也是一个数据集和benchmark。

在github中搜索ByteEntropyHistogram(FeatureType)这个关键字符串，或者google上搜索，都能搜到很多项目、博客，包括深度学习和传统机器学习方向，有些博客中也写了这是在计算熵值Entropy。所以，可以看到很多地方对直接引用这段代码来对PE文件（或者任意二进制文件）来做特征工程，笔者实测其效果也确实不错。

但是这段代码理解起来是不太容易的，它的计算过程是怎么样的？它真的只是在计算熵值吗？它代码中block >> 4是在做什么？

下面就一步一步来理解这段做特征工程的代码。

2. 动态运行调试

先让代码跑起来看看，这是工程上调试常用的方法。让代码运行起来，就能知道运行流程是怎样，各个函数的输入输出是怎样，也就能获取到代码不同位置处的中间结果，就能一步一步把问题分析清楚。

要让ByteEntropyHistogram这个class的代码运行起来，就要稍做修改，比如删除它的父类，去掉一些函数中没用到的参数，删除初始化函数中和父类初始化相关的部分。

import numpy as npclass ByteEntropyHistogram():name = 'byteentropy'dim = 256def __init__(self, step=1024, window=2048):self.window = windowself.step = stepdef _entropy_bin_counts(self, block):# coarse histogram, 16 bytes per binc = np.bincount(block >> 4, minlength=16)  # 16-bin histogramp = c.astype(np.float32) / self.windowwh = np.where(c)[0]H = np.sum(-p[wh] * np.log2(p[wh])) * 2  # * x2 b.c. we reduced information by half: 256 bins (8 bits) to 16 bins (4 bits)Hbin = int(H * 2)  # up to 16 bins (max entropy is 8 bits)if Hbin == 16:  # handle entropy = 8.0 bitsHbin = 15return Hbin, cdef raw_features(self, bytez):output = np.zeros((16, 16), dtype=np.int32)a = np.frombuffer(bytez, dtype=np.uint8)if a.shape[0] < self.window:Hbin, c = self._entropy_bin_counts(a)output[Hbin, :] += celse:# strided trick from here: http://www.rigtorp.se/2011/01/01/rolling-statistics-numpy.htmlshape = a.shape[:-1] + (a.shape[-1] - self.window + 1, self.window)strides = a.strides + (a.strides[-1],)blocks = np.lib.stride_tricks.as_strided(a, shape=shape, strides=strides)[::self.step, :]# from the blocks, compute histogramfor block in blocks:Hbin, c = self._entropy_bin_counts(block)output[Hbin, :] += creturn output.flatten().tolist()def process_raw_features(self, raw_obj):counts = np.array(raw_obj, dtype=np.float32)sum = counts.sum()normalized = counts / sumreturn normalizedwith open('cfdbbd60c7dd63db797fb27e1c427077ce1915b8894ef7165d8715304756a7e2', 'rb') as fr:bytez = fr.read()# get binary file bytes arraybe = ByteEntropyHistogram()raw_obj = be.raw_features(bytez)# get raw feature fea_vec = be.process_raw_features(raw_obj)# get final feature vectorprint('out',fea_vec.shape)# (256,) float 1d-vector

最终读入一个二进制文件，依次调用ByteEntropyHistogram提供的函数，就能得到特征向量。

这个运行的过程是：

1. raw_features()中，使用滑动窗口，来对每一个窗口中的二进制数据（byte数组,就是代码中的block变量），调用_entropy_bin_counts(block)，计算Hbin和c的值
2. 计算得到的c值，被根据Hbin指定的位置，累加到一个16*16的二维矩阵中（output[Hbin, :] += c）
3. 这个二维矩阵最终被拉平（reshape）为一个以为向量（return output.flatten().tolist()），就是代码中的raw_obj的值
4. 最终raw_obj被process_raw_features()函数，做了简单的 normalization，就是每个数据值除以总和（normalized = counts / sum）

这个过程中的关键点，是两个函数：_entropy_bin_counts(block)和raw_features()，下面对这两个函数进行更细节的分析

3._entropy_bin_counts(block)的计算过程

1. 函数输入值

这个函数是对一个窗口中的数据进行计算，所以输入的block表示byte数组；byte数组可以理解为一个list结构，其中的每个数据值都是0~255之间的一个整数。比如:

block中就存储了读入的byte数组，举例如下

bytez = b'MZ\x90\x00\x03\x00\x00\x00\x04'

这个例子中，MZ是PE文件的标识，0x是十六进制数据。这个bytez数组可以转换为如下等效的int数组：

bytez = [77,90,144,0,3,0,0,0,4]

byte数组中有9个数据，字母’M’的ASCII码值为十进制整数77，'Z’的ASCII码值为90。上面两个bytez数组的值是一样的。

程序种最终会将bytez转换为np.array格式作为block，即block = np.frombuffer(bytez, dtype=np.uint8)，这个过程举例如下：

import numpy as np
bytez = b'MZ\x90\x00\x03\x00\x00\x00\x04'
block = np.frombuffer(bytez, dtype=np.uint8)
print(block)# array([ 77,  90, 144,   0,   3,   0,   0,   0,   4], dtype=np.uint8)

2. block >> 4

函数中接收到block数组后，会先对block做一个这个操作block >> 4。这是右移操作，会对block数组中每个数据都右移4位。因为block中的数据是8位int数据，所以右移4位相当于去掉低4位，保留高4位数据。也就是每个数据除以16（2的4次方）后的整数。这个过程举例如下：

block = np.array([ 77,  90, 144,   0,   3,   0,   0,   0,   4], dtype=np.uint8)
y = block >> 4
print(y) # array([4, 5, 9, 0, 0, 0, 0, 0, 0], dtype=uint8)

输入函数的block中的数据是8位int数据，所以每个数据值的范围在[0,255]，经过block >> 4的操作后，每个数据取值为[0,15]。这相当于模糊了数据的取值范围，优点是降低了最终特征向量的维度，缺点是降低/模糊了数据精度。

3. bincount

bincount是numpy中用于统计array中每个数据出现次数的函数，我们这里的用法是 c = np.bincount(block >> 4, minlength=16)，这里的minlength说明输出的数据维度至少为16。具体到我们的数据，经过block >> 4的操作后，每个数据取值为[0,15]，设置minlength为16，则最会输出一个维度为16的一维数组，其中表示0~15中每个数据出现的次数。

这个c变量是函数返回值中比较重要的一个参数，它就是对每个窗口中数据做直方图统计的结果。

4. 熵值计算

函数中接下来的这一部分，主要是在计算信息熵的值

p = c.astype(np.float32) / self.window #计算每个数据出现的概率，window默认值为2048
wh = np.where(c)[0]#输出满足条件 (即非0) 元素的坐标
# 下面是计算信息熵H，并放大2倍
H = np.sum(-p[wh] * np.log2(p[wh])) * 2  
# 再对信息熵值放大2倍，最终相当于Hbin是信息熵值的4倍后的整数
Hbin = int(H * 2)  # up to 16 bins (max entropy is 8 bits)
if Hbin == 16:  # handle entropy = 8.0 bitsHbin = 15# 该操作让Hbin返回值在[0,15]范围内

这里在对c变量计算信息熵，并将信息熵乘以4后，取整数，而且改变其边界值（如果Hbin == 16则将其值改变为15，Hbin = 15）。

c变量中的值，是[0,15]每个值的出现次数，所以一共有16个概率值（p），当概率相等时信息熵结果数值最大（参考2中的定理）。所以信息熵H最大值为log2(16)=4。代码中将这个值扩大4被后得到Hbin，则Hbin的取值范围就是[0,1,2,…,16]。代码中最后将边界值改为15，所以最终返回的Hbin的取值范围就是[0,1,2,…,15]。

至此，该函数的两个返回值

• c：block中数据除以4取整后每个数值的出现次数，是1*16的一维数组，表示直方图向量
• Hbin: 是[0,15]中的一个整数，表示对c计算信息熵并处理后的整数值

4. raw_features()

这个函数是对整个二进制文件提取特征的总函数，理解它的关键在于如下几行核心逻辑：

output = np.zeros((16, 16), dtype=np.int32)# 16*16的二维数组
# 用滑动窗口把整个二进制文件切割为多个block的字节数组存储到blocks中
for block in blocks:# 对每个block的byte数组计算 Hbin（信息熵）和c（直方图向量）Hbin, c = self._entropy_bin_counts(block)# 在信息熵Hbin相同的维度上，对c累加output[Hbin, :] += c
return output.flatten().tolist()# 最终将16*16的二维数组转换为1*256的一维数组 作为返回值

1. 滑动窗口及step设置

class的init函数中，有设置step=1024, window=2048。这说明滑动窗口的大小是2048（字节），每次移动（滑动）的距离数是1024字节

2. 如果文件byte数组size小于窗口宽度

如果二进制文件的size比较小，有可能小于2048字节，则直接将这个二进制文件的字节数组拿去计算Hbin（信息熵）和c（直方图向量）

3. 滑动窗口过程

该函数的else部分，首先 用滑动窗口把整个二进制文件切割为多个block的字节数组存储到blocks中。

4. 结果累加

并对每个窗口block的byte数组计算 Hbin（信息熵）和c（直方图向量），在信息熵Hbin相同的维度上，对c累加。最终将1616的二维数组转换为1256的一维数组 作为返回值。

总结

EMBER对PE文件提取了广泛被使用的 ByteEntropyHistogram 特征，这是直接对二进制文件提取特征的一个案例。这个特征的本质上是利用滑动窗口的过程，对各个窗口中二进制数据做模糊后，求取其直方图的信息熵，在不同信息熵值的维度下，对各个窗口中数据直方图向量值累加的结果。

最终，对这个class做个整体的注释：

class ByteEntropyHistogram():name = 'byteentropy'# 该特征名字dim = 256# 该特征最终返回数据的维度，即如下output变量被flatten为一维数组（数据类型为np.int32）def __init__(self, step=1024, window=2048):self.window = window# 滑动窗口的窗口大小，单位是字节，默认是2048字节self.step = step# 滑动窗口的移动宽度，单位是字节，默认是1024字节# 对输入的block数据处理后，计算信息熵及直方图# block为1维的numpy数组，类型是dtype=np.uint8，说明数据值大小为[0,255](即字节数据)def _entropy_bin_counts(self, block):# block >> 4会让block数组中每个数据值都除以16，最终每个数据值变为[0,15]c = np.bincount(block >> 4, minlength=16) # 统计并返回0~15这16个整数的出现次数：直方图向量# 计算 直方图向量c的信息熵值，并将信息熵值放大4倍后标记为Hbinp = c.astype(np.float32) / self.window#计算每个数据出现的概率，window默认值为2048wh = np.where(c)[0]#输出满足条件 (即非0) 元素的坐标# 下面是计算信息熵H，并放大2倍H = np.sum(-p[wh] * np.log2(p[wh])) * 2  # * x2 b.c. we reduced information by half: 256 bins (8 bits) to 16 bins (4 bits)# 再对信息熵值放大2倍，最终相当于Hbin是信息熵值的4倍后的整数Hbin = int(H * 2)  # up to 16 bins (max entropy is 8 bits)if Hbin == 16:  # handle entropy = 8.0 bitsHbin = 15# 该操作让Hbin返回值在[0,15]范围内return Hbin, c# Hbin为处理后得到的信息熵值，是整数，取值范围[0,15]；c是直方图向量def raw_features(self, bytez):# 结果放到16x16的整数二维数组中output = np.zeros((16, 16), dtype=np.int32)# 将读入的字节数组bytez，转换为np.array的一维向量，每个数据值大小为[0,255]a = np.frombuffer(bytez, dtype=np.uint8)# 如果二进制文件字节数小于window，则直接计算Hbin和c后放到output矩阵中if a.shape[0] < self.window:Hbin, c = self._entropy_bin_counts(a)output[Hbin, :] += celse:# 如果二进制文件字节数（size）大于window大小，则滑动窗口# 用滑动窗口把整个二进制文件切割为多个block的字节数组存储到blocks中# strided trick from here: http://www.rigtorp.se/2011/01/01/rolling-statistics-numpy.htmlshape = a.shape[:-1] + (a.shape[-1] - self.window + 1, self.window)strides = a.strides + (a.strides[-1],)blocks = np.lib.stride_tricks.as_strided(a, shape=shape, strides=strides)[::self.step, :]# from the blocks, compute histogramfor block in blocks:# 对每个block的byte数组计算 Hbin（信息熵）和c（直方图向量）Hbin, c = self._entropy_bin_counts(block)# 在信息熵Hbin相同的维度上，对c累加output[Hbin, :] += creturn output.flatten().tolist()# 最终将16*16的二维数组转换为1*256的一维数组 作为返回值# 做简单的 normalization: X = X/SUM(X)def process_raw_features(self, raw_obj):# 输入的raw_obj是一维向量int32类型（即raw_features()函数返回值）counts = np.array(raw_obj, dtype=np.float32)sum = counts.sum()# 对一维数组中的值求和normalized = counts / sum# 每个数据值求除以总和，即做简单的normalizationreturn normalized

参考

1. https://github.com/elastic/ember/blob/master/ember/features.py#L71
2. https://blog.csdn.net/feixi7358/article/details/83861858