Merge remote-tracking branch 'origin/master' into system-health-stage3

src/helperFunctions/docker.py

@@ -4,14 +4,14 @@
 
 import docker
 from docker.errors import APIError, DockerException, ImageNotFound
-from requests.exceptions import ReadTimeout
+from requests.exceptions import ReadTimeout, RequestException
 
 
 def run_docker_container(
     image: str, logging_label: str = 'Docker', timeout: int = 300, combine_stderr_stdout: bool = False, **kwargs
 ) -> CompletedProcess:
     """
-    This is a convinience function that runs a docker container and returns a
+    This is a convenience function that runs a docker container and returns a
     subprocess.CompletedProcess instance for the command ran in the container.
     All remaining keyword args are passed to `docker.containers.run`.
 
@@ -51,6 +51,9 @@ def run_docker_container(
     except ReadTimeout:
         logging.warning(f'[{logging_label}]: timeout while processing')
         raise
+    except RequestException:
+        logging.warning(f'[{logging_label}]: connection error while processing')
+        raise
     except APIError:
         logging.warning(f'[{logging_label}]: encountered docker error while processing')
         raise

src/plugins/analysis/cwe_checker/code/cwe_checker.py

@@ -37,11 +37,10 @@ class AnalysisPlugin(AnalysisBasePlugin):
         'Due to the nature of static analysis, this plugin may run for a long time.'
     )
     DEPENDENCIES = ['cpu_architecture', 'file_type']  # noqa: RUF012
-    VERSION = '0.5.3'
+    VERSION = '0.5.4'
     TIMEOUT = 600  # 10 minutes
     MIME_WHITELIST = [  # noqa: RUF012
         'application/x-executable',
-        'application/x-object',
         'application/x-pie-executable',
         'application/x-sharedlib',
     ]

src/scheduler/unpacking_scheduler.py

@@ -22,6 +22,7 @@
 )
 from objects.firmware import Firmware
 from storage.db_interface_backend import BackendDbInterface
+from storage.db_interface_base import DbInterfaceError
 from unpacker.extraction_container import ExtractionContainer
 from unpacker.unpack import Unpacker
 from unpacker.unpack_base import ExtractionError
@@ -205,8 +206,12 @@ def work_thread(self, task: FileObject, container: ExtractionContainer):
             logging.info(f'Unpacking completed: {task.uid} (extracted files: {len(extracted_objects)})')
             # each worker needs its own interface because connections are not thread-safe
             db_interface = self.db_interface()
-            db_interface.add_object(task)  # save FO before submitting to analysis scheduler
-            self.post_unpack(task)
+            try:
+                db_interface.add_object(task)  # save FO before submitting to analysis scheduler
+                self.post_unpack(task)
+            except DbInterfaceError as error:
+                logging.error(str(error))
+                extracted_objects = []
             self._update_currently_unpacked(task, extracted_objects, db_interface)
             self._schedule_extracted_files(extracted_objects)
 

src/storage/db_interface_backend.py

@@ -149,6 +149,14 @@ def add_child_to_parent(self, parent_uid: str, child_uid: str):
 
     def update_object(self, fw_object: FileObject):
         if isinstance(fw_object, Firmware):
+            if not self.is_firmware(fw_object.uid):
+                # special case: Trying to upload a file as firmware that is already in the DB as part of another
+                # firmware. This is currently not possible and will likely cause errors
+                parent_fw = self.get_parent_fw(fw_object.uid)
+                raise DbInterfaceError(
+                    'Cannot upload file as firmware that is part of another firmware. '
+                    f'The file you are trying to upload is already part of the following firmware images: {parent_fw}'
+                )
             self.update_firmware(fw_object)
         self.update_file_object(fw_object)
 

src/test/integration/storage/test_db_interface_backend.py

@@ -2,6 +2,7 @@
 
 import pytest
 
+from storage.db_interface_base import DbInterfaceError
 from test.common_helper import create_test_file_object, create_test_firmware
 
 from .helper import TEST_FO, TEST_FW, create_fw_with_child_fo, create_fw_with_parent_and_child, add_included_file
@@ -143,6 +144,17 @@ def test_update_duplicate_same_fw(backend_db, frontend_db):
     assert db_fo.parents == {fw.uid}
 
 
+def test_update_duplicate_file_as_fw(backend_db):
+    # special case: trying to upload a file as FW that is already in the DB as part of another FW -> should cause error
+    fo, fw = create_fw_with_child_fo()
+    backend_db.insert_multiple_objects(fw, fo)
+    fw2 = create_test_firmware()
+    fw2.uid = fo.uid
+
+    with pytest.raises(DbInterfaceError):
+        backend_db.add_object(fw2)
+
+
 def test_analysis_exists(backend_db):
     assert backend_db.analysis_exists(TEST_FO.uid, 'file_type') is False
     backend_db.insert_file_object(TEST_FO)

src/test/integration/web_interface/rest/test_rest_firmware.py

@@ -7,6 +7,20 @@
 from test.integration.web_interface.rest.base import RestTestBase
 
 
+UPLOAD_DATA = {
+    'binary': standard_b64encode(b'test_file_content').decode(),
+    'file_name': 'test_file.txt',
+    'device_name': 'test_device',
+    'device_part': 'full',
+    'device_class': 'test_class',
+    'version': '1',
+    'vendor': 'test_vendor',
+    'release_date': '1970-01-01',
+    'tags': '',
+    'requested_analysis_systems': ['dummy'],
+}
+
+
 @pytest.mark.usefixtures('database_interfaces')
 class TestRestFirmware(RestTestBase):
     def test_rest_firmware_existing(self, backend_db):
@@ -47,35 +61,33 @@ def test_rest_search_not_existing(self, backend_db):
         rv = self.test_client.get(f'/rest/firmware?query={query}', follow_redirects=True)
         assert b'"uids": []' in rv.data
 
-    def test_rest_upload_valid(self):
-        data = {
-            'binary': standard_b64encode(b'test_file_content').decode(),
-            'file_name': 'test_file.txt',
-            'device_name': 'test_device',
-            'device_part': 'full',
-            'device_class': 'test_class',
-            'version': '1',
-            'vendor': 'test_vendor',
-            'release_date': '1970-01-01',
-            'tags': '',
-            'requested_analysis_systems': ['dummy'],
-        }
-        rv = self.test_client.put('/rest/firmware', json=data, follow_redirects=True)
+    def test_rest_upload_valid(self, monkeypatch):
+        monkeypatch.setattr(
+            'intercom.front_end_binding.InterComFrontEndBinding.get_available_analysis_plugins',
+            lambda _: ['dummy'],
+        )
+        rv = self.test_client.put('/rest/firmware', json=UPLOAD_DATA, follow_redirects=True)
         assert b'c1f95369a99b765e93c335067e77a7d91af3076d2d3d64aacd04e1e0a810b3ed_17' in rv.data
         assert b'"status": 0' in rv.data
 
-    def test_rest_upload_invalid(self):
+    def test_upload_unknown_plugin(self, monkeypatch):
+        monkeypatch.setattr(
+            'intercom.front_end_binding.InterComFrontEndBinding.get_available_analysis_plugins',
+            lambda _: ['plugin_1'],
+        )
         data = {
-            'binary': standard_b64encode(b'test_file_content').decode(),
-            'file_name': 'test_file.txt',
-            'device_name': 'test_device',
-            'device_part': 'test_part',
-            'device_class': 'test_class',
-            'vendor': 'test_vendor',
-            'release_date': '01.01.1970',
-            'tags': '',
-            'requested_analysis_systems': ['dummy'],
+            **UPLOAD_DATA,
+            'requested_analysis_systems': ['plugin_1', 'plugin_2'],
         }
+        response = self.test_client.put('/rest/firmware', json=data, follow_redirects=True).json
+        assert 'error_message' in response
+        assert 'The requested analysis plugins are not available' in response['error_message']
+        assert 'plugin_2' in response['error_message']
+        assert 'plugin_1' not in response['error_message']
+
+    def test_rest_upload_invalid(self):
+        data = {**UPLOAD_DATA}
+        data.pop('version')
         rv = self.test_client.put('/rest/firmware', json=data, follow_redirects=True)
         assert rv.json['message'] == 'Input payload validation failed'
         assert 'version' in rv.json['errors']

src/test/integration/web_interface/test_filter.py

@@ -25,11 +25,11 @@ def test_list_group_collapse(frontend):
 @pytest.mark.parametrize(
     ('tag_dict', 'output'),
     [
-        ({'a': 'danger'}, '<span class="badge badge-danger mr-2" style="font-size: 14px;" > a</span>'),
+        ({'a': 'danger'}, '<span class="badge badge-danger mr-2" style="font-size: 14px;"  > a</span>'),
         (
             {'a': 'danger', 'b': 'primary'},
-            '<span class="badge badge-danger mr-2" style="font-size: 14px;" > a</span>'
-            '<span class="badge badge-primary mr-2" style="font-size: 14px;" > b</span>',
+            '<span class="badge badge-danger mr-2" style="font-size: 14px;"  > a</span>'
+            '<span class="badge badge-primary mr-2" style="font-size: 14px;"  > b</span>',
         ),
         (None, ''),
     ],
@@ -51,6 +51,16 @@ def test_render_analysis_tags_success(frontend):
     assert '> wow<' in output
 
 
+def test_render_analysis_tags_link(frontend):
+    plugin, uid, root_uid = 'plugin1', 'foo', 'bar'
+    tags = {plugin: {'tag': {'color': 'success', 'value': 'some_value'}}}
+    with frontend.app.app_context():
+        output = render_analysis_tags(tags, uid=uid, root_uid=root_uid)
+    assert 'onclick' in output
+    link = f'/analysis/{uid}/{plugin}/ro/{root_uid}?load_summary=true'
+    assert link in output
+
+
 def test_render_analysis_tags_fix(frontend):
     tags = {'such plugin': {'tag': {'color': 'very color', 'value': 'wow'}}}
     with frontend.app.app_context():

src/web_interface/components/ajax_routes.py

@@ -6,6 +6,7 @@
 
 from helperFunctions.data_conversion import none_to_none
 from helperFunctions.database import get_shared_session
+from objects.firmware import Firmware
 from web_interface.components.component_base import AppRoute, ComponentBase, GET
 from web_interface.components.hex_highlighting import preview_data_as_hex
 from web_interface.file_tree.file_tree import remove_virtual_path_from_root
@@ -114,10 +115,11 @@ def ajax_get_summary(self, uid, selected_analysis):
         with get_shared_session(self.db.frontend) as frontend_db:
             firmware = frontend_db.get_object(uid, analysis_filter=selected_analysis)
             summary_of_included_files = frontend_db.get_summary(firmware, selected_analysis)
+            root_uid = uid if isinstance(firmware, Firmware) else frontend_db.get_root_uid(uid)
         return render_template(
             'summary.html',
             summary_of_included_files=summary_of_included_files,
-            root_uid=uid,
+            root_uid=root_uid,
             selected_analysis=selected_analysis,
         )
 

src/web_interface/filter.py

@@ -17,7 +17,7 @@
 from re import Match
 from string import ascii_letters
 from time import localtime, strftime, struct_time, time
-from typing import Union
+from typing import Union, Iterable
 
 from common_helper_files import human_readable_file_size
 from flask import render_template
@@ -283,11 +283,11 @@ def render_fw_tags(tag_dict, size=14):
     return output
 
 
-def render_analysis_tags(tags, size=14):
+def render_analysis_tags(tags, uid=None, root_uid=None, size=14):
     output = ''
     if tags:
-        for plugin_name in tags:
-            for key, tag in tags[plugin_name].items():
+        for plugin_name in sorted(tags):
+            for key, tag in sorted(tags[plugin_name].items(), key=_sort_tags_key):
                 if key == 'root_uid':
                     continue
                 color = tag['color'] if tag['color'] in TagColor.ALL else TagColor.BLUE
@@ -297,10 +297,19 @@ def render_analysis_tags(tags, size=14):
                     value=tag['value'],
                     tooltip=f'{plugin_name}: {key}',
                     size=size,
+                    plugin=plugin_name,
+                    root_uid=root_uid,
+                    uid=uid,
                 )
     return output
 
 
+def _sort_tags_key(tag_tuples: Iterable[tuple[str, dict]]) -> str:
+    # Sort tags by "value" (the displayed text in the tag). There can be 'root_uid' entries which are no dicts
+    _, tag_dict = tag_tuples
+    return tag_dict['value'] if isinstance(tag_dict, dict) else ''
+
+
 def fix_cwe(string):
     if 'CWE' in string:
         return string.split(']')[0].split('E')[-1]

src/web_interface/rest/rest_firmware.py

@@ -113,6 +113,16 @@ def put(self):
         except MarshallingError as error:
             logging.error(f'REST|firmware|PUT: Error in payload data: {error}')
             return error_message(str(error), self.URL)
+
+        available_plugins = set(self.intercom.get_available_analysis_plugins()) - {'unpacker'}
+        unavailable_plugins = set(data['requested_analysis_systems']) - available_plugins
+        if unavailable_plugins:
+            return error_message(
+                f'The requested analysis plugins are not available: {", ".join(unavailable_plugins)}',
+                self.URL,
+                request_data={k: v for k, v in data.items() if k != 'binary'},  # don't send the firmware binary back
+            )
+
         result = self._process_data(data)
         if 'error_message' in result:
             logging.warning('Submission not according to API guidelines! (data could not be parsed)')

src/web_interface/static/js/show_analysis_summary.js

@@ -1,9 +1,14 @@
-function load_summary(uid, selected_analysis){
+function load_summary(uid, selected_analysis, focus=false){
     $("#summary-button").css("display", "none");
     let summary_gif = $("#loading-summary-gif");
     summary_gif.css("display", "block");
     $("#summary-div").load(
         `/ajax_get_summary/${uid}/${selected_analysis}`,
-        () => {summary_gif.css("display", "none");}
+        () => {
+            summary_gif.css("display", "none");
+            if (focus === true) {
+                location.href = "#summary-heading";
+            }
+        }
     );
 }

src/web_interface/templates/generic_view/tags.html

@@ -5,6 +5,9 @@
         data-toggle="tooltip"
         title="{{ tooltip | replace_underscore }}"
     {%- endif %}
+    {% if uid -%}
+        onclick="location.href='http://localhost:5000/analysis/{{ uid }}/{{ plugin }}/ro/{{ root_uid }}?load_summary=true'"
+    {%- endif %}
 >
     {{ value }}
 </span>

src/web_interface/templates/show_analysis.html

@@ -114,7 +114,7 @@
             <h3>
                 {{ firmware.uid | replace_uid_with_hid(root_uid=root_uid) | safe }}<br />
                 {% if firmware.analysis_tags or firmware.tags %}
-                    {{ firmware.analysis_tags | render_analysis_tags | safe }}{{ firmware.tags | render_fw_tags | safe }}<br />
+                    {{ firmware.analysis_tags | render_analysis_tags(uid, root_uid) | safe }}{{ firmware.tags | render_fw_tags | safe }}<br />
                 {% endif %}
                 <span style="font-size: 15px"><strong>UID:</strong> {{ uid | safe }}</span>
             </h3>
@@ -316,7 +316,7 @@ <h5 class="modal-title">Add analysis to file</h5>
                         </div>
                     </div>
                     <script type="text/javascript" src="{{ url_for('static', filename='js/show_analysis_summary.js') }}"></script>
-                    
+
                 {%- endif -%}
             </div>
 
@@ -358,7 +358,7 @@ <h5 class="modal-title">Add analysis to file</h5>
                     </div>
 
                     {%- set is_text_preview = firmware.processed_analysis["file_type"]["result"]['mime'][0:5] == "text/" or firmware.processed_analysis["file_type"]["result"]['mime'][0:6] == "image/" %}
-                    
+
                     <script>
                         const isTextOrImage = {{ 'true' if is_text_preview else 'false' }};
                         let mimeType = '{{ firmware.processed_analysis["file_type"]["result"]["mime"].replace("/", "_") }}';
@@ -446,6 +446,15 @@ <h5 class="modal-title">Add analysis to file</h5>
             document.body.append(radare_form);
             radare_form.submit();
         }
+        document.addEventListener("DOMContentLoaded", function() {
+            // auto load summary if URL parameter "load_summary=true" is set
+            const urlParams = new URLSearchParams(window.location.search);
+            const summary = urlParams.get('load_summary');
+            const has_children = {{ "true" if firmware.files_included | length > 0  else "false" }};
+            if (summary === "true" && has_children && selected_analysis !== "None") {
+                load_summary(uid, selected_analysis, focus=true);
+            }
+        });
     </script>
 
 {% endblock %}

src/web_interface/templates/summary.html

@@ -5,7 +5,7 @@
     </colgroup>
     <thead class="thead-light">
         <tr>
-            <th colspan=2>Summary for Included Files</th>
+            <th id="summary-heading" colspan=2>Summary for Included Files</th>
         </tr>
     </thead>
     {% if summary_of_included_files %}