linuxhunt.py

#!/usr/bin/env python3
# LinuxHunt - Linux Log Threat Hunting Tool
# Version: 1.4 (Correction parser sudo systemd-journald)

import re
import sys
import argparse
import base64
import os
import string
import subprocess
from collections import defaultdict

class Colors:
    RED = '\033[91m'
    YELLOW = '\033[93m'
    CYAN = '\033[96m'
    GREEN = '\033[92m'
    RESET = '\033[0m'
    BOLD = '\033[1m'

class LinuxHunt:
    def __init__(self):
        self.max_failed_logons = 5
        self.min_cmd_length = 500
        self.MAX_TRACKED_USERS = 10000
        
        self.failed_logons = defaultdict(int)
        self.total_failed_logons = 0
        
        self.regexes = {
            "suspicious_cmds": [
                re.compile(r"(?i)(nc\s+-e|netcat\s+-e|/dev/tcp/|bash\s+-i)"),
                re.compile(r"(?i)(wget\s+http|curl\s+-O|curl\s+-sL)"),
                re.compile(r"(?i)(chmod\s+\+s|chmod\s+4755)"),
                re.compile(r"(?i)(echo\s+[^|]+\|\s*base64\s+-d)") 
            ],
            "base64_payload": re.compile(r"(?:[A-Za-z0-9+/]{4}){10,}(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?"),
        }
        
        self.ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')

    def sanitize_output(self, text):
        if not text:
            return ""
        text = self.ansi_escape.sub('', text)
        valid_chars = set(string.printable)
        return ''.join(c for c in text if c in valid_chars)

    def print_alert(self, date, log_type, message, results, command="", decoded=""):
        s_date = self.sanitize_output(date)
        s_results = self.sanitize_output(results)
        s_command = self.sanitize_output(command)
        s_decoded = self.sanitize_output(decoded)

        print(f"{Colors.BOLD}{Colors.RED}[!] {message}{Colors.RESET}")
        print(f"    {Colors.CYAN}Date:{Colors.RESET} {s_date}")
        print(f"    {Colors.CYAN}Log:{Colors.RESET} {log_type}")
        if s_results:
            print(f"    {Colors.CYAN}Results:{Colors.RESET} {s_results}")
        if s_command:
            print(f"    {Colors.YELLOW}Command:{Colors.RESET} {s_command}")
        if s_decoded:
            print(f"    {Colors.GREEN}Decoded:{Colors.RESET} {s_decoded}")
        print("-" * 60)

    def decode_base64(self, b64_string):
        try:
            missing_padding = len(b64_string) % 4
            if missing_padding:
                b64_string += '=' * (4 - missing_padding)
            return base64.b64decode(b64_string).decode('utf-8', errors='ignore')
        except Exception:
            return None

    def validate_file(self, file_path):
        if not os.path.exists(file_path):
            sys.exit(f"[!] Erreur: Le fichier '{file_path}' n'existe pas.")
        if not os.path.isfile(file_path):
            sys.exit(f"[!] Erreur: '{file_path}' n'est pas un fichier régulier (risque de blocage I/O).")

    def validate_since_argument(self, since_str):
        if not re.match(r"^[a-zA-Z0-9\s\-:]+$", since_str):
            sys.exit(f"[!] Erreur de sécurité : Format invalide pour l'argument --since.\n"
                     f"    Caractères autorisés : lettres, chiffres, espaces, tirets (-), deux-points (:).")

    def get_log_iterator(self, file_path, use_journal, log_type, since):
        if use_journal:
            self.validate_since_argument(since)
            journalctl_bin = "/bin/journalctl" if os.path.exists("/bin/journalctl") else "/usr/bin/journalctl"
            
            if log_type == "auth":
                cmd = [journalctl_bin, "_COMM=sshd", "_COMM=sudo", "_COMM=su", "_COMM=useradd", "--no-pager", f"--since={since}"]
            else:
                cmd = [journalctl_bin, "--no-pager", f"--since={since}"]
                
            try:
                with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, errors='ignore') as process:
                    for line in process.stdout:
                        yield line
                    
                    process.wait()
                    if process.returncode != 0:
                        stderr_output = process.stderr.read().strip()
                        print(f"\n[!] Erreur critique: journalctl a renvoyé le code {process.returncode}.")
                        if stderr_output:
                            print(f"[!] Détails : {stderr_output}")
            except FileNotFoundError:
                sys.exit("[!] Erreur: la commande 'journalctl' est introuvable sur ce système.")
        else:
            self.validate_file(file_path)
            with open(file_path, 'r', errors='ignore') as f:
                for line in f:
                    yield line

    def analyze_auth_log(self, file_path, use_journal, since):
        iterator = self.get_log_iterator(file_path, use_journal, "auth", since)
        source_name = "journalctl (auth)" if use_journal else "auth.log"
        
        for line in iterator:
            parts = line.split()
            if len(parts) < 5: continue
            
            date = " ".join(parts[:3])
            content = " ".join(parts[3:])
            
            if "Failed password" in content:
                self.total_failed_logons += 1
                
                if len(self.failed_logons) > self.MAX_TRACKED_USERS:
                    print(f"{Colors.YELLOW}[!] Avertissement: Limite de tracking atteinte, cache vidé.{Colors.RESET}")
                    self.failed_logons.clear()

                match = re.search(r"for (invalid user )?(.*?) from (.*?) port", content)
                if match:
                    user = match.group(2)
                    ip = match.group(3)
                    self.failed_logons[user] += 1
                    
                    if self.failed_logons[user] == self.max_failed_logons:
                        self.print_alert(
                            date, source_name, 
                            "High number of logon failures for one account", 
                            f"Username: {user}\n    Source IP: {ip}\n    Failures exceeded ({self.max_failed_logons})"
                        )

            elif "new user: name=" in content:
                match = re.search(r"name=(.*?), UID=(.*?),", content)
                if match:
                    self.print_alert(
                        date, source_name,
                        "New User Created",
                        f"Username: {match.group(1)}\n    UID: {match.group(2)}"
                    )

            # --- CORRECTION DE LA DÉTECTION SUDO POUR SYSTEMD-JOURNALD ---
            elif "COMMAND=" in content and re.search(r"sudo(?:\[\d+\])?:\s", content):
                user_match = re.search(r"sudo(?:\[\d+\])?:\s+([^:]+)\s*:", content)
                cmd_match = re.search(r"COMMAND=(.*)", content)
                if user_match and cmd_match:
                    user = user_match.group(1).strip()
                    cmd = cmd_match.group(1).strip()
                    self.check_command(date, source_name, user, cmd)

    def analyze_syslog_or_audit(self, file_path, use_journal, since):
        iterator = self.get_log_iterator(file_path, use_journal, "syslog", since)
        source_name = "journalctl (syslog)" if use_journal else "syslog/audit"
        
        for line in iterator:
            if "msg=audit(" in line:
                date_match = re.search(r"msg=audit\(([0-9.]+):", line)
                date = f"Epoch {date_match.group(1)}" if date_match else "Audit Date"
                content = line
            else:
                parts = line.split()
                if len(parts) < 5: continue
                date = " ".join(parts[:3])
                content = " ".join(parts[3:])

            if any(x in content for x in ["history -c", "rm ~/.bash_history", "cat /dev/null > /var/log/"]):
                self.print_alert(date, source_name, "Audit/History Log Cleared", "Potential track covering detected.")

            if "type=EXECVE" in content or "COMMAND=" in content:
                cmd_extracted = content
                match_cmd = re.search(r"COMMAND=([^\s].*)", content)
                if match_cmd:
                    cmd_extracted = match_cmd.group(1)
                
                self.check_command(date, source_name, "Unknown", cmd_extracted)

    def check_command(self, date, log_type, user, command):
        alert_text = []
        decoded_text = ""
        
        if len(command) > self.min_cmd_length:
            alert_text.append(f"Long Command Line: > {self.min_cmd_length} bytes.")

        for regex in self.regexes["suspicious_cmds"]:
            if regex.search(command):
                alert_text.append("Suspicious command matched regex.")
                break

        b64_match = self.regexes["base64_payload"].search(command)
        if b64_match:
            alert_text.append("Base64-encoded string detected.")
            decoded = self.decode_base64(b64_match.group(0))
            if decoded and any(char.isalpha() for char in decoded):
                decoded_text = decoded
                for regex in self.regexes["suspicious_cmds"]:
                    if regex.search(decoded_text):
                        alert_text.append("Decoded payload matches malicious patterns.")
                        break

        if alert_text:
            self.print_alert(
                date, log_type,
                "Suspicious Command Line",
                f"User: {user}\n    Details: {' | '.join(alert_text)}",
                command,
                decoded_text
            )

def main():
    parser = argparse.ArgumentParser(description="LinuxHunt - Linux Log Threat Hunting")
    
    group = parser.add_mutually_exclusive_group(required=True)
    group.add_argument("-f", "--file", help="Path to the log file")
    group.add_argument("-j", "--journal", action="store_true", help="Read directly from systemd journalctl")
    
    parser.add_argument("-t", "--type", help="Log type (auth, syslog)", choices=["auth", "syslog"], required=True)
    parser.add_argument("-s", "--since", help="Timeframe for journalctl (e.g., '2 days ago', 'today', '2023-10-01')", default="24 hours ago")
    
    args = parser.parse_args()
    lh = LinuxHunt()
    
    if args.journal:
        print(f"[*] Processing {args.type} logs directly via journalctl (since: {args.since})...")
    else:
        print(f"[*] Processing {args.type} log: {args.file}...")
        
    print("-" * 60)
    
    if args.type == "auth":
        lh.analyze_auth_log(args.file, args.journal, args.since)
    elif args.type == "syslog":
        lh.analyze_syslog_or_audit(args.file, args.journal, args.since)
        
    if lh.total_failed_logons > lh.max_failed_logons and len(lh.failed_logons) > 1:
        lh.print_alert(
            "Multiple", "auth",
            "Distributed Password Spray Attack",
            f"Unique accounts attacked: {len(lh.failed_logons)}\n    Total logon failures: {lh.total_failed_logons}"
        )
    print("[*] Analysis complete.")

if __name__ == "__main__":
    main()