42 KiB
42 KiB
In [ ]:
%%bash
# Vào thư mục bài M1
cd ~/lab/day1_labs
# In 5 tin nhắn đầu của bộ 50 tin (tail -n +2 = bỏ dòng tiêu đề; head -5 = lấy 5 dòng)
tail -n +2 Manual_SOC_Dataset.csv | head -5In [ ]:
%%bash
cd ~/lab/day1_labs
# Lần 1: chạy bộ lọc trên 50 tin nhắn quen thuộc
python rule_based_classifier.py
# Lần 2: nhân 50 tin lên 10.000 tin để đo tốc độ
python rule_based_classifier.py --simulate-mass
# Lần 3: chạy trên 10 tin kẻ tấn công đã viết lại để "lách luật"
python rule_based_classifier.py --dataset Evasion_Dataset.csvIn [ ]:
%cd ~/lab/day1_labs
# Mượn công cụ đọc file CSV, và mượn hàm is_phishing + 2 danh sách luật từ file bài M1
import csv
from rule_based_classifier import is_phishing, SUSPICIOUS_KEYWORDS, SUSPICIOUS_DOMAINS
# In 2 danh sách luật để so
print("Từ khóa trong luật :", SUSPICIOUS_KEYWORDS)
print("Tên miền trong luật:", SUSPICIOUS_DOMAINS)
print()
# Đọc từng dòng của file 10 tin lách luật; mỗi dòng có 5 cột
for ma, kenh, nguoi_gui, noi_dung, nhan in csv.reader(open("Evasion_Dataset.csv", encoding="utf-8-sig")):
# Bỏ qua dòng tiêu đề, chỉ xét các dòng có mã bắt đầu bằng EVA
if ma.startswith("EVA"):
# Hỏi bộ lọc: tin này có dính luật không?
if is_phishing(noi_dung, nguoi_gui):
ket_qua = "BẮT ĐƯỢC"
else:
ket_qua = "BỎ LỌT "
# In kết quả + mã tin + người gửi + nội dung
print(ket_qua, "|", ma, "|", nguoi_gui, "|", noi_dung)In [ ]:
%%bash
# Xem tên 3 file log đầu tiên (mỗi ngày 1 file, tên có ngày)
ls /opt/csai/data/honeypot_m1 | head -3
# Lấy 3 dòng đầu của file ngày 22/09 và trình bày lại cho dễ đọc (python -m json.tool)
head -3 /opt/csai/data/honeypot_m1/cowrie.json.2026-09-22 | python -m json.tool --json-linesIn [ ]:
%%bash
# Có bao nhiêu file log? (ls liệt kê, wc -l đếm số dòng = số file)
ls /opt/csai/data/honeypot_m1 | wc -l
# Cả thư mục nặng bao nhiêu? (du -sh = dung lượng, dạng dễ đọc)
du -sh /opt/csai/data/honeypot_m1
# Tổng cộng bao nhiêu dòng log? (cat ghép 30 file, wc -l đếm dòng)
cat /opt/csai/data/honeypot_m1/cowrie.json.* | wc -lIn [ ]:
%%bash
# (1) Mỗi LOẠI sự kiện có bao nhiêu dòng?
# grep -oh = chỉ lấy phần khớp mẫu "eventid": "..." ; sort | uniq -c = gom giống nhau và đếm ; sort -rn = nhiều nhất lên đầu
grep -oh '"eventid": "[^"]*"' /opt/csai/data/honeypot_m1/cowrie.json.* | sort | uniq -c | sort -rn
# (2) Có bao nhiêu IP KHÁC NHAU đã kết nối?
# lấy dòng kết nối -> cắt riêng phần IP -> sort -u = bỏ trùng -> đếm dòng
grep -h '"cowrie.session.connect"' /opt/csai/data/honeypot_m1/cowrie.json.* | grep -o '"src_ip": "[^"]*"' | sort -u | wc -l
# (3) Top 5 IP kết nối nhiều nhất
grep -h '"cowrie.session.connect"' /opt/csai/data/honeypot_m1/cowrie.json.* | grep -o '"src_ip": "[^"]*"' | sort | uniq -c | sort -rn | head -5In [ ]:
import glob, os
import pandas as pd
# Đọc cả 30 file, giữ NGUYÊN dạng chữ như trong log (chưa đổi kiểu dữ liệu)
cac_file = sorted(glob.glob("/opt/csai/data/honeypot_m1/cowrie.json.*"))
df = pd.concat(pd.read_json(f, lines=True, dtype=False, convert_dates=False) for f in cac_file)
print("TRƯỚC khi làm sạch:", len(df), "dòng, trộn", df.eventid.nunique(), "loại sự kiện,", len(df.columns), "cột:", list(df.columns))
# (1) Tách theo loại sự kiện + chỉ giữ cột cần dùng
ket_noi = df[df.eventid == "cowrie.session.connect"][["timestamp", "src_ip"]]
dang_nhap = df[df.eventid.isin(["cowrie.login.failed", "cowrie.login.success"])][["src_ip", "username", "password"]]
phan_mem = df[df.eventid == "cowrie.client.version"][["src_ip", "version"]]
# (2) Xử lý ô trống: lần thử đăng nhập KHÔNG gửi mật khẩu -> ghi rõ "(không có)", không xóa dòng
print("Số lần thử đăng nhập không có mật khẩu:", dang_nhap.password.isna().sum())
dang_nhap = dang_nhap.fillna({"password": "(không có)"})
# (3) Đổi giờ: chuỗi giờ quốc tế (UTC, chữ Z) -> giờ Việt Nam
ket_noi["gio_vn"] = pd.to_datetime(ket_noi.timestamp, utc=True).dt.tz_convert("Asia/Ho_Chi_Minh")
ket_noi = ket_noi[["gio_vn", "src_ip"]]
# (4) Lưu 3 bảng sạch vào thư mục ~/m12_sach
thu_muc = os.path.expanduser("~/m12_sach")
os.makedirs(thu_muc, exist_ok=True)
ket_noi.to_csv(thu_muc + "/ket_noi.csv", index=False)
dang_nhap.to_csv(thu_muc + "/dang_nhap.csv", index=False)
phan_mem.to_csv(thu_muc + "/phan_mem.csv", index=False)
print("\nSAU khi làm sạch: 3 bảng gọn trong ~/m12_sach")
print("ket_noi.csv :", len(ket_noi), "dòng\n" + ket_noi.head(3).to_string(index=False))
print("\ndang_nhap.csv:", len(dang_nhap), "dòng\n" + dang_nhap.head(3).to_string(index=False))
print("\nphan_mem.csv :", len(phan_mem), "dòng\n" + phan_mem.head(3).to_string(index=False))In [ ]:
import pandas as pd
# Mở bảng kết nối đã làm sạch ở bước ④
ket_noi = pd.read_csv("~/m12_sach/ket_noi.csv")
# Đếm mỗi IP xuất hiện bao nhiêu lần, xếp nhiều nhất lên đầu
so_lan = ket_noi.src_ip.value_counts()
# len(ket_noi) = số kết nối; len(so_lan) = số IP khác nhau
print("Số lần kết nối:", len(ket_noi), "| Số IP khác nhau:", len(so_lan))
# .head(3) = 3 IP đầu bảng
print("Top 3 IP kết nối nhiều nhất:\n" + so_lan.head(3).to_string())
# (so_lan == 1).sum() = đếm IP chỉ xuất hiện đúng 1 lần
print("Số IP chỉ ghé 1 lần:", (so_lan == 1).sum())In [ ]:
import pandas as pd
ket_noi = pd.read_csv("~/m12_sach/ket_noi.csv")
# Đọc lại cột giờ Việt Nam thành thời gian thật, rồi lấy riêng GIỜ (0-23)
gio = pd.to_datetime(ket_noi.gio_vn, utc=True).dt.tz_convert("Asia/Ho_Chi_Minh").dt.hour
# Đếm số kết nối mỗi giờ, xếp theo giờ từ 0h đến 23h
dem = gio.value_counts().sort_index()
# Mỗi giờ in 1 dòng: giờ | số kết nối | thanh # (giờ đông nhất = 40 dấu #)
for h, n in dem.items():
print(f"{h:02d}h | {n:5d} | " + "#" * int(n * 40 / dem.max()))
# idxmax() = giờ đông nhất; dem.max() / dem.min() = giờ đông gấp mấy lần giờ vắng
print("Giờ đông nhất:", dem.idxmax(), "h (giờ VN) | gấp", round(dem.max() / dem.min(), 1), "lần giờ vắng nhất")In [ ]:
import pandas as pd
# dtype=str: đọc mọi thứ là chữ (mật khẩu "123456" không bị biến thành số); keep_default_na=False: giữ mật khẩu rỗng
dang_nhap = pd.read_csv("~/m12_sach/dang_nhap.csv", dtype=str, keep_default_na=False)
print("Tổng số lần thử đăng nhập:", len(dang_nhap))
# 5 username và 5 password bị thử nhiều nhất
print("Top 5 username:\n" + dang_nhap.username.value_counts().head(5).to_string())
print("Top 5 password:\n" + dang_nhap.password.value_counts().head(5).to_string())In [ ]:
import pandas as pd
phan_mem = pd.read_csv("~/m12_sach/phan_mem.csv")
# Đếm từng tên phần mềm SSH mà kẻ tấn công khai báo
dem = phan_mem.version.value_counts()
print("Top 5 phần mềm:\n" + dem.head(5).to_string())
# dem.index[0] = tên đứng đầu; dem.iloc[0] / len(phan_mem) = tỉ lệ phần trăm của nó
print(f"Riêng {dem.index[0]} chiếm {dem.iloc[0] / len(phan_mem):.0%} số lần khai báo")In [ ]:
import pandas as pd
ket_noi = pd.read_csv("~/m12_sach/ket_noi.csv")
# File CSV lưu thời gian dưới dạng chữ -> đổi lại thành thời gian thật (giờ Việt Nam)
thoi_gian = pd.to_datetime(ket_noi.gio_vn, utc=True).dt.tz_convert("Asia/Ho_Chi_Minh")
# .dt.date = lấy NGÀY -> đếm -> lấy 5 ngày đông nhất
print("TOP 5 NGÀY bị tấn công nhiều nhất:")
print(thoi_gian.dt.date.rename("ngay").value_counts().head(5).to_string())
# .dt.day_name() = lấy THỨ trong tuần (tiếng Anh: Monday, Tuesday...) -> đếm
print("\nTấn công theo THỨ trong tuần:")
print(thoi_gian.dt.day_name().rename("thu").value_counts().to_string())In [ ]:
import pandas as pd
# Mở bảng THỬ ĐĂNG NHẬP. dtype=str: đọc mọi thứ là chữ (mật khẩu "123456" không bị biến thành số)
# keep_default_na=False: giữ nguyên mật khẩu rỗng, không coi là ô trống
dang_nhap = pd.read_csv("~/m12_sach/dang_nhap.csv", dtype=str, keep_default_na=False)
# Ghép 2 cột thành 1 chuỗi "username / password" rồi đếm -> top theo CẶP
print("TOP 10 CẶP username / password:")
print((dang_nhap.username + " / " + dang_nhap.password).value_counts().head(10).to_string())In [ ]:
import pandas as pd
# ===== SỬA DÒNG NÀY: username muốn soi (thử "admin", "ubuntu", "postgres"...) =====
USER_CAN_SOI = "root"
dang_nhap = pd.read_csv("~/m12_sach/dang_nhap.csv", dtype=str, keep_default_na=False)
# Lọc: chỉ giữ các lần thử có username đúng bằng USER_CAN_SOI (== nghĩa là "bằng")
loc = dang_nhap[dang_nhap.username == USER_CAN_SOI]
print("Username", USER_CAN_SOI, "bị thử", len(loc), "lần, từ", loc.src_ip.nunique(), "IP khác nhau")
print("TOP 10 mật khẩu kẻ tấn công thử với", USER_CAN_SOI + ":")
print(loc.password.value_counts().head(10).to_string())In [ ]:
import pandas as pd
# Mở cả 3 bảng sạch
ket_noi = pd.read_csv("~/m12_sach/ket_noi.csv")
dang_nhap = pd.read_csv("~/m12_sach/dang_nhap.csv", dtype=str, keep_default_na=False)
phan_mem = pd.read_csv("~/m12_sach/phan_mem.csv")
# set(...) = tập hợp các IP KHÁC NHAU của mỗi bảng
ip_ket_noi = set(ket_noi.src_ip)
ip_dang_nhap = set(dang_nhap.src_ip)
# Phép trừ tập hợp: IP có kết nối nhưng KHÔNG hề thử mật khẩu -> chỉ "gõ cửa" rồi đi (quét, dò dịch vụ)
print("KIỂU 1 - Chỉ kết nối rồi đi (quét / dò dịch vụ):", len(ip_ket_noi - ip_dang_nhap), "IP")
# Phép giao tập hợp: IP vừa kết nối VỪA thử mật khẩu -> dò mật khẩu (brute-force)
print("KIỂU 2 - Có thử mật khẩu (brute-force): ", len(ip_ket_noi & ip_dang_nhap), "IP")
# Rút gọn tên phần mềm: "SSH-2.0-Go" -> "Go", "SSH-2.0-PuTTY_Release_0.84" -> "PuTTY"
ho_phan_mem = phan_mem.version.str.extract(r"^SSH-[\d.]+-([A-Za-z]+)")[0].fillna("khác").rename("cong_cu")
print("\nTOP 5 CÔNG CỤ tấn công (theo số lần khai báo):")
print(ho_phan_mem.value_counts().head(5).to_string())In [ ]:
import pandas as pd
# ===== SỬA DÒNG NÀY: IP muốn soi (lấy từ kết quả TOP 10 IP ở trên) =====
IP_CAN_SOI = "154.18.197.29"
ket_noi = pd.read_csv("~/m12_sach/ket_noi.csv")
dang_nhap = pd.read_csv("~/m12_sach/dang_nhap.csv", dtype=str, keep_default_na=False)
phan_mem = pd.read_csv("~/m12_sach/phan_mem.csv")
# Lọc từng bảng: chỉ giữ dòng của đúng IP này
kn = ket_noi[ket_noi.src_ip == IP_CAN_SOI]
dn = dang_nhap[dang_nhap.src_ip == IP_CAN_SOI].replace("", "(trống)") # ô rỗng -> ghi "(trống)" cho dễ đọc
pm = phan_mem[phan_mem.src_ip == IP_CAN_SOI]
thoi_gian = pd.to_datetime(kn.gio_vn, utc=True).dt.tz_convert("Asia/Ho_Chi_Minh")
print("HỒ SƠ IP", IP_CAN_SOI)
# .min() = sớm nhất, .max() = muộn nhất, .dt.date.nunique() = số ngày khác nhau, .dt.hour.nunique() = số giờ khác nhau
print(" Lần đầu:", thoi_gian.min(), "| lần cuối:", thoi_gian.max())
print(" Xuất hiện", thoi_gian.dt.date.nunique(), "ngày, trong", thoi_gian.dt.hour.nunique(), "khung giờ khác nhau của ngày")
print(" Số lần kết nối:", len(kn), "| số lần thử đăng nhập:", len(dn))
print(" Phần mềm đã khai:", ", ".join(pm.version.unique()))
print(" TOP 5 username:", ", ".join(dn.username.value_counts().head(5).index))
print(" TOP 5 password:", ", ".join(dn.password.value_counts().head(5).index))In [ ]:
import pandas as pd
ket_noi = pd.read_csv("~/m12_sach/ket_noi.csv")
# Cắt số cuối của IP: "109.160.32.18" -> tách ở dấu chấm CUỐI -> "109.160.32" -> thêm ".0/24"
ket_noi["dai_mang"] = ket_noi.src_ip.str.rsplit(".", n=1).str[0] + ".0/24"
# Gom theo dải: đếm số kết nối và số IP khác nhau trong mỗi dải
bang = ket_noi.groupby("dai_mang").agg(so_ket_noi=("src_ip", "size"), so_ip=("src_ip", "nunique"))
print("TOP 10 DẢI MẠNG /24 (cùng 3 số đầu của IP):")
print(bang.sort_values("so_ket_noi", ascending=False).head(10).to_string())In [ ]:
import pandas as pd
ket_noi = pd.read_csv("~/m12_sach/ket_noi.csv")
ket_noi["ngay"] = pd.to_datetime(ket_noi.gio_vn, utc=True).dt.tz_convert("Asia/Ho_Chi_Minh").dt.date
# Với mỗi IP: đếm số NGÀY khác nhau nó xuất hiện -> xếp giảm dần -> lấy 10
so_ngay = ket_noi.groupby("src_ip").ngay.nunique().sort_values(ascending=False)
print("TOP 10 IP 'kiên trì' nhất (xuất hiện nhiều ngày nhất trong 30 ngày):")
print(so_ngay.head(10).to_string())
# (so_ngay == 30).sum() = có bao nhiêu IP xuất hiện đủ 30/30 ngày
print("Số IP xuất hiện đủ 30/30 ngày:", (so_ngay == 30).sum())
print("Số IP chỉ xuất hiện 1 ngày: ", (so_ngay == 1).sum())In [ ]:
import pandas as pd
# ===== SỬA DÒNG NÀY: ngày muốn soi (giờ Việt Nam), dạng năm-tháng-ngày, trong khoảng 2026-08-24 .. 2026-09-22 =====
NGAY_CAN_SOI = "2026-09-20"
ket_noi = pd.read_csv("~/m12_sach/ket_noi.csv")
thoi_gian = pd.to_datetime(ket_noi.gio_vn, utc=True).dt.tz_convert("Asia/Ho_Chi_Minh")
# Lọc các kết nối có NGÀY đúng bằng ngày cần soi (.astype(str) đổi ngày thành chữ để so sánh)
mot_ngay = ket_noi[thoi_gian.dt.date.astype(str) == NGAY_CAN_SOI]
gio_mot_ngay = thoi_gian[thoi_gian.dt.date.astype(str) == NGAY_CAN_SOI].dt.hour.rename("gio")
print("Ngày", NGAY_CAN_SOI, ":", len(mot_ngay), "kết nối từ", mot_ngay.src_ip.nunique(), "IP")
print("TOP 5 IP trong ngày:")
print(mot_ngay.src_ip.value_counts().head(5).to_string())
print("TOP 3 giờ trong ngày:")
print(gio_mot_ngay.value_counts().head(3).to_string())In [ ]:
%cd ~/lab/day1_phishing
# io, contextlib: để "nuốt" dòng chữ mà hàm tạo dữ liệu tự in ra
import io, contextlib
# Mượn file bài M1.3 (đặt tên tắt là m)
import phishing_classifier as m
# Lấy đúng 100 dòng email mà bài dùng (không in gì ra)
with contextlib.redirect_stdout(io.StringIO()):
df = m.generate_mock_data()
# Đếm tổng số dòng và số email KHÁC NHAU (nunique = đếm giá trị khác nhau)
print("Tổng số dòng:", len(df), "| số email KHÁC NHAU:", df.text.nunique())
# Bỏ bản sao (drop_duplicates), in mỗi email 1 lần kèm nhãn: spam = lừa đảo, ham = thật
for text, nhan in df.drop_duplicates("text").values:
print(nhan, "|", text)In [ ]:
%%bash
# Chú ý: thư mục day1_phishing
cd ~/lab/day1_phishing
python phishing_classifier.pyIn [ ]:
%cd ~/lab/day1_phishing
import io, contextlib
import phishing_classifier as m
# Công cụ chia dữ liệu ngẫu nhiên thành phần học / phần thi
from sklearn.model_selection import train_test_split
# Lấy lại 100 dòng email của bài
with contextlib.redirect_stdout(io.StringIO()):
df = m.generate_mock_data()
# Cách chia CŨ: chia ngẫu nhiên 70% học / 30% thi, không để ý các dòng là bản sao của nhau
tr, te = train_test_split(df.index, test_size=0.3, random_state=42)
# Tìm email vừa có trong phần thi, vừa có trong phần học (& = phần chung của 2 tập hợp)
trung = set(df.text[te]) & set(df.text[tr])
print(f"Chia ngẫu nhiên: {df.text[te].nunique()} email khác nhau trong đề thi, {len(trung)} email trong số đó ĐÃ CÓ trong lúc học")