Log Mining Keamanan SSH dengan Polars dan Isolation Forest: Deteksi Anomali Brute Force di Server Linux dan Pemblokiran Otomatis via nftables

Setiap server Linux yang membuka port SSH ke internet menerima ribuan percobaan login gagal per hari. Alat berbasis ambang batas seperti f...

Setiap server Linux yang membuka port SSH ke internet menerima ribuan percobaan login gagal per hari. Alat berbasis ambang batas seperti fail2ban efektif untuk serangan kasar, misalnya 5 gagal dalam 10 menit. Namun penyerang modern sudah beradaptasi. Mereka melakukan low-and-slow brute force dari botnet terdistribusi, merotasi username, atau menjajal kredensial bocor dengan ritme yang sengaja dibuat di bawah ambang.

Di sinilah data mining berperan. Alih-alih aturan statis, kita membangun profil perilaku per IP dari log autentikasi, lalu membiarkan algoritma unsupervised Isolation Forest menemukan titik yang "tidak biasa" dibandingkan populasi. Untuk pemrosesan data, kita memakai Polars, library DataFrame berbasis Rust dengan eksekusi multi-thread dan lazy query optimizer. Polars mampu memproses jutaan baris log di laptop dalam hitungan detik, jauh lebih cepat dan hemat memori dibanding pandas.

1. Mengapa Isolation Forest?

• Tanpa label: kita jarang punya dataset "serangan vs. normal" yang bersih. Isolation Forest bekerja tanpa label.
• Intuisi sederhana: titik anomali lebih mudah diisolasi dengan potongan acak, sehingga kedalaman pohonnya pendek. Skor anomali diturunkan dari rata-rata kedalaman isolasi.
• Efisien: kompleksitas mendekati linear dan cocok untuk fitur numerik agregat per jendela waktu.

2. Ekstraksi Log Autentikasi dari journald

Pada distribusi berbasis systemd, log SSH tersedia di journal. Gunakan format short-unix agar timestamp berupa epoch yang tidak ambigu (unit bernama ssh di Debian/Ubuntu, sshd di RHEL/Rocky):

# Ekspor 7 hari terakhir log SSH
journalctl -u ssh --since "7 days ago" -o short-unix --no-pager > ssh_7d.log
wc -l ssh_7d.log
head -3 ssh_7d.log
# 1762639141.123456 web01 sshd[2211]: Failed password for invalid user admin from 203.0.113.7 port 50122 ssh2

# Siapkan virtualenv analisis
python3 -m venv ~/logmining && source ~/logmining/bin/activate
pip install polars scikit-learn

3. Parsing dan Feature Engineering dengan Polars

3.1 Parsing Regex Menjadi Kolom Terstruktur

import polars as pl

PAT = (r"^(?P<ts>\d+\.\d+) \S+ sshd\[\d+\]: "
       r"(?P<event>Failed password|Accepted (?:password|publickey)|Invalid user)"
       r"(?: for)?(?: invalid user)? (?P<user>\S+) from (?P<ip>[0-9a-fA-F\.:]+)")

with open("ssh_7d.log", encoding="utf-8", errors="replace") as f:
    lines = f.read().splitlines()

events = (
    pl.DataFrame({"line": lines})
    .with_columns(pl.col("line").str.extract_groups(PAT).alias("m"))
    .unnest("m")
    .drop_nulls("ip")
    .with_columns(
        pl.from_epoch((pl.col("ts").cast(pl.Float64) * 1_000_000).cast(pl.Int64),
                      time_unit="us").alias("ts")
    )
    .drop("line")
    .sort(["ip", "ts"])
)
print(events.group_by("event").len())

3.2 Profil Perilaku per IP per Jendela 10 Menit

group_by_dynamic membuat jendela waktu bergulir per IP. Kita menghitung fitur yang mencerminkan pola brute force: jumlah gagal, rasio username tidak valid, keragaman username, dan keberhasilan login.

features = (
    events.group_by_dynamic("ts", every="10m", group_by="ip")
    .agg(
        attempts=pl.len(),
        fails=(pl.col("event") == "Failed password").sum(),
        invalid=(pl.col("event") == "Invalid user").sum(),
        accepted=pl.col("event").str.starts_with("Accepted").sum(),
        distinct_users=pl.col("user").n_unique(),
    )
    .with_columns(
        invalid_ratio=(pl.col("invalid") / pl.col("attempts")),
        hour=pl.col("ts").dt.hour(),
    )
)
print(features.describe())

CATATAN - Skala Big Data: Untuk log berukuran puluhan GB dari banyak server, ganti pembacaan menjadi pl.scan_csv(...) atau simpan hasil parsing sebagai Parquet lalu gunakan pl.scan_parquet("logs/*.parquet"). Eksekusi lazy membuat Polars hanya membaca kolom yang diperlukan dan memproses data secara streaming.

4. Melatih Isolation Forest dan Menandai Anomali

from sklearn.ensemble import IsolationForest

COLS = ["attempts", "fails", "invalid", "accepted", "distinct_users", "invalid_ratio", "hour"]
X = features.select(COLS).fill_null(0).to_numpy()

model = IsolationForest(n_estimators=300, contamination=0.01, random_state=42)
model.fit(X)

scored = features.with_columns(
    score=pl.Series(model.decision_function(X)),     # makin negatif = makin anomali
    anomaly=pl.Series(model.predict(X) == -1),
)

suspects = (
    scored.filter(pl.col("anomaly") & (pl.col("accepted") == 0) & (pl.col("fails") + pl.col("invalid") >= 10))
    .group_by("ip")
    .agg(windows=pl.len(), total_fail=(pl.col("fails") + pl.col("invalid")).sum(),
         min_score=pl.col("score").min(), users=pl.col("distinct_users").max())
    .sort("min_score")
)
print(suspects.head(20))
suspects.write_csv("ssh_suspects.csv")

Perhatikan filter tambahan accepted == 0. Model hanya mengatakan "tidak biasa", dan perilaku tidak biasa belum tentu jahat. Admin yang login dari lokasi baru pada jam 2 pagi juga anomali. Kombinasi skor anomali dengan aturan domain (tanpa login sukses, minimal 10 kegagalan) menekan false positive secara signifikan.

5. Pemblokiran Otomatis dengan nftables Set

5.1 Siapkan Tabel dan Set dengan Timeout

# /etc/nftables.d/ssh-guard.nft
table inet ssh_guard {
    set blocklist4 {
        type ipv4_addr
        flags timeout
    }
    chain input {
        type filter hook input priority -5; policy accept;
        tcp dport 22 ip saddr @blocklist4 counter drop
    }
}

sudo nft -f /etc/nftables.d/ssh-guard.nft
sudo nft list set inet ssh_guard blocklist4

5.2 Script Pengumpan Blocklist

import ipaddress, subprocess, polars as pl

ALLOWLIST = [ipaddress.ip_network(n) for n in ("10.10.10.0/24", "192.168.12.0/24", "100.64.0.0/10")]

def allowed(ip):
    return any(ip in net for net in ALLOWLIST)

for row in pl.read_csv("ssh_suspects.csv").iter_rows(named=True):
    ip = ipaddress.ip_address(row["ip"])
    if ip.version != 4 or allowed(ip):
        continue
    subprocess.run(["nft", "add", "element", "inet", "ssh_guard", "blocklist4",
                    "{", f"{ip} timeout 24h", "}"], check=True)
    print(f"blocked {ip} score={row['min_score']:.3f} fails={row['total_fail']}")

PERINGATAN - Jangan Mengunci Diri Sendiri: Selalu masukkan subnet manajemen, VPN (Tailscale 100.64.0.0/10/WireGuard), dan IP bastion ke allowlist sebelum mengaktifkan pemblokiran otomatis. Gunakan timeout agar blokir kedaluwarsa sendiri, uji dulu dalam mode dry-run (cukup cetak IP), dan pastikan akses konsol out-of-band (Proxmox noVNC/IPMI) tersedia jika terjadi salah blokir.

6. Otomatisasi Berkala dengan systemd Timer

# /etc/systemd/system/ssh-anomaly.service
[Unit]
Description=SSH log mining & anomaly blocklist

[Service]
Type=oneshot
WorkingDirectory=/opt/ssh-anomaly
ExecStart=/bin/sh -c 'journalctl -u ssh --since "7 days ago" -o short-unix --no-pager > ssh_7d.log && /root/logmining/bin/python detect.py && /root/logmining/bin/python block.py'

# /etc/systemd/system/ssh-anomaly.timer
[Timer]
OnCalendar=*:0/15
Persistent=true

[Install]
WantedBy=timers.target

sudo systemctl daemon-reload
sudo systemctl enable --now ssh-anomaly.timer
systemctl list-timers ssh-anomaly.timer

7. Pengembangan Lanjutan

• Skala fleet: kirim log dari seluruh server ke Redpanda/ClickHouse, lalu jalankan feature engineering langsung di SQL dan latih model dari hasil query.
• Fitur tambahan: ASN/negara (GeoIP), entropi username, interval antar-percobaan (bot cenderung periodik), dan korelasi dengan log web.
• Evaluasi model: bandingkan hasil dengan blocklist publik (AbuseIPDB, CrowdSec CTI) untuk mengestimasi presisi, dan latih ulang secara berkala karena pola serangan berubah (concept drift).
• Hardening dasar tetap nomor satu: nonaktifkan PasswordAuthentication, gunakan kunci/sertifikat SSH atau akses Zero Trust. Deteksi anomali adalah lapisan pertahanan tambahan, bukan pengganti.

Kesimpulan

Dengan Polars untuk parsing dan agregasi berkecepatan tinggi serta Isolation Forest untuk deteksi tanpa label, log autentikasi yang biasanya hanya ditumpuk berubah menjadi sinyal keamanan yang bisa ditindaklanjuti. Pendekatan ini menangkap pola brute force yang lolos dari aturan ambang statis. Selama dipadukan dengan allowlist, timeout, dan hardening dasar, pemblokiran otomatis tetap aman dioperasikan di server produksi.

Bagikan Artikel:

WhatsApp Telegram X LinkedIn Facebook