diff --git a/src/plugins/analysis/cve_lookup/code/cve_lookup.py b/src/plugins/analysis/cve_lookup/code/cve_lookup.py index 61fca583e..e9b8dc5bd 100644 --- a/src/plugins/analysis/cve_lookup/code/cve_lookup.py +++ b/src/plugins/analysis/cve_lookup/code/cve_lookup.py @@ -46,9 +46,9 @@ def process_object(self, file_object: FileObject) -> FileObject: cves = {'cve_results': {}} connection = DbConnection(f'sqlite:///{DB_PATH}') lookup = Lookup(file_object, connection, match_any=self.match_any) - for value in file_object.processed_analysis['software_components']['result'].values(): - product = value['meta']['software_name'] - version = value['meta']['version'][0] + for sw_dict in file_object.processed_analysis['software_components']['result'].get('software_components', []): + product = sw_dict['name'] + version = sw_dict['versions'][0] if sw_dict['versions'] else None if product and version: vulnerabilities = lookup.lookup_vulnerabilities(product, version) if vulnerabilities: diff --git a/src/plugins/analysis/cve_lookup/test/test_cve_lookup.py b/src/plugins/analysis/cve_lookup/test/test_cve_lookup.py index 770b7f2eb..a78b6d085 100644 --- a/src/plugins/analysis/cve_lookup/test/test_cve_lookup.py +++ b/src/plugins/analysis/cve_lookup/test/test_cve_lookup.py @@ -18,19 +18,18 @@ SOFTWARE_COMPONENTS_ANALYSIS_RESULT = { 'result': { - 'dnsmasq': {'meta': {'software_name': 'Dnsmasq', 'version': ['2.40']}}, - 'OpenSSL': { - 'matches': True, - 'meta': { + 'software_components': [ + {'name': 'Dnsmasq', 'versions': ['2.40']}, + { + 'name': 'OpenSSL', 'description': 'SSL library', 'open_source': True, - 'software_name': 'OpenSSL', - 'version': [''], + 'versions': [], 'website': 'https://www.openssl.org', + 'rule': 'OpenSSL', + 'matching_strings': [{'offset': 7194, 'identifier': '$a', 'string': 'T1BFTlNTTA=='}], }, - 'rule': 'OpenSSL', - 'strings': [[7194, '$a', 'T1BFTlNTTA==']], - }, + ], }, 'analysis_date': 1563453634.37708, 'plugin_version': '0.3.2', diff --git a/src/plugins/analysis/known_vulnerabilities/code/known_vulnerabilities.py b/src/plugins/analysis/known_vulnerabilities/code/known_vulnerabilities.py index 9b310480f..d9ab4b8c2 100644 --- a/src/plugins/analysis/known_vulnerabilities/code/known_vulnerabilities.py +++ b/src/plugins/analysis/known_vulnerabilities/code/known_vulnerabilities.py @@ -22,7 +22,7 @@ class AnalysisPlugin(YaraBasePlugin): NAME = 'known_vulnerabilities' DESCRIPTION = 'Rule based detection of known vulnerabilities like Heartbleed' DEPENDENCIES = ['file_hashes', 'software_components'] # noqa: RUF012 - VERSION = '0.3.0' + VERSION = '0.3.1' FILE = __file__ def process_object(self, file_object): @@ -47,15 +47,18 @@ def process_object(self, file_object): def get_matched_vulnerabilities(self, yara_result: list[tuple[str, dict]], file_object) -> list[tuple[str, dict]]: software_components_results = file_object.processed_analysis.get('software_components', {}).get('result', {}) + software_by_name = { + sw_dict['name']: sw_dict for sw_dict in software_components_results.get('software_components', []) + } matched_vulnerabilities = self._check_vulnerabilities(file_object.processed_analysis) # CVE-2021-45608 NetUSB - if 'NetUSB' in software_components_results: + if 'NetUSB' in software_by_name: matched_vulnerabilities.extend(self._check_netusb_vulnerability(file_object.file_path)) # CVE-2024-3094 XZ Backdoor secondary detection - if 'liblzma' in software_components_results and not any(vuln == 'xz_backdoor' for vuln, _ in yara_result): - matched_vulnerabilities.extend(_check_xz_backdoor(software_components_results)) + if 'liblzma' in software_by_name and not any(vuln == 'xz_backdoor' for vuln, _ in yara_result): + matched_vulnerabilities.extend(_check_xz_backdoor(software_by_name['liblzma'])) return matched_vulnerabilities def add_tags(self, file_object, vulnerability_list): @@ -131,7 +134,7 @@ def _check_netusb_vulnerability(self, file_path: str) -> list[tuple[str, dict]]: def _check_xz_backdoor(software_results: dict) -> list[tuple[str, dict]]: - if any(v in software_results['liblzma']['meta']['version'] for v in ['5.6.0', '5.6.1']): + if any(v in software_results['versions'] for v in ['5.6.0', '5.6.1']): return [ ( 'XZ Backdoor', diff --git a/src/plugins/analysis/known_vulnerabilities/internal/rulebook.py b/src/plugins/analysis/known_vulnerabilities/internal/rulebook.py index b06c6df5f..e3e30ec79 100644 --- a/src/plugins/analysis/known_vulnerabilities/internal/rulebook.py +++ b/src/plugins/analysis/known_vulnerabilities/internal/rulebook.py @@ -1,3 +1,5 @@ +from __future__ import annotations + from copy import deepcopy RELATIONS = { @@ -36,7 +38,7 @@ def _make_type_assertions(self, link, rule): (self.score in ['low', 'medium', 'high'], 'score has to be one of low, medium or high'), (isinstance(self.description, str), 'description must be a string'), ( - isinstance(self.rule, (SingleRule, MetaRule, SubPathRule)), + isinstance(self.rule, (SingleRule, MetaRule, SubPathRule, SoftwareRule)), f'rule must be of type in [SingleRule, MetaRule, SubPathRule]. Has type {type(rule)}', ), (isinstance(self.link, str) or not link, 'if link is set it has to be a string'), @@ -95,6 +97,22 @@ def __init__(self, base_path, meta_rule): self.meta_rule = meta_rule +class SoftwareRule: + def __init__(self, software_name: str, affected_versions: set[str]): + self.software_name = software_name + self.affected_versions = affected_versions + + def evaluate(self, processed_analysis: dict[str, dict]) -> bool: + software_components = ( + processed_analysis.get('software_components', {}).get('result', {}).get('software_components', []) + ) + return any( + sw['name'] == self.software_name and v in self.affected_versions + for sw in software_components + for v in sw['versions'] + ) + + def evaluate(analysis, rule): try: if isinstance(rule, MetaRule): @@ -103,6 +121,8 @@ def evaluate(analysis, rule): result = _evaluate_single_rule(analysis, rule) elif isinstance(rule, SubPathRule): result = _evaluate_sub_path_rule(analysis, rule) + elif isinstance(rule, SoftwareRule): + result = rule.evaluate(analysis) else: raise TypeError('rule must be of one in types [SingleRule, MetaRule, SubPathRule]') return result @@ -153,10 +173,9 @@ def _get_dotted_path_from_dictionary(dictionary, dotted_path): def vulnerabilities(): - heartbleed_rule = SingleRule( - value_path=['software_components.result.OpenSSL.meta.version'], - relation='intersection', - comparison=[f'1.0.1{minor}' for minor in 'abcde'], + heartbleed_rule = SoftwareRule( + software_name='OpenSSL', + affected_versions={f'1.0.1{minor}' for minor in 'abcde'}, ) heartbleed_vulnerability = Vulnerability( rule=heartbleed_rule, diff --git a/src/plugins/analysis/known_vulnerabilities/test/data/sc.json b/src/plugins/analysis/known_vulnerabilities/test/data/sc.json index af4a3f1b8..bfab461ae 100644 --- a/src/plugins/analysis/known_vulnerabilities/test/data/sc.json +++ b/src/plugins/analysis/known_vulnerabilities/test/data/sc.json @@ -1,61 +1,56 @@ { - "analysis_date": 1517491696.332023, - "plugin_version": "0.3", + "analysis_date": 1737551071.128211, + "plugin_version": "1.0.0", "result": { - "OpenSSL": { - "matches": true, - "meta": { + "software_components": [ + { "description": "SSL library", + "matching_strings": [ + { + "identifier": "$a", + "offset": 13696143, + "string": "T3BlblNzbA==" + }, + { + "identifier": "$a", + "offset": 14147446, + "string": "T3BlblNTTCAwLjkuOA==" + } + ], + "name": "OpenSSL", "open_source": true, - "software_name": "OpenSSL", - "version": [ - "", + "rule": "OpenSSL", + "versions": [ "1.0.1a" ], "website": "https://www.openssl.org" }, - "rule": "OpenSSL", - "strings": [ - [ - 13696143, - "$a", - "T3BlblNzbA==" - ], - [ - 14147446, - "$a", - "T3BlblNTTCAwLjkuOA==" - ] - ] - }, - "VxWorks": { - "matches": true, - "meta": { + { "description": "Operating system for embedded devices", + "matching_strings": [ + { + "identifier": "$b", + "offset": 12748719, + "string": "dnh3b3Jrcy02LjU=" + }, + { + "identifier": "$b", + "offset": 12748792, + "string": "VnhXb3JrcyA2LjU=" + } + ], + "name": "VxWorks", "open_source": false, - "software_name": "VxWorks", - "version": [ + "rule": "VxWorks", + "versions": [ "6.5" ], "website": "http://www.windriver.com/products/vxworks/" - }, - "rule": "VxWorks", - "strings": [ - [ - 12748719, - "$b", - "dnh3b3Jrcy02LjU=" - ], - [ - 12748792, - "$b", - "VnhXb3JrcyA2LjU=" - ] - ] - } + } + ] }, "summary": [ - "OpenSSL ", + "OpenSSL", "OpenSSL 0.9.8", "VxWorks 6.5" ] diff --git a/src/plugins/analysis/known_vulnerabilities/test/test_known_vulnerabilities.py b/src/plugins/analysis/known_vulnerabilities/test/test_known_vulnerabilities.py index 2fca01a24..3fc04d827 100644 --- a/src/plugins/analysis/known_vulnerabilities/test/test_known_vulnerabilities.py +++ b/src/plugins/analysis/known_vulnerabilities/test/test_known_vulnerabilities.py @@ -41,8 +41,9 @@ def test_process_object_software(self, analysis_plugin): def test_process_object_software_wrong_version(self, analysis_plugin): test_file = FileObject(file_path=str(TEST_DATA_DIR / 'empty')) test_file.processed_analysis['file_hashes'] = {'result': {'sha256': '1234'}} - self._software_components_result['result']['OpenSSL']['meta']['version'] = ['0.9.8', '1.0.0', ''] - test_file.processed_analysis['software_components'] = self._software_components_result + software_components_result = {**self._software_components_result} + software_components_result['result']['software_components'][0]['versions'] = ['0.9.8', '1.0.0'] + test_file.processed_analysis['software_components'] = software_components_result results = analysis_plugin.process_object(test_file).processed_analysis[analysis_plugin.NAME] @@ -94,7 +95,9 @@ def test_xz_backdoor_1st(self, analysis_plugin): def test_xz_backdoor_2nd(self, analysis_plugin): test_file = FileObject(file_path=str(TEST_DATA_DIR / 'empty')) assert test_file.binary is not None - test_file.processed_analysis['software_components'] = {'result': {'liblzma': {'meta': {'version': ['5.6.1']}}}} + test_file.processed_analysis['software_components'] = { + 'result': {'software_components': [{'name': 'liblzma', 'versions': ['5.6.1']}]} + } fo = analysis_plugin.process_object(test_file) result = fo.processed_analysis['known_vulnerabilities'] assert 'XZ Backdoor' in result diff --git a/src/plugins/analysis/known_vulnerabilities/test/test_rulebook.py b/src/plugins/analysis/known_vulnerabilities/test/test_rulebook.py index 098e9749e..94f1443ad 100644 --- a/src/plugins/analysis/known_vulnerabilities/test/test_rulebook.py +++ b/src/plugins/analysis/known_vulnerabilities/test/test_rulebook.py @@ -4,6 +4,7 @@ RELATIONS, MetaRule, SingleRule, + SoftwareRule, SubPathRule, _evaluate_meta_rule, _evaluate_single_rule, @@ -13,6 +14,7 @@ evaluate, ) +# FixMe: result structure changed when IP&URI plugin was ported to new base class IPS = { 'ip_and_uri_finder': { 'summary': ['1', '2', '3', '4', 'a', 'b', 'c'], @@ -107,6 +109,32 @@ def test_evaluate_base_rule(): assert not _evaluate_sub_path_rule(IPS, sub_path_no_match) +def test_software_rule(): + rule = SoftwareRule(software_name='foo', affected_versions={'v1.2.2', 'v1.2.3'}) + processed_analysis_match = { + 'software_components': { + 'result': { + 'software_components': [ + {'name': 'foo', 'versions': ['v1.2.2']}, + ] + } + } + } + processed_analysis_no_match = { + 'software_components': { + 'result': { + 'software_components': [ + {'name': 'foo', 'versions': ['v1.2.1', 'v1.2.4']}, + {'name': 'bar', 'versions': ['v1.2.2']}, + ] + } + } + } + + assert rule.evaluate(processed_analysis_match) is True + assert rule.evaluate(processed_analysis_no_match) is False + + def test_evaluate_bad_type(): with pytest.raises(TypeError): evaluate({}, object()) diff --git a/src/plugins/analysis/software_components/code/software_components.py b/src/plugins/analysis/software_components/code/software_components.py index 9c59bc267..cefa760dc 100644 --- a/src/plugins/analysis/software_components/code/software_components.py +++ b/src/plugins/analysis/software_components/code/software_components.py @@ -2,11 +2,14 @@ import re import string -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, List, Optional + +from pydantic import BaseModel, Field +from semver import Version import config -from analysis.YaraPluginBase import YaraBasePlugin -from helperFunctions.data_conversion import make_unicode_string +from analysis.plugin import AnalysisPluginV0, Tag, addons +from analysis.plugin.compat import AnalysisBasePluginAdapterMixin from helperFunctions.tag import TagColor from plugins.analysis.software_components.bin import OS_LIST from plugins.mime_blacklists import MIME_BLACKLIST_NON_EXECUTABLE @@ -14,113 +17,152 @@ from ..internal.resolve_version_format_string import extract_data_from_ghidra if TYPE_CHECKING: - from objects.file import FileObject + from io import FileIO + import yara -class AnalysisPlugin(YaraBasePlugin): - """ - This plugin identifies software components - Credits: - OS Tagging functionality created by Roman Konertz during Firmware Bootcamp WT17/18 at University of Bonn - Maintained by Fraunhofer FKIE - """ +class SoftwareMatch(BaseModel): + name: str + versions: List[str] + rule: str = Field(description='Matching YARA rule name') + matching_strings: List[MatchingString] + description: Optional[str] = None + open_source: Optional[bool] = None + website: Optional[str] = Field(None, description='Website URL of the software') - NAME = 'software_components' - DESCRIPTION = 'identify software components' - MIME_BLACKLIST = MIME_BLACKLIST_NON_EXECUTABLE - VERSION = '0.5.1' - FILE = __file__ - def process_object(self, file_object): - file_object = super().process_object(file_object) - analysis = file_object.processed_analysis[self.NAME] - if len(analysis) > 1: - analysis = self.add_version_information(analysis, file_object) - analysis['summary'] = self._get_summary(analysis) +class MatchingString(BaseModel): + string: str + offset: int + identifier: str = Field(description='Identifier of the rule that this string matched (e.g. "$a")') - self.add_os_key(file_object) - return file_object - def get_version(self, input_string: str, meta_dict: dict) -> str: - if 'version_regex' in meta_dict: - regex = meta_dict['version_regex'].replace('\\\\', '\\') - else: - regex = r'\d+.\d+(.\d+)?(\w)?' - pattern = re.compile(regex) - version = pattern.search(input_string) - if version is not None: - return self._strip_leading_zeroes(version.group(0)) - return '' - - @staticmethod - def _get_summary(results: dict) -> list[str]: - summary = set() - for key, result in results.items(): - if key != 'summary': - software = result['meta']['software_name'] - for version in result['meta']['version']: - summary.add(f'{software} {version}') - return sorted(summary) +class AnalysisPlugin(AnalysisPluginV0, AnalysisBasePluginAdapterMixin): + class Schema(BaseModel): + software_components: List[SoftwareMatch] - def add_version_information(self, results, file_object: FileObject): - for item in results: - if item != 'summary': - results[item] = self.get_version_for_component(results[item], file_object) - return results - - def get_version_for_component(self, result, file_object: FileObject): - versions = set() - for matched_string in result['strings']: - match = matched_string[2].strip() - match = make_unicode_string(match) - versions.add(self.get_version(match, result['meta'])) - if any(k in result['meta'] for k in ('format_string', '_version_function')): - if result['meta'].get('format_string'): - input_data = { - 'mode': 'format_string', - 'key_string_list': [s for _, _, s in result['strings'] if '%s' in s], - } - else: - input_data = { - 'mode': 'version_function', - 'function_name': result['meta']['_version_function'], - } - versions.update( - extract_data_from_ghidra(file_object.file_path, input_data, config.backend.docker_mount_base_dir) + def __init__(self): + super().__init__( + metadata=( + self.MetaData( + name='software_components', + description='identify software components', + mime_blacklist=MIME_BLACKLIST_NON_EXECUTABLE, + version=Version(1, 0, 0), + Schema=self.Schema, + ) ) - if '' in versions and len(versions) > 1: # if there are actual version results, remove the "empty" result - versions.remove('') - result['meta']['version'] = list(versions) - return result + ) + self._yara = addons.Yara(plugin=self) + + def analyze(self, file_handle: FileIO, virtual_file_path: dict, analyses: dict[str, BaseModel]) -> Schema: + del virtual_file_path, analyses + return self.Schema( + software_components=[ + SoftwareMatch( + name=match.meta.get('software_name'), + rule=match.rule, + matching_strings=_get_matching_strings(match), + versions=get_version_for_component(match, file_handle), + description=match.meta.get('description'), + website=match.meta.get('website'), + open_source=match.meta.get('open_source'), + ) + for match in self._yara.match(file_handle) + ] + ) + + def summarize(self, result: Schema) -> list[str]: + summary = set() + for software in result.software_components: + if software.versions: + for version in software.versions: + summary.add(f'{software.name} {version}') + else: + summary.add(software.name) + return sorted(summary) - def add_os_key(self, file_object): - for entry in file_object.processed_analysis[self.NAME]['summary']: + def get_tags(self, result: Schema, summary: list[str]) -> list[Tag]: + del result + tags = [] + for entry in summary: for os_ in OS_LIST: if entry.find(os_) != -1: - if self._entry_has_no_trailing_version(entry, os_): - self.add_analysis_tag(file_object, 'OS', entry, TagColor.GREEN, True) + if _entry_has_no_trailing_version(entry, os_): + tags.append(Tag(name='OS', value=entry, color=TagColor.GREEN, propagate=True)) else: - self.add_analysis_tag(file_object, 'OS', os_, TagColor.GREEN, False) - self.add_analysis_tag(file_object, 'OS Version', entry, TagColor.GREEN, True) - - @staticmethod - def _entry_has_no_trailing_version(entry, os_string): - return os_string.strip() == entry.strip() - - @staticmethod - def _strip_leading_zeroes(version_string: str) -> str: - prefix, suffix = '', '' - while version_string and version_string[0] not in string.digits: - prefix += version_string[0] - version_string = version_string[1:] - while version_string and version_string[-1] not in string.digits: - suffix = version_string[-1] + suffix - version_string = version_string[:-1] - elements = [] - for element in version_string.split('.'): - try: - elements.append(str(int(element))) - except ValueError: - elements.append(element) - return prefix + '.'.join(elements) + suffix + tags.append(Tag(name='OS', value=os_, color=TagColor.GREEN, propagate=False)) + tags.append(Tag(name='OS Version', value=entry, color=TagColor.GREEN, propagate=True)) + return tags + + +def _get_matching_strings(match: yara.Match) -> list[MatchingString]: + return [ + MatchingString( + string=instance.matched_data.decode(errors='replace'), + offset=instance.offset, + identifier=_string.identifier, + ) + for _string in match.strings # type: yara.StringMatch + for instance in _string.instances # type: yara.StringMatchInstance + ] + + +def get_version_for_component(match: yara.Match, file: FileIO) -> list[str]: + matching_strings = _get_strings_from_match(match) + versions = {get_version(matching_str, match.meta) for matching_str in matching_strings} + if any(k in match.meta for k in ('format_string', '_version_function')): + if match.meta.get('format_string'): + input_data = { + 'mode': 'format_string', + 'key_string_list': [s for s in matching_strings if '%s' in s], + } + else: + input_data = { + 'mode': 'version_function', + 'function_name': match.meta['_version_function'], + } + versions.update(extract_data_from_ghidra(file.name, input_data, config.backend.docker_mount_base_dir)) + return [v for v in versions if v] + + +def get_version(input_string: str, meta_dict: dict) -> str | None: + if 'version_regex' in meta_dict: + regex = meta_dict['version_regex'].replace('\\\\', '\\') + else: + regex = r'\d+.\d+(.\d+)?(\w)?' + pattern = re.compile(regex) + version = pattern.search(input_string) + if version is not None: + return _strip_leading_zeroes(version.group(0)) + return None + + +def _get_strings_from_match(match: yara.Match) -> list[str]: + return [ + instance.matched_data.decode(errors='replace').strip() + for string_match in match.strings + for instance in string_match.instances + ] + + +def _entry_has_no_trailing_version(entry, os_string): + return os_string.strip() == entry.strip() + + +def _strip_leading_zeroes(version_string: str) -> str: + prefix, suffix = '', '' + while version_string and version_string[0] not in string.digits: + prefix += version_string[0] + version_string = version_string[1:] + while version_string and version_string[-1] not in string.digits: + suffix = version_string[-1] + suffix + version_string = version_string[:-1] + elements = [] + for element in version_string.split('.'): + try: + elements.append(str(int(element))) + except ValueError: + elements.append(element) + return prefix + '.'.join(elements) + suffix diff --git a/src/plugins/analysis/software_components/test/test_plugin_software_components.py b/src/plugins/analysis/software_components/test/test_plugin_software_components.py index 439532383..beb5d8a2d 100644 --- a/src/plugins/analysis/software_components/test/test_plugin_software_components.py +++ b/src/plugins/analysis/software_components/test/test_plugin_software_components.py @@ -2,46 +2,44 @@ import pytest -from objects.file import FileObject +from ..code.software_components import AnalysisPlugin, _entry_has_no_trailing_version, get_version -from ..code.software_components import AnalysisPlugin - -YARA_TEST_FILE = str(Path(__file__).parent / 'data' / 'yara_test_file') +YARA_TEST_FILE = Path(__file__).parent / 'data' / 'yara_test_file' @pytest.mark.AnalysisPluginTestConfig(plugin_class=AnalysisPlugin) class TestAnalysisPluginsSoftwareComponents: def test_process_object(self, analysis_plugin): - test_file = FileObject(file_path=YARA_TEST_FILE) - - processed_file = analysis_plugin.process_object(test_file) - results = processed_file.processed_analysis[analysis_plugin.NAME] - assert len(results) == 2, 'incorrect number of software components found' - assert 'MyTestRule' in results, 'test Rule match not found' - assert ( - results['MyTestRule']['meta']['software_name'] == 'Test Software' - ), 'incorrect software name from yara meta' - assert ( - results['MyTestRule']['meta']['website'] == 'http://www.fkie.fraunhofer.de' - ), 'incorrect website from yara meta' - assert ( - results['MyTestRule']['meta']['description'] == 'This is a test rule' - ), 'incorrect description from yara meta' - assert results['MyTestRule']['meta']['open_source'], 'incorrect open-source flag from yara meta' - assert (10, '$a', 'MyTestRule 0.1.3.') in results['MyTestRule']['strings'], 'string not found' - assert '0.1.3' in results['MyTestRule']['meta']['version'], 'Version not detected' - assert len(results['MyTestRule']['strings']) == 1, 'to much strings found' - assert len(results['summary']) == 1, 'Number of summary results not correct' - assert 'Test Software 0.1.3' in results['summary'] + with YARA_TEST_FILE.open('rb') as fp: + results = analysis_plugin.analyze(fp, {}, {}) + + assert len(results.software_components) == 1, 'incorrect number of software components found' + software_result = results.software_components[0] + assert software_result.rule == 'MyTestRule', 'incorrect yara rule name' + assert software_result.name == 'Test Software', 'incorrect software name from yara meta' + assert software_result.website == 'http://www.fkie.fraunhofer.de', 'incorrect website from yara meta' + assert software_result.description == 'This is a test rule', 'incorrect description from yara meta' + assert software_result.open_source, 'incorrect open-source flag from yara meta' + + assert len(software_result.matching_strings) == 1, 'too many strings found' + string_match = software_result.matching_strings[0] + assert string_match.string == 'MyTestRule 0.1.3.', 'string not found' + assert string_match.offset == 10 + assert string_match.identifier == '$a' + assert '0.1.3' in software_result.versions, 'Version not detected' + + summary = analysis_plugin.summarize(results) + assert len(summary) == 1, 'Number of summary results not correct' + assert 'Test Software 0.1.3' in summary @pytest.mark.parametrize( ('version', 'expected_output', 'meta_dict'), [ - ('', '', {}), + ('', None, {}), ('Foo 15.14.13', '15.14.13', {}), ('Foo 1.0', '1.0', {}), ('Foo 1.1.1b', '1.1.1b', {}), - ('Foo', '', {}), + ('Foo', None, {}), ('Foo 01.02.03', '1.2.3', {}), ('Foo 00.1.', '0.1', {}), ('\x001.22.333\x00', '1.22.333', {}), @@ -54,43 +52,21 @@ def test_process_object(self, analysis_plugin): ], ) def test_get_version(self, analysis_plugin, version, expected_output, meta_dict): - assert analysis_plugin.get_version(version, meta_dict) == expected_output, f'{version} not found correctly' + assert get_version(version, meta_dict) == expected_output, f'{version} not found correctly' def test_get_version_from_meta(self, analysis_plugin): version = 'v15.14.1a' assert ( - analysis_plugin.get_version(f'Foo {version}', {'version_regex': 'v\\d\\d\\.\\d\\d\\.\\d[a-z]'}) == version + get_version(f'Foo {version}', {'version_regex': 'v\\d\\d\\.\\d\\d\\.\\d[a-z]'}) == version ), 'version not found correctly' def test_entry_has_no_trailing_version(self, analysis_plugin): - assert not analysis_plugin._entry_has_no_trailing_version('Linux', 'Linux 4.15.0-22') - assert analysis_plugin._entry_has_no_trailing_version('Linux', 'Linux') - assert analysis_plugin._entry_has_no_trailing_version(' Linux', 'Linux ') - - def test_add_os_key_fail(self, analysis_plugin): - test_file = FileObject(file_path=YARA_TEST_FILE) - with pytest.raises(KeyError): - analysis_plugin.add_os_key(test_file) - - test_file.processed_analysis[analysis_plugin.NAME] = {'summary': ['OpenSSL']} - analysis_plugin.add_os_key(test_file) - assert 'tags' not in test_file.processed_analysis[analysis_plugin.NAME] - - def test_add_os_key_success(self, analysis_plugin): - test_file = FileObject(file_path=YARA_TEST_FILE) - test_file.processed_analysis[analysis_plugin.NAME] = {'summary': ['Linux Kernel']} - analysis_plugin.add_os_key(test_file) - assert 'tags' in test_file.processed_analysis[analysis_plugin.NAME] - assert test_file.processed_analysis[analysis_plugin.NAME]['tags']['OS']['value'] == 'Linux Kernel' - - def test_update_os_key(self, analysis_plugin): - test_file = FileObject(file_path=YARA_TEST_FILE) - - test_file.processed_analysis[analysis_plugin.NAME] = { - 'summary': ['Linux Kernel'], - 'tags': {'OS': {'value': 'Fire OS'}}, - } - - assert test_file.processed_analysis[analysis_plugin.NAME]['tags']['OS']['value'] == 'Fire OS' - analysis_plugin.add_os_key(test_file) - assert test_file.processed_analysis[analysis_plugin.NAME]['tags']['OS']['value'] == 'Linux Kernel' + assert not _entry_has_no_trailing_version('Linux', 'Linux 4.15.0-22') + assert _entry_has_no_trailing_version('Linux', 'Linux') + assert _entry_has_no_trailing_version(' Linux', 'Linux ') + + def test_get_tags(self, analysis_plugin): + assert analysis_plugin.get_tags({}, ['OpenSSL']) == [] + tags = analysis_plugin.get_tags({}, ['Linux Kernel']) + assert tags != [] + assert tags[0].value == 'Linux Kernel' diff --git a/src/plugins/analysis/software_components/view/software_components.html b/src/plugins/analysis/software_components/view/software_components.html index 2a6f76d18..9bbc328ef 100644 --- a/src/plugins/analysis/software_components/view/software_components.html +++ b/src/plugins/analysis/software_components/view/software_components.html @@ -2,7 +2,7 @@ {% block analysis_result_details %} - {% for key in analysis_result %} + {% for software_data in analysis_result.software_components %} {{ loop.index - 1 }} @@ -14,29 +14,62 @@ Software Name - {{ analysis_result[key]['meta']['software_name'] }} + {{ software_data.name }} Version - {% for version in analysis_result[key]['meta']['version'] %} - {{ version }} - {% endfor %} - - - - Description - {{ analysis_result[key]['meta']['description'] }} - - - Website - - {{ analysis_result[key]['meta']['website'] }} + {% if software_data.versions | length > 1 %} + + {% elif software_data.versions | length == 1 %} + {{ software_data.versions[0] }} + {% else %} + N/A + {% endif %} + {% if software_data.description %} + + Description + {{ software_data.description }} + + {% endif %} + {% if software_data.website %} + + Website + + {{ software_data.website }} + + + {% endif %} + {% if software_data.open_source != None %} + + Open Source + {{ software_data.open_source }} + + {% endif %} Matched Sequence - {{ analysis_result[key]['strings'] }} + + + + + + + + {% for string_data in software_data.matching_strings %} + + + + + + {% endfor %} +
identifieroffsetstring
{{ string_data.identifier }}{{ string_data.offset }}{{ string_data.string }}
+ diff --git a/src/storage/db_interface_stats.py b/src/storage/db_interface_stats.py index 433b73063..3b05e5d6d 100644 --- a/src/storage/db_interface_stats.py +++ b/src/storage/db_interface_stats.py @@ -2,7 +2,7 @@ import logging from collections import Counter -from typing import TYPE_CHECKING, Any, Callable, Iterator, List, Tuple +from typing import TYPE_CHECKING, Any, Callable, Iterable, Iterator, List, Tuple from sqlalchemy import column, func, select from sqlalchemy.exc import SQLAlchemyError @@ -249,21 +249,25 @@ def get_release_date_stats(self, q_filter: dict | None = None) -> list[tuple[int def get_software_components(self, q_filter: dict | None = None) -> Stats: with self.get_read_only_session() as session: - subquery = ( - select(func.jsonb_object_keys(AnalysisEntry.result).label('software'), AnalysisEntry.uid) - .filter(AnalysisEntry.plugin == 'software_components') - .subquery('subquery') - ) + label = 'software' query = ( - select(subquery.c.software, func.count(subquery.c.software)) - .filter(subquery.c.software.notin_(['system_version', 'skipped'])) - .group_by(subquery.c.software) + select( + func.jsonb_array_elements(AnalysisEntry.result['software_components']).label(label), + func.count(label), + ) + .filter(AnalysisEntry.plugin == 'software_components') + .group_by(label) ) if self._filter_is_not_empty(q_filter): - query = query.join(FileObjectEntry, FileObjectEntry.uid == subquery.c.uid) + query = query.join(FileObjectEntry, FileObjectEntry.uid == AnalysisEntry.uid) query = query.join(FirmwareEntry, FileObjectEntry.root_firmware.any(uid=FirmwareEntry.uid)) query = query.filter_by(**q_filter) - return _sort_tuples(session.execute(query)) + + result = {} + for software_dict, count in session.execute(query): + result.setdefault(software_dict['name'], 0) + result[software_dict['name']] += count + return _sort_tuples(result.items()) @staticmethod def _join_fw_or_fo(query: Select, is_firmware: bool) -> Select: @@ -293,7 +297,7 @@ def count_occurrences(result_list: list[str]) -> Stats: return _sort_tuples(Counter(result_list).items()) -def _sort_tuples(query_result: Stats) -> Stats: +def _sort_tuples(query_result: Iterable[Tuple[str, int]]) -> Stats: # Sort stats tuples by count in ascending order return sorted( _convert_to_tuples(query_result), key=lambda e: (e[1], e[0]) if not isinstance(e[0], dict) else (e[1],) diff --git a/src/test/integration/statistic/test_update.py b/src/test/integration/statistic/test_update.py index e45370817..5c123e9a4 100644 --- a/src/test/integration/statistic/test_update.py +++ b/src/test/integration/statistic/test_update.py @@ -297,7 +297,9 @@ def test_get_software_components_stats(stats_updater, backend_db): 'fo1', parent_fw='root_fw', analysis={ - 'software_components': generate_analysis_entry(analysis_result={'LinuxKernel': {'foo': 'bar'}}), + 'software_components': generate_analysis_entry( + analysis_result={'software_components': [{'name': 'LinuxKernel', 'versions': []}]} + ), }, ) insert_test_fo( @@ -305,7 +307,9 @@ def test_get_software_components_stats(stats_updater, backend_db): 'fo2', parent_fw='root_fw', analysis={ - 'software_components': generate_analysis_entry(analysis_result={'LinuxKernel': {'foo': 'bar'}}), + 'software_components': generate_analysis_entry( + analysis_result={'software_components': [{'name': 'LinuxKernel', 'versions': []}]} + ), }, ) insert_test_fo( @@ -313,7 +317,9 @@ def test_get_software_components_stats(stats_updater, backend_db): 'fo3', parent_fw='root_fw', analysis={ - 'software_components': generate_analysis_entry(analysis_result={'SomeSoftware': {'foo': 'bar'}}), + 'software_components': generate_analysis_entry( + analysis_result={'software_components': [{'name': 'SomeSoftware', 'versions': []}]} + ), }, )