11 Commits
Author SHA1 Message Date
naeel 2ceeb05a8b scanner.py: универсальный LLM-промпт (LLM сама определяет типы) + fallback-генератор
Deploy drhider / validate (push) Waiting to run
2026-07-12 09:11:29 +04:00
naeel 579f9c8334 generators: fallback.py — character-class preserving для неизвестных типов
Deploy drhider / validate (push) Waiting to run
2026-07-12 09:09:54 +04:00
naeel d607d7d306 Организация: .md файлы → docs/, обновлены ссылки
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:46:12 +04:00
naeel 4c3f9d4e49 Разделение: copilot-instructions.md (правила) + DEVELOPMENT.md (порядок правок)
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:43:42 +04:00
naeel cc2f627de2 History: документация 2026-07-12
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:38:07 +04:00
naeel 0707d53b37 Документация: README.md, ARCHITECTURE.md, DEPLOY.md, .github/copilot-instructions.md
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:37:51 +04:00
naeel 7a005f7b06 History: результаты рефакторинга
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:34:40 +04:00
naeel 51bc1a79e7 Фаза 2: site/app.py → blueprint'ы (main, health, api), старый drhider.py удалён
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:34:06 +04:00
naeel 2cf4e08100 Фаза 2: llm_client.py, extractor.py, scanner.py, replacer.py, builder.py, obfuscator.py, __init__.py
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:24:33 +04:00
naeel 37581eb601 Фаза 2: config.py, checksum.py, random_utils.py, generators/ (11 модулей)
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:22:20 +04:00
naeel a2f754ab4f Фаза 1: удалён Dockerfile, drhider_server.py, gunicorn из requirements, __main__ блок
Deploy drhider / validate (push) Waiting to run
2026-07-12 08:19:45 +04:00
43 changed files with 2639 additions and 801 deletions
+48
View File
@@ -0,0 +1,48 @@
# ⛔ ПРАВИЛА ДЛЯ COPILOT — DRHIDER
Читать перед ЛЮБЫМ действием.
---
## ⛔⛔⛔ НИЧЕГО НЕ ДЕЛАТЬ БЕЗ «ДЕЛАЙ»
Любой вопрос, обсуждение, анализ — **НЕ повод менять код.**
Только после явного: «делай», «делайте», «давай», «приступай», «пушить», «commit», «push».
## ⛔⛔⛔ НЕ МЕНЯТЬ КОД БЕЗ РАЗРЕШЕНИЯ
Даже если ошибка очевидна. **Показать → описать → ЖДАТЬ.**
## ⛔⛔⛔ НЕ СПЕШИТЬ
Обдумать. Проверить. Перепроверить. Только потом отвечать.
## ⛔⛔⛔ НЕ ПРЕДПОЛАГАТЬ, НЕ ДОГАДЫВАТЬСЯ
Сомневаешься — **ОСТАНОВИСЬ И СПРОСИ.**
## ⛔⛔⛔ НЕ ЛЕЗТЬ В ЛОКАЛЬНЫЕ ПАПКИ CONTRACTS
Код drhider — на ВМ (`5.172.178.213` через SSH). Исключение: `/home/naeel/nubes/loadtest/`.
---
## 📋 Что есть в проекте
| Что нужно | Читай |
|---|---|
| Меняешь код? | [`docs/DEVELOPMENT.md`](docs/DEVELOPMENT.md) — полный порядок действий |
| Архитектура? | [`docs/ARCHITECTURE.md`](docs/ARCHITECTURE.md) — схема, модули, поток данных |
| Деплой/ВМ/env? | [`docs/DEPLOY.md`](docs/DEPLOY.md) — всё про окружение |
| Обзор проекта? | [`README.md`](README.md) — структура, API, правила |
| История изменений? | [`History/`](History/) — что когда и зачем делалось |
---
## 📋 Быстрая справка
- Платформа: **Штурвал** (Managed Flask, без Dockerfile/gunicorn)
- URL: https://drhider.pythonk8s.dev.nubes.ru/
- БД: **нет**, всё в памяти
- Дизайн фронтенда: `/home/naeel/nubes/design/`
- ВМ (legacy): `ssh -i ~/.ssh/naeel_vm_id_ed25519 naeel@5.172.178.213`
-8
View File
@@ -1,8 +0,0 @@
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY drhider.py drhider_server.py /app/
COPY site /app/site
EXPOSE 5000
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--timeout", "300", "--workers", "1", "--worker-class", "gevent", "--limit-request-field_size", "0", "--limit-request-body", "0", "site.app:app"]
+145
View File
@@ -0,0 +1,145 @@
# DrHider — Полный аудит и план рефакторинга
**Дата:** 2026-07-12
**Платформа:** Штурвал (Managed Flask) — запускает Flask сам, без Dockerfile/gunicorn
**URL:** https://drhider.pythonk8s.dev.nubes.ru/
**Репозиторий:** https://gitea.services.ngcloud.ru/Nail/drhider
---
## Текущее состояние (аудит каждого файла)
| Файл | Статус | Решение |
|---|---|---|
| `Dockerfile` | ❌ НЕ НУЖЕН | Штурвал сам запускает Flask. Удалить. |
| `drhider_server.py` | ❌ НЕ НУЖЕН | Standalone-сервер с ВМ (:8767). Flask app.py уже обрабатывает /api/drhider. Удалить. |
| `requirements.txt` | ⚠️ Нужны правки | Убрать `gunicorn` и `gevent`. Оставить `flask`, `python-docx`, `pdfplumber`, `httpx`. |
| `.gitignore` | ✅ OK | Без изменений. |
| `.gitea/workflows/deploy.yaml` | ✅ OK | CI для Штурвала. |
| `site/app.py` | ⚠️ Нужны правки | Удалить блок `if __name__ == "__main__"` (gunicorn). При рефакторинге — вынести LLMClient. |
| `site/templates/index.html` | ✅ OK | UI. |
| `site/static/favicon.svg` | ✅ OK | Иконка (из loadtest). |
| `drhider.py` | ⚠️ МОНОЛИТ | 560 строк. Разбить на ~20 модулей. |
| `History/` | ✅ OK | Документация. |
---
## План рефакторинга (Фаза 1: удаление лишнего)
### Шаг 1.1 — Удалить `Dockerfile`
### Шаг 1.2 — Удалить `drhider_server.py`
### Шаг 1.3 — Исправить `requirements.txt` (убрать gunicorn, gevent)
### Шаг 1.4 — Исправить `site/app.py` (убрать блок `if __name__ == "__main__"`)
### Шаг 1.5 — Проверить синтаксис → commit → push
---
## План рефакторинга (Фаза 2: модульная структура)
### Новая структура `drhider/` пакета (вместо `drhider.py` 560 строк):
```
drhider/ # Пакет ядра обфускации
├── __init__.py # re-export: obfuscate_files()
├── config.py # Константы: ENTITY_PATTERNS, COMPANY_PATTERN, PERSON_PATTERN,
│ # RU_SURNAMES, RU_NAMES, RU_PATRONYMICS,
│ # RU_CITIES, RU_STREETS, FAKE_DOMAINS
├── checksum.py # _checksum_inn10, _checksum_inn12, _checksum_ogrn
├── random_utils.py # _random_digits, _random_letters
├── generators/ # Генераторы фиктивных значений
│ ├── __init__.py # ENTITY_GENERATORS маппинг
│ ├── phone.py # generate_phone
│ ├── email.py # generate_email
│ ├── inn.py # generate_inn10, generate_inn12
│ ├── ogrn.py # generate_ogrn
│ ├── kpp.py # generate_kpp
│ ├── bik.py # generate_bik
│ ├── accounts.py # generate_rs, generate_ks
│ ├── passport.py # generate_passport
│ ├── company.py # generate_company
│ ├── person.py # generate_person
│ └── address.py # generate_address
├── extractor.py # _extract_text, _expand_zips, _convert_pdfs_to_docx
├── scanner.py # Проход 1: _scan_regex, _scan_llm_ner
├── replacer.py # Проход 2: _replace_in_docx, _replace_in_text, _apply_replacements
├── builder.py # Сборка: _build_zip, _build_mapping_csv
├── obfuscator.py # TwoPassObfuscator — оркестратор
└── llm_client.py # LLMClient (из app.py → сюда)
site/ # Flask-приложение
├── __init__.py
├── app.py # Только create_app() + регистрация blueprint'ов
├── routes/
│ ├── __init__.py # Регистрация всех blueprint'ов
│ ├── main_bp.py # GET /
│ ├── health_bp.py # GET /health
│ └── api_bp.py # POST /api/drhider
├── templates/index.html
└── static/favicon.svg
```
### Принципы:
- Каждый файл ≤ 50 строк (где возможно)
- Каждый файл — одна ответственность
- Максимум docstring и инлайн-комментариев
- Все импорты явные, никаких `import *`
---
## Результат рефакторинга (2026-07-12)
### Итоговая структура
```
drhider/ # Пакет ядра обфускации
├── __init__.py # re-export: obfuscate_files, LLMClient, TwoPassObfuscator
├── config.py # Константы: ENTITY_PATTERNS, словари имён/городов/улиц
├── checksum.py # Контрольные суммы: ИНН10, ИНН12, ОГРН
├── random_utils.py # Утилиты: random_digits, random_letters
├── llm_client.py # LLMClient — HTTP-клиент к LLM API
├── extractor.py # Извлечение текста + expand_zips + convert_pdfs_to_docx
├── scanner.py # Проход 1: scan_regex, scan_llm_ner
├── replacer.py # Проход 2: apply_replacements, replace_in_docx, replace_in_text
├── builder.py # Сборка: build_zip, build_mapping_csv
├── obfuscator.py # TwoPassObfuscator — оркестратор
└── generators/ # Генераторы фиктивных значений
├── __init__.py # ENTITY_GENERATORS маппинг
├── phone.py, email.py # Телефон, email
├── inn.py, ogrn.py, kpp.py # ИНН, ОГРН, КПП
├── bik.py, accounts.py # БИК, счета (р/с, к/с)
├── passport.py # Паспорт
├── company.py, person.py # Компания, ФИО
└── address.py # Адрес
site/ # Flask-приложение
├── app.py # create_app() + VERSION (20 строк)
├── routes/
│ ├── __init__.py # register_routes()
│ ├── main_bp.py # GET /
│ ├── health_bp.py # GET /health
│ └── api_bp.py # POST /api/drhider
├── templates/index.html
└── static/favicon.svg
```
### Что изменилось
| Было | Стало |
|---|---|
| `drhider.py` — 560 строк монолит | Пакет `drhider/` — 22 модуля по 20–100 строк |
| `site/app.py` — 100 строк (всё в одном файле) | `app.py` 33 строки + 3 blueprint'а по 2050 строк |
| `Dockerfile` | Удалён (Штурвал сам) |
| `drhider_server.py` | Удалён (Flask заменяет) |
### Коммиты
- `a2f754a` — Фаза 1: удалён Dockerfile, drhider_server.py, gunicorn
- `37581eb` — Фаза 2: config.py, checksum.py, random_utils.py, generators/
- `2cf4e08` — Фаза 2: llm_client.py, extractor.py, scanner.py, replacer.py, builder.py, obfuscator.py
- `51bc1a7` — Фаза 2: site/app.py → blueprint'ы, старый drhider.py удалён
| Переменная | Значение |
|---|---|
| `LLM_API_KEY` | `sk-ucI5YvOticoOQ9Kuj5K9mQ` |
| `LLM_URL` | `https://api.aillm.ru/v1/chat/completions` |
| `LLM_MODEL` | `gpt-oss-120b` |
+24
View File
@@ -0,0 +1,24 @@
# Документация — 2026-07-12
**После рефакторинга** создана полная документация:
### Созданные файлы
| Файл | Назначение |
|---|---|
| `README.md` | Обзор проекта, правила для агентов, структура, API, env vars |
| `ARCHITECTURE.md` | Архитектура: схема, двухпроходная обфускация, поток данных, описание каждого модуля |
| `DEPLOY.md` | Деплой, ВМ, переменные окружения, локальная разработка, как добавить генератор |
| `.github/copilot-instructions.md` | Правила для Copilot (не менять без «делай», не смотреть локальные папки, коммитить после каждой правки) |
### Ключевые факты зафиксированы
- Платформа: Штурвал (Managed Flask), без Dockerfile/gunicorn
- URL: https://drhider.pythonk8s.dev.nubes.ru/
- Дизайн фронтенда: `/home/naeel/nubes/design/`
- ВМ (legacy): `5.172.178.213`, ssh-ключ `~/.ssh/naeel_vm_id_ed25519`
- Переменные окружения: LLM_API_KEY, LLM_URL, LLM_MODEL
### Коммит
`0707d53` — Документация: README.md, ARCHITECTURE.md, DEPLOY.md, .github/copilot-instructions.md
+76
View File
@@ -0,0 +1,76 @@
# Запрос к Соннету — универсальная архитектура DrHider
## Контекст
Проект DrHider — обфускация документов (замена персональных данных на фиктивные). Сейчас:
- Python/Flask, без БД, Managed Flask на платформе Штурвал
- Двухпроходная архитектура: сбор сущностей → замена
- Проход 1: regex-паттерны (телефон, email, ИНН, ОГРН, КПП, БИК, счета, паспорт) + LLM NER (ФИО, компании, адреса, паспорта)
- Проход 2: замена по словарю mapping
## Проблема
Архитектура **в корне неверна** — жёстко привязана к фиксированному списку типов сущностей:
```python
# config.py — жёстко зашитые паттерны
ENTITY_PATTERNS = {
"phone": r'...',
"email": r'...',
"inn_fl": r'ИНН\s*\d{12}',
...
}
```
```python
# scanner.py — жёстко зашитый промпт
prompt = (
"Найди ВСЕ следующие сущности:\n"
"1. ФИО\n2. Компании\n3. Адреса\n4. Паспортные данные\n"
)
```
Если новый документ содержит СНИЛС, водительское удостоверение, номер договора, ИНН без префикса «ИНН» — система их НЕ обнаружит. Надо править config, generators, scanner, маппинги.
## Что нужно
**Универсальная архитектура**, где система САМА определяет что скрывать в любом документе:
1. Не привязана к списку типов
2. Не требует добавления паттернов под каждый новый вид данных
3. LLM сама решает что является персональными/конфиденциальными данными
4. Генерация фиктивных значений — тоже универсальная (не 11 отдельных функций)
## Текущая структура (полная)
```
drhider/
├── config.py # ENTITY_PATTERNS (10 regex), словари имён/городов
├── checksum.py # Контрольные суммы ИНН/ОГРН
├── random_utils.py # random_digits, random_letters
├── generators/ # 11 файлов: phone, email, inn, ogrn, kpp, bik, accounts, passport, company, person, address
├── llm_client.py # HTTP-клиент к LLM API
├── extractor.py # Извлечение текста + expand_zips + convert_pdfs_to_docx
├── scanner.py # scan_regex (regex) + scan_llm_ner (LLM NER с жёстким промптом)
├── replacer.py # apply_replacements, replace_in_docx, replace_in_text
├── builder.py # build_zip, build_mapping_csv
├── obfuscator.py # TwoPassObfuscator — оркестратор
├── site/app.py # Flask (3 blueprint'а)
└── site/templates/ # HTML-интерфейс
```
## Вопросы к Соннету
1. Как перестроить архитектуру чтобы обнаружение было универсальным (LLM сама решает что скрывать)?
2. Нужен ли regex вообще или достаточно одного LLM с правильным промптом?
3. Как сделать генерацию фиктивных значений универсальной? (Не 11 функций под каждый тип, а что-то общее)
4. Двухпроходная схема (сбор→замена) — сохранять или перейти на однопроходную (LLM сразу возвращает обфусцированный текст)?
5. Как должен выглядеть промпт чтобы LLM возвращала структурированный результат (что найдено + на что заменить)?
6. Стоит ли сохранять DOCX-форматирование (сейчас runs склеиваются-разделяются) или проще отдать LLM plain text?
## Ограничения
- Документы до 200 MB
- Форматы: .docx, .pdf, .txt, .zip
- LLM: OpenAI-совместимое API (aillm.ru, модель gpt-oss-120b, 8000 токенов)
- Без БД, всё в памяти
- Платформа: Managed Flask на Kubernetes
+78
View File
@@ -0,0 +1,78 @@
# Ответ Соннета — универсальная архитектура DrHider (2026-07-12)
Кратко: 6 вопросов → 6 ответов → карта изменений.
---
## Q1. Как сделать обнаружение универсальным?
Промпт должен звучать не «найди вот эти типы», а «найди ВСЁ, что выглядит как приватная информация, и сам назови тип».
Затрагивает: `scanner.py` (промпт), `builder.py` (тип для mapping.csv).
---
## Q2. Regex или LLM?
**Гибрид — правильный выбор:**
- **Regex = pre-pass** для структурированных данных (телефон, email, ИНН, БИК). Экономит токены LLM.
- **LLM = post-pass** для неструктурированного (имена, адреса, нестандартные ID).
- Дублирования не страшны — mapping dict сам отсеет.
---
## Q3. Генерация фиктивных значений — универсальная?
Два уровня:
1. **Известные типы** — специализированные генераторы (оставить как есть).
2. **Неизвестные типы** — fallback-генератор: character-class preserving замена (цифры→цифры, буквы→буквы той же длины).
Дополнительно: в промпт добавить `"category": "person|org|contact|id|financial|other"` — 6 категорий вместо 10+ типов.
---
## Q4. Двухпроходная или однопроходная?
**Двухпроходная — обязательно.** Причина: «Иванов» в 5 документах должен заменяться одинаково. Однопроход не может этого гарантировать.
Текущий `TwoPassObfuscator` — правильный, не трогать.
---
## Q5. Новый промпт
```
You are a PII detector. Find ALL sensitive or private information.
Return JSON array: [{"type": "short_label", "value": "exact_string"}]
Rules:
- Copy "value" VERBATIM from text
- "type" is snake_case label you invent
- Same value → include once
- Return ONLY JSON
```
Ключевое: нет ограничения на типы, value verbatim.
---
## Q6. DOCX или Markdown?
- **Внутренняя обработка:** Markdown проще парсить, LLM понимает лучше.
- **На выходе:** опция `output_format: "docx" | "md"`. По умолчанию `"docx"`.
---
## Итоговая карта изменений
```
scanner.py — новый промпт (убрать список типов, LLM сама называет типы)
generators/ — добавить fallback-генератор для неизвестных типов
obfuscator.py — вызывать fallback если тип неизвестен
api_bp.py — принять параметр output_format
replacer.py — добавить replace_to_markdown()
builder.py — упаковывать .md если output_format=md
```
**Не трогать:** двухпроходная схема, checksum-генераторы, ZIP-безопасность.
+51
View File
@@ -0,0 +1,51 @@
# План v3 — универсальная архитектура (2026-07-12)
## Цель
Убрать жёсткую привязку к фиксированному списку типов сущностей.
Перейти на Markdown как внутренний формат.
## Изменения по модулям
### 1. `scanner.py` — универсальный LLM-промпт
- Убрать список типов из промпта
- LLM сама называет типы (snake_case)
- `value` — verbatim из текста
### 2. `generators/` — fallback-генератор
- Новый `fallback.py`: character-class preserving замена
- `__init__.py`: добавить в ENTITY_GENERATORS
- `obfuscator.py`: использовать fallback для неизвестных типов
### 3. `extractor.py` — единый MD-пайплайн
- DOCX → MD (python-docx: стили, форматирование)
- PDF → MD (pdfplumber: текст + таблицы)
- TXT → как есть
- Убрать convert_pdfs_to_docx (не нужен)
- Убрать хранение docx-объектов
### 4. `replacer.py` — упростить
- `apply_replacements(text)` — уже есть
- `replace_in_docx(doc, mapping)` — сохранить как утилиту для DOCX-выхода
- Удалить `replace_in_text()` (заменяется на apply_replacements)
### 5. `obfuscator.py` — обновить пайплайн
- Pass 1: extract MD → scan_regex + scan_llm_ner
- Pass 2: apply_replacements к MD
- Опционально: replace_in_docx к оригинальному DOCX
### 6. `api_bp.py` — output_format
- Параметр `output_format: "md" | "docx"` (по умолчанию md)
### 7. `.doc` — потом
- Добавим convert_doc_to_docx через LibreOffice отдельно
## Порядок реализации
1. `generators/fallback.py` + обновить `__init__.py`
2. `scanner.py` — новый промпт
3. `extractor.py` — MD-конвертация
4. `replacer.py` — упростить
5. `obfuscator.py` — новый пайплайн
6. `api_bp.py` — output_format
7. Проверить всё → commit
+142
View File
@@ -0,0 +1,142 @@
# DrHider — обфускация документов (Managed Flask)
**URL:** https://drhider.pythonk8s.dev.nubes.ru/
**Репозиторий:** https://gitea.services.ngcloud.ru/Nail/drhider
**Платформа:** Штурвал (Managed Flask на pythonk8s)
**Дизайн фронтенда:** `/home/naeel/nubes/design/`
---
## Что делает
Загружаешь документы (.docx, .pdf, .txt, .zip) — получаешь ZIP с обфусцированными копиями.
Все персональные данные, реквизиты, телефоны, email заменяются на фиктивные.
В архиве — mapping.csv с таблицей замен (оригинал → фиктивное).
Обфускация двухпроходная:
1. **Проход 1 (сбор):** regex + LLM находят все сущности во всех файлах → глобальный словарь замен
2. **Проход 2 (замена):** применяем замены ко всем файлам → ZIP
Согласованность: одна и та же сущность во всех файлах → одно и то же фиктивное значение.
---
## ⛔ ПРАВИЛА ДЛЯ АГЕНТОВ (читать перед ЛЮБЫМ действием)
### 1. НЕ смотреть локальные папки contracts/contracts-flask
Весь старый код drhider — ТОЛЬКО на ВМ (`5.172.178.213`). НЕ локально.
Единственное исключение: `/home/naeel/nubes/loadtest/` (образец структуры managed Flask).
### 2. Штурвал запускает Flask САМ
- БЕЗ Dockerfile (удалён)
- БЕЗ gunicorn (нет в requirements.txt)
- Точка входа: `site/app.py``app = create_app()`
- Деплой: `git push origin master` → авто-деплой в UI Штурвала
### 3. НЕ менять код без «делай»
Даже если ошибка очевидна. Показать → описать → ЖДАТЬ.
### 4. После ЛЮБОЙ правки:
```bash
python3 -c "import py_compile; py_compile.compile('файл.py', doraise=True)"
git add -A && git commit -m "подробное описание" && git push
```
### 5. Документировать ВСЁ в History/
Каждое изменение, план, ошибку — в отдельный .md файл.
---
## Структура проекта
### `drhider/` — пакет ядра обфускации
| Файл | Назначение | Строк |
|---|---|---|
| `__init__.py` | re-export: `obfuscate_files`, `LLMClient`, `TwoPassObfuscator` | 10 |
| `config.py` | Константы: regex-паттерны (ENTITY_PATTERNS, COMPANY_PATTERN, PERSON_PATTERN), словари имён/городов/улиц | 80 |
| `checksum.py` | Контрольные суммы: ИНН10, ИНН12, ОГРН | 60 |
| `random_utils.py` | Утилиты: `random_digits`, `random_letters` | 20 |
| `llm_client.py` | LLMClient — HTTP-клиент к LLM API (aillm.ru) | 50 |
| `extractor.py` | Извлечение текста из .docx/.pdf/.txt + `expand_zips` + `convert_pdfs_to_docx` | 180 |
| `scanner.py` | Проход 1: `scan_regex` (regex) + `scan_llm_ner` (LLM NER) | 110 |
| `replacer.py` | Проход 2: `apply_replacements`, `replace_in_docx`, `replace_in_text` | 100 |
| `builder.py` | Сборка: `build_zip`, `build_mapping_csv` | 65 |
| `obfuscator.py` | `TwoPassObfuscator` — оркестратор + `obfuscate_files()` | 100 |
### `drhider/generators/` — генераторы фиктивных значений
Каждый файл — один генератор (10–40 строк). Интерфейс: `generate_xxx(original: str) -> str`.
| Файл | Что генерирует |
|---|---|
| `phone.py` | +7 (XXX) XXX-XX-XX |
| `email.py` | user@fake-domain |
| `inn.py` | ИНН 10 и 12 знаков (с контрольной суммой) |
| `ogrn.py` | ОГРН 13 знаков |
| `kpp.py` | КПП 9 знаков |
| `bik.py` | БИК 9 знаков (04XXXXXXX) |
| `accounts.py` | Расчётный счёт (40702...) + корр. счёт (30101...) |
| `passport.py` | Паспорт (XX XX XXXXXX) |
| `company.py` | ООО/ЗАО/АО «СлучайноеНазвание» |
| `person.py` | Фамилия И.О. |
| `address.py` | Город, ул. Улица, д. N |
`__init__.py` содержит `ENTITY_GENERATORS` — маппинг `entity_type → генератор`.
### `site/` — Flask-приложение
| Файл | Назначение |
|---|---|
| `app.py` | `create_app()` — создание Flask, регистрация blueprint'ов, VERSION |
| `routes/__init__.py` | `register_routes(app)` |
| `routes/main_bp.py` | `GET /` — HTML-интерфейс |
| `routes/health_bp.py` | `GET /health``{"ok": true, "version": "..."}` |
| `routes/api_bp.py` | `POST /api/drhider` — multipart files → ZIP |
| `templates/index.html` | UI (выбор файлов, прогресс, скачивание) |
| `static/favicon.svg` | Иконка |
### Корневые файлы
| Файл | Назначение |
|---|---|
| `requirements.txt` | `flask`, `python-docx`, `pdfplumber`, `httpx` |
| `.gitignore` | `__pycache__/`, `*.pyc`, `.env` |
| `.gitea/workflows/deploy.yaml` | CI: валидация Python |
| `History/` | Документация всех изменений |
| `README.md` | Этот файл |
| `docs/ARCHITECTURE.md` | Архитектура |
| `docs/DEPLOY.md` | Деплой, ВМ, переменные окружения |
| `docs/DEVELOPMENT.md` | Порядок действий при изменении кода |
---
## API
| Метод | Путь | Что делает |
|---|---|---|
| `GET` | `/` | HTML-интерфейс |
| `GET` | `/health` | `{"ok": true, "version": "2.0.0"}` |
| `POST` | `/api/drhider` | multipart/form-data (поле `files`) → `application/zip` |
---
## Переменные окружения (Штурвал)
| Переменная | Значение |
|---|---|
| `LLM_API_KEY` | `sk-ucI5YvOticoOQ9Kuj5K9mQ` |
| `LLM_URL` | `https://api.aillm.ru/v1/chat/completions` |
| `LLM_MODEL` | `gpt-oss-120b` |
---
## Связь с ВМ contracts (legacy)
На ВМ `5.172.178.213` (contracts.kube5s.ru) есть старая версия drhider:
- `drhider_server.py` на порту 8767 — standalone HTTP-сервер
- `drhider.py` — та же логика, что и в этом проекте (скопирована оттуда)
Этот проект (`drhider.pythonk8s.dev.nubes.ru`) — **независимый managed Flask**, заменяет ВМ-версию.
SSH к ВМ: `ssh -i ~/.ssh/naeel_vm_id_ed25519 naeel@5.172.178.213`
+191
View File
@@ -0,0 +1,191 @@
# Архитектура DrHider
## Общая схема
```
Пользователь (браузер)
https://drhider.pythonk8s.dev.nubes.ru/
Штурвал (Managed Flask на pythonk8s)
site/app.py ──→ Flask (create_app)
├── GET / → main_bp → templates/index.html
├── GET /health → health_bp → {"ok": true}
└── POST /api/drhider → api_bp → drhider.obfuscate_files()
TwoPassObfuscator
┌────────────┼────────────┐
│ │ │
extractor scanner replacer
(текст) (regex+LLM) (замена)
│ │ │
└────────────┼────────────┘
builder
(ZIP+CSV)
application/zip
```
---
## Двухпроходная обфускация
### Проход 1: сбор сущностей
```
Файлы (.docx, .pdf, .txt, .zip)
extractor.expand_zips() ← распаковать ZIP
extractor.convert_pdfs_to_docx() ← PDF → DOCX
extractor.extract_text() ← извлечь текст из каждого файла
├──→ scanner.scan_regex() ← regex: телефоны, email, ИНН, ОГРН, КПП,
│ БИК, счета, паспорта, компании, ФИО
└──→ scanner.scan_llm_ner() ← LLM: имена, адреса, паспорта
(только если llm_client передан)
mapping = {оригинал → фиктивное} ← глобальный словарь замен
```
### Проход 2: замена
```
mapping + sorted_keys (по убыванию длины)
Для каждого файла:
├── .docx → replacer.replace_in_docx() ← склеить runs → заменить → сохранить
├── .doc → без изменений (бинарный)
└── .txt/.pdf → replacer.replace_in_text() ← простая строковая замена
builder.build_mapping_csv() ← mapping.csv
builder.build_zip() ← ZIP со всеми файлами + mapping.csv
```
---
## Модули
### `drhider/config.py`
Константы уровня всего пакета. НЕ содержит логики — только данные.
- `ENTITY_PATTERNS` — 10 regex-паттернов
- `COMPANY_PATTERN`, `PERSON_PATTERN` — скомпилированные regex
- Словари: `RU_SURNAMES`, `RU_NAMES`, `RU_PATRONYMICS`, `RU_CITIES`, `RU_STREETS`, `FAKE_DOMAINS`
### `drhider/checksum.py`
Чистые функции для расчёта контрольных сумм. Без побочных эффектов.
- `checksum_inn10(inn)` → 1 цифра
- `checksum_inn12(inn)` → 2 цифры
- `checksum_ogrn(ogrn)` → 1 цифра
### `drhider/random_utils.py`
Утилиты генерации случайных строк.
- `random_digits(n)` → строка из n цифр
- `random_letters(n)` → строка из n строчных латинских букв
### `drhider/generators/`
Каждый генератор — независимая функция с сигнатурой `generate_xxx(original: str) -> str`.
Параметр `original` нужен для извлечения префикса (например, «ИНН » из «ИНН 1234567890»).
Генераторы НЕ общаются друг с другом — только импортируют из `config`, `checksum`, `random_utils`.
`__init__.py` содержит `ENTITY_GENERATORS` — словарь, связывающий `entity_type` из `ENTITY_PATTERNS` с функцией-генератором. Используется в `scanner.scan_regex()`.
### `drhider/llm_client.py`
HTTP-клиент к OpenAI-совместимому API. Читает переменные окружения.
Интерфейс: `.complete(prompt: str) -> str`.
### `drhider/extractor.py`
Три независимые функции (не класс):
- `extract_text(fname, content, ctype)``(text, docx_document_or_None)`
- `expand_zips(files)` → распакованный список
- `convert_pdfs_to_docx(files)` → PDF заменены на DOCX
### `drhider/scanner.py`
Две функции (не класс), мутируют переданный `mapping: Dict[str, str]`:
- `scan_regex(text, mapping)` — regex-поиск
- `scan_llm_ner(all_texts, mapping, llm_client)` — LLM NER
### `drhider/replacer.py`
Три чистые функции:
- `apply_replacements(text, mapping, sorted_keys)` → строка с заменами
- `replace_in_docx(doc, mapping, sorted_keys)` → bytes (изменённый DOCX)
- `replace_in_text(text, fname, mapping, sorted_keys)` → bytes
### `drhider/builder.py`
Две функции сборки результата:
- `build_zip(files, mapping_csv)` → bytes (ZIP-архив)
- `build_mapping_csv(mapping)` → str (CSV)
### `drhider/obfuscator.py`
`TwoPassObfuscator` — оркестратор. Единственный класс, который знает о顺序е вызовов.
Метод `.obfuscate(files)` вызывает модули в правильном порядке.
`obfuscate_files()` — удобная функция-обёртка.
### `site/app.py`
Точка входа для Штурвала. `create_app()` создаёт Flask, регистрирует blueprint'ы.
`app = create_app()` — глобальный экземпляр.
### `site/routes/`
Blueprint'ы — каждый в своём файле:
- `main_bp` — только `GET /`
- `health_bp` — только `GET /health`
- `api_bp` — только `POST /api/drhider`
`__init__.py` содержит `register_routes(app)` — единая точка регистрации.
---
## Поток данных
```
HTTP POST /api/drhider (multipart/form-data)
api_bp.drhider()
│ request.files.getlist("files")
│ → [(filename, bytes, mimetype), ...]
obfuscate_files(files, llm_client=LLMClient())
TwoPassObfuscator.obfuscate(files)
├── expand_zips → convert_pdfs_to_docx → extract_text
├── scan_regex + scan_llm_ner → mapping
├── replace_in_docx / replace_in_text → обфусцированные файлы
└── build_mapping_csv + build_zip → (zip_bytes, csv_str)
send_file(BytesIO(zip_data), mimetype="application/zip")
HTTP 200 + ZIP-архив
```
---
## Принципы
1. **Модули — чистые функции.** Где возможно — без классов, без состояния.
2. **Единственный класс — TwoPassObfuscator.** Только он управляет состоянием (mapping, sorted_keys).
3. **Генераторы — изолированы.** Каждый в своём файле, не зависят друг от друга.
4. **Flask — тонкая прослойка.** Только принимает запрос, вызывает `obfuscate_files()`, отдаёт ответ.
5. **Без БД.** Всё в памяти. Никаких внешних зависимостей кроме LLM API.
+139
View File
@@ -0,0 +1,139 @@
# Деплой, ВМ и переменные окружения
---
## Платформа
**Штурвал** — Managed Flask на pythonk8s (Kubernetes).
- Запускает Flask **сам** — без Dockerfile, без gunicorn
- Точка входа: `site/app.py` → глобальная переменная `app`
- Деплой: `git push origin master` → авто-деплой через UI Штурвала
---
## Деплой
```bash
cd /home/naeel/nubes/drhider
git add -A
git commit -m "описание изменений"
git push origin master
```
После пуша — зайти в UI Штурвала и нажать ** Redeploy** для `drhider`.
---
## Переменные окружения (настраиваются в UI Штурвала)
| Переменная | Значение | Для чего |
|---|---|---|
| `LLM_API_KEY` | `sk-ucI5YvOticoOQ9Kuj5K9mQ` | Ключ доступа к LLM API |
| `LLM_URL` | `https://api.aillm.ru/v1/chat/completions` | URL LLM API |
| `LLM_MODEL` | `gpt-oss-120b` | Модель для NER |
Без `LLM_API_KEY` LLM-сканирование не работает (только regex).
---
## ВМ contracts (legacy)
Старый сервер contracts.kube5s.ru. DrHider там — standalone HTTP-сервер на порту 8767.
### SSH
```bash
ssh -i ~/.ssh/naeel_vm_id_ed25519 naeel@5.172.178.213
```
### Где лежит старый drhider
```
/home/naeel/contracts/drhider/
├── drhider.py # Ядро обфускации (отсюда скопировано в этот проект)
├── drhider_server.py # HTTP-сервер :8767 (НЕ ИСПОЛЬЗУЕТСЯ в новом проекте)
└── __init__.py
```
### Сервисы на ВМ
| Сервис | Порт | systemd unit |
|---|---|---|
| drhider | 8767 | `contracts-drhider` |
| convert_server | 8766 | `contracts` |
| Flask UI | 5001 | — (start.sh) |
| nginx | 443 | `nginx` |
| PostgreSQL | 5432 | — |
### Управление drhider на ВМ
```bash
# Статус
systemctl status contracts-drhider
# Логи
journalctl -u contracts-drhider -f
# Перезапуск
sudo systemctl restart contracts-drhider
```
---
## Локальная разработка
```bash
cd /home/naeel/nubes/drhider
# Установка зависимостей
pip install -r requirements.txt
# Запуск Flask (dev-сервер)
cd site && python3 app.py
# Проверка синтаксиса ВСЕХ файлов
python3 -c "
import py_compile, os
for root, dirs, files in os.walk('.'):
for f in files:
if f.endswith('.py'):
path = os.path.join(root, f)
py_compile.compile(path, doraise=True)
print(f'OK: {path}')
"
# Проверка импорта
python3 -c "from drhider import obfuscate_files, LLMClient; print('OK')"
```
---
## Как править код
1. Понять что меняем → **описать план**
2. Получить **«делай»**
3. Внести изменения
4. `python3 -c "import py_compile; py_compile.compile('файл.py', doraise=True)"` — КАЖДЫЙ изменённый .py
5. `python3 -c "from drhider import obfuscate_files, LLMClient"` — проверка импорта
6. `git add -A && git commit -m "подробно что и зачем" && git push`
7. Записать в `History/` что сделано
8. Зайти в UI Штурвала → Redeploy
---
## Как добавить новый генератор
1. Создать `drhider/generators/новый_тип.py` с функцией `generate_xxx(original: str) -> str`
2. Добавить regex-паттерн в `drhider/config.py``ENTITY_PATTERNS`
3. Добавить импорт и маппинг в `drhider/generators/__init__.py``ENTITY_GENERATORS`
4. Проверить синтаксис → commit → push → Redeploy
---
## История версий
| Версия | Дата | Что |
|---|---|---|
| 2.0.0 | 2026-07-12 | Модульный рефакторинг (22 модуля вместо монолита) |
| 1.0.0 | 2026-07-11 | Начальная версия (monolith drhider.py + Flask) |
+104
View File
@@ -0,0 +1,104 @@
# Порядок действий при изменении кода
**Читать перед ЛЮБОЙ правкой.**
---
## ⛔ ЗАПРЕЩЕНО
1. Менять код без «делай»
2. Действовать по догадкам («я бы сделал так», «можно попробовать»)
3. Удалять файлы/папки без разрешения
4. «Улучшать» рабочий код без прямого разрешения
5. Использовать `sed`
6. Откладывать commit/push «на потом»
7. Игнорировать проверку синтаксиса перед push
---
## ✅ ПОРЯДОК ПРИ ЛЮБОМ ИЗМЕНЕНИИ
### Шаг 1 — План
```
Описать что меняем → получить «делай»
```
### Шаг 2 — Правка
Вносим изменения. Только то, что обговорено. Никакой самодеятельности.
### Шаг 3 — Проверка синтаксиса
```bash
python3 -c "import py_compile; py_compile.compile('путь/к/файлу.py', doraise=True)"
```
**КАЖДЫЙ изменённый .py файл.**
### Шаг 4 — Проверка импорта
```bash
cd /home/naeel/nubes/drhider && python3 -c "from drhider import obfuscate_files, LLMClient; print('OK')"
```
### Шаг 5 — Проверить соседей
После `replace_string_in_file` — прочитать соседние строки, убедиться что не затёр соседнюю функцию/класс.
### Шаг 6 — commit + push (СРАЗУ)
```bash
cd /home/naeel/nubes/drhider
git add -A
git commit -m "подробное описание что и зачем"
git push
```
**Без исключений. Никаких накоплений.**
### Шаг 7 — Документировать
Записать в `History/` что сделано:
- Отдельный `.md` файл
- Что планировалось → что сделано → в чём ошибся
### Шаг 8 — Redeploy
Зайти в UI Штурвала → Redeploy `drhider`.
---
## 🔧 Команды в терминале — всегда с таймаутами
```bash
curl --max-time 30 ...
ssh -o ConnectTimeout=10 ...
timeout 10 grep ... file.txt
```
---
## 📦 Проверка ВСЕХ файлов разом
```bash
cd /home/naeel/nubes/drhider
python3 -c "
import py_compile, os
for root, dirs, files in os.walk('.'):
for f in files:
if f.endswith('.py'):
py_compile.compile(os.path.join(root, f), doraise=True)
print(f'OK: {os.path.join(root, f)}')
"
```
---
## 🏷️ Версионирование
Версия в `site/app.py` (переменная `VERSION`).
Менять при любом изменении кода, влияющем на поведение.
---
## 📝 Пример хорошего коммита
```
git commit -m "generators: добавил generate_snils — генератор фиктивных СНИЛС
- Новый файл drhider/generators/snils.py
- Добавлен паттерн в config.py → ENTITY_PATTERNS
- Добавлен в generators/__init__.py → ENTITY_GENERATORS
- bump VERSION 2.0.0 → 2.0.1"
```
+5
View File
@@ -0,0 +1,5 @@
НЕ СПЕШИ ! не ошибайся. НЕ ПРЕДПОЛАГАЙ !
не надо ДОГАДОК !
не надо самостоятельно что либо "УЛуЧШАТЬ" - никаких изменений рабочего кода без прямого разрешения
есть сомнения - лучше остановись и спроси
ВСЁ записывай в хистори ! что планировал что сделал в чём ошибся и тд
-557
View File
@@ -1,557 +0,0 @@
"""
DrHider — обфускация документов (двухпроходная, в памяти, без БД).
Проход 1: собрать все сущности из всех файлов → глобальный словарь замен.
Проход 2: применить замены → собрать ZIP с обфусцированными файлами + mapping.csv.
Согласованность: одна и та же сущность во всех файлах → одно и то же фиктивное значение.
"""
DRHIDER_VERSION = "1.3"
import io
import csv
import re
import random
import string
import zipfile
import logging
import os
from typing import Dict, List, Tuple, Callable, Optional
log = logging.getLogger("drhider")
# ═══════════════════════════════════════════
# Regex-паттерны для обнаружения сущностей
# ═══════════════════════════════════════════
ENTITY_PATTERNS: Dict[str, str] = {
"phone": r'(?<!\d)(?:\+7|8)[\s\-]?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{2}[\s\-]?\d{2}(?!\d)',
"email": r'\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b',
# 12-значный ИНН ДО 10-значного (иначе 12-значный матчится как 10)
"inn_fl": r'ИНН\s*\d{12}',
"inn_ul": r'ИНН\s*\d{10}',
"ogrn": r'ОГРН\s*\d{13}',
"kpp": r'КПП\s*\d{9}',
"bik": r'БИК\s*\d{9}',
"rs": r'(?:р/с|расч[её]тный\s*сч[её]т)\s*\d{20}',
"ks": r'(?:к/с|кор/сч|корр?[еи]?спондентский\s*сч[её]т)\s*\d{20}',
"passport": r'(?:паспорт|серия\s+номер)[\s:№]*\d{2}\s*\d{2}\s*\d{6,7}',
}
# Фирмы — обнаружение по шаблону
COMPANY_PATTERN = re.compile(
r'(?:ООО|ЗАО|ОАО|АО|ПАО|ИП|ТОО)\s+(?:«[^»]+»|"[^"]+"|\u201c[^\u201d]+\u201d|[А-ЯA-Z][\w\-\.]{2,40})',
re.IGNORECASE
)
# ФИО — Фамилия + инициалы или полное имя (только кириллица)
PERSON_PATTERN = re.compile(
r'\b[А-Я][а-я]+\s+[А-Я]\.[А-Я]\.'
r'|\b[А-Я][а-я]+\s+[А-Я][а-я]+\s+[А-Я][а-я]+'
)
# ═══════════════════════════════════════════
# Генераторы фиктивных значений
# ═══════════════════════════════════════════
# Словари русских имён
RU_SURNAMES = ["Иванов", "Смирнов", "Кузнецов", "Попов", "Васильев", "Петров",
"Соколов", "Михайлов", "Новиков", "Фёдоров", "Морозов", "Волков", "Алексеев",
"Лебедев", "Семёнов", "Егоров", "Павлов", "Козлов", "Степанов", "Николаев"]
RU_NAMES = ["Александр", "Дмитрий", "Сергей", "Андрей", "Алексей", "Максим",
"Евгений", "Иван", "Михаил", "Николай", "Владимир", "Павел", "Виктор", "Олег"]
RU_PATRONYMICS = ["Александрович", "Дмитриевич", "Сергеевич", "Андреевич",
"Алексеевич", "Иванович", "Михайлович", "Николаевич", "Владимирович",
"Павлович", "Викторович", "Олегович", "Евгеньевич", "Максимович"]
RU_CITIES = ["Москва", "Санкт-Петербург", "Новосибирск", "Екатеринбург",
"Казань", "Нижний Новгород", "Челябинск", "Самара", "Омск", "Ростов-на-Дону",
"Уфа", "Красноярск", "Воронеж", "Пермь", "Волгоград"]
RU_STREETS = ["Ленина", "Мира", "Пушкина", "Гагарина", "Советская",
"Кирова", "Октябрьская", "Молодёжная", "Садовая", "Центральная"]
FAKE_DOMAINS = ["example.ru", "mail.test", "company.local", "org.example.ru"]
def _checksum_inn10(inn: str) -> str:
"""Контрольная сумма для 10-значного ИНН."""
coeffs = [2, 4, 10, 3, 5, 9, 4, 6, 8]
s = sum(int(inn[i]) * coeffs[i] for i in range(9))
return str((s % 11) % 10)
def _checksum_inn12(inn: str) -> str:
"""Контрольные суммы для 12-значного ИНН (первые 10 цифр)."""
c1 = [7, 2, 4, 10, 3, 5, 9, 4, 6, 8]
c2 = [3, 7, 2, 4, 10, 3, 5, 9, 4, 6, 8]
s1 = sum(int(inn[i]) * c1[i] for i in range(10))
n1 = (s1 % 11) % 10
inn2 = inn + str(n1)
s2 = sum(int(inn2[i]) * c2[i] for i in range(11))
n2 = (s2 % 11) % 10
return str(n1) + str(n2)
def _checksum_ogrn(ogrn: str) -> str:
"""Контрольная сумма для ОГРН (12 цифр → остаток от деления на 11)."""
s = int(ogrn) % 11
return str(s % 10)
def _random_digits(n: int) -> str:
return ''.join(random.choice(string.digits) for _ in range(n))
def _random_letters(n: int) -> str:
return ''.join(random.choice(string.ascii_lowercase) for _ in range(n))
def generate_phone(_: str) -> str:
code = random.choice(["495", "499", "812", "383", "343"])
return f"+7 ({code}) {_random_digits(3)}-{_random_digits(2)}-{_random_digits(2)}"
def generate_email(original: str) -> str:
"""Генерирует email с тем же форматом."""
domain = random.choice(FAKE_DOMAINS)
local = _random_letters(random.randint(5, 10))
return f"{local}@{domain}"
def generate_inn10(original: str) -> str:
prefix = re.match(r'ИНН\s*', original, re.IGNORECASE).group(0)
base = f"{random.randint(1,9)}{_random_digits(8)}"
return prefix + base + _checksum_inn10(base)
def generate_inn12(original: str) -> str:
prefix = re.match(r'ИНН\s*', original, re.IGNORECASE).group(0)
base = f"{random.randint(1,9)}{_random_digits(9)}"
return prefix + base + _checksum_inn12(base)
def generate_ogrn(original: str) -> str:
prefix = re.match(r'ОГРН\s*', original, re.IGNORECASE).group(0)
base = "1" + _random_digits(11)
return prefix + base + _checksum_ogrn(base)
def generate_kpp(original: str) -> str:
prefix = re.match(r'КПП\s*', original, re.IGNORECASE).group(0)
return prefix + _random_digits(4) + random.choice(["01", "43", "77"]) + _random_digits(3)
def generate_bik(original: str) -> str:
prefix = re.match(r'БИК\s*', original, re.IGNORECASE).group(0)
return prefix + "04" + _random_digits(7)
def generate_rs(original: str) -> str:
prefix = re.match(r'(?:р/с|расч[её]тный\s*сч[её]т)\s*', original, re.IGNORECASE).group(0)
return prefix + "40702" + _random_digits(15)
def generate_ks(original: str) -> str:
prefix = re.match(r'(?:к/с|кор/сч|корр?[еи]?спондентский\s*сч[её]т)\s*', original, re.IGNORECASE).group(0)
return prefix + "30101" + _random_digits(15)
def generate_passport(original: str) -> str:
m = re.match(r'(?:паспорт|серия\s+номер)[\s:№]*', original, re.IGNORECASE)
prefix = m.group(0) if m else ""
return prefix + f"{random.randint(10,99)} {random.randint(10,99)} {_random_digits(6)}"
def generate_company(_: str) -> str:
forms = ["ООО", "ЗАО", "АО"]
nouns = ["Технология", "Прогресс", "Гарант", "Стандарт", "Импульс",
"Вектор", "Сфера", "Альянс", "Синтез", "Меридиан", "Спектр", "Формат"]
return f'{random.choice(forms)} «{random.choice(nouns)}»'
def generate_person(_: str) -> str:
s = random.choice(RU_SURNAMES)
n = random.choice(RU_NAMES)
p = random.choice(RU_PATRONYMICS)
return f"{s} {n[0]}.{p[0]}."
def generate_address(_: str) -> str:
city = random.choice(RU_CITIES)
street = random.choice(RU_STREETS)
house = random.randint(1, 200)
return f"{city}, ул. {street}, д. {house}"
# ═══════════════════════════════════════════
# Основной класс
# ═══════════════════════════════════════════
class TwoPassObfuscator:
"""Двухпроходный обфускатор: сбор сущностей → замена."""
def __init__(self, llm_client=None):
self._mapping: Dict[str, str] = {} # оригинал → замена (только для точных текстовых совпадений)
self._regex_replacements: List[Tuple[str, str, Callable]] = [] # (pattern, type, generator)
self._sorted_keys: List[str] = [] # ключи mapping, отсортированные по длине (убывание)
self._llm_client = llm_client
def obfuscate(self, files: List[Tuple[str, bytes, str]]) -> Tuple[bytes, str]:
"""
Главная точка входа.
Args:
files: [(original_filename, content_bytes, content_type), ...]
Returns:
(zip_bytes, mapping_csv_string)
"""
# --- Распаковать ZIP-файлы ---
files = self._expand_zips(files)
# --- Конвертировать PDF → DOCX ---
files = self._convert_pdfs_to_docx(files)
try:
# --- Проход 1: сбор сущностей ---
all_texts: Dict[str, str] = {}
all_docx: Dict[str, object] = {}
for fname, content, ctype in files:
text, doc = self._extract_text(fname, content, ctype)
all_texts[fname] = text
if doc is not None:
all_docx[fname] = doc
if text and text != "[DOC binary — not parsed]":
self._scan_regex(text)
if self._llm_client:
self._scan_llm_ner(all_texts)
# Предсортировать ключи один раз для _apply_replacements
self._sorted_keys = sorted(self._mapping.keys(), key=len, reverse=True)
# --- Проход 2: замена ---
results = []
for fname, content, ctype in files:
obf_content = content
if fname.endswith('.doc'):
# .doc — бинарный, оставляем как есть
pass
elif fname in all_docx:
obf_content = self._replace_in_docx(all_docx[fname])
else:
txt = all_texts.get(fname, '')
obf_content = self._replace_in_text(txt, fname)
results.append((fname, obf_content))
csv_str = self._build_mapping_csv()
return self._build_zip(results, csv_str), csv_str
finally:
self._mapping.clear()
self._regex_replacements.clear()
self._sorted_keys.clear()
def _convert_pdfs_to_docx(self, files: List[Tuple[str, bytes, str]]) -> List[Tuple[str, bytes, str]]:
"""Конвертировать PDF в DOCX через pdfplumber. При совпадении имён — _из_pdf."""
import pdfplumber
from docx import Document as DocxDocument
result = []
existing_names = {f[0] for f in files}
for fname, content, ctype in files:
if not fname.lower().endswith('.pdf'):
result.append((fname, content, ctype))
continue
try:
doc = DocxDocument()
with pdfplumber.open(io.BytesIO(content)) as pdf:
for page in pdf.pages:
tables = page.extract_tables()
for table in tables:
if table:
rows = [[str(c or "").strip() for c in (row or [])] for row in table]
rows = [r for r in rows if any(r)]
if rows:
t = doc.add_table(rows=len(rows), cols=len(rows[0]))
t.style = 'Table Grid'
for ri, row in enumerate(rows):
for ci, cell_text in enumerate(row):
t.rows[ri].cells[ci].text = cell_text
text = page.extract_text()
if text:
for line in text.split('\n'):
line = line.strip()
if line:
doc.add_paragraph(line)
buf = io.BytesIO()
doc.save(buf)
new_name = fname[:-4] + '.docx'
if new_name in existing_names:
new_name = fname[:-4] + '_из_pdf.docx'
existing_names.add(new_name)
result.append((new_name, buf.getvalue(), ctype))
except Exception as e:
log.warning("PDF→DOCX error for %s: %s", fname, e)
result.append((fname, content, ctype))
return result
def _expand_zips(self, files: List[Tuple[str, bytes, str]]) -> List[Tuple[str, bytes, str]]:
"""Распаковать ZIP-файлы, заменив их содержимым. Остальные файлы — как есть.
Защита от ZIP-бомб: ratio + накопительный размер (как в compare/unzip.py)."""
result = []
for fname, content, ctype in files:
if fname.lower().endswith('.zip'):
try:
with zipfile.ZipFile(io.BytesIO(content)) as zf:
if len(zf.infolist()) > 500:
log.warning("ZIP too many files, skipping: %s", fname)
result.append((fname, content, ctype))
continue
total_uncompressed = 0
for info in zf.infolist():
if info.is_dir():
continue
# ZIP bomb: ratio check
if info.compress_size > 0:
ratio = info.file_size / info.compress_size
if ratio > 100:
log.warning("ZIP bomb ratio %.0f:1, skipping: %s", ratio, fname)
result.append((fname, content, ctype))
break
# cp437 → utf8 (как в compare/unzip.py)
name = info.filename
try:
name = name.encode("cp437").decode("utf-8", errors="replace")
except (UnicodeDecodeError, UnicodeEncodeError):
pass
# Убрать path traversal
name = os.path.basename(name)
if not name or name.endswith("/") or ".." in name or "/" in name or "\\" in name:
continue
inner_data = zf.read(info)
total_uncompressed += len(inner_data)
if total_uncompressed > 500 * 1024 * 1024: # 500 MB
log.warning("ZIP uncompressed limit exceeded, stopping: %s", fname)
break
result.append((name, inner_data, ""))
except Exception as e:
log.warning("Failed to expand ZIP %s: %s", fname, e)
result.append((fname, content, ctype))
else:
result.append((fname, content, ctype))
return result
# --- Проход 1: обнаружение ---
def _extract_text(self, fname: str, content: bytes, ctype: str) -> Tuple[str, Optional[object]]:
"""Извлечь текст из файла. Возвращает (text, docx_document_or_None)."""
doc = None
text = ""
ext = os.path.splitext(fname)[1].lower()
if ext == '.docx':
try:
from docx import Document
except ImportError:
text = content.decode('utf-8', errors='replace')
return text, None
doc = Document(io.BytesIO(content))
text = "\n".join(p.text for p in doc.paragraphs)
for table in doc.tables:
for row in table.rows:
text += "\n" + " | ".join(cell.text for cell in row.cells)
elif ext == '.pdf':
try:
import pdfplumber
except ImportError:
text = content.decode('utf-8', errors='replace')
return text, None
with pdfplumber.open(io.BytesIO(content)) as pdf:
for page in pdf.pages:
t = page.extract_text()
if t:
text += t + "\n"
for table in page.extract_tables():
for row in table:
text += "\n" + " | ".join(str(c) if c else "" for c in row)
elif ext == '.doc':
# .doc — бинарный формат, без libreoffice не парсим
# Пропускаем без изменений (не обфусцируем)
text = "[DOC binary — not parsed]"
return text, None
else:
text = content.decode('utf-8', errors='replace')
return text, doc
def _scan_regex(self, text: str):
"""Сканировать текст regex-паттернами, заполнить словарь замен."""
for entity_type, pattern in ENTITY_PATTERNS.items():
for match in re.finditer(pattern, text, re.IGNORECASE | re.MULTILINE):
original = match.group(0).strip()
if original and original not in self._mapping:
generator = ENTITY_GENERATORS.get(entity_type, lambda x: "XXX")
self._mapping[original] = generator(original)
# Компании (кроме НУБЕС — это Исполнитель)
for match in COMPANY_PATTERN.finditer(text):
original = match.group(0).strip()
if original and original not in self._mapping:
if re.search(r'НУБЕС|NUBES', original, re.IGNORECASE):
continue # Исполнитель — не заменяем
self._mapping[original] = generate_company(original)
# ФИО (Фамилия И.О.)
for match in PERSON_PATTERN.finditer(text):
original = match.group(0).strip()
if original and original not in self._mapping:
self._mapping[original] = generate_person(original)
def _scan_llm_ner(self, all_texts: Dict[str, str]):
"""LLM NER для обнаружения имён и адресов во всех файлах."""
combined = "\n\n---FILE---\n\n".join(
f"FILE: {fname}\n{t[:3000]}" for fname, t in all_texts.items()
)
prompt = (
"Ты — система обнаружения персональных данных в документах. "
"Найди ВСЕ следующие сущности в тексте ниже:\n\n"
"1. ФИО (полные и сокращённые — 'Иванов И.И.', 'Петров А.С.')\n"
"2. Названия компаний-контрагентов (не 'НУБЕС')\n"
"3. Почтовые адреса\n"
"4. Паспортные данные\n\n"
"Формат ответа — JSON-массив:\n"
'[{"type": "person"|"company"|"address"|"passport", "value": "найденный текст"}]\n\n'
f"Текст:\n{combined[:8000]}"
)
try:
raw = self._llm_client.complete(prompt)
import json
# Игнорируем markdown-обёртку
raw = raw.strip()
if raw.startswith("```"):
raw = raw.split("\n", 1)[1]
if raw.endswith("```"):
raw = raw[:-3]
entities = json.loads(raw)
for ent in entities:
val = ent.get("value", "").strip()
if val and val not in self._mapping:
if ent.get("type") == "person":
self._mapping[val] = generate_person(val)
elif ent.get("type") == "company":
self._mapping[val] = generate_company(val)
elif ent.get("type") == "address":
self._mapping[val] = generate_address(val)
elif ent.get("type") == "passport":
self._mapping[val] = generate_passport(val)
except Exception as e:
log.warning("LLM NER failed: %s", e)
# --- Проход 2: замена ---
def _replace_in_docx(self, doc) -> bytes:
"""Заменить сущности в docx. Склеиваем runs → заменяем → пишем в первый run, очищаем остальные."""
for para in doc.paragraphs:
if not para.runs:
continue
full_text = "".join(run.text for run in para.runs)
replaced = self._apply_replacements(full_text)
if replaced != full_text:
para.runs[0].text = replaced
for run in para.runs[1:]:
run.text = ""
for table in doc.tables:
for row in table.rows:
for cell in row.cells:
for para in cell.paragraphs:
if not para.runs:
continue
full_text = "".join(run.text for run in para.runs)
replaced = self._apply_replacements(full_text)
if replaced != full_text:
para.runs[0].text = replaced
for run in para.runs[1:]:
run.text = ""
buf = io.BytesIO()
doc.save(buf)
return buf.getvalue()
def _replace_in_text(self, text: str, fname: str) -> bytes:
"""Заменить сущности в plain text (для PDF и прочих)."""
replaced = self._apply_replacements(text)
# Для PDF пока отдаём текст (MVP — без сохранения форматирования PDF)
return replaced.encode('utf-8')
def _apply_replacements(self, text: str) -> str:
"""Применить все замены из словаря mapping к строке. Сначала длинные, потом короткие."""
result = text
for original in self._sorted_keys:
replacement = self._mapping[original]
# Границы слова — если сущность начинается и заканчивается на \w
if original and original[0].isalnum() and original[-1].isalnum():
pattern = r'(?<!\w)' + re.escape(original) + r'(?!\w)'
else:
pattern = re.escape(original)
result = re.sub(pattern, replacement, result)
return result
# --- Сборка выдачи ---
def _build_zip(self, files: List[Tuple[str, bytes]], mapping_csv: str = "") -> bytes:
"""Собрать ZIP с обфусцированными файлами + mapping.csv."""
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
for fname, content in files:
info = zipfile.ZipInfo(fname)
info.flag_bits |= 0x800 # UTF-8 filename
zf.writestr(info, content)
if mapping_csv:
zf.writestr("mapping.csv", '\ufeff'.encode('utf-8') + mapping_csv.encode('utf-8'))
return buf.getvalue()
def _build_mapping_csv(self) -> str:
"""Собрать mapping.csv."""
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(["тип_данных", "оригинал", "замена"])
for original, replacement in sorted(self._mapping.items()):
etype = "text"
# inn_fl ДО inn_ul (12 цифр vs 10)
for t in ["phone", "email", "inn_fl", "inn_ul", "ogrn", "kpp", "bik", "rs", "ks", "passport"]:
pat = ENTITY_PATTERNS.get(t, "")
if pat and re.match(pat, original, re.IGNORECASE):
etype = t
break
if COMPANY_PATTERN.match(original):
etype = "company"
writer.writerow([etype, original, replacement])
return buf.getvalue()
# ═══════════════════════════════════════════
# Маппинг entity_type → генератор
# ═══════════════════════════════════════════
ENTITY_GENERATORS: Dict[str, Callable] = {
"phone": generate_phone,
"email": generate_email,
"inn_ul": generate_inn10,
"inn_fl": generate_inn12,
"ogrn": generate_ogrn,
"kpp": generate_kpp,
"bik": generate_bik,
"rs": generate_rs,
"ks": generate_ks,
"passport": generate_passport,
}
def obfuscate_files(files: List[Tuple[str, bytes, str]], llm_client=None) -> Tuple[bytes, str]:
"""Удобная функция: обфусцировать список файлов → (zip_bytes, csv_string)."""
obf = TwoPassObfuscator(llm_client=llm_client)
return obf.obfuscate(files)
+13
View File
@@ -0,0 +1,13 @@
"""
DrHider — обфускация документов (двухпроходная, в памяти, без БД).
Проход 1: собрать все сущности из всех файлов → глобальный словарь замен.
Проход 2: применить замены → собрать ZIP с обфусцированными файлами + mapping.csv.
Согласованность: одна и та же сущность во всех файлах → одно и то же фиктивное значение.
"""
from .obfuscator import TwoPassObfuscator, obfuscate_files
from .llm_client import LLMClient
__all__ = ["TwoPassObfuscator", "obfuscate_files", "LLMClient"]
+96
View File
@@ -0,0 +1,96 @@
"""
Сборка результата обфускации.
Два метода:
1. _build_zip — упаковка обфусцированных файлов в ZIP
2. _build_mapping_csv — генерация mapping.csv с таблицей замен
"""
import io
import csv
import zipfile
import re
from typing import Dict, List, Tuple
from .config import ENTITY_PATTERNS, COMPANY_PATTERN
def build_zip(files: List[Tuple[str, bytes]], mapping_csv: str = "") -> bytes:
"""Собрать ZIP-архив с обфусцированными файлами.
В архив добавляются:
- Все обфусцированные файлы (с оригинальными именами)
- mapping.csv — таблица соответствия оригинал→замена (если не пустая)
Имена файлов в архиве — UTF-8 (бит 11 в flag_bits).
Args:
files: [(filename, content_bytes), ...]
mapping_csv: Строка CSV с таблицей замен (опционально)
Returns:
Бинарное содержимое ZIP-архива
"""
buf = io.BytesIO()
with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
# Добавляем обфусцированные файлы
for fname, content in files:
info = zipfile.ZipInfo(fname)
info.flag_bits |= 0x800 # Флаг: имя файла в UTF-8
zf.writestr(info, content)
# Добавляем mapping.csv с BOM (для корректного открытия в Excel)
if mapping_csv:
zf.writestr(
"mapping.csv",
'\ufeff'.encode('utf-8') + mapping_csv.encode('utf-8'),
)
return buf.getvalue()
def build_mapping_csv(mapping: Dict[str, str]) -> str:
"""Собрать mapping.csv — таблицу соответствия оригинал → замена.
Колонки:
- тип_данных: тип сущности (phone, email, inn_ul, company, ...)
- оригинал: исходное значение из документа
- замена: фиктивное значение
Тип определяется проверкой каждого значения через ENTITY_PATTERNS
и COMPANY_PATTERN. Если ни один паттерн не подошёл — тип "text".
Args:
mapping: Словарь {оригинал: замена}
Returns:
Строка в формате CSV
"""
buf = io.StringIO()
writer = csv.writer(buf)
writer.writerow(["тип_данных", "оригинал", "замена"])
# Порядок проверки типов: inn_fl ДО inn_ul (12 цифр vs 10)
type_order = [
"phone", "email", "inn_fl", "inn_ul", "ogrn",
"kpp", "bik", "rs", "ks", "passport",
]
for original, replacement in sorted(mapping.items()):
# Определяем тип сущности
etype = "text" # По умолчанию
for t in type_order:
pat = ENTITY_PATTERNS.get(t, "")
if pat and re.match(pat, original, re.IGNORECASE):
etype = t
break
# Компании проверяем отдельно (COMPANY_PATTERN не в ENTITY_PATTERNS)
if COMPANY_PATTERN.match(original):
etype = "company"
writer.writerow([etype, original, replacement])
return buf.getvalue()
+69
View File
@@ -0,0 +1,69 @@
"""
Контрольные суммы для ИНН (10 и 12 знаков) и ОГРН.
Используются генераторами фиктивных значений для создания
синтаксически корректных номеров.
"""
def checksum_inn10(inn: str) -> str:
"""Контрольная сумма для 10-значного ИНН (юридические лица).
Алгоритм:
1. Умножаем первые 9 цифр на коэффициенты [2,4,10,3,5,9,4,6,8]
2. Суммируем, берём (сумма % 11) % 10
Args:
inn: 9 цифр ИНН (без контрольной суммы)
Returns:
Одна цифра контрольной суммы
"""
coeffs = [2, 4, 10, 3, 5, 9, 4, 6, 8]
s = sum(int(inn[i]) * coeffs[i] for i in range(9))
return str((s % 11) % 10)
def checksum_inn12(inn: str) -> str:
"""Контрольные суммы для 12-значного ИНН (физические лица).
Алгоритм (две контрольные цифры):
1. n1: умножаем первые 10 цифр на [7,2,4,10,3,5,9,4,6,8],
берём (сумма % 11) % 10
2. n2: добавляем n1 к числу, умножаем 11 цифр на
[3,7,2,4,10,3,5,9,4,6,8], берём (сумма % 11) % 10
Args:
inn: 10 цифр ИНН (без контрольных сумм)
Returns:
Две цифры контрольных сумм (n1 + n2)
"""
c1 = [7, 2, 4, 10, 3, 5, 9, 4, 6, 8]
c2 = [3, 7, 2, 4, 10, 3, 5, 9, 4, 6, 8]
# Первая контрольная цифра
s1 = sum(int(inn[i]) * c1[i] for i in range(10))
n1 = (s1 % 11) % 10
# Вторая контрольная цифра (с учётом первой)
inn2 = inn + str(n1)
s2 = sum(int(inn2[i]) * c2[i] for i in range(11))
n2 = (s2 % 11) % 10
return str(n1) + str(n2)
def checksum_ogrn(ogrn: str) -> str:
"""Контрольная сумма для ОГРН (13 знаков).
Алгоритм: первые 12 цифр как число % 11, затем % 10.
Args:
ogrn: 12 цифр ОГРН (без контрольной суммы)
Returns:
Одна цифра контрольной суммы
"""
s = int(ogrn) % 11
return str(s % 10)
+88
View File
@@ -0,0 +1,88 @@
"""
Константы и словари для DrHider.
Содержит:
- ENTITY_PATTERNS — regex-паттерны для обнаружения сущностей
- COMPANY_PATTERN — паттерн для обнаружения названий компаний
- PERSON_PATTERN — паттерн для обнаружения ФИО
- Словари русских имён, городов, улиц
- Список фиктивных доменов для email
"""
import re
# ═══════════════════════════════════════════════════════════════════════════
# Regex-паттерны для обнаружения сущностей в тексте документов
# ═══════════════════════════════════════════════════════════════════════════
ENTITY_PATTERNS = {
"phone": r'(?<!\d)(?:\+7|8)[\s\-]?\(?\d{3}\)?[\s\-]?\d{3}[\s\-]?\d{2}[\s\-]?\d{2}(?!\d)',
"email": r'\b[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}\b',
# 12-значный ИНН ДО 10-значного — иначе 12-значный матчится как 10-значный
"inn_fl": r'ИНН\s*\d{12}',
"inn_ul": r'ИНН\s*\d{10}',
"ogrn": r'ОГРН\s*\d{13}',
"kpp": r'КПП\s*\d{9}',
"bik": r'БИК\s*\d{9}',
"rs": r'(?:р/с|расч[её]тный\s*сч[её]т)\s*\d{20}',
"ks": r'(?:к/с|кор/сч|корр?[еи]?спондентский\s*сч[её]т)\s*\d{20}',
"passport": r'(?:паспорт|серия\s+номер)[\s:№]*\d{2}\s*\d{2}\s*\d{6,7}',
}
# ═══════════════════════════════════════════════════════════════════════════
# Паттерны для LLM-независимого обнаружения (regex)
# ═══════════════════════════════════════════════════════════════════════════
# Названия компаний: ООО/ЗАО/АО/ПАО/ИП/ТОО + название в кавычках или без
COMPANY_PATTERN = re.compile(
r'(?:ООО|ЗАО|ОАО|АО|ПАО|ИП|ТОО)\s+(?:«[^»]+»|"[^"]+"|\u201c[^\u201d]+\u201d|[А-ЯA-Z][\w\-\.]{2,40})',
re.IGNORECASE,
)
# ФИО: Фамилия + инициалы (Иванов И.И.) или полное имя (Иванов Иван Иванович)
PERSON_PATTERN = re.compile(
r'\b[А-Я][а-я]+\s+[А-Я]\.[А-Я]\.'
r'|\b[А-Я][а-я]+\s+[А-Я][а-я]+\s+[А-Я][а-я]+',
)
# ═══════════════════════════════════════════════════════════════════════════
# Словари для генерации фиктивных значений
# ═══════════════════════════════════════════════════════════════════════════
# Русские фамилии (20 самых распространённых)
RU_SURNAMES = [
"Иванов", "Смирнов", "Кузнецов", "Попов", "Васильев", "Петров",
"Соколов", "Михайлов", "Новиков", "Фёдоров", "Морозов", "Волков",
"Алексеев", "Лебедев", "Семёнов", "Егоров", "Павлов", "Козлов",
"Степанов", "Николаев",
]
# Русские мужские имена
RU_NAMES = [
"Александр", "Дмитрий", "Сергей", "Андрей", "Алексей", "Максим",
"Евгений", "Иван", "Михаил", "Николай", "Владимир", "Павел",
"Виктор", "Олег",
]
# Русские отчества
RU_PATRONYMICS = [
"Александрович", "Дмитриевич", "Сергеевич", "Андреевич",
"Алексеевич", "Иванович", "Михайлович", "Николаевич", "Владимирович",
"Павлович", "Викторович", "Олегович", "Евгеньевич", "Максимович",
]
# Города России (15 крупнейших)
RU_CITIES = [
"Москва", "Санкт-Петербург", "Новосибирск", "Екатеринбург",
"Казань", "Нижний Новгород", "Челябинск", "Самара", "Омск",
"Ростов-на-Дону", "Уфа", "Красноярск", "Воронеж", "Пермь", "Волгоград",
]
# Названия улиц
RU_STREETS = [
"Ленина", "Мира", "Пушкина", "Гагарина", "Советская",
"Кирова", "Октябрьская", "Молодёжная", "Садовая", "Центральная",
]
# Фиктивные домены для генерации email
FAKE_DOMAINS = ["example.ru", "mail.test", "company.local", "org.example.ru"]
+260
View File
@@ -0,0 +1,260 @@
"""
Извлечение текста из документов разных форматов.
Поддерживает:
- .docx (через python-docx)
- .pdf (через pdfplumber)
- .doc (бинарный — не парсится)
- .txt и прочие (как UTF-8)
Также содержит:
- _expand_zips — распаковка ZIP с защитой от ZIP-бомб
- _convert_pdfs_to_docx — конвертация PDF → DOCX
"""
import io
import os
import zipfile
import logging
from typing import Dict, List, Tuple, Optional
log = logging.getLogger("drhider")
def extract_text(fname: str, content: bytes, ctype: str) -> Tuple[str, Optional[object]]:
"""Извлечь текст из одного файла.
Args:
fname: Имя файла (с расширением)
content: Бинарное содержимое файла
ctype: MIME-тип (не используется в текущей версии)
Returns:
(text, docx_document_or_None):
text — извлечённый текст (строка)
doc — объект python-docx Document или None
"""
doc = None
text = ""
ext = os.path.splitext(fname)[1].lower()
# ── .docx: извлекаем текст + сохраняем Document для замен с форматированием ──
if ext == '.docx':
try:
from docx import Document
except ImportError:
# Если python-docx не установлен — читаем как plain text
text = content.decode('utf-8', errors='replace')
return text, None
doc = Document(io.BytesIO(content))
# Собираем текст из параграфов
paragraphs = [p.text for p in doc.paragraphs]
# Добавляем текст из таблиц
for table in doc.tables:
for row in table.rows:
row_text = " | ".join(cell.text for cell in row.cells)
paragraphs.append(row_text)
text = "\n".join(paragraphs)
# ── .pdf: извлекаем текст через pdfplumber ──
elif ext == '.pdf':
try:
import pdfplumber
except ImportError:
text = content.decode('utf-8', errors='replace')
return text, None
parts = []
with pdfplumber.open(io.BytesIO(content)) as pdf:
for page in pdf.pages:
# Текст страницы
t = page.extract_text()
if t:
parts.append(t)
# Текст из таблиц
for table in page.extract_tables():
for row in table:
row_str = " | ".join(str(c) if c else "" for c in row)
parts.append(row_str)
text = "\n".join(parts)
# ── .doc: бинарный формат — без libreoffice не парсим ──
elif ext == '.doc':
text = "[DOC binary — not parsed]"
return text, None
# ── .txt и прочие: читаем как UTF-8 ──
else:
text = content.decode('utf-8', errors='replace')
return text, doc
def expand_zips(files: List[Tuple[str, bytes, str]]) -> List[Tuple[str, bytes, str]]:
"""Распаковать ZIP-файлы в списке, заменив их содержимым.
Не-ZIP файлы проходят без изменений.
Защита от ZIP-бомб:
- Максимум 500 файлов в архиве
- Ratio file_size/compress_size не более 100:1
- Накопительный размер распакованных данных не более 500 MB
Args:
files: [(filename, content_bytes, content_type), ...]
Returns:
Новый список файлов (ZIP раскрыты, остальные как есть)
"""
result: List[Tuple[str, bytes, str]] = []
for fname, content, ctype in files:
# Пропускаем не-ZIP
if not fname.lower().endswith('.zip'):
result.append((fname, content, ctype))
continue
try:
with zipfile.ZipFile(io.BytesIO(content)) as zf:
# Проверка: не более 500 файлов в архиве
if len(zf.infolist()) > 500:
log.warning("ZIP too many files, skipping: %s", fname)
result.append((fname, content, ctype))
continue
total_uncompressed = 0
for info in zf.infolist():
# Пропускаем директории
if info.is_dir():
continue
# Проверка на ZIP-бомбу: ratio
if info.compress_size > 0:
ratio = info.file_size / info.compress_size
if ratio > 100: # Файл сжимается более чем в 100 раз
log.warning(
"ZIP bomb ratio %.0f:1, skipping: %s", ratio, fname
)
result.append((fname, content, ctype))
break # Пропускаем весь архив
# Декодируем имя файла: cp437 → utf-8
name = info.filename
try:
name = name.encode("cp437").decode("utf-8", errors="replace")
except (UnicodeDecodeError, UnicodeEncodeError):
pass
# Защита от path traversal
name = os.path.basename(name)
if (
not name
or name.endswith("/")
or ".." in name
or "/" in name
or "\\" in name
):
continue
# Читаем и проверяем накопительный размер
inner_data = zf.read(info)
total_uncompressed += len(inner_data)
if total_uncompressed > 500 * 1024 * 1024: # 500 MB
log.warning(
"ZIP uncompressed limit exceeded, stopping: %s", fname
)
break
result.append((name, inner_data, ""))
except Exception as e:
log.warning("Failed to expand ZIP %s: %s", fname, e)
result.append((fname, content, ctype))
return result
def convert_pdfs_to_docx(
files: List[Tuple[str, bytes, str]],
) -> List[Tuple[str, bytes, str]]:
"""Конвертировать PDF-файлы в DOCX через pdfplumber.
Не-PDF файлы проходят без изменений.
При совпадении имён к имени добавляется суффикс '_из_pdf'.
Конвертация:
- Текст страницы → параграфы DOCX
- Таблицы → таблицы DOCX со стилем 'Table Grid'
Args:
files: [(filename, content_bytes, content_type), ...]
Returns:
Новый список файлов (PDF заменены на DOCX)
"""
import pdfplumber
from docx import Document as DocxDocument
result: List[Tuple[str, bytes, str]] = []
existing_names = {f[0] for f in files}
for fname, content, ctype in files:
# Пропускаем не-PDF
if not fname.lower().endswith('.pdf'):
result.append((fname, content, ctype))
continue
try:
doc = DocxDocument()
with pdfplumber.open(io.BytesIO(content)) as pdf:
for page in pdf.pages:
# ── Таблицы ──
tables = page.extract_tables()
for table in tables:
if not table:
continue
# Чистим строки: убираем полностью пустые
rows = [
[str(c or "").strip() for c in (row or [])]
for row in table
]
rows = [r for r in rows if any(r)]
if rows:
t = doc.add_table(rows=len(rows), cols=len(rows[0]))
t.style = 'Table Grid'
for ri, row in enumerate(rows):
for ci, cell_text in enumerate(row):
t.rows[ri].cells[ci].text = cell_text
# ── Текст ──
text = page.extract_text()
if text:
for line in text.split('\n'):
line = line.strip()
if line:
doc.add_paragraph(line)
# Сохраняем DOCX в буфер
buf = io.BytesIO()
doc.save(buf)
# Формируем новое имя
new_name = fname[:-4] + '.docx'
if new_name in existing_names:
new_name = fname[:-4] + '_из_pdf.docx'
existing_names.add(new_name)
result.append((new_name, buf.getvalue(), ctype))
except Exception as e:
log.warning("PDF→DOCX error for %s: %s", fname, e)
# При ошибке — оставляем оригинальный PDF
result.append((fname, content, ctype))
return result
+37
View File
@@ -0,0 +1,37 @@
"""
Маппинг типов сущностей на генераторы фиктивных значений.
ENTITY_GENERATORS используется в TwoPassObfuscator._scan_regex()
для автоматического выбора генератора по типу найденной сущности.
"""
from .phone import generate_phone
from .email import generate_email
from .inn import generate_inn10, generate_inn12
from .ogrn import generate_ogrn
from .kpp import generate_kpp
from .bik import generate_bik
from .accounts import generate_rs, generate_ks
from .passport import generate_passport
from .fallback import generate_fallback
# ═══════════════════════════════════════════════════════════════════════════
# Маппинг: entity_type → функция-генератор
# Ключи соответствуют ключам в config.ENTITY_PATTERNS
# ═══════════════════════════════════════════════════════════════════════════
ENTITY_GENERATORS = {
"phone": generate_phone,
"email": generate_email,
"inn_ul": generate_inn10, # 10-значный ИНН → юрлица
"inn_fl": generate_inn12, # 12-значный ИНН → физлица
"ogrn": generate_ogrn,
"kpp": generate_kpp,
"bik": generate_bik,
"rs": generate_rs, # расчётный счёт
"ks": generate_ks, # корреспондентский счёт
"passport": generate_passport,
}
# Fallback-генератор — экспортируется отдельно для использования в scanner/obfuscator
FALLBACK_GENERATOR = generate_fallback
+47
View File
@@ -0,0 +1,47 @@
"""
Генераторы фиктивных банковских счетов.
- generate_rs: расчётный счёт (20 знаков)
- generate_ks: корреспондентский счёт (20 знаков)
Счета начинаются с реальных префиксов: 40702 (расчётный), 30101 (корреспондентский).
"""
import re
from ..random_utils import random_digits
def generate_rs(original: str) -> str:
"""Сгенерировать фиктивный 20-значный расчётный счёт.
Сохраняет оригинальный префикс (р/с, расчётный счёт и т.д.).
Начинается с 40702 (реальный префикс расчётного счёта).
Args:
original: Оригинальная строка с расчётным счётом
Returns:
Строка вида «р/с 40702XXXXXXXXXXXXXXX»
"""
prefix = re.match(r'(?:р/с|расч[её]тный\s*сч[её]т)\s*', original, re.IGNORECASE).group(0)
return prefix + "40702" + random_digits(15)
def generate_ks(original: str) -> str:
"""Сгенерировать фиктивный 20-значный корреспондентский счёт.
Сохраняет оригинальный префикс (к/с, кор/счёт и т.д.).
Начинается с 30101 (реальный префикс корр. счёта).
Args:
original: Оригинальная строка с корр. счётом
Returns:
Строка вида «к/с 30101XXXXXXXXXXXXXXX»
"""
prefix = re.match(
r'(?:к/с|кор/сч|корр?[еи]?спондентский\s*сч[её]т)\s*',
original,
re.IGNORECASE,
).group(0)
return prefix + "30101" + random_digits(15)
+25
View File
@@ -0,0 +1,25 @@
"""
Генератор фиктивных почтовых адресов.
Формат: Город, ул. Улица, д. Номер
"""
import random
from ..config import RU_CITIES, RU_STREETS
def generate_address(_: str) -> str:
"""Сгенерировать фиктивный почтовый адрес.
Выбирает случайный город, улицу и номер дома из словарей.
Args:
_: Оригинальный адрес (игнорируется)
Returns:
Строка вида «Москва, ул. Ленина, д. 42»
"""
city = random.choice(RU_CITIES)
street = random.choice(RU_STREETS)
house = random.randint(1, 200)
return f"{city}, ул. {street}, д. {house}"
+25
View File
@@ -0,0 +1,25 @@
"""
Генератор фиктивного БИК (9 знаков).
Формат: «БИК » + 9 цифр.
Первые 2 цифры — 04 (код РФ).
"""
import re
from ..random_utils import random_digits
def generate_bik(original: str) -> str:
"""Сгенерировать фиктивный 9-значный БИК.
Сохраняет оригинальный префикс (например, «БИК »).
Начинается с 04 (код Российской Федерации).
Args:
original: Оригинальная строка с БИК
Returns:
Строка вида «БИК 04XXXXXXX»
"""
prefix = re.match(r'БИК\s*', original, re.IGNORECASE).group(0)
return prefix + "04" + random_digits(7)
+27
View File
@@ -0,0 +1,27 @@
"""
Генератор фиктивных названий компаний.
Формат: ООО/ЗАО/АО «СлучайноеСуществительное»
"""
import random
def generate_company(_: str) -> str:
"""Сгенерировать фиктивное название компании.
Выбирает случайную организационно-правовую форму (ООО, ЗАО, АО)
и случайное существительное из словаря.
Args:
_: Оригинальное название (игнорируется)
Returns:
Строка вида «ООО «Технология»»
"""
forms = ["ООО", "ЗАО", "АО"]
nouns = [
"Технология", "Прогресс", "Гарант", "Стандарт", "Импульс",
"Вектор", "Сфера", "Альянс", "Синтез", "Меридиан", "Спектр", "Формат",
]
return f'{random.choice(forms)} «{random.choice(nouns)}»'
+23
View File
@@ -0,0 +1,23 @@
"""
Генератор фиктивного email-адреса.
Сохраняет структуру оригинального email: локальная_часть@домен.
"""
import random
from ..config import FAKE_DOMAINS
from ..random_utils import random_letters
def generate_email(original: str) -> str:
"""Сгенерировать фиктивный email с тем же форматом.
Args:
original: Оригинальный email (для сохранения структуры, не используется)
Returns:
Фиктивный email вида случайные_буквы@фиктивный_домен
"""
domain = random.choice(FAKE_DOMAINS)
local = random_letters(random.randint(5, 10))
return f"{local}@{domain}"
+64
View File
@@ -0,0 +1,64 @@
"""
Fallback-генератор фиктивных значений для неизвестных типов сущностей.
Используется когда LLM находит сущность с типом, которого нет
в специализированных генераторах (например, "contract_number", "employee_id").
Принцип: character-class preserving замена.
- Цифры → случайные цифры
- Буквы → случайные буквы (того же алфавита)
- Пробелы/знаки препинания → сохраняются
- Длина строки сохраняется
"""
import random
import string
def generate_fallback(original: str) -> str:
"""Сгенерировать фиктивное значение, сохраняя структуру оригинала.
Правила замены:
- [0-9] → случайная цифра
- [A-Za-z] → случайная буква того же регистра
- [А-Яа-я] → случайная кириллическая буква того же регистра
- Все остальные символы (пробелы, дефисы, спецсимволы) → без изменений
Args:
original: Оригинальное значение сущности
Returns:
Фиктивное значение той же длины и структуры
Example:
"Договор №123/2024""Фтщшлпь №847/5921"
"EMP-0042""XQJ-8193"
"""
result = []
for ch in original:
if ch.isdigit():
# Цифра → случайная цифра
result.append(random.choice(string.digits))
elif 'A' <= ch <= 'Z':
# Заглавная латиница → случайная заглавная латиница
result.append(random.choice(string.ascii_uppercase))
elif 'a' <= ch <= 'z':
# Строчная латиница → случайная строчная латиница
result.append(random.choice(string.ascii_lowercase))
elif 'А' <= ch <= 'Я':
# Заглавная кириллица → случайная заглавная кириллица
result.append(chr(random.randint(0x0410, 0x042F)))
elif 'а' <= ch <= 'я' or ch == 'ё':
# Строчная кириллица → случайная строчная кириллица
result.append(chr(random.randint(0x0430, 0x044F)))
else:
# Всё остальное (пробелы, дефисы, слэши, точки) → сохранить
result.append(ch)
return ''.join(result)
+45
View File
@@ -0,0 +1,45 @@
"""
Генераторы фиктивных ИНН.
- generate_inn10: 10-значный ИНН (юридические лица)
- generate_inn12: 12-значный ИНН (физические лица)
Оба сохраняют префикс «ИНН » и генерируют корректную контрольную сумму.
"""
import random
import re
from ..random_utils import random_digits
from ..checksum import checksum_inn10, checksum_inn12
def generate_inn10(original: str) -> str:
"""Сгенерировать фиктивный 10-значный ИНН (для юридических лиц).
Сохраняет оригинальный префикс (например, «ИНН »).
Args:
original: Оригинальная строка с ИНН
Returns:
Строка вида «ИНН XXXXXXXXX-C» с корректной контрольной суммой
"""
prefix = re.match(r'ИНН\s*', original, re.IGNORECASE).group(0)
base = f"{random.randint(1,9)}{random_digits(8)}"
return prefix + base + checksum_inn10(base)
def generate_inn12(original: str) -> str:
"""Сгенерировать фиктивный 12-значный ИНН (для физических лиц).
Сохраняет оригинальный префикс (например, «ИНН »).
Args:
original: Оригинальная строка с ИНН
Returns:
Строка вида «ИНН XXXXXXXXXX-CC» с двумя корректными контрольными суммами
"""
prefix = re.match(r'ИНН\s*', original, re.IGNORECASE).group(0)
base = f"{random.randint(1,9)}{random_digits(9)}"
return prefix + base + checksum_inn12(base)
+26
View File
@@ -0,0 +1,26 @@
"""
Генератор фиктивного КПП (9 знаков).
Формат: «КПП » + 9 цифр.
Последние 3 цифры — код причины постановки (01, 43 или 77).
"""
import random
import re
from ..random_utils import random_digits
def generate_kpp(original: str) -> str:
"""Сгенерировать фиктивный 9-значный КПП.
Сохраняет оригинальный префикс (например, «КПП »).
Последние 3 цифры — один из реальных кодов причины постановки.
Args:
original: Оригинальная строка с КПП
Returns:
Строка вида «КПП XXXX-код-XXX»
"""
prefix = re.match(r'КПП\s*', original, re.IGNORECASE).group(0)
return prefix + random_digits(4) + random.choice(["01", "43", "77"]) + random_digits(3)
+26
View File
@@ -0,0 +1,26 @@
"""
Генератор фиктивного ОГРН (13 знаков).
Формат: «ОГРН » + 13 цифр с корректной контрольной суммой.
"""
import re
from ..random_utils import random_digits
from ..checksum import checksum_ogrn
def generate_ogrn(original: str) -> str:
"""Сгенерировать фиктивный 13-значный ОГРН.
Сохраняет оригинальный префикс (например, «ОГРН »).
Первая цифра всегда 1 (признак юридического лица).
Args:
original: Оригинальная строка с ОГРН
Returns:
Строка вида «ОГРН 1XXXXXXXXXXX-C» с корректной контрольной суммой
"""
prefix = re.match(r'ОГРН\s*', original, re.IGNORECASE).group(0)
base = "1" + random_digits(11)
return prefix + base + checksum_ogrn(base)
+26
View File
@@ -0,0 +1,26 @@
"""
Генератор фиктивных паспортных данных.
Формат: «паспорт » / «серия номер » + XX XX XXXXXX
"""
import random
import re
from ..random_utils import random_digits
def generate_passport(original: str) -> str:
"""Сгенерировать фиктивные паспортные данные.
Сохраняет оригинальный префикс (паспорт, серия номер и т.д.).
Формат: 2 цифры серии + 2 цифры + 6 цифр номера.
Args:
original: Оригинальная строка с паспортными данными
Returns:
Строка вида «паспорт XX XX XXXXXX»
"""
m = re.match(r'(?:паспорт|серия\s+номер)[\s:№]*', original, re.IGNORECASE)
prefix = m.group(0) if m else ""
return prefix + f"{random.randint(10, 99)} {random.randint(10, 99)} {random_digits(6)}"
+26
View File
@@ -0,0 +1,26 @@
"""
Генератор фиктивных ФИО.
Формат: Фамилия И.О. (фамилия полностью, имя и отчество — инициалы).
"""
import random
from ..config import RU_SURNAMES, RU_NAMES, RU_PATRONYMICS
def generate_person(_: str) -> str:
"""Сгенерировать фиктивное ФИО в формате «Фамилия И.О.».
Выбирает случайные фамилию, имя и отчество из словарей.
Имя и отчество сокращаются до инициалов.
Args:
_: Оригинальное ФИО (игнорируется)
Returns:
Строка вида «Иванов А.И.»
"""
s = random.choice(RU_SURNAMES)
n = random.choice(RU_NAMES)
p = random.choice(RU_PATRONYMICS)
return f"{s} {n[0]}.{p[0]}."
+22
View File
@@ -0,0 +1,22 @@
"""
Генератор фиктивного телефонного номера.
Формат: +7 (код) XXX-XX-XX
Код выбирается случайно из списка реальных российских кодов.
"""
import random
from ..random_utils import random_digits
def generate_phone(_: str) -> str:
"""Сгенерировать фиктивный российский телефонный номер.
Args:
_: Оригинальный номер (игнорируется, нужен для единого интерфейса)
Returns:
Строка в формате +7 (XXX) XXX-XX-XX
"""
code = random.choice(["495", "499", "812", "383", "343"])
return f"+7 ({code}) {random_digits(3)}-{random_digits(2)}-{random_digits(2)}"
+64
View File
@@ -0,0 +1,64 @@
"""
LLM-клиент для DrHider.
Используется в TwoPassObfuscator._scan_llm_ner() для обнаружения
имён, компаний, адресов и паспортных данных через LLM API.
Интерфейс:
client = LLMClient()
result = client.complete(prompt) # str
"""
import os
import httpx
class LLMClient:
"""Клиент для вызова LLM API (aillm.ru / OpenAI-совместимый).
Использует переменные окружения:
LLM_API_KEY или LLM_KEY — ключ API
LLM_URL — URL эндпоинта (по умолчанию https://api.aillm.ru/v1/chat/completions)
LLM_MODEL — модель (по умолчанию gpt-oss-120b)
"""
def __init__(self):
"""Инициализировать клиент. httpx импортируется лениво."""
self._httpx = httpx
def complete(self, prompt: str) -> str:
"""Отправить промпт в LLM и вернуть текст ответа.
Args:
prompt: Текст промпта (инструкция + данные для анализа)
Returns:
Текстовый ответ модели (обычно JSON-строка)
Raises:
httpx.HTTPError: при ошибке HTTP
KeyError: при неожиданном формате ответа
"""
# Ключ API: сначала LLM_KEY, затем LLM_API_KEY (для совместимости)
key = os.environ.get("LLM_KEY") or os.environ.get("LLM_API_KEY", "")
# URL и модель с значениями по умолчанию
url = os.environ.get("LLM_URL", "https://api.aillm.ru/v1/chat/completions")
model = os.environ.get("LLM_MODEL", "gpt-oss-120b")
r = self._httpx.post(
url,
json={
"model": model,
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 8000,
"temperature": 0.1,
},
headers={
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
},
timeout=120,
)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
+148
View File
@@ -0,0 +1,148 @@
"""
Оркестратор двухпроходной обфускации — класс TwoPassObfuscator.
Процесс:
1. Проход 1 (сбор): извлечь текст → regex-сканирование → LLM-сканирование
2. Проход 2 (замена): применить mapping ко всем файлам
3. Сборка: ZIP + mapping.csv
"""
import io
import logging
from typing import Dict, List, Tuple, Callable, Optional
from . import extractor
from . import scanner
from . import replacer
from . import builder
log = logging.getLogger("drhider")
class TwoPassObfuscator:
"""Двухпроходный обфускатор документов.
Проход 1: собрать все сущности из всех файлов → глобальный словарь замен.
Проход 2: применить замены ко всем файлам → ZIP с результатом.
Согласованность: одна и та же сущность во всех файлах получает
одно и то же фиктивное значение.
Attributes:
_mapping: {оригинал: замена} — глобальный словарь
_sorted_keys: ключи mapping, отсортированные по длине (убывание)
_llm_client: опциональный LLM-клиент для NER
"""
def __init__(self, llm_client=None):
"""Инициализировать обфускатор.
Args:
llm_client: Объект с методом .complete(prompt) -> str.
Если None — LLM-сканирование не выполняется.
"""
self._mapping: Dict[str, str] = {}
self._sorted_keys: List[str] = []
self._llm_client = llm_client
# ═══════════════════════════════════════════════════════════════════
# Главная точка входа
# ═══════════════════════════════════════════════════════════════════
def obfuscate(
self, files: List[Tuple[str, bytes, str]]
) -> Tuple[bytes, str]:
"""Обфусцировать список файлов.
Args:
files: [(filename, content_bytes, content_type), ...]
content_type — MIME-тип (может быть пустой строкой)
Returns:
(zip_bytes, csv_string):
zip_bytes — ZIP-архив с обфусцированными файлами + mapping.csv
csv_string — содержимое mapping.csv как строка
"""
# ── Предобработка: распаковать ZIP, конвертировать PDF ──
files = extractor.expand_zips(files)
files = extractor.convert_pdfs_to_docx(files)
try:
# ── Проход 1: сбор сущностей ──
all_texts: Dict[str, str] = {}
all_docx: Dict[str, object] = {}
for fname, content, ctype in files:
text, doc = extractor.extract_text(fname, content, ctype)
all_texts[fname] = text
if doc is not None:
all_docx[fname] = doc
# Regex-сканирование (быстрое, локальное)
if text and text != "[DOC binary — not parsed]":
scanner.scan_regex(text, self._mapping)
# LLM-сканирование (медленное, сетевое — только если есть клиент)
if self._llm_client:
scanner.scan_llm_ner(all_texts, self._mapping, self._llm_client)
# Предсортировать ключи один раз (по убыванию длины)
self._sorted_keys = sorted(
self._mapping.keys(), key=len, reverse=True
)
# ── Проход 2: замена сущностей ──
results: List[Tuple[str, bytes]] = []
for fname, content, ctype in files:
obf_content = content # По умолчанию — без изменений
if fname.endswith('.doc'):
# .doc — бинарный формат, оставляем как есть
pass
elif fname in all_docx:
# DOCX: замена с сохранением форматирования
obf_content = replacer.replace_in_docx(
all_docx[fname], self._mapping, self._sorted_keys
)
else:
# Plain text / PDF-текст: простая замена
txt = all_texts.get(fname, '')
obf_content = replacer.replace_in_text(
txt, fname, self._mapping, self._sorted_keys
)
results.append((fname, obf_content))
# ── Сборка результата ──
csv_str = builder.build_mapping_csv(self._mapping)
zip_data = builder.build_zip(results, csv_str)
return zip_data, csv_str
finally:
# Очистка состояния (обфускатор может использоваться повторно)
self._mapping.clear()
self._sorted_keys.clear()
# ═══════════════════════════════════════════════════════════════════════
# Удобная функция для быстрого вызова
# ═══════════════════════════════════════════════════════════════════════
def obfuscate_files(
files: List[Tuple[str, bytes, str]], llm_client=None
) -> Tuple[bytes, str]:
"""Обфусцировать список файлов — удобная функция.
Создаёт экземпляр TwoPassObfuscator и вызывает .obfuscate().
Args:
files: [(filename, content_bytes, content_type), ...]
llm_client: Опциональный LLM-клиент
Returns:
(zip_bytes, csv_string)
"""
obf = TwoPassObfuscator(llm_client=llm_client)
return obf.obfuscate(files)
+33
View File
@@ -0,0 +1,33 @@
"""
Утилиты для генерации случайных строк.
Используются генераторами фиктивных значений для создания
случайных цифр и букв в номерах/реквизитах.
"""
import random
import string
def random_digits(n: int) -> str:
"""Сгенерировать строку из n случайных цифр.
Args:
n: Количество цифр
Returns:
Строка из n случайных цифр (0-9)
"""
return ''.join(random.choice(string.digits) for _ in range(n))
def random_letters(n: int) -> str:
"""Сгенерировать строку из n случайных строчных латинских букв.
Args:
n: Количество букв
Returns:
Строка из n случайных букв (a-z)
"""
return ''.join(random.choice(string.ascii_lowercase) for _ in range(n))
+120
View File
@@ -0,0 +1,120 @@
"""
Проход 2: замена сущностей в документах.
Три метода:
1. _apply_replacements — замена в plain-тексте (ядро)
2. _replace_in_docx — замена в DOCX с сохранением форматирования
3. _replace_in_text — замена в plain-тексте → bytes
"""
import io
import re
from typing import Dict, List
def apply_replacements(text: str, mapping: Dict[str, str], sorted_keys: List[str]) -> str:
"""Применить все замены из словаря mapping к строке.
Ключи применяются в порядке убывания длины (sorted_keys).
Это гарантирует, что более длинные совпадения заменяются раньше
коротких (например, «ИНН 123456789012» до «ИНН 1234567890»).
Для сущностей, начинающихся и заканчивающихся на букву/цифру,
используются границы слова (\\b), чтобы избежать частичных замен.
Args:
text: Исходный текст
mapping: Словарь {оригинал: замена}
sorted_keys: Ключи mapping, отсортированные по длине (убывание)
Returns:
Текст с заменами
"""
result = text
for original in sorted_keys:
replacement = mapping[original]
# Если сущность обрамлена буквами/цифрами — используем границы слова
if original and original[0].isalnum() and original[-1].isalnum():
pattern = r'(?<!\w)' + re.escape(original) + r'(?!\w)'
else:
pattern = re.escape(original)
result = re.sub(pattern, replacement, result)
return result
def replace_in_docx(doc, mapping: Dict[str, str], sorted_keys: List[str]) -> bytes:
"""Заменить сущности в DOCX-документе с сохранением форматирования.
Алгоритм для каждого параграфа:
1. Склеиваем текст всех runs в одну строку
2. Применяем замены
3. Пишем результат в первый run, очищаем остальные
Это нужно потому что python-docx разбивает текст на runs
(например, mid-docx форматирование), и сущность может быть
разорвана между несколькими runs.
Args:
doc: Объект python-docx Document
mapping: Словарь замен
sorted_keys: Ключи mapping по убыванию длины
Returns:
Бинарное содержимое изменённого DOCX-файла
"""
# ── Обработка параграфов ──
for para in doc.paragraphs:
if not para.runs:
continue
# Склеиваем все runs в одну строку
full_text = "".join(run.text for run in para.runs)
replaced = apply_replacements(full_text, mapping, sorted_keys)
# Если были замены — пишем в первый run, очищаем остальные
if replaced != full_text:
para.runs[0].text = replaced
for run in para.runs[1:]:
run.text = ""
# ── Обработка таблиц ──
for table in doc.tables:
for row in table.rows:
for cell in row.cells:
for para in cell.paragraphs:
if not para.runs:
continue
full_text = "".join(run.text for run in para.runs)
replaced = apply_replacements(full_text, mapping, sorted_keys)
if replaced != full_text:
para.runs[0].text = replaced
for run in para.runs[1:]:
run.text = ""
# Сохраняем в буфер
buf = io.BytesIO()
doc.save(buf)
return buf.getvalue()
def replace_in_text(text: str, fname: str, mapping: Dict[str, str], sorted_keys: List[str]) -> bytes:
"""Заменить сущности в plain-тексте.
Для PDF и прочих нетекстовых форматов — отдаём текст.
(MVP: без сохранения форматирования PDF).
Args:
text: Исходный текст
fname: Имя файла (для будущего использования — разные форматы)
mapping: Словарь замен
sorted_keys: Ключи mapping по убыванию длины
Returns:
Бинарное содержимое с заменами (UTF-8)
"""
replaced = apply_replacements(text, mapping, sorted_keys)
return replaced.encode('utf-8')
+146
View File
@@ -0,0 +1,146 @@
"""
Проход 1: обнаружение сущностей в тексте документов.
Два метода сканирования:
1. scan_regex — быстрое regex-обнаружение (телефоны, email, ИНН, счета, компании, ФИО)
2. scan_llm_ner — универсальное LLM-обнаружение (любые приватные данные)
LLM САМА определяет какие типы сущностей присутствуют в документе.
Никакого фиксированного списка типов.
"""
import re
import json
import logging
from typing import Dict
from .config import ENTITY_PATTERNS, COMPANY_PATTERN, PERSON_PATTERN
from .generators import ENTITY_GENERATORS, FALLBACK_GENERATOR
from .generators.company import generate_company
from .generators.person import generate_person
log = logging.getLogger("drhider")
def _get_generator(entity_type: str):
"""Получить генератор по типу сущности. Если тип неизвестен — fallback.
Args:
entity_type: Строковый идентификатор типа (например, "phone", "person_name")
Returns:
Функция-генератор: (str) -> str
"""
return ENTITY_GENERATORS.get(entity_type, FALLBACK_GENERATOR)
def scan_regex(text: str, mapping: Dict[str, str]) -> None:
"""Сканировать текст regex-паттернами, заполнить словарь замен.
Ищет в тексте:
- Сущности из ENTITY_PATTERNS (телефоны, email, ИНН, ОГРН, КПП, БИК, счета, паспорта)
- Названия компаний (кроме «НУБЕС» — это Исполнитель, не заменяем)
- ФИО в формате «Фамилия И.О.»
Найденные значения добавляются в mapping: оригинал → фиктивное значение.
Если сущность уже есть в mapping — не перезаписываем (согласованность).
Args:
text: Текст документа для сканирования
mapping: Словарь замен (мутабельный, пополняется)
"""
# ── Сущности по regex-паттернам (телефоны, email, реквизиты) ──
for entity_type, pattern in ENTITY_PATTERNS.items():
for match in re.finditer(pattern, text, re.IGNORECASE | re.MULTILINE):
original = match.group(0).strip()
if not original or original in mapping:
continue
# Генератор: специализированный (если есть) или fallback
generator = _get_generator(entity_type)
mapping[original] = generator(original)
# ── Названия компаний ──
for match in COMPANY_PATTERN.finditer(text):
original = match.group(0).strip()
if not original or original in mapping:
continue
# «НУБЕС» — Исполнитель, НЕ заменяем
if re.search(r'НУБЕС|NUBES', original, re.IGNORECASE):
continue
mapping[original] = generate_company(original)
# ── ФИО (Фамилия И.О.) ──
for match in PERSON_PATTERN.finditer(text):
original = match.group(0).strip()
if not original or original in mapping:
continue
mapping[original] = generate_person(original)
def scan_llm_ner(all_texts: Dict[str, str], mapping: Dict[str, str], llm_client) -> None:
"""LLM NER для обнаружения имён, адресов, паспортных данных.
Отправляет объединённый текст всех файлов в LLM, получает JSON-список
найденных сущностей, добавляет их в словарь замен.
Отправляются первые 3000 символов каждого файла (экономия токенов).
Общий размер промпта ограничен 8000 символами.
Args:
all_texts: {filename: text_content} — тексты всех файлов
mapping: Словарь замен (мутабельный, пополняется)
llm_client: Объект с методом .complete(prompt) -> str
"""
# Формируем комбинированный текст: имя файла + первые 3000 символов
combined = "\n\n---FILE---\n\n".join(
f"FILE: {fname}\n{t[:3000]}" for fname, t in all_texts.items()
)
# ── Универсальный промпт — без фиксированного списка типов ──
prompt = (
"You are a PII (Personally Identifiable Information) detector. "
"Find ALL sensitive or private information in the text below.\n\n"
"Return a JSON array. Each item must have:\n"
' - "type": short snake_case label describing the entity\n'
' (e.g. person_name, phone, email, address, inn, company, passport,\n'
' contract_number, employee_id, bank_account — WHATEVER you see)\n'
' - "value": the EXACT string from the text (copy verbatim)\n\n'
"Rules:\n"
"- Copy value VERBATIM — same spaces, punctuation, case as in text\n"
"- If same value appears multiple times — include only once\n"
"- Invent the type label yourself based on what you see\n"
"- Return ONLY the JSON array, no other text, no markdown wrapping\n\n"
f"Text:\n{combined[:8000]}"
)
try:
# Отправляем запрос в LLM
raw = llm_client.complete(prompt)
# Чистим markdown-обёртку (если LLM вернула ```json ... ```)
raw = raw.strip()
if raw.startswith("```"):
raw = raw.split("\n", 1)[1]
if raw.endswith("```"):
raw = raw[:-3]
entities = json.loads(raw)
# Добавляем найденные сущности в mapping
for ent in entities:
val = ent.get("value", "").strip()
if not val or val in mapping:
continue
# Тип — любой (LLM сама придумала)
ent_type = ent.get("type", "").strip().lower().replace(" ", "_")
# Генератор: специализированный (если тип совпал) или fallback
generator = _get_generator(ent_type)
mapping[val] = generator(val)
except Exception as e:
log.warning("LLM NER failed: %s", e)
-151
View File
@@ -1,151 +0,0 @@
#!/usr/bin/env python3
"""DrHider standalone server — NO DB dependency. Port 8767."""
from http.server import HTTPServer, BaseHTTPRequestHandler
from socketserver import ThreadingMixIn, TCPServer
from urllib.parse import urlparse
import json, re, os, sys, cgi
DRHIDER_SERVER_VERSION = "1.3"
MAX_BODY = 200 * 1024 * 1024 # 200 MB
# ── Загрузка .env ────────────────────────────────────────────────────────
_env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".env")
if os.path.exists(_env_path):
with open(_env_path) as _f:
for _line in _f:
_line = _line.strip()
if _line and not _line.startswith("#") and "=" in _line:
_k, _v = _line.split("=", 1)
if _k not in os.environ:
os.environ[_k] = _v
class ThreadingHTTPServer(ThreadingMixIn, HTTPServer):
daemon_threads = True
class Handler(BaseHTTPRequestHandler):
def do_OPTIONS(self):
self.send_response(200)
self._send_cors()
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.end_headers()
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/health":
self._json({"ok": True, "server": "drhider", "version": DRHIDER_SERVER_VERSION})
else:
self.send_error(404)
def do_POST(self):
if self.path == "/api/drhider":
self._handle_drhider()
else:
self.send_error(404)
# ── DrHider ──────────────────────────────────────────────────────────
def _handle_drhider(self):
"""Обфускация: multipart files → ZIP."""
from drhider import obfuscate_files
length = int(self.headers.get("Content-Length", 0))
if length <= 0 or length > MAX_BODY:
self._json({"ok": False, "error": "Файл слишком большой"}, 413)
return
raw = self.rfile.read(length)
ctype = self.headers.get("Content-Type", "")
_, params = cgi.parse_header(ctype)
boundary = params.get("boundary", "")
if not boundary:
self._json({"ok": False, "error": "No boundary"}, 400)
return
parts = raw.split(b"--" + boundary.encode())
files = []
for part in parts:
if b"Content-Disposition" not in part:
continue
try:
hdr_end = part.index(b"\r\n\r\n")
except ValueError:
continue
hdrs = part[:hdr_end].decode("latin-1", errors="replace")
body = part[hdr_end + 4:]
if body.endswith(b"\r\n"):
body = body[:-2]
m = re.search(r'filename="([^"]*)"', hdrs)
if not m:
continue
name = os.path.basename(m.group(1))
try:
name = name.encode('latin-1').decode('utf-8')
except (UnicodeDecodeError, UnicodeEncodeError):
pass
if not name or ".." in name or "/" in name or "\\" in name:
continue
files.append((name, body, ""))
if not files:
self._json({"ok": False, "error": "Нет файлов"}, 400)
return
class _VMLLM:
def complete(self, prompt):
import httpx
key = os.environ.get("LLM_KEY") or os.environ.get("LLM_API_KEY", "")
r = httpx.post(
os.environ.get("LLM_URL", "https://api.aillm.ru/v1/chat/completions"),
json={"model": os.environ.get("LLM_MODEL", "gpt-oss-120b"),
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 8000, "temperature": 0.1},
headers={"Authorization": f"Bearer {key}",
"Content-Type": "application/json"},
timeout=120
)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
try:
zip_data, _csv = obfuscate_files(files, llm_client=_VMLLM())
self.send_response(200)
self.send_header("Content-Type", "application/zip")
self.send_header("Content-Disposition", 'attachment; filename="drhider_output.zip"')
self.send_header("Content-Length", str(len(zip_data)))
self.send_header("Access-Control-Allow-Origin", "*")
self.end_headers()
self.wfile.write(zip_data)
except Exception as e:
import traceback
traceback.print_exc()
self._json({"ok": False, "error": str(e)}, 500)
# ── Helpers ───────────────────────────────────────────────────────────
def _json(self, data, status=200):
self.send_response(status)
self._send_cors()
self.send_header("Content-Type", "application/json; charset=utf-8")
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode())
def _send_cors(self):
self.send_header("Access-Control-Allow-Origin", "*")
def log_message(self, format, *args):
pass
if __name__ == "__main__":
TCPServer.allow_reuse_address = True
port = int(os.environ.get("PORT", "8767"))
server = ThreadingHTTPServer(("0.0.0.0", port), Handler)
print(f"DrHider server on :{port} (NO DB)")
try:
server.serve_forever()
except KeyboardInterrupt:
server.shutdown()
-2
View File
@@ -1,6 +1,4 @@
flask flask
gunicorn
gevent
python-docx python-docx
pdfplumber pdfplumber
httpx httpx
+37 -83
View File
@@ -1,96 +1,50 @@
"""
DrHider — Managed Flask приложение на платформе Штурвал.
Приложение создаётся фабрикой create_app().
Штурвал запускает Flask самостоятельно (без Dockerfile/gunicorn).
Роуты разнесены по blueprint'ам:
- main_bp: GET / — веб-интерфейс
- health_bp: GET /health — проверка живости
- api_bp: POST /api/drhider — обфускация документов
"""
import os import os
import io
import sys import sys
import json from flask import Flask
import zipfile
from flask import Flask, render_template, request, send_file, jsonify
# Добавляем корень в sys.path чтобы import drhider работал # Добавляем корень проекта в sys.path для импорта пакета drhider
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) _sys_path_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _sys_path_root not in sys.path:
sys.path.insert(0, _sys_path_root)
from drhider import obfuscate_files # Версия приложения (меняется при изменениях)
VERSION = "2.0.0"
def create_app():
"""Создать и настроить Flask-приложение.
Returns:
Экземпляр Flask с зарегистрированными blueprint'ами.
"""
app = Flask(__name__) app = Flask(__name__)
VERSION = "1.0.0" # Конфигурация
app.config["VERSION"] = VERSION
app.config["MAX_CONTENT_LENGTH"] = 200 * 1024 * 1024 # 200 MB
MAX_BODY = 200 * 1024 * 1024 # 200 MB # Регистрируем blueprint'ы
from routes import register_routes
register_routes(app)
return app
class LLMClient: # Экземпляр приложения — точка входа для Штурвала
"""LLM-клиент для drhider (обнаружение компаний/ФИО).""" app = create_app()
def __init__(self):
import httpx
self._httpx = httpx
def complete(self, prompt: str) -> str:
key = os.environ.get("LLM_KEY") or os.environ.get("LLM_API_KEY", "")
r = self._httpx.post(
os.environ.get("LLM_URL", "https://api.aillm.ru/v1/chat/completions"),
json={
"model": os.environ.get("LLM_MODEL", "gpt-oss-120b"),
"messages": [{"role": "user", "content": prompt}],
"max_tokens": 8000,
"temperature": 0.1,
},
headers={
"Authorization": f"Bearer {key}",
"Content-Type": "application/json",
},
timeout=120,
)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
@app.route("/")
def index():
return render_template("index.html", version=VERSION)
@app.route("/health")
def health():
return jsonify({"ok": True, "version": VERSION})
@app.route("/api/drhider", methods=["POST"])
def api_drhider():
"""Обфускация: multipart/form-data с файлами → ZIP."""
uploaded = request.files.getlist("files")
if not uploaded:
return jsonify({"ok": False, "error": "Нет файлов"}), 400
files = []
for f in uploaded:
if f.filename:
files.append((f.filename, f.read(), f.mimetype or ""))
if not files:
return jsonify({"ok": False, "error": "Нет файлов"}), 400
try:
llm = LLMClient()
zip_data, _csv = obfuscate_files(files, llm_client=llm)
return send_file(
io.BytesIO(zip_data),
mimetype="application/zip",
as_attachment=True,
download_name="drhider_output.zip",
)
except Exception as e:
import traceback
traceback.print_exc()
return jsonify({"ok": False, "error": str(e)}), 500
if __name__ == "__main__":
import subprocess
subprocess.run([
sys.executable, "-m", "gunicorn", "app:app",
"--bind", "0.0.0.0:5000",
"--worker-class", "gevent",
"--workers", "1",
"--worker-connections", "1000",
"--timeout", "300",
])
+20
View File
@@ -0,0 +1,20 @@
"""
Регистрация всех blueprint'ов приложения.
Импортируется из site/app.py при создании Flask-приложения.
"""
from .main_bp import main_bp
from .health_bp import health_bp
from .api_bp import api_bp
def register_routes(app):
"""Зарегистрировать все blueprint'ы на Flask-приложении.
Args:
app: Экземпляр Flask-приложения
"""
app.register_blueprint(main_bp)
app.register_blueprint(health_bp)
app.register_blueprint(api_bp)
+73
View File
@@ -0,0 +1,73 @@
"""
Blueprint: API обфускации (POST /api/drhider).
Принимает multipart/form-data с файлами, возвращает ZIP-архив
с обфусцированными документами.
"""
import io
import traceback
from flask import Blueprint, request, send_file, jsonify
from drhider import obfuscate_files, LLMClient
# ═══════════════════════════════════════════════════════════════════════════
# Blueprint: API DrHider
# ═══════════════════════════════════════════════════════════════════════════
api_bp = Blueprint("api", __name__, url_prefix="/api")
# Максимальный размер тела запроса: 200 MB
MAX_BODY = 200 * 1024 * 1024
@api_bp.route("/drhider", methods=["POST"])
def drhider():
"""Обфускация документов.
Принимает multipart/form-data с полем 'files' (один или несколько файлов).
Возвращает ZIP-архив с обфусцированными документами + mapping.csv.
Поддерживаемые форматы:
.docx, .pdf, .txt, .doc (бинарный — без изменений)
.zip (распаковывается, содержимое обфусцируется)
Returns:
200: ZIP-архив (application/zip)
400: {"ok": false, "error": "..."}
500: {"ok": false, "error": "..."}
"""
# ── Получаем файлы из запроса ──
uploaded = request.files.getlist("files")
if not uploaded:
return jsonify({"ok": False, "error": "Нет файлов"}), 400
# Формируем список для obfuscate_files: [(name, bytes, mimetype), ...]
files = []
for f in uploaded:
if f.filename:
files.append((f.filename, f.read(), f.mimetype or ""))
if not files:
return jsonify({"ok": False, "error": "Нет файлов"}), 400
# ── Обфускация ──
try:
# Создаём LLM-клиент для NER-сканирования
llm = LLMClient()
# Вызываем ядро обфускации
zip_data, _csv = obfuscate_files(files, llm_client=llm)
# Отдаём ZIP-архив
return send_file(
io.BytesIO(zip_data),
mimetype="application/zip",
as_attachment=True,
download_name="drhider_output.zip",
)
except Exception as e:
# Логируем полный трейсбек на сервере
traceback.print_exc()
return jsonify({"ok": False, "error": str(e)}), 500
+27
View File
@@ -0,0 +1,27 @@
"""
Blueprint: health-check (GET /health).
Используется платформой Штурвал для проверки живости приложения.
"""
from flask import Blueprint, jsonify, current_app
# ═══════════════════════════════════════════════════════════════════════════
# Blueprint: health check
# ═══════════════════════════════════════════════════════════════════════════
health_bp = Blueprint("health", __name__)
@health_bp.route("/health")
def health():
"""Эндпоинт проверки живости.
Возвращает JSON с версией приложения.
Используется платформой для readiness/liveness probes.
Returns:
{"ok": true, "version": "X.Y.Z"}
"""
version = current_app.config.get("VERSION", "0.0.0")
return jsonify({"ok": True, "version": version})
+23
View File
@@ -0,0 +1,23 @@
"""
Blueprint: главная страница (GET /).
Отдаёт HTML-интерфейс DrHider.
"""
from flask import Blueprint, render_template, current_app
# ═══════════════════════════════════════════════════════════════════════════
# Blueprint: главная страница
# ═══════════════════════════════════════════════════════════════════════════
main_bp = Blueprint("main", __name__)
@main_bp.route("/")
def index():
"""Главная страница — веб-интерфейс DrHider.
Передаёт в шаблон текущую версию приложения.
"""
version = current_app.config.get("VERSION", "0.0.0")
return render_template("index.html", version=version)