Туннелирование NTFS: поиск USN Record в незанятой области

код для поиска USN-записей

import struct
import json
import csv
from datetime import datetime, timedelta
import os
def windows_filetime_to_datetime(filetime_bytes):
    """Конвертирует 8 байт Windows FileTime в datetime"""
    try:
        value = struct.unpack('<Q', filetime_bytes)[0]
        if value == 0:
            return None
        epoch = datetime(1601, 1, 1)
        microseconds = value // 10
        return epoch + timedelta(microseconds=microseconds)
    except:
        return None
def is_relevant_date(dt):
    """
    Проверяет, является ли дата релевантной
    Обычно это даты с 2010 по 2030 год
    """
    if not dt:
        return False
    return 2010 <= dt.year <= 2030
def has_invalid_filename_chars(filename):
    """Проверяет наличие запрещенных символов в имени файла"""
    if not filename:
        return True
    forbidden_chars = ['<', '>', ':', '"', '/', '\\', '|', '?', '*']
    return any(char in filename for char in forbidden_chars)
def is_valid_filename_length(bytes_to_read):
    """Проверяет что длина имени файла не превышает 0x1FE (510 байт) учитывая кодировку UTF-16 LE 00 разделитель"""
    return bytes_to_read <= 0x1FE
def is_printable_filename(filename):
    """Проверяет, что имя файла состоит из печатных символов"""
    if not filename:
        return False
    # Проверяем на бинарные данные (много нулевых символов или управляющих)
    if any(ord(c) < 32 and c not in '\t\n\r' for c in filename):
        return False
    # Проверяем, что есть хотя бы один печатный символ
    if not any(c.isprintable() for c in filename):
        return False
    return True
def get_mft_entry_from_position(data, local_position, absolute_position):
    """
    Получает MFT Entry Number отступив от даты на 24 байта вверх
    """
    # Отступаем на 24 байта вверх от позиции даты (локально в чанке)
    mft_local_position = local_position - 24
    # Проверяем что позиция валидная в текущем чанке
    if mft_local_position < 0 or mft_local_position + 7 >= len(data):
        return None
    try:
        # Читаем 4 байта начиная с этой позиции как little-endian dword
        mft_entry_bytes = data[mft_local_position:mft_local_position+4]
        mft_entry = struct.unpack('<I', mft_entry_bytes)[0]
        # Проверяем что MFT Entry в разумном диапазоне
        if 0 <= mft_entry <= 100000000:
            return {
                'position': absolute_position - 24,  # АБСОЛЮТНАЯ позиция в файле
                'mft_entry_bytes': mft_entry_bytes.hex().upper(),
                'mft_entry': mft_entry
            }
    except:
        pass
    return None
def get_parent_entry_from_position(data, local_position, absolute_position):
    """
    Получает Parent Entry отступив от даты на 16 байта вверх
    """
    # Отступаем на 16 байта вверх от позиции даты (локально в чанке)
    parent_local_position = local_position - 16
    # Проверяем что позиция валидная в текущем чанке
    if parent_local_position < 0 or parent_local_position + 7 >= len(data):
        return None
    try:
        # Читаем 4 байта начиная с этой позиции как little-endian dword
        parent_entry_bytes = data[parent_local_position:parent_local_position+4]
        parent_entry = struct.unpack('<I', parent_entry_bytes)[0]
        # Проверяем что Parent Entry в разумном диапазоне
        if 0 <= parent_entry <= 100000000:
            return {
                'position': absolute_position - 16,  # АБСОЛЮТНАЯ позиция в файле
                'parent_entry_bytes': parent_entry_bytes.hex().upper(),
                'parent_entry': parent_entry
            }
    except:
        pass
    return None
def read_utf16_string(data, local_position, absolute_position):
    """
    Читает строку в кодировке UTF-16 LE:
    1. От даты вперед 24 байта, читаем word (2 байта) - количество байтов для чтения
    2. От даты вперед 28 байт, читаем указанное количество байтов
    3. Конвертируем в UTF-16 LE строку
    """
    try:
        # 1. От даты вперед 24 байта (локально в чанке), читаем word (2 байта)
        length_local_position = local_position + 24
        if length_local_position + 2 > len(data):
            return None
        length_bytes = data[length_local_position:length_local_position+2]
        bytes_to_read = struct.unpack('<H', length_bytes)[0]  # количество байтов для чтения
        # ПРОВЕРКА ДЛИНЫ: если больше 0x1FE (510 байт) - пропускаем
        if not is_valid_filename_length(bytes_to_read):
            return None
        # 2. От даты вперед 28 байт (локально в чанке), читаем указанное количество байтов
        string_local_position = local_position + 28
        string_end_position = string_local_position + bytes_to_read
        if string_end_position > len(data):
            return None
        string_bytes = data[string_local_position:string_end_position]
        # 3. Декодируем из UTF-16 LE
        string_value = string_bytes.decode('utf-16le', errors='ignore').rstrip('\x00')
        # ПРОВЕРКА ЗАПРЕЩЕННЫХ СИМВОЛОВ
        if has_invalid_filename_chars(string_value):
            return None
        # ПРОВЕРКА ЧИТАЕМОСТИ ИМЕНИ
        if not is_printable_filename(string_value):
            return None
        return {
            'length_position': absolute_position + 24,  # АБСОЛЮТНАЯ позиция в файле
            'length_bytes': length_bytes.hex().upper(),
            'bytes_to_read': bytes_to_read,
            'string_position': absolute_position + 28,  # АБСОЛЮТНАЯ позиция в файле
            'string_value': string_value
        }
    except Exception as e:
        return None
def find_datetime_patterns_in_file(filename, chunk_size=1024*1024*100):  # 100MB chunks
    """Ищет паттерны в файле любого размера с использованием чанков"""
    results = []
    with open(filename, 'rb') as f:
        file_size = f.seek(0, 2)  # Получаем размер файла
        f.seek(0)  # Возвращаемся в начало
        print(f"Размер файла: {file_size} байт")
        chunk_number = 0
        position_offset = 0
        while True:
            # Читаем чанк с перекрытием для поиска паттернов на границах
            overlap = 32  # достаточный overlap для поиска паттернов
            read_size = chunk_size + overlap
            if position_offset > 0:
                f.seek(position_offset - overlap)
            else:
                f.seek(0)
            data = f.read(read_size)
            if not data:
                break
            actual_chunk_size = min(chunk_size, len(data))
            print(f"Обработка чанка {chunk_number + 1} ({len(data)} байт)...")
            # Ищем паттерны в текущем чанке
            i = 0
            while i <= len(data) - 16 - (overlap if position_offset + i >= chunk_size else 0):
                date_bytes = data[i:i+8]
                # проверяем Reason code в little endian
                if (i + 15 < len(data) and
                    data[i+8] == 0x00 and
                    data[i+9] in (0x10, 0x20) and  # Младшие байты Reason code
                    all(b == 0x00 for b in data[i+10:i+12])):  # Старшие байты Reason code (должны быть 00)
                    # Дополнительная проверка: читаем полный Reason code как little endian
                    reason_bytes = data[i+8:i+12]
                    reason = struct.unpack('<I', reason_bytes)[0]
                    # ФИЛЬТР: ТОЛЬКО RENAME_OLD_NAME и RENAME_NEW_NAME
                    if reason not in (0x00001000, 0x00002000):
                        i += 1
                        continue
                    dt = windows_filetime_to_datetime(date_bytes)
                    if dt and is_relevant_date(dt):
                        pattern_type = "RENAME_OLD_NAME" if reason == 0x00001000 else "RENAME_NEW_NAME"
                        absolute_position = position_offset + i
                        # ПЕРЕДАЕМ АБСОЛЮТНЫЕ ПОЗИЦИИ В ФУНКЦИИ
                        mft_info = get_mft_entry_from_position(data, i, absolute_position)
                        parent_info = get_parent_entry_from_position(data, i, absolute_position)
                        string_info = read_utf16_string(data, i, absolute_position)
                        # ДОБАВЛЯЕМ ТОЛЬКО ЕСЛИ ЕСТЬ ВАЛИДНОЕ ИМЯ ФАЙЛА
                        if string_info and string_info['bytes_to_read'] > 0:
                            results.append({
                                'position': absolute_position,
                                'date_hex': ''.join(f'{b:02X}' for b in date_bytes),
                                'datetime': dt,
                                'pattern_type': pattern_type,
                                'reason_code': f'{reason:08X}',
                                'full_pattern': data[i:i+16].hex().upper(),
                                'mft_info': mft_info,
                                'parent_info': parent_info,
                                'string_info': string_info
                            })
                        # ЛОГИКА ПРОПУСКА
                        if string_info and string_info['bytes_to_read'] > 0:
                            # СЦЕНАРИЙ 1: Полный паттерн + есть валидное имя файла
                            skip_bytes = 28 + string_info['bytes_to_read']
                            # Проверяем границы данных
                            if i + skip_bytes <= len(data):
                                i += skip_bytes
                            else:
                                i += 16  # fallback
                        else:
                            # СЦЕНАРИЙ 2: Полный паттерн без имени файла или невалидное имя
                            i += 16
                        continue
                # СЦЕНАРИЙ 3-4: Паттерн не найден ИЛИ найден частично
                # Двигаемся по 1 байту для тщательного поиска
                i += 1
                # Показываем прогресс внутри чанка
                if i % (1024*1024) == 0:
                    bytes_processed = position_offset + i
                    progress = (bytes_processed / file_size) * 100 if file_size > 0 else 0
                    print(f"Обработано {bytes_processed} байт ({progress:.1f}%)...")
            # Переходим к следующему чанку
            position_offset += chunk_size
            chunk_number += 1
            # Проверяем, не достигли ли конца файла
            if position_offset >= file_size:
                break
    return results
def save_to_json(results, filename):
    """Сохраняет результаты в JSON файл"""
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(results, f, indent=2, ensure_ascii=False, default=str)
    print(f"Результаты сохранены в {filename}")
def save_to_csv(results, filename):
    """Сохраняет результаты в CSV файл"""
    with open(filename, 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['MFT Entry', 'Parent Entry', 'File Name', 'Record Type', 'Reason Code', 'Windows Time'])
        for result in results:
            mft_entry = result['mft_info']['mft_entry'] if result['mft_info'] else 'N/A'
            parent_entry = result['parent_info']['parent_entry'] if result['parent_info'] else 'N/A'
            filename_str = result['string_info']['string_value'] if result['string_info'] else 'N/A'
            writer.writerow([mft_entry, parent_entry, filename_str, result['pattern_type'], result['reason_code'], result['datetime']])
    print(f"Результаты сохранены в {filename}")
# Основная программа
if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1:
        filename = sys.argv[1]
    else:
        filename = input("Введите путь к файлу: ")
    try:
        print(f"Чтение файла: {filename}")
        print("Поиск паттернов: 8 байт даты + RENAME_OLD_NAME (0x1000) ИЛИ RENAME_NEW_NAME (0x2000)")
        print("Фильтрация: только релевантные даты (2010-2030 годы)")
        print("Проверки: запрещенные символы в именах, длина имени ≤ 510 байт, читаемые имена")
        print("Режим: обработка файлов любого размера")
        print("Оптимизация: пропуск области имени файла после найденного паттерна\n")
        results = find_datetime_patterns_in_file(filename)
        # Статистика по типам записей
        old_name_count = len([r for r in results if r['pattern_type'] == 'RENAME_OLD_NAME'])
        new_name_count = len([r for r in results if r['pattern_type'] == 'RENAME_NEW_NAME'])
        print(f"   Найдено {len(results)} валидных записей:")
        print(f"   RENAME_OLD_NAME: {old_name_count} записей")
        print(f"   RENAME_NEW_NAME: {new_name_count} записей")
        # Сохраняем в JSON (ТОЛЬКО валидные результаты)
        save_to_json(results, 'CarverUSNREC.json')
        # Сохраняем в CSV (ТОЛЬКО валидные результаты)
        save_to_csv(results, 'CarverUSNREC.csv')
        print(" Обработка завершена")
    except FileNotFoundError:
        print(f" Файл {filename} не найден")
    except Exception as e:
        print(f" Ошибка: {e}")