Script Python pentru detectarea modificărilor de fișiere

Script Python pentru detectarea modificărilor de fișiere. Monitorizarea integrității fișierelor WordPress este una dintre cele mai eficiente metode de detectare a compromiterilor, deoarece orice atac reușit implică modificarea, adăugarea sau ștergerea de fișiere pe server.

Un script Python dedicat calculează hash-urile fișierelor și le compară cu valorile de referință, alertându-vă imediat când detectează modificări neautorizate.

Script Python pentru detectarea modificărilor de fișiere

Monitorizarea integrității fișierelor (File Integrity Monitoring, FIM) funcționează prin crearea unui inventar de referință al tuturor fișierelor cu hash-urile lor. La fiecare verificare ulterioară, hash-urile actuale sunt comparate cu referința pentru a detecta modificări.

Vom construi un sistem FIM complet care detectează fișiere modificate, adăugate și șterse din instalarea WordPress.

Crearea inventarului de referință

import hashlib
import os
import json
import time
from datetime import datetime
from pathlib import Path

class FileIntegrityMonitor:
    """Monitor integritate fișiere WordPress."""

    def __init__(self, wp_path, db_file='fim_baseline.json'):
        self.wp_path = wp_path
        self.db_file = db_file
        self.baseline = {}
        self.ignore_patterns = [
            'wp-content/cache',
            'wp-content/upgrade',
            'wp-content/backup',
            'wp-content/wflogs',
            '*.log',
            '*.tmp',
            'error_log'
        ]

    def should_ignore(self, filepath):
        """Verifică dacă fișierul trebuie ignorat."""
        rel_path = os.path.relpath(filepath, self.wp_path)
        for pattern in self.ignore_patterns:
            if pattern.startswith('*'):
                if filepath.endswith(pattern[1:]):
                    return True
            elif pattern in rel_path:
                return True
        return False

    def calculate_hash(self, filepath):
        """Calculează hash SHA-256 al unui fișier."""
        sha256 = hashlib.sha256()
        try:
            with open(filepath, 'rb') as f:
                for chunk in iter(lambda: f.read(8192), b''):
                    sha256.update(chunk)
            return sha256.hexdigest()
        except (PermissionError, OSError):
            return None

    def create_baseline(self):
        """Creează inventarul de referință al tuturor fișierelor."""
        print(f"Creare inventar referință: {self.wp_path}")
        start_time = time.time()
        file_count = 0

        self.baseline = {
            'created': datetime.now().isoformat(),
            'wp_path': self.wp_path,
            'files': {}
        }

        for root, dirs, files in os.walk(self.wp_path):
            for filename in files:
                filepath = os.path.join(root, filename)

                if self.should_ignore(filepath):
                    continue

                file_hash = self.calculate_hash(filepath)
                if file_hash:
                    stat = os.stat(filepath)
                    rel_path = os.path.relpath(filepath, self.wp_path)
                    self.baseline['files'][rel_path] = {
                        'hash': file_hash,
                        'size': stat.st_size,
                        'mtime': stat.st_mtime,
                        'permissions': oct(stat.st_mode)[-3:]
                    }
                    file_count += 1

        # Salvare pe disc
        with open(self.db_file, 'w') as f:
            json.dump(self.baseline, f, indent=2)

        elapsed = round(time.time() - start_time, 2)
        print(f"Inventar creat: {file_count} fișiere în {elapsed}s")
        print(f"Salvat: {self.db_file}")
        return file_count

• SHA-256 oferă securitate criptografică pentru detectarea oricărei modificări
• Ignorarea directoarelor de cache previne alerte false frecvente
• Inventarul include hash-ul, dimensiunea, data modificării și permisiunile
• Procesarea pe blocuri de 8KB permite scanarea eficientă a fișierelor mari

Verificarea modificărilor

    def load_baseline(self):
        """Încarcă inventarul de referință."""
        try:
            with open(self.db_file, 'r') as f:
                self.baseline = json.load(f)
            print(f"Inventar încărcat: {len(self.baseline['files'])} fișiere")
            print(f"Creat la: {self.baseline['created']}")
            return True
        except FileNotFoundError:
            print("EROARE: Inventar de referință nu există. Rulați create_baseline() mai întâi.")
            return False

    def check_integrity(self):
        """Verifică integritatea fișierelor contra inventarului."""
        if not self.baseline:
            if not self.load_baseline():
                return None

        report = {
            'check_date': datetime.now().isoformat(),
            'modified': [],
            'added': [],
            'deleted': [],
            'permission_changed': []
        }

        baseline_files = set(self.baseline['files'].keys())
        current_files = set()

        # Verificare fișiere existente
        for root, dirs, files in os.walk(self.wp_path):
            for filename in files:
                filepath = os.path.join(root, filename)
                if self.should_ignore(filepath):
                    continue

                rel_path = os.path.relpath(filepath, self.wp_path)
                current_files.add(rel_path)

                current_hash = self.calculate_hash(filepath)
                if not current_hash:
                    continue

                stat = os.stat(filepath)
                current_perms = oct(stat.st_mode)[-3:]

                if rel_path in self.baseline['files']:
                    baseline_info = self.baseline['files'][rel_path]

                    # Verificare hash (conținut modificat)
                    if current_hash != baseline_info['hash']:
                        report['modified'].append({
                            'file': rel_path,
                            'old_hash': baseline_info['hash'][:16] + '...',
                            'new_hash': current_hash[:16] + '...',
                            'old_size': baseline_info['size'],
                            'new_size': stat.st_size
                        })

                    # Verificare permisiuni
                    if current_perms != baseline_info['permissions']:
                        report['permission_changed'].append({
                            'file': rel_path,
                            'old_perms': baseline_info['permissions'],
                            'new_perms': current_perms
                        })
                else:
                    # Fișier nou
                    report['added'].append({
                        'file': rel_path,
                        'size': stat.st_size,
                        'is_php': rel_path.endswith('.php')
                    })

        # Fișiere șterse
        deleted = baseline_files - current_files
        for rel_path in deleted:
            report['deleted'].append({'file': rel_path})

        return report

• Detectează patru categorii de modificări: conținut, fișiere noi, fișiere șterse, permisiuni
• Fișierele PHP noi în directorul uploads sunt un indicator puternic de compromitere
• Modificările de permisiuni pot indica activitate malițioasă
• Compararea hash-urilor detectează chiar și modificări de un singur octet

Generarea raportului și alertelor

    def print_report(self, report):
        """Afișează raportul de integritate."""
        sep = "=" * 60
        print("n" + sep)
        print(f"RAPORT INTEGRITATE FIȘIERE - {report['check_date']}")
        print(sep)

        total_changes = (len(report['modified']) + len(report['added']) +
                        len(report['deleted']) + len(report['permission_changed']))

        if total_changes == 0:
            print("nNicio modificare detectată. Toate fișierele sunt integre.")
            return

        print(f"nTOTAL MODIFICĂRI: {total_changes}")

        if report['modified']:
            print(f"nFIȘIERE MODIFICATE ({len(report['modified'])}):")
            for item in report['modified']:
                print(f"  [MODIFICAT] {item['file']}")
                print(f"    Dimensiune: {item['old_size']} -> {item['new_size']} bytes")

        if report['added']:
            print(f"nFIȘIERE NOI ({len(report['added'])}):")
            for item in report['added']:
                indicator = " [PHP!]" if item['is_php'] else ""
                print(f"  [NOU{indicator}] {item['file']} ({item['size']} bytes)")

        if report['deleted']:
            print(f"nFIȘIERE ȘTERSE ({len(report['deleted'])}):")
            for item in report['deleted']:
                print(f"  [ȘTERS] {item['file']}")

        if report['permission_changed']:
            print(f"nPERMISIUNI MODIFICATE ({len(report['permission_changed'])}):")
            for item in report['permission_changed']:
                print(f"  [PERM] {item['file']}: {item['old_perms']} -> {item['new_perms']}")

    def save_report(self, report, filename=None):
        """Salvează raportul în format JSON."""
        if not filename:
            filename = f"fim_report_{datetime.now().strftime('%Y%m%d_%H%M')}.json"
        with open(filename, 'w') as f:
            json.dump(report, f, indent=2)
        print(f"nRaport salvat: {filename}")

# Utilizare
fim = FileIntegrityMonitor('/var/www/wpsecurity.ro')

# Prima rulare: creare inventar de referință
fim.create_baseline()

# Verificări ulterioare
report = fim.check_integrity()
fim.print_report(report)
fim.save_report(report)

• Creați inventarul de referință imediat după o instalare curată sau verificare de securitate
• Actualizați inventarul după fiecare actualizare legitimă WordPress
• Programați verificarea la fiecare oră prin cron pentru detectare rapidă
• Fișierele PHP noi în orice director sunt cel mai important indicator de compromitere

Automatizarea verificării și alertării

#!/usr/bin/env python3
"""Script de verificare automată, rulat prin cron."""
import smtplib
from email.mime.text import MIMEText

def send_alert_email(report, to_email):
    """Trimite alertă email cu modificările detectate."""
    total = (len(report['modified']) + len(report['added']) +
            len(report['deleted']) + len(report['permission_changed']))

    if total == 0:
        return

    body = f"Modificări detectate: {total}nn"
    for item in report['modified']:
        body += f"MODIFICAT: {item['file']}n"
    for item in report['added']:
        php = " [PHP!]" if item['is_php'] else ""
        body += f"NOU{php}: {item['file']}n"
    for item in report['deleted']:
        body += f"ȘTERS: {item['file']}n"

    msg = MIMEText(body)
    msg['Subject'] = f'[FIM ALERTĂ] {total} modificări detectate pe WordPress'
    msg['From'] = 'fim@wpsecurity.ro'
    msg['To'] = to_email

    with smtplib.SMTP('localhost') as server:
        server.send_message(msg)

# Rulare automată
fim = FileIntegrityMonitor('/var/www/wpsecurity.ro')
report = fim.check_integrity()
if report:
    fim.print_report(report)
    send_alert_email(report, 'admin@wpsecurity.ro')

# Adăugare în crontab - verificare la fiecare oră
0 * * * * /usr/bin/python3 /home/user/scripts/fim_check.py >> /home/user/logs/fim.log 2>&1

Monitorizarea integrității fișierelor cu Python oferă detecție rapidă a compromiterilor WordPress. Pentru strategii avansate de securitate, consultați securitate cibernetică și blogul nostru. Echipa WPSecurity oferă servicii speciale de monitorizare continuă, contactați-ne pentru implementare.