Analiza logurilor de securitate cu Python, ghid practic

Analiza logurilor de securitate cu Python, ghid practic. Logurile serverului web conțin informații valoroase despre fiecare cerere primită, iar analiza lor automată cu Python permite detectarea atacurilor, identificarea tiparelor suspecte și generarea de rapoarte de securitate detaliate.

Analiza manuală a logurilor care conțin mii de linii zilnic este imposibilă. Python automatizează acest proces și evidențiază doar informațiile relevante pentru securitate.

Analiza logurilor de securitate cu Python, ghid practic

Un script Python de analiză a logurilor parsează fișierele de log, extrage informațiile relevante, identifică tiparele de atac și generează rapoarte structurate. Automatizarea acestui proces transformă datele brute în informații acționabile.

Vom construi un analizor complet care procesează loguri Apache/Nginx și identifică amenințările de securitate WordPress.

Parsarea logurilor Apache

Logurile Apache urmează un format standard (Combined Log Format) care conține IP-ul, data, metoda HTTP, URL-ul, codul de răspuns, dimensiunea și user-agent-ul.

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

class LogParser:
    """Parser pentru loguri Apache în format Combined."""

    # Regex pentru format Apache Combined Log
    LOG_PATTERN = re.compile(
        r'(?P<ip>[d.]+)s+S+s+S+s+'
        r'[(?P<date>[^]]+)]s+'
        r'"(?P<method>S+)s+(?P<url>S+)s+S+"s+'
        r'(?P<status>d+)s+(?P<size>S+)s+'
        r'"(?P<referer>[^"]*)"s+'
        r'"(?P<user_agent>[^"]*)"'
    )

    def __init__(self, log_file):
        self.log_file = log_file
        self.entries = []

    def parse(self):
        """Parsează fișierul de log și returnează lista de intrări."""
        with open(self.log_file, 'r', encoding='utf-8', errors='ignore') as f:
            for line_num, line in enumerate(f, 1):
                match = self.LOG_PATTERN.match(line.strip())
                if match:
                    entry = match.groupdict()
                    entry['status'] = int(entry['status'])
                    entry['size'] = int(entry['size']) if entry['size'] != '-' else 0
                    entry['line_num'] = line_num
                    self.entries.append(entry)

        print(f"Parsate {len(self.entries)} intrări din {self.log_file}")
        return self.entries

• Expresia regulată extrage toate câmpurile relevante din fiecare linie de log
• Tratarea erorilor de encoding previne blocarea la caractere speciale
• Conversia tipurilor de date facilitează analiza ulterioară
• Numărul liniei ajută la localizarea rapidă a intrărilor suspecte

Detectarea atacurilor brute force

Atacurile brute force generează un număr mare de cereri POST către wp-login.php sau xmlrpc.php de la aceeași adresă IP într-un interval scurt.

class SecurityAnalyzer:
    """Analizor de securitate pentru loguri WordPress."""

    def __init__(self, entries):
        self.entries = entries

    def detect_brute_force(self, threshold=20):
        """Detectează atacuri brute force pe baza cererilor POST repetate."""
        login_attempts = defaultdict(list)

        for entry in self.entries:
            if entry['method'] == 'POST' and (
                'wp-login.php' in entry['url'] or
                'xmlrpc.php' in entry['url']
            ):
                login_attempts[entry['ip']].append(entry)

        attackers = {}
        for ip, attempts in login_attempts.items():
            if len(attempts) >= threshold:
                attackers[ip] = {
                    'total_attempts': len(attempts),
                    'target': 'wp-login.php' if 'wp-login' in attempts[0]['url'] else 'xmlrpc.php',
                    'first_seen': attempts[0]['date'],
                    'last_seen': attempts[-1]['date'],
                    'user_agent': attempts[0]['user_agent'][:80]
                }

        if attackers:
            sep = "=" * 60
            print("n" + sep)
            print(f"ATACURI BRUTE FORCE DETECTATE: {len(attackers)} surse")
            print(sep)
            for ip, info in sorted(attackers.items(), key=lambda x: x[1]['total_attempts'], reverse=True):
                print(f"  IP: {ip}")
                print(f"    Încercări: {info['total_attempts']}")
                print(f"    Țintă: {info['target']}")
                print(f"    Prima: {info['first_seen']}")
                print(f"    Ultima: {info['last_seen']}")
                print()

        return attackers

• Pragul de 20 de cereri POST identifică atacurile fără false pozitive semnificative
• Analiza include atât wp-login.php, cât și xmlrpc.php
• Timestamp-urile primei și ultimei cereri arată durata atacului
• User-agent-ul poate dezvălui instrumentul folosit de atacator

Detectarea scanărilor de vulnerabilități

    def detect_vulnerability_scans(self):
        """Detectează scanări de vulnerabilități prin tipare URL suspecte."""
        suspicious_patterns = [
            (r'eval(', 'Tentativă injecție eval()'),
            (r'union.*select', 'Tentativă SQL injection'),
            (r'base64', 'Posibilă ofuscare base64'),
            (r'..', 'Tentativă path traversal'),
            (r'<script', 'Tentativă XSS'),
            (r'wp-config', 'Acces wp-config.php'),
            (r'.env', 'Acces fișier .env'),
            (r'.git', 'Acces repository Git'),
            (r'phpmyadmin', 'Scanare phpMyAdmin'),
            (r'setup.php', 'Scanare pagini de setup'),
            (r'shell.', 'Căutare web shell'),
            (r'wp-content/uploads/.*.php', 'PHP în uploads')
        ]

        scanners = defaultdict(list)
        for entry in self.entries:
            url_lower = entry['url'].lower()
            for pattern, description in suspicious_patterns:
                if re.search(pattern, url_lower, re.IGNORECASE):
                    scanners[entry['ip']].append({
                        'url': entry['url'][:100],
                        'type': description,
                        'status': entry['status'],
                        'date': entry['date']
                    })
                    break

        print(f"nSCANĂRI DETECTATE: {len(scanners)} surse")
        for ip, scans in sorted(scanners.items(), key=lambda x: len(x[1]), reverse=True)[:10]:
            print(f"  IP: {ip} - {len(scans)} cereri suspecte")
            for scan in scans[:3]:
                print(f"    [{scan['status']}] {scan['type']}: {scan['url']}")

        return scanners

• Tiparele acoperă cele mai comune tipuri de atacuri: SQL injection, XSS, path traversal
• Căutarea de fișiere sensibile (.env, .git, wp-config) indică scanare automată
• IP-urile cu multiple tipuri de cereri suspecte sunt scanere automate
• Codurile de răspuns 200 pentru aceste cereri sunt cele mai periculoase

Generarea raportului complet

    def detect_suspicious_user_agents(self):
        """Identifică user-agents asociate cu instrumente de atac."""
        bad_agents = ['sqlmap', 'nikto', 'nmap', 'masscan', 'nessus',
                      'w3af', 'acunetix', 'havij', 'dirbuster', 'gobuster',
                      'wpscan', 'python-requests', 'curl/', 'wget/']

        suspicious = defaultdict(int)
        for entry in self.entries:
            ua_lower = entry['user_agent'].lower()
            for agent in bad_agents:
                if agent in ua_lower:
                    suspicious[entry['ip']] += 1
                    break

        return suspicious

    def generate_report(self, output_file='security_report.txt'):
        """Generează raport complet de securitate."""
        report = []
        report.append(f"RAPORT SECURITATE - {datetime.now().strftime('%Y-%m-%d %H:%M')}")
        report.append(f"Total intrări analizate: {len(self.entries)}")
        report.append("=" * 60)

        # Statistici generale
        status_codes = Counter(e['status'] for e in self.entries)
        report.append(f"nCoduri răspuns: {dict(status_codes.most_common(10))}")

        methods = Counter(e['method'] for e in self.entries)
        report.append(f"Metode HTTP: {dict(methods)}")

        # Top IP-uri
        top_ips = Counter(e['ip'] for e in self.entries).most_common(20)
        report.append(f"nTop 20 IP-uri (cereri totale):")
        for ip, count in top_ips:
            report.append(f"  {ip}: {count}")

        # Atacuri detectate
        brute_force = self.detect_brute_force()
        report.append(f"nAtacuri brute force: {len(brute_force)} surse")

        scanners = self.detect_vulnerability_scans()
        report.append(f"Scanări vulnerabilități: {len(scanners)} surse")

        bad_agents = self.detect_suspicious_user_agents()
        report.append(f"User-agents suspecte: {len(bad_agents)} surse")

        # Salvare raport
        with open(output_file, 'w') as f:
            f.write('n'.join(report))

        print(f"nRaport salvat: {output_file}")
        return report

# Utilizare
parser = LogParser('/var/log/apache2/access.log')
entries = parser.parse()
analyzer = SecurityAnalyzer(entries)
analyzer.generate_report()

• Raportul combină toate analizele într-un document ușor de citit
• Statisticile generale oferă o privire de ansamblu asupra traficului
• Detectarea user-agent-urilor suspecte identifică instrumente de atac
• Programați generarea zilnică a raportului și trimiteți-l prin email

Analiza automată a logurilor cu Python transformă datele brute în informații acționabile de securitate. Vizitați secțiunea de securitate cibernetică pentru strategii avansate sau blogul nostru. Echipa WPSecurity oferă servicii speciale de analiză securitate, contactați-ne pentru detalii.