84 KiB
84 KiB
In [ ]:
import glob
import os
import zlib
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
%matplotlib inline
pd.set_option("display.width", 200)
pd.set_option("display.max_colwidth", 60)
RANDOM_STATE = 42 # "hạt giống" ngẫu nhiên: lần nào chạy cũng ra cùng kết quả
rng = np.random.default_rng(RANDOM_STATE) # máy sinh số ngẫu nhiên có hạt giống cố định
def entropy(du_lieu):
# Entropy Shannon của một dãy byte, đơn vị bit/byte (0 đến 8)
if len(du_lieu) == 0:
return 0.0
so_lan = np.bincount(np.frombuffer(du_lieu, dtype=np.uint8), minlength=256) # 256 ô: giá trị byte k xuất hiện bao nhiêu lần
p = so_lan[so_lan > 0] / so_lan.sum() # tỉ lệ của mỗi giá trị byte CÓ xuất hiện (bỏ ô 0 vì log2(0) không xác định)
return max(0.0, float(-(p * np.log2(p)).sum())) # H = -Σ p·log2(p); max(0, ...) để không in "-0.000"
with open("/usr/bin/ls", "rb") as f: # "rb" = đọc dạng byte (không phải chữ)
ls_bytes = f.read()
with open("/etc/services", "rb") as f:
van_ban = f.read()
khoa = rng.integers(0, 256, len(ls_bytes), dtype=np.uint8) # một "khoá" ngẫu nhiên dài bằng file ls
mau = {
"1.000 chữ A giống nhau": b"A" * 1000,
"văn bản (/etc/services)": van_ban,
"chương trình /usr/bin/ls": ls_bytes,
"ls bị NÉN (zlib)": zlib.compress(ls_bytes, 9),
"ls bị 'MÃ HOÁ' (XOR khoá ngẫu nhiên)": (np.frombuffer(ls_bytes, dtype=np.uint8) ^ khoa).tobytes(), # ^ = phép XOR từng byte
"byte ngẫu nhiên": rng.integers(0, 256, len(ls_bytes), dtype=np.uint8).tobytes(),
}
# 📦 mau = dict: tên mẫu -> dãy byte
bang_mau = pd.DataFrame([{"mẫu": ten, "kích thước (byte)": len(b), "số giá trị byte khác nhau": len(set(b)),
"entropy": round(entropy(b), 3)} for ten, b in mau.items()])
print(bang_mau.to_string(index=False))
print("\n10 byte đầu của ls :", ls_bytes[:10])
print("10 byte đầu của ls bị nén :", mau["ls bị NÉN (zlib)"][:10])In [ ]:
mal = pd.read_csv("/opt/csai/data/malware_meta/malware_meta.csv")
print("mal:", mal.shape, "(hàng = mẫu, cột = thông tin)")
print(mal.head(3).to_string(index=False))
print("\nLoại file:", mal.file_type.value_counts().to_dict())
mal_elf = mal[mal.file_type == "ELF"].copy() # chỉ giữ chương trình Linux (ELF) để so cùng loại với phía sạch
print("Kiến trúc CPU của mẫu ELF:", mal_elf.elf_arch.value_counts().to_dict())
print("Mẫu ELF có dấu UPX:", int(mal_elf.has_upx.sum()), "/", len(mal_elf))
# Phía SẠCH: mọi chương trình ELF trên chính máy lớp (đi kèm hệ điều hành, coi là sạch)
hang = []
for duong_dan in sorted(glob.glob("/usr/bin/*") + glob.glob("/usr/sbin/*")):
if os.path.islink(duong_dan) or not os.path.isfile(duong_dan):
continue # bỏ lối tắt và thư mục
try:
with open(duong_dan, "rb") as f:
b = f.read()
except OSError:
continue # file không có quyền đọc -> bỏ qua
if b[:4] == b"\x7fELF":
hang.append({"ten": duong_dan, "size_bytes": len(b), "entropy": round(entropy(b), 4), "has_upx": int(b"UPX!" in b)})
lanh = pd.DataFrame(hang)
# 📦 lanh = bảng chương trình sạch: ten, size_bytes, entropy, has_upx
print(f"\nlanh: {len(lanh)} chương trình ELF sạch, có dấu UPX: {int(lanh.has_upx.sum())}")
print(lanh.sort_values("entropy", ascending=False).head(3).to_string(index=False))
nhom = {"sạch (máy lớp)": lanh.entropy, "mã độc KHÔNG UPX": mal_elf[mal_elf.has_upx == 0].entropy,
"mã độc CÓ UPX": mal_elf[mal_elf.has_upx == 1].entropy}
print("\n" + pd.DataFrame({ten: s.describe()[["count", "min", "50%", "max"]] for ten, s in nhom.items()}).round(3).to_string())In [ ]:
fig, ax = plt.subplots(figsize=(11, 4))
khoang = np.arange(1.0, 8.05, 0.1) # các "thùng" rộng 0,1
for (ten, s), mau_ve in zip(nhom.items(), ["tab:blue", "tab:orange", "tab:red"]):
ax.hist(s, bins=khoang, alpha=0.6, label=f"{ten} ({len(s)})", color=mau_ve)
for x in (7.0, 7.5):
ax.axvline(x, ls="--", color="gray")
ax.set_xlabel("entropy (bit/byte)")
ax.set_ylabel("số file")
ax.set_title("Entropy: chương trình sạch trên máy lớp vs 205 mã độc ELF thật")
ax.legend()
plt.show()In [ ]:
# ✏️ SỬA: 2 ngưỡng của luật chọn đường
NGUONG_NEN, NGUONG_MA_HOA = 7.0, 7.5
def chon_duong(e, nguong_nen=7.0, nguong_ma_hoa=7.5):
# Luật G3A của quy trình V3.2: entropy -> đường phân tích
if e < nguong_nen:
return "1 THƯỜNG -> Ghidra"
if e <= nguong_ma_hoa:
return "2 NÉN -> thử giải nén UPX"
return "3 MÃ HOÁ -> bỏ Ghidra, chạy thử"
mal_elf["duong"] = mal_elf.entropy.apply(lambda e: chon_duong(e, NGUONG_NEN, NGUONG_MA_HOA)) # lambda = hàm 1 dòng không tên
lanh["duong"] = lanh.entropy.apply(lambda e: chon_duong(e, NGUONG_NEN, NGUONG_MA_HOA))
print("Mã độc ELF (205):")
print(pd.crosstab(mal_elf.duong, mal_elf.has_upx.map({0: "không UPX", 1: "có UPX"})).to_string())
print("\nChương trình sạch trên máy lớp:")
print(lanh.duong.value_counts().sort_index().to_string())
print("\nFile sạch KHÔNG đi đường THƯỜNG:")
print(lanh[~lanh.duong.str.startswith("1")][["ten", "entropy"]].to_string(index=False) or "(không có)")In [ ]:
# ✏️ SỬA: tỉ lệ phần cuối file bị nén (0 = không nén, 1 = nén cả file)
TI_LE_NEN = 0.8
cat = int(len(ls_bytes) * (1 - TI_LE_NEN)) # vị trí cắt: trước = để nguyên, sau = nén
nen_mot_phan = ls_bytes[:cat] + zlib.compress(ls_bytes[cat:], 9)
# 📦 nen_mot_phan = phần đầu ls để nguyên + phần cuối đã nén
print(f"File nén {TI_LE_NEN:.0%} phần cuối: {len(nen_mot_phan):,} byte, entropy CẢ FILE = {entropy(nen_mot_phan):.3f}"
f" -> {chon_duong(entropy(nen_mot_phan))}")
def ho_so(du_lieu, buoc=1024):
# entropy của từng đoạn 'buoc' byte liên tiếp
return [entropy(du_lieu[i:i + buoc]) for i in range(0, len(du_lieu) - buoc + 1, buoc)]
fig, ax = plt.subplots(figsize=(11, 3.5))
for ten, b in [("ls (thường)", ls_bytes), (f"ls nén {TI_LE_NEN:.0%} phần cuối", nen_mot_phan), ("ls nén cả file", mau["ls bị NÉN (zlib)"])]:
ax.plot(np.arange(len(ho_so(b))), ho_so(b), label=f"{ten}: cả file {entropy(b):.2f}")
ax.axhline(7.5, ls="--", color="gray")
ax.set_xlabel("vị trí trong file (đoạn 1 KB thứ ...)")
ax.set_ylabel("entropy của đoạn")
ax.set_ylim(0, 8.2)
ax.legend(loc="lower right")
ax.set_title("Hồ sơ entropy: entropy cả file là TRUNG BÌNH của các đoạn")
plt.show()In [ ]:
def chon_duong_v2(e, co_upx, nguong_nen=7.0, nguong_ma_hoa=7.5):
# Bản 2: thấy dấu UPX thì THỬ GIẢI NÉN trước, bất kể entropy; không thấy thì dùng luật cũ
if co_upx == 1:
return "2 NÉN -> thử giải nén UPX"
return chon_duong(e, nguong_nen, nguong_ma_hoa)
mal_elf["duong_v2"] = [chon_duong_v2(e, u) for e, u in zip(mal_elf.entropy, mal_elf.has_upx)] # zip = đi song song 2 cột
so_sanh = pd.DataFrame({"luật V3.2": mal_elf.duong.value_counts(), "luật bản 2": mal_elf.duong_v2.value_counts()}).fillna(0).astype(int)
print(so_sanh.sort_index().to_string())
print("\nKiến trúc CPU của 74 mẫu UPX:", mal_elf[mal_elf.has_upx == 1].elf_arch.value_counts().to_dict())In [ ]:
BANG_NATARAJ = [(10_240, 32), (30_720, 64), (61_440, 128), (102_400, 256), (204_800, 384), (512_000, 512), (1_048_576, 768)]
def chieu_rong(kich_thuoc):
# chiều rộng ảnh theo bảng Nataraj 2011: file càng lớn ảnh càng rộng
for nguong, rong in BANG_NATARAJ:
if kich_thuoc < nguong:
return rong
return 1024
def anh_tu_bytes(du_lieu):
mang = np.frombuffer(du_lieu, dtype=np.uint8)
w = chieu_rong(len(mang))
mang = np.pad(mang, (0, (w - len(mang) % w) % w)) # đệm 0 cho tròn hàng cuối
return mang.reshape(-1, w) # bảng (số hàng, w): mỗi ô = 1 byte = 1 điểm ảnh
def do_lech_histogram(du_lieu):
# phương sai của 256 số đếm: byte phân bố càng ĐỀU (nhiễu) thì càng NHỎ
return float(np.var(np.bincount(np.frombuffer(du_lieu, dtype=np.uint8), minlength=256)))
ba_file = ["chương trình /usr/bin/ls", "ls bị NÉN (zlib)", "ls bị 'MÃ HOÁ' (XOR khoá ngẫu nhiên)"]
fig, axes = plt.subplots(1, 3, figsize=(13, 6))
for ax, ten in zip(axes, ba_file):
anh = anh_tu_bytes(mau[ten])
ax.imshow(anh, cmap="gray", aspect="auto", interpolation="nearest")
ax.set_title(f"{ten}\n{anh.shape[1]}×{anh.shape[0]} điểm · entropy {entropy(mau[ten]):.2f}\n"
f"độ lệch histogram {do_lech_histogram(mau[ten]):,.0f}", fontsize=9)
ax.axis("off")
plt.tight_layout()
plt.show()In [ ]:
NGUONG_NHIEU = 800_000 # ngưỡng của pipeline vẽ ảnh
hang = []
for kb in (10, 50, 140):
doan = ls_bytes[:kb * 1024] # đoạn đầu của ls: CÓ kết cấu
hang.append({"file": f"{kb} KB đầu của ls", "entropy": round(entropy(doan), 2),
"độ lệch histogram": round(do_lech_histogram(doan)), "pipeline kết luận": "nhiễu" if do_lech_histogram(doan) < NGUONG_NHIEU else "có kết cấu"})
for mb in (1, 10, 30):
ngau_nhien = rng.integers(0, 256, mb * 1024 * 1024, dtype=np.uint8).tobytes() # nhiễu thật
hang.append({"file": f"{mb} MB byte ngẫu nhiên", "entropy": round(entropy(ngau_nhien), 2),
"độ lệch histogram": round(do_lech_histogram(ngau_nhien)), "pipeline kết luận": "nhiễu" if do_lech_histogram(ngau_nhien) < NGUONG_NHIEU else "có kết cấu"})
bang_ql = pd.DataFrame(hang)
print(bang_ql.to_string(index=False))In [ ]:
from sklearn.tree import DecisionTreeClassifier, export_text
# ✏️ SỬA: số tầng tối đa của cây
DO_SAU = 2
X = pd.DataFrame({"entropy": mal_elf.entropy, "log10_kich_thuoc": np.log10(mal_elf.size_bytes)}) # log10: 1 MB -> 6, 30 MB -> 7.5
y = mal_elf.has_upx
# 📦 X = 205 hàng × 2 cột số · y = 205 nhãn (1 = có UPX)
print("X:", X.shape, "| y: có UPX", int(y.sum()), "/ không UPX", int((y == 0).sum()))
print(f"Mốc 'đoán bừa': luôn đoán 'không UPX' đã đúng {(y == 0).mean():.1%}")
cay = DecisionTreeClassifier(max_depth=DO_SAU, random_state=RANDOM_STATE).fit(X, y)
print(f"Cây đoán đúng {cay.score(X, y):.1%} số mẫu")
print(export_text(cay, feature_names=list(X.columns)))In [ ]:
cao_nhat_khong_upx = mal_elf[mal_elf.has_upx == 0].entropy.max()
thap_nhat_co_upx = mal_elf[mal_elf.has_upx == 1].entropy.min()
print(f"Mã độc KHÔNG UPX: entropy cao nhất = {cao_nhat_khong_upx:.4f}")
print(f"Mã độc CÓ UPX : entropy thấp nhất = {thap_nhat_co_upx:.4f}")
print(f"Điểm giữa khoảng trống = {(cao_nhat_khong_upx + thap_nhat_co_upx) / 2:.4f}")
for nguong in (6.5, 7.0, 7.2, 7.5, 7.9):
dung = ((mal_elf.entropy > nguong).astype(int) == mal_elf.has_upx).mean()
print(f" luật 'entropy > {nguong}' đoán đúng {dung:.1%}")
# Thêm cả chương trình sạch của máy lớp (đều không UPX)
X2 = pd.concat([X[["entropy"]], lanh[["entropy"]]], ignore_index=True)
y2 = pd.concat([y, lanh.has_upx], ignore_index=True)
cay2 = DecisionTreeClassifier(max_depth=1, random_state=RANDOM_STATE).fit(X2, y2)
print(f"\nThêm {len(lanh)} chương trình sạch -> cây mới:")
print(export_text(cay2, feature_names=["entropy"]))In [ ]:
hai_ca = mal_elf[mal_elf.sha12.isin(["94f2e4d8d443", "f14937a19cf6"])].copy()
hai_ca["MB"] = (hai_ca.size_bytes / 1e6).round(1)
print(hai_ca[["sha12", "MB", "entropy", "elf_arch", "has_upx", "duong", "duong_v2"]].to_string(index=False))