fix: DrHider ZIP expansion — extract archives before obfuscation
Deploy contracts-flask / validate (push) Successful in 0s
Deploy contracts-flask / validate (push) Successful in 0s
This commit is contained in:
@@ -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]]:
|
||||
|
||||
@@ -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]]:
|
||||
|
||||
Reference in New Issue
Block a user