Unverified Commit 82bd92c9 by Johannes vom Dorp Committed by GitHub

Merge pull request #7 from fkie-cad/yara-4.3

yara >= 4.3 fix
parents e9c621a9 b7a840e6
from __future__ import annotations
import logging import logging
import os import os
import socket import socket
from sys import exc_info from sys import exc_info
from typing import Iterable
import yara
from common_analysis_base import AnalysisPluginFile from common_analysis_base import AnalysisPluginFile
from common_helper_files import get_dir_of_file from common_helper_files import get_dir_of_file
from packaging.version import parse as parse_version
import yara
from .version import __version__ as system_version from .version import __version__ as system_version
logger = logging.getLogger('CommonAnalysisIPAndURIFinder') logger = logging.getLogger('CommonAnalysisIPAndURIFinder')
logger.setLevel(logging.INFO) logger.setLevel(logging.INFO)
YARA_IS_NEW = parse_version(yara.YARA_VERSION) >= parse_version('4.3.0')
class FinderBase: class FinderBase:
@staticmethod @staticmethod
def get_strings_from_matches(matches): def get_strings_from_matches(matches):
# the desired strings are contained in the tuples at the third position in each yara match object return [string for match in matches for string in URIFinder.get_strings_without_overlaps(match.strings)]
return [item[2].decode() for match in matches for item in URIFinder.eliminate_overlaps(match.strings)]
@staticmethod @staticmethod
def get_file_content(file_path): def get_file_content(file_path):
...@@ -101,27 +106,39 @@ class URIFinder(FinderBase): ...@@ -101,27 +106,39 @@ class URIFinder(FinderBase):
return self.find_uris(file_content) return self.find_uris(file_content)
@staticmethod @staticmethod
def eliminate_overlaps(yara_match_strings): def get_strings_without_overlaps(matches: list[tuple[int, str, bytes]] | list[yara.StringMatch]) -> Iterable[str]:
""" yara matches contain overlaps
e.g. if the string contains 123.123.123.123
the results would be 123.123.123.123, 23.123.123.123 and 3.123.123.123
""" """
result = yara_match_strings[:] yara matches contain overlaps e.g. if the string contains 123.123.123.123, the results would be
for i in range(1, len(yara_match_strings)): 123.123.123.123, 23.123.123.123 and 3.123.123.123
# if the matches are in direct succession """
if yara_match_strings[i][0] - yara_match_strings[i - 1][0] == 1: iter_function = _iter_match_instance if YARA_IS_NEW else _iter_tuples
result.remove(yara_match_strings[i]) last_offset = -2
return result for offset, string in iter_function(matches):
if offset != last_offset + 1: # skip non-greedy overlaps
yield string
last_offset = offset
class CommonAnalysisIPAndURIFinder(AnalysisPluginFile): def _iter_match_instance(matches: list[yara.StringMatch]) -> Iterable[tuple[int, str]]:
# newer YARA versions use StringMatchInstance objects
# see https://yara.readthedocs.io/en/latest/yarapython.html#yara.StringMatchInstance
for match in matches:
for instance in match.instances:
yield instance.offset, instance.matched_data.decode()
def _iter_tuples(matches: list[tuple[int, str, bytes]]) -> Iterable[tuple[int, str]]:
for offset, _, string in matches:
yield offset, string.decode()
class CommonAnalysisIPAndURIFinder(AnalysisPluginFile):
def __init__(self, yara_uri_rules=None, yara_ip_rules=None): def __init__(self, yara_uri_rules=None, yara_ip_rules=None):
super(CommonAnalysisIPAndURIFinder, self).__init__(system_version) super(CommonAnalysisIPAndURIFinder, self).__init__(system_version)
self._set_rule_file_pathes(yara_uri_rules, yara_ip_rules) self._set_rule_file_paths(yara_uri_rules, yara_ip_rules)
self._check_for_errors() self._check_for_errors()
def _set_rule_file_pathes(self, yara_uri_rules, yara_ip_rules): def _set_rule_file_paths(self, yara_uri_rules, yara_ip_rules):
internal_signature_dir = os.path.join(get_dir_of_file(__file__), 'yara_rules') internal_signature_dir = os.path.join(get_dir_of_file(__file__), 'yara_rules')
if yara_ip_rules is None: if yara_ip_rules is None:
self.yara_ip_rules = os.path.join(internal_signature_dir, 'ip_rules.yara') self.yara_ip_rules = os.path.join(internal_signature_dir, 'ip_rules.yara')
......
[pytest] [pytest]
addopts = --pycodestyle --cov=./ -v addopts = --cov=./ -v
...@@ -19,7 +19,8 @@ setup( ...@@ -19,7 +19,8 @@ setup(
install_requires=[ install_requires=[
'common_analysis_base @ git+https://github.com/mass-project/common_analysis_base.git', 'common_analysis_base @ git+https://github.com/mass-project/common_analysis_base.git',
'common_helper_files @ git+https://github.com/fkie-cad/common_helper_files.git', 'common_helper_files @ git+https://github.com/fkie-cad/common_helper_files.git',
'yara-python >= 3.5' 'packaging >= 23.0',
'yara-python >= 3.5',
], ],
extras_require={ extras_require={
'dev': [ 'dev': [
......
...@@ -25,17 +25,17 @@ class TestIpAndUrlFinder(unittest.TestCase): ...@@ -25,17 +25,17 @@ class TestIpAndUrlFinder(unittest.TestCase):
results = IPFinder(self.yara_ip_rules).find_ips(self.test_string, validate=False) results = IPFinder(self.yara_ip_rules).find_ips(self.test_string, validate=False)
expected_results = {"1.2.3.4", "123.123.123.123", "1234:1234:abcd:abcd:1234:1234:abcd:abcd", expected_results = {"1.2.3.4", "123.123.123.123", "1234:1234:abcd:abcd:1234:1234:abcd:abcd",
"2001:db8:0:8d3:0:8a2e:70:7344"} "2001:db8:0:8d3:0:8a2e:70:7344"}
self.assertEqual(set(results), expected_results) assert set(results) == expected_results
def test_find_ipv4_addresses(self): def test_find_ipv4_addresses(self):
results = IPFinder(self.yara_ip_rules).find_ipv4_addresses(self.test_string, validate=True) results = IPFinder(self.yara_ip_rules).find_ipv4_addresses(self.test_string, validate=True)
expected_results = {"1.2.3.4", "123.123.123.123"} expected_results = {"1.2.3.4", "123.123.123.123"}
self.assertEqual(set(results), expected_results) assert set(results) == expected_results
def test_find_ipv6_addresses(self): def test_find_ipv6_addresses(self):
results = IPFinder(self.yara_ip_rules).find_ipv6_addresses(self.test_string, validate=True) results = IPFinder(self.yara_ip_rules).find_ipv6_addresses(self.test_string, validate=True)
expected_results = {"1234:1234:abcd:abcd:1234:1234:abcd:abcd", "2001:db8:0:8d3:0:8a2e:70:7344"} expected_results = {"1234:1234:abcd:abcd:1234:1234:abcd:abcd", "2001:db8:0:8d3:0:8a2e:70:7344"}
self.assertEqual(set(results), expected_results) assert set(results) == expected_results
def test_find_ips_in_file(self): def test_find_ips_in_file(self):
with tempfile.NamedTemporaryFile() as test_file: with tempfile.NamedTemporaryFile() as test_file:
...@@ -48,19 +48,19 @@ class TestIpAndUrlFinder(unittest.TestCase): ...@@ -48,19 +48,19 @@ class TestIpAndUrlFinder(unittest.TestCase):
test_file.seek(0) test_file.seek(0)
results = set(IPFinder(self.yara_ip_rules).find_ips_in_file(test_file.name, validate=False)) results = set(IPFinder(self.yara_ip_rules).find_ips_in_file(test_file.name, validate=False))
expected_results = {"1.2.3.4", "255.255.255.255", "1234:1234:abcd:abcd:1234:1234:abcd:abcd"} expected_results = {"1.2.3.4", "255.255.255.255", "1234:1234:abcd:abcd:1234:1234:abcd:abcd"}
self.assertEqual(results, expected_results) assert results == expected_results
def test_validate_ipv4(self): def test_validate_ipv4(self):
ips = ["1.1.1.1", "1.1.1", "a.1.1.1", "1.1.1.1.1"] ips = ["1.1.1.1", "1.1.1", "a.1.1.1", "1.1.1.1.1"]
valid_ips = ["1.1.1.1"] valid_ips = ["1.1.1.1"]
validated_ips = IPFinder(self.yara_ip_rules)._validate_ips(ips, socket.AF_INET) validated_ips = IPFinder(self.yara_ip_rules)._validate_ips(ips, socket.AF_INET)
self.assertEqual(validated_ips, valid_ips) assert validated_ips == valid_ips
def test_validate_ipv6(self): def test_validate_ipv6(self):
ips = ["1234:1234:abcd:abcd:1234:1234:1.0.0.127", "2001:db8::8d3::", "2001:db8:0:0:8d3::"] ips = ["1234:1234:abcd:abcd:1234:1234:1.0.0.127", "2001:db8::8d3::", "2001:db8:0:0:8d3::"]
valid_ips = ['1234:1234:abcd:abcd:1234:1234:1.0.0.127', "2001:db8:0:0:8d3::"] valid_ips = ['1234:1234:abcd:abcd:1234:1234:1.0.0.127', "2001:db8:0:0:8d3::"]
validated_ips = IPFinder(self.yara_ip_rules)._validate_ips(ips, socket.AF_INET6) validated_ips = IPFinder(self.yara_ip_rules)._validate_ips(ips, socket.AF_INET6)
self.assertEqual(validated_ips, valid_ips) assert validated_ips == valid_ips
def test_find_uri(self): def test_find_uri(self):
test_string = "http://www.google.de https://www.test.de/test/ " \ test_string = "http://www.google.de https://www.test.de/test/ " \
...@@ -68,4 +68,4 @@ class TestIpAndUrlFinder(unittest.TestCase): ...@@ -68,4 +68,4 @@ class TestIpAndUrlFinder(unittest.TestCase):
test_result = set(URIFinder(self.yara_uri_rules).find_uris(test_string)) test_result = set(URIFinder(self.yara_uri_rules).find_uris(test_string))
expected_result = {"http://www.google.de", "https://www.test.de/test/", "ftp://ftp.is.co.za/rfc/rfc1808.txt", expected_result = {"http://www.google.de", "https://www.test.de/test/", "ftp://ftp.is.co.za/rfc/rfc1808.txt",
"telnet://192.0.2.16:80/"} "telnet://192.0.2.16:80/"}
self.assertEqual(test_result, expected_result) assert test_result == expected_result
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment