diff --git a/capa/capabilities/common.py b/capa/capabilities/common.py index 47b93625..ba8b97f8 100644 --- a/capa/capabilities/common.py +++ b/capa/capabilities/common.py @@ -21,8 +21,16 @@ from dataclasses import dataclass from capa.rules import Rule, Scope, RuleSet from capa.engine import FeatureSet, MatchResults from capa.features.address import NO_ADDRESS -from capa.render.result_document import LibraryFunction, StaticFeatureCounts, DynamicFeatureCounts -from capa.features.extractors.base_extractor import FeatureExtractor, StaticFeatureExtractor, DynamicFeatureExtractor +from capa.render.result_document import ( + LibraryFunction, + StaticFeatureCounts, + DynamicFeatureCounts, +) +from capa.features.extractors.base_extractor import ( + FeatureExtractor, + StaticFeatureExtractor, + DynamicFeatureExtractor, +) logger = logging.getLogger(__name__) @@ -39,7 +47,9 @@ def find_file_capabilities( ) -> FileCapabilities: file_features: FeatureSet = collections.defaultdict(set) - for feature, va in itertools.chain(extractor.extract_file_features(), extractor.extract_global_features()): + for feature, va in itertools.chain( + extractor.extract_file_features(), extractor.extract_global_features() + ): # not all file features may have virtual addresses. # if not, then at least ensure the feature shows up in the index. # the set of addresses will still be empty. @@ -64,7 +74,9 @@ class Capabilities: library_functions: Optional[tuple[LibraryFunction, ...]] = None -def find_capabilities(ruleset: RuleSet, extractor: FeatureExtractor, disable_progress=None, **kwargs) -> Capabilities: +def find_capabilities( + ruleset: RuleSet, extractor: FeatureExtractor, disable_progress=None, **kwargs +) -> Capabilities: from capa.capabilities.static import find_static_capabilities from capa.capabilities.dynamic import find_dynamic_capabilities @@ -72,14 +84,20 @@ def find_capabilities(ruleset: RuleSet, extractor: FeatureExtractor, disable_pro # for the time being, extractors are either static or dynamic. # Remove this assertion once that has changed assert not isinstance(extractor, DynamicFeatureExtractor) - return find_static_capabilities(ruleset, extractor, disable_progress=disable_progress, **kwargs) + return find_static_capabilities( + ruleset, extractor, disable_progress=disable_progress, **kwargs + ) if isinstance(extractor, DynamicFeatureExtractor): - return find_dynamic_capabilities(ruleset, extractor, disable_progress=disable_progress, **kwargs) + return find_dynamic_capabilities( + ruleset, extractor, disable_progress=disable_progress, **kwargs + ) raise ValueError(f"unexpected extractor type: {extractor.__class__.__name__}") -def has_limitation(rules: list, capabilities: Capabilities | FileCapabilities, is_standalone: bool) -> bool: +def has_limitation( + rules: list, capabilities: Capabilities | FileCapabilities, is_standalone: bool +) -> bool: for rule in rules: if rule.name not in capabilities.matches: @@ -90,7 +108,9 @@ def has_limitation(rules: list, capabilities: Capabilities | FileCapabilities, i logger.warning(" Identified via rule: %s", rule.name) if is_standalone: logger.warning(" ") - logger.warning(" Use -v or -vv if you really want to see the capabilities identified by capa.") + logger.warning( + " Use -v or -vv if you really want to see the capabilities identified by capa." + ) logger.warning("-" * 80) # bail on first file limitation @@ -102,8 +122,12 @@ def is_static_limitation_rule(r: Rule) -> bool: return r.meta.get("namespace", "") == "internal/limitation/static" -def has_static_limitation(rules: RuleSet, capabilities: Capabilities | FileCapabilities, is_standalone=True) -> bool: - file_limitation_rules = list(filter(lambda r: is_static_limitation_rule(r), rules.rules.values())) +def has_static_limitation( + rules: RuleSet, capabilities: Capabilities | FileCapabilities, is_standalone=True +) -> bool: + file_limitation_rules = list( + filter(lambda r: is_static_limitation_rule(r), rules.rules.values()) + ) return has_limitation(file_limitation_rules, capabilities, is_standalone) @@ -111,6 +135,10 @@ def is_dynamic_limitation_rule(r: Rule) -> bool: return r.meta.get("namespace", "") == "internal/limitation/dynamic" -def has_dynamic_limitation(rules: RuleSet, capabilities: Capabilities | FileCapabilities, is_standalone=True) -> bool: - dynamic_limitation_rules = list(filter(lambda r: is_dynamic_limitation_rule(r), rules.rules.values())) +def has_dynamic_limitation( + rules: RuleSet, capabilities: Capabilities | FileCapabilities, is_standalone=True +) -> bool: + dynamic_limitation_rules = list( + filter(lambda r: is_dynamic_limitation_rule(r), rules.rules.values()) + ) return has_limitation(dynamic_limitation_rules, capabilities, is_standalone) diff --git a/capa/features/extractors/base_extractor.py b/capa/features/extractors/base_extractor.py index a930d96b..bc990366 100644 --- a/capa/features/extractors/base_extractor.py +++ b/capa/features/extractors/base_extractor.py @@ -23,7 +23,13 @@ from dataclasses import dataclass import capa.features.address from capa.features.common import Feature -from capa.features.address import Address, ThreadAddress, ProcessAddress, DynamicCallAddress, AbsoluteVirtualAddress +from capa.features.address import ( + Address, + ThreadAddress, + ProcessAddress, + DynamicCallAddress, + AbsoluteVirtualAddress, +) # feature extractors may reference functions, BBs, insns by opaque handle values. # you can use the `.address` property to get and render the address of the feature. @@ -47,7 +53,9 @@ class SampleHashes: sha1.update(buf) sha256.update(buf) - return cls(md5=md5.hexdigest(), sha1=sha1.hexdigest(), sha256=sha256.hexdigest()) + return cls( + md5=md5.hexdigest(), sha1=sha1.hexdigest(), sha256=sha256.hexdigest() + ) @dataclass @@ -119,7 +127,9 @@ class StaticFeatureExtractor(abc.ABC): self._sample_hashes = hashes @abc.abstractmethod - def get_base_address(self) -> Union[AbsoluteVirtualAddress, capa.features.address._NoAddress]: + def get_base_address( + self, + ) -> Union[AbsoluteVirtualAddress, capa.features.address._NoAddress]: """ fetch the preferred load address at which the sample was analyzed. @@ -212,7 +222,9 @@ class StaticFeatureExtractor(abc.ABC): raise KeyError(addr) @abc.abstractmethod - def extract_function_features(self, f: FunctionHandle) -> Iterator[tuple[Feature, Address]]: + def extract_function_features( + self, f: FunctionHandle + ) -> Iterator[tuple[Feature, Address]]: """ extract function-scope features. the arguments are opaque values previously provided by `.get_functions()`, etc. @@ -241,7 +253,9 @@ class StaticFeatureExtractor(abc.ABC): raise NotImplementedError() @abc.abstractmethod - def extract_basic_block_features(self, f: FunctionHandle, bb: BBHandle) -> Iterator[tuple[Feature, Address]]: + def extract_basic_block_features( + self, f: FunctionHandle, bb: BBHandle + ) -> Iterator[tuple[Feature, Address]]: """ extract basic block-scope features. the arguments are opaque values previously provided by `.get_functions()`, etc. @@ -299,7 +313,9 @@ class StaticFeatureExtractor(abc.ABC): raise NotImplementedError() -def FunctionFilter(extractor: StaticFeatureExtractor, functions: set) -> StaticFeatureExtractor: +def FunctionFilter( + extractor: StaticFeatureExtractor, functions: set +) -> StaticFeatureExtractor: original_get_functions = extractor.get_functions def filtered_get_functions(self): @@ -425,7 +441,9 @@ class DynamicFeatureExtractor(abc.ABC): raise NotImplementedError() @abc.abstractmethod - def extract_process_features(self, ph: ProcessHandle) -> Iterator[tuple[Feature, Address]]: + def extract_process_features( + self, ph: ProcessHandle + ) -> Iterator[tuple[Feature, Address]]: """ Yields all the features of a process. These include: - file features of the process' image @@ -448,7 +466,9 @@ class DynamicFeatureExtractor(abc.ABC): raise NotImplementedError() @abc.abstractmethod - def extract_thread_features(self, ph: ProcessHandle, th: ThreadHandle) -> Iterator[tuple[Feature, Address]]: + def extract_thread_features( + self, ph: ProcessHandle, th: ThreadHandle + ) -> Iterator[tuple[Feature, Address]]: """ Yields all the features of a thread. These include: - sequenced api traces @@ -484,7 +504,9 @@ class DynamicFeatureExtractor(abc.ABC): raise NotImplementedError() -def ProcessFilter(extractor: DynamicFeatureExtractor, pids: set[int]) -> DynamicFeatureExtractor: +def ProcessFilter( + extractor: DynamicFeatureExtractor, pids: set[int] +) -> DynamicFeatureExtractor: original_get_processes = extractor.get_processes def filtered_get_processes(self): @@ -500,7 +522,9 @@ def ProcessFilter(extractor: DynamicFeatureExtractor, pids: set[int]) -> Dynamic return new_extractor -def ThreadFilter(extractor: DynamicFeatureExtractor, threads: set[Address]) -> DynamicFeatureExtractor: +def ThreadFilter( + extractor: DynamicFeatureExtractor, threads: set[Address] +) -> DynamicFeatureExtractor: original_get_threads = extractor.get_threads def filtered_get_threads(self, ph: ProcessHandle): diff --git a/capa/features/extractors/common.py b/capa/features/extractors/common.py index 27c7e2e7..1cd153cd 100644 --- a/capa/features/extractors/common.py +++ b/capa/features/extractors/common.py @@ -88,7 +88,9 @@ def extract_format(buf: bytes) -> Iterator[tuple[Feature, Address]]: def extract_arch(buf) -> Iterator[tuple[Feature, Address]]: if buf.startswith(MATCH_PE): - yield from capa.features.extractors.pefile.extract_file_arch(pe=pefile.PE(data=buf)) + yield from capa.features.extractors.pefile.extract_file_arch( + pe=pefile.PE(data=buf) + ) elif buf.startswith(MATCH_RESULT): yield Arch(ARCH_ANY), NO_ADDRESS @@ -114,7 +116,10 @@ def extract_arch(buf) -> Iterator[tuple[Feature, Address]]: # rules that rely on arch conditions will fail to match on shellcode. # # for (2), this logic will need to be updated as the format is implemented. - logger.debug("unsupported file format: %s, will not guess Arch", binascii.hexlify(buf[:4]).decode("ascii")) + logger.debug( + "unsupported file format: %s, will not guess Arch", + binascii.hexlify(buf[:4]).decode("ascii"), + ) return @@ -146,5 +151,8 @@ def extract_os(buf, os=OS_AUTO) -> Iterator[tuple[Feature, Address]]: # rules that rely on OS conditions will fail to match on shellcode. # # for (2), this logic will need to be updated as the format is implemented. - logger.debug("unsupported file format: %s, will not guess OS", binascii.hexlify(buf[:4]).decode("ascii")) + logger.debug( + "unsupported file format: %s, will not guess OS", + binascii.hexlify(buf[:4]).decode("ascii"), + ) return diff --git a/capa/features/extractors/dotnetfile.py b/capa/features/extractors/dotnetfile.py index 3de80d20..ab5a4ae3 100644 --- a/capa/features/extractors/dotnetfile.py +++ b/capa/features/extractors/dotnetfile.py @@ -62,23 +62,31 @@ def extract_file_format(**kwargs) -> Iterator[tuple[Format, Address]]: yield Format(FORMAT_PE), NO_ADDRESS -def extract_file_import_names(pe: dnfile.dnPE, **kwargs) -> Iterator[tuple[Import, Address]]: +def extract_file_import_names( + pe: dnfile.dnPE, **kwargs +) -> Iterator[tuple[Import, Address]]: for method in get_dotnet_managed_imports(pe): # like System.IO.File::OpenRead yield Import(str(method)), DNTokenAddress(method.token) for imp in get_dotnet_unmanaged_imports(pe): # like kernel32.CreateFileA - for name in capa.features.extractors.helpers.generate_symbols(imp.module, imp.method, include_dll=True): + for name in capa.features.extractors.helpers.generate_symbols( + imp.module, imp.method, include_dll=True + ): yield Import(name), DNTokenAddress(imp.token) -def extract_file_function_names(pe: dnfile.dnPE, **kwargs) -> Iterator[tuple[FunctionName, Address]]: +def extract_file_function_names( + pe: dnfile.dnPE, **kwargs +) -> Iterator[tuple[FunctionName, Address]]: for method in get_dotnet_managed_methods(pe): yield FunctionName(str(method)), DNTokenAddress(method.token) -def extract_file_namespace_features(pe: dnfile.dnPE, **kwargs) -> Iterator[tuple[Namespace, Address]]: +def extract_file_namespace_features( + pe: dnfile.dnPE, **kwargs +) -> Iterator[tuple[Namespace, Address]]: """emit namespace features from TypeRef and TypeDef tables""" # namespaces may be referenced multiple times, so we need to filter @@ -102,7 +110,9 @@ def extract_file_namespace_features(pe: dnfile.dnPE, **kwargs) -> Iterator[tuple yield Namespace(namespace), NO_ADDRESS -def extract_file_class_features(pe: dnfile.dnPE, **kwargs) -> Iterator[tuple[Class, Address]]: +def extract_file_class_features( + pe: dnfile.dnPE, **kwargs +) -> Iterator[tuple[Class, Address]]: """emit class features from TypeRef and TypeDef tables""" nested_class_table = get_dotnet_nested_class_table_index(pe) @@ -110,19 +120,29 @@ def extract_file_class_features(pe: dnfile.dnPE, **kwargs) -> Iterator[tuple[Cla # emit internal .NET classes assert isinstance(typedef, dnfile.mdtable.TypeDefRow) - typedefnamespace, typedefname = resolve_nested_typedef_name(nested_class_table, rid, typedef, pe) + typedefnamespace, typedefname = resolve_nested_typedef_name( + nested_class_table, rid, typedef, pe + ) token = calculate_dotnet_token_value(dnfile.mdtable.TypeDef.number, rid) - yield Class(DnType.format_name(typedefname, namespace=typedefnamespace)), DNTokenAddress(token) + yield ( + Class(DnType.format_name(typedefname, namespace=typedefnamespace)), + DNTokenAddress(token), + ) for rid, typeref in iter_dotnet_table(pe, dnfile.mdtable.TypeRef.number): # emit external .NET classes assert isinstance(typeref, dnfile.mdtable.TypeRefRow) - typerefnamespace, typerefname = resolve_nested_typeref_name(typeref.ResolutionScope.row_index, typeref, pe) + typerefnamespace, typerefname = resolve_nested_typeref_name( + typeref.ResolutionScope.row_index, typeref, pe + ) token = calculate_dotnet_token_value(dnfile.mdtable.TypeRef.number, rid) - yield Class(DnType.format_name(typerefname, namespace=typerefnamespace)), DNTokenAddress(token) + yield ( + Class(DnType.format_name(typerefname, namespace=typerefnamespace)), + DNTokenAddress(token), + ) def extract_file_os(**kwargs) -> Iterator[tuple[OS, Address]]: @@ -137,7 +157,10 @@ def extract_file_arch(pe: dnfile.dnPE, **kwargs) -> Iterator[tuple[Arch, Address if pe.net.Flags.CLR_32BITREQUIRED and pe.PE_TYPE == pefile.OPTIONAL_HEADER_MAGIC_PE: yield Arch(ARCH_I386), NO_ADDRESS - elif not pe.net.Flags.CLR_32BITREQUIRED and pe.PE_TYPE == pefile.OPTIONAL_HEADER_MAGIC_PE_PLUS: + elif ( + not pe.net.Flags.CLR_32BITREQUIRED + and pe.PE_TYPE == pefile.OPTIONAL_HEADER_MAGIC_PE_PLUS + ): yield Arch(ARCH_AMD64), NO_ADDRESS else: yield Arch(ARCH_ANY), NO_ADDRESS @@ -236,25 +259,41 @@ class DotnetFileFeatureExtractor(StaticFeatureExtractor): return vbuf.rstrip(b"\x00").decode("utf-8") def get_functions(self): - raise NotImplementedError("DotnetFileFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "DotnetFileFeatureExtractor can only be used to extract file features" + ) def extract_function_features(self, f): - raise NotImplementedError("DotnetFileFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "DotnetFileFeatureExtractor can only be used to extract file features" + ) def get_basic_blocks(self, f): - raise NotImplementedError("DotnetFileFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "DotnetFileFeatureExtractor can only be used to extract file features" + ) def extract_basic_block_features(self, f, bb): - raise NotImplementedError("DotnetFileFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "DotnetFileFeatureExtractor can only be used to extract file features" + ) def get_instructions(self, f, bb): - raise NotImplementedError("DotnetFileFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "DotnetFileFeatureExtractor can only be used to extract file features" + ) def extract_insn_features(self, f, bb, insn): - raise NotImplementedError("DotnetFileFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "DotnetFileFeatureExtractor can only be used to extract file features" + ) def is_library_function(self, va): - raise NotImplementedError("DotnetFileFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "DotnetFileFeatureExtractor can only be used to extract file features" + ) def get_function_name(self, va): - raise NotImplementedError("DotnetFileFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "DotnetFileFeatureExtractor can only be used to extract file features" + ) diff --git a/capa/features/extractors/elf.py b/capa/features/extractors/elf.py index 477b5f2b..0c7071bb 100644 --- a/capa/features/extractors/elf.py +++ b/capa/features/extractors/elf.py @@ -167,21 +167,34 @@ class ELF: raise CorruptElfFile(f"not an ELF file: invalid ei_data: 0x{ei_data:02x}") if self.bitness == 32: - e_phoff, e_shoff = struct.unpack_from(self.endian + "II", self.file_header, 0x1C) - self.e_phentsize, self.e_phnum = struct.unpack_from(self.endian + "HH", self.file_header, 0x2A) + e_phoff, e_shoff = struct.unpack_from( + self.endian + "II", self.file_header, 0x1C + ) + self.e_phentsize, self.e_phnum = struct.unpack_from( + self.endian + "HH", self.file_header, 0x2A + ) self.e_shentsize, self.e_shnum, self.e_shstrndx = struct.unpack_from( self.endian + "HHH", self.file_header, 0x2E ) elif self.bitness == 64: - e_phoff, e_shoff = struct.unpack_from(self.endian + "QQ", self.file_header, 0x20) - self.e_phentsize, self.e_phnum = struct.unpack_from(self.endian + "HH", self.file_header, 0x36) + e_phoff, e_shoff = struct.unpack_from( + self.endian + "QQ", self.file_header, 0x20 + ) + self.e_phentsize, self.e_phnum = struct.unpack_from( + self.endian + "HH", self.file_header, 0x36 + ) self.e_shentsize, self.e_shnum, self.e_shstrndx = struct.unpack_from( self.endian + "HHH", self.file_header, 0x3A ) else: raise NotImplementedError() - logger.debug("e_phoff: 0x%02x e_phentsize: 0x%02x e_phnum: %d", e_phoff, self.e_phentsize, self.e_phnum) + logger.debug( + "e_phoff: 0x%02x e_phentsize: 0x%02x e_phnum: %d", + e_phoff, + self.e_phentsize, + self.e_phnum, + ) self.f.seek(e_phoff) program_header_size = self.e_phnum * self.e_phentsize @@ -332,12 +345,12 @@ class ELF: phent = self.phbuf[phent_offset : phent_offset + self.e_phentsize] if self.bitness == 32: - p_type, p_offset, p_vaddr, p_paddr, p_filesz, p_memsz, p_flags = struct.unpack_from( - self.endian + "IIIIIII", phent, 0x0 + p_type, p_offset, p_vaddr, p_paddr, p_filesz, p_memsz, p_flags = ( + struct.unpack_from(self.endian + "IIIIIII", phent, 0x0) ) elif self.bitness == 64: - p_type, p_flags, p_offset, p_vaddr, p_paddr, p_filesz, p_memsz = struct.unpack_from( - self.endian + "IIQQQQQ", phent, 0x0 + p_type, p_flags, p_offset, p_vaddr, p_paddr, p_filesz, p_memsz = ( + struct.unpack_from(self.endian + "IIQQQQQ", phent, 0x0) ) else: raise NotImplementedError() @@ -362,13 +375,31 @@ class ELF: shent = self.shbuf[shent_offset : shent_offset + self.e_shentsize] if self.bitness == 32: - sh_name, sh_type, sh_flags, sh_addr, sh_offset, sh_size, sh_link, _, _, sh_entsize = struct.unpack_from( - self.endian + "IIIIIIIIII", shent, 0x0 - ) + ( + sh_name, + sh_type, + sh_flags, + sh_addr, + sh_offset, + sh_size, + sh_link, + _, + _, + sh_entsize, + ) = struct.unpack_from(self.endian + "IIIIIIIIII", shent, 0x0) elif self.bitness == 64: - sh_name, sh_type, sh_flags, sh_addr, sh_offset, sh_size, sh_link, _, _, sh_entsize = struct.unpack_from( - self.endian + "IIQQQQIIQQ", shent, 0x0 - ) + ( + sh_name, + sh_type, + sh_flags, + sh_addr, + sh_offset, + sh_size, + sh_link, + _, + _, + sh_entsize, + ) = struct.unpack_from(self.endian + "IIQQQQIIQQ", shent, 0x0) else: raise NotImplementedError() @@ -377,7 +408,17 @@ class ELF: if len(buf) != sh_size: raise ValueError("failed to read section header content") - return Shdr(sh_name, sh_type, sh_flags, sh_addr, sh_offset, sh_size, sh_link, sh_entsize, buf) + return Shdr( + sh_name, + sh_type, + sh_flags, + sh_addr, + sh_offset, + sh_size, + sh_link, + sh_entsize, + buf, + ) @property def section_headers(self): @@ -442,7 +483,9 @@ class ELF: vna_offset = vn_offset + vn_aux for _ in range(vn_cnt): # ElfXX_Vernaux layout is the same on 32 and 64 bit - _, _, _, vna_name, vna_next = struct.unpack_from(self.endian + "IHHII", shdr.buf, vna_offset) + _, _, _, vna_name, vna_next = struct.unpack_from( + self.endian + "IHHII", shdr.buf, vna_offset + ) # ABI names, like: "GLIBC_2.2.5" abi = read_cstr(linked_shdr.buf, vna_name) @@ -473,10 +516,14 @@ class ELF: offset = 0x0 while True: if self.bitness == 32: - d_tag, d_val = struct.unpack_from(self.endian + "II", phdr.buf, offset) + d_tag, d_val = struct.unpack_from( + self.endian + "II", phdr.buf, offset + ) offset += 8 elif self.bitness == 64: - d_tag, d_val = struct.unpack_from(self.endian + "QQ", phdr.buf, offset) + d_tag, d_val = struct.unpack_from( + self.endian + "QQ", phdr.buf, offset + ) offset += 16 else: raise NotImplementedError() @@ -592,13 +639,24 @@ class PHNote: self._parse() def _parse(self): - namesz, self.descsz, self.type_ = struct.unpack_from(self.endian + "III", self.buf, 0x0) + namesz, self.descsz, self.type_ = struct.unpack_from( + self.endian + "III", self.buf, 0x0 + ) name_offset = 0xC self.desc_offset = name_offset + align(namesz, 0x4) - logger.debug("ph:namesz: 0x%02x descsz: 0x%02x type: 0x%04x", namesz, self.descsz, self.type_) + logger.debug( + "ph:namesz: 0x%02x descsz: 0x%02x type: 0x%04x", + namesz, + self.descsz, + self.type_, + ) - self.name = self.buf[name_offset : name_offset + namesz].partition(b"\x00")[0].decode("ascii") + self.name = ( + self.buf[name_offset : name_offset + namesz] + .partition(b"\x00")[0] + .decode("ascii") + ) logger.debug("name: %s", self.name) @property @@ -616,14 +674,22 @@ class PHNote: return None desc = self.buf[self.desc_offset : self.desc_offset + self.descsz] - abi_tag, kmajor, kminor, kpatch = struct.unpack_from(self.endian + "IIII", desc, 0x0) + abi_tag, kmajor, kminor, kpatch = struct.unpack_from( + self.endian + "IIII", desc, 0x0 + ) logger.debug("GNU_ABI_TAG: 0x%02x", abi_tag) os = GNU_ABI_TAG.get(abi_tag) if not os: return None - logger.debug("abi tag: %s earliest compatible kernel: %d.%d.%d", os, kmajor, kminor, kpatch) + logger.debug( + "abi tag: %s earliest compatible kernel: %d.%d.%d", + os, + kmajor, + kminor, + kpatch, + ) return ABITag(os, kmajor, kminor, kpatch) @@ -641,11 +707,18 @@ class SHNote: self._parse() def _parse(self): - namesz, self.descsz, self.type_ = struct.unpack_from(self.endian + "III", self.buf, 0x0) + namesz, self.descsz, self.type_ = struct.unpack_from( + self.endian + "III", self.buf, 0x0 + ) name_offset = 0xC self.desc_offset = name_offset + align(namesz, 0x4) - logger.debug("sh:namesz: 0x%02x descsz: 0x%02x type: 0x%04x", namesz, self.descsz, self.type_) + logger.debug( + "sh:namesz: 0x%02x descsz: 0x%02x type: 0x%04x", + namesz, + self.descsz, + self.type_, + ) name_buf = self.buf[name_offset : name_offset + namesz] self.name = read_cstr(name_buf, 0x0) @@ -660,14 +733,22 @@ class SHNote: return None desc = self.buf[self.desc_offset : self.desc_offset + self.descsz] - abi_tag, kmajor, kminor, kpatch = struct.unpack_from(self.endian + "IIII", desc, 0x0) + abi_tag, kmajor, kminor, kpatch = struct.unpack_from( + self.endian + "IIII", desc, 0x0 + ) logger.debug("GNU_ABI_TAG: 0x%02x", abi_tag) os = GNU_ABI_TAG.get(abi_tag) if not os: return None - logger.debug("abi tag: %s earliest compatible kernel: %d.%d.%d", os, kmajor, kminor, kpatch) + logger.debug( + "abi tag: %s earliest compatible kernel: %d.%d.%d", + os, + kmajor, + kminor, + kpatch, + ) return ABITag(os, kmajor, kminor, kpatch) @@ -746,9 +827,12 @@ class SymTab: for section in elf.sections: if section.sh_type == SHT_SYMTAB: strtab_section = elf.sections[section.sh_link] - sh_symtab = Shdr.from_viv(section, elf.readAtOffset(section.sh_offset, section.sh_size)) + sh_symtab = Shdr.from_viv( + section, elf.readAtOffset(section.sh_offset, section.sh_size) + ) sh_strtab = Shdr.from_viv( - strtab_section, elf.readAtOffset(strtab_section.sh_offset, strtab_section.sh_size) + strtab_section, + elf.readAtOffset(strtab_section.sh_offset, strtab_section.sh_size), ) try: @@ -903,7 +987,9 @@ def guess_os_from_abi_versions_needed(elf: ELF) -> Optional[OS]: # this will let us guess about linux/hurd in some cases. versions_needed = elf.versions_needed - if any(abi.startswith("GLIBC") for abi in itertools.chain(*versions_needed.values())): + if any( + abi.startswith("GLIBC") for abi in itertools.chain(*versions_needed.values()) + ): # there are any GLIBC versions needed if elf.e_machine != "i386": @@ -1100,7 +1186,12 @@ def guess_os_from_go_buildinfo(elf: ELF) -> Optional[OS]: assert psize in (4, 8) is_big_endian = flags & 0b01 has_inline_strings = flags & 0b10 - logger.debug("go buildinfo: psize: %d big endian: %s inline: %s", psize, is_big_endian, has_inline_strings) + logger.debug( + "go buildinfo: psize: %d big endian: %s inline: %s", + psize, + is_big_endian, + has_inline_strings, + ) GOOS_TO_OS = { b"aix": OS.AIX, @@ -1169,7 +1260,9 @@ def guess_os_from_go_buildinfo(elf: ELF) -> Optional[OS]: build_version = read_go_slice(elf, build_version_address) if build_version: - logger.debug("go buildinfo: build version: %s", build_version.decode("utf-8")) + logger.debug( + "go buildinfo: build version: %s", build_version.decode("utf-8") + ) modinfo = read_go_slice(elf, modinfo_address) if modinfo: @@ -1461,7 +1554,12 @@ def guess_os_from_vdso_strings(elf: ELF) -> Optional[OS]: ("x86/32", b"__vdso_time", b"LINUX_2.6"), ): if symbol in buf and version in buf: - logger.debug("vdso string: %s %s %s", arch, symbol.decode("ascii"), version.decode("ascii")) + logger.debug( + "vdso string: %s %s %s", + arch, + symbol.decode("ascii"), + version.decode("ascii"), + ) return OS.LINUX return None diff --git a/capa/features/extractors/elffile.py b/capa/features/extractors/elffile.py index a8fb10db..05432aa5 100644 --- a/capa/features/extractors/elffile.py +++ b/capa/features/extractors/elffile.py @@ -37,7 +37,11 @@ def extract_file_export_names(elf: ELFFile, **kwargs): logger.debug("Symbol table '%s' has a sh_entsize of zero!", section.name) continue - logger.debug("Symbol table '%s' contains %s entries:", section.name, section.num_symbols()) + logger.debug( + "Symbol table '%s' contains %s entries:", + section.name, + section.num_symbols(), + ) for symbol in section.iter_symbols(): # The following conditions are based on the following article @@ -113,7 +117,9 @@ def extract_file_import_names(elf: ELFFile, **kwargs): continue relocation_tables = segment.get_relocation_tables() - logger.debug("Dynamic Segment contains %s relocation tables:", len(relocation_tables)) + logger.debug( + "Dynamic Segment contains %s relocation tables:", len(relocation_tables) + ) for relocation_table in relocation_tables.values(): relocations = [] @@ -126,7 +132,10 @@ def extract_file_import_names(elf: ELFFile, **kwargs): break for relocation in relocations: - if "r_info_sym" not in relocation.entry or "r_offset" not in relocation.entry: + if ( + "r_info_sym" not in relocation.entry + or "r_offset" not in relocation.entry + ): continue symbol_address: int = relocation["r_offset"] @@ -232,25 +241,41 @@ class ElfFeatureExtractor(StaticFeatureExtractor): yield feature, addr def get_functions(self): - raise NotImplementedError("ElfFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "ElfFeatureExtractor can only be used to extract file features" + ) def extract_function_features(self, f): - raise NotImplementedError("ElfFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "ElfFeatureExtractor can only be used to extract file features" + ) def get_basic_blocks(self, f): - raise NotImplementedError("ElfFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "ElfFeatureExtractor can only be used to extract file features" + ) def extract_basic_block_features(self, f, bb): - raise NotImplementedError("ElfFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "ElfFeatureExtractor can only be used to extract file features" + ) def get_instructions(self, f, bb): - raise NotImplementedError("ElfFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "ElfFeatureExtractor can only be used to extract file features" + ) def extract_insn_features(self, f, bb, insn): - raise NotImplementedError("ElfFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "ElfFeatureExtractor can only be used to extract file features" + ) def is_library_function(self, addr): - raise NotImplementedError("ElfFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "ElfFeatureExtractor can only be used to extract file features" + ) def get_function_name(self, addr): - raise NotImplementedError("ElfFeatureExtractor can only be used to extract file features") + raise NotImplementedError( + "ElfFeatureExtractor can only be used to extract file features" + ) diff --git a/capa/features/freeze/__init__.py b/capa/features/freeze/__init__.py index 1de0b161..8f0eb8b0 100644 --- a/capa/features/freeze/__init__.py +++ b/capa/features/freeze/__init__.py @@ -94,12 +94,19 @@ class Address(HashableModel): return cls(type=AddressType.PROCESS, value=(a.ppid, a.pid)) elif isinstance(a, capa.features.address.ThreadAddress): - return cls(type=AddressType.THREAD, value=(a.process.ppid, a.process.pid, a.tid)) + return cls( + type=AddressType.THREAD, value=(a.process.ppid, a.process.pid, a.tid) + ) elif isinstance(a, capa.features.address.DynamicCallAddress): - return cls(type=AddressType.CALL, value=(a.thread.process.ppid, a.thread.process.pid, a.thread.tid, a.id)) + return cls( + type=AddressType.CALL, + value=(a.thread.process.ppid, a.thread.process.pid, a.thread.tid, a.id), + ) - elif a == capa.features.address.NO_ADDRESS or isinstance(a, capa.features.address._NoAddress): + elif a == capa.features.address.NO_ADDRESS or isinstance( + a, capa.features.address._NoAddress + ): return cls(type=AddressType.NO_ADDRESS, value=None) elif isinstance(a, capa.features.address.Address): @@ -146,7 +153,8 @@ class Address(HashableModel): assert isinstance(pid, int) assert isinstance(tid, int) return capa.features.address.ThreadAddress( - process=capa.features.address.ProcessAddress(ppid=ppid, pid=pid), tid=tid + process=capa.features.address.ProcessAddress(ppid=ppid, pid=pid), + tid=tid, ) elif self.type is AddressType.CALL: @@ -154,7 +162,8 @@ class Address(HashableModel): ppid, pid, tid, id_ = self.value return capa.features.address.DynamicCallAddress( thread=capa.features.address.ThreadAddress( - process=capa.features.address.ProcessAddress(ppid=ppid, pid=pid), tid=tid + process=capa.features.address.ProcessAddress(ppid=ppid, pid=pid), + tid=tid, ), id=id_, ) @@ -569,16 +578,26 @@ def loads_static(s: str) -> StaticFeatureExtractor: base_address=freeze.base_address.to_capa(), sample_hashes=freeze.sample_hashes, global_features=[f.feature.to_capa() for f in freeze.features.global_], - file_features=[(f.address.to_capa(), f.feature.to_capa()) for f in freeze.features.file], + file_features=[ + (f.address.to_capa(), f.feature.to_capa()) for f in freeze.features.file + ], functions={ f.address.to_capa(): null.FunctionFeatures( - features=[(fe.address.to_capa(), fe.feature.to_capa()) for fe in f.features], + features=[ + (fe.address.to_capa(), fe.feature.to_capa()) for fe in f.features + ], basic_blocks={ bb.address.to_capa(): null.BasicBlockFeatures( - features=[(fe.address.to_capa(), fe.feature.to_capa()) for fe in bb.features], + features=[ + (fe.address.to_capa(), fe.feature.to_capa()) + for fe in bb.features + ], instructions={ i.address.to_capa(): null.InstructionFeatures( - features=[(fe.address.to_capa(), fe.feature.to_capa()) for fe in i.features] + features=[ + (fe.address.to_capa(), fe.feature.to_capa()) + for fe in i.features + ] ) for i in bb.instructions }, @@ -604,18 +623,28 @@ def loads_dynamic(s: str) -> DynamicFeatureExtractor: base_address=freeze.base_address.to_capa(), sample_hashes=freeze.sample_hashes, global_features=[f.feature.to_capa() for f in freeze.features.global_], - file_features=[(f.address.to_capa(), f.feature.to_capa()) for f in freeze.features.file], + file_features=[ + (f.address.to_capa(), f.feature.to_capa()) for f in freeze.features.file + ], processes={ p.address.to_capa(): null.ProcessFeatures( name=p.name, - features=[(fe.address.to_capa(), fe.feature.to_capa()) for fe in p.features], + features=[ + (fe.address.to_capa(), fe.feature.to_capa()) for fe in p.features + ], threads={ t.address.to_capa(): null.ThreadFeatures( - features=[(fe.address.to_capa(), fe.feature.to_capa()) for fe in t.features], + features=[ + (fe.address.to_capa(), fe.feature.to_capa()) + for fe in t.features + ], calls={ c.address.to_capa(): null.CallFeatures( name=c.name, - features=[(fe.address.to_capa(), fe.feature.to_capa()) for fe in c.features], + features=[ + (fe.address.to_capa(), fe.feature.to_capa()) + for fe in c.features + ], ) for c in t.calls }, @@ -687,7 +716,9 @@ def main(argv=None): argv = sys.argv[1:] parser = argparse.ArgumentParser(description="save capa features to a file") - capa.main.install_common_args(parser, {"input_file", "format", "backend", "os", "signatures"}) + capa.main.install_common_args( + parser, {"input_file", "format", "backend", "os", "signatures"} + ) parser.add_argument("output", type=str, help="Path to output file") args = parser.parse_args(args=argv) diff --git a/capa/helpers.py b/capa/helpers.py index 7656f334..151ba5e4 100644 --- a/capa/helpers.py +++ b/capa/helpers.py @@ -282,7 +282,9 @@ def log_unsupported_cape_report_error(error: str): logger.error("-" * 80) logger.error(" Input file is not a valid CAPE report: %s", error) logger.error(" ") - logger.error(" capa currently only supports analyzing standard CAPE reports in JSON format.") + logger.error( + " capa currently only supports analyzing standard CAPE reports in JSON format." + ) logger.error( " Please make sure your report file is in the standard format and contains both the static and dynamic sections." ) @@ -293,7 +295,9 @@ def log_unsupported_drakvuf_report_error(error: str): logger.error("-" * 80) logger.error(" Input file is not a valid DRAKVUF output file: %s", error) logger.error(" ") - logger.error(" capa currently only supports analyzing standard DRAKVUF outputs in JSONL format.") + logger.error( + " capa currently only supports analyzing standard DRAKVUF outputs in JSONL format." + ) logger.error( " Please make sure your report file is in the standard format and contains both the static and dynamic sections." ) @@ -307,15 +311,23 @@ def log_unsupported_vmray_report_error(error: str): logger.error( " capa only supports analyzing VMRay dynamic analysis archives containing summary_v2.json and flog.xml log files." ) - logger.error(" Please make sure you have downloaded a dynamic analysis archive from VMRay.") + logger.error( + " Please make sure you have downloaded a dynamic analysis archive from VMRay." + ) logger.error("-" * 80) def log_empty_sandbox_report_error(error: str, sandbox_name: str): logger.error("-" * 80) - logger.error(" %s report is empty or only contains little useful data: %s", sandbox_name, error) + logger.error( + " %s report is empty or only contains little useful data: %s", + sandbox_name, + error, + ) logger.error(" ") - logger.error(" Please make sure the sandbox run captures useful behaviour of your sample.") + logger.error( + " Please make sure the sandbox run captures useful behaviour of your sample." + ) logger.error("-" * 80) @@ -326,7 +338,9 @@ def log_unsupported_os_error(): logger.error(" capa currently only analyzes executables for some operating systems") logger.error(" (including Windows, Linux, and Android).") logger.error(" ") - logger.error(" If you know the target OS, you can specify it explicitly, for example:") + logger.error( + " If you know the target OS, you can specify it explicitly, for example:" + ) logger.error(" capa --os linux ") logger.error("-" * 80) @@ -395,7 +409,10 @@ def is_cache_newer_than_rule_code(cache_dir: Path) -> bool: import capa.rules import capa.rules.cache - latest_rule_code_file = max([Path(capa.rules.__file__), Path(capa.rules.cache.__file__)], key=os.path.getmtime) + latest_rule_code_file = max( + [Path(capa.rules.__file__), Path(capa.rules.cache.__file__)], + key=os.path.getmtime, + ) rule_code_timestamp = Path(latest_rule_code_file).stat().st_mtime if rule_code_timestamp > cache_timestamp: diff --git a/capa/loader.py b/capa/loader.py index ec0e93b4..18fb3877 100644 --- a/capa/loader.py +++ b/capa/loader.py @@ -32,7 +32,11 @@ import capa.render.result_document as rdoc import capa.features.extractors.common from capa.rules import RuleSet from capa.engine import MatchResults -from capa.exceptions import UnsupportedOSError, UnsupportedArchError, UnsupportedFormatError +from capa.exceptions import ( + UnsupportedOSError, + UnsupportedArchError, + UnsupportedFormatError, +) from capa.features.common import ( OS_AUTO, FORMAT_PE, @@ -218,9 +222,13 @@ def get_workspace(path: Path, input_format: str, sigpaths: list[Path]): vw = viv_utils.getWorkspace(str(path), analyze=False, should_save=False) elif input_format == FORMAT_SC32: # these are not analyzed nor saved. - vw = viv_utils.getShellcodeWorkspaceFromFile(str(path), arch="i386", analyze=False) + vw = viv_utils.getShellcodeWorkspaceFromFile( + str(path), arch="i386", analyze=False + ) elif input_format == FORMAT_SC64: - vw = viv_utils.getShellcodeWorkspaceFromFile(str(path), arch="amd64", analyze=False) + vw = viv_utils.getShellcodeWorkspaceFromFile( + str(path), arch="amd64", analyze=False + ) else: raise ValueError("unexpected format: " + input_format) except envi.exc.SegmentationViolation as e: @@ -289,12 +297,16 @@ def get_extractor( import capa.features.extractors.drakvuf.extractor report = capa.helpers.load_jsonl_from_path(input_path) - return capa.features.extractors.drakvuf.extractor.DrakvufExtractor.from_report(report) + return capa.features.extractors.drakvuf.extractor.DrakvufExtractor.from_report( + report + ) elif backend == BACKEND_VMRAY: import capa.features.extractors.vmray.extractor - return capa.features.extractors.vmray.extractor.VMRayExtractor.from_zipfile(input_path) + return capa.features.extractors.vmray.extractor.VMRayExtractor.from_zipfile( + input_path + ) elif backend == BACKEND_DOTNET: import capa.features.extractors.dnfile.extractor @@ -302,7 +314,9 @@ def get_extractor( if input_format not in (FORMAT_PE, FORMAT_DOTNET): raise UnsupportedFormatError() - return capa.features.extractors.dnfile.extractor.DnfileFeatureExtractor(input_path) + return capa.features.extractors.dnfile.extractor.DnfileFeatureExtractor( + input_path + ) elif backend == BACKEND_BINJA: import capa.features.extractors.binja.find_binja_api as finder @@ -361,11 +375,15 @@ def get_extractor( vw.saveWorkspace() except IOError: # see #168 for discussion around how to handle non-writable directories - logger.info("source directory is not writable, won't save intermediate workspace") + logger.info( + "source directory is not writable, won't save intermediate workspace" + ) else: logger.debug("CAPA_SAVE_WORKSPACE unset, not saving workspace") - return capa.features.extractors.viv.extractor.VivisectFeatureExtractor(vw, input_path, os_) + return capa.features.extractors.viv.extractor.VivisectFeatureExtractor( + vw, input_path, os_ + ) elif backend == BACKEND_FREEZE: return frz.load(input_path.read_bytes()) @@ -378,7 +396,9 @@ def get_extractor( assert sample_path is not None buf = sample_path.read_bytes() - return capa.features.extractors.binexport2.extractor.BinExport2FeatureExtractor(be2, buf) + return capa.features.extractors.binexport2.extractor.BinExport2FeatureExtractor( + be2, buf + ) elif backend == BACKEND_IDA: import capa.features.extractors.ida.idalib as idalib @@ -409,7 +429,9 @@ def get_extractor( # -1 - Generic errors (database already open, auto-analysis failed, etc.) # -2 - User cancelled operation ret = idapro.open_database( - str(input_path), run_auto_analysis=True, args="-Olumina:host=0.0.0.0 -Osecondary_lumina:host=0.0.0.0 -R" + str(input_path), + run_auto_analysis=True, + args="-Olumina:host=0.0.0.0 -Osecondary_lumina:host=0.0.0.0 -R", ) if ret != 0: raise RuntimeError("failed to analyze input file") @@ -444,12 +466,19 @@ def get_extractor( monitor = TaskMonitor.DUMMY # Import file - loader = pyghidra.program_loader().project(project).source(str(input_path)).name(input_path.name) + loader = ( + pyghidra.program_loader() + .project(project) + .source(str(input_path)) + .name(input_path.name) + ) with loader.load() as load_results: load_results.save(monitor) # Open program - program, consumer = pyghidra.consume_program(project, "/" + input_path.name) + program, consumer = pyghidra.consume_program( + project, "/" + input_path.name + ) # Analyze pyghidra.analyze(program, monitor) @@ -482,7 +511,9 @@ def get_extractor( import capa.features.extractors.ghidra.extractor - return capa.features.extractors.ghidra.extractor.GhidraFeatureExtractor(ctx_manager=cm, tmpdir=tmpdir) + return capa.features.extractors.ghidra.extractor.GhidraFeatureExtractor( + ctx_manager=cm, tmpdir=tmpdir + ) else: raise ValueError("unexpected backend: " + backend) @@ -518,37 +549,55 @@ def get_file_extractors(input_file: Path, input_format: str) -> list[FeatureExtr if input_format == FORMAT_PE: import capa.features.extractors.pefile - file_extractors.append(capa.features.extractors.pefile.PefileFeatureExtractor(input_file)) + file_extractors.append( + capa.features.extractors.pefile.PefileFeatureExtractor(input_file) + ) elif input_format == FORMAT_DOTNET: import capa.features.extractors.pefile import capa.features.extractors.dotnetfile - file_extractors.append(capa.features.extractors.pefile.PefileFeatureExtractor(input_file)) - file_extractors.append(capa.features.extractors.dotnetfile.DotnetFileFeatureExtractor(input_file)) + file_extractors.append( + capa.features.extractors.pefile.PefileFeatureExtractor(input_file) + ) + file_extractors.append( + capa.features.extractors.dotnetfile.DotnetFileFeatureExtractor(input_file) + ) elif input_format == FORMAT_ELF: import capa.features.extractors.elffile - file_extractors.append(capa.features.extractors.elffile.ElfFeatureExtractor(input_file)) + file_extractors.append( + capa.features.extractors.elffile.ElfFeatureExtractor(input_file) + ) elif input_format == FORMAT_CAPE: import capa.features.extractors.cape.extractor report = capa.helpers.load_json_from_path(input_file) - file_extractors.append(capa.features.extractors.cape.extractor.CapeExtractor.from_report(report)) + file_extractors.append( + capa.features.extractors.cape.extractor.CapeExtractor.from_report(report) + ) elif input_format == FORMAT_DRAKVUF: import capa.helpers import capa.features.extractors.drakvuf.extractor report = capa.helpers.load_jsonl_from_path(input_file) - file_extractors.append(capa.features.extractors.drakvuf.extractor.DrakvufExtractor.from_report(report)) + file_extractors.append( + capa.features.extractors.drakvuf.extractor.DrakvufExtractor.from_report( + report + ) + ) elif input_format == FORMAT_VMRAY: import capa.features.extractors.vmray.extractor - file_extractors.append(capa.features.extractors.vmray.extractor.VMRayExtractor.from_zipfile(input_file)) + file_extractors.append( + capa.features.extractors.vmray.extractor.VMRayExtractor.from_zipfile( + input_file + ) + ) elif input_format == FORMAT_BINEXPORT2: file_extractors = _get_binexport2_file_extractors(input_file) @@ -558,7 +607,9 @@ def get_file_extractors(input_file: Path, input_format: str) -> list[FeatureExtr def get_signatures(sigs_path: Path) -> list[Path]: if not sigs_path.exists(): - raise IOError(f"signatures path {sigs_path} does not exist or cannot be accessed") + raise IOError( + f"signatures path {sigs_path} does not exist or cannot be accessed" + ) paths: list[Path] = [] if sigs_path.is_file(): @@ -582,7 +633,9 @@ def get_signatures(sigs_path: Path) -> list[Path]: return paths -def get_sample_analysis(format_, arch, os_, extractor, rules_path, feature_counts, library_functions): +def get_sample_analysis( + format_, arch, os_, extractor, rules_path, feature_counts, library_functions +): if isinstance(extractor, StaticFeatureExtractor): return rdoc.StaticAnalysis( format=format_, @@ -632,12 +685,22 @@ def collect_metadata( md5, sha1, sha256 = sample_hashes.md5, sample_hashes.sha1, sample_hashes.sha256 global_feats = list(extractor.extract_global_features()) - extractor_format = [f.value for (f, _) in global_feats if isinstance(f, capa.features.common.Format)] - extractor_arch = [f.value for (f, _) in global_feats if isinstance(f, capa.features.common.Arch)] - extractor_os = [f.value for (f, _) in global_feats if isinstance(f, capa.features.common.OS)] + extractor_format = [ + f.value for (f, _) in global_feats if isinstance(f, capa.features.common.Format) + ] + extractor_arch = [ + f.value for (f, _) in global_feats if isinstance(f, capa.features.common.Arch) + ] + extractor_os = [ + f.value for (f, _) in global_feats if isinstance(f, capa.features.common.OS) + ] input_format = ( - str(extractor_format[0]) if extractor_format else "unknown" if input_format == FORMAT_AUTO else input_format + str(extractor_format[0]) + if extractor_format + else "unknown" + if input_format == FORMAT_AUTO + else input_format ) arch = str(extractor_arch[0]) if extractor_arch else "unknown" os_ = str(extractor_os[0]) if extractor_os else "unknown" if os_ == OS_AUTO else os_ @@ -757,7 +820,9 @@ def compute_dynamic_layout( return layout -def compute_static_layout(rules: RuleSet, extractor: StaticFeatureExtractor, capabilities) -> rdoc.StaticLayout: +def compute_static_layout( + rules: RuleSet, extractor: StaticFeatureExtractor, capabilities +) -> rdoc.StaticLayout: """ compute a metadata structure that links basic blocks to the functions in which they're found. @@ -787,7 +852,9 @@ def compute_static_layout(rules: RuleSet, extractor: StaticFeatureExtractor, cap rdoc.FunctionLayout( address=frz.Address.from_capa(f), matched_basic_blocks=tuple( - rdoc.BasicBlockLayout(address=frz.Address.from_capa(bb)) for bb in bbs if bb in matched_bbs + rdoc.BasicBlockLayout(address=frz.Address.from_capa(bb)) + for bb in bbs + if bb in matched_bbs ), # this object is open to extension in the future, # such as with the function name, etc. ) diff --git a/capa/main.py b/capa/main.py index 9e561b10..6f2a8ead 100644 --- a/capa/main.py +++ b/capa/main.py @@ -240,15 +240,29 @@ def install_common_args(parser, wanted=None): # common arguments that all scripts will have # - parser.add_argument("--version", action="version", version="%(prog)s {:s}".format(capa.version.__version__)) parser.add_argument( - "-v", "--verbose", action="store_true", help="enable verbose result document (no effect with --json)" + "--version", + action="version", + version="%(prog)s {:s}".format(capa.version.__version__), ) parser.add_argument( - "-vv", "--vverbose", action="store_true", help="enable very verbose result document (no effect with --json)" + "-v", + "--verbose", + action="store_true", + help="enable verbose result document (no effect with --json)", + ) + parser.add_argument( + "-vv", + "--vverbose", + action="store_true", + help="enable very verbose result document (no effect with --json)", + ) + parser.add_argument( + "-d", "--debug", action="store_true", help="enable debugging output on STDERR" + ) + parser.add_argument( + "-q", "--quiet", action="store_true", help="disable all output but errors" ) - parser.add_argument("-d", "--debug", action="store_true", help="enable debugging output on STDERR") - parser.add_argument("-q", "--quiet", action="store_true", help="disable all output but errors") parser.add_argument( "--color", type=str, @@ -365,7 +379,9 @@ def install_common_args(parser, wanted=None): ) if "tag" in wanted: - parser.add_argument("-t", "--tag", type=str, help="filter on rule meta field values") + parser.add_argument( + "-t", "--tag", type=str, help="filter on rule meta field values" + ) ############################################################################### @@ -427,7 +443,9 @@ def handle_common_args(args): logformat = "[dim]%(name)s[/]: %(message)s" # set markup=True to allow the use of Rich's markup syntax in log messages - rich_handler = RichHandler(markup=True, show_time=False, show_path=True, console=capa.helpers.log_console) + rich_handler = RichHandler( + markup=True, show_time=False, show_path=True, console=capa.helpers.log_console + ) rich_handler.setFormatter(logging.Formatter(logformat)) # use RichHandler for root logger @@ -492,7 +510,9 @@ def handle_common_args(args): # this pulls down just the source code - not the default rules. # i'm not sure the default rules should even be written to the library directory, # so in this case, we require the user to use -r to specify the rule directory. - logger.error("default embedded rules not found! (maybe you installed capa as a library?)") + logger.error( + "default embedded rules not found! (maybe you installed capa as a library?)" + ) logger.error("provide your own rule set via the `-r` option.") raise ShouldExitError(E_MISSING_RULES) @@ -559,7 +579,9 @@ def get_input_format_from_cli(args) -> str: try: return get_auto_format(args.input_file) except PEFormatError as e: - logger.error("Input file '%s' is not a valid PE file: %s", args.input_file, str(e)) + logger.error( + "Input file '%s' is not a valid PE file: %s", args.input_file, str(e) + ) raise ShouldExitError(E_CORRUPT_FILE) from e except UnsupportedFormatError as e: log_unsupported_format_error() @@ -673,11 +695,17 @@ def get_rules_from_cli(args) -> RuleSet: # using the rules cache during development may result in unexpected errors, see #1898 enable_cache = capa.helpers.is_cache_newer_than_rule_code(cache_dir) if not enable_cache: - logger.debug("not using cache. delete the cache file manually to use rule caching again") + logger.debug( + "not using cache. delete the cache file manually to use rule caching again" + ) else: - logger.debug("cache can be used, no potentially outdated cache files found") + logger.debug( + "cache can be used, no potentially outdated cache files found" + ) - rules = capa.rules.get_rules(args.rules, cache_dir=cache_dir, enable_cache=enable_cache) + rules = capa.rules.get_rules( + args.rules, cache_dir=cache_dir, enable_cache=enable_cache + ) except (IOError, capa.rules.InvalidRule, capa.rules.InvalidRuleSet) as e: logger.error("%s", str(e)) logger.error( @@ -730,10 +758,14 @@ def get_file_extractors_from_cli(args, input_format: str) -> list[FeatureExtract try: return capa.loader.get_file_extractors(args.input_file, input_format) except PEFormatError as e: - logger.error("Input file '%s' is not a valid PE file: %s", args.input_file, str(e)) + logger.error( + "Input file '%s' is not a valid PE file: %s", args.input_file, str(e) + ) raise ShouldExitError(E_CORRUPT_FILE) from e except (ELFError, OverflowError) as e: - logger.error("Input file '%s' is not a valid ELF file: %s", args.input_file, str(e)) + logger.error( + "Input file '%s' is not a valid ELF file: %s", args.input_file, str(e) + ) raise ShouldExitError(E_CORRUPT_FILE) from e except UnsupportedFormatError as e: if input_format == FORMAT_CAPE: @@ -757,7 +789,9 @@ def get_file_extractors_from_cli(args, input_format: str) -> list[FeatureExtract raise ShouldExitError(E_INVALID_FILE_TYPE) from e -def find_static_limitations_from_cli(args, rules: RuleSet, file_extractors: list[FeatureExtractor]) -> bool: +def find_static_limitations_from_cli( + args, rules: RuleSet, file_extractors: list[FeatureExtractor] +) -> bool: """ args: args: The parsed command line arguments from `install_common_args`. @@ -773,10 +807,14 @@ def find_static_limitations_from_cli(args, rules: RuleSet, file_extractors: list try: pure_file_capabilities = find_file_capabilities(rules, file_extractor, {}) except PEFormatError as e: - logger.error("Input file '%s' is not a valid PE file: %s", args.input_file, str(e)) + logger.error( + "Input file '%s' is not a valid PE file: %s", args.input_file, str(e) + ) raise ShouldExitError(E_CORRUPT_FILE) from e except (ELFError, OverflowError) as e: - logger.error("Input file '%s' is not a valid ELF file: %s", args.input_file, str(e)) + logger.error( + "Input file '%s' is not a valid ELF file: %s", args.input_file, str(e) + ) raise ShouldExitError(E_CORRUPT_FILE) from e # file limitations that rely on non-file scope won't be detected here. @@ -791,7 +829,9 @@ def find_static_limitations_from_cli(args, rules: RuleSet, file_extractors: list return found_file_limitation -def find_dynamic_limitations_from_cli(args, rules: RuleSet, file_extractors: list[FeatureExtractor]) -> bool: +def find_dynamic_limitations_from_cli( + args, rules: RuleSet, file_extractors: list[FeatureExtractor] +) -> bool: """ Does the dynamic analysis describe some trace that we may not support well? For example, .NET samples detonated in a sandbox, which may rely on different API patterns than we currently describe in our rules. @@ -805,7 +845,9 @@ def find_dynamic_limitations_from_cli(args, rules: RuleSet, file_extractors: lis found_dynamic_limitation = False for file_extractor in file_extractors: pure_dynamic_capabilities = find_file_capabilities(rules, file_extractor, {}) - found_dynamic_limitation |= has_dynamic_limitation(rules, pure_dynamic_capabilities) + found_dynamic_limitation |= has_dynamic_limitation( + rules, pure_dynamic_capabilities + ) if found_dynamic_limitation: # bail if capa encountered file limitation e.g. a dotnet sample is detected @@ -818,11 +860,15 @@ def find_dynamic_limitations_from_cli(args, rules: RuleSet, file_extractors: lis def get_signatures_from_cli(args, input_format: str, backend: str) -> list[Path]: if backend != BACKEND_VIV: - logger.debug("skipping library code matching: only supported by the vivisect backend") + logger.debug( + "skipping library code matching: only supported by the vivisect backend" + ) return [] if input_format != FORMAT_PE: - logger.debug("skipping library code matching: signatures only supports PE files") + logger.debug( + "skipping library code matching: signatures only supports PE files" + ) return [] if args.is_default_signatures: @@ -839,7 +885,9 @@ def get_signatures_from_cli(args, input_format: str, backend: str) -> list[Path] + "Please install the signatures first: " + "https://github.com/mandiant/capa/blob/master/doc/installation.md#method-2-using-capa-as-a-python-library." ) - raise IOError(f"signatures path {args.signatures} does not exist or cannot be accessed") + raise IOError( + f"signatures path {args.signatures} does not exist or cannot be accessed" + ) else: logger.debug("using signatures path: %s", args.signatures) @@ -862,7 +910,13 @@ def get_extractor_from_cli(args, input_format: str, backend: str) -> FeatureExtr """ sig_paths = get_signatures_from_cli(args, input_format, backend) - should_save_workspace = os.environ.get("CAPA_SAVE_WORKSPACE") not in ("0", "no", "NO", "n", None) + should_save_workspace = os.environ.get("CAPA_SAVE_WORKSPACE") not in ( + "0", + "no", + "NO", + "n", + None, + ) os_ = get_os_from_cli(args, backend) sample_path = get_sample_path_from_cli(args, backend) @@ -905,7 +959,9 @@ def get_extractor_from_cli(args, input_format: str, backend: str) -> FeatureExtr def get_extractor_filters_from_cli(args, input_format) -> FilterConfig: - if not hasattr(args, "restrict_to_processes") and not hasattr(args, "restrict_to_functions"): + if not hasattr(args, "restrict_to_processes") and not hasattr( + args, "restrict_to_functions" + ): # no processes or function filters were installed in the args return {} @@ -921,7 +977,9 @@ def get_extractor_filters_from_cli(args, input_format) -> FilterConfig: raise ShouldExitError(E_INVALID_INPUT_FORMAT) -def apply_extractor_filters(extractor: FeatureExtractor, extractor_filters: FilterConfig): +def apply_extractor_filters( + extractor: FeatureExtractor, extractor_filters: FilterConfig +): if not any(extractor_filters.values()): return extractor @@ -971,7 +1029,9 @@ def main(argv: Optional[list[str]] = None): """) parser = argparse.ArgumentParser( - description=desc, epilog=epilog, formatter_class=argparse.RawDescriptionHelpFormatter + description=desc, + epilog=epilog, + formatter_class=argparse.RawDescriptionHelpFormatter, ) install_common_args( parser, @@ -987,7 +1047,9 @@ def main(argv: Optional[list[str]] = None): "restrict-to-processes", }, ) - parser.add_argument("-j", "--json", action="store_true", help="emit JSON instead of text") + parser.add_argument( + "-j", "--json", action="store_true", help="emit JSON instead of text" + ) args = parser.parse_args(args=argv) try: @@ -1000,7 +1062,9 @@ def main(argv: Optional[list[str]] = None): if input_format == FORMAT_RESULT: # render the result document immediately, # no need to load the rules or do other processing. - result_doc = capa.render.result_document.ResultDocument.from_file(args.input_file) + result_doc = capa.render.result_document.ResultDocument.from_file( + args.input_file + ) if args.json: print(result_doc.model_dump_json(exclude_none=True)) @@ -1019,9 +1083,13 @@ def main(argv: Optional[list[str]] = None): file_extractors = get_file_extractors_from_cli(args, input_format) if input_format in STATIC_FORMATS: # only static extractors have file limitations - found_limitation = find_static_limitations_from_cli(args, rules, file_extractors) + found_limitation = find_static_limitations_from_cli( + args, rules, file_extractors + ) if input_format in DYNAMIC_FORMATS: - found_limitation = find_dynamic_limitations_from_cli(args, rules, file_extractors) + found_limitation = find_dynamic_limitations_from_cli( + args, rules, file_extractors + ) backend = get_backend_from_cli(args, input_format) sample_path = get_sample_path_from_cli(args, backend) @@ -1029,16 +1097,22 @@ def main(argv: Optional[list[str]] = None): os_ = "unknown" else: os_ = capa.loader.get_os(sample_path) - extractor: FeatureExtractor = get_extractor_from_cli(args, input_format, backend) + extractor: FeatureExtractor = get_extractor_from_cli( + args, input_format, backend + ) except ShouldExitError as e: return e.status_code - capabilities: Capabilities = find_capabilities(rules, extractor, disable_progress=args.quiet) + capabilities: Capabilities = find_capabilities( + rules, extractor, disable_progress=args.quiet + ) meta: rdoc.Metadata = capa.loader.collect_metadata( argv, args.input_file, input_format, os_, args.rules, extractor, capabilities ) - meta.analysis.layout = capa.loader.compute_layout(rules, extractor, capabilities.matches) + meta.analysis.layout = capa.loader.compute_layout( + rules, extractor, capabilities.matches + ) if found_limitation: # bail if capa's static feature extractor encountered file limitation e.g. a packed binary @@ -1090,7 +1164,9 @@ def ida_main(): meta = capa.ida.helpers.collect_metadata([rules_path]) - capabilities = find_capabilities(rules, capa.features.extractors.ida.extractor.IdaFeatureExtractor()) + capabilities = find_capabilities( + rules, capa.features.extractors.ida.extractor.IdaFeatureExtractor() + ) meta.analysis.feature_counts = capabilities.feature_counts meta.analysis.library_functions = capabilities.library_functions diff --git a/capa/optimizer.py b/capa/optimizer.py index 4aa712ec..694c23a1 100644 --- a/capa/optimizer.py +++ b/capa/optimizer.py @@ -21,7 +21,14 @@ logger = logging.getLogger(__name__) def get_node_cost(node): - if isinstance(node, (capa.features.common.OS, capa.features.common.Arch, capa.features.common.Format)): + if isinstance( + node, + ( + capa.features.common.OS, + capa.features.common.Arch, + capa.features.common.Format, + ), + ): # we assume these are the most restrictive features: # authors commonly use them at the start of rules to restrict the category of samples to inspect return 0 @@ -32,7 +39,14 @@ def get_node_cost(node): # this should be all hash-lookup features. # see below. - elif isinstance(node, (capa.features.common.Substring, capa.features.common.Regex, capa.features.common.Bytes)): + elif isinstance( + node, + ( + capa.features.common.Substring, + capa.features.common.Regex, + capa.features.common.Bytes, + ), + ): # substring and regex features require a full scan of each string # which we anticipate is more expensive then a hash lookup feature (e.g. mnemonic or count). # diff --git a/capa/render/default.py b/capa/render/default.py index 283fb5c4..3c9218b8 100644 --- a/capa/render/default.py +++ b/capa/render/default.py @@ -102,7 +102,9 @@ def find_subrule_matches(doc: rd.ResultDocument): for child in match.children: rec(child) - elif isinstance(match.node, rd.FeatureNode) and isinstance(match.node.feature, frzf.MatchFeature): + elif isinstance(match.node, rd.FeatureNode) and isinstance( + match.node.feature, frzf.MatchFeature + ): matches.add(match.node.feature.match) for rule in rutils.capability_rules(doc): @@ -183,7 +185,9 @@ def render_attack(doc: rd.ResultDocument, console: Console): tactics = collections.defaultdict(set) for rule in rutils.capability_rules(doc): for attack in rule.meta.attack: - tactics[attack.tactic].add((attack.technique, attack.subtechnique, attack.id.strip("[").strip("]"))) + tactics[attack.tactic].add( + (attack.technique, attack.subtechnique, attack.id.strip("[").strip("]")) + ) rows = [] for tactic, techniques in sorted(tactics.items()): @@ -194,7 +198,9 @@ def render_attack(doc: rd.ResultDocument, console: Console): inner_rows.append(f"{bold_markup(technique)} {render_attack_link(id)}") else: # example: Code Discovery::Enumerate PE Sections [T1084.001] - inner_rows.append(f"{bold_markup(technique)}::{subtechnique} {render_attack_link(id)}") + inner_rows.append( + f"{bold_markup(technique)}::{subtechnique} {render_attack_link(id)}" + ) tactic = bold_markup(tactic.upper()) technique = "\n".join(inner_rows) @@ -245,7 +251,9 @@ def render_maec(doc: rd.ResultDocument, console: Console): for category in sorted(maec_categories): values = maec_table.get(category, set()) if values: - rows.append((bold_markup(category.replace("_", "-")), "\n".join(sorted(values)))) + rows.append( + (bold_markup(category.replace("_", "-")), "\n".join(sorted(values))) + ) if rows: table = rich.table.Table(min_width=100) @@ -264,7 +272,9 @@ def render_mbc_link(id: str, objective: str, behavior: str) -> str: base_url = "https://github.com/MBCProject/mbc-markdown/blob/main" elif id[0] == "C": # micro-behavior - base_url = "https://github.com/MBCProject/mbc-markdown/blob/main/micro-behaviors" + base_url = ( + "https://github.com/MBCProject/mbc-markdown/blob/main/micro-behaviors" + ) else: raise ValueError("unexpected MBC prefix") @@ -292,7 +302,9 @@ def render_mbc(doc: rd.ResultDocument, console: Console): objectives = collections.defaultdict(set) for rule in rutils.capability_rules(doc): for mbc in rule.meta.mbc: - objectives[mbc.objective].add((mbc.behavior, mbc.method, mbc.id.strip("[").strip("]"))) + objectives[mbc.objective].add( + (mbc.behavior, mbc.method, mbc.id.strip("[").strip("]")) + ) rows = [] for objective, behaviors in sorted(objectives.items()): @@ -300,7 +312,9 @@ def render_mbc(doc: rd.ResultDocument, console: Console): for technique, subtechnique, id in sorted(behaviors): if not subtechnique: # example: File and Directory Discovery [T1083] - inner_rows.append(f"{bold_markup(technique)} {render_mbc_link(id, objective, technique)}") + inner_rows.append( + f"{bold_markup(technique)} {render_mbc_link(id, objective, technique)}" + ) else: # example: Code Discovery::Enumerate PE Sections [T1084.001] inner_rows.append( diff --git a/capa/render/vverbose.py b/capa/render/vverbose.py index 026d3ef4..7a16e97f 100644 --- a/capa/render/vverbose.py +++ b/capa/render/vverbose.py @@ -55,7 +55,11 @@ def hanging_indent(s: str, indent: int) -> str: def render_locations( - console: Console, layout: rd.Layout, locations: Iterable[frz.Address], indent: int, use_short_format: bool = False + console: Console, + layout: rd.Layout, + locations: Iterable[frz.Address], + indent: int, + use_short_format: bool = False, ): """ Render the given locations, such as virtual address or pid/tid/callid with process name. @@ -117,7 +121,13 @@ def render_locations( raise RuntimeError("unreachable") -def render_statement(console: Console, layout: rd.Layout, match: rd.Match, statement: rd.Statement, indent: int): +def render_statement( + console: Console, + layout: rd.Layout, + match: rd.Match, + statement: rd.Statement, + indent: int, +): console.write(" " * indent) if isinstance(statement, rd.SubscopeStatement): @@ -191,7 +201,12 @@ def render_string_value(s: str) -> str: def render_feature( - console: Console, layout: rd.Layout, rule: rd.RuleMatches, match: rd.Match, feature: frzf.Feature, indent: int + console: Console, + layout: rd.Layout, + rule: rd.RuleMatches, + match: rd.Match, + feature: frzf.Feature, + indent: int, ): console.write(" " * indent) @@ -220,7 +235,13 @@ def render_feature( value = render_string_value(value) elif isinstance( - feature, (frzf.NumberFeature, frzf.OffsetFeature, frzf.OperandNumberFeature, frzf.OperandOffsetFeature) + feature, + ( + frzf.NumberFeature, + frzf.OffsetFeature, + frzf.OperandNumberFeature, + frzf.OperandOffsetFeature, + ), ): assert isinstance(value, int) value = capa.helpers.hex(value) @@ -246,15 +267,22 @@ def render_feature( if isinstance(feature, (frzf.OSFeature, frzf.ArchFeature, frzf.FormatFeature)): # don't show the location of these global features pass - elif isinstance(layout, rd.DynamicLayout) and rule.meta.scopes.dynamic == capa.rules.Scope.CALL: + elif ( + isinstance(layout, rd.DynamicLayout) + and rule.meta.scopes.dynamic == capa.rules.Scope.CALL + ): # if we're in call scope, then the call will have been rendered at the top # of the output, so don't re-render it again for each feature. pass - elif isinstance(layout, rd.DynamicLayout) and isinstance(feature, frzf.MatchFeature): + elif isinstance(layout, rd.DynamicLayout) and isinstance( + feature, frzf.MatchFeature + ): # don't render copies of the span of calls address for submatches pass else: - render_locations(console, layout, match.locations, indent, use_short_format=True) + render_locations( + console, layout, match.locations, indent, use_short_format=True + ) console.writeln() else: # like: @@ -267,15 +295,27 @@ def render_feature( console.write(" " * (indent + 1)) console.write("- ") console.write(rutils.bold2(render_string_value(capture))) - if isinstance(layout, rd.DynamicLayout) and rule.meta.scopes.dynamic == capa.rules.Scope.CALL: + if ( + isinstance(layout, rd.DynamicLayout) + and rule.meta.scopes.dynamic == capa.rules.Scope.CALL + ): # like above, don't re-render calls when in call scope. pass else: - render_locations(console, layout, locations, indent=indent + 1, use_short_format=True) + render_locations( + console, layout, locations, indent=indent + 1, use_short_format=True + ) console.writeln() -def render_node(console: Console, layout: rd.Layout, rule: rd.RuleMatches, match: rd.Match, node: rd.Node, indent: int): +def render_node( + console: Console, + layout: rd.Layout, + rule: rd.RuleMatches, + match: rd.Match, + node: rd.Node, + indent: int, +): if isinstance(node, rd.StatementNode): render_statement(console, layout, match, node.statement, indent=indent) elif isinstance(node, rd.FeatureNode): @@ -293,7 +333,12 @@ MODE_FAILURE = "failure" def render_match( - console: Console, layout: rd.Layout, rule: rd.RuleMatches, match: rd.Match, indent=0, mode=MODE_SUCCESS + console: Console, + layout: rd.Layout, + rule: rd.RuleMatches, + match: rd.Match, + indent=0, + mode=MODE_SUCCESS, ): child_mode = mode if mode == MODE_SUCCESS: @@ -302,12 +347,18 @@ def render_match( return # optional statement with no successful children is empty - if isinstance(match.node, rd.StatementNode) and match.node.statement.type == rd.CompoundStatementType.OPTIONAL: + if ( + isinstance(match.node, rd.StatementNode) + and match.node.statement.type == rd.CompoundStatementType.OPTIONAL + ): if not any(m.success for m in match.children): return # not statement, so invert the child mode to show failed evaluations - if isinstance(match.node, rd.StatementNode) and match.node.statement.type == rd.CompoundStatementType.NOT: + if ( + isinstance(match.node, rd.StatementNode) + and match.node.statement.type == rd.CompoundStatementType.NOT + ): child_mode = MODE_FAILURE elif mode == MODE_FAILURE: @@ -316,12 +367,18 @@ def render_match( return # optional statement with successful children is not relevant - if isinstance(match.node, rd.StatementNode) and match.node.statement.type == rd.CompoundStatementType.OPTIONAL: + if ( + isinstance(match.node, rd.StatementNode) + and match.node.statement.type == rd.CompoundStatementType.OPTIONAL + ): if any(m.success for m in match.children): return # not statement, so invert the child mode to show successful evaluations - if isinstance(match.node, rd.StatementNode) and match.node.statement.type == rd.CompoundStatementType.NOT: + if ( + isinstance(match.node, rd.StatementNode) + and match.node.statement.type == rd.CompoundStatementType.NOT + ): child_mode = MODE_SUCCESS else: raise RuntimeError("unexpected mode: " + mode) @@ -355,7 +412,10 @@ def collect_span_of_calls_locations( yield location child_mode = mode - if isinstance(match.node, rd.StatementNode) and match.node.statement.type == rd.CompoundStatementType.NOT: + if ( + isinstance(match.node, rd.StatementNode) + and match.node.statement.type == rd.CompoundStatementType.NOT + ): child_mode = MODE_FAILURE if mode == MODE_SUCCESS else MODE_SUCCESS for child in match.children: @@ -381,7 +441,9 @@ def render_rules(console: Console, doc: rd.ResultDocument): """ import capa.render.verbose as v - functions_by_bb: dict[capa.features.address.Address, capa.features.address.Address] = {} + functions_by_bb: dict[ + capa.features.address.Address, capa.features.address.Address + ] = {} if isinstance(doc.meta.analysis, rd.StaticAnalysis): for finfo in doc.meta.analysis.layout.functions: faddress = finfo.address.to_capa() @@ -396,7 +458,9 @@ def render_rules(console: Console, doc: rd.ResultDocument): had_match = False - for _, _, rule in sorted((rule.meta.namespace or "", rule.meta.name, rule) for rule in doc.rules.values()): + for _, _, rule in sorted( + (rule.meta.namespace or "", rule.meta.name, rule) for rule in doc.rules.values() + ): # default scope hides things like lib rules, malware-category rules, etc. # but in vverbose mode, we really want to show everything. # @@ -413,7 +477,9 @@ def render_rules(console: Console, doc: rd.ResultDocument): else: if rule.meta.lib: lib_info = ", only showing first match of library rule" - capability = Text.assemble(rutils.bold(rule.meta.name), f" ({count} matches{lib_info})") + capability = Text.assemble( + rutils.bold(rule.meta.name), f" ({count} matches{lib_info})" + ) console.writeln(capability) had_match = True @@ -424,19 +490,25 @@ def render_rules(console: Console, doc: rd.ResultDocument): rows.append(("namespace", rule.meta.namespace)) if rule.meta.maec.analysis_conclusion or rule.meta.maec.analysis_conclusion_ov: - rows.append(( - "maec/analysis-conclusion", - rule.meta.maec.analysis_conclusion or rule.meta.maec.analysis_conclusion_ov, - )) + rows.append( + ( + "maec/analysis-conclusion", + rule.meta.maec.analysis_conclusion + or rule.meta.maec.analysis_conclusion_ov, + ) + ) if rule.meta.maec.malware_family: rows.append(("maec/malware-family", rule.meta.maec.malware_family)) if rule.meta.maec.malware_category or rule.meta.maec.malware_category_ov: - rows.append(( - "maec/malware-category", - rule.meta.maec.malware_category or rule.meta.maec.malware_category_ov, - )) + rows.append( + ( + "maec/malware-category", + rule.meta.maec.malware_category + or rule.meta.maec.malware_category_ov, + ) + ) rows.append(("author", ", ".join(rule.meta.authors))) @@ -449,10 +521,17 @@ def render_rules(console: Console, doc: rd.ResultDocument): rows.append(("scope", rule.meta.scopes.dynamic.value)) if rule.meta.attack: - rows.append(("att&ck", ", ".join([rutils.format_parts_id(v) for v in rule.meta.attack]))) + rows.append( + ( + "att&ck", + ", ".join([rutils.format_parts_id(v) for v in rule.meta.attack]), + ) + ) if rule.meta.mbc: - rows.append(("mbc", ", ".join([rutils.format_parts_id(v) for v in rule.meta.mbc]))) + rows.append( + ("mbc", ", ".join([rutils.format_parts_id(v) for v in rule.meta.mbc])) + ) if rule.meta.references: rows.append(("references", ", ".join(rule.meta.references))) @@ -487,8 +566,12 @@ def render_rules(console: Console, doc: rd.ResultDocument): console.write(capa.render.verbose.format_address(location)) if rule.meta.scopes.static == capa.rules.Scope.BASIC_BLOCK: - func = frz.Address.from_capa(functions_by_bb[location.to_capa()]) - console.write(f" in function {capa.render.verbose.format_address(func)}") + func = frz.Address.from_capa( + functions_by_bb[location.to_capa()] + ) + console.write( + f" in function {capa.render.verbose.format_address(func)}" + ) elif doc.meta.flavor == rd.Flavor.DYNAMIC: assert rule.meta.scopes.dynamic is not None @@ -497,14 +580,28 @@ def render_rules(console: Console, doc: rd.ResultDocument): console.write(rule.meta.scopes.dynamic.value + " @ ") if rule.meta.scopes.dynamic == capa.rules.Scope.PROCESS: - console.write(v.render_process(doc.meta.analysis.layout, location)) + console.write( + v.render_process(doc.meta.analysis.layout, location) + ) elif rule.meta.scopes.dynamic == capa.rules.Scope.THREAD: - console.write(v.render_thread(doc.meta.analysis.layout, location)) + console.write( + v.render_thread(doc.meta.analysis.layout, location) + ) elif rule.meta.scopes.dynamic == capa.rules.Scope.SPAN_OF_CALLS: calls = sorted(set(collect_span_of_calls_locations(match))) - console.write(hanging_indent(v.render_span_of_calls(doc.meta.analysis.layout, calls), indent=1)) + console.write( + hanging_indent( + v.render_span_of_calls(doc.meta.analysis.layout, calls), + indent=1, + ) + ) elif rule.meta.scopes.dynamic == capa.rules.Scope.CALL: - console.write(hanging_indent(v.render_call(doc.meta.analysis.layout, location), indent=1)) + console.write( + hanging_indent( + v.render_call(doc.meta.analysis.layout, location), + indent=1, + ) + ) else: capa.helpers.assert_never(rule.meta.scopes.dynamic) diff --git a/capa/rules/__init__.py b/capa/rules/__init__.py index cd148265..81189fe6 100644 --- a/capa/rules/__init__.py +++ b/capa/rules/__init__.py @@ -145,7 +145,9 @@ class Scopes: elif self.dynamic: return f"dynamic-scope: {self.dynamic}" else: - raise ValueError("invalid rules class. at least one scope must be specified") + raise ValueError( + "invalid rules class. at least one scope must be specified" + ) @classmethod def from_dict(self, scopes: dict[str, str]) -> "Scopes": @@ -169,7 +171,9 @@ class Scopes: scopes_["dynamic"] = None if (not scopes_["static"]) and (not scopes_["dynamic"]): - raise InvalidRule("invalid scopes value. At least one scope must be specified") + raise InvalidRule( + "invalid scopes value. At least one scope must be specified" + ) # check that all the specified scopes are valid if scopes_["static"] and scopes_["static"] not in STATIC_SCOPES: @@ -363,8 +367,12 @@ def translate_com_feature(com_name: str, com_type: ComType) -> ceng.Statement: guid_bytes = bytes.fromhex("".join(reordered_hex_pairs)) prefix = capa.features.com.COM_PREFIXES[com_type] symbol = prefix + com_name - com_features.append(capa.features.common.String(guid, f"{symbol} as GUID string")) - com_features.append(capa.features.common.Bytes(guid_bytes, f"{symbol} as bytes")) + com_features.append( + capa.features.common.String(guid, f"{symbol} as GUID string") + ) + com_features.append( + capa.features.common.Bytes(guid_bytes, f"{symbol} as bytes") + ) return ceng.Or(com_features) @@ -468,7 +476,9 @@ def parse_bytes(s: str) -> bytes: try: b = bytes.fromhex(s.replace(" ", "")) except binascii.Error: - raise InvalidRule(f'unexpected bytes value: must be a valid hex sequence: "{s}"') + raise InvalidRule( + f'unexpected bytes value: must be a valid hex sequence: "{s}"' + ) if len(b) > MAX_BYTES_FEATURE_SIZE: raise InvalidRule( @@ -502,7 +512,9 @@ def parse_description(s: Union[str, int, bytes], value_type: str, description=No if description == "": # sanity check: # there is an empty description, like `number: 10 =` - raise InvalidRule(f'unexpected value: "{s}", description cannot be empty') + raise InvalidRule( + f'unexpected value: "{s}", description cannot be empty' + ) else: # this is a string, but there is no description, # like: `api: CreateFileA` @@ -523,13 +535,18 @@ def parse_description(s: Union[str, int, bytes], value_type: str, description=No or value_type.startswith(("number/", "offset/")) or ( value_type.startswith("operand[") - and (value_type.endswith("].number") or value_type.endswith("].offset")) + and ( + value_type.endswith("].number") + or value_type.endswith("].offset") + ) ) ): try: value = parse_int(value) except ValueError: - raise InvalidRule(f'unexpected value: "{value}", must begin with numerical value') + raise InvalidRule( + f'unexpected value: "{value}", must begin with numerical value' + ) else: # the value might be a number, like: `number: 10` @@ -560,7 +577,9 @@ def pop_statement_description_entry(d): return None # identify child of form '{ "description": }' - descriptions = list(filter(lambda c: isinstance(c, dict) and len(c) == 1 and "description" in c, d)) + descriptions = list( + filter(lambda c: isinstance(c, dict) and len(c) == 1 and "description" in c, d) + ) if len(descriptions) > 1: raise InvalidRule("statements can only have one description") @@ -624,7 +643,9 @@ def is_subscope_compatible(scope: Scope | None, subscope: Scope) -> bool: elif subscope in DYNAMIC_SCOPE_ORDER: try: - return DYNAMIC_SCOPE_ORDER.index(subscope) >= DYNAMIC_SCOPE_ORDER.index(scope) + return DYNAMIC_SCOPE_ORDER.index(subscope) >= DYNAMIC_SCOPE_ORDER.index( + scope + ) except ValueError: return False @@ -639,21 +660,35 @@ def build_statements(d, scopes: Scopes): key = list(d.keys())[0] description = pop_statement_description_entry(d[key]) if key == "and": - return ceng.And(unique(build_statements(dd, scopes) for dd in d[key]), description=description) + return ceng.And( + unique(build_statements(dd, scopes) for dd in d[key]), + description=description, + ) elif key == "or": - return ceng.Or(unique(build_statements(dd, scopes) for dd in d[key]), description=description) + return ceng.Or( + unique(build_statements(dd, scopes) for dd in d[key]), + description=description, + ) elif key == "not": if len(d[key]) != 1: raise InvalidRule("not statement must have exactly one child statement") return ceng.Not(build_statements(d[key][0], scopes), description=description) elif key.endswith(" or more"): count = int(key[: -len("or more")]) - return ceng.Some(count, unique(build_statements(dd, scopes) for dd in d[key]), description=description) + return ceng.Some( + count, + unique(build_statements(dd, scopes) for dd in d[key]), + description=description, + ) elif key == "optional": # `optional` is an alias for `0 or more` # which is useful for documenting behaviors, # like with `write file`, we might say that `WriteFile` is optionally found alongside `CreateFileA`. - return ceng.Some(0, unique(build_statements(dd, scopes) for dd in d[key]), description=description) + return ceng.Some( + 0, + unique(build_statements(dd, scopes) for dd in d[key]), + description=description, + ) elif key == "process": if not is_subscope_compatible(scopes.dynamic, Scope.PROCESS): @@ -663,23 +698,31 @@ def build_statements(d, scopes: Scopes): raise InvalidRule("subscope must have exactly one child statement") return ceng.Subscope( - Scope.PROCESS, build_statements(d[key][0], Scopes(dynamic=Scope.PROCESS)), description=description + Scope.PROCESS, + build_statements(d[key][0], Scopes(dynamic=Scope.PROCESS)), + description=description, ) elif key == "thread": if not is_subscope_compatible(scopes.dynamic, Scope.THREAD): - raise InvalidRule("`thread` subscope supported only for the `process` scope") + raise InvalidRule( + "`thread` subscope supported only for the `process` scope" + ) if len(d[key]) != 1: raise InvalidRule("subscope must have exactly one child statement") return ceng.Subscope( - Scope.THREAD, build_statements(d[key][0], Scopes(dynamic=Scope.THREAD)), description=description + Scope.THREAD, + build_statements(d[key][0], Scopes(dynamic=Scope.THREAD)), + description=description, ) elif key == "span of calls": if not is_subscope_compatible(scopes.dynamic, Scope.SPAN_OF_CALLS): - raise InvalidRule("`span of calls` subscope supported only for the `process` and `thread` scopes") + raise InvalidRule( + "`span of calls` subscope supported only for the `process` and `thread` scopes" + ) if len(d[key]) != 1: raise InvalidRule("subscope must have exactly one child statement") @@ -692,13 +735,17 @@ def build_statements(d, scopes: Scopes): elif key == "call": if not is_subscope_compatible(scopes.dynamic, Scope.CALL): - raise InvalidRule("`call` subscope supported only for the `process`, `thread`, and `call` scopes") + raise InvalidRule( + "`call` subscope supported only for the `process`, `thread`, and `call` scopes" + ) if len(d[key]) != 1: raise InvalidRule("subscope must have exactly one child statement") return ceng.Subscope( - Scope.CALL, build_statements(d[key][0], Scopes(dynamic=Scope.CALL)), description=description + Scope.CALL, + build_statements(d[key][0], Scopes(dynamic=Scope.CALL)), + description=description, ) elif key == "function": @@ -709,23 +756,31 @@ def build_statements(d, scopes: Scopes): raise InvalidRule("subscope must have exactly one child statement") return ceng.Subscope( - Scope.FUNCTION, build_statements(d[key][0], Scopes(static=Scope.FUNCTION)), description=description + Scope.FUNCTION, + build_statements(d[key][0], Scopes(static=Scope.FUNCTION)), + description=description, ) elif key == "basic block": if not is_subscope_compatible(scopes.static, Scope.BASIC_BLOCK): - raise InvalidRule("`basic block` subscope supported only for `function` scope") + raise InvalidRule( + "`basic block` subscope supported only for `function` scope" + ) if len(d[key]) != 1: raise InvalidRule("subscope must have exactly one child statement") return ceng.Subscope( - Scope.BASIC_BLOCK, build_statements(d[key][0], Scopes(static=Scope.BASIC_BLOCK)), description=description + Scope.BASIC_BLOCK, + build_statements(d[key][0], Scopes(static=Scope.BASIC_BLOCK)), + description=description, ) elif key == "instruction": if not is_subscope_compatible(scopes.static, Scope.INSTRUCTION): - raise InvalidRule("`instruction` subscope supported only for `function` and `basic block` scope") + raise InvalidRule( + "`instruction` subscope supported only for `function` and `basic block` scope" + ) if len(d[key]) == 1: statements = build_statements(d[key][0], Scopes(static=Scope.INSTRUCTION)) @@ -742,7 +797,12 @@ def build_statements(d, scopes: Scopes): # - arch: i386 # - mnemonic: cmp # - statements = ceng.And(unique(build_statements(dd, Scopes(static=Scope.INSTRUCTION)) for dd in d[key])) + statements = ceng.And( + unique( + build_statements(dd, Scopes(static=Scope.INSTRUCTION)) + for dd in d[key] + ) + ) return ceng.Subscope(Scope.INSTRUCTION, statements, description=description) @@ -806,7 +866,9 @@ def build_statements(d, scopes: Scopes): else: raise InvalidRule(f"unexpected range: {count}") elif key == "string" and not isinstance(d[key], str): - raise InvalidRule(f"ambiguous string value {d[key]}, must be defined as explicit string") + raise InvalidRule( + f"ambiguous string value {d[key]}, must be defined as explicit string" + ) elif key.startswith("operand[") and key.endswith("].number"): index = key[len("operand[") : -len("].number")] @@ -818,7 +880,9 @@ def build_statements(d, scopes: Scopes): value, description = parse_description(d[key], key, d.get("description")) assert isinstance(value, int) try: - feature = capa.features.insn.OperandNumber(index, value, description=description) + feature = capa.features.insn.OperandNumber( + index, value, description=description + ) except ValueError as e: raise InvalidRule(str(e)) from e ensure_feature_valid_for_scopes(scopes, feature) @@ -834,7 +898,9 @@ def build_statements(d, scopes: Scopes): value, description = parse_description(d[key], key, d.get("description")) assert isinstance(value, int) try: - feature = capa.features.insn.OperandOffset(index, value, description=description) + feature = capa.features.insn.OperandOffset( + index, value, description=description + ) except ValueError as e: raise InvalidRule(str(e)) from e ensure_feature_valid_for_scopes(scopes, feature) @@ -854,7 +920,9 @@ def build_statements(d, scopes: Scopes): value, description = parse_description(d[key], key, d.get("description")) try: - feature = capa.features.insn.Property(value, access=access, description=description) + feature = capa.features.insn.Property( + value, access=access, description=description + ) except ValueError as e: raise InvalidRule(str(e)) from e ensure_feature_valid_for_scopes(scopes, feature) @@ -893,7 +961,9 @@ def second(s: list[Any]) -> Any: class Rule: - def __init__(self, name: str, scopes: Scopes, statement: Statement, meta, definition=""): + def __init__( + self, name: str, scopes: Scopes, statement: Statement, meta, definition="" + ): super().__init__() self.name = name self.scopes = scopes @@ -1069,13 +1139,19 @@ class Rule: # each rule has two scopes, a static-flavor scope, and a # dynamic-flavor one. which one is used depends on the analysis type. if "scope" in meta: - raise InvalidRule(f"legacy rule detected (rule.meta.scope), please update to the new syntax: {name}") + raise InvalidRule( + f"legacy rule detected (rule.meta.scope), please update to the new syntax: {name}" + ) elif "scopes" in meta: scopes_ = meta.get("scopes") else: - raise InvalidRule("please specify at least one of this rule's (static/dynamic) scopes") + raise InvalidRule( + "please specify at least one of this rule's (static/dynamic) scopes" + ) if not isinstance(scopes_, dict): - raise InvalidRule("the scopes field must contain a dictionary specifying the scopes") + raise InvalidRule( + "the scopes field must contain a dictionary specifying the scopes" + ) scopes: Scopes = Scopes.from_dict(scopes_) statements = d["rule"]["features"] @@ -1094,7 +1170,9 @@ class Rule: if not isinstance(meta.get("mbc", []), list): raise InvalidRule("MBC mapping must be a list") - return cls(name, scopes, build_statements(statements[0], scopes), meta, definition) + return cls( + name, scopes, build_statements(statements[0], scopes), meta, definition + ) @staticmethod @lru_cache() @@ -1106,7 +1184,9 @@ class Rule: logger.debug("using libyaml CSafeLoader.") return yaml.CSafeLoader except Exception: - logger.debug("unable to import libyaml CSafeLoader, falling back to Python yaml parser.") + logger.debug( + "unable to import libyaml CSafeLoader, falling back to Python yaml parser." + ) logger.debug("this will be slower to load rules.") return yaml.SafeLoader @@ -1261,7 +1341,9 @@ class Rule: # only do this for the features section, so the meta description doesn't get reformatted # assumes features section always exists features_offset = doc.find("features") - doc = doc[:features_offset] + doc[features_offset:].replace(" description:", " description:") + doc = doc[:features_offset] + doc[features_offset:].replace( + " description:", " description:" + ) # for negative hex numbers, yaml dump outputs: # - offset: !!int '0x-30' @@ -1448,19 +1530,25 @@ class RuleSet: self.rules = {rule.name: rule for rule in rules} self.rules_by_namespace = index_rules_by_namespace(rules) - self.rules_by_scope = {scope: self._get_rules_for_scope(rules, scope) for scope in scopes} + self.rules_by_scope = { + scope: self._get_rules_for_scope(rules, scope) for scope in scopes + } # these structures are unstable and may change before the next major release. scores_by_rule: dict[str, int] = {} self._feature_indexes_by_scopes = { - scope: self._index_rules_by_feature(scope, self.rules_by_scope[scope], scores_by_rule) for scope in scopes + scope: self._index_rules_by_feature( + scope, self.rules_by_scope[scope], scores_by_rule + ) + for scope in scopes } # Pre-compute the topological index mapping for each scope. # This avoids rebuilding the dict on every call to _match (which runs once per # instruction/basic-block/function/file scope, i.e. potentially millions of times). self._rule_index_by_scope: dict[Scope, dict[str, int]] = { - scope: {rule.name: i for i, rule in enumerate(self.rules_by_scope[scope])} for scope in scopes + scope: {rule.name: i for i, rule in enumerate(self.rules_by_scope[scope])} + for scope in scopes } @property @@ -1506,7 +1594,9 @@ class RuleSet: # this routine is unstable and may change before the next major release. @staticmethod - def _score_feature(scores_by_rule: dict[str, int], node: capa.features.common.Feature) -> int: + def _score_feature( + scores_by_rule: dict[str, int], node: capa.features.common.Feature + ) -> int: """ Score the given feature by how "uncommon" we think it will be. Features that we expect to be very selective (ie. uniquely identify a rule and be required to match), @@ -1561,7 +1651,9 @@ class RuleSet: return scores_by_rule[rule_name] - elif isinstance(node, (capa.features.insn.Number, capa.features.insn.OperandNumber)): + elif isinstance( + node, (capa.features.insn.Number, capa.features.insn.OperandNumber) + ): v = node.value assert isinstance(v, int) @@ -1581,7 +1673,14 @@ class RuleSet: # Other numbers are assumed to be uncommon. return 7 - elif isinstance(node, (capa.features.common.Substring, capa.features.common.Regex, capa.features.common.Bytes)): + elif isinstance( + node, + ( + capa.features.common.Substring, + capa.features.common.Regex, + capa.features.common.Bytes, + ), + ): # Scanning features (non-hashable), which we can't use for quick matching/filtering. return 0 @@ -1661,7 +1760,9 @@ class RuleSet: # this routine is unstable and may change before the next major release. @staticmethod - def _index_rules_by_feature(scope: Scope, rules: list[Rule], scores_by_rule: dict[str, int]) -> _RuleFeatureIndex: + def _index_rules_by_feature( + scope: Scope, rules: list[Rule], scores_by_rule: dict[str, int] + ) -> _RuleFeatureIndex: """ Index the given rules by their minimal set of most "uncommon" features required to match. @@ -1810,7 +1911,9 @@ class RuleSet: # Ideally we find a way to get rid of all of these, eventually. string_rules: dict[str, list[Feature]] = {} bytes_rules_count = 0 - bytes_prefix_index: dict[int, list[tuple[str, bytes]]] = collections.defaultdict(list) + bytes_prefix_index: dict[int, list[tuple[str, bytes]]] = ( + collections.defaultdict(list) + ) for rule in rules: rule_name = rule.meta["name"] @@ -1823,26 +1926,45 @@ class RuleSet: string_features = [ feature for feature in features - if isinstance(feature, (capa.features.common.Substring, capa.features.common.Regex)) + if isinstance( + feature, + (capa.features.common.Substring, capa.features.common.Regex), + ) ] hashable_features = [ feature for feature in features if not isinstance( - feature, (capa.features.common.Substring, capa.features.common.Regex, capa.features.common.Bytes) + feature, + ( + capa.features.common.Substring, + capa.features.common.Regex, + capa.features.common.Bytes, + ), ) ] - logger.debug("indexing: features: %d, score: %d, rule: %s", len(features), score, rule_name) + logger.debug( + "indexing: features: %d, score: %d, rule: %s", + len(features), + score, + rule_name, + ) scores_by_rule[rule_name] = score for feature in features: - logger.debug(" : [%d] %s", RuleSet._score_feature(scores_by_rule, feature), feature) + logger.debug( + " : [%d] %s", + RuleSet._score_feature(scores_by_rule, feature), + feature, + ) if string_features: string_rules[rule_name] = cast(list[Feature], string_features) bytes_features: list[capa.features.common.Bytes] = [ - feature for feature in features if isinstance(feature, capa.features.common.Bytes) + feature + for feature in features + if isinstance(feature, capa.features.common.Bytes) ] if bytes_features: bytes_rules_count += 1 @@ -1857,15 +1979,27 @@ class RuleSet: for feature in hashable_features: rules_by_feature[feature].add(rule_name) - logger.debug("indexing: %d features indexed for scope %s", len(rules_by_feature), scope) + logger.debug( + "indexing: %d features indexed for scope %s", len(rules_by_feature), scope + ) logger.debug( "indexing: %d indexed features are shared by more than 3 rules", - len([feature for feature, rules in rules_by_feature.items() if len(rules) > 3]), + len( + [ + feature + for feature, rules in rules_by_feature.items() + if len(rules) > 3 + ] + ), ) logger.debug( - "indexing: %d scanning string features, %d scanning bytes features", len(string_rules), bytes_rules_count + "indexing: %d scanning string features, %d scanning bytes features", + len(string_rules), + bytes_rules_count, + ) + return RuleSet._RuleFeatureIndex( + rules_by_feature, string_rules, dict(bytes_prefix_index) ) - return RuleSet._RuleFeatureIndex(rules_by_feature, string_rules, dict(bytes_prefix_index)) @staticmethod def _get_rules_for_scope(rules, scope) -> list[Rule]: @@ -1926,20 +2060,36 @@ class RuleSet: for rule in rules: for k, v in rule.meta.items(): if isinstance(v, str) and tag in v: - logger.debug('using rule "%s" and dependencies, found tag in meta.%s: %s', rule.name, k, v) - rules_filtered.update(set(get_rules_and_dependencies(rules, rule.name))) + logger.debug( + 'using rule "%s" and dependencies, found tag in meta.%s: %s', + rule.name, + k, + v, + ) + rules_filtered.update( + set(get_rules_and_dependencies(rules, rule.name)) + ) break if isinstance(v, list): for vv in v: if tag in vv: - logger.debug('using rule "%s" and dependencies, found tag in meta.%s: %s', rule.name, k, vv) - rules_filtered.update(set(get_rules_and_dependencies(rules, rule.name))) + logger.debug( + 'using rule "%s" and dependencies, found tag in meta.%s: %s', + rule.name, + k, + vv, + ) + rules_filtered.update( + set(get_rules_and_dependencies(rules, rule.name)) + ) break return RuleSet(list(rules_filtered)) # this routine is unstable and may change before the next major release. @staticmethod - def _sort_rules_by_index(rule_index_by_rule_name: dict[str, int], rules: list[Rule]): + def _sort_rules_by_index( + rule_index_by_rule_name: dict[str, int], rules: list[Rule] + ): """ Sort (in place) the given rules by their index provided by the given dict. This mapping is intended to represent the topologic index of the given rule; @@ -1948,7 +2098,9 @@ class RuleSet: """ rules.sort(key=lambda r: rule_index_by_rule_name[r.name]) - def _match(self, scope: Scope, features: FeatureSet, addr: Address) -> tuple[FeatureSet, ceng.MatchResults]: + def _match( + self, scope: Scope, features: FeatureSet, addr: Address + ) -> tuple[FeatureSet, ceng.MatchResults]: """ Match rules from this ruleset at the given scope against the given features. @@ -1956,7 +2108,9 @@ class RuleSet: It uses its knowledge of all the rules to evaluate a minimal set of candidate rules for the given features. """ - feature_index: RuleSet._RuleFeatureIndex = self._feature_indexes_by_scopes[scope] + feature_index: RuleSet._RuleFeatureIndex = self._feature_indexes_by_scopes[ + scope + ] # Topologic location of rule given its name. # That is, rules with a lower index should be evaluated first, since their dependencies # will be evaluated later. @@ -2056,7 +2210,9 @@ class RuleSet: for feature in bytes_features: if len(feature.value) >= _BYTES_PREFIX_SIZE: prefix = struct.unpack_from(">I", feature.value)[0] - for rule_name, pattern in feature_index.bytes_prefix_index.get(prefix, ()): + for rule_name, pattern in feature_index.bytes_prefix_index.get( + prefix, () + ): if feature.value.startswith(pattern): candidate_rule_names.add(rule_name) @@ -2103,7 +2259,9 @@ class RuleSet: # such as rule or namespace matches. if augmented_features is features: # lazily create the copy of features only when a rule matches, since it could be expensive. - augmented_features = collections.defaultdict(set, copy.copy(features)) + augmented_features = collections.defaultdict( + set, copy.copy(features) + ) ceng.index_rule_matches(augmented_features, rule, [addr]) @@ -2118,12 +2276,18 @@ class RuleSet: if new_features: new_candidates: list[str] = [] for new_feature in new_features: - new_candidates.extend(feature_index.rules_by_feature.get(new_feature, ())) + new_candidates.extend( + feature_index.rules_by_feature.get(new_feature, ()) + ) if new_candidates: candidate_rule_names.update(new_candidates) - candidate_rules.extend([self.rules[rule_name] for rule_name in new_candidates]) - RuleSet._sort_rules_by_index(rule_index_by_rule_name, candidate_rules) + candidate_rules.extend( + [self.rules[rule_name] for rule_name in new_candidates] + ) + RuleSet._sort_rules_by_index( + rule_index_by_rule_name, candidate_rules + ) candidate_rules.reverse() return (augmented_features, results) @@ -2153,17 +2317,25 @@ class RuleSet: if paranoid: rules: list[Rule] = self.rules_by_scope[scope] - paranoid_features, paranoid_matches = capa.engine.match(rules, features, addr) + paranoid_features, paranoid_matches = capa.engine.match( + rules, features, addr + ) if features != paranoid_features: logger.warning("paranoid: %s: %s", scope, addr) - for feature in sorted(set(features.keys()) & set(paranoid_features.keys())): + for feature in sorted( + set(features.keys()) & set(paranoid_features.keys()) + ): logger.warning("paranoid: %s", feature) - for feature in sorted(set(features.keys()) - set(paranoid_features.keys())): + for feature in sorted( + set(features.keys()) - set(paranoid_features.keys()) + ): logger.warning("paranoid: + %s", feature) - for feature in sorted(set(paranoid_features.keys()) - set(features.keys())): + for feature in sorted( + set(paranoid_features.keys()) - set(features.keys()) + ): logger.warning("paranoid: - %s", feature) assert features == paranoid_features @@ -2208,7 +2380,10 @@ def collect_rule_file_paths(rule_paths: list[Path]) -> list[Path]: continue for file in files: if not file.endswith(".yml"): - if not (file.startswith(".git") or file.endswith((".git", ".md", ".txt"))): + if not ( + file.startswith(".git") + or file.endswith((".git", ".md", ".txt")) + ): # expect to see .git* files, readme.md, format.md, and maybe a .git directory # other things maybe are rules, but are mis-named. logger.warning("skipping non-.yml file: %s", file) diff --git a/rules b/rules index 2af9fbfc..03a20f69 160000 --- a/rules +++ b/rules @@ -1 +1 @@ -Subproject commit 2af9fbfc1c9b4634dbeb76b5d34fca9389fa7f80 +Subproject commit 03a20f69ae05e4c48467b06ee69faaa773957684 diff --git a/tests/data b/tests/data index f41a1998..413fd280 160000 --- a/tests/data +++ b/tests/data @@ -1 +1 @@ -Subproject commit f41a1998b92d391a26858f8ae4e6c92ac7394411 +Subproject commit 413fd2803e0f45c7af1eb27a091a1d93221d5d04 diff --git a/tests/test_capabilities.py b/tests/test_capabilities.py index dc94d2d4..eb34c26a 100644 --- a/tests/test_capabilities.py +++ b/tests/test_capabilities.py @@ -23,10 +23,11 @@ from capa.features.extractors.base_extractor import SampleHashes, FunctionFilter def test_match_across_scopes_file_function(z9324d_extractor): - rules = capa.rules.RuleSet([ - # this rule should match on a function (0x4073F0) - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + # this rule should match on a function (0x4073F0) + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: install service @@ -41,10 +42,10 @@ def test_match_across_scopes_file_function(z9324d_extractor): - api: advapi32.CreateServiceA - api: advapi32.StartServiceA """) - ), - # this rule should match on a file feature - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + ), + # this rule should match on a file feature + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: .text section @@ -56,12 +57,12 @@ def test_match_across_scopes_file_function(z9324d_extractor): features: - section: .text """) - ), - # this rule should match on earlier rule matches: - # - install service, with function scope - # - .text section, with file scope - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + ), + # this rule should match on earlier rule matches: + # - install service, with function scope + # - .text section, with file scope + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: .text section and install service @@ -75,8 +76,9 @@ def test_match_across_scopes_file_function(z9324d_extractor): - match: install service - match: .text section """) - ), - ]) + ), + ] + ) capabilities = capa.capabilities.common.find_capabilities(rules, z9324d_extractor) assert "install service" in capabilities.matches assert ".text section" in capabilities.matches @@ -84,10 +86,11 @@ def test_match_across_scopes_file_function(z9324d_extractor): def test_match_across_scopes(z9324d_extractor): - rules = capa.rules.RuleSet([ - # this rule should match on a basic block (including at least 0x403685) - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + # this rule should match on a basic block (including at least 0x403685) + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: tight loop @@ -99,11 +102,11 @@ def test_match_across_scopes(z9324d_extractor): features: - characteristic: tight loop """) - ), - # this rule should match on a function (0x403660) - # based on API, as well as prior basic block rule match - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + ), + # this rule should match on a function (0x403660) + # based on API, as well as prior basic block rule match + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: kill thread loop @@ -118,10 +121,10 @@ def test_match_across_scopes(z9324d_extractor): - api: kernel32.CloseHandle - match: tight loop """) - ), - # this rule should match on a file feature and a prior function rule match - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + ), + # this rule should match on a file feature and a prior function rule match + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: kill thread program @@ -135,8 +138,9 @@ def test_match_across_scopes(z9324d_extractor): - section: .text - match: kill thread loop """) - ), - ]) + ), + ] + ) capabilities = capa.capabilities.common.find_capabilities(rules, z9324d_extractor) assert "tight loop" in capabilities.matches assert "kill thread loop" in capabilities.matches @@ -144,9 +148,10 @@ def test_match_across_scopes(z9324d_extractor): def test_subscope_bb_rules(z9324d_extractor): - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: test rule @@ -158,17 +163,19 @@ def test_subscope_bb_rules(z9324d_extractor): - basic block: - characteristic: tight loop """) - ) - ]) + ) + ] + ) # tight loop at 0x403685 capabilities = capa.capabilities.common.find_capabilities(rules, z9324d_extractor) assert "test rule" in capabilities.matches def test_match_specific_functions(z9324d_extractor): - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: receive data @@ -181,8 +188,9 @@ def test_match_specific_functions(z9324d_extractor): - or: - api: recv """) - ) - ]) + ) + ] + ) extractor = FunctionFilter(z9324d_extractor, {0x4019C0}) capabilities = capa.capabilities.common.find_capabilities(rules, extractor) matches = capabilities.matches["receive data"] @@ -193,9 +201,10 @@ def test_match_specific_functions(z9324d_extractor): def test_byte_matching(z9324d_extractor): - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: byte match test @@ -206,16 +215,18 @@ def test_byte_matching(z9324d_extractor): - and: - bytes: ED 24 9E F4 52 A9 07 47 55 8E E1 AB 30 8E 23 61 """) - ) - ]) + ) + ] + ) capabilities = capa.capabilities.common.find_capabilities(rules, z9324d_extractor) assert "byte match test" in capabilities.matches def test_com_feature_matching(z395eb_extractor): - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: initialize IWebBrowser2 @@ -228,16 +239,18 @@ def test_com_feature_matching(z395eb_extractor): - com/class: InternetExplorer #bytes: 01 DF 02 00 00 00 00 00 C0 00 00 00 00 00 00 46 = CLSID_InternetExplorer - com/interface: IWebBrowser2 #bytes: 61 16 0C D3 AF CD D0 11 8A 3E 00 C0 4F C9 E2 6E = IID_IWebBrowser2 """) - ) - ]) + ) + ] + ) capabilities = capa.main.find_capabilities(rules, z395eb_extractor) assert "initialize IWebBrowser2" in capabilities.matches def test_count_bb(z9324d_extractor): - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: count bb @@ -249,17 +262,19 @@ def test_count_bb(z9324d_extractor): - and: - count(basic blocks): 1 or more """) - ) - ]) + ) + ] + ) capabilities = capa.capabilities.common.find_capabilities(rules, z9324d_extractor) assert "count bb" in capabilities.matches def test_instruction_scope(z9324d_extractor): # .text:004071A4 68 E8 03 00 00 push 3E8h - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: push 1000 @@ -272,8 +287,9 @@ def test_instruction_scope(z9324d_extractor): - mnemonic: push - number: 1000 """) - ) - ]) + ) + ] + ) capabilities = capa.capabilities.common.find_capabilities(rules, z9324d_extractor) assert "push 1000" in capabilities.matches assert 0x4071A4 in {result[0] for result in capabilities.matches["push 1000"]} @@ -283,9 +299,10 @@ def test_instruction_subscope(z9324d_extractor): # .text:00406F60 sub_406F60 proc near # [...] # .text:004071A4 68 E8 03 00 00 push 3E8h - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: push 1000 on i386 @@ -300,11 +317,14 @@ def test_instruction_subscope(z9324d_extractor): - mnemonic: push - number: 1000 """) - ) - ]) + ) + ] + ) capabilities = capa.capabilities.common.find_capabilities(rules, z9324d_extractor) assert "push 1000 on i386" in capabilities.matches - assert 0x406F60 in {result[0] for result in capabilities.matches["push 1000 on i386"]} + assert 0x406F60 in { + result[0] for result in capabilities.matches["push 1000 on i386"] + } def test_find_file_capabilities_preserves_address_zero(): @@ -317,9 +337,10 @@ def test_find_file_capabilities_preserves_address_zero(): file_features=[(addr, feature)], functions={}, ) - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: embedded pe @@ -329,8 +350,9 @@ def test_find_file_capabilities_preserves_address_zero(): features: - characteristic: embedded pe """) - ) - ]) + ) + ] + ) caps = capa.capabilities.common.find_file_capabilities(rules, extractor, {}) assert feature in caps.features assert addr in caps.features[feature] diff --git a/tests/test_engine.py b/tests/test_engine.py index 88184f1a..6caf5d6e 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -17,7 +17,13 @@ import pytest import capa.features.address from capa.engine import Or, And, Not, Some, Range from capa.features.insn import Number -from capa.features.address import ThreadAddress, ProcessAddress, DynamicCallAddress, DNTokenOffsetAddress, AbsoluteVirtualAddress +from capa.features.address import ( + ThreadAddress, + ProcessAddress, + DynamicCallAddress, + DNTokenOffsetAddress, + AbsoluteVirtualAddress, +) ADDR1 = capa.features.address.AbsoluteVirtualAddress(0x401001) ADDR2 = capa.features.address.AbsoluteVirtualAddress(0x401002) @@ -78,7 +84,14 @@ def test_and(): assert bool(And([Number(1), Number(2)]).evaluate({Number(0): {ADDR1}})) is False assert bool(And([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}})) is False assert bool(And([Number(1), Number(2)]).evaluate({Number(2): {ADDR1}})) is False - assert bool(And([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}, Number(2): {ADDR2}})) is True + assert ( + bool( + And([Number(1), Number(2)]).evaluate( + {Number(1): {ADDR1}, Number(2): {ADDR2}} + ) + ) + is True + ) def test_or(): @@ -87,7 +100,14 @@ def test_or(): assert bool(Or([Number(1), Number(2)]).evaluate({Number(0): {ADDR1}})) is False assert bool(Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}})) is True assert bool(Or([Number(1), Number(2)]).evaluate({Number(2): {ADDR1}})) is True - assert bool(Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}, Number(2): {ADDR2}})) is True + assert ( + bool( + Or([Number(1), Number(2)]).evaluate( + {Number(1): {ADDR1}, Number(2): {ADDR2}} + ) + ) + is True + ) def test_not(): @@ -99,38 +119,54 @@ def test_some(): assert bool(Some(0, [Number(1)]).evaluate({Number(0): {ADDR1}})) is True assert bool(Some(1, [Number(1)]).evaluate({Number(0): {ADDR1}})) is False - assert bool(Some(2, [Number(1), Number(2), Number(3)]).evaluate({Number(0): {ADDR1}})) is False - assert bool(Some(2, [Number(1), Number(2), Number(3)]).evaluate({Number(0): {ADDR1}, Number(1): {ADDR1}})) is False + assert ( + bool(Some(2, [Number(1), Number(2), Number(3)]).evaluate({Number(0): {ADDR1}})) + is False + ) assert ( bool( - Some(2, [Number(1), Number(2), Number(3)]).evaluate({ - Number(0): {ADDR1}, - Number(1): {ADDR1}, - Number(2): {ADDR1}, - }) + Some(2, [Number(1), Number(2), Number(3)]).evaluate( + {Number(0): {ADDR1}, Number(1): {ADDR1}} + ) + ) + is False + ) + assert ( + bool( + Some(2, [Number(1), Number(2), Number(3)]).evaluate( + { + Number(0): {ADDR1}, + Number(1): {ADDR1}, + Number(2): {ADDR1}, + } + ) ) is True ) assert ( bool( - Some(2, [Number(1), Number(2), Number(3)]).evaluate({ - Number(0): {ADDR1}, - Number(1): {ADDR1}, - Number(2): {ADDR1}, - Number(3): {ADDR1}, - }) + Some(2, [Number(1), Number(2), Number(3)]).evaluate( + { + Number(0): {ADDR1}, + Number(1): {ADDR1}, + Number(2): {ADDR1}, + Number(3): {ADDR1}, + } + ) ) is True ) assert ( bool( - Some(2, [Number(1), Number(2), Number(3)]).evaluate({ - Number(0): {ADDR1}, - Number(1): {ADDR1}, - Number(2): {ADDR1}, - Number(3): {ADDR1}, - Number(4): {ADDR1}, - }) + Some(2, [Number(1), Number(2), Number(3)]).evaluate( + { + Number(0): {ADDR1}, + Number(1): {ADDR1}, + Number(2): {ADDR1}, + Number(3): {ADDR1}, + Number(4): {ADDR1}, + } + ) ) is True ) @@ -138,21 +174,35 @@ def test_some(): def test_complex(): assert True is bool( - Or([And([Number(1), Number(2)]), Or([Number(3), Some(2, [Number(4), Number(5), Number(6)])])]).evaluate({ - Number(5): {ADDR1}, - Number(6): {ADDR1}, - Number(7): {ADDR1}, - Number(8): {ADDR1}, - }) + Or( + [ + And([Number(1), Number(2)]), + Or([Number(3), Some(2, [Number(4), Number(5), Number(6)])]), + ] + ).evaluate( + { + Number(5): {ADDR1}, + Number(6): {ADDR1}, + Number(7): {ADDR1}, + Number(8): {ADDR1}, + } + ) ) assert False is bool( - Or([And([Number(1), Number(2)]), Or([Number(3), Some(2, [Number(4), Number(5)])])]).evaluate({ - Number(5): {ADDR1}, - Number(6): {ADDR1}, - Number(7): {ADDR1}, - Number(8): {ADDR1}, - }) + Or( + [ + And([Number(1), Number(2)]), + Or([Number(3), Some(2, [Number(4), Number(5)])]), + ] + ).evaluate( + { + Number(5): {ADDR1}, + Number(6): {ADDR1}, + Number(7): {ADDR1}, + Number(8): {ADDR1}, + } + ) ) @@ -175,27 +225,62 @@ def test_range(): assert bool(Range(Number(1), max=1).evaluate({Number(1): {ADDR1}})) is True assert bool(Range(Number(1), max=2).evaluate({Number(1): {ADDR1}})) is True assert bool(Range(Number(1), max=2).evaluate({Number(1): {ADDR1, ADDR2}})) is True - assert bool(Range(Number(1), max=2).evaluate({Number(1): {ADDR1, ADDR2, ADDR3}})) is False + assert ( + bool(Range(Number(1), max=2).evaluate({Number(1): {ADDR1, ADDR2, ADDR3}})) + is False + ) # we can do an exact match by setting min==max assert bool(Range(Number(1), min=1, max=1).evaluate({Number(1): {}})) is False # type: ignore assert bool(Range(Number(1), min=1, max=1).evaluate({Number(1): {ADDR1}})) is True - assert bool(Range(Number(1), min=1, max=1).evaluate({Number(1): {ADDR1, ADDR2}})) is False + assert ( + bool(Range(Number(1), min=1, max=1).evaluate({Number(1): {ADDR1, ADDR2}})) + is False + ) # bounded range assert bool(Range(Number(1), min=1, max=3).evaluate({Number(1): {}})) is False # type: ignore assert bool(Range(Number(1), min=1, max=3).evaluate({Number(1): {ADDR1}})) is True - assert bool(Range(Number(1), min=1, max=3).evaluate({Number(1): {ADDR1, ADDR2}})) is True - assert bool(Range(Number(1), min=1, max=3).evaluate({Number(1): {ADDR1, ADDR2, ADDR3}})) is True - assert bool(Range(Number(1), min=1, max=3).evaluate({Number(1): {ADDR1, ADDR2, ADDR3, ADDR4}})) is False + assert ( + bool(Range(Number(1), min=1, max=3).evaluate({Number(1): {ADDR1, ADDR2}})) + is True + ) + assert ( + bool( + Range(Number(1), min=1, max=3).evaluate({Number(1): {ADDR1, ADDR2, ADDR3}}) + ) + is True + ) + assert ( + bool( + Range(Number(1), min=1, max=3).evaluate( + {Number(1): {ADDR1, ADDR2, ADDR3, ADDR4}} + ) + ) + is False + ) def test_short_circuit(): assert bool(Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}})) is True # with short circuiting, only the children up until the first satisfied child are captured. - assert len(Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}}, short_circuit=True).children) == 1 - assert len(Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}}, short_circuit=False).children) == 2 + assert ( + len( + Or([Number(1), Number(2)]) + .evaluate({Number(1): {ADDR1}}, short_circuit=True) + .children + ) + == 1 + ) + assert ( + len( + Or([Number(1), Number(2)]) + .evaluate({Number(1): {ADDR1}}, short_circuit=False) + .children + ) + == 2 + ) def test_eval_order(): @@ -206,14 +291,29 @@ def test_eval_order(): # with short circuiting, only the children up until the first satisfied child are captured. assert len(Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}}).children) == 1 assert len(Or([Number(1), Number(2)]).evaluate({Number(2): {ADDR1}}).children) == 2 - assert len(Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}, Number(2): {ADDR1}}).children) == 1 + assert ( + len( + Or([Number(1), Number(2)]) + .evaluate({Number(1): {ADDR1}, Number(2): {ADDR1}}) + .children + ) + == 1 + ) # and its guaranteed that children are evaluated in order. - assert Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}}).children[0].statement == Number(1) - assert Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}}).children[0].statement != Number(2) + assert Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}}).children[ + 0 + ].statement == Number(1) + assert Or([Number(1), Number(2)]).evaluate({Number(1): {ADDR1}}).children[ + 0 + ].statement != Number(2) - assert Or([Number(1), Number(2)]).evaluate({Number(2): {ADDR1}}).children[1].statement == Number(2) - assert Or([Number(1), Number(2)]).evaluate({Number(2): {ADDR1}}).children[1].statement != Number(1) + assert Or([Number(1), Number(2)]).evaluate({Number(2): {ADDR1}}).children[ + 1 + ].statement == Number(2) + assert Or([Number(1), Number(2)]).evaluate({Number(2): {ADDR1}}).children[ + 1 + ].statement != Number(1) def test_address_cross_type_eq(): diff --git a/tests/test_freeze_static.py b/tests/test_freeze_static.py index 9d53c92a..4f060ba8 100644 --- a/tests/test_freeze_static.py +++ b/tests/test_freeze_static.py @@ -28,7 +28,11 @@ import capa.features.basicblock import capa.features.extractors.null import capa.features.extractors.base_extractor from capa.features.address import Address, AbsoluteVirtualAddress -from capa.features.extractors.base_extractor import BBHandle, SampleHashes, FunctionHandle +from capa.features.extractors.base_extractor import ( + BBHandle, + SampleHashes, + FunctionHandle, +) EXTRACTOR = capa.features.extractors.null.NullStaticFeatureExtractor( base_address=AbsoluteVirtualAddress(0x401000), @@ -39,28 +43,54 @@ EXTRACTOR = capa.features.extractors.null.NullStaticFeatureExtractor( ), global_features=[], file_features=[ - (AbsoluteVirtualAddress(0x402345), capa.features.common.Characteristic("embedded pe")), + ( + AbsoluteVirtualAddress(0x402345), + capa.features.common.Characteristic("embedded pe"), + ), ], functions={ - AbsoluteVirtualAddress(0x401000): capa.features.extractors.null.FunctionFeatures( + AbsoluteVirtualAddress( + 0x401000 + ): capa.features.extractors.null.FunctionFeatures( features=[ - (AbsoluteVirtualAddress(0x401000), capa.features.common.Characteristic("indirect call")), + ( + AbsoluteVirtualAddress(0x401000), + capa.features.common.Characteristic("indirect call"), + ), ], basic_blocks={ - AbsoluteVirtualAddress(0x401000): capa.features.extractors.null.BasicBlockFeatures( + AbsoluteVirtualAddress( + 0x401000 + ): capa.features.extractors.null.BasicBlockFeatures( features=[ - (AbsoluteVirtualAddress(0x401000), capa.features.common.Characteristic("tight loop")), + ( + AbsoluteVirtualAddress(0x401000), + capa.features.common.Characteristic("tight loop"), + ), ], instructions={ - AbsoluteVirtualAddress(0x401000): capa.features.extractors.null.InstructionFeatures( + AbsoluteVirtualAddress( + 0x401000 + ): capa.features.extractors.null.InstructionFeatures( features=[ - (AbsoluteVirtualAddress(0x401000), capa.features.insn.Mnemonic("xor")), - (AbsoluteVirtualAddress(0x401000), capa.features.common.Characteristic("nzxor")), + ( + AbsoluteVirtualAddress(0x401000), + capa.features.insn.Mnemonic("xor"), + ), + ( + AbsoluteVirtualAddress(0x401000), + capa.features.common.Characteristic("nzxor"), + ), ], ), - AbsoluteVirtualAddress(0x401002): capa.features.extractors.null.InstructionFeatures( + AbsoluteVirtualAddress( + 0x401002 + ): capa.features.extractors.null.InstructionFeatures( features=[ - (AbsoluteVirtualAddress(0x401002), capa.features.insn.Mnemonic("mov")), + ( + AbsoluteVirtualAddress(0x401002), + capa.features.insn.Mnemonic("mov"), + ), ], ), }, @@ -80,15 +110,18 @@ def test_null_feature_extractor(): bbh = BBHandle(AbsoluteVirtualAddress(0x401000), None) assert addresses(EXTRACTOR.get_functions()) == [AbsoluteVirtualAddress(0x401000)] - assert addresses(EXTRACTOR.get_basic_blocks(fh)) == [AbsoluteVirtualAddress(0x401000)] + assert addresses(EXTRACTOR.get_basic_blocks(fh)) == [ + AbsoluteVirtualAddress(0x401000) + ] assert addresses(EXTRACTOR.get_instructions(fh, bbh)) == [ AbsoluteVirtualAddress(0x401000), AbsoluteVirtualAddress(0x401002), ] - rules = capa.rules.RuleSet([ - capa.rules.Rule.from_yaml( - textwrap.dedent(""" + rules = capa.rules.RuleSet( + [ + capa.rules.Rule.from_yaml( + textwrap.dedent(""" rule: meta: name: xor loop @@ -101,8 +134,9 @@ def test_null_feature_extractor(): - mnemonic: xor - characteristic: nzxor """) - ), - ]) + ), + ] + ) capabilities = capa.main.find_capabilities(rules, EXTRACTOR) assert "xor loop" in capabilities.matches @@ -113,10 +147,14 @@ def compare_extractors(a, b): assert addresses(a.get_functions()) == addresses(b.get_functions()) for f in a.get_functions(): assert addresses(a.get_basic_blocks(f)) == addresses(b.get_basic_blocks(f)) - assert sorted(set(a.extract_function_features(f))) == sorted(set(b.extract_function_features(f))) + assert sorted(set(a.extract_function_features(f))) == sorted( + set(b.extract_function_features(f)) + ) for bb in a.get_basic_blocks(f): - assert addresses(a.get_instructions(f, bb)) == addresses(b.get_instructions(f, bb)) + assert addresses(a.get_instructions(f, bb)) == addresses( + b.get_instructions(f, bb) + ) assert sorted(set(a.extract_basic_block_features(f, bb))) == sorted( set(b.extract_basic_block_features(f, bb)) ) @@ -159,7 +197,9 @@ def test_serialize_features(): roundtrip_feature(capa.features.file.Import("#11")) roundtrip_feature(capa.features.insn.OperandOffset(0, 0x8)) roundtrip_feature( - capa.features.insn.Property("System.IO.FileInfo::Length", access=capa.features.common.FeatureAccess.READ) + capa.features.insn.Property( + "System.IO.FileInfo::Length", access=capa.features.common.FeatureAccess.READ + ) ) roundtrip_feature(capa.features.insn.Property("System.IO.FileInfo::Length")) diff --git a/tests/test_helpers.py b/tests/test_helpers.py index d4555ed5..0edf5b95 100644 --- a/tests/test_helpers.py +++ b/tests/test_helpers.py @@ -32,7 +32,11 @@ from capa.features.extractors import helpers CD = Path(__file__).resolve().parent DRAKVUF_LOG_GZ = ( - CD / "data" / "dynamic" / "drakvuf" / "93b2d1840566f45fab674ebc79a9d19c88993bcb645e0357f3cb584d16e7c795.log.gz" + CD + / "data" + / "dynamic" + / "drakvuf" + / "93b2d1840566f45fab674ebc79a9d19c88993bcb645e0357f3cb584d16e7c795.log.gz" ) @@ -56,7 +60,9 @@ def test_generate_symbols(): ) # A/W import - symbols = list(helpers.generate_symbols("kernel32", "CreateFileA", include_dll=True)) + symbols = list( + helpers.generate_symbols("kernel32", "CreateFileA", include_dll=True) + ) assert len(symbols) == 4 assert "kernel32.CreateFileA" in symbols assert "kernel32.CreateFile" in symbols @@ -75,7 +81,9 @@ def test_generate_symbols(): assert "ws2_32.#1" in symbols # A/W api - symbols = list(helpers.generate_symbols("kernel32", "CreateFileA", include_dll=False)) + symbols = list( + helpers.generate_symbols("kernel32", "CreateFileA", include_dll=False) + ) assert len(symbols) == 2 assert "CreateFileA" in symbols assert "CreateFile" in symbols diff --git a/tests/test_os_detection.py b/tests/test_os_detection.py index 9dc6832d..0f281f1f 100644 --- a/tests/test_os_detection.py +++ b/tests/test_os_detection.py @@ -111,76 +111,78 @@ def test_elf_parse_capa_pyinstaller_header(): # compressed ELF header of capa-v5.1.0-linux # SHA256 e16974994914466647e24cdcfb6a6f8710297a4def21525e53f73c72c4b52fcf elf_header = zlib.decompress( - b"".join([ - b"\x78\x9c\x8d\x56\x4f\x88\x1c\xd5\x13\xae\x1d\x35\x0a\x7a\x58\x65", - b"\xd1\xa0\x9b\xb0\x82\x11\x14\x67\x63\xd6\xcd\x26\xf1\xf0\x63\x49", - b"\xdc\xc4\xc8\x26\x98\x7f\x07\x89\xa4\xed\xe9\x7e\x6f\xa6\x99\xd7", - b"\xaf\xdb\xee\x37\xbb\x13\x3d\xb8\x78\x8a\x28\x28\x1e\xbc\x09\x7b", - b"\xf0\xcf\x82\xa0\x41\x10\x23\xa8\x07\x89\x17\x41\x85\x88\x07\x2f", - b"\xe2\x25\xb0\x17\x51\x7e\x07\xd9\x5b\x52\xf5\xfe\xcc\x36\x71\x0a", - b"\x6c\x98\xa9\xf7\xbe\xf9\xea\xab\xaf\xea\x35\x3d\xfd\xda\xd2\xf2", - b"\xd1\xd6\xc4\x04\x84\xab\x05\xff\x03\xda\x2d\xed\x59\xb4\x7b\xf7", - b"\x8d\xf1\xef\x57\x5b\x81\xb3\x08\x07\xe1\x6e\xfc\x9e\x86\x87\x60", - b"\x07\xee\x6f\x6f\xf2\xfc\x2a\xc4\x9e\xcf\x0a\xf1\x2e\xcf\xbb\xcd", - b"\xe7\x6d\x78\x7c\xa3\xe5\xf8\x21\x4e\x7b\x5e\x88\xc1\x21\x45\xca", - b"\xdb\xbe\xb6\x2b\xd3\x75\xe9\x01\xb7\x0b\x11\x26\xb7\xf3\xee\xa0", - b"\xc5\x8c\xc7\x67\x7c\x9e\x8f\xf9\x8b\x6e\x1b\x62\x33\xcf\xd6\x5b", - b"\xf3\xf8\x9a\xcf\xf3\xf1\xca\x7e\xb7\x0d\xb1\x99\x47\xb3\xd9\xfc", - b"\xc6\xed\x37\x7f\x74\xfc\x10\xaf\xf8\x26\x36\x86\xbe\x33\x9f\x47", - b"\xe3\xa0\xbc\x2d\x9f\xb7\xe5\xf9\x21\x5a\x42\x23\x86\x79\x92\x1c", - b"\x7d\xae\x7a\xfc\xaa\x9f\x63\x88\xcf\x78\x5e\x88\x61\x86\xcf\x5f", - b"\x37\x29\xad\x37\xd7\xbd\xcf\x75\xef\xd3\xc7\x27\xe8\xa0\x1a\x31", - b"\xe4\x9d\xc2\x3c\xf2\xf9\x5f\x2f\xdf\x1e\x9c\x0e\xf5\x98\xb9\xec", - b"\xf4\xfe\x43\x0c\xe7\xbe\x57\x65\x9d\x85\xf9\xbd\x2a\x6d\xab\x4c", - b"\x0f\x86\xed\xe1\xc1\x85\xf6\xc2\xfc\x6c\x5d\xcc\xce\x59\x4f\x53", - b"\xfe\x9e\x3a\x76\xf2\x1c\x9c\xfd\xe5\xd9\x33\xe2\xcc\x4b\x17\x76", - b"\x7f\xfd\xd1\xd4\xe6\xe5\xa9\xe5\xf3\x4f\xff\x7c\x82\x38\xe4\x81", - b"\xf4\x88\xd3\x9c\x35\xdd\x12\x94\x7b\xaa\x71\x6e\x30\x31\x03\x6b", - b"\x13\x93\x2d\xc2\x4e\x7b\x0f\x8f\xed\x7a\x6b\x5a\x9e\x8b\x27\x0f", - b"\xfd\xff\xcd\x70\x5b\xfe\xeb\xd2\x28\x7a\xdf\x18\xfc\x1a\x0a\x8f", - b"\xc3\xff\x60\xf0\x0b\xf8\x19\x87\x7f\xc7\xe8\x3f\xc7\xe0\x27\x18", - b"\x9d\x1b\x0c\x7e\x9d\xc1\xe9\xb8\xc6\xe1\x6f\x33\xf8\x97\x4c\x5f", - b"\x6d\x86\xbf\x83\xe1\x7f\xc0\xf4\xf5\x08\x83\xff\xc9\xe8\xff\xc5", - b"\xe8\x4b\x06\x7f\x90\xc1\xdf\x60\xf0\x1f\x18\xfc\x13\xc6\xe7\x49", - b"\x86\x7f\x88\xc1\xff\x61\xfa\xa2\xa7\xf2\x38\xfc\x08\x83\x7f\xc5", - b"\xe0\x47\x99\xba\x37\x18\xfc\x4e\x46\xe7\x57\xc6\xe7\xf7\x0c\xfe", - b"\x2e\xa3\x73\x96\xa9\xfb\x05\xa3\x33\xc7\xf0\x5f\x67\xf4\x7f\x67", - b"\x74\x1e\x67\xf8\x6d\x46\x9f\x9e\x17\xe1\x2f\xa5\x79\x9d\x67\xf8", - b"\x9f\x72\x3e\x19\x7c\x91\xc1\x0f\x33\xfe\x1f\x66\xf8\x2f\x33\xfc", - b"\xfb\x99\x7e\x25\x83\xbf\xcf\xe8\xec\x66\xf0\x75\xc6\xcf\x25\x86", - b"\xff\x2d\xc3\x7f\x87\xc1\xe9\x99\x3e\x0e\xbf\x87\xe1\xbf\xc0\xf4", - b"\x45\x7f\xef\xe3\xf0\x0f\x19\xfc\x3d\x06\xa7\xd7\x80\x71\xf8\x01", - b"\xa6\x6e\xc6\xf0\x3f\x67\xf8\x9f\x31\xfc\x9f\x18\xfc\x51\x66\x0e", - b"\x29\x83\x6f\x31\xf8\xc7\x0c\xbe\x8b\xf1\x99\x73\xcf\x4f\x86\xbf", - b"\x93\xf1\x9f\x32\xf8\x1e\xee\x79\x88\x75\xef\x45\xb5\xf5\x6b\xee", - b"\x7d\x22\xbc\x1f\x4d\x79\x7c\xe3\x16\xfc\x37\x8f\x5f\xbe\x05\x87", - b"\x28\xea\xe6\x85\x8e\x6a\x13\x57\x26\x8a\x20\x55\x89\x2a\x6a\x81", - b"\xb1\xbe\x98\xe3\x77\x51\x0a\x8d\x41\x54\x55\x51\x41\xa6\xa5\x8a", - b"\x8d\x08\xf1\xb8\xce\x4c\x14\x36\x4b\x3a\x45\x2d\xe4\xe9\x22\x52", - b"\x45\x12\x9b\xac\xd0\x50\xc5\x19\x6a\xc9\xa2\xea\xc3\x6a\x9c\x99", - b"\x32\x23\xce\xb0\xec\x46\x9d\xb8\x16\x3a\xce\x05\xe4\xfd\xd4\x88", - b"\xbc\x04\x29\xd5\xa0\xee\x41\x6d\xaa\xa4\xbc\x08\x32\xe9\xe5\x45", - b"\x0a\x95\x88\xd3\x34\xab\xa0\x16\x86\x24\x15\x49\x91\x9f\xd5\xa4", - b"\xd6\x44\x43\xb6\x4e\x30\x39\x42\xfb\x55\x3a\x28\xa1\x74\x3e\x6d", - b"\x0b\x36\x31\xeb\xea\x58\x39\x1e\xf2\xf3\x4e\x6d\x0a\x4c\xc6\x04", - b"\x35\xc4\x8e\x0d\x0c\x34\xbe\x66\xf5\xc9\x05\xb2\xb1\x9c\xc2\x3a", - b"\x48\x4f\x33\x0d\x5d\x61\xfd\xf6\x33\x65\x05\x4c\xd1\x07\x29\x0a", - b"\x09\xe8\xc3\x91\x2a\x85\x56\xca\x2a\x31\x0a\x30\xdb\x76\x53\xe5", - b"\xa4\x93\x8b\x9c\x5c\x25\x4a\xc4\x15\x1a\xc2\x22\xd8\x80\xd0\x2b", - b"\x58\x56\x96\x55\xa6\x8d\x8c\x92\x5e\x9f\xca\x14\x03\x63\xd9\xb6", - b"\x65\x3b\xf7\x28\x5a\xa9\x75\x83\x94\x8f\xaa\xe1\x48\xad\xc3\x32", - b"\x36\x3d\x90\x46\x20\x0e\x5a\x45\x2a\xd6\x5d\x3c\x82\x02\x68\x32", - b"\x54\x1d\x7d\x53\x2d\x54\xa7\xda\x38\x9a\xa6\x1c\x4d\xd4\x76\x2c", - b"\x86\x22\x59\x29\xdd\x64\x50\x38\x8a\x82\xb4\xa5\xc9\x0c\x7b\x2b", - b"\x40\xae\x56\x19\x1e\xb7\xa4\x2c\xa4\x38\xa7\x96\x80\x9d\x10\xe8", - b"\xfb\xa8\x92\x1e\x55\x5a\x69\x76\x67\xcf\x64\x9b\xee\xc6\xed\xc0", - b"\xd8\xb8\x3c\x61\x3a\x03\x69\xd3\x71\x5a\x18\xdc\xe1\xe1\xd9\x64", - b"\x9d\xc4\xdf\x90\x79\x8c\x27\x21\xdd\x0f\xb5\x29\xed\xa0\x6a\x21", - b"\xfa\x05\x84\xb6\xc8\x9d\x00\x4c\x49\x95\x7b\x4b\xc6\xe5\x2b\xb4", - b"\xda\x47\xab\xd2\xf4\xc8\x27\xed\x9f\xa4\x7d\x42\xab\x05\x38\xb6", - b"\x7c\xfc\xf0\x91\x68\x6e\x76\x6e\x76\xff\x68\x7d\x60\xb4\xda\x37", - b"\x3f\x5a\x3e\x35\x5a\x35\x30\x5c\xc3\x4d\x95\x6e\xa4\x60", - ]) + b"".join( + [ + b"\x78\x9c\x8d\x56\x4f\x88\x1c\xd5\x13\xae\x1d\x35\x0a\x7a\x58\x65", + b"\xd1\xa0\x9b\xb0\x82\x11\x14\x67\x63\xd6\xcd\x26\xf1\xf0\x63\x49", + b"\xdc\xc4\xc8\x26\x98\x7f\x07\x89\xa4\xed\xe9\x7e\x6f\xa6\x99\xd7", + b"\xaf\xdb\xee\x37\xbb\x13\x3d\xb8\x78\x8a\x28\x28\x1e\xbc\x09\x7b", + b"\xf0\xcf\x82\xa0\x41\x10\x23\xa8\x07\x89\x17\x41\x85\x88\x07\x2f", + b"\xe2\x25\xb0\x17\x51\x7e\x07\xd9\x5b\x52\xf5\xfe\xcc\x36\x71\x0a", + b"\x6c\x98\xa9\xf7\xbe\xf9\xea\xab\xaf\xea\x35\x3d\xfd\xda\xd2\xf2", + b"\xd1\xd6\xc4\x04\x84\xab\x05\xff\x03\xda\x2d\xed\x59\xb4\x7b\xf7", + b"\x8d\xf1\xef\x57\x5b\x81\xb3\x08\x07\xe1\x6e\xfc\x9e\x86\x87\x60", + b"\x07\xee\x6f\x6f\xf2\xfc\x2a\xc4\x9e\xcf\x0a\xf1\x2e\xcf\xbb\xcd", + b"\xe7\x6d\x78\x7c\xa3\xe5\xf8\x21\x4e\x7b\x5e\x88\xc1\x21\x45\xca", + b"\xdb\xbe\xb6\x2b\xd3\x75\xe9\x01\xb7\x0b\x11\x26\xb7\xf3\xee\xa0", + b"\xc5\x8c\xc7\x67\x7c\x9e\x8f\xf9\x8b\x6e\x1b\x62\x33\xcf\xd6\x5b", + b"\xf3\xf8\x9a\xcf\xf3\xf1\xca\x7e\xb7\x0d\xb1\x99\x47\xb3\xd9\xfc", + b"\xc6\xed\x37\x7f\x74\xfc\x10\xaf\xf8\x26\x36\x86\xbe\x33\x9f\x47", + b"\xe3\xa0\xbc\x2d\x9f\xb7\xe5\xf9\x21\x5a\x42\x23\x86\x79\x92\x1c", + b"\x7d\xae\x7a\xfc\xaa\x9f\x63\x88\xcf\x78\x5e\x88\x61\x86\xcf\x5f", + b"\x37\x29\xad\x37\xd7\xbd\xcf\x75\xef\xd3\xc7\x27\xe8\xa0\x1a\x31", + b"\xe4\x9d\xc2\x3c\xf2\xf9\x5f\x2f\xdf\x1e\x9c\x0e\xf5\x98\xb9\xec", + b"\xf4\xfe\x43\x0c\xe7\xbe\x57\x65\x9d\x85\xf9\xbd\x2a\x6d\xab\x4c", + b"\x0f\x86\xed\xe1\xc1\x85\xf6\xc2\xfc\x6c\x5d\xcc\xce\x59\x4f\x53", + b"\xfe\x9e\x3a\x76\xf2\x1c\x9c\xfd\xe5\xd9\x33\xe2\xcc\x4b\x17\x76", + b"\x7f\xfd\xd1\xd4\xe6\xe5\xa9\xe5\xf3\x4f\xff\x7c\x82\x38\xe4\x81", + b"\xf4\x88\xd3\x9c\x35\xdd\x12\x94\x7b\xaa\x71\x6e\x30\x31\x03\x6b", + b"\x13\x93\x2d\xc2\x4e\x7b\x0f\x8f\xed\x7a\x6b\x5a\x9e\x8b\x27\x0f", + b"\xfd\xff\xcd\x70\x5b\xfe\xeb\xd2\x28\x7a\xdf\x18\xfc\x1a\x0a\x8f", + b"\xc3\xff\x60\xf0\x0b\xf8\x19\x87\x7f\xc7\xe8\x3f\xc7\xe0\x27\x18", + b"\x9d\x1b\x0c\x7e\x9d\xc1\xe9\xb8\xc6\xe1\x6f\x33\xf8\x97\x4c\x5f", + b"\x6d\x86\xbf\x83\xe1\x7f\xc0\xf4\xf5\x08\x83\xff\xc9\xe8\xff\xc5", + b"\xe8\x4b\x06\x7f\x90\xc1\xdf\x60\xf0\x1f\x18\xfc\x13\xc6\xe7\x49", + b"\x86\x7f\x88\xc1\xff\x61\xfa\xa2\xa7\xf2\x38\xfc\x08\x83\x7f\xc5", + b"\xe0\x47\x99\xba\x37\x18\xfc\x4e\x46\xe7\x57\xc6\xe7\xf7\x0c\xfe", + b"\x2e\xa3\x73\x96\xa9\xfb\x05\xa3\x33\xc7\xf0\x5f\x67\xf4\x7f\x67", + b"\x74\x1e\x67\xf8\x6d\x46\x9f\x9e\x17\xe1\x2f\xa5\x79\x9d\x67\xf8", + b"\x9f\x72\x3e\x19\x7c\x91\xc1\x0f\x33\xfe\x1f\x66\xf8\x2f\x33\xfc", + b"\xfb\x99\x7e\x25\x83\xbf\xcf\xe8\xec\x66\xf0\x75\xc6\xcf\x25\x86", + b"\xff\x2d\xc3\x7f\x87\xc1\xe9\x99\x3e\x0e\xbf\x87\xe1\xbf\xc0\xf4", + b"\x45\x7f\xef\xe3\xf0\x0f\x19\xfc\x3d\x06\xa7\xd7\x80\x71\xf8\x01", + b"\xa6\x6e\xc6\xf0\x3f\x67\xf8\x9f\x31\xfc\x9f\x18\xfc\x51\x66\x0e", + b"\x29\x83\x6f\x31\xf8\xc7\x0c\xbe\x8b\xf1\x99\x73\xcf\x4f\x86\xbf", + b"\x93\xf1\x9f\x32\xf8\x1e\xee\x79\x88\x75\xef\x45\xb5\xf5\x6b\xee", + b"\x7d\x22\xbc\x1f\x4d\x79\x7c\xe3\x16\xfc\x37\x8f\x5f\xbe\x05\x87", + b"\x28\xea\xe6\x85\x8e\x6a\x13\x57\x26\x8a\x20\x55\x89\x2a\x6a\x81", + b"\xb1\xbe\x98\xe3\x77\x51\x0a\x8d\x41\x54\x55\x51\x41\xa6\xa5\x8a", + b"\x8d\x08\xf1\xb8\xce\x4c\x14\x36\x4b\x3a\x45\x2d\xe4\xe9\x22\x52", + b"\x45\x12\x9b\xac\xd0\x50\xc5\x19\x6a\xc9\xa2\xea\xc3\x6a\x9c\x99", + b"\x32\x23\xce\xb0\xec\x46\x9d\xb8\x16\x3a\xce\x05\xe4\xfd\xd4\x88", + b"\xbc\x04\x29\xd5\xa0\xee\x41\x6d\xaa\xa4\xbc\x08\x32\xe9\xe5\x45", + b"\x0a\x95\x88\xd3\x34\xab\xa0\x16\x86\x24\x15\x49\x91\x9f\xd5\xa4", + b"\xd6\x44\x43\xb6\x4e\x30\x39\x42\xfb\x55\x3a\x28\xa1\x74\x3e\x6d", + b"\x0b\x36\x31\xeb\xea\x58\x39\x1e\xf2\xf3\x4e\x6d\x0a\x4c\xc6\x04", + b"\x35\xc4\x8e\x0d\x0c\x34\xbe\x66\xf5\xc9\x05\xb2\xb1\x9c\xc2\x3a", + b"\x48\x4f\x33\x0d\x5d\x61\xfd\xf6\x33\x65\x05\x4c\xd1\x07\x29\x0a", + b"\x09\xe8\xc3\x91\x2a\x85\x56\xca\x2a\x31\x0a\x30\xdb\x76\x53\xe5", + b"\xa4\x93\x8b\x9c\x5c\x25\x4a\xc4\x15\x1a\xc2\x22\xd8\x80\xd0\x2b", + b"\x58\x56\x96\x55\xa6\x8d\x8c\x92\x5e\x9f\xca\x14\x03\x63\xd9\xb6", + b"\x65\x3b\xf7\x28\x5a\xa9\x75\x83\x94\x8f\xaa\xe1\x48\xad\xc3\x32", + b"\x36\x3d\x90\x46\x20\x0e\x5a\x45\x2a\xd6\x5d\x3c\x82\x02\x68\x32", + b"\x54\x1d\x7d\x53\x2d\x54\xa7\xda\x38\x9a\xa6\x1c\x4d\xd4\x76\x2c", + b"\x86\x22\x59\x29\xdd\x64\x50\x38\x8a\x82\xb4\xa5\xc9\x0c\x7b\x2b", + b"\x40\xae\x56\x19\x1e\xb7\xa4\x2c\xa4\x38\xa7\x96\x80\x9d\x10\xe8", + b"\xfb\xa8\x92\x1e\x55\x5a\x69\x76\x67\xcf\x64\x9b\xee\xc6\xed\xc0", + b"\xd8\xb8\x3c\x61\x3a\x03\x69\xd3\x71\x5a\x18\xdc\xe1\xe1\xd9\x64", + b"\x9d\xc4\xdf\x90\x79\x8c\x27\x21\xdd\x0f\xb5\x29\xed\xa0\x6a\x21", + b"\xfa\x05\x84\xb6\xc8\x9d\x00\x4c\x49\x95\x7b\x4b\xc6\xe5\x2b\xb4", + b"\xda\x47\xab\xd2\xf4\xc8\x27\xed\x9f\xa4\x7d\x42\xab\x05\x38\xb6", + b"\x7c\xfc\xf0\x91\x68\x6e\x76\x6e\x76\xff\x68\x7d\x60\xb4\xda\x37", + b"\x3f\x5a\x3e\x35\x5a\x35\x30\x5c\xc3\x4d\x95\x6e\xa4\x60", + ] + ) ) assert capa.features.extractors.elf.detect_elf_os(io.BytesIO(elf_header)) == "linux" diff --git a/tests/test_render.py b/tests/test_render.py index c931dfea..3201289d 100644 --- a/tests/test_render.py +++ b/tests/test_render.py @@ -44,7 +44,12 @@ def test_render_offset(): def test_render_property(): assert ( - str(capa.features.insn.Property("System.IO.FileInfo::Length", access=capa.features.common.FeatureAccess.READ)) + str( + capa.features.insn.Property( + "System.IO.FileInfo::Length", + access=capa.features.common.FeatureAccess.READ, + ) + ) == "property/read(System.IO.FileInfo::Length)" ) @@ -184,7 +189,10 @@ def test_render_meta_maec(): (capa.features.common.Regex("^foo"), "regex: ^foo"), (capa.features.common.String("foo"), 'string: "foo" @ 0x401000'), (capa.features.common.Class("BeanFactory"), "class: BeanFactory @ 0x401000"), - (capa.features.common.Namespace("std::enterprise"), "namespace: std::enterprise @ 0x401000"), + ( + capa.features.common.Namespace("std::enterprise"), + "namespace: std::enterprise @ 0x401000", + ), (capa.features.insn.API("CreateFileW"), "api: CreateFileW @ 0x401000"), (capa.features.insn.Property("foo"), "property: foo @ 0x401000"), (capa.features.insn.Property("foo", "read"), "property/read: foo @ 0x401000"), @@ -202,7 +210,9 @@ def test_render_meta_maec(): def test_render_vverbose_feature(feature, expected): console = Console(highlight=False) - addr = capa.features.freeze.Address.from_capa(capa.features.address.AbsoluteVirtualAddress(0x401000)) + addr = capa.features.freeze.Address.from_capa( + capa.features.address.AbsoluteVirtualAddress(0x401000) + ) feature = capa.features.freeze.features.feature_from_capa(feature) matches = capa.render.result_document.Match( @@ -241,7 +251,9 @@ def test_render_vverbose_feature(feature, expected): ) with console.capture() as capture: - capa.render.vverbose.render_feature(console, layout, rm, matches, feature, indent=0) + capa.render.vverbose.render_feature( + console, layout, rm, matches, feature, indent=0 + ) output = capture.get().strip() assert output == expected