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

import structimport jsonimport csvfrom datetime import datetime, timedeltaimport osdef 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 Nonedef is_relevant_date(dt):    """    Проверяет, является ли дата релевантной    Обычно это даты с 2010 по 2030 год    """    if not dt:        return False    return 2010 <= dt.year <= 2030def 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 <= 0x1FEdef 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 Truedef 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 Nonedef 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 Nonedef 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 Nonedef 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 resultsdef 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}")