""" 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) 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]) elif fname.endswith('.pdf'): txt = all_texts.get(fname, '') obf_content = self._replace_in_text(txt, fname) fname = fname[:-4] + '.txt' 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 _expand_zips(self, files: List[Tuple[str, bytes, str]]) -> List[Tuple[str, bytes, str]]: """Распаковать ZIP-файлы, заменив их содержимым. Остальные файлы — как есть.""" result = [] for fname, content, ctype in files: if fname.lower().endswith('.zip'): try: with zipfile.ZipFile(io.BytesIO(content)) as zf: total_size = sum(info.file_size for info in zf.infolist()) if total_size > 500 * 1024 * 1024: # 500 MB log.warning("ZIP too large, skipping expansion: %s", fname) result.append((fname, content, ctype)) continue if len(zf.infolist()) > 500: log.warning("ZIP too many files, skipping: %s", fname) result.append((fname, content, ctype)) continue for info in zf.infolist(): if info.is_dir(): continue # cp437 → utf8 (как в services/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) 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, metadata_encoding='utf-8') as zf: for fname, content in files: zf.writestr(fname, 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)