11.2.3 Log Processing

#!/usr/bin/env python3
# parse-auth-log.py
# Ekstrak IP attacker dari auth.log dan deteksi brute force

import re
import sys
from collections import Counter, defaultdict
from datetime import datetime, timedelta

class AuthLogParser:
    """Parser untuk /var/log/auth.log - deteksi brute force SSH."""

    # Pattern untuk failed password
    FAILED_PATTERN = re.compile(
        r"Failed password for (.*?) from (\d+\.\d+\.\d+\.\d+)"
    )
    # Pattern untuk success login
    SUCCESS_PATTERN = re.compile(
        r"Accepted password for (.*?) from (\d+\.\d+\.\d+\.\d+)"
    )

    def __init__(self, log_file="/var/log/auth.log"):
        self.log_file = log_file
        self.failed_attempts = []  # List of (timestamp, username, ip)
        self.success_attempts = []

    def parse(self):
        """Parse seluruh file auth.log."""
        try:
            with open(self.log_file, "r") as f:
                for line in f:
                    # Parse failed attempts
                    match = self.FAILED_PATTERN.search(line)
                    if match:
                        username, ip = match.groups()
                        # Ekstrak timestamp dari syslog
                        ts_str = " ".join(line.split()[:3])
                        self.failed_attempts.append((ts_str, username, ip))

                    # Parse success attempts
                    match = self.SUCCESS_PATTERN.search(line)
                    if match:
                        username, ip = match.groups()
                        ts_str = " ".join(line.split()[:3])
                        self.success_attempts.append((ts_str, username, ip))

        except FileNotFoundError:
            print(f"[ERROR] File {self.log_file} tidak ditemukan")
            sys.exit(1)

        return self

    def detect_brute_force(self, threshold=5):
        """Deteksi IP dengan failed attempts di atas threshold."""
        ip_counter = Counter([ip for _, _, ip in self.failed_attempts])

        attackers = [
            {"ip": ip, "attempts": count}
            for ip, count in ip_counter.items()
            if count >= threshold
        ]
        attackers.sort(key=lambda x: x["attempts"], reverse=True)

        return attackers

    def detect_credential_stuffing(self, threshold=3):
        """Deteksi username yang jadi target credential stuffing."""
        username_counter = Counter([
            user for _, user, _ in self.failed_attempts
        ])

        targets = [
            {"username": user, "attempts": count}
            for user, count in username_counter.items()
            if count >= threshold
        ]
        targets.sort(key=lambda x: x["attempts"], reverse=True)

        return targets

    def generate_report(self):
        """Generate laporan lengkap."""
        attackers = self.detect_brute_force(threshold=5)
        targets = self.detect_credential_stuffing(threshold=3)

        print("=" * 60)
        print("  AUTH LOG ANALYSIS REPORT")
        print(f"  File: {self.log_file}")
        print(f"  Total failed: {len(self.failed_attempts)}")
        print(f"  Total success: {len(self.success_attempts)}")
        print("=" * 60)

        print("\n[TOP ATTACKERS - IP dengan failed login terbanyak]")
        print(f"{'ATTEMPTS':<10} {'IP':<20}")
        print("-" * 30)
        for a in attackers[:10]:
            print(f"{a['attempts']:<10} {a['ip']:<20}")

        print("\n[TOP TARGET USERNAMES]")
        print(f"{'ATTEMPTS':<10} {'USERNAME':<20}")
        print("-" * 30)
        for t in targets[:10]:
            print(f"{t['attempts']:<10} {t['username']:<20}")

        return {"attackers": attackers, "targets": targets}

# Run
if __name__ == "__main__":
    parser = AuthLogParser("/var/log/auth.log")
    parser.parse()
    parser.generate_report()

#!/usr/bin/env python3
# parse-apache-log.py
# Parse Apache combined log format - deteksi error, scanner, bot

import re
from collections import Counter, defaultdict

class ApacheLogParser:
    """Parser untuk Apache/Nginx combined log format."""

    # Regex untuk combined log format
    LOG_PATTERN = re.compile(
        r'(\S+) (\S+) (\S+) \[([^]]+)\] '
        r'"(\S+) (\S+) (\S+)" (\d{3}) (\S+) '
        r'"([^"]*)" "([^"]*)"'
    )

    # Daftar bot/user-agent mencurigakan
    SUSPICIOUS_BOTS = [
        "mj12bot", "ahrefsbot", "semrushbot",
        "zgrab", "masscan", "nmap", "python-requests"
    ]

    def __init__(self, log_file):
        self.log_file = log_file
        self.entries = []
        self.errors_4xx = Counter()
        self.errors_5xx = Counter()
        self.ip_requests = Counter()
        self.path_requests = Counter()
        self.user_agents = Counter()

    def parse(self):
        """Parse log file per line."""
        try:
            with open(self.log_file, "r") as f:
                for line in f:
                    match = self.LOG_PATTERN.match(line)
                    if not match:
                        continue

                    ip, ident, auth, timestamp, method, path, \
                        protocol, status, size, referer, ua = match.groups()

                    entry = {
                        "ip": ip,
                        "timestamp": timestamp,
                        "method": method,
                        "path": path,
                        "protocol": protocol,
                        "status": int(status),
                        "size": size,
                        "referer": referer,
                        "user_agent": ua
                    }
                    self.entries.append(entry)

                    # Counters
                    self.ip_requests[ip] += 1
                    self.path_requests[path] += 1
                    self.user_agents[ua] += 1

                    if entry["status"] >= 500:
                        self.errors_5xx[path] += 1
                    elif entry["status"] >= 400:
                        self.errors_4xx[path] += 1

        except FileNotFoundError:
            print(f"[ERROR] File {self.log_file} tidak ditemukan")

        return self

    def detect_scanners(self, request_threshold=100):
        """Deteksi IP yang melakukan scanning (banyak request ke path berbeda)."""
        ip_paths = defaultdict(set)
        for entry in self.entries:
            ip_paths[entry["ip"]].add(entry["path"])

        scanners = [
            {"ip": ip, "unique_paths": len(paths), "total_requests": self.ip_requests[ip]}
            for ip, paths in ip_paths.items()
            if len(paths) >= request_threshold
        ]
        scanners.sort(key=lambda x: x["unique_paths"], reverse=True)

        return scanners

    def detect_attack_patterns(self):
        """Deteksi pola serangan dari path request."""
        attack_patterns = {
            "sql_injection": Counter(),
            "xss": Counter(),
            "path_traversal": Counter(),
            "lfi": Counter()
        }

        for entry in self.entries:
            path = entry["path"].lower()

            if any(kw in path for kw in ["select", "union", "insert", "drop", "--"]):
                attack_patterns["sql_injection"][entry["ip"]] += 1

            if any(kw in path for kw in ["script", "onerror", "onload", "alert("]):
                attack_patterns["xss"][entry["ip"]] += 1

            if ".." in path or "etc/passwd" in path:
                attack_patterns["path_traversal"][entry["ip"]] += 1

            if "include" in path or "file=" in path:
                attack_patterns["lfi"][entry["ip"]] += 1

        return attack_patterns

    def generate_report(self):
        """Generate analisis report."""
        scanners = self.detect_scanners(100)
        attacks = self.detect_attack_patterns()

        print("=" * 60)
        print("  APACHE LOG ANALYSIS REPORT")
        print(f"  File: {self.log_file}")
        print(f"  Total requests: {len(self.entries)}")
        print(f"  Unique IPs: {len(self.ip_requests)}")
        print("=" * 60)

        print("\n[ERROR 4xx - Top Paths]")
        for path, count in self.errors_4xx.most_common(10):
            print(f"  {count:<6} {path}")

        print("\n[ERROR 5xx - Top Paths]")
        for path, count in self.errors_5xx.most_common(5):
            print(f"  {count:<6} {path}")

        print("\n[SCANNERS - IP dengan banyak path unik]")
        for s in scanners[:5]:
            print(f"  {s['ip']:<20} {s['unique_paths']:<5} paths "
                  f"[{s['total_requests']} requests]")

        print("\n[ATTACK PATTERNS]")
        for attack_type, ips in attacks.items():
            if ips:
                total = sum(ips.values())
                print(f"  {attack_type.replace('_', ' ').title()}: {total} attempts "
                      f"dari {len(ips)} IPs")

# Run
if __name__ == "__main__":
    parser = ApacheLogParser("/var/log/apache2/access.log")
    parser.parse()
    parser.generate_report()

#!/usr/bin/env python3
# parse-json-log.py
# Parse JSON structured log untuk analisis timeline

import json
from datetime import datetime
from collections import Counter, defaultdict

class JSONLogParser:
    """Parser untuk log format JSON (modern app logs)."""

    def __init__(self, log_file):
        self.log_file = log_file
        self.entries = []
        self.by_level = Counter()
        self.by_logger = Counter()
        self.by_hour = Counter()
        self.errors_by_message = Counter()

    def parse(self):
        """Parse JSON lines (setiap baris adalah JSON object)."""
        try:
            with open(self.log_file, "r") as f:
                for line in f:
                    line = line.strip()
                    if not line:
                        continue
                    try:
                        entry = json.loads(line)
                        self.entries.append(entry)

                        # Aggregasi
                        level = entry.get("level", "UNKNOWN")
                        self.by_level[level] += 1

                        logger = entry.get("logger", "unknown")
                        self.by_logger[logger] += 1

                        # Aggregasi per jam
                        ts = entry.get("timestamp", "")
                        if ts:
                            try:
                                hour = ts[:13]  # "2024-07-22T12"
                                self.by_hour[hour] += 1
                            except (IndexError, ValueError):
                                pass

                        # Error messages
                        if level in ("ERROR", "CRITICAL", "FATAL"):
                            msg = entry.get("message", "")
                            self.errors_by_message[msg] += 1

                    except json.JSONDecodeError:
                        print(f"[WARN] Baris bukan JSON valid: {line[:80]}")

        except FileNotFoundError:
            print(f"[ERROR] File {self.log_file} tidak ditemukan")

        return self

    def timeline(self):
        """Generate timeline aktivitas per jam."""
        print("[TIMELINE - Requests per Hour]")
        print(f"{'HOUR':<20} {'COUNT':<8} {'BAR'}")
        print("-" * 60)

        max_count = max(self.by_hour.values()) if self.by_hour else 1
        for hour in sorted(self.by_hour.keys())[-24:]:  # Last 24 hours
            count = self.by_hour[hour]
            bar_length = int(count / max_count * 40)
            bar = "█" * bar_length
            print(f"{hour:<20} {count:<8} {bar}")

    def top_errors(self, limit=10):
        """Tampilkan error paling sering."""
        print(f"\n[TOP {limit} ERRORS]")
        print(f"{'COUNT':<8} {'MESSAGE'}")
        print("-" * 60)
        for msg, count in self.errors_by_message.most_common(limit):
            # Truncate message for display
            short_msg = msg[:80] + "..." if len(msg) > 80 else msg
            print(f"{count:<8} {short_msg}")

    def correlation(self, minute_window=5):
        """Cari korelasi temporal antar error."""
        error_entries = [
            e for e in self.entries
            if e.get("level") in ("ERROR", "CRITICAL")
        ]

        # Group errors yang terjadi dalam window yang sama
        from itertools import combinations
        correlations = []

        for i, e1 in enumerate(error_entries):
            for e2 in error_entries[i+1:]:
                ts1 = e1.get("timestamp", "")
                ts2 = e2.get("timestamp", "")
                if ts1 and ts2:
                    try:
                        dt1 = datetime.fromisoformat(ts1.replace("Z", "+00:00"))
                        dt2 = datetime.fromisoformat(ts2.replace("Z", "+00:00"))
                        diff = abs((dt1 - dt2).total_seconds())

                        if diff <= minute_window * 60:
                            correlations.append({
                                "window_minutes": minute_window,
                                "error1": e1.get("message", "")[:60],
                                "error2": e2.get("message", "")[:60],
                                "diff_seconds": diff
                            })
                    except (ValueError, TypeError):
                        pass

        if correlations:
            print(f"\n[CORRELATION - Errors dalam {minute_window} menit]")
            print(f"Ditemukan {len(correlations)} pasang error berkorelasi")
            for corr in correlations[:5]:
                print(f"  [{corr['diff_seconds']:.0f}s gap]")
                print(f"    A: {corr['error1']}")
                print(f"    B: {corr['error2']}")

# Run
if __name__ == "__main__":
    parser = JSONLogParser("/var/log/app/application.log")
    parser.parse()
    parser.timeline()
    parser.top_errors(10)
    parser.correlation(5)

#!/usr/bin/env python3
# brute-force-detector.py
# Deteksi brute force real-time dengan tail -f

import time
import re
import subprocess
from collections import defaultdict, deque
from datetime import datetime, timedelta

class RealTimeBruteForceDetector:
    """Deteksi brute force secara real-time dari log stream."""

    def __init__(self, log_file="/var/log/auth.log", window_minutes=10, threshold=5):
        self.log_file = log_file
        self.window = timedelta(minutes=window_minutes)
        self.threshold = threshold

        # Ring buffer: IP -> deque of (timestamp, username)
        self.attempts = defaultdict(lambda: deque(maxlen=100))

        self.failed_pattern = re.compile(
            r"Failed password for (\S+) from (\d+\.\d+\.\d+\.\d+)"
        )

    def follow(self):
        """Tail -f log file real-time."""
        with subprocess.Popen(
            ["tail", "-f", self.log_file],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        ) as proc:
            for line in proc.stdout:
                self.process_line(line.strip())

    def process_line(self, line):
        """Proses satu baris log."""
        match = self.failed_pattern.search(line)
        if not match:
            return

        username, ip = match.groups()
        now = datetime.now()

        # Simpan attempt
        self.attempts[ip].append((now, username))

        # Hapus attempt yang sudah kadaluarsa dari window
        self._cleanup(ip, now)

        # Cek threshold
        if len(self.attempts[ip]) >= self.threshold:
            self.alert(ip, username, len(self.attempts[ip]))

    def _cleanup(self, ip, now):
        """Hapus attempt di luar window."""
        cutoff = now - self.window
        while self.attempts[ip] and self.attempts[ip][0][0] < cutoff:
            self.attempts[ip].popleft()

    def alert(self, ip, last_username, count):
        """Trigger alert saat threshold terlewati."""
        print(
            f"[{datetime.now().strftime('%H:%M:%S')}] "
            f"🚨 BRUTE FORCE DETECTED!\n"
            f"    IP: {ip}\n"
            f"    Username: {last_username}\n"
            f"    Attempts: {count} in last {self.window.seconds // 60} minutes"
        )

        # Di production: kirim ke SIEM, block IP, notifikasi
        # self.block_ip(ip)
        # self.send_alert_to_siem(ip, count)

# Run
if __name__ == "__main__":
    detector = RealTimeBruteForceDetector(
        log_file="/var/log/auth.log",
        window_minutes=10,
        threshold=5
    )
    print("[DETECTOR] Monitoring brute force... (Ctrl+C to stop)")
    try:
        detector.follow()
    except KeyboardInterrupt:
        print("\n[DETECTOR] Stopped")