Методология анализа файлов 1С (FDB, DAT, DBF): восстановление удалённых операций и криминалистический аудит 2026

.DBF     — журнал операций (все проводки)

1SCRDOC.DBF     — связи документов (какой документ породил какую проводку)

SC*.DBF         — справочники (SCxxxx.DBF, где xxxx — код справочника)

DH*.DBF         — шапки документов

DT*.DBF         — табличные части документов

RA*.DBF         — движения регистров

RG*.DBF         — итоги регистров

1SCONST.DBF     — константы

USERS.DBF       — список пользователей

Формат    Версия 1С   Расширение    Восст. удал.  Журн. транз.  Сложность

-----------------------------------------------------------------------

DBF       7.7         .DBF          ★★★★★         Нет           Низкая

Файловая  8.x         .1CD/.dt      ★★★☆☆         Внутренний    Средняя

Firebird  8.x         .FDB          ★★★★☆         Да (OAT/OIT)  Высокая

MS SQL    8.x         .MDF/.LDF     ★★★★★         Да (LDF)      Высокая

PostgreSQL 8.x        директория    ★★★☆☆         WAL           Высокая

<h2 id="shag-1-ubeditsya-chto-1s-ostanovlena-i-fayl-ne-otkryt">Шаг 1: Убедиться, что 1С остановлена и файл не открыт</h2>

<h2 id="proverit-protsessy-windows">Проверить процессы (Windows)</h2>

tasklist | findstr "1cv8"

<h2 id="ili-na-linux">или на Linux</h2>

ps aux | grep 1cv8



<h2 id="shag-2-snyat-pobitovuyu-kopiyu-s-heshirovaniem">Шаг 2: Снять побитовую копию с хэшированием</h2>

<h2 id="linux-macos">Linux/macOS</h2>

sudo dd if=/путь/к/1Cv8.1CD bs=4096 conv=noerror,sync \

  | tee /криминалистика/evidence/1Cv8_ORIGINAL.1CD \

  | sha256sum > /криминалистика/evidence/1Cv8_ORIGINAL.1CD.sha256



<h2 id="alternativa-dcfldd-s-vstroennym-heshirovaniem">Альтернатива — dcfldd с встроенным хэшированием</h2>

dcfldd if=/путь/к/1Cv8.1CD of=/криминалистика/evidence/1Cv8_ORIGINAL.1CD \

  hash=sha256 hashlog=/криминалистика/evidence/1Cv8.sha256log



<h2 id="windows-ftk-imager-cli">Windows — FTK Imager CLI</h2>

FTKImager.exe /SourcePath "C:\Базы\MyBase\1Cv8.1CD" \

              /DestinationPath "E:\Evidence\" \

              /AD  # добавить доп. хэш (MD5+SHA1)

<h2 id="vazhno-firebird-dolzhen-byt-ostanovlen-ili-ispolzovat">ВАЖНО: Firebird должен быть остановлен ИЛИ использовать</h2>

<h2 id="ofitsialnyy-instrument-rezervnogo-kopirovaniya-gbak">официальный инструмент резервного копирования gbak</h2>

<h2 id="dlya-polucheniya-konsistentnogo-snimka">для получения консистентного снимка</h2>




<h2 id="variant-1-ostanovit-firebird-skopirovat-fayl">Вариант 1: Остановить Firebird, скопировать файл</h2>

sudo systemctl stop firebird

cp /var/lib/firebird/data/база.FDB /evidence/база_ORIGINAL.FDB

sha256sum /evidence/база_ORIGINAL.FDB > /evidence/база.sha256

sudo systemctl start firebird



<h2 id="variant-2-goryachiy-bekap-cherez-gbak-bez-ostanovki">Вариант 2: Горячий бэкап через gbak (без остановки)</h2>

gbak -backup -user SYSDBA -password masterkey \

  /var/lib/firebird/data/база.FDB \

  /evidence/база_backup.FBK



<h2 id="variant-3-nbackup-inkrementnyy-bekap-urovnya-0">Вариант 3: nbackup — инкрементный бэкап уровня 0</h2>

nbackup -backup 0 база.FDB /evidence/база_nbackup.nbk \

  -user SYSDBA -password masterkey

-- Снять полный бэкап с контрольной суммой

BACKUP DATABASE [ИмяБазы1С]

TO DISK = N'E:\Evidence\База1С_ORIGINAL.bak'

WITH CHECKSUM, COMPRESSION, STATS = 10;



-- Верифицировать резервную копию

RESTORE VERIFYONLY

FROM DISK = N'E:\Evidence\База1С_ORIGINAL.bak'

WITH CHECKSUM;



-- Дополнительно: снять бэкап журнала транзакций

-- (критично! В журнале хранятся все изменения)

BACKUP LOG [ИмяБазы1С]

TO DISK = N'E:\Evidence\База1С_LOG_ORIGINAL.bak'

WITH CHECKSUM;

Смещение  Размер  Описание

0x00      4       Заголовок страницы (тип + флаги + генерация)

0x04      2       Page type: 0x01 = Header page

0x08      4       Размер страницы

0x0C      4       ODS Major Version (On Disk Structure)

0x10      4       ODS Minor Version

0x18      8       Временна́я метка создания базы (ISC timestamp)

0x20      4       Page buffer (размер кэша при создании)

0x28      4       Номер следующей страницы транзакций (OIT)

0x2C      4       Номер старейшей активной транзакции (OAT)

0x30      4       Номер следующей транзакции (Next transaction)

0x38      8       Creation timestamp

pag_header    — заголовочная страница (одна на базу)

0x02  pag_pages     — инвентаризационная страница (PIP — Page Inventory Page)

0x03  pag_transactions — страница транзакций (TIP — Transaction Inventory Page)

0x04  pag_pointer   — страница указателей на записи данных (PPG — Pointer Page)

0x05  pag_data      — страница данных (записи таблиц)

0x06  pag_root      — корневая страница B-дерева индекса

0x07  pag_index     — страница индекса (не-корневая)

0x08  pag_blob      — страница BLOB-данных

0x09  pag_ids       — страница генераторов (sequences)

0x0A  pag_log       — страница WAL-журнала (устарело)

0x0D  pag_scns      — страница SCN (используется для backup)

#!/usr/bin/env python3

"""

fdb_raw_reader.py — низкоуровневый анализ FDB-файла Firebird

Работает без установленного Firebird, только с файлом образа

"""

import struct

import datetime

import sys

from pathlib import Path

from dataclasses import dataclass

from typing import Optional, List, Iterator





<h2 id="konstanty-firebird">Константы Firebird</h2>

PAGE_TYPES = {

    0x01: "Header",

    0x02: "PIP (Page Inventory)",

    0x03: "TIP (Transaction Inventory)",

    0x04: "PPG (Pointer Page)",

    0x05: "Data",

    0x06: "Index Root",

    0x07: "Index B-tree",

    0x08: "BLOB",

    0x09: "Generator",

    0x0D: "SCN",

}



TRANSACTION_STATES = {

    0: "Active",

    1: "Limbo",

    2: "Committed",

    3: "Dead (rolled back)",

}





@dataclass

class FDBHeader:

    """Заголовочная страница FDB-файла."""

    page_size: int

    ods_major: int

    ods_minor: int

    next_transaction: int

    oldest_transaction: int        # OIT — Oldest Interesting Transaction

    oldest_active_transaction: int # OAT — Oldest Active Transaction

    created_at: Optional[datetime.datetime]

    database_dialect: int

    sweep_interval: int

    read_only: bool

    forced_writes: bool





@dataclass

class DataPage:

    """Страница данных с записями."""

    page_number: int

    sequence: int        # Порядковый номер в PPG

    count: int           # Количество записей на странице

    records: List[bytes] # Сырые данные записей





class FDBRawReader:

    """

    Низкоуровневый читатель FDB-файла Firebird.

    Предназначен для криминалистического анализа без запуска СУБД.

    """



    def __init__(self, fdb_path: str):

        self.path = Path(fdb_path)

        if not self.path.exists():

            raise FileNotFoundError(f"FDB не найден: {fdb_path}")



        self.fdb_file = open(fdb_path, "rb")

        self.file_size = self.path.stat().st_size

        self.header = None

        self.page_size = 0



        # Читать заголовок

        self._read_header()



    def _read_header(self):

        """Разобрать заголовочную страницу (страница 0)."""

        self.fdb_file.seek(0)

        raw = self.fdb_file.read(1024)  # Читаем первые 1KB



        # Заголовок страницы (4 байта)

        page_type = struct.unpack_from("<H", raw, 0)[0] & 0xFF

        if page_type != 0x01:

            raise ValueError(f"Не корректный FDB: ожидался тип страницы 0x01, получен {hex(page_type)}")



        # Размер страницы (смещение 0x08)

        self.page_size = struct.unpack_from("<I", raw, 8)[0]

        if self.page_size not in (4096, 8192, 16384, 32768):

            # Попробовать определить по магии

            for ps in (4096, 8192, 16384, 32768):

                if self.file_size % ps == 0:

                    self.page_size = ps

                    break

            else:

                raise ValueError(f"Не удалось определить размер страницы: {self.page_size}")



        # ODS версия

        ods_major = struct.unpack_from("<H", raw, 12)[0]

        ods_minor = struct.unpack_from("<H", raw, 14)[0]



        # Номера транзакций (смещения зависят от версии ODS)

        # ODS 11.x (Firebird 2.x) — стандартная для 1С 8.x

        if ods_major >= 11:

            next_tx = struct.unpack_from("<I", raw, 0x28)[0]

            oit = struct.unpack_from("<I", raw, 0x2C)[0]

            oat = struct.unpack_from("<I", raw, 0x30)[0]

        else:

            next_tx = struct.unpack_from("<I", raw, 0x20)[0]

            oit = struct.unpack_from("<I", raw, 0x24)[0]

            oat = struct.unpack_from("<I", raw, 0x28)[0]



        # Временна́я метка создания (ISC timestamp)

        created_at = None

        try:

            ts_days = struct.unpack_from("<I", raw, 0x38)[0]

            ts_tenths = struct.unpack_from("<I", raw, 0x3C)[0]

            if ts_days > 0:

                # ISC timestamp: дни с 17 ноября 1858 года + доли секунды

                base = datetime.datetime(1858, 11, 17)

                delta = datetime.timedelta(

                    days=ts_days,

                    seconds=ts_tenths / 10

                )

                created_at = base + delta

        except Exception:

            pass



        # Диалект базы данных

        dialect = struct.unpack_from("<H", raw, 0x40)[0] if len(raw) > 0x42 else 3



        # Флаги

        flags = struct.unpack_from("<H", raw, 0x1C)[0] if len(raw) > 0x1E else 0

        read_only = bool(flags & 0x0010)

        forced_writes = bool(flags & 0x0001)



        sweep_interval = struct.unpack_from("<I", raw, 0x44)[0] if len(raw) > 0x48 else 20000



        self.header = FDBHeader(

            page_size=self.page_size,

            ods_major=ods_major,

            ods_minor=ods_minor,

            next_transaction=next_tx,

            oldest_transaction=oit,

            oldest_active_transaction=oat,

            created_at=created_at,

            database_dialect=dialect,

            read_only=read_only,

            forced_writes=forced_writes,

            sweep_interval=sweep_interval,

        )



    def get_page(self, page_number: int) -> bytes:

        """Прочитать конкретную страницу по номеру."""

        offset = page_number * self.page_size

        if offset + self.page_size > self.file_size:

            raise ValueError(f"Страница {page_number} выходит за пределы файла")

        self.fdb_file.seek(offset)

        return self.fdb_file.read(self.page_size)



    def get_page_type(self, page_data: bytes) -> int:

        """Определить тип страницы."""

        return struct.unpack_from("<H", page_data, 0)[0] & 0xFF



    def iter_pages(self, page_type_filter: Optional[int] = None) -> Iterator[tuple]:

        """

        Итератор по всем страницам FDB.



        Yields:

            (page_number, page_type, page_data)

        """

        total_pages = self.file_size // self.page_size

        for pnum in range(total_pages):

            try:

                page = self.get_page(pnum)

                ptype = self.get_page_type(page)

                if page_type_filter is None or ptype == page_type_filter:

                    yield pnum, ptype, page

            except Exception:

                continue



    def find_deleted_records(self) -> List[dict]:

        """

        Поиск удалённых/изменённых записей через MVCC back versions.



        В Firebird удалённые и изменённые записи остаются в страницах данных

        как 'back versions' до выполнения garbage collection.

        """

        deleted_records = []



        for page_num, page_type, page_data in self.iter_pages(0x05):  # Data pages

            try:

                records = self._parse_data_page(page_num, page_data)

                for rec in records:

                    if rec.get("is_back_version") or rec.get("is_deleted"):

                        deleted_records.append(rec)

            except Exception:

                continue



        return deleted_records



    def _parse_data_page(self, page_num: int, page_data: bytes) -> List[dict]:

        """

        Разобрать страницу данных Firebird.

        Извлекает как активные, так и back-version записи.

        """

        records = []



        # Заголовок страницы данных

        # Offset 0: page header (6 bytes)

        # Offset 6: sequence number

        # Offset 8: relation id (таблица)

        # Offset 10: count (число записей)



        if len(page_data) < 16:

            return records



        # Тип страницы и флаги

        ptype = struct.unpack_from("<H", page_data, 0)[0]

        seq_num = struct.unpack_from("<I", page_data, 6)[0]

        relation_id = struct.unpack_from("<H", page_data, 10)[0]

        record_count = struct.unpack_from("<H", page_data, 12)[0]



        if record_count == 0 or record_count > 500:  # Санитарная проверка

            return records



        # Таблица смещений записей начинается с offset 16

        # Каждая запись в offset table: 2 bytes offset от конца страницы

        for i in range(min(record_count, 200)):

            try:

                slot_offset = 16 + i * 2

                if slot_offset + 2 > len(page_data):

                    break



                rec_offset_from_end = struct.unpack_from("<H", page_data, slot_offset)[0]

                if rec_offset_from_end == 0:

                    continue



                # Смещение записи от начала страницы

                rec_offset = self.page_size - rec_offset_from_end



                if rec_offset < 16 or rec_offset >= self.page_size:

                    continue



                # Заголовок записи Firebird

                # Байт 0: флаги записи

                rec_flags = page_data[rec_offset] if rec_offset < len(page_data) else 0



                # Флаги записи Firebird

                # 0x10 = deleted (помечена на удаление)

                # 0x20 = back version (предыдущая версия)

                # 0x40 = gc_active (сборка мусора активна)

                is_deleted = bool(rec_flags & 0x10)

                is_back_version = bool(rec_flags & 0x20)



                # Длина записи

                if rec_offset + 4 <= len(page_data):

                    rec_length = struct.unpack_from("<H", page_data, rec_offset + 2)[0]

                else:

                    continue



                if rec_length == 0 or rec_length > self.page_size:

                    continue



                # Данные записи

                rec_data = page_data[rec_offset: rec_offset + rec_length]



                rec_info = {

                    "page_number": page_num,

                    "slot": i,

                    "relation_id": relation_id,

                    "sequence": seq_num,

                    "offset": rec_offset,

                    "length": rec_length,

                    "flags": hex(rec_flags),

                    "is_deleted": is_deleted,

                    "is_back_version": is_back_version,

                    "raw_data": rec_data.hex()[:128],  # Первые 64 байта в hex

                }

                records.append(rec_info)



            except Exception:

                continue



        return records



    def analyze_transaction_inventory(self) -> dict:

        """

        Анализ TIP (Transaction Inventory Pages) — состояния транзакций.

        Позволяет определить, какие транзакции были откачены (rolled back).

        """

        tx_analysis = {

            "total_transactions": self.header.next_transaction if self.header else 0,

            "oldest_active": self.header.oldest_active_transaction if self.header else 0,

            "oldest_interesting": self.header.oldest_transaction if self.header else 0,

            "rolled_back": [],

            "committed": 0,

            "active": 0,

        }



        # TIP содержит состояние каждой транзакции

        # Каждая транзакция = 2 бита: 00=active, 01=limbo, 10=committed, 11=dead

        for page_num, _, page_data in self.iter_pages(0x03):  # TIP pages

            try:

                # Первые 8 байт — заголовок страницы

                # Далее — packed bits (2 бита на транзакцию)

                data_start = 16

                bits_data = page_data[data_start:]



                for byte_idx, byte_val in enumerate(bits_data):

                    for bit_pair in range(4):

                        state = (byte_val >> (bit_pair * 2)) & 0x03

                        # Вычислить номер транзакции

                        tx_num = (page_num * (self.page_size - 16) * 4 +

                                  byte_idx * 4 + bit_pair)



                        if state == 2:

                            tx_analysis["committed"] += 1

                        elif state == 3:

                            tx_analysis["rolled_back"].append(tx_num)

                        elif state == 0:

                            tx_analysis["active"] += 1



            except Exception:

                continue



        return tx_analysis



    def print_summary(self):

        """Вывести сводную информацию о базе."""

        if not self.header:

            print("[!] Не удалось прочитать заголовок")

            return



        print(f"\n{'='*60}")

        print(f"АНАЛИЗ FDB-ФАЙЛА: {self.path.name}")

        print(f"{'='*60}")

        print(f"Размер файла:         {self.file_size:,} байт ({self.file_size/(1024**3):.2f} GB)")

        print(f"Размер страницы:      {self.header.page_size} байт")

        print(f"Всего страниц:        {self.file_size // self.header.page_size:,}")

        print(f"Версия ODS:           {self.header.ods_major}.{self.header.ods_minor}")

        print(f"Диалект SQL:          {self.header.database_dialect}")

        print(f"Создана:              {self.header.created_at or 'Не определено'}")

        print(f"Только чтение:        {'Да' if self.header.read_only else 'Нет'}")

        print(f"Следующая транзакция: {self.header.next_transaction:,}")

        print(f"Старейшая активная:   {self.header.oldest_active_transaction:,}")

        print(f"Старейшая интересная: {self.header.oldest_transaction:,}")



        # Разрыв между OIT и следующей транзакцией — показывает "мусор"

        tx_gap = self.header.next_transaction - self.header.oldest_transaction

        if tx_gap > 1000:

            print(f"\n[!] ВНИМАНИЕ: Разрыв транзакций {tx_gap:,} — база содержит")

            print(f"    значительный объём back versions (старых версий записей)")

            print(f"    Это благоприятно для криминалистического восстановления данных")



        # Подсчёт страниц по типам

        print(f"\nРаспределение страниц:")

        page_counts = {}

        for _, ptype, _ in self.iter_pages():

            type_name = PAGE_TYPES.get(ptype, f"Unknown({hex(ptype)})")

            page_counts[type_name] = page_counts.get(type_name, 0) + 1



        for ptype, count in sorted(page_counts.items(), key=lambda x: -x[1]):

            print(f"  {ptype:<30} {count:>8,}")



    def close(self):

        self.fdb_file.close()





if __name__ == "__main__":

    if len(sys.argv) < 2:

        print(f"Использование: {sys.argv[0]} <файл.FDB>")

        sys.exit(1)



    reader = FDBRawReader(sys.argv[1])

    reader.print_summary()



    print(f"\n[*] Поиск удалённых и изменённых записей...")

    deleted = reader.find_deleted_records()

    print(f"[+] Найдено back versions и удалённых записей: {len(deleted)}")



    if deleted:

        print(f"\nПервые 10 найденных:")

        for rec in deleted[:10]:

            status = "DELETED" if rec["is_deleted"] else "BACK_VERSION"

            print(f"  [{status}] Страница {rec['page_number']}, "

                  f"слот {rec['slot']}, "

                  f"таблица ID={rec['relation_id']}, "

                  f"размер={rec['length']}")



    reader.close()

Файл 1Cv8.1CD организован в виде страниц размером 512 байт.



Первая страница (страница 0) — корневая:

  Смещение 0x00-0x0F: Сигнатура "1CDBMSSV8" (или вариации)

  Смещение 0x10-0x13: Версия формата

  Смещение 0x14-0x17: Размер страницы

  Смещение 0x18-0x1F: Номер корневого объекта базы данных

  Смещение 0x20-0x3F: Дополнительные метаданные



Страницы данных:

  Первые 4 байта: флаги и состояние страницы

  Далее: данные объектов 1С в формате V8 Container

#!/usr/bin/env python3

"""

onec_1cd_reader.py — анализ файловой базы 1С (1Cv8.1CD)

Извлечение метаданных и данных без запуска 1С

"""

import struct

import os

import sys

import zlib

import datetime

from pathlib import Path

from typing import Optional, List, Tuple, Iterator





<h2 id="signatura-fayla-1s-faylovoy-bazy">Сигнатура файла 1С файловой базы</h2>

ONEC_SIGNATURE_V1 = b"1CDBMSSV8"

ONEC_SIGNATURE_V2 = b"1CDBMSSS"

PAGE_SIZE = 512  # стандартный размер страницы 1С 8.x





class OneCFileDB:

    """

    Анализатор файловой базы 1С:Предприятие 8.x (1Cv8.1CD).

    """



    def __init__(self, db_path: str, page_size: int = PAGE_SIZE):

        self.path = Path(db_path)

        if not self.path.exists():

            raise FileNotFoundError(f"База не найдена: {db_path}")



        self.page_size = page_size

        self.file_size = self.path.stat().st_size

        self.total_pages = self.file_size // self.page_size

        self._file = open(db_path, "rb")

        self.version = None

        self._validate_and_read_root()



    def _validate_and_read_root(self):

        """Прочитать и валидировать корневую страницу."""

        root = self._read_page(0)



        # Проверить сигнатуру

        if root[:9] == ONEC_SIGNATURE_V1:

            self.version = "V8"

        elif root[:8] == ONEC_SIGNATURE_V2:

            self.version = "V8S"

        else:

            # Попробовать как сырой файл без явной сигнатуры

            self.version = "UNKNOWN"



        # Прочитать параметры

        if len(root) >= 20:

            ver_raw = struct.unpack_from("<I", root, 0x10)[0]

            self.format_version = ver_raw



        print(f"[+] Открыта база 1С: {self.path.name}")

        print(f"[+] Версия формата: {self.version}")

        print(f"[+] Размер файла: {self.file_size:,} байт")

        print(f"[+] Всего страниц: {self.total_pages:,}")



    def _read_page(self, page_num: int) -> bytes:

        """Прочитать страницу по номеру."""

        offset = page_num * self.page_size

        if offset + self.page_size > self.file_size:

            raise ValueError(f"Страница {page_num} за пределами файла")

        self._file.seek(offset)

        return self._file.read(self.page_size)



    def iter_pages(self) -> Iterator[Tuple[int, bytes]]:

        """Итератор по всем страницам."""

        for i in range(self.total_pages):

            try:

                yield i, self._read_page(i)

            except Exception:

                continue



    def find_pages_with_signature(self, signature: bytes) -> List[int]:

        """

        Найти страницы, содержащие заданную сигнатуру.

        Полезно для поиска удалённых документов 1С.

        """

        found_pages = []

        for page_num, page_data in self.iter_pages():

            if signature in page_data:

                found_pages.append(page_num)

        return found_pages



    def extract_strings_from_page(

        self, page_data: bytes, min_length: int = 6

    ) -> List[str]:

        """Извлечь строки из страницы (UTF-16LE для 1С)."""

        import re

        strings = []



        # UTF-16LE строки (основная кодировка 1С 8.x)

        pattern_utf16 = re.compile(

            rb"(?:[\x20-\x7e\x00][\x00]){" + str(min_length).encode() + rb",}"

        )

        for m in pattern_utf16.finditer(page_data):

            try:

                s = m.group().decode("utf-16-le", errors="ignore").strip()

                if len(s) >= min_length and s.isprintable():

                    strings.append(s)

            except Exception:

                pass



        # ASCII строки (для метаданных и технических полей)

        pattern_ascii = re.compile(rb"[\x20-\x7e]{" + str(min_length).encode() + rb",}")

        for m in pattern_ascii.finditer(page_data):

            try:

                s = m.group().decode("ascii").strip()

                if len(s) >= min_length:

                    strings.append(s)

            except Exception:

                pass



        return list(set(strings))



    def scan_for_deleted_documents(self) -> List[dict]:

        """

        Сканировать базу на предмет удалённых документов 1С.



        1С помечает удалённые объекты специальными флагами в заголовке

        объекта. Эти объекты остаются в файле до реструктуризации.

        """

        deleted_docs = []



        # Сигнатуры начала объектов 1С в файловой базе

        # Первые байты контейнера V8: версия + тип + размер

        OBJECT_SIGNATURES = [

            b"\x01\x00\x00\x00",  # V8Object тип 1

            b"\x02\x00\x00\x00",  # V8Object тип 2

            b"\x08\x00\x00\x00",  # Объект документа

        ]



        for page_num, page_data in self.iter_pages():

            page_strings = self.extract_strings_from_page(page_data)



            # Искать признаки документов 1С в строках страницы

            doc_indicators = []

            for s in page_strings:

                # Типичные строки в заголовках документов 1С

                if any(keyword in s for keyword in [

                    "Документ.", "Справочник.", "РегистрНакопления.",

                    "Document.", "Catalog.", "AccumulationRegister.",

                    "Ссылка", "Дата", "Номер", "Организация",

                ]):

                    doc_indicators.append(s)



            if doc_indicators:

                deleted_docs.append({

                    "page_number": page_num,

                    "indicators": doc_indicators[:5],

                    "page_offset": page_num * self.page_size,

                })



        return deleted_docs



    def read_raw_block(self, offset: int, size: int) -> bytes:

        """Прочитать произвольный блок данных по смещению."""

        self._file.seek(offset)

        return self._file.read(size)



    def calculate_page_entropy(self, page_data: bytes) -> float:

        """Вычислить энтропию страницы (полезно для определения типа данных)."""

        import math

        if not page_data:

            return 0.0

        counts = [0] * 256

        for b in page_data:

            counts[b] += 1

        entropy = 0.0

        l = len(page_data)

        for c in counts:

            if c > 0:

                p = c / l

                entropy -= p * math.log2(p)

        return entropy



    def get_page_statistics(self) -> dict:

        """Статистика по содержимому страниц."""

        stats = {

            "empty_pages": 0,       # Страницы с нулевым содержимым

            "text_pages": 0,        # Страницы с читаемыми строками

            "binary_pages": 0,      # Бинарные страницы

            "high_entropy_pages": 0, # Возможно сжатые/зашифрованные

            "reused_pages": 0,      # Повторно используемые (б/у) страницы

        }



        for _, page_data in self.iter_pages():

            # Пустые страницы

            if page_data == b"\x00" * self.page_size:

                stats["empty_pages"] += 1

                continue



            # Энтропия

            entropy = self.calculate_page_entropy(page_data)

            if entropy > 7.5:

                stats["high_entropy_pages"] += 1

            elif entropy < 1.0:

                stats["binary_pages"] += 1



            # Наличие UTF-16 строк

            strings = self.extract_strings_from_page(page_data, min_length=8)

            if strings:

                stats["text_pages"] += 1

            else:

                stats["binary_pages"] += 1



            # Признаки б/у страниц (нулевой заголовок + непустые данные)

            header_bytes = page_data[:8]

            if header_bytes[:4] == b"\x00\x00\x00\x00" and \

               page_data[8:20] != b"\x00" * 12:

                stats["reused_pages"] += 1



        return stats



    def export_page_to_file(self, page_num: int, output_dir: str):

        """Экспортировать страницу в файл для анализа в hex-редакторе."""

        page = self._read_page(page_num)

        output = Path(output_dir) / f"page_{page_num:06d}.bin"

        output.parent.mkdir(parents=True, exist_ok=True)

        with open(output, "wb") as f:

            f.write(page)

        print(f"[+] Страница {page_num} → {output}")



    def close(self):

        self._file.close()





class OneCMetadataParser:

    """

    Парсер метаданных 1С из файловой базы.

    Позволяет прочитать структуру конфигурации без запуска 1С.

    """



    KNOWN_OBJECT_TYPES = {

        "Document": "Документ",

        "Catalog": "Справочник",

        "InformationRegister": "РегистрСведений",

        "AccumulationRegister": "РегистрНакопления",

        "AccountingRegister": "РегистрБухгалтерии",

        "BusinessProcess": "БизнесПроцесс",

        "Task": "Задача",

        "ChartOfCharacteristicTypes": "ПланВидовХарактеристик",

        "ChartOfAccounts": "ПланСчетов",

        "ExchangePlan": "ПланОбмена",

    }



    def __init__(self, db: OneCFileDB):

        self.db = db



    def find_document_tables(self) -> List[str]:

        """Найти таблицы документов в метаданных."""

        tables = []

        # В файловой базе метаданные хранятся в начальных страницах

        for page_num in range(min(100, self.db.total_pages)):

            _, page_data = next(

                ((n, d) for n, d in self.db.iter_pages() if n == page_num),

                (None, b"")

            )

            strings = self.db.extract_strings_from_page(page_data)

            for s in strings:

                for obj_type in self.KNOWN_OBJECT_TYPES:

                    if f"_{obj_type}" in s or s.startswith(obj_type):

                        if s not in tables:

                            tables.append(s)

        return tables

Заголовок DBF (32 байта):

  Байт 0:     Версия dBASE (0x03 = dBASE III)

  Байты 1-3:  Дата последнего изменения (YY MM DD)

  Байты 4-7:  Количество записей (DWORD, little-endian)

  Байты 8-9:  Размер заголовка в байтах (WORD)

  Байты 10-11: Размер одной записи в байтах (WORD)

  Байты 12-31: Зарезервировано



Дескрипторы полей (по 32 байта каждый):

  Байты 0-10:  Имя поля (null-terminated)

  Байт 11:    Тип поля (C=Char, N=Numeric, D=Date, L=Logical, M=Memo)

  Байты 12-15: Зарезервировано

  Байт 16:    Длина поля

  Байт 17:    Количество десятичных знаков

  Байты 18-31: Зарезервировано



После дескрипторов: байт 0x0D (маркер конца заголовка)



Записи данных:

  Первый байт каждой записи:

    0x20 (пробел) = запись активна

    0x2A (звёздочка *) = запись помечена на удаление

  Далее: данные записи (длина соответствует заголовку)

#!/usr/bin/env python3

"""

onec_dbf_forensics.py — криминалистический анализ DBF-файлов 1С 7.7

Читает активные И помеченные на удаление записи

"""

import struct

import os

import csv

import sys

import datetime

import codecs

from pathlib import Path

from typing import List, Optional, Tuple, Iterator

from dataclasses import dataclass, field





@dataclass

class DBFField:

    """Описание поля DBF."""

    name: str

    field_type: str

    length: int

    decimal: int





@dataclass

class DBFRecord:

    """Запись DBF (активная или удалённая)."""

    row_number: int

    is_deleted: bool

    fields: dict

    raw_data: bytes





class ForensicDBFReader:

    """

    Криминалистический читатель DBF, извлекающий все записи

    включая помеченные на удаление.

    """



    # Кодировки для 1С 7.7 (зависит от региональной настройки)

    ENCODINGS = ["cp866", "cp1251", "utf-8", "latin-1"]



    def __init__(self, dbf_path: str, encoding: str = "cp866"):

        self.path = Path(dbf_path)

        self.encoding = encoding

        self.fields: List[DBFField] = []

        self.record_count = 0

        self.header_size = 0

        self.record_size = 0

        self.last_modified = None

        self._data = self.path.read_bytes()

        self._parse_header()



    def _parse_header(self):

        """Разобрать заголовок DBF-файла."""

        if len(self._data) < 32:

            raise ValueError(f"Файл слишком мал для DBF: {len(self._data)} байт")



        # Версия

        version = self._data[0]

        if version not in (0x02, 0x03, 0x04, 0x05, 0x83, 0x8B):

            raise ValueError(f"Неизвестная версия DBF: {hex(version)}")



        # Дата изменения

        yy = self._data[1]

        mm = self._data[2]

        dd = self._data[3]

        try:

            year = 1900 + yy if yy >= 0 else 2000 + yy

            self.last_modified = datetime.date(year, mm, dd)

        except Exception:

            self.last_modified = None



        # Основные параметры

        self.record_count = struct.unpack_from("<I", self._data, 4)[0]

        self.header_size = struct.unpack_from("<H", self._data, 8)[0]

        self.record_size = struct.unpack_from("<H", self._data, 10)[0]



        # Парсинг дескрипторов полей

        self.fields = []

        offset = 32



        while offset < self.header_size - 1:

            if self._data[offset] == 0x0D:  # Маркер конца заголовка

                break

            if offset + 32 > len(self._data):

                break



            field_data = self._data[offset: offset + 32]



            # Имя поля (null-terminated, 11 байт)

            name_raw = field_data[:11]

            name = name_raw.rstrip(b"\x00").decode("ascii", errors="replace").strip()

            if not name:

                offset += 32

                continue



            field_type = chr(field_data[11])

            field_length = field_data[16]

            field_decimal = field_data[17]



            self.fields.append(DBFField(

                name=name,

                field_type=field_type,

                length=field_length,

                decimal=field_decimal,

            ))



            offset += 32



    def _decode_field(

        self, raw: bytes, field: DBFField

    ) -> Optional[str]:

        """Декодировать значение поля с учётом типа."""

        value = raw.rstrip(b" ")



        if not value:

            return None



        if field.field_type == "C":  # Character

            for enc in [self.encoding] + self.ENCODINGS:

                try:

                    return raw.decode(enc).strip()

                except Exception:

                    continue

            return raw.decode("latin-1").strip()



        elif field.field_type == "N":  # Numeric

            try:

                s = raw.decode("ascii").strip()

                if not s:

                    return None

                return s  # Вернуть как строку — точность важна

            except Exception:

                return None



        elif field.field_type == "D":  # Date YYYYMMDD

            try:

                s = raw.decode("ascii").strip()

                if len(s) == 8 and s.isdigit():

                    return f"{s[:4]}-{s[4:6]}-{s[6:8]}"

                return s

            except Exception:

                return None



        elif field.field_type == "L":  # Logical

            if value in (b"T", b"t", b"Y", b"y"):

                return "True"

            elif value in (b"F", b"f", b"N", b"n"):

                return "False"

            return None



        elif field.field_type == "M":  # Memo (ссылка на .DBT файл)

            try:

                block_num = struct.unpack_from("<I", raw[:4])[0]

                return f"MEMO_BLOCK:{block_num}"

            except Exception:

                return None



        else:

            return raw.decode("latin-1", errors="replace").strip()



    def iter_records(

        self,

        include_deleted: bool = True,

    ) -> Iterator[DBFRecord]:

        """

        Итератор по всем записям включая удалённые.



        Args:

            include_deleted: включать записи, помеченные на удаление

        """

        if self.record_size == 0:

            return



        # Подсчитать реальное число записей (может отличаться от заголовка!)

        available_space = len(self._data) - self.header_size

        actual_records = available_space // self.record_size



        # Если реальное число записей БОЛЬШЕ чем в заголовке — это важная находка

        if actual_records > self.record_count:

            print(f"[!] АНОМАЛИЯ: Заголовок объявляет {self.record_count} записей, "

                  f"но в файле помещается {actual_records} записей!")

            print(f"    Возможно, заголовок был намеренно изменён для сокрытия записей.")



        for row_num in range(actual_records):

            record_offset = self.header_size + row_num * self.record_size



            if record_offset + self.record_size > len(self._data):

                break



            raw_record = self._data[record_offset: record_offset + self.record_size]

            delete_flag = raw_record[0]

            is_deleted = (delete_flag == 0x2A)  # 0x2A = '*'



            if not include_deleted and is_deleted:

                continue



            # Разобрать поля записи

            field_offset = 1  # Пропустить байт флага удаления

            fields = {}



            for field in self.fields:

                if field_offset + field.length > len(raw_record):

                    break



                raw_value = raw_record[field_offset: field_offset + field.length]

                fields[field.name] = self._decode_field(raw_value, field)

                field_offset += field.length



            yield DBFRecord(

                row_number=row_num + 1,

                is_deleted=is_deleted,

                fields=fields,

                raw_data=raw_record,

            )



    def get_active_records(self) -> List[DBFRecord]:

        """Получить только активные (не удалённые) записи."""

        return [r for r in self.iter_records(include_deleted=False)]



    def get_deleted_records(self) -> List[DBFRecord]:

        """Получить только удалённые записи."""

        return [r for r in self.iter_records() if r.is_deleted]



    def export_to_csv(

        self,

        output_path: str,

        include_deleted: bool = True,

    ):

        """Экспортировать все записи в CSV."""

        path = Path(output_path)

        with open(path, "w", newline="", encoding="utf-8-sig") as f:

            fieldnames = ["#ROW", "#DELETED"] + [fld.name for fld in self.fields]

            writer = csv.DictWriter(f, fieldnames=fieldnames)

            writer.writeheader()



            active_count = 0

            deleted_count = 0



            for rec in self.iter_records(include_deleted=include_deleted):

                row = {"#ROW": rec.row_number, "#DELETED": "ДА" if rec.is_deleted else ""}

                row.update(rec.fields)

                writer.writerow(row)



                if rec.is_deleted:

                    deleted_count += 1

                else:

                    active_count += 1



        print(f"[+] Экспортировано в {output_path}")

        print(f"    Активных записей: {active_count}")

        print(f"    Удалённых записей: {deleted_count}")



    def print_info(self):

        """Вывести информацию о DBF-файле."""

        print(f"\n{'='*50}")

        print(f"DBF-файл: {self.path.name}")

        print(f"{'='*50}")

        print(f"Последнее изменение: {self.last_modified}")

        print(f"Записей (по заголовку): {self.record_count}")

        print(f"Размер записи: {self.record_size} байт")

        print(f"Размер заголовка: {self.header_size} байт")

        print(f"\nПоля ({len(self.fields)}):")

        for fld in self.fields:

            print(f"  {fld.name:<12} {fld.field_type}({fld.length})")



        # Статистика

        deleted = list(self.iter_records())

        del_count = sum(1 for r in deleted if r.is_deleted)

        print(f"\nАктивных записей: {len(deleted) - del_count}")

        print(f"Удалённых записей: {del_count}")



        if del_count > 0:

            pct = del_count / len(deleted) * 100 if deleted else 0

            print(f"[!] Внимание: {pct:.1f}% записей помечены на удаление!")





def analyze_1c77_journal(journal_path: str) -> dict:

    """

    Специализированный анализ журнала операций 1С 7.7 (1SJOURN.DBF).

    Журнал содержит все бухгалтерские проводки.

    """

    reader = ForensicDBFReader(journal_path)



    analysis = {

        "file": journal_path,

        "total_records": 0,

        "active_records": 0,

        "deleted_records": 0,

        "deleted_operations": [],

        "date_range": {"min": None, "max": None},

        "suspicious_patterns": [],

    }



    all_records = list(reader.iter_records(include_deleted=True))

    analysis["total_records"] = len(all_records)

    analysis["deleted_records"] = sum(1 for r in all_records if r.is_deleted)

    analysis["active_records"] = analysis["total_records"] - analysis["deleted_records"]



    dates = []

    for rec in all_records:

        # Поле даты в журнале 1С 7.7 обычно называется DDATE или DATE

        date_val = rec.fields.get("DDATE") or rec.fields.get("DATE") or rec.fields.get("D")

        if date_val:

            try:

                dates.append(date_val)

            except Exception:

                pass



        # Удалённые операции — особый интерес

        if rec.is_deleted:

            op_info = {

                "row": rec.row_number,

                "date": rec.fields.get("DDATE") or rec.fields.get("DATE"),

                "amount": rec.fields.get("SALDO") or rec.fields.get("SUMMA"),

                "account_dt": rec.fields.get("KODDT"),

                "account_ct": rec.fields.get("KODCT"),

                "document": rec.fields.get("DOCCODE"),

            }

            analysis["deleted_operations"].append(op_info)



    if dates:

        dates_sorted = sorted(filter(None, dates))

        if dates_sorted:

            analysis["date_range"]["min"] = dates_sorted[0]

            analysis["date_range"]["max"] = dates_sorted[-1]



    return analysis

Файловая база (1С 8.x):

  <Каталог базы>\1Cv8Log\20YYMMDD.lgp  — файлы по дням

  <Каталог базы>\1Cv8Log\20YYMMDD.lgd  — данные (строки)

  <Каталог базы>\1Cv8Log\1Cv8.lgf      — основной файл журнала



Клиент-серверная база (1С 8.x):

  Журнал хранится в базе данных СУБД в таблицах _EventLog*

  Или в отдельных файлах на сервере 1С



1С 7.7:

  <Каталог базы>\SYSLOG\*.LOG          — текстовые файлы

  <Каталог базы>\SYSLOG\JOURNAL.DBF    — структурированный журнал

Запись журнала (переменная длина):

  4 байта: Временна́я метка (секунды с 01.01.0001)

  4 байта: Транзакция ID

  4 байта: Тип события (код)

  4 байта: Уровень важности

  4 байта: Смещение строки в LGD (имя пользователя и другие строки)

  4 байта: Смещение строки (компьютер)

  4 байта: Смещение строки (метаданные - тип объекта)

  8 байта: GUID объекта (ссылка на документ/справочник)

  ...     : Дополнительные данные

#!/usr/bin/env python3

"""

onec_journal_analyzer.py — анализ журнала регистрации 1С 8.x

"""

import struct

import os

import sys

import datetime

import re

import sqlite3

import csv

from pathlib import Path

from typing import List, Optional, Iterator

from dataclasses import dataclass





<h2 id="kody-tipov-sobytiy-1s">Коды типов событий 1С</h2>

EVENT_TYPES = {

    "_$Access$_.Access":           "Доступ",

    "_$Access$_.AccessDenied":     "Отказ в доступе",

    "_$Data$_.Delete":             "Удаление данных",

    "_$Data$_.DeletePredefined":   "Удаление предопределённого",

    "_$Data$_.New":                "Добавление данных",

    "_$Data$_.Post":               "Проведение документа",

    "_$Data$_.TotalsPeriodUpdate": "Обновление итогов",

    "_$Data$_.Unpost":             "Отмена проведения",

    "_$Data$_.Update":             "Изменение данных",

    "_$InfoBase$_.ConfigUpdate":   "Изменение конфигурации",

    "_$InfoBase$_.DBCfgUpdate":    "Обновление базы данных",

    "_$Session$_.Authentication":  "Аутентификация",

    "_$Session$_.AuthenticationError": "Ошибка аутентификации",

    "_$Session$_.Finish":          "Завершение сеанса",

    "_$Session$_.Start":           "Начало сеанса",

    "_$Transaction$_.Begin":       "Начало транзакции",

    "_$Transaction$_.Commit":      "Фиксация транзакции",

    "_$Transaction$_.Rollback":    "Откат транзакции",

}



<h2 id="urovni-vazhnosti">Уровни важности</h2>

SEVERITY_LEVELS = {

    0: "Информация",

    1: "Предупреждение",

    2: "Ошибка",

    3: "Критическая ошибка",

}





@dataclass

class JournalEntry:

    """Запись журнала регистрации 1С."""

    timestamp: datetime.datetime

    user: str

    computer: str

    application: str

    event_type: str

    event_description: str

    severity: str

    metadata: str      # Тип объекта метаданных

    data_ref: str      # Ссылка на объект (GUID)

    comment: str

    transaction_id: Optional[str]

    session_id: Optional[int]





class OneCJournalAnalyzer:

    """

    Анализатор журнала регистрации 1С:Предприятие 8.x.

    Поддерживает как текстовый (LGF), так и SQL-форматы.

    """



    def __init__(self, log_dir: str):

        self.log_dir = Path(log_dir)

        self.entries: List[JournalEntry] = []

        self._load_journal()



    def _load_journal(self):

        """Загрузить журнал из файлов или базы данных."""

        # Поиск LGF-файлов (основной файл журнала)

        lgf_files = list(self.log_dir.glob("*.lgf")) + \

                    list(self.log_dir.glob("*.lgp"))



        if lgf_files:

            for lgf in lgf_files:

                self._load_lgf(lgf)

            print(f"[+] Загружено записей из LGF: {len(self.entries)}")

        else:

            print(f"[!] LGF-файлы не найдены в {self.log_dir}")

            print(f"[*] Для SQL-журнала используйте метод load_from_sql()")



    def _onec_timestamp_to_datetime(self, ts_raw: int) -> datetime.datetime:

        """

        Конвертировать временну́ю метку 1С в datetime.

        1С хранит время в 100-наносекундных интервалах с 01.01.0001

        """

        try:

            # 1С использует формат аналогичный .NET DateTime (тики с 01.01.0001)

            ONEC_EPOCH = datetime.datetime(1, 1, 1)

            delta = datetime.timedelta(microseconds=ts_raw // 10)

            result = ONEC_EPOCH + delta

            # Ограничить разумным диапазоном

            if 1990 <= result.year <= 2100:

                return result

        except Exception:

            pass

        return datetime.datetime.fromtimestamp(0)



    def _load_lgf(self, lgf_path: Path):

        """Загрузить журнал из LGF/LGP файла."""

        # LGF — текстовый формат (более старые версии 1С)

        # LGP — бинарный формат (современные версии)



        try:

            # Попробовать как текстовый файл (LGF)

            content = lgf_path.read_bytes()



            # Проверить текстовый формат

            try:

                text = content.decode("utf-8-sig")

                entries = self._parse_lgf_text(text)

                self.entries.extend(entries)

                return

            except UnicodeDecodeError:

                pass



            # Попробовать cp1251 (старые версии 1С)

            try:

                text = content.decode("cp1251")

                entries = self._parse_lgf_text(text)

                self.entries.extend(entries)

                return

            except Exception:

                pass



            # Бинарный формат (LGP)

            entries = self._parse_lgp_binary(content)

            self.entries.extend(entries)



        except Exception as e:

            print(f"[!] Ошибка загрузки {lgf_path}: {e}")



    def _parse_lgf_text(self, content: str) -> List[JournalEntry]:

        """

        Разобрать текстовый формат журнала регистрации 1С.

        Формат: {timestamp,user,computer,application,event,severity,metadata,ref,comment}

        """

        entries = []



        # Регулярное выражение для строки журнала

        # Формат немного меняется в зависимости от версии платформы

        PATTERNS = [

            # Формат 1С 8.3

            re.compile(

                r'\{(\d+),(\d+),(\w+),"([^"]*)","([^"]*)","([^"]*)","([^"]*)",'

                r'"([^"]*)","([^"]*)",(\d+),"([^"]*)","([^"]*)",'

                r'"([^"]*)","([^"]*)"\}'

            ),

            # Упрощённый формат

            re.compile(r'\{([^}]+)\}'),

        ]



        for line in content.splitlines():

            line = line.strip()

            if not line.startswith("{"):

                continue



            # Попробовать CSV-разбор (1С журнал близок к CSV в фигурных скобках)

            try:

                # Убрать фигурные скобки

                inner = line.strip("{}")

                # Разобрать как CSV

                import csv as csv_mod

                reader = csv_mod.reader([inner], quotechar='"', delimiter=',')

                parts = next(reader, [])



                if len(parts) >= 8:

                    # Временна́я метка (числовая в 100-нс интервалах)

                    try:

                        ts_raw = int(parts[0])

                        ts = self._onec_timestamp_to_datetime(ts_raw)

                    except Exception:

                        continue



                    entry = JournalEntry(

                        timestamp=ts,

                        user=self._clean_str(parts[3] if len(parts) > 3 else ""),

                        computer=self._clean_str(parts[4] if len(parts) > 4 else ""),

                        application=self._clean_str(parts[5] if len(parts) > 5 else ""),

                        event_type=self._clean_str(parts[6] if len(parts) > 6 else ""),

                        event_description=EVENT_TYPES.get(

                            self._clean_str(parts[6] if len(parts) > 6 else ""),

                            parts[6] if len(parts) > 6 else ""

                        ),

                        severity=SEVERITY_LEVELS.get(

                            int(parts[7]) if len(parts) > 7 and parts[7].isdigit() else 0,

                            "Информация"

                        ),

                        metadata=self._clean_str(parts[8] if len(parts) > 8 else ""),

                        data_ref=self._clean_str(parts[9] if len(parts) > 9 else ""),

                        comment=self._clean_str(parts[10] if len(parts) > 10 else ""),

                        transaction_id=parts[1] if len(parts) > 1 else None,

                        session_id=int(parts[2]) if len(parts) > 2 and parts[2].isdigit() else None,

                    )

                    entries.append(entry)



            except Exception:

                continue



        return entries



    def _parse_lgp_binary(self, data: bytes) -> List[JournalEntry]:

        """Базовый парсер бинарного LGP формата."""

        entries = []

        # LGP формат сложен и версионирован — здесь базовая реализация

        # Полная реализация требует изучения конкретной версии платформы



        # Ищем паттерны временны́х меток и строковых данных

        # Временна́я метка 1С: 8-байтное число, большое значение (>= 1е17)

        offset = 0

        while offset < len(data) - 8:

            try:

                ts_candidate = struct.unpack_from("<Q", data, offset)[0]

                # Проверить диапазон разумных временны́х меток

                # (2000-01-01 по 2030-01-01 в тиках с 01.01.0001)

                MIN_TS = 630_822_816_000_000_000

                MAX_TS = 638_800_000_000_000_000

                if MIN_TS <= ts_candidate <= MAX_TS:

                    ts = self._onec_timestamp_to_datetime(ts_candidate)

                    entry = JournalEntry(

                        timestamp=ts,

                        user="[BINARY_RECOVERED]",

                        computer="",

                        application="",

                        event_type="[RAW]",

                        event_description="Запись восстановлена из бинарного LGP",

                        severity="Информация",

                        metadata="",

                        data_ref="",

                        comment=f"Offset: {hex(offset)}",

                        transaction_id=None,

                        session_id=None,

                    )

                    entries.append(entry)

                    offset += 8

                else:

                    offset += 1

            except Exception:

                offset += 1



        return entries



    def _clean_str(self, s: str) -> str:

        """Очистить строку от служебных символов."""

        return s.strip().strip('"').strip("'")



    def load_from_sql(self, conn_string: str):

        """

        Загрузить журнал регистрации из SQL-базы.

        Для клиент-серверных баз 1С на MS SQL / PostgreSQL.

        """

        # Таблицы журнала регистрации в MS SQL

        MSSQL_QUERY = """

            SELECT

                el.date AS Timestamp,

                u.Name AS UserName,

                c.Name AS ComputerName,

                a.Name AS ApplicationName,

                ev.Name AS EventType,

                el.comment AS Comment,

                m.Name AS Metadata,

                CONVERT(VARCHAR(36), el.DataUUID) AS DataRef,

                el.TransactionID,

                el.SessionID

            FROM dbo._EventLog el

            LEFT JOIN dbo._EventLogUsers u ON el.UserCode = u.Code

            LEFT JOIN dbo._EventLogComputers c ON el.ComputerCode = c.Code

            LEFT JOIN dbo._EventLogApplications a ON el.AppCode = a.Code

            LEFT JOIN dbo._EventLogEvents ev ON el.Event = ev.Code

            LEFT JOIN dbo._EventLogMetadata m ON el.MetadataCode = m.Code

            ORDER BY el.date

        """

        print(f"[*] Загрузка журнала из SQL: {conn_string[:50]}...")

        # Реализация зависит от СУБД (pyodbc для MSSQL, psycopg2 для PG)

        print(f"[*] Используйте pyodbc/psycopg2 для подключения к СУБД")



    def get_critical_events(self) -> List[JournalEntry]:

        """

        Извлечь критически важные события для расследования:

        удаления, отмены проведения, изменения.

        """

        CRITICAL_EVENTS = {

            "_$Data$_.Delete",

            "_$Data$_.Unpost",

            "_$Data$_.Update",

            "_$Data$_.Post",

            "_$InfoBase$_.ConfigUpdate",

        }



        return [

            e for e in self.entries

            if e.event_type in CRITICAL_EVENTS

        ]



    def detect_suspicious_patterns(self) -> List[dict]:

        """

        Обнаружить подозрительные паттерны в журнале регистрации.

        """

        findings = []



        # 1. Работа в нерабочее время (после 20:00 и до 7:00)

        off_hours = [

            e for e in self.entries

            if e.timestamp.hour >= 20 or e.timestamp.hour < 7

        ]

        if off_hours:

            findings.append({

                "type": "OFF_HOURS_ACTIVITY",

                "severity": "MEDIUM",

                "description": f"Активность в нерабочее время: {len(off_hours)} событий",

                "events": off_hours[:5],

            })



        # 2. Массовые удаления (более 10 удалений за один сеанс)

        from collections import defaultdict

        deletions_by_session = defaultdict(list)

        for e in self.entries:

            if e.event_type == "_$Data$_.Delete" and e.session_id:

                deletions_by_session[e.session_id].append(e)



        for session_id, deletions in deletions_by_session.items():

            if len(deletions) > 10:

                findings.append({

                    "type": "MASS_DELETION",

                    "severity": "HIGH",

                    "description": f"Массовое удаление: {len(deletions)} удалений в сеансе {session_id}",

                    "events": deletions[:3],

                })



        # 3. Массовые отмены проведения

        unposts_by_session = defaultdict(list)

        for e in self.entries:

            if e.event_type == "_$Data$_.Unpost" and e.session_id:

                unposts_by_session[e.session_id].append(e)



        for session_id, unposts in unposts_by_session.items():

            if len(unposts) > 5:

                findings.append({

                    "type": "MASS_UNPOST",

                    "severity": "HIGH",

                    "description": f"Массовая отмена проведения: {len(unposts)} в сеансе {session_id}",

                    "events": unposts[:3],

                })



        # 4. Изменение конфигурации — подозрительно в рабочее время

        config_changes = [

            e for e in self.entries

            if e.event_type == "_$InfoBase$_.ConfigUpdate"

        ]

        for cc in config_changes:

            findings.append({

                "type": "CONFIG_CHANGE",

                "severity": "MEDIUM",

                "description": f"Изменение конфигурации: {cc.timestamp} пользователем {cc.user}",

                "events": [cc],

            })



        # 5. Аутентификация под несколькими пользователями с одного ПК

        sessions_by_computer = defaultdict(set)

        for e in self.entries:

            if e.event_type == "_$Session$_.Authentication" and e.computer:

                sessions_by_computer[e.computer].add(e.user)



        for computer, users in sessions_by_computer.items():

            if len(users) > 3:

                findings.append({

                    "type": "MULTIPLE_USERS_ONE_PC",

                    "severity": "MEDIUM",

                    "description": f"С компьютера '{computer}' входили {len(users)} пользователей: {', '.join(list(users)[:5])}",

                    "events": [],

                })



        # 6. Ошибки аутентификации (возможный подбор пароля)

        auth_errors = [

            e for e in self.entries

            if e.event_type == "_$Session$_.AuthenticationError"

        ]

        if len(auth_errors) > 5:

            findings.append({

                "type": "AUTH_ERRORS",

                "severity": "MEDIUM",

                "description": f"Ошибки аутентификации: {len(auth_errors)} попыток",

                "events": auth_errors[:3],

            })



        return findings



    def build_user_activity_report(self) -> dict:

        """Построить отчёт об активности пользователей."""

        from collections import defaultdict, Counter



        user_stats = defaultdict(lambda: {

            "total_events": 0,

            "deletions": 0,

            "modifications": 0,

            "unposts": 0,

            "sessions": set(),

            "computers": set(),

            "first_activity": None,

            "last_activity": None,

            "events_by_hour": Counter(),

        })



        for entry in self.entries:

            user = entry.user or "Неизвестный"

            stats = user_stats[user]

            stats["total_events"] += 1



            if entry.event_type == "_$Data$_.Delete":

                stats["deletions"] += 1

            elif entry.event_type == "_$Data$_.Update":

                stats["modifications"] += 1

            elif entry.event_type == "_$Data$_.Unpost":

                stats["unposts"] += 1



            if entry.session_id:

                stats["sessions"].add(entry.session_id)

            if entry.computer:

                stats["computers"].add(entry.computer)



            stats["events_by_hour"][entry.timestamp.hour] += 1



            if stats["first_activity"] is None or entry.timestamp < stats["first_activity"]:

                stats["first_activity"] = entry.timestamp

            if stats["last_activity"] is None or entry.timestamp > stats["last_activity"]:

                stats["last_activity"] = entry.timestamp



        # Конвертировать для JSON-сериализации

        result = {}

        for user, stats in user_stats.items():

            result[user] = {

                "total_events": stats["total_events"],

                "deletions": stats["deletions"],

                "modifications": stats["modifications"],

                "unposts": stats["unposts"],

                "unique_sessions": len(stats["sessions"]),

                "computers": list(stats["computers"]),

                "first_activity": stats["first_activity"].isoformat() if stats["first_activity"] else None,

                "last_activity": stats["last_activity"].isoformat() if stats["last_activity"] else None,

                "peak_activity_hour": stats["events_by_hour"].most_common(1)[0][0]

                    if stats["events_by_hour"] else None,

            }



        return result



    def export_to_csv(self, output_path: str):

        """Экспорт журнала в CSV."""

        with open(output_path, "w", newline="", encoding="utf-8-sig") as f:

            fieldnames = [

                "timestamp", "user", "computer", "application",

                "event_type", "event_description", "severity",

                "metadata", "data_ref", "comment",

                "transaction_id", "session_id"

            ]

            writer = csv.DictWriter(f, fieldnames=fieldnames)

            writer.writeheader()

            for entry in self.entries:

                writer.writerow({

                    "timestamp": entry.timestamp.strftime("%Y-%m-%d %H:%M:%S"),

                    "user": entry.user,

                    "computer": entry.computer,

                    "application": entry.application,

                    "event_type": entry.event_type,

                    "event_description": entry.event_description,

                    "severity": entry.severity,

                    "metadata": entry.metadata,

                    "data_ref": entry.data_ref,

                    "comment": entry.comment,

                    "transaction_id": entry.transaction_id,

                    "session_id": entry.session_id,

                })

        print(f"[+] Журнал экспортирован: {output_path} ({len(self.entries)} записей)")

#!/usr/bin/env python3

"""

recover_deleted_dbf.py — восстановление удалённых записей из DBF-файлов 1С 7.7

"""

import struct

import csv

import sys

from pathlib import Path

from typing import List, Dict

from onec_dbf_forensics import ForensicDBFReader





def recover_journal_operations(

    journal_dir: str,

    output_dir: str,

) -> dict:

    """

    Восстановление удалённых операций из журнала 1С 7.7.

    Анализирует связанные DBF-файлы: 1SJOURN + 1SCRDOC + документы.

    """

    jdir = Path(journal_dir)

    odir = Path(output_dir)

    odir.mkdir(parents=True, exist_ok=True)



    recovery_report = {

        "found_deleted": 0,

        "recovered_operations": [],

        "document_references": [],

        "errors": [],

    }



    # Шаг 1: Анализ основного журнала операций (1SJOURN.DBF)

    journal_path = jdir / "1SJOURN.DBF"

    if not journal_path.exists():

        # Попробовать другие варианты имени

        for alt in ["1SJOURN.dbf", "JOURNAL.DBF", "SC1SJOURN.DBF"]:

            if (jdir / alt).exists():

                journal_path = jdir / alt

                break



    if journal_path.exists():

        print(f"[*] Анализ журнала: {journal_path.name}")

        reader = ForensicDBFReader(str(journal_path))



        deleted_ops = reader.get_deleted_records()

        recovery_report["found_deleted"] = len(deleted_ops)

        print(f"[+] Удалённых операций: {len(deleted_ops)}")



        for rec in deleted_ops:

            op = {

                "row_number": rec.row_number,

                "status": "УДАЛЕНО",

                "date": rec.fields.get("DDATE") or rec.fields.get("DATE"),

                "number": rec.fields.get("DOCNO") or rec.fields.get("NUMBER"),

                "debit": rec.fields.get("KODDT") or rec.fields.get("ACCDT"),

                "credit": rec.fields.get("KODCT") or rec.fields.get("ACCCT"),

                "amount": rec.fields.get("SALDO") or rec.fields.get("SUMMA"),

                "document_type": rec.fields.get("DOCCODE") or rec.fields.get("DOCTYPE"),

                "document_id": rec.fields.get("DOCID") or rec.fields.get("IDDOC"),

                "description": rec.fields.get("SUBKONTO1") or rec.fields.get("COMMENT"),

            }

            recovery_report["recovered_operations"].append(op)



        # Экспорт удалённых операций

        if deleted_ops:

            export_path = odir / "recovered_journal_operations.csv"

            reader.export_to_csv(str(export_path), include_deleted=True)

            print(f"[+] Экспортировано: {export_path}")



    # Шаг 2: Анализ связей документов (1SCRDOC.DBF)

    crdoc_path = jdir / "1SCRDOC.DBF"

    if crdoc_path.exists():

        print(f"[*] Анализ связей документов: {crdoc_path.name}")

        crdoc_reader = ForensicDBFReader(str(crdoc_path))

        deleted_links = crdoc_reader.get_deleted_records()



        if deleted_links:

            print(f"[+] Удалённых связей документов: {len(deleted_links)}")

            for link in deleted_links:

                recovery_report["document_references"].append({

                    "status": "УДАЛЕНО",

                    "parent_doc": link.fields.get("PARENTID") or link.fields.get("IDDOC"),

                    "child_doc": link.fields.get("CHILDID") or link.fields.get("IDDOCDEF"),

                    "row": link.row_number,

                })



    # Шаг 3: Поиск документов в DH*.DBF и DT*.DBF файлах

    print(f"\n[*] Поиск удалённых документов в файлах DH/DT...")

    for dbf_file in sorted(jdir.glob("DH*.DBF")):

        try:

            doc_reader = ForensicDBFReader(str(dbf_file))

            deleted = doc_reader.get_deleted_records()

            if deleted:

                print(f"  [+] {dbf_file.name}: {len(deleted)} удалённых записей")

                export_path = odir / f"recovered_{dbf_file.stem}.csv"

                doc_reader.export_to_csv(str(export_path), include_deleted=True)

        except Exception as e:

            recovery_report["errors"].append(f"{dbf_file.name}: {e}")



    # Шаг 4: Регистры накопления (RA*.DBF)

    print(f"\n[*] Поиск удалённых движений в регистрах...")

    for dbf_file in sorted(jdir.glob("RA*.DBF")):

        try:

            reg_reader = ForensicDBFReader(str(dbf_file))

            deleted = reg_reader.get_deleted_records()

            if deleted:

                print(f"  [+] {dbf_file.name}: {len(deleted)} удалённых движений")

                export_path = odir / f"recovered_{dbf_file.stem}.csv"

                reg_reader.export_to_csv(str(export_path), include_deleted=True)

        except Exception as e:

            recovery_report["errors"].append(f"{dbf_file.name}: {e}")



    # Сводный отчёт

    print(f"\n{'='*50}")

    print(f"ИТОГИ ВОССТАНОВЛЕНИЯ")

    print(f"{'='*50}")

    print(f"Удалённых операций в журнале:  {recovery_report['found_deleted']}")

    print(f"Удалённых связей документов:   {len(recovery_report['document_references'])}")

    print(f"Ошибок обработки:              {len(recovery_report['errors'])}")



    return recovery_report





def carve_deleted_from_dbf_binary(

    dbf_path: str,

    output_path: str,

) -> int:

    """

    Карвинг удалённых записей из DBF файла.

    Дополнительный метод: ищет записи по сигнатуре даже в повреждённых файлах.

    """

    data = Path(dbf_path).read_bytes()



    # Читать заголовок для получения размера записи

    if len(data) < 32:

        return 0



    record_size = struct.unpack_from("<H", data, 10)[0]

    header_size = struct.unpack_from("<H", data, 8)[0]



    if record_size == 0 or record_size > 10000:

        print(f"[!] Некорректный размер записи: {record_size}")

        return 0



    carved_count = 0

    records_section = data[header_size:]



    with open(output_path, "wb") as f:

        offset = 0

        while offset + record_size <= len(records_section):

            record = records_section[offset: offset + record_size]



            # Флаг удалённой записи

            if record[0] == 0x2A:  # '*'

                # Проверить что это не случайные данные

                # (запись должна содержать читаемые данные)

                printable_bytes = sum(1 for b in record[1:] if 0x20 <= b <= 0x7E)

                if printable_bytes > record_size * 0.3:

                    f.write(record)

                    carved_count += 1



            offset += record_size



    print(f"[+] Карвинг {dbf_path}: извлечено {carved_count} удалённых записей → {output_path}")

    return carved_count

#!/usr/bin/env python3

"""

detect_backdating.py — выявление документов, проведённых задним числом в 1С

"""

import datetime

from typing import List, Dict, Tuple

from dataclasses import dataclass

from collections import defaultdict





@dataclass

class DocumentRecord:

    """Запись о документе из базы 1С."""

    doc_id: str

    doc_type: str

    doc_number: str

    doc_date: datetime.datetime      # Дата документа (видимая пользователю)

    created_at: datetime.datetime    # Дата создания в базе

    modified_at: datetime.datetime   # Дата последнего изменения

    posted_at: datetime.datetime     # Дата проведения

    user: str

    amount: float





class BackdatingDetector:

    """

    Детектор документов, проведённых задним числом.

    Использует временны́е метаданные из базы данных.

    """



    def __init__(self, documents: List[DocumentRecord]):

        self.documents = documents

        self.findings = []



    def detect_all(self) -> List[dict]:

        """Запустить все проверки."""

        self.findings = []



        self._check_document_date_vs_creation()

        self._check_sequence_anomalies()

        self._check_modification_after_period_close()

        self._check_clustered_backdating()

        self._check_amount_rounding_patterns()



        return self.findings



    def _check_document_date_vs_creation(self, threshold_days: int = 7):

        """

        Проверка 1: Дата документа значительно раньше даты создания в базе.

        Если документ создан 15 февраля, а датирован 1 января —

        это признак ввода задним числом.

        """

        for doc in self.documents:

            try:

                delta = (doc.created_at - doc.doc_date).days

                if delta > threshold_days:

                    self.findings.append({

                        "type": "BACKDATING_SUSPECTED",

                        "severity": "HIGH" if delta > 30 else "MEDIUM",

                        "document": doc.doc_id,

                        "doc_type": doc.doc_type,

                        "doc_number": doc.doc_number,

                        "doc_date": doc.doc_date.strftime("%Y-%m-%d"),

                        "created_at": doc.created_at.strftime("%Y-%m-%d %H:%M:%S"),

                        "delta_days": delta,

                        "user": doc.user,

                        "description": (

                            f"Документ {doc.doc_type} №{doc.doc_number} "

                            f"датирован {doc.doc_date.strftime('%d.%m.%Y')}, "

                            f"но создан в базе {delta} дней спустя "

                            f"({doc.created_at.strftime('%d.%m.%Y %H:%M')}). "

                            f"Пользователь: {doc.user}"

                        ),

                    })

            except Exception:

                continue



    def _check_sequence_anomalies(self):

        """

        Проверка 2: Аномалии в нумерации документов.

        Если документы с последовательными номерами имеют даты в обратном порядке —

        это признак вставки задним числом.

        """

        # Группировать документы по типу

        docs_by_type = defaultdict(list)

        for doc in self.documents:

            docs_by_type[doc.doc_type].append(doc)



        for doc_type, docs in docs_by_type.items():

            # Сортировать по номеру

            try:

                sorted_docs = sorted(

                    docs,

                    key=lambda d: int(''.join(filter(str.isdigit, d.doc_number or "0")) or "0")

                )

            except Exception:

                continue



            for i in range(1, len(sorted_docs)):

                prev = sorted_docs[i - 1]

                curr = sorted_docs[i]



                # Дата должна возрастать или оставаться той же при увеличении номера

                if curr.doc_date < prev.doc_date:

                    self.findings.append({

                        "type": "SEQUENCE_DATE_ANOMALY",

                        "severity": "MEDIUM",

                        "document": curr.doc_id,

                        "doc_type": doc_type,

                        "description": (

                            f"Нарушение хронологии в {doc_type}: "

                            f"документ №{curr.doc_number} "

                            f"({curr.doc_date.strftime('%d.%m.%Y')}) "

                            f"идёт после №{prev.doc_number} "

                            f"({prev.doc_date.strftime('%d.%m.%Y')})"

                        ),

                    })



    def _check_modification_after_period_close(self):

        """

        Проверка 3: Изменение документов после закрытия периода.

        В большинстве организаций закрытые периоды не должны изменяться.

        """

        # Определить закрытые периоды (обычно прошлый квартал и ранее)

        today = datetime.datetime.now()

        # Считаем текущий квартал открытым, предыдущий — потенциально закрытым

        current_quarter_start = datetime.datetime(

            today.year,

            ((today.month - 1) // 3) * 3 + 1,

            1

        )



        for doc in self.documents:

            # Документ за закрытый период

            if doc.doc_date < current_quarter_start:

                # Но изменён в открытом периоде

                if (doc.modified_at >= current_quarter_start or

                        doc.posted_at >= current_quarter_start):

                    self.findings.append({

                        "type": "MODIFICATION_CLOSED_PERIOD",

                        "severity": "HIGH",

                        "document": doc.doc_id,

                        "doc_type": doc.doc_type,

                        "doc_number": doc.doc_number,

                        "doc_date": doc.doc_date.strftime("%Y-%m-%d"),

                        "modified_at": doc.modified_at.strftime("%Y-%m-%d %H:%M:%S"),

                        "user": doc.user,

                        "description": (

                            f"Документ {doc.doc_type} №{doc.doc_number} "

                            f"от {doc.doc_date.strftime('%d.%m.%Y')} "

                            f"изменён/проведён {doc.modified_at.strftime('%d.%m.%Y %H:%M')} "

                            f"— после возможного закрытия периода"

                        ),

                    })



    def _check_clustered_backdating(self, window_hours: int = 2, min_count: int = 5):

        """

        Проверка 4: Кластерная датировка задним числом.

        Много документов с разными датами, введённых за короткий промежуток времени —

        классический признак массового ввода задним числом.

        """

        # Группировать документы по времени создания (окно 2 часа)

        time_clusters = defaultdict(list)

        for doc in self.documents:

            # Округлить до окна

            window_start = doc.created_at.replace(

                minute=(doc.created_at.minute // 30) * 30,

                second=0, microsecond=0

            )

            time_clusters[window_start].append(doc)



        for window_start, docs in time_clusters.items():

            if len(docs) >= min_count:

                # Проверить разброс дат документов

                doc_dates = [d.doc_date for d in docs]

                date_range = (max(doc_dates) - min(doc_dates)).days



                if date_range > 30:

                    self.findings.append({

                        "type": "CLUSTERED_BACKDATING",

                        "severity": "HIGH",

                        "description": (

                            f"За {window_hours}-часовое окно {window_start.strftime('%d.%m.%Y %H:%M')} "

                            f"введено {len(docs)} документов с разбросом дат {date_range} дней. "

                            f"Возможно массовое введение задним числом."

                        ),

                        "count": len(docs),

                        "date_range_days": date_range,

                        "window": window_start.strftime("%Y-%m-%d %H:%M"),

                        "users": list(set(d.user for d in docs)),

                    })



    def _check_amount_rounding_patterns(self):

        """

        Проверка 5: Подозрительные паттерны сумм (часто встречаются при хищениях).

        """

        amounts = [doc.amount for doc in self.documents if doc.amount > 0]



        if not amounts:

            return



        # Паттерн "круглых сумм" — хищения часто оформляются на круглые суммы

        round_amounts = [a for a in amounts if a % 1000 == 0 and a > 0]

        if len(round_amounts) > len(amounts) * 0.5 and len(amounts) > 10:

            self.findings.append({

                "type": "SUSPICIOUS_AMOUNT_PATTERN",

                "severity": "LOW",

                "description": (

                    f"{len(round_amounts)} из {len(amounts)} документов "

                    f"({len(round_amounts)/len(amounts)*100:.0f}%) "

                    f"имеют суммы, кратные 1000. "

                    f"Возможен искусственный характер операций."

                ),

            })



        # Суммы чуть ниже лимита авторизации (например, 499 900 при лимите 500 000)

        # Без знания конкретных лимитов — ищем кластеры около "психологических" рубежей

        THRESHOLDS = [100_000, 500_000, 1_000_000, 5_000_000, 10_000_000]

        for threshold in THRESHOLDS:

            near_threshold = [

                a for a in amounts

                if threshold * 0.95 <= a <= threshold * 0.999

            ]

            if len(near_threshold) >= 3:

                self.findings.append({

                    "type": "THRESHOLD_AVOIDANCE",

                    "severity": "HIGH",

                    "description": (

                        f"{len(near_threshold)} транзакций с суммами "

                        f"в диапазоне 95-99.9% от {threshold:,} руб. "

                        f"Возможное намеренное дробление для обхода лимитов авторизации."

                    ),

                    "threshold": threshold,

                    "count": len(near_threshold),

                })

#!/usr/bin/env python3

"""

onec_forensic_toolkit.py — полный набор инструментов криминалистического анализа 1С

Точка входа для автоматизированного анализа

"""

import sys

import os

import json

import argparse

import datetime

import hashlib

from pathlib import Path

from typing import Optional





def detect_base_type(base_path: str) -> str:

    """Автоматически определить тип базы данных 1С."""

    path = Path(base_path)



    if path.is_file():

        suffix = path.suffix.lower()

        if suffix == ".fdb":

            return "firebird"

        elif suffix in (".1cd", ".dt"):

            return "file_1cd"

        elif suffix == ".dbf":

            return "dbf_77"

        elif suffix in (".mdf", ".bak"):

            return "mssql"



    elif path.is_dir():

        # Определить по содержимому директории

        files = [f.name.upper() for f in path.iterdir() if f.is_file()]



        if "1CV8.1CD" in files:

            return "file_1cd"

        elif any(f.endswith(".FDB") for f in files):

            return "firebird"

        elif "1SJOURN.DBF" in files or any(f.startswith("1S") and f.endswith(".DBF")

                                            for f in files):

            return "dbf_77"

        elif "1CV8LOG" in [f.name.upper() for f in path.iterdir() if f.is_dir()]:

            return "file_1cd"  # Есть каталог журнала регистрации



    return "unknown"





def compute_file_hashes(file_path: str) -> dict:

    """Вычислить хэши файла для акта."""

    hashes = {"md5": hashlib.md5(), "sha1": hashlib.sha1(), "sha256": hashlib.sha256()}

    with open(file_path, "rb") as f:

        for chunk in iter(lambda: f.read(65536), b""):

            for h in hashes.values():

                h.update(chunk)

    return {k: v.hexdigest() for k, v in hashes.items()}





class OneCForensicToolkit:

    """

    Главный класс инструментария криминалистического анализа 1С.

    """



    def __init__(self, base_path: str, output_dir: str, case_id: str):

        self.base_path = Path(base_path)

        self.output_dir = Path(output_dir) / case_id

        self.case_id = case_id

        self.base_type = detect_base_type(str(base_path))

        self.start_time = datetime.datetime.utcnow()

        self.report = {

            "case_id": case_id,

            "base_path": str(base_path),

            "base_type": self.base_type,

            "analysis_start": self.start_time.isoformat(),

            "findings": [],

            "statistics": {},

        }



        # Создать структуру директорий

        for subdir in ["evidence_hashes", "recovered_data", "journal_analysis",

                       "backdating_report", "reports"]:

            (self.output_dir / subdir).mkdir(parents=True, exist_ok=True)



        print(f"""

╔═══════════════════════════════════════════════════╗

║    1С FORENSIC TOOLKIT — Криминалистика 1С       ║

║    Кейс: {case_id:<42}║

║    Тип базы: {self.base_type:<39}║

╚═══════════════════════════════════════════════════╝

        """)



    def step_1_hash_and_document(self) -> dict:

        """Шаг 1: Документирование исходных файлов базы."""

        print("[*] Шаг 1: Документирование и хэширование исходных файлов")



        files_to_hash = []

        if self.base_path.is_file():

            files_to_hash.append(self.base_path)

        elif self.base_path.is_dir():

            files_to_hash = [

                f for f in self.base_path.rglob("*")

                if f.is_file() and f.suffix.lower() in

                {".fdb", ".1cd", ".dbf", ".lgf", ".lgp", ".lgd",

                 ".mdf", ".ldf", ".bak", ".dt"}

            ]



        hash_results = []

        for f in files_to_hash:

            try:

                hashes = compute_file_hashes(str(f))

                stat = f.stat()

                result = {

                    "file": str(f.relative_to(self.base_path) if self.base_path.is_dir()

                               else f.name),

                    "size": stat.st_size,

                    "modified": datetime.datetime.fromtimestamp(stat.st_mtime).isoformat(),

                    "created": datetime.datetime.fromtimestamp(stat.st_ctime).isoformat(),

                    **hashes,

                }

                hash_results.append(result)

                print(f"  [+] {f.name}: SHA256={hashes['sha256'][:16]}...")

            except Exception as e:

                print(f"  [!] Ошибка хэширования {f.name}: {e}")



        # Сохранить манифест хэшей

        manifest_path = self.output_dir / "evidence_hashes" / "file_manifest.json"

        with open(manifest_path, "w", encoding="utf-8") as f:

            json.dump({

                "generated_at": self.start_time.isoformat(),

                "case_id": self.case_id,

                "files": hash_results,

            }, f, indent=2, ensure_ascii=False)



        print(f"  [+] Манифест хэшей: {manifest_path}")

        self.report["statistics"]["hashed_files"] = len(hash_results)

        return {"files": hash_results}



    def step_2_analyze_journal(self) -> dict:

        """Шаг 2: Анализ журнала регистрации."""

        print("\n[*] Шаг 2: Анализ журнала регистрации 1С")



        journal_dir = None



        if self.base_type == "file_1cd":

            journal_dir = self.base_path / "1Cv8Log" \

                if self.base_path.is_dir() else \

                self.base_path.parent / "1Cv8Log"



        if not journal_dir or not journal_dir.exists():

            print("  [!] Журнал регистрации не найден")

            return {"status": "not_found"}



        try:

            from onec_journal_analyzer import OneCJournalAnalyzer

            analyzer = OneCJournalAnalyzer(str(journal_dir))



            # Экспорт

            csv_path = self.output_dir / "journal_analysis" / "journal_full.csv"

            analyzer.export_to_csv(str(csv_path))



            # Подозрительные паттерны

            suspicious = analyzer.detect_suspicious_patterns()

            findings_path = self.output_dir / "journal_analysis" / "suspicious_patterns.json"

            with open(findings_path, "w", encoding="utf-8") as f:

                json.dump(suspicious, f, indent=2, ensure_ascii=False, default=str)



            print(f"  [+] Записей в журнале: {len(analyzer.entries)}")

            print(f"  [+] Подозрительных паттернов: {len(suspicious)}")



            for finding in suspicious:

                self.report["findings"].append({

                    "source": "journal",

                    **finding,

                })



            return {

                "total_entries": len(analyzer.entries),

                "suspicious_count": len(suspicious),

            }



        except Exception as e:

            print(f"  [!] Ошибка анализа журнала: {e}")

            return {"error": str(e)}



    def step_3_recover_deleted(self) -> dict:

        """Шаг 3: Восстановление удалённых данных."""

        print("\n[*] Шаг 3: Восстановление удалённых данных")

        recovery_output = str(self.output_dir / "recovered_data")



        if self.base_type == "dbf_77":

            from recover_deleted_dbf import recover_journal_operations

            result = recover_journal_operations(

                str(self.base_path),

                recovery_output,

            )

            self.report["statistics"]["recovered_operations"] = result.get("found_deleted", 0)

            return result



        elif self.base_type == "firebird":

            try:

                from fdb_raw_reader import FDBRawReader

                fdb_files = ([self.base_path] if self.base_path.is_file()

                            else list(self.base_path.glob("*.FDB")) +

                                 list(self.base_path.glob("*.fdb")))



                total_deleted = 0

                for fdb_file in fdb_files:

                    reader = FDBRawReader(str(fdb_file))

                    deleted = reader.find_deleted_records()

                    total_deleted += len(deleted)

                    print(f"  [+] {fdb_file.name}: {len(deleted)} back versions")

                    reader.close()



                return {"back_versions_found": total_deleted}

            except Exception as e:

                print(f"  [!] Ошибка анализа FDB: {e}")

                return {"error": str(e)}



        else:

            print(f"  [!] Восстановление для {self.base_type} требует специфических инструментов")

            return {"status": "manual_required"}



    def step_4_detect_fraud(self) -> dict:

        """Шаг 4: Детектирование фальсификаций."""

        print("\n[*] Шаг 4: Детектирование фальсификаций и манипуляций")

        # В реальной реализации здесь используются методы из detect_backdating.py

        # и специфические запросы к базе данных

        return {"status": "requires_db_connection"}



    def step_5_generate_report(self) -> str:

        """Шаг 5: Генерация итогового криминалистического отчёта."""

        print("\n[*] Шаг 5: Генерация криминалистического отчёта")



        self.report["analysis_end"] = datetime.datetime.utcnow().isoformat()

        self.report["total_findings"] = len(self.report["findings"])



        # JSON-отчёт

        json_path = self.output_dir / "reports" / f"{self.case_id}_forensic_report.json"

        with open(json_path, "w", encoding="utf-8") as f:

            json.dump(self.report, f, indent=2, ensure_ascii=False, default=str)



        print(f"  [+] JSON-отчёт: {json_path}")

        print(f"  [+] Всего находок: {self.report['total_findings']}")



        return str(json_path)



    def run_full_analysis(self):

        """Запустить полный анализ."""

        self.step_1_hash_and_document()

        self.step_2_analyze_journal()

        self.step_3_recover_deleted()

        self.step_4_detect_fraud()

        report_path = self.step_5_generate_report()



        print(f"\n{'='*55}")

        print(f"АНАЛИЗ ЗАВЕРШЁН | Кейс: {self.case_id}")

        print(f"Отчёт: {report_path}")

        print(f"{'='*55}")





def main():

    parser = argparse.ArgumentParser(

        description="1С Forensic Toolkit — Криминалистический анализ баз 1С"

    )

    parser.add_argument("base_path",

                        help="Путь к базе данных 1С (файл или директория)")

    parser.add_argument("output_dir",

                        help="Директория для результатов анализа")

    parser.add_argument("--case-id", required=True,

                        help="Идентификатор дела (например: ДЕЛО-2026-001)")

    parser.add_argument("--type",

                        choices=["firebird", "file_1cd", "dbf_77", "mssql", "auto"],

                        default="auto",

                        help="Тип базы данных (default: auto-detect)")

    parser.add_argument("--journal-only", action="store_true",

                        help="Анализировать только журнал регистрации")

    args = parser.parse_args()



    toolkit = OneCForensicToolkit(

        base_path=args.base_path,

        output_dir=args.output_dir,

        case_id=args.case_id,

    )



    if args.journal_only:

        toolkit.step_1_hash_and_document()

        toolkit.step_2_analyze_journal()

        toolkit.step_5_generate_report()

    else:

        toolkit.run_full_analysis()





if __name__ == "__main__":

    main()

-- Выявление подозрительных контрагентов:

-- созданы недавно относительно первой операции

SELECT

    p.Description AS Контрагент,

    p.Code AS КодКонтрагента,

    p.INN AS ИНН,

    MIN(d.Date) AS ПерваяОперация,

    COUNT(d.Ref) AS КоличествоДокументов,

    SUM(dt.Amount) AS ОбщаяСумма,

    p.DataVersion AS ДатаСозданияЗаписи



FROM _Reference150 p  -- Справочник контрагентов (имя таблицы зависит от конфигурации)

JOIN _Document301 d ON d.Counterparty_RRef = p._IDRRef  -- Накладные

JOIN _Document301VT3012 dt ON dt._Document301_IDRRef = d._IDRRef



GROUP BY p.Description, p.Code, p.INN, p.DataVersion

HAVING

    -- Первая операция менее чем через 7 дней после создания контрагента

    DATEDIFF(day, CAST(p.DataVersion AS DATE), MIN(d.Date)) < 7

    AND SUM(dt.Amount) > 100000  -- Значительные суммы



ORDER BY ОбщаяСумма DESC;

def detect_manual_corrections(db_connection, period_start: str, period_end: str) -> list:

    """

    Выявление ручных корректировок в базе 1С.

    Адаптировать под конкретную конфигурацию.

    """

    suspicious = []



    # Запрос к журналу регистрации через SQL (для клиент-серверной базы)

    query = """

    SELECT

        el.date,

        u.Name AS UserName,

        el.comment,

        m.Name AS ObjectType,

        el.DataPresentation

    FROM _EventLog el

    JOIN _EventLogUsers u ON el.UserCode = u.Code

    JOIN _EventLogMetadata m ON el.MetadataCode = m.Code

    JOIN _EventLogEvents ev ON el.Event = ev.Code

    WHERE

        ev.Name IN ('_$Data$_.Post', '_$Data$_.Update', '_$Data$_.New')

        AND el.date BETWEEN :start AND :end

        AND m.Name LIKE '%Корректировка%'

        OR m.Name LIKE '%Adjustment%'

        OR m.Name LIKE '%НачальныеОстатки%'

    ORDER BY el.date

    """

    # Выполнение запроса через подключение к СУБД

    return suspicious

def detect_duplicate_payments(payments: list) -> list:

    """

    Выявление дублирующихся платёжных поручений.

    payments: список словарей с полями date, amount, counterparty, doc_number, bank_account

    """

    from collections import defaultdict

    duplicates = []



    # Группировка по ключевым атрибутам

    payment_groups = defaultdict(list)

    for p in payments:

        # Ключ: сумма + контрагент + дата (в пределах 3 дней)

        date_bucket = p.get("date", "")[:7]  # Год-Месяц

        key = (

            p.get("amount"),

            p.get("counterparty"),

            date_bucket,

            p.get("bank_account"),

        )

        payment_groups[key].append(p)



    for key, group in payment_groups.items():

        if len(group) > 1:

            # Проверить: это одни и те же платёжные поручения или случайное совпадение?

            doc_numbers = [p.get("doc_number") for p in group]

            if len(set(doc_numbers)) > 1:

                # Разные номера документов — возможное дублирование

                total = sum(float(p.get("amount", 0)) for p in group)

                duplicates.append({

                    "type": "DUPLICATE_PAYMENT",

                    "severity": "HIGH",

                    "amount_per_payment": key[0],

                    "total_duplicated": total,

                    "counterparty": key[1],

                    "period": key[2],

                    "documents": group,

                    "description": (

                        f"Обнаружено {len(group)} платёжных поручений "

                        f"на одинаковую сумму {key[0]} "

                        f"контрагенту {key[1]} в период {key[2]}. "

                        f"Номера документов: {', '.join(str(n) for n in doc_numbers)}"

                    ),

                })



    return duplicates

С создаёт резервные копии в нескольких форматах:



1. Файл .dt (Data Transfer) — стандартная выгрузка через интерфейс 1С

   - Архив ZIP с файлами базы данных

   - Содержит все данные конфигурации и прикладные данные

   - Сигнатура: 50 4B 03 04 (ZIP)



2. Архив gbak (Firebird) — резервная копия FDB

   - Создаётся утилитой gbak Firebird

   - Содержит метаданные и данные таблиц в специальном формате

   - Требует gbak для восстановления



3. SQL Server Backup (.bak) — для MS SQL баз

   - Полный/дифференциальный/журнал транзакций

   - Может содержать несколько баз в одном файле



4. Архивы файловой базы — простое копирование 1Cv8.1CD

#!/usr/bin/env python3

"""

backup_analyzer.py — анализ резервных копий 1С и сравнение с текущей базой

"""

import zipfile

import hashlib

import json

import datetime

from pathlib import Path

from typing import Optional





def analyze_dt_backup(dt_path: str, output_dir: str) -> dict:

    """

    Анализ файла резервной копии .dt (Data Transfer format).

    .dt — это ZIP-архив с файлами базы данных.

    """

    path = Path(dt_path)

    output = Path(output_dir)

    output.mkdir(parents=True, exist_ok=True)



    result = {

        "backup_file": str(path),

        "backup_size": path.stat().st_size,

        "format": "unknown",

        "contents": [],

        "backup_date": None,

    }



    # Проверить формат

    if zipfile.is_zipfile(str(path)):

        result["format"] = "ZIP/DT"



        with zipfile.ZipFile(str(path), "r") as zf:

            for info in zf.infolist():

                file_info = {

                    "name": info.filename,

                    "size": info.file_size,

                    "compressed_size": info.compress_size,

                    "date_time": datetime.datetime(*info.date_time).isoformat(),

                    "crc": hex(info.CRC),

                }

                result["contents"].append(file_info)



                # Дата самого свежего файла ≈ дата создания резервной копии

                file_dt = datetime.datetime(*info.date_time)

                if result["backup_date"] is None or file_dt > datetime.datetime.fromisoformat(result["backup_date"]):

                    result["backup_date"] = file_dt.isoformat()



            # Извлечь для анализа

            print(f"[*] Извлечение резервной копии .dt: {path.name}")

            zf.extractall(str(output / "extracted"))

            print(f"[+] Извлечено {len(result['contents'])} файлов в {output / 'extracted'}")



    else:

        # Возможно это сам 1CD файл или FDB

        result["format"] = "raw"

        print(f"[!] Файл не является ZIP-архивом. Возможно, это прямая копия базы.")



    return result





def compare_backups(

    backup1_path: str,

    backup2_path: str,

    output_file: str,

) -> dict:

    """

    Сравнение двух резервных копий 1С для выявления изменений.

    Позволяет точно определить что и когда было изменено.

    """

    print(f"[*] Сравнение резервных копий:")

    print(f"    База 1 (ранняя): {backup1_path}")

    print(f"    База 2 (поздняя): {backup2_path}")



    diff_result = {

        "backup1": backup1_path,

        "backup2": backup2_path,

        "analysis_time": datetime.datetime.utcnow().isoformat(),

        "file_differences": [],

        "added_files": [],

        "removed_files": [],

        "modified_files": [],

    }



    def get_zip_contents(zip_path: str) -> dict:

        """Получить содержимое ZIP с хэшами файлов."""

        contents = {}

        if zipfile.is_zipfile(zip_path):

            with zipfile.ZipFile(zip_path, "r") as zf:

                for info in zf.infolist():

                    try:

                        data = zf.read(info.filename)

                        contents[info.filename] = {

                            "size": info.file_size,

                            "date": datetime.datetime(*info.date_time).isoformat(),

                            "sha256": hashlib.sha256(data).hexdigest(),

                            "crc": info.CRC,

                        }

                    except Exception:

                        pass

        return contents



    contents1 = get_zip_contents(backup1_path)

    contents2 = get_zip_contents(backup2_path)



    all_files = set(contents1.keys()) | set(contents2.keys())



    for filename in all_files:

        if filename not in contents1:

            diff_result["added_files"].append({

                "file": filename,

                "size": contents2[filename]["size"],

                "date": contents2[filename]["date"],

            })

        elif filename not in contents2:

            diff_result["removed_files"].append({

                "file": filename,

                "size": contents1[filename]["size"],

            })

        elif contents1[filename]["sha256"] != contents2[filename]["sha256"]:

            diff_result["modified_files"].append({

                "file": filename,

                "size_before": contents1[filename]["size"],

                "size_after": contents2[filename]["size"],

                "date_before": contents1[filename]["date"],

                "date_after": contents2[filename]["date"],

                "size_delta": contents2[filename]["size"] - contents1[filename]["size"],

            })



    print(f"[+] Добавлено файлов:  {len(diff_result['added_files'])}")

    print(f"[+] Удалено файлов:    {len(diff_result['removed_files'])}")

    print(f"[+] Изменено файлов:   {len(diff_result['modified_files'])}")



    with open(output_file, "w", encoding="utf-8") as f:

        json.dump(diff_result, f, indent=2, ensure_ascii=False)



    return diff_result

ЗАКЛЮЧЕНИЕ СПЕЦИАЛИСТА / ЭКСПЕРТА

(в соответствии со ст. 80, 168 УПК РФ / ст. 85-86 АПК РФ)



1. ВВОДНАЯ ЧАСТЬ



2. ИССЛЕДОВАТЕЛЬСКАЯ ЧАСТЬ

   2.1. Описание объектов исследования

        (тип базы, версия 1С, конфигурация, временной период)

   2.2. Применённые методы и инструменты

        (описание методики, названия ПО с версиями)

   2.3. Ход исследования

        (пошаговое описание каждого этапа)

   2.4. Результаты исследования

        (фактические данные без выводов)



3. ВЫВОДЫ



4. ПРИЛОЖЕНИЯ

   - Таблицы с исходными данными

   - Скриншоты/распечатки

   - Хэши исследованных файлов

   - Использованные инструменты

Перечень типовых вопросов судебной экспертизы баз данных 1С:



1. Были ли в период с [дата1] по [дата2] внесены изменения

   в бухгалтерские документы базы данных 1С организации [название]?

   Если да — в какие документы, кем и когда?



2. Имеются ли в базе данных 1С признаки удаления документов

   или операций за период [дата1]-[дата2]?

   Если да — какие именно документы были удалены, когда и кем?



3. Соответствуют ли даты хозяйственных операций, отражённых

   в базе данных 1С, датам их фактического ввода в систему?



4. Каков суммарный объём хозяйственных операций

   с контрагентом [наименование] за период [дата1]-[дата2]

   по данным базы данных 1С?



5. Имеются ли в базе данных 1С признаки постороннего вмешательства

   или несанкционированной модификации данных?



6. Какой пользователь системы 1С производил операции

   с документами [перечень] в период [дата1]-[дата2]?