from PYAS_Diagnostics import log_exception import os import math import zlib import numpy import pefile import datetime def _entropy_from_counts(counts, size): if size <= 0: return 0.0 nonzero = counts[counts > 0] return float(numpy.log2(size) - numpy.sum(nonzero * numpy.log2(nonzero)) / size) def _bounded_entropy(data, batch_size=1048576): size = len(data) if size == 0: return 0.0 counts = numpy.zeros(256, dtype=numpy.int64) view = memoryview(data) try: for start in range(0, size, batch_size): chunk = numpy.frombuffer(view[start : min(start + batch_size, size)], dtype=numpy.uint8) counts += numpy.bincount(chunk, minlength=256) finally: view.release() return _entropy_from_counts(counts, size) def _accumulate_string_chunk(chunk, array, state, pattern): separators = (array < 0x20) | (array > 0x7E) if not separators.any(): state["pending_counts"] += numpy.bincount(array, minlength=256) state["pending_length"] += len(array) return first_separator = int(numpy.argmax(separators)) last_separator = len(separators) - 1 - int(numpy.argmax(separators[::-1])) if first_separator: state["pending_counts"] += numpy.bincount(array[:first_separator], minlength=256) state["pending_length"] += first_separator if state["pending_length"] >= 5: state["count"] += 1 state["total_length"] += state["pending_length"] state["counts"] += state["pending_counts"] state["pending_length"] = 0 state["pending_counts"].fill(0) middle_start = first_separator + 1 middle_end = last_separator if middle_end > middle_start: matches = pattern.findall(chunk[middle_start:middle_end]) if matches: combined = b"".join(matches) state["count"] += len(matches) state["total_length"] += len(combined) state["counts"] += numpy.bincount( numpy.frombuffer(combined, dtype=numpy.uint8), minlength=256 ) trailing_start = last_separator + 1 if trailing_start < len(array): trailing = array[trailing_start:] state["pending_counts"] += numpy.bincount(trailing, minlength=256) state["pending_length"] = len(trailing) def _accumulate_entropy_windows(data, entropy_histogram, c_log_c_table): block_count = len(data) // 1024 if block_count < 2: return data complete_size = block_count * 1024 blocks = numpy.frombuffer(data, dtype=numpy.uint8, count=complete_size).reshape( block_count, 1024 ) encoded = blocks.astype(numpy.int32) encoded += numpy.arange(block_count, dtype=numpy.int32)[:, None] * 256 block_counts = numpy.bincount(encoded.ravel(), minlength=block_count * 256).reshape( block_count, 256 ) window_counts = block_counts[:-1] + block_counts[1:] sum_c_log_c = numpy.sum(c_log_c_table[window_counts], axis=1) entropies = 11.0 - sum_c_log_c / 2048.0 entropy_bins = (entropies * 2.0).astype(numpy.int32) numpy.clip(entropy_bins, 0, 15, out=entropy_bins) byte_bins = window_counts.reshape(block_count - 1, 16, 16).sum(axis=2) flat_indices = (entropy_bins[:, None] * 16 + numpy.arange(16)).ravel() numpy.add.at(entropy_histogram, flat_indices, byte_bins.ravel()) return data[(block_count - 1) * 1024 :] def _collect_resource_ranges(pe, file_size): ranges = [] if not hasattr(pe, "DIRECTORY_ENTRY_RESOURCE"): return ranges for entry_type in pe.DIRECTORY_ENTRY_RESOURCE.entries: for entry_id in getattr(getattr(entry_type, "directory", None), "entries", []): for entry_lang in getattr(getattr(entry_id, "directory", None), "entries", []): if not hasattr(entry_lang, "data"): continue try: struct = entry_lang.data.struct offset = int(pe.get_offset_from_rva(int(struct.OffsetToData))) size = int(struct.Size) start = max(0, offset) end = min(file_size, start + max(0, size)) ranges.append((start, max(start, end))) except Exception: log_exception("PYAS_Features._collect_resource_ranges:109") ranges.append((0, 0)) return ranges def _extract_stream_features(file_path, file_size, ranges, extractor): normalized = [] for start, end in ranges: start = max(0, min(file_size, int(start))) end = max(start, min(file_size, int(end))) normalized.append((start, end)) byte_counts = numpy.zeros(256, dtype=numpy.int64) range_counts = [numpy.zeros(256, dtype=numpy.int64) for _ in normalized] entropy_histogram = numpy.zeros(256, dtype=numpy.float64) string_state = { "count": 0, "total_length": 0, "counts": numpy.zeros(256, dtype=numpy.int64), "pending_length": 0, "pending_counts": numpy.zeros(256, dtype=numpy.int64), } histogram_tail = b"" position = 0 with open(file_path, "rb") as source: while True: chunk = source.read(4194304) if not chunk: break array = numpy.frombuffer(chunk, dtype=numpy.uint8) _accumulate_string_chunk(chunk, array, string_state, extractor._STRING_PATTERN) histogram_tail = _accumulate_entropy_windows( histogram_tail + chunk, entropy_histogram, extractor._C_LOG_C_TABLE, ) chunk_end = position + len(chunk) boundaries = {0, len(chunk)} for start, end in normalized: if position < start < chunk_end: boundaries.add(start - position) if position < end < chunk_end: boundaries.add(end - position) ordered = sorted(boundaries) for index in range(len(ordered) - 1): local_start = ordered[index] local_end = ordered[index + 1] if local_end <= local_start: continue counts = numpy.bincount(array[local_start:local_end], minlength=256) byte_counts += counts absolute_start = position + local_start absolute_end = position + local_end for range_index, (start, end) in enumerate(normalized): if start <= absolute_start and absolute_end <= end: range_counts[range_index] += counts position = chunk_end if string_state["pending_length"] >= 5: string_state["count"] += 1 string_state["total_length"] += string_state["pending_length"] string_state["counts"] += string_state["pending_counts"] string_count = string_state["count"] total_string_length = string_state["total_length"] result = { "StringCount": float(string_count), "StringMeanLength": (float(total_string_length) / string_count if string_count else 0.0), "StringEntropy": _entropy_from_counts(string_state["counts"], total_string_length), } result.update(dict.fromkeys(extractor._STRING_KEYS, 0.0)) if total_string_length: valid_counts = string_state["counts"][0x20:0x7F] result.update( dict( zip( extractor._STRING_KEYS, valid_counts.astype(numpy.float64) / total_string_length, ) ) ) result["FileEntropy"] = _entropy_from_counts(byte_counts, file_size) result.update( dict( zip( extractor._BYTE_HIST_KEYS, byte_counts.astype(numpy.float64) / file_size, ) ) ) entropy_sum = numpy.sum(entropy_histogram) if entropy_sum: entropy_histogram /= entropy_sum result.update(dict(zip(extractor._BYTE_ENT_KEYS, entropy_histogram))) range_entropies = [ _entropy_from_counts(counts, end - start) for counts, (start, end) in zip(range_counts, normalized) ] return result, range_entropies class PEFeatureMixin: def _safe_float(self, val): try: f = float(val) if math.isinf(f) or math.isnan(f): return 0.0 return f except Exception: log_exception("PYAS_Features.PEFeatureMixin._safe_float:221") return 0.0 def _base_defaults(self): base = dict.fromkeys(self._OPTIONAL_HEADER_FIELDS, 0.0) base.update(self._SECTION_DEFAULTS) base.update(self._CONDITIONAL_DEFAULTS) for i in range(16): base[f"DataDirectory_{i}_Size"] = 0.0 base[f"DataDirectory_{i}_VA"] = 0.0 return base def _calc_entropy(self, data): return _bounded_entropy(data) def _extract_strings(self, file_bytes): res = {"StringCount": 0.0, "StringMeanLength": 0.0, "StringEntropy": 0.0} res.update(dict.fromkeys(self._STRING_KEYS, 0.0)) if len(file_bytes) <= self._STRING_FAST_PATH_BYTES: matches = self._STRING_PATTERN.findall(file_bytes) if not matches: return res combined_data = b"".join(matches) count = len(matches) total_len = len(combined_data) global_counts = numpy.bincount( numpy.frombuffer(combined_data, dtype=numpy.uint8), minlength=256 ) else: count = 0 total_len = 0 global_counts = numpy.zeros(256, dtype=numpy.int64) mv = memoryview(file_bytes) pos = 0 size = len(file_bytes) while pos < size: chunk_end = min(pos + self._STRING_BATCH_BYTES, size) chunk_arr = numpy.frombuffer(mv[pos:chunk_end], dtype=numpy.uint8) separators = (chunk_arr < 0x20) | (chunk_arr > 0x7E) if separators.any(): last_separator = len(separators) - 1 - int(numpy.argmax(separators[::-1])) segment_end = pos + last_separator + 1 matches = self._STRING_PATTERN.findall(mv[pos:segment_end]) if matches: combined_data = b"".join(matches) count += len(matches) total_len += len(combined_data) global_counts += numpy.bincount( numpy.frombuffer(combined_data, dtype=numpy.uint8), minlength=256 ) pos = segment_end continue pending_len = 0 pending_counts = numpy.zeros(256, dtype=numpy.int64) while pos < size: chunk_end = min(pos + self._STRING_BATCH_BYTES, size) chunk_arr = numpy.frombuffer(mv[pos:chunk_end], dtype=numpy.uint8) separators = (chunk_arr < 0x20) | (chunk_arr > 0x7E) if separators.any(): first_separator = int(numpy.argmax(separators)) if first_separator: pending_counts += numpy.bincount( chunk_arr[:first_separator], minlength=256 ) pending_len += first_separator pos += first_separator + 1 break pending_counts += numpy.bincount(chunk_arr, minlength=256) pending_len += len(chunk_arr) pos = chunk_end if pending_len >= 5: count += 1 total_len += pending_len global_counts += pending_counts if count == 0 or total_len == 0: return res res["StringCount"] = float(count) res["StringMeanLength"] = float(total_len) / count counts_nonzero = global_counts[global_counts > 0] if len(counts_nonzero) > 0: res["StringEntropy"] = float( numpy.log2(total_len) - numpy.sum(counts_nonzero * numpy.log2(counts_nonzero)) / total_len ) valid_chars = global_counts[0x20:0x7F] res.update(dict(zip(self._STRING_KEYS, valid_chars.astype(numpy.float64) / total_len))) return res def _extract_histograms(self, file_bytes): arr = numpy.frombuffer(file_bytes, dtype=numpy.uint8) sz = len(arr) res = {} if sz == 0: res["FileEntropy"] = 0.0 res.update(dict.fromkeys(self._BYTE_HIST_KEYS, 0.0)) res.update(dict.fromkeys(self._BYTE_ENT_KEYS, 0.0)) return res byte_counts = numpy.zeros(256, dtype=numpy.int64) ent_hist = numpy.zeros(256, dtype=numpy.float64) window = 2048 step = 1024 if sz < window: byte_counts += numpy.bincount(arr, minlength=256) p = byte_counts[byte_counts > 0] / float(sz) entropy = float(-numpy.sum(p * numpy.log2(p))) ent_bin = min(int(entropy * 2.0), 15) byte_bins = byte_counts.reshape(16, 16).sum(axis=1) idx_start = ent_bin * 16 ent_hist[idx_start : idx_start + 16] += byte_bins else: num_windows = (sz - window) // step + 1 for w_start in range(0, num_windows, self._HISTOGRAM_BATCH_WINDOWS): w_end = min(w_start + self._HISTOGRAM_BATCH_WINDOWS, num_windows) b_num = w_end - w_start start_byte = w_start * step end_byte = start_byte + b_num * step + step encoded = arr[start_byte:end_byte].astype(numpy.int32).reshape(b_num + 1, step) encoded += numpy.arange(b_num + 1, dtype=numpy.int32)[:, None] * 256 chunk_counts = numpy.bincount(encoded.ravel(), minlength=(b_num + 1) * 256).reshape( b_num + 1, 256 ) del encoded byte_counts += numpy.sum(chunk_counts[:-1], axis=0) if w_end == num_windows: byte_counts += chunk_counts[-1] window_counts = chunk_counts[:-1] + chunk_counts[1:] sum_c_log_c = numpy.sum(self._C_LOG_C_TABLE[window_counts], axis=1) entropies = 11.0 - sum_c_log_c / 2048.0 ent_bins = (entropies * 2.0).astype(numpy.int32) numpy.clip(ent_bins, 0, 15, out=ent_bins) byte_bins = window_counts.reshape(b_num, 16, 16).sum(axis=2) flat_ent_bins = (ent_bins[:, None] * 16 + numpy.arange(16)).ravel() numpy.add.at(ent_hist, flat_ent_bins, byte_bins.ravel()) covered_size = (num_windows + 1) * step if covered_size < sz: byte_counts += numpy.bincount(arr[covered_size:], minlength=256) counts_nonzero = byte_counts[byte_counts > 0] res["FileEntropy"] = float( numpy.log2(sz) - numpy.sum(counts_nonzero * numpy.log2(counts_nonzero)) / sz ) byte_hist = byte_counts.astype(numpy.float64) / sz res.update(dict(zip(self._BYTE_HIST_KEYS, byte_hist))) ent_sum = numpy.sum(ent_hist) if ent_sum > 0: ent_hist /= ent_sum res.update(dict(zip(self._BYTE_ENT_KEYS, ent_hist))) return res def _extract_overlay_features(self, pe, fsize): overlay_offset = pe.get_overlay_data_start_offset() if not overlay_offset or overlay_offset >= fsize: return { "HasOverlay": 0.0, "OverlaySize": 0.0, "OverlayRatio": 0.0, "OverlayEntropy": 0.0, } overlay_data = pe.get_overlay() if not overlay_data: return { "HasOverlay": 0.0, "OverlaySize": 0.0, "OverlayRatio": 0.0, "OverlayEntropy": 0.0, } sz = len(overlay_data) return { "HasOverlay": 1.0, "OverlaySize": float(sz), "OverlayRatio": float(sz) / float(fsize), "OverlayEntropy": self._calc_entropy(overlay_data), } def _extract_rich_header(self, pe): if not hasattr(pe, "RICH_HEADER") or not pe.RICH_HEADER: return {"HasRichHeader": 0.0, "RichHeaderCount": 0.0} return { "HasRichHeader": 1.0, "RichHeaderCount": ( float(len(pe.RICH_HEADER.values) // 2) if hasattr(pe.RICH_HEADER, "values") else 0.0 ), } def _extract_ep_anomalies(self, pe): if not hasattr(pe, "OPTIONAL_HEADER") or not hasattr(pe, "sections") or not pe.sections: return { "EntryPointSectionIndex": -1.0, "EntryPointInExecutable": 0.0, "EntryPointInLastSection": 0.0, } ep = pe.OPTIONAL_HEADER.AddressOfEntryPoint ep_section_idx = -1 ep_in_exec = 0.0 for idx, sec in enumerate(pe.sections): if sec.VirtualAddress <= ep < (sec.VirtualAddress + sec.Misc_VirtualSize): ep_section_idx = idx if sec.Characteristics & 0x20000000: ep_in_exec = 1.0 break return { "EntryPointSectionIndex": float(ep_section_idx), "EntryPointInExecutable": ep_in_exec, "EntryPointInLastSection": 1.0 if ep_section_idx == len(pe.sections) - 1 else 0.0, } def _extract_advanced_resources(self, pe, range_entropies=None): res_data = { "ResourceMaxEntropy": 0.0, "ResourceMinEntropy": 0.0, "ResourceMeanEntropy": 0.0, "ResourceLangCount": 0.0, "ResourceRCDataCount": 0.0, } if not hasattr(pe, "DIRECTORY_ENTRY_RESOURCE"): return res_data entropies = [] entropy_iter = iter(range_entropies or ()) langs = set() rcdata_count = 0 for entry_type in pe.DIRECTORY_ENTRY_RESOURCE.entries: if hasattr(entry_type, "id") and entry_type.id == 10: rcdata_count += len(getattr(entry_type.directory, "entries", [])) if not hasattr(entry_type, "directory"): continue for entry_id in entry_type.directory.entries: if not hasattr(entry_id, "directory"): continue for entry_lang in entry_id.directory.entries: if hasattr(entry_lang, "id"): langs.add(entry_lang.id) if hasattr(entry_lang, "data"): try: entropies.append(float(next(entropy_iter))) except (StopIteration, TypeError, ValueError): entropies.append(0.0) res_data["ResourceLangCount"] = float(len(langs)) res_data["ResourceRCDataCount"] = float(rcdata_count) if entropies: res_data["ResourceMaxEntropy"] = max(entropies) res_data["ResourceMinEntropy"] = min(entropies) res_data["ResourceMeanEntropy"] = sum(entropies) / len(entropies) return res_data def _extract_load_config(self, pe): cfg = {"HasLoadConfig": 0.0, "HasCFG": 0.0, "HasSEHTable": 0.0} if hasattr(pe, "DIRECTORY_ENTRY_LOAD_CONFIG"): cfg["HasLoadConfig"] = 1.0 struct = pe.DIRECTORY_ENTRY_LOAD_CONFIG.struct if getattr(struct, "GuardCFFunctionTable", 0) != 0: cfg["HasCFG"] = 1.0 if getattr(struct, "SEHandlerTable", 0) != 0: cfg["HasSEHTable"] = 1.0 return cfg def _extract_security_directory(self, pe, file_path, fsize): res = {"HasSignature": 0.0, "SignatureCount": 0.0} if not hasattr(pe, "OPTIONAL_HEADER") or not hasattr(pe.OPTIONAL_HEADER, "DATA_DIRECTORY"): return res directories = pe.OPTIONAL_HEADER.DATA_DIRECTORY if len(directories) <= 4: return res sec_dir = directories[4] offset = int(sec_dir.VirtualAddress) size = int(sec_dir.Size) if offset <= 0 or size <= 0 or offset + size > fsize: return res res["HasSignature"] = 1.0 count = 0 curr = offset end = offset + size with open(file_path, "rb") as source: while curr + 8 <= end: source.seek(curr) header = source.read(4) if len(header) != 4: break length = int.from_bytes(header, "little") if length < 8 or curr + length > end: break count += 1 curr += (length + 7) & ~7 res["SignatureCount"] = float(count) return res def extract_features(self, file_path): pe = None try: fsize = os.path.getsize(file_path) if fsize == 0 or fsize > 4294967296: return None base = self._base_defaults() dlls = set() apis = set() pe = pefile.PE(name=file_path, fast_load=True) try: pe.parse_rich_header() pe.parse_data_directories( directories=[ pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_IMPORT"], pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_EXPORT"], pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_RESOURCE"], pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_DEBUG"], pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_TLS"], pefile.DIRECTORY_ENTRY["IMAGE_DIRECTORY_ENTRY_LOAD_CONFIG"], ] ) except Exception: log_exception("PYAS_Features.PEFeatureMixin.extract_features:553") pass sections = list(getattr(pe, "sections", [])) section_ranges = [] for section in sections: start = max(0, int(section.get_PointerToRawData_adj())) end = min(fsize, start + max(0, int(section.SizeOfRawData))) section_ranges.append((start, max(start, end))) resource_ranges = _collect_resource_ranges(pe, fsize) overlay_offset = pe.get_overlay_data_start_offset() if not overlay_offset or overlay_offset >= fsize: overlay_offset = fsize all_ranges = section_ranges + resource_ranges + [(int(overlay_offset), fsize)] stream_features, range_entropies = _extract_stream_features( file_path, fsize, all_ranges, self ) section_entropies = range_entropies[: len(section_ranges)] resource_start = len(section_ranges) resource_entropies = range_entropies[ resource_start : resource_start + len(resource_ranges) ] overlay_entropy = range_entropies[-1] base["TrustSigned"] = 1.0 if self.signer.sign_verify(file_path) else 0.0 base["FileSize"] = float(fsize) fh = pe.FILE_HEADER base["Machine"] = getattr(fh, "Machine", 0) base["NumberOfSections"] = getattr(fh, "NumberOfSections", 0) base["TimeDateStamp"] = getattr(fh, "TimeDateStamp", 0) base["PointerToSymbolTable"] = getattr(fh, "PointerToSymbolTable", 0) base["NumberOfSymbols"] = getattr(fh, "NumberOfSymbols", 0) base["SizeOfOptionalHeader"] = getattr(fh, "SizeOfOptionalHeader", 0) base["Characteristics"] = getattr(fh, "Characteristics", 0) curr_ts = datetime.datetime.utcnow().timestamp() base["HasInvalidTimestamp"] = ( 1.0 if (base["TimeDateStamp"] < 631152000 or base["TimeDateStamp"] > curr_ts + 2592000) else 0.0 ) base["FileTimeException"] = 1.0 if base["TimeDateStamp"] == 0 else 0.0 base["Is64Bit"] = 1.0 if base["Machine"] in (0x8664, 0xAA64, 0x0200) else 0.0 base["IsExe"] = 1.0 if pe.is_exe() else 0.0 base["IsDll"] = 1.0 if pe.is_dll() else 0.0 base["ExceptionCount"] = 0.0 if hasattr(pe, "OPTIONAL_HEADER"): op = pe.OPTIONAL_HEADER fields = [ "Magic", "MajorLinkerVersion", "MinorLinkerVersion", "SizeOfCode", "SizeOfInitializedData", "SizeOfUninitializedData", "AddressOfEntryPoint", "BaseOfCode", "ImageBase", "SectionAlignment", "FileAlignment", "MajorOperatingSystemVersion", "MinorOperatingSystemVersion", "MajorImageVersion", "MinorImageVersion", "MajorSubsystemVersion", "MinorSubsystemVersion", "SizeOfImage", "SizeOfHeaders", "CheckSum", "Subsystem", "DllCharacteristics", "SizeOfStackReserve", "SizeOfStackCommit", "SizeOfHeapReserve", "SizeOfHeapCommit", "LoaderFlags", "NumberOfRvaAndSizes", ] for f_field in fields: base[f_field] = getattr(op, f_field, 0) base["IsDriver"] = 1.0 if base.get("Subsystem") == 1 else 0.0 if hasattr(op, "DATA_DIRECTORY"): for i, directory in enumerate(op.DATA_DIRECTORY): base[f"DataDirectory_{i}_Size"] = directory.Size base[f"DataDirectory_{i}_VA"] = directory.VirtualAddress exc_idx = pefile.DIRECTORY_ENTRY.get("IMAGE_DIRECTORY_ENTRY_EXCEPTION", 3) if len(op.DATA_DIRECTORY) > exc_idx and op.DATA_DIRECTORY[exc_idx].Size > 0: base["ExceptionCount"] = ( op.DATA_DIRECTORY[exc_idx].Size // 12 if base["Machine"] in (0x8664, 0xAA64) else 0.0 ) res_flags_count = dict.fromkeys(self._CHAR_COUNT_KEYS.values(), 0.0) res_flags_ent = dict.fromkeys(self._CHAR_ENT_KEYS.values(), 0.0) res_sec_hash = dict.fromkeys(self._SECTION_HASH_KEYS, 0.0) if hasattr(pe, "sections"): base["SectionCount"] = len(pe.sections) entropies = [] raw_sizes = [] v_sizes = [] exec_sec = write_sec = read_sec = sec_exc = 0 for section_index, section in enumerate(pe.sections): sec_name = section.Name.rstrip(b"\x00") hash_val = zlib.crc32(sec_name) % 50 res_sec_hash[self._SECTION_HASH_KEYS[hash_val]] += 1.0 s_entropy = ( section_entropies[section_index] if section_index < len(section_entropies) else 0.0 ) entropies.append(s_entropy) raw_sizes.append(section.SizeOfRawData) v_sizes.append(section.Misc_VirtualSize) if section.Characteristics & 0x20000000: exec_sec += 1 if section.Characteristics & 0x80000000: write_sec += 1 if section.Characteristics & 0x40000000: read_sec += 1 if section.SizeOfRawData + section.PointerToRawData > fsize: sec_exc = 1 for flag in self._CHAR_FLAGS: if section.Characteristics & flag: res_flags_count[self._CHAR_COUNT_KEYS[flag]] += 1.0 res_flags_ent[self._CHAR_ENT_KEYS[flag]] += s_entropy base["SectionMaxEntropy"] = max(entropies) if entropies else 0.0 base["SectionMinEntropy"] = min(entropies) if entropies else 0.0 base["SectionMeanEntropy"] = sum(entropies) / len(entropies) if entropies else 0.0 base["SectionMaxRawSize"] = max(raw_sizes) if raw_sizes else 0.0 base["SectionMinRawSize"] = min(raw_sizes) if raw_sizes else 0.0 base["SectionMeanRawSize"] = sum(raw_sizes) / len(raw_sizes) if raw_sizes else 0.0 base["SectionMaxVSize"] = max(v_sizes) if v_sizes else 0.0 base["SectionMinVSize"] = min(v_sizes) if v_sizes else 0.0 base["SectionMeanVSize"] = sum(v_sizes) / len(v_sizes) if v_sizes else 0.0 base["ExecutableSections"] = float(exec_sec) base["WritableSections"] = float(write_sec) base["ReadableSections"] = float(read_sec) base["SectionException"] = float(sec_exc) for flag in self._CHAR_FLAGS: cnt_key = self._CHAR_COUNT_KEYS[flag] if res_flags_count[cnt_key] > 0: res_flags_ent[self._CHAR_ENT_KEYS[flag]] /= res_flags_count[cnt_key] base.update(res_flags_count) base.update(res_flags_ent) base.update(res_sec_hash) string_keys = [ "FileDescription", "FileVersion", "ProductName", "ProductVersion", "CompanyName", "LegalCopyright", "Comments", "InternalName", "LegalTrademarks", "SpecialBuild", "PrivateBuild", ] for key in string_keys: base[f"{key}Length"] = 0.0 if hasattr(pe, "FileInfo"): for fileinfo_list in pe.FileInfo: for fileinfo in fileinfo_list: if getattr(fileinfo, "name", "") in ("StringFileInfo", b"StringFileInfo"): for st in getattr(fileinfo, "StringTable", []): for key, val in st.entries.items(): try: k = ( key.decode("utf-8", "ignore") if isinstance(key, bytes) else str(key) ) if k in string_keys: v = ( val.decode("utf-8", "ignore") if isinstance(val, bytes) else str(val) ) base[f"{k}Length"] = float(len(v)) except Exception: log_exception( "PYAS_Features.PEFeatureMixin.extract_features:704" ) continue if hasattr(pe, "DIRECTORY_ENTRY_TLS") and hasattr(pe.DIRECTORY_ENTRY_TLS, "struct"): if getattr(pe.DIRECTORY_ENTRY_TLS.struct, "AddressOfCallBacks", 0) != 0: base["HasTlsCallbacks"] = 1.0 if hasattr(pe, "VS_FIXEDFILEINFO") and len(pe.VS_FIXEDFILEINFO) > 0: flags = getattr(pe.VS_FIXEDFILEINFO[0], "FileFlags", 0) base["IsDebug"] = 1.0 if flags & 0x1 else 0.0 base["IsPreRelease"] = 1.0 if flags & 0x2 else 0.0 base["IsPatched"] = 1.0 if flags & 0x4 else 0.0 base["IsPrivateBuild"] = 1.0 if flags & 0x8 else 0.0 base["IsSpecialBuild"] = 1.0 if flags & 0x20 else 0.0 if hasattr(pe, "DIRECTORY_ENTRY_IMPORT"): base["ImportCount"] = float(len(pe.DIRECTORY_ENTRY_IMPORT)) func_count = 0 for entry in pe.DIRECTORY_ENTRY_IMPORT: if getattr(entry, "dll", None): try: dlls.add(entry.dll.decode("ascii", "ignore").lower()) except Exception: log_exception("PYAS_Features.PEFeatureMixin.extract_features:727") pass for imp in getattr(entry, "imports", []): func_count += 1 if getattr(imp, "name", None): try: apis.add(imp.name.decode("ascii", "ignore")) except Exception: log_exception("PYAS_Features.PEFeatureMixin.extract_features:735") pass base["ImportFunctionCount"] = float(func_count) else: base["ImportCount"] = 0.0 base["ImportFunctionCount"] = 0.0 if hasattr(pe, "DIRECTORY_ENTRY_EXPORT") and hasattr( pe.DIRECTORY_ENTRY_EXPORT, "symbols" ): base["ExportCount"] = float(len(pe.DIRECTORY_ENTRY_EXPORT.symbols)) else: base["ExportCount"] = 0.0 if hasattr(pe, "DIRECTORY_ENTRY_RESOURCE"): icon_count = 0 for entry in pe.DIRECTORY_ENTRY_RESOURCE.entries: if getattr(entry, "id", None) == 3 and hasattr(entry, "directory"): icon_count += len(getattr(entry.directory, "entries", [])) base["IconCount"] = float(icon_count) else: base["IconCount"] = 0.0 if hasattr(pe, "DIRECTORY_ENTRY_DEBUG"): base["DebugCount"] = float(len(pe.DIRECTORY_ENTRY_DEBUG)) else: base["DebugCount"] = 0.0 base.update(stream_features) overlay_size = fsize - int(overlay_offset) base.update( { "HasOverlay": 1.0 if overlay_size else 0.0, "OverlaySize": float(overlay_size), "OverlayRatio": float(overlay_size) / float(fsize) if overlay_size else 0.0, "OverlayEntropy": overlay_entropy if overlay_size else 0.0, } ) base.update(self._extract_rich_header(pe)) base.update(self._extract_ep_anomalies(pe)) base.update(self._extract_advanced_resources(pe, resource_entropies)) base.update(self._extract_load_config(pe)) base.update(self._extract_security_directory(pe, file_path, fsize)) for k in base: base[k] = self._safe_float(base[k]) return {"Base": base, "DLLs": list(dlls), "APIs": list(apis)} except Exception: log_exception("PYAS_Features.PEFeatureMixin.extract_features:782") return None finally: if pe: pe.close()