diff --git a/deploy/services/drhider.py b/deploy/services/drhider.py index 846807c..ab62306 100644 --- a/deploy/services/drhider.py +++ b/deploy/services/drhider.py @@ -190,6 +190,9 @@ class TwoPassObfuscator: """ self._workdir = tempfile.mkdtemp(prefix="drhider_") try: + # --- Распаковать ZIP-файлы --- + files = self._expand_zips(files) + # --- Проход 1: сбор сущностей --- all_texts: Dict[str, str] = {} # filename → text for LLM NER all_docx: Dict[str, object] = {} # filename → docx Document for replacement @@ -231,6 +234,43 @@ class TwoPassObfuscator: self._mapping.clear() self._regex_replacements.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 для имён + try: + inner_name = info.filename.encode('cp437').decode('utf-8') + except (UnicodeDecodeError, UnicodeEncodeError): + inner_name = info.filename + # Убрать path traversal + inner_name = os.path.basename(inner_name) + if not inner_name: + continue + inner_data = zf.read(info) + result.append((inner_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]]: diff --git a/site/services/drhider.py b/site/services/drhider.py index 846807c..ab62306 100644 --- a/site/services/drhider.py +++ b/site/services/drhider.py @@ -190,6 +190,9 @@ class TwoPassObfuscator: """ self._workdir = tempfile.mkdtemp(prefix="drhider_") try: + # --- Распаковать ZIP-файлы --- + files = self._expand_zips(files) + # --- Проход 1: сбор сущностей --- all_texts: Dict[str, str] = {} # filename → text for LLM NER all_docx: Dict[str, object] = {} # filename → docx Document for replacement @@ -231,6 +234,43 @@ class TwoPassObfuscator: self._mapping.clear() self._regex_replacements.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 для имён + try: + inner_name = info.filename.encode('cp437').decode('utf-8') + except (UnicodeDecodeError, UnicodeEncodeError): + inner_name = info.filename + # Убрать path traversal + inner_name = os.path.basename(inner_name) + if not inner_name: + continue + inner_data = zf.read(info) + result.append((inner_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]]: