diff --git a/.github/mypy/mypy.ini b/.github/mypy/mypy.ini index e5d862a7..0aded4dc 100644 --- a/.github/mypy/mypy.ini +++ b/.github/mypy/mypy.ini @@ -57,6 +57,9 @@ ignore_missing_imports = True [mypy-ida_funcs.*] ignore_missing_imports = True +[mypy-ida_loader.*] +ignore_missing_imports = True + [mypy-PyQt5.*] ignore_missing_imports = True diff --git a/CHANGELOG.md b/CHANGELOG.md index 69f05601..65c783eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,9 +8,14 @@ - explorer: enforce max column width Features and Editor panes #691 @mike-hunhoff - explorer: add option to limit features to currently selected disassembly address #692 @mike-hunhoff - all: add support for ELF files #700 @Adir-Shemesh @TcM1911 +- rule format: add feature `format: ` for file format, like `format: pe` #723 @williballenthin +- rule format: add feature `arch: ` for architecture, like `arch: amd64` #723 @williballenthin +- rule format: add feature `os: ` for operating system, like `os: windows` #723 @williballenthin ### Breaking Changes +- legacy term `arch` (i.e., "x32") is now called `bitness` @williballenthin + ### New Rules (24) - collection/webcam/capture-webcam-image johnk3r diff --git a/capa/features/common.py b/capa/features/common.py index 221f15c1..8f16d6e5 100644 --- a/capa/features/common.py +++ b/capa/features/common.py @@ -14,6 +14,7 @@ from typing import Set, Dict, Union import capa.engine import capa.features +import capa.features.extractors.elf logger = logging.getLogger(__name__) MAX_BYTES_FEATURE_SIZE = 0x100 @@ -21,12 +22,6 @@ MAX_BYTES_FEATURE_SIZE = 0x100 # thunks may be chained so we specify a delta to control the depth to which these chains are explored THUNK_CHAIN_DEPTH_DELTA = 5 -# identifiers for supported architectures names that tweak a feature -# for example, offset/x32 -ARCH_X32 = "x32" -ARCH_X64 = "x64" -VALID_ARCH = (ARCH_X32, ARCH_X64) - def bytes_to_str(b: bytes) -> str: return str(codecs.encode(b, "hex").decode("utf-8")) @@ -52,33 +47,33 @@ def escape_string(s: str) -> str: class Feature: - def __init__(self, value: Union[str, int, bytes], arch=None, description=None): + def __init__(self, value: Union[str, int, bytes], bitness=None, description=None): """ Args: value (any): the value of the feature, such as the number or string. - arch (str): one of the VALID_ARCH values, or None. - When None, then the feature applies to any architecture. - Modifies the feature name from `feature` to `feature/arch`, like `offset/x32`. + bitness (str): one of the VALID_BITNESS values, or None. + When None, then the feature applies to any bitness. + Modifies the feature name from `feature` to `feature/bitness`, like `offset/x32`. description (str): a human-readable description that explains the feature value. """ super(Feature, self).__init__() - if arch is not None: - if arch not in VALID_ARCH: - raise ValueError("arch '%s' must be one of %s" % (arch, VALID_ARCH)) - self.name = self.__class__.__name__.lower() + "/" + arch + if bitness is not None: + if bitness not in VALID_BITNESS: + raise ValueError("bitness '%s' must be one of %s" % (bitness, VALID_BITNESS)) + self.name = self.__class__.__name__.lower() + "/" + bitness else: self.name = self.__class__.__name__.lower() self.value = value - self.arch = arch + self.bitness = bitness self.description = description def __hash__(self): - return hash((self.name, self.value, self.arch)) + return hash((self.name, self.value, self.bitness)) def __eq__(self, other): - return self.name == other.name and self.value == other.value and self.arch == other.arch + return self.name == other.name and self.value == other.value and self.bitness == other.bitness def get_value_str(self) -> str: """ @@ -105,8 +100,8 @@ class Feature: return capa.engine.Result(self in ctx, self, [], locations=ctx.get(self, [])) def freeze_serialize(self): - if self.arch is not None: - return (self.__class__.__name__, [self.value, {"arch": self.arch}]) + if self.bitness is not None: + return (self.__class__.__name__, [self.value, {"bitness": self.bitness}]) else: return (self.__class__.__name__, [self.value]) @@ -131,6 +126,7 @@ class MatchedRule(Feature): class Characteristic(Feature): def __init__(self, value: str, description=None): + super(Characteristic, self).__init__(value, description=description) @@ -260,3 +256,57 @@ class Bytes(Feature): @classmethod def freeze_deserialize(cls, args): return cls(*[codecs.decode(x, "hex") for x in args]) + + +# identifiers for supported bitness names that tweak a feature +# for example, offset/x32 +BITNESS_X32 = "x32" +BITNESS_X64 = "x64" +VALID_BITNESS = (BITNESS_X32, BITNESS_X64) + + +# other candidates here: https://docs.microsoft.com/en-us/windows/win32/debug/pe-format#machine-types +ARCH_I386 = "i386" +ARCH_AMD64 = "amd64" +VALID_ARCH = (ARCH_I386, ARCH_AMD64) + + +class Arch(Feature): + def __init__(self, value: str, description=None): + assert value in VALID_ARCH + super(Arch, self).__init__(value, description=description) + self.name = "arch" + + +OS_WINDOWS = "windows" +OS_LINUX = "linux" +OS_MACOS = "macos" +VALID_OS = {os.value for os in capa.features.extractors.elf.OS} +VALID_OS.update({OS_WINDOWS, OS_LINUX, OS_MACOS}) + + +class OS(Feature): + def __init__(self, value: str, description=None): + assert value in (VALID_OS) + super(OS, self).__init__(value, description=description) + self.name = "os" + + +FORMAT_PE = "pe" +FORMAT_ELF = "elf" +VALID_FORMAT = (FORMAT_PE, FORMAT_ELF) + + +class Format(Feature): + def __init__(self, value: str, description=None): + assert value in (VALID_FORMAT) + super(Format, self).__init__(value, description=description) + self.name = "format" + + +def is_global_feature(feature): + """ + is this a feature that is extracted at every scope? + today, these are OS and arch features. + """ + return isinstance(feature, (OS, Arch)) diff --git a/capa/features/extractors/common.py b/capa/features/extractors/common.py new file mode 100644 index 00000000..450fb636 --- /dev/null +++ b/capa/features/extractors/common.py @@ -0,0 +1,78 @@ +import io +import logging +import binascii +import contextlib + +import pefile + +import capa.features.extractors.elf +import capa.features.extractors.pefile +from capa.features.common import OS, FORMAT_PE, FORMAT_ELF, OS_WINDOWS, Arch, Format + +logger = logging.getLogger(__name__) + + +def extract_format(buf): + if buf.startswith(b"MZ"): + yield Format(FORMAT_PE), 0x0 + elif buf.startswith(b"\x7fELF"): + yield Format(FORMAT_ELF), 0x0 + else: + # we likely end up here: + # 1. handling a file format (e.g. macho) + # + # for (1), this logic will need to be updated as the format is implemented. + logger.debug("unsupported file format: %s", binascii.hexlify(buf[:4]).decode("ascii")) + return + + +def extract_arch(buf): + if buf.startswith(b"MZ"): + yield from capa.features.extractors.pefile.extract_file_arch(pe=pefile.PE(data=buf)) + + elif buf.startswith(b"\x7fELF"): + with contextlib.closing(io.BytesIO(buf)) as f: + arch = capa.features.extractors.elf.detect_elf_arch(f) + + if arch == "unknown": + logger.debug("unsupported arch: %s", arch) + return + + yield Arch(arch), 0x0 + + else: + # we likely end up here: + # 1. handling shellcode, or + # 2. handling a new file format (e.g. macho) + # + # for (1) we can't do much - its shellcode and all bets are off. + # we could maybe accept a futher CLI argument to specify the arch, + # but i think this would be rarely used. + # 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")) + return + + +def extract_os(buf): + if buf.startswith(b"MZ"): + yield OS(OS_WINDOWS), 0x0 + elif buf.startswith(b"\x7fELF"): + with contextlib.closing(io.BytesIO(buf)) as f: + os = capa.features.extractors.elf.detect_elf_os(f) + + yield OS(os), 0x0 + else: + # we likely end up here: + # 1. handling shellcode, or + # 2. handling a new file format (e.g. macho) + # + # for (1) we can't do much - its shellcode and all bets are off. + # we could maybe accept a futher CLI argument to specify the OS, + # but i think this would be rarely used. + # 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")) + return diff --git a/capa/features/extractors/elf.py b/capa/features/extractors/elf.py new file mode 100644 index 00000000..ce3fe638 --- /dev/null +++ b/capa/features/extractors/elf.py @@ -0,0 +1,269 @@ +import struct +import logging +from enum import Enum +from typing import BinaryIO + +logger = logging.getLogger(__name__) + + +def align(v, alignment): + remainder = v % alignment + if remainder == 0: + return v + else: + return v + (alignment - remainder) + + +class CorruptElfFile(ValueError): + pass + + +class OS(str, Enum): + HPUX = "hpux" + NETBSD = "netbsd" + LINUX = "linux" + HURD = "hurd" + _86OPEN = "86open" + SOLARIS = "solaris" + AIX = "aix" + IRIX = "irix" + FREEBSD = "freebsd" + TRU64 = "tru64" + MODESTO = "modesto" + OPENBSD = "openbsd" + OPENVMS = "openvms" + NSK = "nsk" + AROS = "aros" + FENIXOS = "fenixos" + CLOUD = "cloud" + SYLLABLE = "syllable" + NACL = "nacl" + + +def detect_elf_os(f: BinaryIO) -> str: + f.seek(0x0) + file_header = f.read(0x40) + + # we'll set this to the detected OS + # prefer the first heuristics, + # but rather than short circuiting, + # we'll still parse out the remainder, for debugging. + ret = None + + if not file_header.startswith(b"\x7fELF"): + raise CorruptElfFile("missing magic header") + + ei_class, ei_data = struct.unpack_from("BB", file_header, 4) + logger.debug("ei_class: 0x%02x ei_data: 0x%02x", ei_class, ei_data) + if ei_class == 1: + bitness = 32 + elif ei_class == 2: + bitness = 64 + else: + raise CorruptElfFile("invalid ei_class: 0x%02x" % ei_class) + + if ei_data == 1: + endian = "<" + elif ei_data == 2: + endian = ">" + else: + raise CorruptElfFile("not an ELF file: invalid ei_data: 0x%02x" % ei_data) + + if bitness == 32: + (e_phoff,) = struct.unpack_from(endian + "I", file_header, 0x1C) + e_phentsize, e_phnum = struct.unpack_from(endian + "HH", file_header, 0x2A) + elif bitness == 64: + (e_phoff,) = struct.unpack_from(endian + "Q", file_header, 0x20) + e_phentsize, e_phnum = struct.unpack_from(endian + "HH", file_header, 0x36) + else: + raise NotImplemented + + logger.debug("e_phoff: 0x%02x e_phentsize: 0x%02x e_phnum: %d", e_phoff, e_phentsize, e_phnum) + + (ei_osabi,) = struct.unpack_from(endian + "B", file_header, 7) + OSABI = { + # via pyelftools: https://github.com/eliben/pyelftools/blob/0664de05ed2db3d39041e2d51d19622a8ef4fb0f/elftools/elf/enums.py#L35-L58 + # some candidates are commented out because the are not useful values, + # at least when guessing OSes + # 0: "SYSV", # too often used when OS is not SYSV + 1: OS.HPUX, + 2: OS.NETBSD, + 3: OS.LINUX, + 4: OS.HURD, + 5: OS._86OPEN, + 6: OS.SOLARIS, + 7: OS.AIX, + 8: OS.IRIX, + 9: OS.FREEBSD, + 10: OS.TRU64, + 11: OS.MODESTO, + 12: OS.OPENBSD, + 13: OS.OPENVMS, + 14: OS.NSK, + 15: OS.AROS, + 16: OS.FENIXOS, + 17: OS.CLOUD, + # 53: "SORTFIX", # i can't find any reference to this OS, i dont think it exists + # 64: "ARM_AEABI", # not an OS + # 97: "ARM", # not an OS + # 255: "STANDALONE", # not an OS + } + logger.debug("ei_osabi: 0x%02x (%s)", ei_osabi, OSABI.get(ei_osabi, "unknown")) + + # os_osabi == 0 is commonly set even when the OS is not SYSV. + # other values are unused or unknown. + if ei_osabi in OSABI and ei_osabi != 0x0: + # subsequent strategies may overwrite this value + ret = OSABI[ei_osabi] + + f.seek(e_phoff) + program_header_size = e_phnum * e_phentsize + program_headers = f.read(program_header_size) + if len(program_headers) != program_header_size: + logger.warning("failed to read program headers") + e_phnum = 0 + + # search for PT_NOTE sections that specify an OS + # for example, on Linux there is a GNU section with minimum kernel version + for i in range(e_phnum): + offset = i * e_phentsize + phent = program_headers[offset : offset + e_phentsize] + + PT_NOTE = 0x4 + + (p_type,) = struct.unpack_from(endian + "I", phent, 0x0) + logger.debug("p_type: 0x%04x", p_type) + if p_type != PT_NOTE: + continue + + if bitness == 32: + p_offset, _, _, p_filesz = struct.unpack_from(endian + "IIII", phent, 0x4) + elif bitness == 64: + p_offset, _, _, p_filesz = struct.unpack_from(endian + "QQQQ", phent, 0x8) + else: + raise NotImplemented + + logger.debug("p_offset: 0x%02x p_filesz: 0x%04x", p_offset, p_filesz) + + f.seek(p_offset) + note = f.read(p_filesz) + if len(note) != p_filesz: + logger.warning("failed to read note content") + continue + + namesz, descsz, type_ = struct.unpack_from(endian + "III", note, 0x0) + name_offset = 0xC + desc_offset = name_offset + align(namesz, 0x4) + + logger.debug("namesz: 0x%02x descsz: 0x%02x type: 0x%04x", namesz, descsz, type_) + + name = note[name_offset : name_offset + namesz].partition(b"\x00")[0].decode("ascii") + logger.debug("name: %s", name) + + if type_ != 1: + continue + + if name == "GNU": + if descsz < 16: + continue + + desc = note[desc_offset : desc_offset + descsz] + abi_tag, kmajor, kminor, kpatch = struct.unpack_from(endian + "IIII", desc, 0x0) + # via readelf: https://github.com/bminor/binutils-gdb/blob/c0e94211e1ac05049a4ce7c192c9d14d1764eb3e/binutils/readelf.c#L19635-L19658 + # and here: https://github.com/bminor/binutils-gdb/blob/34c54daa337da9fadf87d2706d6a590ae1f88f4d/include/elf/common.h#L933-L939 + GNU_ABI_TAG = { + 0: OS.LINUX, + 1: OS.HURD, + 2: OS.SOLARIS, + 3: OS.FREEBSD, + 4: OS.NETBSD, + 5: OS.SYLLABLE, + 6: OS.NACL, + } + logger.debug("GNU_ABI_TAG: 0x%02x", abi_tag) + + if abi_tag in GNU_ABI_TAG: + # update only if not set + # so we can get the debugging output of subsequent strategies + ret = GNU_ABI_TAG[abi_tag] if not ret else ret + logger.debug("abi tag: %s earliest compatible kernel: %d.%d.%d", ret, kmajor, kminor, kpatch) + elif name == "OpenBSD": + logger.debug("note owner: %s", "OPENBSD") + ret = OS.OPENBSD if not ret else ret + elif name == "NetBSD": + logger.debug("note owner: %s", "NETBSD") + ret = OS.NETBSD if not ret else ret + elif name == "FreeBSD": + logger.debug("note owner: %s", "FREEBSD") + ret = OS.FREEBSD if not ret else ret + + # search for recognizable dynamic linkers (interpreters) + # for example, on linux, we see file paths like: /lib64/ld-linux-x86-64.so.2 + for i in range(e_phnum): + offset = i * e_phentsize + phent = program_headers[offset : offset + e_phentsize] + + PT_INTERP = 0x3 + + (p_type,) = struct.unpack_from(endian + "I", phent, 0x0) + if p_type != PT_INTERP: + continue + + if bitness == 32: + p_offset, _, _, p_filesz = struct.unpack_from(endian + "IIII", phent, 0x4) + elif bitness == 64: + p_offset, _, _, p_filesz = struct.unpack_from(endian + "QQQQ", phent, 0x8) + else: + raise NotImplemented + + f.seek(p_offset) + interp = f.read(p_filesz) + if len(interp) != p_filesz: + logger.warning("failed to read interp content") + continue + + linker = interp.partition(b"\x00")[0].decode("ascii") + logger.debug("linker: %s", linker) + if "ld-linux" in linker: + # update only if not set + # so we can get the debugging output of subsequent strategies + ret = OS.LINUX if ret is None else ret + + return ret.value if ret is not None else "unknown" + + +class Arch(str, Enum): + I386 = "i386" + AMD64 = "amd64" + + +def detect_elf_arch(f: BinaryIO) -> str: + f.seek(0x0) + file_header = f.read(0x40) + + if not file_header.startswith(b"\x7fELF"): + raise CorruptElfFile("missing magic header") + + (ei_data,) = struct.unpack_from("B", file_header, 5) + logger.debug("ei_data: 0x%02x", ei_data) + + if ei_data == 1: + endian = "<" + elif ei_data == 2: + endian = ">" + else: + raise CorruptElfFile("not an ELF file: invalid ei_data: 0x%02x" % ei_data) + + (ei_machine,) = struct.unpack_from(endian + "H", file_header, 0x12) + logger.debug("ei_machine: 0x%02x", ei_machine) + + EM_386 = 0x3 + EM_X86_64 = 0x3E + if ei_machine == EM_386: + return Arch.I386 + elif ei_machine == EM_X86_64: + return Arch.AMD64 + else: + # not really unknown, but unsupport at the moment: + # https://github.com/eliben/pyelftools/blob/ab444d982d1849191e910299a985989857466620/elftools/elf/enums.py#L73 + return "unknown" diff --git a/capa/features/extractors/ida/extractor.py b/capa/features/extractors/ida/extractor.py index dc260adc..7d00c7ef 100644 --- a/capa/features/extractors/ida/extractor.py +++ b/capa/features/extractors/ida/extractor.py @@ -5,11 +5,13 @@ # Unless required by applicable law or agreed to in writing, software distributed under the License # is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and limitations under the License. - import idaapi +import capa.ida.helpers +import capa.features.extractors.elf import capa.features.extractors.ida.file import capa.features.extractors.ida.insn +import capa.features.extractors.ida.global_ import capa.features.extractors.ida.function import capa.features.extractors.ida.basicblock from capa.features.extractors.base_extractor import FeatureExtractor @@ -57,6 +59,9 @@ class InstructionHandle: class IdaFeatureExtractor(FeatureExtractor): def __init__(self): super(IdaFeatureExtractor, self).__init__() + self.global_features = [] + self.global_features.extend(capa.features.extractors.ida.global_.extract_os()) + self.global_features.extend(capa.features.extractors.ida.global_.extract_arch()) def get_base_address(self): return idaapi.get_imagebase() @@ -64,6 +69,7 @@ class IdaFeatureExtractor(FeatureExtractor): def extract_file_features(self): for (feature, ea) in capa.features.extractors.ida.file.extract_features(): yield feature, ea + yield from self.global_features def get_functions(self): import capa.features.extractors.ida.helpers as ida_helpers @@ -86,6 +92,7 @@ class IdaFeatureExtractor(FeatureExtractor): def extract_function_features(self, f): for (feature, ea) in capa.features.extractors.ida.function.extract_features(f): yield feature, ea + yield from self.global_features def get_basic_blocks(self, f): import capa.features.extractors.ida.helpers as ida_helpers @@ -96,6 +103,7 @@ class IdaFeatureExtractor(FeatureExtractor): def extract_basic_block_features(self, f, bb): for (feature, ea) in capa.features.extractors.ida.basicblock.extract_features(f, bb): yield feature, ea + yield from self.global_features def get_instructions(self, f, bb): import capa.features.extractors.ida.helpers as ida_helpers @@ -106,3 +114,4 @@ class IdaFeatureExtractor(FeatureExtractor): def extract_insn_features(self, f, bb, insn): for (feature, ea) in capa.features.extractors.ida.insn.extract_features(f, bb, insn): yield feature, ea + yield from self.global_features diff --git a/capa/features/extractors/ida/file.py b/capa/features/extractors/ida/file.py index d3f6a101..3fa72351 100644 --- a/capa/features/extractors/ida/file.py +++ b/capa/features/extractors/ida/file.py @@ -11,12 +11,13 @@ import struct import idc import idaapi import idautils +import ida_loader import capa.features.extractors.helpers import capa.features.extractors.strings import capa.features.extractors.ida.helpers from capa.features.file import Export, Import, Section, FunctionName -from capa.features.common import String, Characteristic +from capa.features.common import OS, FORMAT_PE, FORMAT_ELF, OS_WINDOWS, Format, String, Characteristic def check_segment_for_pe(seg): @@ -153,6 +154,19 @@ def extract_file_function_names(): yield FunctionName(name), ea +def extract_file_format(): + format_name = ida_loader.get_file_type_name() + + if "PE" in format_name: + yield Format(FORMAT_PE), 0x0 + elif "ELF64" in format_name: + yield Format(FORMAT_ELF), 0x0 + elif "ELF32" in format_name: + yield Format(FORMAT_ELF), 0x0 + else: + raise NotImplementedError("file format: %s", format_name) + + def extract_features(): """extract file features""" for file_handler in FILE_HANDLERS: @@ -167,6 +181,7 @@ FILE_HANDLERS = ( extract_file_section_names, extract_file_embedded_pe, extract_file_function_names, + extract_file_format, ) diff --git a/capa/features/extractors/ida/global_.py b/capa/features/extractors/ida/global_.py new file mode 100644 index 00000000..b3181293 --- /dev/null +++ b/capa/features/extractors/ida/global_.py @@ -0,0 +1,56 @@ +import logging +import contextlib + +import idaapi +import ida_loader + +import capa.ida.helpers +import capa.features.extractors.elf +from capa.features.common import OS, ARCH_I386, ARCH_AMD64, OS_WINDOWS, Arch + +logger = logging.getLogger(__name__) + + +def extract_os(): + format_name = ida_loader.get_file_type_name() + + if "PE" in format_name: + yield OS(OS_WINDOWS), 0x0 + + elif "ELF" in format_name: + with contextlib.closing(capa.ida.helpers.IDAIO()) as f: + os = capa.features.extractors.elf.detect_elf_os(f) + + yield OS(os), 0x0 + + else: + # we likely end up here: + # 1. handling shellcode, or + # 2. handling a new file format (e.g. macho) + # + # for (1) we can't do much - its shellcode and all bets are off. + # we could maybe accept a futher CLI argument to specify the OS, + # but i think this would be rarely used. + # 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", format_name) + return + + +def extract_arch(): + info = idaapi.get_inf_structure() + if info.procName == "metapc" and info.is_64bit(): + yield Arch(ARCH_AMD64), 0x0 + elif info.procName == "metapc" and info.is_32bit(): + yield Arch(ARCH_I386), 0x0 + elif info.procName == "metapc": + logger.debug("unsupported architecture: non-32-bit nor non-64-bit intel") + return + else: + # we likely end up here: + # 1. handling a new architecture (e.g. aarch64) + # + # for (1), this logic will need to be updated as the format is implemented. + logger.debug("unsupported architecture: %s", info.procName) + return diff --git a/capa/features/extractors/ida/insn.py b/capa/features/extractors/ida/insn.py index 82fb4715..8bfcd7fb 100644 --- a/capa/features/extractors/ida/insn.py +++ b/capa/features/extractors/ida/insn.py @@ -14,8 +14,8 @@ import capa.features.extractors.helpers import capa.features.extractors.ida.helpers from capa.features.insn import API, Number, Offset, Mnemonic from capa.features.common import ( - ARCH_X32, - ARCH_X64, + BITNESS_X32, + BITNESS_X64, MAX_BYTES_FEATURE_SIZE, THUNK_CHAIN_DEPTH_DELTA, Bytes, @@ -28,22 +28,22 @@ from capa.features.common import ( SECURITY_COOKIE_BYTES_DELTA = 0x40 -def get_arch(ctx): +def get_bitness(ctx): """ - fetch the ARCH_* constant for the currently open workspace. + fetch the BITNESS_* constant for the currently open workspace. via Tamir Bahar/@tmr232 https://reverseengineering.stackexchange.com/a/11398/17194 """ - if "arch" not in ctx: + if "bitness" not in ctx: info = idaapi.get_inf_structure() if info.is_64bit(): - ctx["arch"] = ARCH_X64 + ctx["bitness"] = BITNESS_X64 elif info.is_32bit(): - ctx["arch"] = ARCH_X32 + ctx["bitness"] = BITNESS_X32 else: - raise ValueError("unexpected architecture") - return ctx["arch"] + raise ValueError("unexpected bitness") + return ctx["bitness"] def get_imports(ctx): @@ -149,7 +149,7 @@ def extract_insn_number_features(f, bb, insn): const = op.addr yield Number(const), insn.ea - yield Number(const, arch=get_arch(f.ctx)), insn.ea + yield Number(const, bitness=get_bitness(f.ctx)), insn.ea def extract_insn_bytes_features(f, bb, insn): @@ -218,7 +218,7 @@ def extract_insn_offset_features(f, bb, insn): op_off = capa.features.extractors.helpers.twos_complement(op_off, 32) yield Offset(op_off), insn.ea - yield Offset(op_off, arch=get_arch(f.ctx)), insn.ea + yield Offset(op_off, bitness=get_bitness(f.ctx)), insn.ea def contains_stack_cookie_keywords(s): diff --git a/capa/features/extractors/pefile.py b/capa/features/extractors/pefile.py index e27d6c20..3e5d97fd 100644 --- a/capa/features/extractors/pefile.py +++ b/capa/features/extractors/pefile.py @@ -9,25 +9,23 @@ import logging import pefile +import capa.features.common import capa.features.extractors import capa.features.extractors.helpers import capa.features.extractors.strings from capa.features.file import Export, Import, Section -from capa.features.common import String, Characteristic +from capa.features.common import OS, ARCH_I386, FORMAT_PE, ARCH_AMD64, OS_WINDOWS, Arch, Format, String, Characteristic from capa.features.extractors.base_extractor import FeatureExtractor logger = logging.getLogger(__name__) -def extract_file_embedded_pe(pe, file_path): - with open(file_path, "rb") as f: - fbytes = f.read() - - for offset, i in capa.features.extractors.helpers.carve_pe(fbytes, 1): +def extract_file_embedded_pe(buf, **kwargs): + for offset, _ in capa.features.extractors.helpers.carve_pe(buf, 1): yield Characteristic("embedded pe"), offset -def extract_file_export_names(pe, file_path): +def extract_file_export_names(pe, **kwargs): base_address = pe.OPTIONAL_HEADER.ImageBase if hasattr(pe, "DIRECTORY_ENTRY_EXPORT"): @@ -42,7 +40,7 @@ def extract_file_export_names(pe, file_path): yield Export(name), va -def extract_file_import_names(pe, file_path): +def extract_file_import_names(pe, **kwargs): """ extract imported function names 1. imports by ordinal: @@ -74,7 +72,7 @@ def extract_file_import_names(pe, file_path): yield Import(name), imp.address -def extract_file_section_names(pe, file_path): +def extract_file_section_names(pe, **kwargs): base_address = pe.OPTIONAL_HEADER.ImageBase for section in pe.sections: @@ -86,21 +84,18 @@ def extract_file_section_names(pe, file_path): yield Section(name), base_address + section.VirtualAddress -def extract_file_strings(pe, file_path): +def extract_file_strings(buf, **kwargs): """ extract ASCII and UTF-16 LE strings from file """ - with open(file_path, "rb") as f: - b = f.read() - - for s in capa.features.extractors.strings.extract_ascii_strings(b): + for s in capa.features.extractors.strings.extract_ascii_strings(buf): yield String(s.s), s.offset - for s in capa.features.extractors.strings.extract_unicode_strings(b): + for s in capa.features.extractors.strings.extract_unicode_strings(buf): yield String(s.s), s.offset -def extract_file_function_names(pe, file_path): +def extract_file_function_names(**kwargs): """ extract the names of statically-linked library functions. """ @@ -110,20 +105,39 @@ def extract_file_function_names(pe, file_path): return -def extract_file_features(pe, file_path): +def extract_file_os(**kwargs): + # assuming PE -> Windows + # though i suppose they're also used by UEFI + yield OS(OS_WINDOWS), 0x0 + + +def extract_file_format(**kwargs): + yield Format(FORMAT_PE), 0x0 + + +def extract_file_arch(pe, **kwargs): + if pe.FILE_HEADER.Machine == pefile.MACHINE_TYPE["IMAGE_FILE_MACHINE_I386"]: + yield Arch(ARCH_I386), 0x0 + elif pe.FILE_HEADER.Machine == pefile.MACHINE_TYPE["IMAGE_FILE_MACHINE_AMD64"]: + yield Arch(ARCH_AMD64), 0x0 + else: + logger.warning("unsupported architecture: %s", pefile.MACHINE_TYPE[pe.FILE_HEADER.Machine]) + + +def extract_file_features(pe, buf): """ extract file features from given workspace args: pe (pefile.PE): the parsed PE - file_path: path to the input file + buf: the raw sample bytes yields: Tuple[Feature, VA]: a feature and its location. """ for file_handler in FILE_HANDLERS: - for feature, va in file_handler(pe, file_path): + for feature, va in file_handler(pe=pe, buf=buf): yield feature, va @@ -134,6 +148,9 @@ FILE_HANDLERS = ( extract_file_section_names, extract_file_strings, extract_file_function_names, + extract_file_os, + extract_file_format, + extract_file_arch, ) @@ -147,7 +164,10 @@ class PefileFeatureExtractor(FeatureExtractor): return self.pe.OPTIONAL_HEADER.ImageBase def extract_file_features(self): - for feature, va in extract_file_features(self.pe, self.path): + with open(self.path, "rb") as f: + buf = f.read() + + for feature, va in extract_file_features(self.pe, buf): yield feature, va def get_functions(self): diff --git a/capa/features/extractors/smda/extractor.py b/capa/features/extractors/smda/extractor.py index b4355d8f..1a13653b 100644 --- a/capa/features/extractors/smda/extractor.py +++ b/capa/features/extractors/smda/extractor.py @@ -1,7 +1,9 @@ from smda.common.SmdaReport import SmdaReport +import capa.features.extractors.common import capa.features.extractors.smda.file import capa.features.extractors.smda.insn +import capa.features.extractors.smda.global_ import capa.features.extractors.smda.function import capa.features.extractors.smda.basicblock from capa.features.extractors.base_extractor import FeatureExtractor @@ -12,13 +14,21 @@ class SmdaFeatureExtractor(FeatureExtractor): super(SmdaFeatureExtractor, self).__init__() self.smda_report = smda_report self.path = path + with open(self.path, "rb") as f: + self.buf = f.read() + + # pre-compute these because we'll yield them at *every* scope. + self.global_features = [] + self.global_features.extend(capa.features.extractors.common.extract_os(self.buf)) + self.global_features.extend(capa.features.extractors.smda.global_.extract_arch(self.smda_report)) def get_base_address(self): return self.smda_report.base_addr def extract_file_features(self): - for feature, va in capa.features.extractors.smda.file.extract_features(self.smda_report, self.path): + for feature, va in capa.features.extractors.smda.file.extract_features(self.smda_report, self.buf): yield feature, va + yield from self.global_features def get_functions(self): for function in self.smda_report.getFunctions(): @@ -27,6 +37,7 @@ class SmdaFeatureExtractor(FeatureExtractor): def extract_function_features(self, f): for feature, va in capa.features.extractors.smda.function.extract_features(f): yield feature, va + yield from self.global_features def get_basic_blocks(self, f): for bb in f.getBlocks(): @@ -35,6 +46,7 @@ class SmdaFeatureExtractor(FeatureExtractor): def extract_basic_block_features(self, f, bb): for feature, va in capa.features.extractors.smda.basicblock.extract_features(f, bb): yield feature, va + yield from self.global_features def get_instructions(self, f, bb): for smda_ins in bb.getInstructions(): @@ -43,3 +55,4 @@ class SmdaFeatureExtractor(FeatureExtractor): def extract_insn_features(self, f, bb, insn): for feature, va in capa.features.extractors.smda.insn.extract_features(f, bb, insn): yield feature, va + yield from self.global_features diff --git a/capa/features/extractors/smda/file.py b/capa/features/extractors/smda/file.py index 5250e26b..fe901bea 100644 --- a/capa/features/extractors/smda/file.py +++ b/capa/features/extractors/smda/file.py @@ -1,30 +1,29 @@ # if we have SMDA we definitely have lief import lief +import capa.features.extractors.common import capa.features.extractors.helpers import capa.features.extractors.strings from capa.features.file import Export, Import, Section from capa.features.common import String, Characteristic -def extract_file_embedded_pe(smda_report, file_path): - with open(file_path, "rb") as f: - fbytes = f.read() - - for offset, i in capa.features.extractors.helpers.carve_pe(fbytes, 1): +def extract_file_embedded_pe(buf, **kwargs): + for offset, _ in capa.features.extractors.helpers.carve_pe(buf, 1): yield Characteristic("embedded pe"), offset -def extract_file_export_names(smda_report, file_path): - lief_binary = lief.parse(file_path) +def extract_file_export_names(buf, **kwargs): + lief_binary = lief.parse(buf) + if lief_binary is not None: for function in lief_binary.exported_functions: yield Export(function.name), function.address -def extract_file_import_names(smda_report, file_path): +def extract_file_import_names(smda_report, buf): # extract import table info via LIEF - lief_binary = lief.parse(file_path) + lief_binary = lief.parse(buf) if not isinstance(lief_binary, lief.PE.Binary): return for imported_library in lief_binary.imports: @@ -40,8 +39,8 @@ def extract_file_import_names(smda_report, file_path): yield Import(name), va -def extract_file_section_names(smda_report, file_path): - lief_binary = lief.parse(file_path) +def extract_file_section_names(buf, **kwargs): + lief_binary = lief.parse(buf) if not isinstance(lief_binary, lief.PE.Binary): return if lief_binary and lief_binary.sections: @@ -50,21 +49,18 @@ def extract_file_section_names(smda_report, file_path): yield Section(section.name), base_address + section.virtual_address -def extract_file_strings(smda_report, file_path): +def extract_file_strings(buf, **kwargs): """ extract ASCII and UTF-16 LE strings from file """ - with open(file_path, "rb") as f: - b = f.read() - - for s in capa.features.extractors.strings.extract_ascii_strings(b): + for s in capa.features.extractors.strings.extract_ascii_strings(buf): yield String(s.s), s.offset - for s in capa.features.extractors.strings.extract_unicode_strings(b): + for s in capa.features.extractors.strings.extract_unicode_strings(buf): yield String(s.s), s.offset -def extract_file_function_names(smda_report, file_path): +def extract_file_function_names(smda_report, **kwargs): """ extract the names of statically-linked library functions. """ @@ -74,20 +70,24 @@ def extract_file_function_names(smda_report, file_path): return -def extract_features(smda_report, file_path): +def extract_file_format(buf, **kwargs): + yield from capa.features.extractors.common.extract_format(buf) + + +def extract_features(smda_report, buf): """ extract file features from given workspace args: smda_report (smda.common.SmdaReport): a SmdaReport - file_path: path to the input file + buf: the raw bytes of the sample yields: Tuple[Feature, VA]: a feature and its location. """ for file_handler in FILE_HANDLERS: - for feature, va in file_handler(smda_report, file_path): + for feature, va in file_handler(smda_report=smda_report, buf=buf): yield feature, va @@ -98,4 +98,5 @@ FILE_HANDLERS = ( extract_file_section_names, extract_file_strings, extract_file_function_names, + extract_file_format, ) diff --git a/capa/features/extractors/smda/global_.py b/capa/features/extractors/smda/global_.py new file mode 100644 index 00000000..84befc59 --- /dev/null +++ b/capa/features/extractors/smda/global_.py @@ -0,0 +1,20 @@ +import logging + +from capa.features.common import ARCH_I386, ARCH_AMD64, Arch + +logger = logging.getLogger(__name__) + + +def extract_arch(smda_report): + if smda_report.architecture == "intel": + if smda_report.bitness == 32: + yield Arch(ARCH_I386), 0x0 + elif smda_report.bitness == 64: + yield Arch(ARCH_AMD64), 0x0 + else: + # we likely end up here: + # 1. handling a new architecture (e.g. aarch64) + # + # for (1), this logic will need to be updated as the format is implemented. + logger.debug("unsupported architecture: %s", smda_report.architecture) + return diff --git a/capa/features/extractors/smda/insn.py b/capa/features/extractors/smda/insn.py index da5ebbfa..d666728c 100644 --- a/capa/features/extractors/smda/insn.py +++ b/capa/features/extractors/smda/insn.py @@ -7,8 +7,8 @@ from smda.common.SmdaReport import SmdaReport import capa.features.extractors.helpers from capa.features.insn import API, Number, Offset, Mnemonic from capa.features.common import ( - ARCH_X32, - ARCH_X64, + BITNESS_X32, + BITNESS_X64, MAX_BYTES_FEATURE_SIZE, THUNK_CHAIN_DEPTH_DELTA, Bytes, @@ -23,12 +23,12 @@ PATTERN_HEXNUM = re.compile(r"[+\-] (?P0x[a-fA-F0-9]+)") PATTERN_SINGLENUM = re.compile(r"[+\-] (?P[0-9])") -def get_arch(smda_report): +def get_bitness(smda_report): if smda_report.architecture == "intel": if smda_report.bitness == 32: - return ARCH_X32 + return BITNESS_X32 elif smda_report.bitness == 64: - return ARCH_X64 + return BITNESS_X64 else: raise NotImplementedError @@ -85,7 +85,7 @@ def extract_insn_number_features(f, bb, insn): for operand in operands: try: yield Number(int(operand, 16)), insn.offset - yield Number(int(operand, 16), arch=get_arch(f.smda_report)), insn.offset + yield Number(int(operand, 16), bitness=get_bitness(f.smda_report)), insn.offset except: continue @@ -228,7 +228,7 @@ def extract_insn_offset_features(f, bb, insn): number = int(number_int.group("num")) number = -1 * number if number_int.group().startswith("-") else number yield Offset(number), insn.offset - yield Offset(number, arch=get_arch(f.smda_report)), insn.offset + yield Offset(number, bitness=get_bitness(f.smda_report)), insn.offset def is_security_cookie(f, bb, insn): diff --git a/capa/features/extractors/viv/extractor.py b/capa/features/extractors/viv/extractor.py index 15f3973a..b18d4349 100644 --- a/capa/features/extractors/viv/extractor.py +++ b/capa/features/extractors/viv/extractor.py @@ -10,8 +10,10 @@ import logging import viv_utils import viv_utils.flirt +import capa.features.extractors.common import capa.features.extractors.viv.file import capa.features.extractors.viv.insn +import capa.features.extractors.viv.global_ import capa.features.extractors.viv.function import capa.features.extractors.viv.basicblock from capa.features.extractors.base_extractor import FeatureExtractor @@ -37,14 +39,22 @@ class VivisectFeatureExtractor(FeatureExtractor): super(VivisectFeatureExtractor, self).__init__() self.vw = vw self.path = path + with open(self.path, "rb") as f: + self.buf = f.read() + + # pre-compute these because we'll yield them at *every* scope. + self.global_features = [] + self.global_features.extend(capa.features.extractors.common.extract_os(self.buf)) + self.global_features.extend(capa.features.extractors.viv.global_.extract_arch(self.vw)) def get_base_address(self): # assume there is only one file loaded into the vw return list(self.vw.filemeta.values())[0]["imagebase"] def extract_file_features(self): - for feature, va in capa.features.extractors.viv.file.extract_features(self.vw, self.path): + for feature, va in capa.features.extractors.viv.file.extract_features(self.vw, self.buf): yield feature, va + yield from self.global_features def get_functions(self): for va in sorted(self.vw.getFunctions()): @@ -53,6 +63,7 @@ class VivisectFeatureExtractor(FeatureExtractor): def extract_function_features(self, f): for feature, va in capa.features.extractors.viv.function.extract_features(f): yield feature, va + yield from self.global_features def get_basic_blocks(self, f): return f.basic_blocks @@ -60,6 +71,7 @@ class VivisectFeatureExtractor(FeatureExtractor): def extract_basic_block_features(self, f, bb): for feature, va in capa.features.extractors.viv.basicblock.extract_features(f, bb): yield feature, va + yield from self.global_features def get_instructions(self, f, bb): for insn in bb.instructions: @@ -68,6 +80,7 @@ class VivisectFeatureExtractor(FeatureExtractor): def extract_insn_features(self, f, bb, insn): for feature, va in capa.features.extractors.viv.insn.extract_features(f, bb, insn): yield feature, va + yield from self.global_features def is_library_function(self, va): return viv_utils.flirt.is_library_function(self.vw, va) diff --git a/capa/features/extractors/viv/file.py b/capa/features/extractors/viv/file.py index 00eb57ed..4b9cd13f 100644 --- a/capa/features/extractors/viv/file.py +++ b/capa/features/extractors/viv/file.py @@ -11,26 +11,24 @@ import viv_utils import viv_utils.flirt import capa.features.insn +import capa.features.extractors.common import capa.features.extractors.helpers import capa.features.extractors.strings from capa.features.file import Export, Import, Section, FunctionName from capa.features.common import String, Characteristic -def extract_file_embedded_pe(vw, file_path): - with open(file_path, "rb") as f: - fbytes = f.read() - - for offset, i in pe_carve.carve(fbytes, 1): +def extract_file_embedded_pe(buf, **kwargs): + for offset, _ in pe_carve.carve(buf, 1): yield Characteristic("embedded pe"), offset -def extract_file_export_names(vw, file_path): - for va, etype, name, _ in vw.getExports(): +def extract_file_export_names(vw, **kwargs): + for va, _, name, _ in vw.getExports(): yield Export(name), va -def extract_file_import_names(vw, file_path): +def extract_file_import_names(vw, **kwargs): """ extract imported function names 1. imports by ordinal: @@ -64,26 +62,23 @@ def is_viv_ord_impname(impname: str) -> bool: return True -def extract_file_section_names(vw, file_path): +def extract_file_section_names(vw, **kwargs): for va, _, segname, _ in vw.getSegments(): yield Section(segname), va -def extract_file_strings(vw, file_path): +def extract_file_strings(buf, **kwargs): """ extract ASCII and UTF-16 LE strings from file """ - with open(file_path, "rb") as f: - b = f.read() - - for s in capa.features.extractors.strings.extract_ascii_strings(b): + for s in capa.features.extractors.strings.extract_ascii_strings(buf): yield String(s.s), s.offset - for s in capa.features.extractors.strings.extract_unicode_strings(b): + for s in capa.features.extractors.strings.extract_unicode_strings(buf): yield String(s.s), s.offset -def extract_file_function_names(vw, file_path): +def extract_file_function_names(vw, **kwargs): """ extract the names of statically-linked library functions. """ @@ -93,20 +88,24 @@ def extract_file_function_names(vw, file_path): yield FunctionName(name), va -def extract_features(vw, file_path): +def extract_file_format(buf, **kwargs): + yield from capa.features.extractors.common.extract_format(buf) + + +def extract_features(vw, buf: bytes): """ extract file features from given workspace args: vw (vivisect.VivWorkspace): the vivisect workspace - file_path: path to the input file + buf: the raw input file bytes yields: Tuple[Feature, VA]: a feature and its location. """ for file_handler in FILE_HANDLERS: - for feature, va in file_handler(vw, file_path): + for feature, va in file_handler(vw=vw, buf=buf): # type: ignore yield feature, va @@ -117,4 +116,5 @@ FILE_HANDLERS = ( extract_file_section_names, extract_file_strings, extract_file_function_names, + extract_file_format, ) diff --git a/capa/features/extractors/viv/global_.py b/capa/features/extractors/viv/global_.py new file mode 100644 index 00000000..8fc08ee2 --- /dev/null +++ b/capa/features/extractors/viv/global_.py @@ -0,0 +1,24 @@ +import logging + +import envi.archs.i386 +import envi.archs.amd64 + +from capa.features.common import ARCH_I386, ARCH_AMD64, Arch + +logger = logging.getLogger(__name__) + + +def extract_arch(vw): + if isinstance(vw.arch, envi.archs.amd64.Amd64Module): + yield Arch(ARCH_AMD64), 0x0 + + elif isinstance(vw.arch, envi.archs.i386.i386Module): + yield Arch(ARCH_I386), 0x0 + + else: + # we likely end up here: + # 1. handling a new architecture (e.g. aarch64) + # + # for (1), this logic will need to be updated as the format is implemented. + logger.debug("unsupported architecture: %s", vw.arch.__class__.__name__) + return diff --git a/capa/features/extractors/viv/insn.py b/capa/features/extractors/viv/insn.py index ffe9e3c9..5157b75b 100644 --- a/capa/features/extractors/viv/insn.py +++ b/capa/features/extractors/viv/insn.py @@ -19,8 +19,8 @@ import capa.features.extractors.helpers import capa.features.extractors.viv.helpers from capa.features.insn import API, Number, Offset, Mnemonic from capa.features.common import ( - ARCH_X32, - ARCH_X64, + BITNESS_X32, + BITNESS_X64, MAX_BYTES_FEATURE_SIZE, THUNK_CHAIN_DEPTH_DELTA, Bytes, @@ -34,12 +34,12 @@ from capa.features.extractors.viv.indirect_calls import NotFoundError, resolve_i SECURITY_COOKIE_BYTES_DELTA = 0x40 -def get_arch(vw): - arch = vw.getMeta("Architecture") - if arch == "i386": - return ARCH_X32 - elif arch == "amd64": - return ARCH_X64 +def get_bitness(vw): + bitness = vw.getMeta("Architecture") + if bitness == "i386": + return BITNESS_X32 + elif bitness == "amd64": + return BITNESS_X64 def interface_extract_instruction_XXX(f, bb, insn): @@ -193,7 +193,7 @@ def extract_insn_number_features(f, bb, insn): return yield Number(v), insn.va - yield Number(v, arch=get_arch(f.vw)), insn.va + yield Number(v, bitness=get_bitness(f.vw)), insn.va def derefs(vw, p): @@ -389,7 +389,7 @@ def extract_insn_offset_features(f, bb, insn): v = oper.disp yield Offset(v), insn.va - yield Offset(v, arch=get_arch(f.vw)), insn.va + yield Offset(v, bitness=get_bitness(f.vw)), insn.va # like: [esi + ecx + 16384] # reg ^ ^ @@ -400,7 +400,7 @@ def extract_insn_offset_features(f, bb, insn): v = oper.disp yield Offset(v), insn.va - yield Offset(v, arch=get_arch(f.vw)), insn.va + yield Offset(v, bitness=get_bitness(f.vw)), insn.va def is_security_cookie(f, bb, insn) -> bool: diff --git a/capa/features/insn.py b/capa/features/insn.py index 6c7e07ff..c5bc727e 100644 --- a/capa/features/insn.py +++ b/capa/features/insn.py @@ -21,16 +21,16 @@ class API(Feature): class Number(Feature): - def __init__(self, value: int, arch=None, description=None): - super(Number, self).__init__(value, arch=arch, description=description) + def __init__(self, value: int, bitness=None, description=None): + super(Number, self).__init__(value, bitness=bitness, description=description) def get_value_str(self): return capa.render.utils.hex(self.value) class Offset(Feature): - def __init__(self, value: int, arch=None, description=None): - super(Offset, self).__init__(value, arch=arch, description=description) + def __init__(self, value: int, bitness=None, description=None): + super(Offset, self).__init__(value, bitness=bitness, description=description) def get_value_str(self): return capa.render.utils.hex(self.value) diff --git a/capa/helpers.py b/capa/helpers.py index dd0ff79e..1bdad425 100644 --- a/capa/helpers.py +++ b/capa/helpers.py @@ -21,3 +21,12 @@ def get_file_taste(sample_path: str) -> bytes: with open(sample_path, "rb") as f: taste = f.read(8) return taste + + +def is_runtime_ida(): + try: + import idc + except ImportError: + return False + else: + return True diff --git a/capa/ida/helpers.py b/capa/ida/helpers.py index b2e8f0fc..8cc6d894 100644 --- a/capa/ida/helpers.py +++ b/capa/ida/helpers.py @@ -12,6 +12,8 @@ import datetime import idc import idaapi import idautils +import ida_bytes +import ida_loader import capa import capa.version @@ -28,11 +30,12 @@ SUPPORTED_IDA_VERSIONS = [ "7.6", ] -# file type names as returned by idaapi.get_file_type_name() +# file type names as returned by idainfo.file_type SUPPORTED_FILE_TYPES = [ - "Portable executable for 80386 (PE)", - "Portable executable for AMD64 (PE)", - "Binary file", # x86/AMD64 shellcode support + idaapi.f_PE, + idaapi.f_ELF, + # idaapi.f_MACHO, + idaapi.f_BIN, ] @@ -53,10 +56,10 @@ def is_supported_ida_version(): def is_supported_file_type(): - file_type = idaapi.get_file_type_name() - if file_type not in SUPPORTED_FILE_TYPES: + file_info = idaapi.get_inf_structure() + if file_info.filetype not in SUPPORTED_FILE_TYPES: logger.error("-" * 80) - logger.error(" Input file does not appear to be a PE file.") + logger.error(" Input file does not appear to be a supported file type.") logger.error(" ") logger.error( " capa currently only supports analyzing PE files (or binary files containing x86/AMD64 shellcode) with IDA." @@ -121,3 +124,30 @@ def collect_metadata(): }, "version": capa.version.__version__, } + + +class IDAIO: + """ + An object that acts as a file-like object, + using bytes from the current IDB workspace. + """ + + def __init__(self): + super(IDAIO, self).__init__() + self.offset = 0 + + def seek(self, offset, whence=0): + assert whence == 0 + self.offset = offset + + def read(self, size): + ea = ida_loader.get_fileregion_ea(self.offset) + if ea == idc.BADADDR: + # best guess, such as if file is mapped at address 0x0. + ea = self.offset + + logger.debug("reading 0x%x bytes at 0x%x (ea: 0x%x)", size, self.offset, ea) + return ida_bytes.get_bytes(ea, size) + + def close(self): + return diff --git a/capa/ida/plugin/view.py b/capa/ida/plugin/view.py index 3f13b89d..3986114a 100644 --- a/capa/ida/plugin/view.py +++ b/capa/ida/plugin/view.py @@ -1007,7 +1007,11 @@ class CapaExplorerRulegenFeatures(QtWidgets.QTreeWidget): self.parent_items[feature], (format_feature(feature), format_address(ea)), feature=feature ) else: - ea = eas.pop() + if eas: + ea = eas.pop() + else: + # some features may not have an address e.g. "format" + ea = "" for (i, v) in enumerate((format_feature(feature), format_address(ea))): self.parent_items[feature].setText(i, v) self.parent_items[feature].setData(0, 0x100, feature) diff --git a/capa/main.py b/capa/main.py index 6f6ce168..2b91fe26 100644 --- a/capa/main.py +++ b/capa/main.py @@ -21,7 +21,7 @@ import textwrap import itertools import contextlib import collections -from typing import Any, Dict, List, Tuple, Iterable +from typing import Any, Dict, List, Tuple import halo import tqdm @@ -37,6 +37,7 @@ import capa.features.common import capa.features.freeze import capa.render.vverbose import capa.features.extractors +import capa.features.extractors.common import capa.features.extractors.pefile from capa.rules import Rule, RuleSet from capa.engine import FeatureSet, MatchResults @@ -45,7 +46,6 @@ from capa.features.extractors.base_extractor import FunctionHandle, FeatureExtra RULES_PATH_DEFAULT_STRING = "(embedded rules)" SIGNATURES_PATH_DEFAULT_STRING = "(embedded signatures)" -SUPPORTED_FILE_MAGIC = (b"MZ", b"\x7fELF") BACKEND_VIV = "vivisect" BACKEND_SMDA = "smda" EXTENSIONS_SHELLCODE_32 = ("sc32", "raw32") @@ -237,16 +237,61 @@ def has_file_limitation(rules: RuleSet, capabilities: MatchResults, is_standalon return False -def is_supported_file_type(sample: str) -> bool: +def is_supported_format(sample: str) -> bool: """ Return if this is a supported file based on magic header values """ with open(sample, "rb") as f: - magic = f.read(4) - if magic.startswith(SUPPORTED_FILE_MAGIC): - return True - else: - return False + taste = f.read(0x100) + + return len(list(capa.features.extractors.common.extract_format(taste))) == 1 + + +def get_format(sample: str) -> str: + with open(sample, "rb") as f: + buf = f.read() + + for feature, _ in capa.features.extractors.common.extract_format(buf): + assert isinstance(feature.value, str) + return feature.value + + return "unknown" + + +def is_supported_arch(sample: str) -> bool: + with open(sample, "rb") as f: + buf = f.read() + + return len(list(capa.features.extractors.common.extract_arch(buf))) == 1 + + +def get_arch(sample: str) -> str: + with open(sample, "rb") as f: + buf = f.read() + + for feature, _ in capa.features.extractors.common.extract_arch(buf): + assert isinstance(feature.value, str) + return feature.value + + return "unknown" + + +def is_supported_os(sample: str) -> bool: + with open(sample, "rb") as f: + buf = f.read() + + return len(list(capa.features.extractors.common.extract_os(buf))) == 1 + + +def get_os(sample: str) -> str: + with open(sample, "rb") as f: + buf = f.read() + + for feature, _ in capa.features.extractors.common.extract_os(buf): + assert isinstance(feature.value, str) + return feature.value + + return "unknown" SHELLCODE_BASE = 0x690000 @@ -391,6 +436,14 @@ class UnsupportedFormatError(ValueError): pass +class UnsupportedArchError(ValueError): + pass + + +class UnsupportedOSError(ValueError): + pass + + def get_workspace(path, format, sigpaths): """ load the program at the given path into a vivisect workspace using the given format. @@ -411,7 +464,7 @@ def get_workspace(path, format, sigpaths): logger.debug("generating vivisect workspace for: %s", path) if format == "auto": - if not is_supported_file_type(path): + if not is_supported_format(path): raise UnsupportedFormatError() # don't analyze, so that we can add our Flirt function analyzer first. @@ -443,8 +496,20 @@ def get_extractor( ) -> FeatureExtractor: """ raises: - UnsupportedFormatError: + UnsupportedFormatError + UnsupportedArchError + UnsupportedOSError """ + if format not in ("sc32", "sc64"): + if not is_supported_format(path): + raise UnsupportedFormatError() + + if not is_supported_arch(path): + raise UnsupportedArchError() + + if not is_supported_os(path): + raise UnsupportedOSError() + if backend == "smda": from smda.SmdaConfig import SmdaConfig from smda.Disassembler import Disassembler @@ -463,10 +528,6 @@ def get_extractor( import capa.features.extractors.viv.extractor with halo.Halo(text="analyzing program", spinner="simpleDots", stream=sys.stderr, enabled=not disable_progress): - if format == "auto" and path.endswith(EXTENSIONS_SHELLCODE_32): - format = "sc32" - elif format == "auto" and path.endswith(EXTENSIONS_SHELLCODE_64): - format = "sc64" vw = get_workspace(path, format, sigpaths) if should_save_workspace: @@ -574,7 +635,7 @@ def get_signatures(sigs_path): return paths -def collect_metadata(argv, sample_path, rules_path, format, extractor): +def collect_metadata(argv, sample_path, rules_path, extractor): md5 = hashlib.md5() sha1 = hashlib.sha1() sha256 = hashlib.sha256() @@ -589,6 +650,10 @@ def collect_metadata(argv, sample_path, rules_path, format, extractor): if rules_path != RULES_PATH_DEFAULT_STRING: rules_path = os.path.abspath(os.path.normpath(rules_path)) + format = get_format(sample_path) + arch = get_arch(sample_path) + os_ = get_os(sample_path) + return { "timestamp": datetime.datetime.now().isoformat(), "version": capa.version.__version__, @@ -601,6 +666,8 @@ def collect_metadata(argv, sample_path, rules_path, format, extractor): }, "analysis": { "format": format, + "arch": arch, + "os": os_, "extractor": extractor.__class__.__name__, "rules": rules_path, "base_address": extractor.get_base_address(), @@ -909,6 +976,11 @@ def main(argv=None): extractor = capa.features.freeze.load(f.read()) else: format = args.format + if format == "auto" and args.sample.endswith(EXTENSIONS_SHELLCODE_32): + format = "sc32" + elif format == "auto" and args.sample.endswith(EXTENSIONS_SHELLCODE_64): + format = "sc64" + should_save_workspace = os.environ.get("CAPA_SAVE_WORKSPACE") not in ("0", "no", "NO", "n", None) try: @@ -917,16 +989,32 @@ def main(argv=None): ) except UnsupportedFormatError: logger.error("-" * 80) - logger.error(" Input file does not appear to be a PE file.") + logger.error(" Input file does not appear to be a PE or ELF file.") logger.error(" ") logger.error( - " capa currently only supports analyzing PE files (or shellcode, when using --format sc32|sc64)." + " capa currently only supports analyzing PE and ELF files (or shellcode, when using --format sc32|sc64)." ) logger.error(" If you don't know the input file type, you can try using the `file` utility to guess it.") logger.error("-" * 80) return -1 + except UnsupportedArchError: + logger.error("-" * 80) + logger.error(" Input file does not appear to target a supported architecture.") + logger.error(" ") + logger.error(" capa currently only supports analyzing x86 (32- and 64-bit).") + logger.error("-" * 80) + return -1 + except UnsupportedOSError: + logger.error("-" * 80) + logger.error(" Input file does not appear to target a supported OS.") + logger.error(" ") + logger.error( + " capa currently only supports analyzing executables for some operating systems (including Windows and Linux)." + ) + logger.error("-" * 80) + return -1 - meta = collect_metadata(argv, args.sample, args.rules, format, extractor) + meta = collect_metadata(argv, args.sample, args.rules, extractor) capabilities, counts = find_capabilities(rules, extractor, disable_progress=args.quiet) meta["analysis"].update(counts) diff --git a/capa/render/default.py b/capa/render/default.py index 0f3f9ef3..2ed06a66 100644 --- a/capa/render/default.py +++ b/capa/render/default.py @@ -7,7 +7,6 @@ # See the License for the specific language governing permissions and limitations under the License. import collections -from typing import Dict, List import tabulate @@ -33,6 +32,9 @@ def render_meta(doc, ostream: StringIO): (width("md5", 22), width(doc["meta"]["sample"]["md5"], 82)), ("sha1", doc["meta"]["sample"]["sha1"]), ("sha256", doc["meta"]["sample"]["sha256"]), + ("os", doc["meta"]["analysis"]["os"]), + ("format", doc["meta"]["analysis"]["format"]), + ("arch", doc["meta"]["analysis"]["arch"]), ("path", doc["meta"]["sample"]["path"]), ] diff --git a/capa/render/verbose.py b/capa/render/verbose.py index df8a9586..d7908268 100644 --- a/capa/render/verbose.py +++ b/capa/render/verbose.py @@ -41,7 +41,9 @@ def render_meta(ostream, doc): path /tmp/suspicious.dll_ timestamp 2020-07-03T10:17:05.796933 capa version 0.0.0 - format auto + os windows + format pe + arch amd64 extractor VivisectFeatureExtractor base address 0x10000000 rules (embedded rules) @@ -55,7 +57,9 @@ def render_meta(ostream, doc): ("path", doc["meta"]["sample"]["path"]), ("timestamp", doc["meta"]["timestamp"]), ("capa version", doc["meta"]["version"]), + ("os", doc["meta"]["analysis"]["os"]), ("format", doc["meta"]["analysis"]["format"]), + ("arch", doc["meta"]["analysis"]["arch"]), ("extractor", doc["meta"]["analysis"]["extractor"]), ("base address", hex(doc["meta"]["analysis"]["base_address"])), ("rules", doc["meta"]["analysis"]["rules"]), diff --git a/capa/render/vverbose.py b/capa/render/vverbose.py index 95239eab..adecde36 100644 --- a/capa/render/vverbose.py +++ b/capa/render/vverbose.py @@ -113,7 +113,8 @@ def render_feature(ostream, match, feature, indent=0): ostream.write(capa.rules.DESCRIPTION_SEPARATOR) ostream.write(feature["description"]) - render_locations(ostream, match) + if key not in ("os", "arch"): + render_locations(ostream, match) ostream.write("\n") else: # like: diff --git a/capa/rules.py b/capa/rules.py index 6fa9a4c4..b8b90faa 100644 --- a/capa/rules.py +++ b/capa/rules.py @@ -22,7 +22,7 @@ except ImportError: # https://github.com/python/mypy/issues/1153 from backports.functools_lru_cache import lru_cache # type: ignore -from typing import Any, Set, Dict, List, Union, Iterator +from typing import Any, Dict, List, Union, Iterator import yaml import ruamel.yaml @@ -78,6 +78,9 @@ SUPPORTED_FEATURES = { capa.features.file.FunctionName, capa.features.common.Characteristic("embedded pe"), capa.features.common.String, + capa.features.common.Format, + capa.features.common.OS, + capa.features.common.Arch, }, FUNCTION_SCOPE: { # plus basic block scope features, see below @@ -86,6 +89,8 @@ SUPPORTED_FEATURES = { capa.features.common.Characteristic("calls to"), capa.features.common.Characteristic("loop"), capa.features.common.Characteristic("recursive call"), + capa.features.common.OS, + capa.features.common.Arch, }, BASIC_BLOCK_SCOPE: { capa.features.common.MatchedRule, @@ -103,6 +108,8 @@ SUPPORTED_FEATURES = { capa.features.common.Characteristic("tight loop"), capa.features.common.Characteristic("stack string"), capa.features.common.Characteristic("indirect call"), + capa.features.common.OS, + capa.features.common.Arch, }, } @@ -153,14 +160,14 @@ def ensure_feature_valid_for_scope(scope: str, feature: Union[Feature, Statement and isinstance(feature.value, str) and capa.features.common.Characteristic(feature.value) not in SUPPORTED_FEATURES[scope] ): - raise InvalidRule("feature %s not support for scope %s" % (feature, scope)) + raise InvalidRule("feature %s not supported for scope %s" % (feature, scope)) if not isinstance(feature, capa.features.common.Characteristic): # features of this scope that are not Characteristics will be Type instances. # check that the given feature is one of these types. types_for_scope = filter(lambda t: isinstance(t, type), SUPPORTED_FEATURES[scope]) if not isinstance(feature, tuple(types_for_scope)): # type: ignore - raise InvalidRule("feature %s not support for scope %s" % (feature, scope)) + raise InvalidRule("feature %s not supported for scope %s" % (feature, scope)) def parse_int(s: str) -> int: @@ -217,19 +224,19 @@ def parse_feature(key: str): elif key == "number": return capa.features.insn.Number elif key.startswith("number/"): - arch = key.partition("/")[2] + bitness = key.partition("/")[2] # the other handlers here return constructors for features, # and we want to as well, - # however, we need to preconfigure one of the arguments (`arch`). + # however, we need to preconfigure one of the arguments (`bitness`). # so, instead we return a partially-applied function that - # provides `arch` to the feature constructor. + # provides `bitness` to the feature constructor. # it forwards any other arguments provided to the closure along to the constructor. - return functools.partial(capa.features.insn.Number, arch=arch) + return functools.partial(capa.features.insn.Number, bitness=bitness) elif key == "offset": return capa.features.insn.Offset elif key.startswith("offset/"): - arch = key.partition("/")[2] - return functools.partial(capa.features.insn.Offset, arch=arch) + bitness = key.partition("/")[2] + return functools.partial(capa.features.insn.Offset, bitness=bitness) elif key == "mnemonic": return capa.features.insn.Mnemonic elif key == "basic blocks": @@ -246,6 +253,12 @@ def parse_feature(key: str): return capa.features.common.MatchedRule elif key == "function-name": return capa.features.file.FunctionName + elif key == "os": + return capa.features.common.OS + elif key == "format": + return capa.features.common.Format + elif key == "arch": + return capa.features.common.Arch else: raise InvalidRule("unexpected statement: %s" % key) diff --git a/scripts/bulk-process.py b/scripts/bulk-process.py index da80a477..3bb3495e 100644 --- a/scripts/bulk-process.py +++ b/scripts/bulk-process.py @@ -126,7 +126,7 @@ def get_capa_results(args): "error": "unexpected error: %s" % (e), } - meta = capa.main.collect_metadata("", path, "", format, extractor) + meta = capa.main.collect_metadata("", path, "", extractor) capabilities, counts = capa.main.find_capabilities(rules, extractor, disable_progress=True) meta["analysis"].update(counts) diff --git a/scripts/capa2yara.py b/scripts/capa2yara.py index 169b8cc4..5a72c697 100644 --- a/scripts/capa2yara.py +++ b/scripts/capa2yara.py @@ -43,7 +43,7 @@ import capa.rules import capa.engine import capa.features import capa.features.insn -from capa.features.common import ARCH_X32, ARCH_X64, String +from capa.features.common import BITNESS_X32, BITNESS_X64, String logger = logging.getLogger("capa2yara") diff --git a/scripts/capa_as_library.py b/scripts/capa_as_library.py index 449c35b9..012c1bf3 100644 --- a/scripts/capa_as_library.py +++ b/scripts/capa_as_library.py @@ -169,7 +169,7 @@ def capa_details(file_path, output_format="dictionary"): capabilities, counts = capa.main.find_capabilities(rules, extractor, disable_progress=True) # collect metadata (used only to make rendering more complete) - meta = capa.main.collect_metadata("", file_path, RULES_PATH, "auto", extractor) + meta = capa.main.collect_metadata("", file_path, RULES_PATH, extractor) meta["analysis"].update(counts) capa_output = False diff --git a/scripts/detect-elf-os.py b/scripts/detect-elf-os.py new file mode 100644 index 00000000..c01990f4 --- /dev/null +++ b/scripts/detect-elf-os.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python2 +""" +Copyright (C) 2021 FireEye, Inc. All Rights Reserved. +Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. +You may obtain a copy of the License at: [package root]/LICENSE.txt +Unless required by applicable law or agreed to in writing, software distributed under the License + is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and limitations under the License. + +detect-elf-os + +Attempt to detect the underlying OS that the given ELF file targets. +""" +import sys +import logging +import argparse +import contextlib +from typing import BinaryIO + +import capa.helpers +import capa.features.extractors.elf + +logger = logging.getLogger("capa.detect-elf-os") + + +def main(argv=None): + if capa.helpers.is_runtime_ida(): + from capa.ida.helpers import IDAIO + + f: BinaryIO = IDAIO() + + else: + if argv is None: + argv = sys.argv[1:] + + parser = argparse.ArgumentParser(description="Detect the underlying OS for the given ELF file") + parser.add_argument("sample", type=str, help="path to ELF file") + + logging_group = parser.add_argument_group("logging arguments") + + logging_group.add_argument("-d", "--debug", action="store_true", help="enable debugging output on STDERR") + logging_group.add_argument( + "-q", "--quiet", action="store_true", help="disable all status output except fatal errors" + ) + + args = parser.parse_args(args=argv) + + if args.quiet: + logging.basicConfig(level=logging.WARNING) + logging.getLogger().setLevel(logging.WARNING) + elif args.debug: + logging.basicConfig(level=logging.DEBUG) + logging.getLogger().setLevel(logging.DEBUG) + else: + logging.basicConfig(level=logging.INFO) + logging.getLogger().setLevel(logging.INFO) + + f = open(args.sample, "rb") + + with contextlib.closing(f): + try: + print(capa.features.extractors.elf.detect_elf_os(f)) + return 0 + except capa.features.extractors.elf.CorruptElfFile as e: + logger.error("corrupt ELF file: %s", str(e.args[0])) + return -1 + + +if __name__ == "__main__": + if capa.helpers.is_runtime_ida(): + main() + else: + sys.exit(main()) diff --git a/scripts/show-capabilities-by-function.py b/scripts/show-capabilities-by-function.py index b95481df..11fcdaab 100644 --- a/scripts/show-capabilities-by-function.py +++ b/scripts/show-capabilities-by-function.py @@ -171,7 +171,7 @@ def main(argv=None): logger.error("-" * 80) return -1 - meta = capa.main.collect_metadata(argv, args.sample, args.rules, format, extractor) + meta = capa.main.collect_metadata(argv, args.sample, args.rules, extractor) capabilities, counts = capa.main.find_capabilities(rules, extractor) meta["analysis"].update(counts) diff --git a/scripts/show-features.py b/scripts/show-features.py index 3090a471..6d69ea7f 100644 --- a/scripts/show-features.py +++ b/scripts/show-features.py @@ -75,6 +75,7 @@ import capa.rules import capa.engine import capa.helpers import capa.features +import capa.features.common import capa.features.freeze logger = logging.getLogger("capa.show-features") @@ -203,14 +204,23 @@ def print_features(functions, extractor): continue for feature, va in extractor.extract_function_features(f): + if capa.features.common.is_global_feature(feature): + continue + print("func: 0x%08x: %s" % (va, feature)) for bb in extractor.get_basic_blocks(f): for feature, va in extractor.extract_basic_block_features(f, bb): + if capa.features.common.is_global_feature(feature): + continue + print("bb : 0x%08x: %s" % (va, feature)) for insn in extractor.get_instructions(f, bb): for feature, va in extractor.extract_insn_features(f, bb, insn): + if capa.features.common.is_global_feature(feature): + continue + try: print("insn: 0x%08x: %s" % (va, feature)) except UnicodeEncodeError: diff --git a/tests/fixtures.py b/tests/fixtures.py index f15e7f06..b47e79e2 100644 --- a/tests/fixtures.py +++ b/tests/fixtures.py @@ -21,7 +21,19 @@ import capa.features.file import capa.features.insn import capa.features.common import capa.features.basicblock -from capa.features.common import ARCH_X32, ARCH_X64 +from capa.features.common import ( + OS, + OS_LINUX, + ARCH_I386, + FORMAT_PE, + ARCH_AMD64, + FORMAT_ELF, + OS_WINDOWS, + BITNESS_X32, + BITNESS_X64, + Arch, + Format, +) CD = os.path.dirname(__file__) @@ -196,6 +208,8 @@ def get_data_path_by_name(name): return os.path.join(CD, "data", "773290480d5445f11d3dc1b800728966.exe_") elif name.startswith("3b13b"): return os.path.join(CD, "data", "3b13b6f1d7cd14dc4a097a12e2e505c0a4cff495262261e2bfc991df238b9b04.dll_") + elif name == "7351f.elf": + return os.path.join(CD, "data", "7351f8a40c5450557b24622417fc478d.elf_") else: raise ValueError("unexpected sample fixture: %s" % name) @@ -241,6 +255,8 @@ def get_sample_md5_by_name(name): elif name.startswith("3b13b"): # file name is SHA256 hash return "56a6ffe6a02941028cc8235204eef31d" + elif name == "7351f.elf": + return "7351f8a40c5450557b24622417fc478d" else: raise ValueError("unexpected sample fixture: %s" % name) @@ -379,10 +395,10 @@ FEATURE_PRESENCE_TESTS = sorted( # insn/number: stack adjustments ("mimikatz", "function=0x40105D", capa.features.insn.Number(0xC), False), ("mimikatz", "function=0x40105D", capa.features.insn.Number(0x10), False), - # insn/number: arch flavors + # insn/number: bitness flavors ("mimikatz", "function=0x40105D", capa.features.insn.Number(0xFF), True), - ("mimikatz", "function=0x40105D", capa.features.insn.Number(0xFF, arch=ARCH_X32), True), - ("mimikatz", "function=0x40105D", capa.features.insn.Number(0xFF, arch=ARCH_X64), False), + ("mimikatz", "function=0x40105D", capa.features.insn.Number(0xFF, bitness=BITNESS_X32), True), + ("mimikatz", "function=0x40105D", capa.features.insn.Number(0xFF, bitness=BITNESS_X64), False), # insn/offset ("mimikatz", "function=0x40105D", capa.features.insn.Offset(0x0), True), ("mimikatz", "function=0x40105D", capa.features.insn.Offset(0x4), True), @@ -395,10 +411,10 @@ FEATURE_PRESENCE_TESTS = sorted( # insn/offset: negative ("mimikatz", "function=0x4011FB", capa.features.insn.Offset(-0x1), True), ("mimikatz", "function=0x4011FB", capa.features.insn.Offset(-0x2), True), - # insn/offset: arch flavors + # insn/offset: bitness flavors ("mimikatz", "function=0x40105D", capa.features.insn.Offset(0x0), True), - ("mimikatz", "function=0x40105D", capa.features.insn.Offset(0x0, arch=ARCH_X32), True), - ("mimikatz", "function=0x40105D", capa.features.insn.Offset(0x0, arch=ARCH_X64), False), + ("mimikatz", "function=0x40105D", capa.features.insn.Offset(0x0, bitness=BITNESS_X32), True), + ("mimikatz", "function=0x40105D", capa.features.insn.Offset(0x0, bitness=BITNESS_X64), False), # insn/api ("mimikatz", "function=0x403BAC", capa.features.insn.API("advapi32.CryptAcquireContextW"), True), ("mimikatz", "function=0x403BAC", capa.features.insn.API("advapi32.CryptAcquireContext"), True), @@ -499,6 +515,26 @@ FEATURE_PRESENCE_TESTS = sorted( ("mimikatz", "function=0x456BB9", capa.features.common.Characteristic("calls to"), False), # file/function-name ("pma16-01", "file", capa.features.file.FunctionName("__aulldiv"), True), + # os & format & arch + ("pma16-01", "file", OS(OS_WINDOWS), True), + ("pma16-01", "file", OS(OS_LINUX), False), + ("pma16-01", "function=0x404356", OS(OS_WINDOWS), True), + ("pma16-01", "function=0x404356,bb=0x4043B9", OS(OS_WINDOWS), True), + ("pma16-01", "file", Arch(ARCH_I386), True), + ("pma16-01", "file", Arch(ARCH_AMD64), False), + ("pma16-01", "function=0x404356", Arch(ARCH_I386), True), + ("pma16-01", "function=0x404356,bb=0x4043B9", Arch(ARCH_I386), True), + ("pma16-01", "file", Format(FORMAT_PE), True), + ("pma16-01", "file", Format(FORMAT_ELF), False), + # elf support + ("7351f.elf", "file", OS(OS_LINUX), True), + ("7351f.elf", "file", OS(OS_WINDOWS), False), + ("7351f.elf", "file", Format(FORMAT_ELF), True), + ("7351f.elf", "file", Format(FORMAT_PE), False), + ("7351f.elf", "file", Arch(ARCH_I386), False), + ("7351f.elf", "file", Arch(ARCH_AMD64), True), + ("7351f.elf", "function=0x408753", capa.features.common.String("/dev/null"), True), + ("7351f.elf", "function=0x408753,bb=0x408781", capa.features.insn.API("open"), True), ], # order tests by (file, item) # so that our LRU cache is most effective. diff --git a/tests/test_engine.py b/tests/test_engine.py index 642ddb49..57fffb8e 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -474,11 +474,11 @@ def test_match_namespace(): def test_render_number(): assert str(capa.features.insn.Number(1)) == "number(0x1)" - assert str(capa.features.insn.Number(1, arch=capa.features.common.ARCH_X32)) == "number/x32(0x1)" - assert str(capa.features.insn.Number(1, arch=capa.features.common.ARCH_X64)) == "number/x64(0x1)" + assert str(capa.features.insn.Number(1, bitness=capa.features.common.BITNESS_X32)) == "number/x32(0x1)" + assert str(capa.features.insn.Number(1, bitness=capa.features.common.BITNESS_X64)) == "number/x64(0x1)" def test_render_offset(): assert str(capa.features.insn.Offset(1)) == "offset(0x1)" - assert str(capa.features.insn.Offset(1, arch=capa.features.common.ARCH_X32)) == "offset/x32(0x1)" - assert str(capa.features.insn.Offset(1, arch=capa.features.common.ARCH_X64)) == "offset/x64(0x1)" + assert str(capa.features.insn.Offset(1, bitness=capa.features.common.BITNESS_X32)) == "offset/x32(0x1)" + assert str(capa.features.insn.Offset(1, bitness=capa.features.common.BITNESS_X64)) == "offset/x64(0x1)" diff --git a/tests/test_pefile_features.py b/tests/test_pefile_features.py index 8bb46d43..2e1afc7b 100644 --- a/tests/test_pefile_features.py +++ b/tests/test_pefile_features.py @@ -20,9 +20,12 @@ import capa.features.file ) def test_pefile_features(sample, scope, feature, expected): if scope.__name__ != "file": - pytest.xfail("pefile only extract file scope features") + pytest.xfail("pefile only extracts file scope features") if isinstance(feature, capa.features.file.FunctionName): - pytest.xfail("pefile only doesn't extract function names") + pytest.xfail("pefile doesn't extract function names") + + if ".elf" in sample: + pytest.xfail("pefile doesn't handle ELF files") fixtures.do_test_feature_presence(fixtures.get_pefile_extractor, sample, scope, feature, expected) diff --git a/tests/test_rules.py b/tests/test_rules.py index 12791d2e..9362ef91 100644 --- a/tests/test_rules.py +++ b/tests/test_rules.py @@ -15,7 +15,20 @@ import capa.engine import capa.features.common from capa.features.file import FunctionName from capa.features.insn import Number, Offset -from capa.features.common import ARCH_X32, ARCH_X64, String +from capa.features.common import ( + OS, + OS_LINUX, + ARCH_I386, + FORMAT_PE, + ARCH_AMD64, + FORMAT_ELF, + OS_WINDOWS, + BITNESS_X32, + BITNESS_X64, + Arch, + Format, + String, +) def test_rule_ctor(): @@ -517,7 +530,7 @@ def test_invalid_number(): ) -def test_number_arch(): +def test_number_bitness(): r = capa.rules.Rule.from_yaml( textwrap.dedent( """ @@ -529,13 +542,13 @@ def test_number_arch(): """ ) ) - assert r.evaluate({Number(2, arch=ARCH_X32): {1}}) == True + assert r.evaluate({Number(2, bitness=BITNESS_X32): {1}}) == True assert r.evaluate({Number(2): {1}}) == False - assert r.evaluate({Number(2, arch=ARCH_X64): {1}}) == False + assert r.evaluate({Number(2, bitness=BITNESS_X64): {1}}) == False -def test_number_arch_symbol(): +def test_number_bitness_symbol(): r = capa.rules.Rule.from_yaml( textwrap.dedent( """ @@ -547,7 +560,7 @@ def test_number_arch_symbol(): """ ) ) - assert r.evaluate({Number(2, arch=ARCH_X32, description="some constant"): {1}}) == True + assert r.evaluate({Number(2, bitness=BITNESS_X32, description="some constant"): {1}}) == True def test_offset_symbol(): @@ -595,7 +608,7 @@ def test_count_offset_symbol(): assert r.evaluate({Offset(0x100, description="symbol name"): {1, 2, 3}}) == True -def test_offset_arch(): +def test_offset_bitness(): r = capa.rules.Rule.from_yaml( textwrap.dedent( """ @@ -607,13 +620,13 @@ def test_offset_arch(): """ ) ) - assert r.evaluate({Offset(2, arch=ARCH_X32): {1}}) == True + assert r.evaluate({Offset(2, bitness=BITNESS_X32): {1}}) == True assert r.evaluate({Offset(2): {1}}) == False - assert r.evaluate({Offset(2, arch=ARCH_X64): {1}}) == False + assert r.evaluate({Offset(2, bitness=BITNESS_X64): {1}}) == False -def test_offset_arch_symbol(): +def test_offset_bitness_symbol(): r = capa.rules.Rule.from_yaml( textwrap.dedent( """ @@ -625,7 +638,7 @@ def test_offset_arch_symbol(): """ ) ) - assert r.evaluate({Offset(2, arch=ARCH_X32, description="some constant"): {1}}) == True + assert r.evaluate({Offset(2, bitness=BITNESS_X32, description="some constant"): {1}}) == True def test_invalid_offset(): @@ -944,3 +957,57 @@ def test_function_name_features(): assert (FunctionName("strcpy") in children) == True assert (FunctionName("strcmp", description="copy from here to there") in children) == True assert (FunctionName("strdup", description="duplicate a string") in children) == True + + +def test_os_features(): + rule = textwrap.dedent( + """ + rule: + meta: + name: test rule + scope: file + features: + - and: + - os: windows + """ + ) + r = capa.rules.Rule.from_yaml(rule) + children = list(r.statement.get_children()) + assert (OS(OS_WINDOWS) in children) == True + assert (OS(OS_LINUX) not in children) == True + + +def test_format_features(): + rule = textwrap.dedent( + """ + rule: + meta: + name: test rule + scope: file + features: + - and: + - format: pe + """ + ) + r = capa.rules.Rule.from_yaml(rule) + children = list(r.statement.get_children()) + assert (Format(FORMAT_PE) in children) == True + assert (Format(FORMAT_ELF) not in children) == True + + +def test_arch_features(): + rule = textwrap.dedent( + """ + rule: + meta: + name: test rule + scope: file + features: + - and: + - arch: amd64 + """ + ) + r = capa.rules.Rule.from_yaml(rule) + children = list(r.statement.get_children()) + assert (Arch(ARCH_AMD64) in children) == True + assert (Arch(ARCH_I386) not in children) == True diff --git a/tests/test_viv_features.py b/tests/test_viv_features.py index fa8bfda3..3989db81 100644 --- a/tests/test_viv_features.py +++ b/tests/test_viv_features.py @@ -15,6 +15,9 @@ from fixtures import * indirect=["sample", "scope"], ) def test_viv_features(sample, scope, feature, expected): + if "elf" in sample: + pytest.xfail("viv ELF parsing is broken (for our test file), see #735") + fixtures.do_test_feature_presence(fixtures.get_viv_extractor, sample, scope, feature, expected) @@ -24,4 +27,7 @@ def test_viv_features(sample, scope, feature, expected): indirect=["sample", "scope"], ) def test_viv_feature_counts(sample, scope, feature, expected): + if "elf" in sample: + pytest.xfail("viv ELF parsing is broken (for our test file), see #735") + fixtures.do_test_feature_count(fixtures.get_viv_extractor, sample, scope, feature, expected)