<jupyter_start><jupyter_text>**Contents:** 1. File selection 2. `.text` section extraction 3. Preliminary visualization 4. Sampling 5. Sample Visualization 6. Creating the Reference 7. Baseline Testing **Note:**Files with less than 10 kilobytes of machine code were excluded from both reference distribution construction and testing. This is because if a file contains less than 10k of code, most likely there will not be enough 1024-byte blocks to form a cluster that DBSCAN can detect. Small file size is a limitation of the Centrifuge toolkit.<jupyter_code>import os
import sys
import pickle
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
plt.rcParams['figure.figsize'] = [16, 9]
import seaborn as sns
sns.set_style("whitegrid")

sys.path.append("../../pyelftools")             # relative location of pyelftools repo cloned from github
from elftools.elf.elffile import ELFFile
from tqdm import tqdm
from scipy import stats
files = os.listdir("amd64")
len(files)
files = [file for file in files if "code" not in file]
len(files)
np.random.seed(1)
sample_files = np.random.choice(files, size=1000, replace=False)
code_dict = {}
path = "amd64/"
counter = 0

for file in sample_files:
    with open(path + file, "rb") as f:
        try:
            elffile = ELFFile(f)
        except:
            continue
        
        try:
            text_section_header = elffile.get_section_by_name(".text").header
        except:
            continue
        offset = text_section_header.sh_offset
        size = text_section_header.sh_size
        
        if size >= 10000: # minimum size cutoff: 10kb of machine code
            counter += 1
            f.seek(offset)
            code_dict[file] = list(f.read(size))
            if counter == 100: # number of files to select samples from
                break
len(code_dict.keys())
for file, code in code_dict.items():
    print("%s\t%s\t%s" %(file, str(len(code)), code[:10]))
def plot_dists(dictionary):
    for file, code in dictionary.items():
        sns.distplot(code, 
                     bins = 256,
                     kde=False,
                     hist_kws={'histtype':'step', 'cumulative': True, 'linewidth':1, 'alpha':0.1}, 
                     kde_kws={'cumulative': True},
                     norm_hist=True,
                     color="blue")
    plt.show()
plot_dists(code_dict)
def plot_sample(sample):
    sns.distplot(sample,
                 norm_hist=True,     
                 kde=False,
                 hist_kws={'histtype':'step', 'cumulative': True, 'linewidth':0.1, 'alpha':0.1},
                 kde_kws={'cumulative': True},  # want CDF
                 bins=256,                      
                 color="blue")

def create_code_samples(sample_size, n_rows, sample_source, plot=False):
    
    samples_df = pd.DataFrame(index=np.arange(0, n_rows), columns=[i for i in range(sample_size)])
    
    for i in tqdm(np.arange(0, n_rows)):
        sample = np.random.choice(sample_source, size=sample_size, replace=True)
        samples_df.loc[i] = sorted(sample)
        
    samples_mean = samples_df.mean(axis = 0)
    
    if plot is True:
        sns.distplot(samples_mean, norm_hist=True, kde=False, hist_kws={'histtype':'step', 'cumulative': True, 'linewidth':1, 'alpha':1},kde_kws={'cumulative': True},bins=256,color="red")
        plt.show()    
    
    return samples_mean
def create_reference_distribution(code_dictionary, sample_size, n_samples):
    # create data frame to hold mean sample of each file's code
    means_df = pd.DataFrame(index=list(code_dictionary.keys()),
                            columns=[i for i in range(sample_size)])
    
    # for each file's code, store mean of the samples in data frame
    for file, code in code_dictionary.items():
        means_df.loc[file] = create_code_samples(sample_size,
                                                 n_samples,
                                                 code)
    # average all samples means together
    mean = means_df.mean(axis = 0)
    median = means_df.median(axis = 0)
    return mean, median
mean, median = create_reference_distribution(code_dict, 1000, 1000)
sns.distplot(mean, bins = 256, kde=False, hist_kws={'histtype':'step', 'cumulative': True, 'linewidth':1, 'alpha':1}, kde_kws={'cumulative': True}, norm_hist=True, color="red")

plt.title("Mean byte values of the means (red)")
plot_dists(code_dict)


sns.distplot(median, bins = 256, kde=False, hist_kws={'histtype':'step', 'cumulative': True, 'linewidth':1, 'alpha':1}, kde_kws={'cumulative': True}, norm_hist=True, color="red")

plt.title("Median byte values of the means")
plot_dists(code_dict)<jupyter_output><empty_output><jupyter_text>The distribution of byte values of the **median of the means** looks to be more representative than the **mean of the means**.<jupyter_code>from collections import Counter
rounded_medians = [round(i) for i in median]
sorted_counts = sorted(Counter(rounded_medians).items(), key = lambda x: x[1], reverse=True)

counts_df = pd.DataFrame(sorted_counts[:25])
counts_df = counts_df.rename(columns={0:"byte value", 1:"frequency"})

sns.barplot(x="byte value", y="frequency", data=counts_df, order=counts_df["byte value"])
plt.title("Top 25 most common byte values in AMD64 machine code reference distribution (1000 bytes)")
plt.show()
[hex(i) for i in list(dict(sorted_counts).keys())[:25]]
# get pool of test files
test_pool = [file for file in files if file not in sample_files]

# select test files
np.random.seed(1)
test_files = np.random.choice(test_pool, size=3000, replace=False)
# check for overlap between files that the reference distribution was built from and the files used for testing
list(set(code_dict.keys()) & set(test_files))
#path = "amd64/"

def extract_code(path, file_list, min_file_size=10000, n_files=1000):
    
    code_dict = {}
    counter = 0
    
    for file in file_list:
        with open(path + file, "rb") as f:
            try:
                elffile = ELFFile(f)
            except:
                continue
        
            try:
                text_section_header = elffile.get_section_by_name(".text").header
            except:
                continue
            offset = text_section_header.sh_offset
            size = text_section_header.sh_size
        
            if size >= min_file_size: # minimum size cutoff: 10kb of machine code
                counter += 1
                f.seek(offset)
                code_dict[file] = list(f.read(size))
                if counter == n_files: # number of files to select samples from
                    break
            else:
                continue
                
    return code_dict
test_code_dict = extract_code("amd64/", test_files)
for file, code in test_code_dict.items():
    print("%s\t%s\t%s" % (file, str(len(code)), code[:10]))
test_results_df = pd.DataFrame(index=test_code_dict.keys(),
                               columns=["size", "wasserstein distance","energy distance", "mean", "median", "standard deviation"])

for file, code in test_code_dict.items():
    test_results_df.loc[file]["size"] = len(code)
    test_results_df.loc[file]["wasserstein distance"] = stats.wasserstein_distance(median, code)
    test_results_df.loc[file]["energy distance"] = stats.energy_distance(median, code)
    test_results_df.loc[file]["mean"] = np.mean(code)
    test_results_df.loc[file]["median"] = np.median(code)
    test_results_df.loc[file]["standard deviation"] = np.std(code)
    
print(test_results_df.shape)
test_results_df
test_results_df.mean()
test_results_df.median()
test_results_df["size"].sum()
test_results_df["size"].sum() / 1000000
sns.boxplot(x="wasserstein distance", data=test_results_df)
plt.show()
sns.distplot(test_results_df["wasserstein distance"], bins=30)
plt.show()
plt.scatter(test_results_df["size"], test_results_df["wasserstein distance"])
plt.xlabel(".text section size (bytes)")
plt.ylabel("Wasserstein distance from AMD64 reference distribution")
plt.title(".text section size vs. Wasserstein distance from AMD64 reference distribution")
plt.show()
plt.scatter(test_results_df["size"], test_results_df["wasserstein distance"])
plt.xlabel(".text section size (bytes)")
plt.ylabel("Wasserstein distance from AMD64 reference distribution")
plt.title(".text section size vs. Wasserstein distance from AMD64 reference distribution (outlier not shown)")
plt.xlim(0, 2000000)
plt.show()
plt.scatter(test_results_df["wasserstein distance"], test_results_df["energy distance"])
for code in test_code_dict.values():
    plot_sample(code)
    
sns.distplot(median, # reference distribution
             norm_hist=True,     
             kde=False,
             hist_kws={'histtype':'step', 'cumulative': True, 'linewidth':1, 'alpha':1},
             kde_kws={'cumulative': True},
             bins=256,                      
             color="red")

plt.title("1000 .text section CDFs with AMD64 reference CDF in red")
plt.show()
# export reference distribution
with open("AMD64_reference", "wb") as f:
    pickle.dump(median, f)<jupyter_output><empty_output>
StarcoderdataJupyterStructuredCleanDedup