164 KiB
164 KiB
In [ ]:
%%bash
cd /opt/csai/data/honeypot_m1
# ls -lh = liệt kê file kèm dung lượng dễ đọc; head -4 = chỉ xem 4 dòng đầu
ls -lh | head -4
# du -sh = tổng dung lượng thư mục
du -sh .
# cat ghép cả 30 file rồi wc -l đếm số dòng = số sự kiện
cat cowrie.json.* | wc -lIn [ ]:
%%bash
# sed -n "6001,6030p" = in nguyên văn từ dòng 6001 tới dòng 6030 của file ngày 20/09 (không xử lý gì)
sed -n "6001,6030p" /opt/csai/data/honeypot_m1/cowrie.json.2026-09-20In [ ]:
THU_MUC_LOG = "/opt/csai/data/honeypot_m1"
FILE_NGAY = THU_MUC_LOG + "/cowrie.json.2026-09-22"
# Đọc 3 dòng đầu của file ngày 22/09 (readline = đọc 1 dòng; strip = bỏ ký tự xuống dòng ở cuối)
with open(FILE_NGAY) as f:
ba_dong = [f.readline().strip() for _ in range(3)]
for dong in ba_dong:
print(dong)
print()
print("Kiểu dữ liệu:", type(ba_dong[0]), "| dài", len(ba_dong[0]), "ký tự")In [ ]:
import json
# ✏️ SỬA: chọn dòng số mấy (0, 1 hoặc 2)
DONG_SO = 0
su_kien = json.loads(ba_dong[DONG_SO]) # chuỗi JSON -> từ điển (dict) Python
print("Các trường:", list(su_kien.keys()))
loai = su_kien["eventid"] # str (chuỗi chữ)
thoi_gian = su_kien["timestamp"] # str
ip = su_kien["src_ip"] # str
co_cong = "src_port" in su_kien # bool (True / False): dòng này có cổng nguồn không?
cong = su_kien.get("src_port", 0) # int (số nguyên); .get(..., 0) = không có thì lấy 0
print(f"loai={loai} | thoi_gian={thoi_gian} | ip={ip} | co_cong={co_cong} | cong={cong}")
print(type(loai), type(ip), type(co_cong), type(cong))In [ ]:
ngay = thoi_gian[:10] # 10 ký tự đầu: năm-tháng-ngày
gio_phut_giay = thoi_gian[11:19] # ký tự 11 -> 18
cac_phan_ip = ip.split(".") # tách IP ở dấu chấm -> danh sách 4 phần
dai_mang = ".".join(cac_phan_ip[:3]) + ".0/24" # ghép lại 3 phần đầu -> dải mạng
ten_ngan = loai.replace("cowrie.", "").upper() # bỏ chữ "cowrie." rồi viết HOA
print("Ngày:", ngay, "| Giờ (UTC):", gio_phut_giay)
print("4 phần của IP:", cac_phan_ip, "-> dải mạng:", dai_mang)
print("Loại sự kiện viết gọn:", ten_ngan)In [ ]:
gio_utc = int(thoi_gian[11:13]) # [11:13] cắt lấy 2 ký tự ở vị trí 11 và 12, tức là chuỗi "00"; "00" (chữ) -> 0 (số) thì mới cộng trừ được
gio_vn = (gio_utc + 7) % 24 # Việt Nam = UTC + 7; % 24 để 25 giờ quay về 1 giờ
print("Giờ UTC:", gio_utc, "-> giờ Việt Nam:", gio_vn)
# Vì sao MẬT KHẨU phải giữ là chuỗi chữ, không đổi sang số?
mat_khau = "0123456"
print("Mật khẩu gốc:", mat_khau, "| đổi sang số:", int(mat_khau), "<- mất số 0 ở đầu = sai mật khẩu")
# Thử bỏ dấu # ở dòng dưới rồi chạy: cộng chữ với số sẽ báo lỗi TypeError
# print(thoi_gian[11:13] + 7)In [ ]:
# Lấy dòng thứ 3 (một lần thử đăng nhập thật)
dang_nhap = json.loads(ba_dong[2])
user = dang_nhap["username"]
la_dang_nhap = dang_nhap["eventid"].startswith("cowrie.login")
thu_root = user.lower() == "root"
gio_vn = (int(dang_nhap["timestamp"][11:13]) + 7) % 24
ngoai_gio = gio_vn < 6 or gio_vn >= 22 # 22h -> 6h sáng giờ Việt Nam
# ✏️ SỬA: trọng số (mỗi điều kiện đúng được cộng bao nhiêu điểm)
diem = int(la_dang_nhap) * 40 + int(thu_root) * 30 + int(ngoai_gio) * 30
print(f"user={user} | đăng nhập={la_dang_nhap} | thử root={thu_root} | giờ VN={gio_vn}h | ngoài giờ={ngoai_gio}")
print("Điểm rủi ro:", diem, "-> cần xem xét" if diem >= 60 else "-> bình thường")In [ ]:
with open(FILE_NGAY) as f:
dong_ngay = f.read().splitlines() # mỗi phần tử = 1 dòng (chuỗi)
# Đổi tất cả sang dict một lần (list comprehension = vòng for viết gọn trong ngoặc vuông)
su_kien_ngay = [json.loads(d) for d in dong_ngay]
# Bên trong su_kien_ngay có gì? -> một DANH SÁCH (list); mỗi phần tử là một TỪ ĐIỂN (dict) = 1 dòng log, giống 1.2
print("su_kien_ngay là", type(su_kien_ngay).__name__, "gồm", len(su_kien_ngay), "phần tử (= số sự kiện của ngày 22/09)")
print("Phần tử số 0 (kết nối): ", su_kien_ngay[0])
print("Phần tử số 2 (thử mật khẩu):", su_kien_ngay[2])
print("Lấy 1 trường của 1 phần tử: su_kien_ngay[2]['username'] =", su_kien_ngay[2]["username"])In [ ]:
# ✏️ SỬA: IP muốn kiểm tra (thử "2.57.121.112", "109.160.32.67")
IP_KIEM_TRA = "109.160.32.31"
so_lan_sai = 0
for e in su_kien_ngay:
if e["eventid"] == "cowrie.login.failed" and e["src_ip"] == IP_KIEM_TRA:
so_lan_sai += 1
if so_lan_sai >= 500:
muc_do = "NGHIÊM TRỌNG"
elif so_lan_sai >= 100:
muc_do = "CAO"
elif so_lan_sai >= 1:
muc_do = "TRUNG BÌNH"
else:
muc_do = "THẤP"
print(IP_KIEM_TRA, "thử sai", so_lan_sai, "lần trong ngày ->", muc_do)In [ ]:
# Duyệt từ đầu ngày: bỏ qua dòng không phải đăng nhập (continue), dừng ở lần đăng nhập THÀNH CÔNG đầu tiên (break)
da_xem = 0
# enumerate: mỗi vòng lấy cả vi_tri (số thứ tự 0, 1, 2...) lẫn e (1 sự kiện = 1 dict)
for vi_tri, e in enumerate(su_kien_ngay):
if not e["eventid"].startswith("cowrie.login"):
continue # không phải đăng nhập -> sang dòng kế
da_xem += 1
print(vi_tri, e["timestamp"][11:19], e["src_ip"], e["eventid"][7:], e["username"], "/", e.get("password"))
if e["eventid"] == "cowrie.login.success":
print("-> Đăng nhập THÀNH CÔNG đầu tiên của ngày ở dòng", vi_tri, "sau", da_xem, "lần thử")
break # tìm thấy rồi -> dừng vòng lặpIn [ ]:
# ✏️ SỬA: muốn gặp bao nhiêu IP khác nhau
N_IP = 5
da_gap = []
i = 0
while len(da_gap) < N_IP and i < len(su_kien_ngay):
ip_i = su_kien_ngay[i]["src_ip"]
if ip_i not in da_gap:
da_gap.append(ip_i)
print(f"dòng {i:5d} | {su_kien_ngay[i]['timestamp'][11:19]} | IP mới: {ip_i}")
i += 1
print("Đọc", i, "dòng thì gặp đủ", len(da_gap), "IP khác nhau")In [ ]:
def phan_loai_ip(so_sai, thu_root=False, ngoai_gio=False):
"""Trả về THẤP / TRUNG BÌNH / CAO / NGHIÊM TRỌNG cho một IP."""
diem = so_sai # mỗi lần thử sai = 1 điểm
if thu_root:
diem += 50
if ngoai_gio:
diem += 30
if diem >= 500:
return "NGHIÊM TRỌNG"
if diem >= 100:
return "CAO"
if diem >= 30:
return "TRUNG BÌNH"
return "THẤP"
print(phan_loai_ip(3))
print(phan_loai_ip(80, thu_root=True))
# Áp dụng cho 5 IP thật của ngày 22/09
for ip_x in ["109.160.32.31", "109.160.32.67", "2.57.121.112", "2.57.121.25", "8.8.8.8"]:
# sum(1 for ... if ...) = đếm số sự kiện thỏa điều kiện: thử sai VÀ đúng IP này
sai = sum(1 for e in su_kien_ngay if e["eventid"] == "cowrie.login.failed" and e["src_ip"] == ip_x)
# any(...) = True nếu có ÍT NHẤT MỘT sự kiện của IP này dùng username root
root = any(e["src_ip"] == ip_x and e.get("username") == "root" for e in su_kien_ngay)
print(f"{ip_x:15s} sai {sai:4d} | thử root {str(root):5s} -> {phan_loai_ip(sai, thu_root=root)}")In [ ]:
CAC_TRUONG = ("timestamp", "src_ip", "eventid", "username", "password") # tuple: cố định, không sửa được
def rut_gon(e):
"""Giữ đúng các trường trong CAC_TRUONG; trường nào không có thì để None"""
return {truong: e.get(truong) for truong in CAC_TRUONG}
gon = [rut_gon(e) for e in su_kien_ngay]
print("list có", len(gon), "phần tử, phần tử đầu:")
print(gon[0])
# Thử bỏ # dòng dưới: tuple không cho sửa -> TypeError
# CAC_TRUONG[0] = "gio"In [ ]:
ip_trong_ngay = {e["src_ip"] for e in gon}
ip_dang_nhap = {e["src_ip"] for e in gon if e["eventid"].startswith("cowrie.login")}
print("IP khác nhau trong ngày:", len(ip_trong_ngay))
print("IP có thử đăng nhập: ", len(ip_dang_nhap))
print("IP chỉ gõ cửa rồi đi (phép trừ tập hợp):", len(ip_trong_ngay - ip_dang_nhap))In [ ]:
sai_theo_ip = {}
for e in gon:
if e["eventid"] != "cowrie.login.failed":
continue
sai_theo_ip[e["src_ip"]] = sai_theo_ip.get(e["src_ip"], 0) + 1 # .get(ip, 0): chưa có thì bắt đầu từ 0
# Bên trong sai_theo_ip: dict, KHÓA = IP, GIÁ TRỊ = số lần thử sai của IP đó trong ngày
print("sai_theo_ip là", type(sai_theo_ip).__name__, "có", len(sai_theo_ip), "khóa (= số IP có thử sai). 3 cặp đầu:", dict(list(sai_theo_ip.items())[:3]))
# .items() = các cặp (IP, số lần); sorted theo phần tử thứ 2 của cặp (cap[1] = số lần), lớn trước (reverse=True); [:5] lấy 5 cặp
for ip_x, n in sorted(sai_theo_ip.items(), key=lambda cap: cap[1], reverse=True)[:5]:
print(f" {ip_x:15s} {n}")In [ ]:
user_theo_ip = {}
for e in gon:
if e["username"] is not None:
# setdefault(ip, set()): nếu IP chưa có trong dict thì tạo một set rỗng cho nó; rồi .add() thêm username vào set đó
user_theo_ip.setdefault(e["src_ip"], set()).add(e["username"])
# Bên trong user_theo_ip: dict, KHÓA = IP, GIÁ TRỊ = set các username IP đó đã thử (set tự bỏ trùng)
print("user_theo_ip có", len(user_theo_ip), "khóa (IP). Ví dụ IP 109.160.32.67 ->", len(user_theo_ip["109.160.32.67"]), "username")
# ✏️ SỬA: IP muốn xem
IP_XEM = "2.57.121.112"
print(IP_XEM, "đã thử", len(user_theo_ip.get(IP_XEM, set())), "username khác nhau, ví dụ:", sorted(user_theo_ip.get(IP_XEM, set()))[:15])In [ ]:
# ✏️ SỬA: danh sách chặn có sẵn (set) và ngưỡng số lần thử sai
DANH_SACH_CHAN = {"2.57.121.112", "203.0.113.200"}
NGUONG = 100
canh_bao = []
for ip_x, n in sai_theo_ip.items():
if n >= NGUONG or ip_x in DANH_SACH_CHAN:
canh_bao.append({"ip": ip_x, "so_lan_sai": n, "trong_ds_chan": ip_x in DANH_SACH_CHAN,
"so_user": len(user_theo_ip.get(ip_x, set()))})
# Sắp theo so_lan_sai giảm dần (dấu trừ = đảo chiều: số lớn lên trước)
for c in sorted(canh_bao, key=lambda c: -c["so_lan_sai"]):
print(c)In [ ]:
# Ô NÀY: đọc 30 file log (30 ngày) -> 1 bảng 'log' -> giữ các dòng kết nối -> lấy cột giờ + cột IP ra thành 2 mảng NumPy
import glob, time
import numpy as np
import pandas as pd
# Đọc cả 30 file log, ghép thành MỘT bảng pandas (concat = nối các bảng của từng ngày theo chiều dọc)
cac_file = sorted(glob.glob(THU_MUC_LOG + "/cowrie.json.*"))
log = pd.concat(pd.read_json(f, lines=True, dtype=False, convert_dates=False) for f in cac_file)
# Chỉ giữ các dòng KẾT NỐI (mỗi dòng = 1 lần kẻ tấn công kết nối tới honeypot)
ket_noi = log[log.eventid == "cowrie.session.connect"]
print("log = bảng MỌI sự kiện 30 ngày:", log.shape[0], "dòng x", log.shape[1], "cột")
print("ket_noi = bảng chỉ gồm các KẾT NỐI: ", ket_noi.shape[0], "dòng")
# .to_numpy() = lấy MỘT CỘT của bảng ra thành mảng NumPy
mang_gio_utc = pd.to_datetime(ket_noi.timestamp, utc=True).dt.hour.to_numpy() # mảng số: giờ UTC (0..23) của từng kết nối
mang_ip = ket_noi.src_ip.to_numpy() # mảng chữ: IP của từng kết nối
# Mỗi ô của mảng ứng với 1 dòng log: đặt 3 ô đầu cạnh dòng log gốc của chúng
print("\n3 ô đầu của 2 mảng, cạnh dòng log gốc:")
for i in range(3):
print(f" mang_gio_utc[{i}] = {mang_gio_utc[i]:2d} | mang_ip[{i}] = {mang_ip[i]:15s} <- dòng log: {ket_noi.timestamp.iloc[i]}")
# Ba thuộc tính của mảng
print("\nshape =", mang_gio_utc.shape, "-> 1 chiều, dài", mang_gio_utc.shape[0], "ô (= số kết nối)")
print("dtype =", mang_gio_utc.dtype, "-> mỗi ô là số nguyên", mang_gio_utc.itemsize, "byte")
print(f"Cả mảng chiếm {mang_gio_utc.nbytes / 1024:,.0f} KB | nhỏ nhất {mang_gio_utc.min()}, lớn nhất {mang_gio_utc.max()}")
print("Mảng IP có dtype =", mang_ip.dtype, "(object = chữ)")In [ ]:
# Cách 1: vòng for Python, đổi từng phần tử sang giờ Việt Nam
bat_dau = time.perf_counter()
mang_gio_vn_for = [(g + 7) % 24 for g in mang_gio_utc]
t_for = time.perf_counter() - bat_dau
# Cách 2: NumPy, một phép tính cho cả mảng
bat_dau = time.perf_counter()
mang_gio_vn = (mang_gio_utc + 7) % 24
t_np = time.perf_counter() - bat_dau
print(f"vòng for: {t_for * 1000:.1f} ms | NumPy: {t_np * 1000:.2f} ms | nhanh hơn ~{t_for / t_np:.0f} lần")
print("Kết quả giống nhau:", np.array_equal(mang_gio_vn, mang_gio_vn_for))In [ ]:
# bincount: đếm mỗi số 0..23 xuất hiện bao nhiêu lần trong mang_gio_vn -> theo_gio = mảng 24 ô, ô h = số kết nối lúc h giờ
theo_gio = np.bincount(mang_gio_vn, minlength=24)
print("Kết nối theo giờ VN (0h..23h):", theo_gio)
print(f"Trung bình {theo_gio.mean():.0f}/giờ | độ lệch chuẩn {theo_gio.std():.0f} | trung vị {np.median(theo_gio):.0f}")
print(f"Đông nhất {theo_gio.argmax()}h ({theo_gio.max()}) | vắng nhất {theo_gio.argmin()}h ({theo_gio.min()}) | phân vị 95: {np.percentile(theo_gio, 95):.0f}")In [ ]:
# value_counts() = mỗi IP có bao nhiêu kết nối; .to_numpy() lấy ra mảng số (mỗi ô = 1 IP)
kn_moi_ip = ket_noi.src_ip.value_counts().to_numpy()
print(f"Số IP: {kn_moi_ip.size} | trung bình {kn_moi_ip.mean():.1f} kết nối/IP | trung vị {np.median(kn_moi_ip):.0f} kết nối/IP")
print(f"Chỉ {(kn_moi_ip > kn_moi_ip.mean()).mean():.1%} số IP có nhiều kết nối hơn mức 'trung bình'")
for q in [50, 90, 95, 99]:
print(f" phân vị {q:2d}: {np.percentile(kn_moi_ip, q):6.0f} kết nối -> {q}% số IP có từ chừng này kết nối trở xuống")
# ✏️ SỬA: phân vị dùng làm ngưỡng "IP bất thường"
PHAN_VI = 99
nguong_pv = np.percentile(kn_moi_ip, PHAN_VI)
vuot = kn_moi_ip[kn_moi_ip > nguong_pv] # mặt nạ True/False (xem 4.4) lọc các IP vượt ngưỡng
print(f"\nNgưỡng phân vị {PHAN_VI} = {nguong_pv:.0f} kết nối: {vuot.size} IP vượt ngưỡng, tạo ra {vuot.sum() / kn_moi_ip.sum():.0%} tổng số kết nối")In [ ]:
# np.unique: danh sách IP khác nhau + mỗi IP xuất hiện bao nhiêu lần
cac_ip, so_ket_noi = np.unique(mang_ip, return_counts=True)
# Số lần thử sai của từng IP, xếp đúng thứ tự cac_ip
sai = log[log.eventid == "cowrie.login.failed"].src_ip.value_counts()
# reindex(cac_ip): xếp lại theo ĐÚNG thứ tự của cac_ip; IP không có lần sai nào -> điền 0
so_sai = sai.reindex(cac_ip, fill_value=0).to_numpy()
# Số username khác nhau mỗi IP đã thử
so_user = log[log.eventid.str.startswith("cowrie.login")].groupby("src_ip").username.nunique().reindex(cac_ip, fill_value=0).to_numpy()
print("Số IP:", cac_ip.size)
# 4 mảng SONG SONG: cùng vị trí k = cùng một IP
print("Vị trí k | cac_ip[k] | so_ket_noi[k] | so_sai[k] | so_user[k]")
for k in range(3):
print(f" {k} | {cac_ip[k]:14s} | {so_ket_noi[k]:13d} | {so_sai[k]:9d} | {so_user[k]:10d}")
# ✏️ SỬA: điều kiện "đáng ngờ" (| = HOẶC, & = VÀ)
mat_na = (so_ket_noi > 1000) | (so_sai > 500)
# mat_na = mảng True/False dài 2.906 ô, ô k ứng với IP cac_ip[k]. Xem 2 ví dụ:
k_vd = int(np.where(cac_ip == "130.12.182.196")[0][0]) # np.where: tìm vị trí của IP này trong cac_ip
for k in [0, k_vd]:
print(f"mat_na[{k}] = {mat_na[k]!s:5s} vì {cac_ip[k]} có {so_ket_noi[k]} kết nối, {so_sai[k]} lần thử sai")
print("Số IP đáng ngờ:", mat_na.sum(), "/", mat_na.size)
print("Các IP đó:", cac_ip[mat_na])
print(f"Họ tạo ra {so_ket_noi[mat_na].sum() / so_ket_noi.sum():.0%} tổng số kết nối")In [ ]:
def min_max(x):
"""Đưa mọi giá trị về khoảng 0..1 (nhỏ nhất -> 0, lớn nhất -> 1)"""
x = np.asarray(x, dtype=float)
khoang = x.max() - x.min()
return np.zeros_like(x) if khoang == 0 else (x - x.min()) / khoang
# ✏️ SỬA: trọng số (tổng nên = 1). Thử dồn hết cho W_USER = 1.0 xem IP đứng đầu đổi thế nào
W_KET_NOI, W_SAI, W_USER = 0.3, 0.4, 0.3
diem_rui_ro = W_KET_NOI * min_max(so_ket_noi) + W_SAI * min_max(so_sai) + W_USER * min_max(so_user)
print("Điểm nằm trong 0..1:", bool(np.all((diem_rui_ro >= 0) & (diem_rui_ro <= 1))))
# argsort: thứ tự từ nhỏ đến lớn; [::-1] đảo ngược; [:5] lấy 5 đầu
for i in np.argsort(diem_rui_ro)[::-1][:5]:
print(f"{cac_ip[i]:16s} điểm {diem_rui_ro[i]:.3f} | kết nối {so_ket_noi[i]:5d} | thử sai {so_sai[i]:5d} | {so_user[i]:4d} username")In [ ]:
# 3 thước đo trải rộng thế nào? (nhỏ nhất / trung vị / lớn nhất)
for ten_thuoc_do, x in [("so_ket_noi", so_ket_noi), ("so_sai", so_sai), ("so_user", so_user)]:
print(f" {ten_thuoc_do:10s}: {x.min():5d} / {np.median(x):5.0f} / {x.max():5d}")
# Mỗi thước đo đóng góp bao nhiêu % điểm của 1 IP: cộng thẳng vs min-max
# ✏️ SỬA: IP muốn xem
IP_DONG_GOP = "109.160.32.31"
k_dg = int(np.where(cac_ip == IP_DONG_GOP)[0][0]) # vị trí của IP này trong cac_ip
for ten_cach, ham_chuan_hoa in [("cộng thẳng", lambda x: np.asarray(x, dtype=float)), ("min-max ", min_max)]:
phan = [W_KET_NOI * ham_chuan_hoa(so_ket_noi)[k_dg], W_SAI * ham_chuan_hoa(so_sai)[k_dg], W_USER * ham_chuan_hoa(so_user)[k_dg]]
print(f"{IP_DONG_GOP} | {ten_cach}: kết nối {phan[0] / sum(phan):.0%} · thử sai {phan[1] / sum(phan):.0%} · username {phan[2] / sum(phan):.0%} của điểm")
# 5 cách chuẩn hoá: mỗi cách là 1 hàm nhận mảng số, trả về mảng số cùng độ dài
CACH_CHUAN_HOA = {
"0. KHÔNG chuẩn hoá (cộng thẳng)": lambda x: np.asarray(x, dtype=float),
"1. min-max (cách của 4.5)": min_max,
"2. log rồi min-max": lambda x: min_max(np.log1p(x)), # log1p(x) = log(1 + x): x = 0 không bị lỗi
"3. z-score": lambda x: (x - np.mean(x)) / np.std(x),
"4. xếp hạng phần trăm": lambda x: pd.Series(x).rank(pct=True).to_numpy(),
}
for ten_cach, ham_chuan_hoa in CACH_CHUAN_HOA.items():
diem_thu = W_KET_NOI * ham_chuan_hoa(so_ket_noi) + W_SAI * ham_chuan_hoa(so_sai) + W_USER * ham_chuan_hoa(so_user)
print(f"\n{ten_cach}: điểm từ {diem_thu.min():.2f} đến {diem_thu.max():.2f}")
print(" top 5:", ", ".join(cac_ip[np.argsort(diem_thu)[::-1][:5]]))In [ ]:
import matplotlib.pyplot as plt
%matplotlib inline
# --- BƯỚC 1: mỗi kết nối nằm ở ô nào? ---
# Thời gian từng kết nối, đổi sang giờ Việt Nam
tg_vn = pd.to_datetime(ket_noi.timestamp, utc=True).dt.tz_convert("Asia/Ho_Chi_Minh")
cac_ngay = np.sort(tg_vn.dt.date.unique()) # danh sách 30 ngày, xếp tăng dần
# Hàng = vị trí của ngày đó trong danh sách (searchsorted tìm vị trí cho cả 136 nghìn kết nối một lúc)
hang = np.searchsorted(cac_ngay, tg_vn.dt.date.to_numpy())
cot = tg_vn.dt.hour.to_numpy() # cột = giờ 0..23
# --- BƯỚC 2: đếm vào bảng 2 chiều ---
# Bảng 30 x 24 toàn số 0; np.add.at cộng 1 vào ô (hang, cot) cho MỖI kết nối
bang_nhiet = np.zeros((len(cac_ngay), 24), dtype=int)
np.add.at(bang_nhiet, (hang, cot), 1)
print("Bảng có shape", bang_nhiet.shape, "(ngày, giờ) | tổng", bang_nhiet.sum(), "kết nối | trung bình 1 ô:", round(bang_nhiet.mean()))
# --- BƯỚC 3: tô màu ---
THU = ["T2", "T3", "T4", "T5", "T6", "T7", "CN"] # tên thứ trong tuần
plt.figure(figsize=(14, 8))
plt.imshow(bang_nhiet, aspect="auto", cmap="inferno") # imshow: tô mỗi ô của bảng một màu theo giá trị
plt.colorbar(label="số kết nối trong 1 giờ")
plt.yticks(range(len(cac_ngay)), [f"{d:%d/%m} {THU[d.weekday()]}" for d in cac_ngay], fontsize=8)
plt.xticks(range(24), [f"{h}h" for h in range(24)])
plt.xlabel("Giờ Việt Nam"); plt.ylabel("Ngày")
plt.title("Bản đồ nhiệt tấn công SSH: mỗi ô = số kết nối trong 1 giờ của 1 ngày")
# Đánh số 5 ô nóng nhất lên hình: argsort(axis=None) xếp mọi ô của bảng phẳng, [::-1][:5] lấy 5 ô lớn nhất
top5 = np.argsort(bang_nhiet, axis=None)[::-1][:5]
for thu_tu, vi_tri in enumerate(top5, 1):
h, c = np.unravel_index(vi_tri, bang_nhiet.shape) # đổi vị trí trên bảng phẳng -> (hàng, cột)
plt.text(c, h, str(thu_tu), color="cyan", ha="center", va="center", fontsize=13, fontweight="bold")
plt.show()
# --- BƯỚC 4: ai gây ra 5 ô nóng nhất? ---
# kn = bảng nhỏ 3 cột (IP, ngày, giờ VN) của mọi kết nối, để tra xem ai nằm trong mỗi ô
kn = pd.DataFrame({"ip": ket_noi.src_ip.to_numpy(), "ngay": tg_vn.dt.date.to_numpy(), "gio": cot})
kn["dai"] = kn.ip.str.rsplit(".", n=1).str[0] + ".0/24" # dải mạng = 3 số đầu của IP
for thu_tu, vi_tri in enumerate(top5, 1): # enumerate(..., 1): đánh số thứ tự từ 1
h, c = np.unravel_index(vi_tri, bang_nhiet.shape) # vị trí trên bảng phẳng -> (hàng = ngày, cột = giờ)
o = kn[(kn.ngay == cac_ngay[h]) & (kn.gio == c)]
dai_dau = o.dai.value_counts() # đếm kết nối theo dải mạng trong ô này, nhiều nhất lên đầu
ip_dau = o.ip.value_counts() # đếm kết nối theo IP trong ô này
print(f"ô {thu_tu}: {cac_ngay[h]:%d/%m} lúc {c:2d}h | {len(o):4d} kết nối từ {o.ip.nunique():2d} IP | "
f"dải {dai_dau.index[0]} chiếm {dai_dau.iloc[0] / len(o):.0%} | 2 IP mạnh nhất: {ip_dau.index[0]} ({ip_dau.iloc[0]}), {ip_dau.index[1]} ({ip_dau.iloc[1]})")
# Cộng theo hàng (axis=1) = tổng mỗi ngày; cộng theo cột (axis=0) = tổng mỗi giờ
print("\nNgày đông nhất:", f"{cac_ngay[bang_nhiet.sum(axis=1).argmax()]:%d/%m}",
"| giờ đông nhất:", bang_nhiet.sum(axis=0).argmax(), "h | số ô dưới 100 kết nối:", (bang_nhiet < 100).sum(), "/", bang_nhiet.size)In [ ]:
# shape = (số dòng, số cột); duplicated() = đánh dấu dòng giống hệt một dòng phía trên nó
print("Kích thước:", log.shape, "(dòng, cột)")
print("Dòng trùng nhau:", log.duplicated().sum())
# isna() = True ở ô trống (NaN); .sum() đếm số True trong mỗi cột
print("\nÔ trống mỗi cột:\n" + log.isna().sum().to_string())
# dtypes = kiểu dữ liệu pandas đoán cho mỗi cột: object = chữ, float64 = số thực, int64 = số nguyên
print("\nKiểu dữ liệu mỗi cột:\n" + log.dtypes.to_string())In [ ]:
# Chỉ lấy các lần THỬ ĐĂNG NHẬP (2 loại sự kiện login), 5 cột cần dùng; .copy() = bản sao riêng để sửa thoải mái
dn = log[log.eventid.str.startswith("cowrie.login")][["timestamp", "src_ip", "eventid", "username", "password"]].copy()
# Bên trong dn: bảng các lần THỬ ĐĂNG NHẬP, CHƯA làm sạch, 5 cột
print("dn: bảng", len(dn), "dòng x", dn.shape[1], "cột", list(dn.columns))
print(dn.head(3).to_string(), "\n")
print("Số lần thử đăng nhập:", len(dn))
# Đếm từng loại bẩn
print("1. password THIẾU (NaN) :", dn.password.isna().sum())
print("2. password RỖNG \"\" :", (dn.password == "").sum(), "| username RỖNG:", (dn.username == "").sum())
print("3. username có chữ HOA :", (dn.username != dn.username.str.lower()).sum(),
f"-> {len(set(dn.username))} username khác nhau, nhưng chỉ {len(set(dn.username.str.lower()))} nếu không phân biệt hoa/thường")
print("4. username có khoảng trắng thừa :", (dn.username != dn.username.str.strip()).sum(),
"| password có khoảng trắng đầu/cuối:", (dn.password.dropna() != dn.password.dropna().str.strip()).sum())
print("5. 'username' thật ra là lệnh web :", dn.username.str.upper().str.startswith("GET /").sum())
print("6. kiểu của cột timestamp :", dn.timestamp.dtype, "(chữ) | ví dụ:", repr(dn.timestamp.iloc[0]))
print("7. dòng trùng :", dn.duplicated().sum())
# Mẫu regex [...] = "một ký tự bất kỳ trong khoảng này": ký tự điều khiển \x00-\x08, \x0e-\x1f và BOM \ufeff
KY_TU_AN = r"[\x00-\x08\x0e-\x1f\ufeff]"
print("8. có ký tự ẩn / nhị phân : password", dn.password.fillna("").str.contains(KY_TU_AN).sum(),
"| username", dn.username.str.contains(KY_TU_AN).sum())
# Gom 1 dòng THẬT cho mỗi loại bẩn để lát nữa xem trước / sau
mau = pd.concat([
dn[dn.password.isna()].head(1).assign(loai_ban="1. password thiếu"),
dn[dn.password == ""].head(1).assign(loai_ban="2. password rỗng"),
dn[dn.username == ""].head(1).assign(loai_ban="2. username rỗng"),
dn[dn.username == "Root"].head(1).assign(loai_ban="3. chữ hoa"),
dn[dn.password.fillna("").str.endswith(" ")].head(1).assign(loai_ban="4. mật khẩu có dấu cách"),
# (chọn dòng lệnh web KHÔNG chứa địa chỉ IP trong mật khẩu)
dn[dn.username.str.upper().str.startswith("GET /") & ~dn.password.fillna("").str.contains(r"\d+\.\d+\.\d+\.\d+")].head(1).assign(loai_ban="5. lệnh web"),
dn[dn.password.fillna("").str.contains("\x00")].head(1).assign(loai_ban="8. ký tự nhị phân"),
]).set_index("loai_ban")
# repr() hiện rõ sự khác nhau: nan (thiếu) / '' (rỗng) / 'Root' / '888888 ' (có dấu cách) / '\x00' (ký tự nhị phân)
mau[["timestamp", "username", "password"]].apply(lambda cot: cot.map(repr))In [ ]:
def lam_sach(bang):
"""Nhận bảng các lần thử đăng nhập (dạng như 'dn'), trả về bảng đã làm sạch. Bảng gốc giữ nguyên."""
s = bang.copy()
# a. Chữ -> thời gian (UTC). errors="coerce": gặp chuỗi không phải thời gian thì cho thành NaT, không dừng chương trình
s["timestamp"] = pd.to_datetime(s.timestamp, utc=True, errors="coerce")
# b. Bỏ dòng thiếu trường BẮT BUỘC (không có giờ hoặc không có IP thì không phân tích được)
s = s.dropna(subset=["timestamp", "src_ip"])
# c. Mật khẩu: thiếu -> "(không gửi)", rỗng -> "(rỗng)". KHÔNG strip: "888888 " khác "888888"
s["password"] = s.password.fillna("(không gửi)").replace("", "(rỗng)")
# d. Username: bỏ khoảng trắng 2 đầu, viết thường; rỗng -> "(rỗng)"
s["username"] = s.username.fillna("(không gửi)").str.strip().str.lower().replace("", "(rỗng)")
# e. Đánh dấu "username" thật ra là lệnh web (giữ dòng lại làm bằng chứng)
s["rac_http"] = s.username.str.startswith("get /")
# f. Tên sự kiện dài -> kết quả ngắn gọn
s["ket_qua"] = s.eventid.map({"cowrie.login.failed": "THẤT BẠI", "cowrie.login.success": "THÀNH CÔNG"})
# g. Giờ Việt Nam (0..23) để phân tích "ngoài giờ"
s["gio_vn"] = s.timestamp.dt.tz_convert("Asia/Ho_Chi_Minh").dt.hour
# h. Đánh dấu mật khẩu chứa ký tự ẩn / nhị phân (không sửa: đó là dấu vết thật)
s["ky_tu_an"] = s.password.str.contains(KY_TU_AN)
return s.drop(columns="eventid")
# Áp dụng cho 6 dòng mẫu và đặt TRƯỚC / SAU cạnh nhau
mau_sach = lam_sach(mau.reset_index()).set_index("loai_ban")
# .apply(lambda c: c.map(repr)): với mỗi cột, đổi từng ô thành dạng repr() để thấy rõ '' / nan / ký tự ẩn
# pd.concat({...}, axis=1): đặt 2 bảng TRƯỚC và SAU cạnh nhau theo chiều ngang
truoc_sau = pd.concat({"TRƯỚC": mau[["username", "password"]].apply(lambda c: c.map(repr)),
"SAU": mau_sach[["username", "password", "rac_http", "ky_tu_an", "ket_qua", "gio_vn"]]}, axis=1)
truoc_sauIn [ ]:
# Áp dụng cho TOÀN BỘ các lần thử đăng nhập, rồi đếm lại từng vấn đề: trước -> sau
sach = lam_sach(dn).sort_values("timestamp").reset_index(drop=True)
so_sanh = pd.DataFrame({
"TRƯỚC": [dn.password.isna().sum(), (dn.password == "").sum(), (dn.username == "").sum(),
(dn.username != dn.username.str.lower()).sum(), len(set(dn.username)), str(dn.timestamp.dtype), int(dn.isna().sum().sum())],
"SAU": [sach.password.isna().sum(), (sach.password == "").sum(), (sach.username == "").sum(),
(sach.username != sach.username.str.lower()).sum(), len(set(sach.username)), str(sach.timestamp.dtype), int(sach.isna().sum().sum())],
}, index=["password thiếu (NaN)", "password rỗng", "username rỗng", "username có chữ hoa",
"số username khác nhau", "kiểu cột timestamp", "tổng ô trống"])
print("Số dòng: trước", len(dn), "-> sau", len(sach), "(không mất dòng nào)")
# Bên trong sach: bảng đăng nhập ĐÃ làm sạch — Phần 5.5, 6 và 7 đều dùng bảng này
print("sach có các cột:", list(sach.columns))
print(sach.head(3).to_string(), "\n")
so_sanhIn [ ]:
from pathlib import Path
THU_MUC_RA = Path.home() / "lab" / "out_python"
THU_MUC_RA.mkdir(parents=True, exist_ok=True)
sach.to_csv(THU_MUC_RA / "dang_nhap_sach.csv", index=False)
# Đọc lại kiểu MẶC ĐỊNH
doc_sai = pd.read_csv(THU_MUC_RA / "dang_nhap_sach.csv")
hong = doc_sai.password.astype(str) != sach.password.to_numpy()
print("Đọc mặc định : cột password kiểu", doc_sai.password.dtype, "|", hong.sum(), "mật khẩu bị đổi, ví dụ:",
list(zip(sach.password[hong.to_numpy()].head(3), doc_sai.password[hong].head(3))))
# Bẫy thứ 2: cột CHỈ có chữ số -> pandas đổi thành SỐ, mất số 0 ở đầu
chi_so = sach[sach.password.str.fullmatch(r"\d+")][["password"]]
chi_so.to_csv(THU_MUC_RA / "mk_chi_so.csv", index=False)
doc_so = pd.read_csv(THU_MUC_RA / "mk_chi_so.csv")
print("Cột toàn chữ số: đọc lại thành kiểu", doc_so.password.dtype, "| mật khẩu bắt đầu bằng 0:", chi_so.password.str.startswith("0").sum(),
"| ví dụ", list(zip(chi_so.password[chi_so.password.str.startswith("0")].head(3), doc_so.password[chi_so.password.str.startswith("0").to_numpy()].head(3))))
# Cách ĐÚNG: dtype=str (mọi thứ là chữ) + keep_default_na=False (không coi "null", "NA", "" là ô trống)
doc_dung = pd.read_csv(THU_MUC_RA / "dang_nhap_sach.csv", dtype=str, keep_default_na=False)
con_hong = doc_dung.password != sach.password.to_numpy()
print("Đọc đúng cách :", con_hong.sum(), "mật khẩu vẫn bị đổi -> đều có ký tự nhị phân \x00:", bool(sach.ky_tu_an[con_hong.to_numpy()].all()))
# Muốn giữ NGUYÊN 100% (kể cả ký tự nhị phân) -> lưu JSON thay vì CSV
sach[["password"]].to_json(THU_MUC_RA / "mat_khau.json", orient="records", force_ascii=False)
doc_json = pd.read_json(THU_MUC_RA / "mat_khau.json", orient="records", dtype=False)
print("Lưu / đọc JSON :", (doc_json.password != sach.password.to_numpy()).sum(), "mật khẩu bị đổi")In [ ]:
from pathlib import Path
THU_MUC_RA = Path.home() / "lab" / "out_python"
THU_MUC_RA.mkdir(parents=True, exist_ok=True)
print(sach.ket_qua.value_counts().to_string())
# that_bai_ip: lọc các lần THẤT BẠI -> đếm theo IP (value_counts) -> đặt tên cột là so_lan_that_bai
# (khác dict sai_theo_ip ở 3.3 chỉ có 1 ngày: đây là Series pandas của cả 30 ngày, KHÓA = IP, GIÁ TRỊ = số lần sai)
that_bai_ip = sach[sach.ket_qua == "THẤT BẠI"].src_ip.value_counts().rename("so_lan_that_bai")
print("\nTop 5 IP thử sai:\n" + that_bai_ip.head(5).to_string())
sach.to_csv(THU_MUC_RA / "dang_nhap_sach.csv", index=False)
that_bai_ip.to_csv(THU_MUC_RA / "that_bai_theo_ip.csv")
sach.head(100).to_json(THU_MUC_RA / "dang_nhap_sach_100.json", orient="records", indent=2, date_format="iso", force_ascii=False)
print("\nĐã xuất:", sorted(p.name for p in THU_MUC_RA.iterdir()))In [ ]:
# ✏️ SỬA: ngưỡng số lần thử sai để coi là đáng ngờ
NGUONG_SAI = 500
ip_dang_ngo = that_bai_ip[that_bai_ip >= NGUONG_SAI]
print(len(ip_dang_ngo), "IP thử sai từ", NGUONG_SAI, "lần trở lên, chiếm",
f"{ip_dang_ngo.sum() / that_bai_ip.sum():.0%}", "tổng số lần thử sai")
ip_dang_ngo.head(10)In [ ]:
# ✏️ SỬA: khung "ngoài giờ" theo giờ Việt Nam (22h tối -> 6h sáng)
GIO_TOI, GIO_SANG = 22, 6
# thu_ban_dem: các lần thử đăng nhập rơi vào khung ngoài giờ (| = HOẶC: từ GIO_TOI trở đi, hoặc trước GIO_SANG)
thu_ban_dem = sach[(sach.gio_vn >= GIO_TOI) | (sach.gio_vn < GIO_SANG)]
# dem_ket_qua: ĐẾM số lần THÀNH CÔNG / THẤT BẠI (reindex để luôn đúng thứ tự này, thiếu thì 0)
dem_ket_qua = sach.ket_qua.value_counts().reindex(["THÀNH CÔNG", "THẤT BẠI"], fill_value=0)
ti_le_sai = np.round(dem_ket_qua["THẤT BẠI"] / dem_ket_qua.sum() * 100, 1)
ip_dau, so_dau = that_bai_ip.index[0], int(that_bai_ip.iloc[0])
so_gio_khung = (24 - GIO_TOI + GIO_SANG) % 24
print(f"IP thử sai nhiều nhất: {ip_dau} ({so_dau} lần)")
print(f"Tỉ lệ thất bại: {ti_le_sai}%")
print(f"Thử đăng nhập ngoài giờ: {len(thu_ban_dem)} lần = {len(thu_ban_dem) / len(sach):.0%} (khung này chiếm {so_gio_khung / 24:.0%} thời gian trong ngày)")In [ ]:
import matplotlib.pyplot as plt
%matplotlib inline
hinh, o = plt.subplots(1, 3, figsize=(16, 4))
dem_ket_qua.plot(kind="bar", ax=o[0], color=["#2a9d8f", "#e76f51"], title="Kết quả đăng nhập")
o[0].tick_params(axis="x", rotation=0)
that_bai_ip.head(8).sort_values().plot(kind="barh", ax=o[1], color="#e76f51", title="Top 8 IP thử sai")
sach.groupby("gio_vn").size().plot(kind="bar", ax=o[2], color="#457b9d", title="Thử đăng nhập theo giờ VN")
hinh.tight_layout()
hinh.savefig(THU_MUC_RA / "dieu_tra_dang_nhap.png", dpi=120)
plt.show()Warning:
Output truncated. This notebook contains too many cells to display efficiently.