commit a455f9da7b7b92fd3593d3f34b3f6cca0e7875a0 Author: “Naeel” Date: Sat Jul 11 16:06:19 2026 +0400 Начальная структура managed Flask + drhider diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..b6cf5f0 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +__pycache__/ +*.pyc +.env diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..a18e98a --- /dev/null +++ b/Dockerfile @@ -0,0 +1,8 @@ +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"] diff --git a/drhider.py b/drhider.py new file mode 100644 index 0000000..5fcaec7 --- /dev/null +++ b/drhider.py @@ -0,0 +1,557 @@ +""" +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'(? 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'(? 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) diff --git a/drhider_server.py b/drhider_server.py new file mode 100644 index 0000000..ea5819c --- /dev/null +++ b/drhider_server.py @@ -0,0 +1,151 @@ +#!/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() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..574b557 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,6 @@ +flask +gunicorn +gevent +python-docx +pdfplumber +httpx