gitea-mirror/capa
1
0
Fork 0
mirror of https://github.com/mandiant/capa.git synced 2026-07-28 14:47:08 -07:00
Code Issues Packages Projects Releases Wiki Activity

style: ruff format changed files

Browse Source
This commit is contained in:
Willi Ballenthin
2026-04-22 19:03:46 +03:00
parent 0c8e44ac7a
commit ccd7e8ecb6
22 changed files with 1400 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files
capa/capabilities/common.py
+40 -12
View File
@@ -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)
capa/features/extractors/base_extractor.py
+34 -10
View File
@@ -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):
capa/features/extractors/common.py
+11 -3
View File
@@ -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
capa/features/extractors/dotnetfile.py
+57 -18
View File
@@ -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"
)
capa/features/extractors/elf.py
+132 -34
View File
@@ -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
capa/features/extractors/elffile.py
+36 -11
View File
@@ -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"
)
capa/features/freeze/__init__.py
+45 -14
View File
@@ -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)
capa/helpers.py
+24 -7
View File
@@ -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 <sample>")
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:
capa/loader.py
+95 -28
View File
@@ -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.
)
capa/main.py
+110 -34
View File
@@ -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
capa/optimizer.py
+16 -2
View File
@@ -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).
#
capa/render/default.py
+21 -7
View File
@@ -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(
capa/render/vverbose.py
+130 -33
View File
@@ -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((
rows.append(
(
"maec/analysis-conclusion",
rule.meta.maec.analysis_conclusion or rule.meta.maec.analysis_conclusion_ov,
))
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((
rows.append(
(
"maec/malware-category",
rule.meta.maec.malware_category or rule.meta.maec.malware_category_ov,
))
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)
capa/rules/__init__.py
+245 -70
View File
@@ -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": <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)
rules
+1 -1
Submodule rules updated: 2af9fbfc1c...03a20f69ae
tests/data
+1 -1
Submodule tests/data updated: f41a1998b9...413fd2803e
tests/test_capabilities.py
+43 -21
View File
@@ -23,7 +23,8 @@ from capa.features.extractors.base_extractor import SampleHashes, FunctionFilter
def test_match_across_scopes_file_function(z9324d_extractor):
rules = capa.rules.RuleSet([
rules = capa.rules.RuleSet(
[
# this rule should match on a function (0x4073F0)
capa.rules.Rule.from_yaml(
textwrap.dedent("""
@@ -76,7 +77,8 @@ def test_match_across_scopes_file_function(z9324d_extractor):
- 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,7 +86,8 @@ def test_match_across_scopes_file_function(z9324d_extractor):
def test_match_across_scopes(z9324d_extractor):
rules = capa.rules.RuleSet([
rules = capa.rules.RuleSet(
[
# this rule should match on a basic block (including at least 0x403685)
capa.rules.Rule.from_yaml(
textwrap.dedent("""
@@ -136,7 +139,8 @@ def test_match_across_scopes(z9324d_extractor):
- 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,7 +148,8 @@ def test_match_across_scopes(z9324d_extractor):
def test_subscope_bb_rules(z9324d_extractor):
rules = capa.rules.RuleSet([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -159,14 +164,16 @@ def test_subscope_bb_rules(z9324d_extractor):
- 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([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -182,7 +189,8 @@ def test_match_specific_functions(z9324d_extractor):
- api: recv
""")
)
])
]
)
extractor = FunctionFilter(z9324d_extractor, {0x4019C0})
capabilities = capa.capabilities.common.find_capabilities(rules, extractor)
matches = capabilities.matches["receive data"]
@@ -193,7 +201,8 @@ def test_match_specific_functions(z9324d_extractor):
def test_byte_matching(z9324d_extractor):
rules = capa.rules.RuleSet([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -207,13 +216,15 @@ def test_byte_matching(z9324d_extractor):
- 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([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -229,13 +240,15 @@ def test_com_feature_matching(z395eb_extractor):
- 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([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -250,14 +263,16 @@ def test_count_bb(z9324d_extractor):
- 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([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -273,7 +288,8 @@ def test_instruction_scope(z9324d_extractor):
- 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,7 +299,8 @@ 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([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -301,10 +318,13 @@ def test_instruction_subscope(z9324d_extractor):
- 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,7 +337,8 @@ def test_find_file_capabilities_preserves_address_zero():
file_features=[(addr, feature)],
functions={},
)
rules = capa.rules.RuleSet([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -330,7 +351,8 @@ def test_find_file_capabilities_preserves_address_zero():
- characteristic: embedded pe
""")
)
])
]
)
caps = capa.capabilities.common.find_file_capabilities(rules, extractor, {})
assert feature in caps.features
assert addr in caps.features[feature]
tests/test_engine.py
+127 -27
View File
@@ -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({
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({
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({
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({
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({
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():
tests/test_freeze_static.py
+57 -17
View File
@@ -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,13 +110,16 @@ 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([
rules = capa.rules.RuleSet(
[
capa.rules.Rule.from_yaml(
textwrap.dedent("""
rule:
@@ -102,7 +135,8 @@ def test_null_feature_extractor():
- 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"))
tests/test_helpers.py
+11 -3
View File
@@ -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
tests/test_os_detection.py
+4 -2
View File
@@ -111,7 +111,8 @@ 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"".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",
@@ -180,7 +181,8 @@ def test_elf_parse_capa_pyinstaller_header():
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"
tests/test_render.py
+16 -4
View File
@@ -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